diff --git a/pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/Storage.java b/pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/Storage.java index 620ff24434..8183b6c7ca 100644 --- a/pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/Storage.java +++ b/pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/Storage.java @@ -44,6 +44,7 @@ enum Scheme file, // local fs s3, // Amazon S3 minio, // Minio + cos, // Tencent Cloud COS redis, // Redis gcs, // google cloud storage httpstream, // http (netty) based stream storage diff --git a/pixels-common/src/main/resources/pixels.properties b/pixels-common/src/main/resources/pixels.properties index 2808285454..3e86fb9680 100644 --- a/pixels-common/src/main/resources/pixels.properties +++ b/pixels-common/src/main/resources/pixels.properties @@ -108,7 +108,7 @@ row.batch.size=10000 pxl.file.timestamp.zone=Asia/Shanghai ### file storage and I/O ### -# the scheme of the storage systems that are enabled, e.g., hdfs,file,s3,gcs,minio,redis,s3qs,httpstream +# the scheme of the storage systems that are enabled, e.g., hdfs,file,s3,gcs,minio,cos,redis,s3qs,httpstream enabled.storage.schemes=s3,minio,file # which scheduler to use for read requests, valid values: noop, sortmerge, ratelimited read.request.scheduler=sortmerge @@ -141,6 +141,10 @@ minio.region=eu-central-2 minio.endpoint=http://minio-host-dummy:9000 minio.access.key=minio-access-key-dummy minio.secret.key=minio-secret-key-dummy +cos.region=ap-guangzhou +cos.endpoint=https://cos.ap-guangzhou.myqcloud.com +cos.access.key=cos-access-key-dummy +cos.secret.key=cos-secret-key-dummy redis.endpoint=localhost:6379 redis.access.key=redis-user-dummy redis.secret.key=redis-password-dummy diff --git a/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/Cos.java b/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/Cos.java new file mode 100644 index 0000000000..c521527131 --- /dev/null +++ b/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/Cos.java @@ -0,0 +1,162 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.storage.s3; + +import io.pixelsdb.pixels.common.physical.StorageFactory; +import io.pixelsdb.pixels.common.utils.ConfigFactory; +import software.amazon.awssdk.http.apache.ApacheHttpClient; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3Client; + +import java.io.IOException; +import java.net.URI; +import java.time.Duration; +import java.util.Objects; + +import static java.util.Objects.requireNonNull; + +/** + * For Tencent Cloud COS, we assume that each table is stored in a separate folder + * (i.e., a prefix or empty object in a bucket). And all the pixels + * files in this table are stored as individual objects in the folder. + *
+ * To reuse the S3-compatible storage implementation, we use AWS S3 SDK to access COS. + *
+ * + * @author Dongyang Geng + * @create 2026-07-07 + */ +public final class Cos extends AbstractS3 +{ + private static final String SchemePrefix = Scheme.cos.name() + "://"; + + private static String cosRegion; + private static String cosEndpoint; + private static String cosAccessKey; + private static String cosSecretKey; + + static + { + cosRegion = ConfigFactory.Instance().getProperty("cos.region"); + cosEndpoint = ConfigFactory.Instance().getProperty("cos.endpoint"); + cosAccessKey = ConfigFactory.Instance().getProperty("cos.access.key"); + cosSecretKey = ConfigFactory.Instance().getProperty("cos.secret.key"); + } + + /** + * Set the configurations for COS. If any configuration is different from the default (null) or + * previous value, the COS storage instance in StorageFactory is reloaded for the configuration + * changes to take effect. In this case, the previous COS storage instance acquired from the + * StorageFactory can be used without any impact. + *
+ * If the configurations are not changed, this method is a no-op. + *
+ * @param region COS region + * @param endpoint COS endpoint + * @param accessKey COS access key + * @param secretKey COS secret key + * @throws IOException if the configuration is invalid + */ + public static void ConfigCos(String region, String endpoint, String accessKey, String secretKey) + throws IOException + { + requireNonNull(region, "region is null"); + requireNonNull(endpoint, "endpoint is null"); + requireNonNull(accessKey, "accessKey is null"); + requireNonNull(secretKey, "secretKey is null"); + + if (!Objects.equals(cosRegion, region) || + !Objects.equals(cosEndpoint, endpoint) || + !Objects.equals(cosAccessKey, accessKey) || + !Objects.equals(cosSecretKey, secretKey)) + { + cosRegion = region; + cosEndpoint = endpoint; + cosAccessKey = accessKey; + cosSecretKey = secretKey; + StorageFactory.Instance().reload(Scheme.cos); + } + } + + public Cos() + { + connect(); + } + + private void connect() + { + requireNonNull(cosRegion, "COS region is not set"); + requireNonNull(cosEndpoint, "COS endpoint is not set"); + requireNonNull(cosAccessKey, "COS access key is not set"); + requireNonNull(cosSecretKey, "COS secret key is not set"); + + // + this.s3 = S3Client.builder().httpClientBuilder(ApacheHttpClient.builder() + .connectionTimeout(Duration.ofSeconds(ConnTimeoutSec)) + .socketTimeout(Duration.ofSeconds(ConnTimeoutSec)) + .connectionAcquisitionTimeout(Duration.ofSeconds(ConnAcquisitionTimeoutSec)) + .maxConnections(MaxRequestConcurrency)) + .endpointOverride(URI.create(cosEndpoint)) + .region(Region.of(cosRegion)) + .credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create(cosAccessKey, cosSecretKey))).build(); + + } + + @Override + public void reconnect() + { + connect(); + } + + // Reuse S3.Path. + + @Override + public Scheme getScheme() + { + return Scheme.cos; + } + + @Override + public String ensureSchemePrefix(String path) throws IOException + { + if (path.startsWith(SchemePrefix)) + { + return path; + } + if (path.contains("://")) + { + throw new IOException("Path '" + path + + "' already has a different scheme prefix than '" + SchemePrefix + "'."); + } + return SchemePrefix + path; + } + + @Override + public void close() throws IOException + { + if (s3 != null) + { + s3.close(); + } + } +} \ No newline at end of file diff --git a/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/CosProvider.java b/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/CosProvider.java new file mode 100644 index 0000000000..5f72e409fe --- /dev/null +++ b/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/CosProvider.java @@ -0,0 +1,71 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.storage.s3; + +import io.pixelsdb.pixels.common.physical.*; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.io.IOException; + +/** + * @author Dongyang Geng + * @create 2026-07-07 + */ +public class CosProvider implements StorageProvider +{ + @Override + public Storage createStorage(@Nonnull Storage.Scheme scheme) throws IOException + { + if (!this.compatibleWith(scheme)) + { + throw new IOException("incompatible storage scheme: " + scheme); + } + return new Cos(); + } + + @Override + public PhysicalReader createReader(@Nonnull Storage storage, @Nonnull String path, + @Nullable PhysicalReaderOption option) throws IOException + { + if (!this.compatibleWith(storage.getScheme())) + { + throw new IOException("incompatible storage scheme: " + storage.getScheme()); + } + return new PhysicalCosReader(storage, path); + } + + @Override + public PhysicalWriter createWriter(@Nonnull Storage storage, @Nonnull String path, + @Nonnull PhysicalWriterOption option) throws IOException + { + if (!this.compatibleWith(storage.getScheme())) + { + throw new IOException("incompatible storage scheme: " + storage.getScheme()); + } + return new PhysicalS3Writer(storage, path, option.isOverwrite()); + } + + @Override + public boolean compatibleWith(@Nonnull Storage.Scheme scheme) + { + return scheme.equals(Storage.Scheme.cos); + } +} diff --git a/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/PhysicalCosReader.java b/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/PhysicalCosReader.java new file mode 100644 index 0000000000..819b16b861 --- /dev/null +++ b/pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/PhysicalCosReader.java @@ -0,0 +1,54 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.storage.s3; + +import io.pixelsdb.pixels.common.physical.Storage; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.CompletableFuture; + +/** + * The physical reader for COS. + * + * @author Dongyang Geng + * @create 2026-07-07 + */ +public class PhysicalCosReader extends AbstractS3Reader +{ + public PhysicalCosReader(Storage storage, String path) throws IOException + { + super(storage, path); + this.enableAsync = false; + } + + @Override + public CompletableFuture readAsync(long offset, int len) throws IOException + { + throw new UnsupportedOperationException("asynchronous read is not supported for COS."); + } + + @Override + public void close() throws IOException + { + // Should not close the client because it is shared by all threads. + // this.client.close(); // Closing s3 client may take several seconds. + } +} diff --git a/pixels-storage/pixels-storage-s3/src/main/resources/META-INF/services/io.pixelsdb.pixels.common.physical.StorageProvider b/pixels-storage/pixels-storage-s3/src/main/resources/META-INF/services/io.pixelsdb.pixels.common.physical.StorageProvider index bb65463f1d..91247f3b54 100644 --- a/pixels-storage/pixels-storage-s3/src/main/resources/META-INF/services/io.pixelsdb.pixels.common.physical.StorageProvider +++ b/pixels-storage/pixels-storage-s3/src/main/resources/META-INF/services/io.pixelsdb.pixels.common.physical.StorageProvider @@ -1,2 +1,3 @@ io.pixelsdb.pixels.storage.s3.S3Provider -io.pixelsdb.pixels.storage.s3.MinioProvider \ No newline at end of file +io.pixelsdb.pixels.storage.s3.MinioProvider +io.pixelsdb.pixels.storage.s3.CosProvider \ No newline at end of file diff --git a/pixels-storage/pixels-storage-s3/src/test/java/io/pixelsdb/pixels/storage/s3/TestCos.java b/pixels-storage/pixels-storage-s3/src/test/java/io/pixelsdb/pixels/storage/s3/TestCos.java new file mode 100644 index 0000000000..99d949804a --- /dev/null +++ b/pixels-storage/pixels-storage-s3/src/test/java/io/pixelsdb/pixels/storage/s3/TestCos.java @@ -0,0 +1,70 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.storage.s3; + +import io.pixelsdb.pixels.common.physical.*; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; +import java.util.UUID; +import java.nio.charset.StandardCharsets; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +/** + * @author Dongyang Geng + * @create 2026-07-07 + */ +public class TestCos +{ + @Test + public void testReadWrite() throws IOException + { + Storage cos = StorageFactory.Instance().getStorage(Storage.Scheme.cos); + String path = "cos://bucket/pixels-cos-test/"; + List files = cos.listPaths(path); + for (String file : files) + { + System.out.println(file); + } + + String testPath = path + "pixels-cos-test-" + UUID.randomUUID().toString(); + String content = "hello from pixels cos storage"; + + // write the file + PhysicalWriter writer = PhysicalWriterUtil.newPhysicalWriter(cos, testPath, true); + writer.append(content.getBytes(StandardCharsets.UTF_8), 0, content.length()); + writer.close(); + + // read the file + PhysicalReader reader = PhysicalReaderUtil.newPhysicalReader(cos, testPath); + byte[] actual = new byte[(int) reader.getFileLength()]; + reader.readFully(actual); + reader.close(); + String actualContent = new String(actual, StandardCharsets.UTF_8); + assertEquals(content, actualContent); + + // delete the file + cos.delete(testPath, false); + assertFalse(cos.exists(testPath)); + } +}