zjw1111 commented on code in PR #205:
URL: https://github.com/apache/paimon-cpp/pull/205#discussion_r3794760669
##########
src/paimon/CMakeLists.txt:
##########
@@ -190,11 +190,13 @@ set(PAIMON_COMMON_SRCS
# The shared HTTP client is used by both the object store file systems and the
# rest catalog.
set(PAIMON_CURL_LINK_LIBS)
-if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)
+if(PAIMON_ENABLE_OSS
Review Comment:
The matching test gate further down (`PAIMON_COMMON_HTTP_CLIENT_TEST_SRCS`,
still `if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)`) was not updated, and the
comment right above it says "so its test follows the same gate". With
`-DPAIMON_ENABLE_OSS=ON` and S3/REST off, `http_client.cpp` is compiled into
`paimon_shared` while `http_client_test.cpp` is excluded.
That said, I wonder whether this hunk is needed at all. The OSS plugin uses
the SDK's own curl transport, and neither `oss_file_system.cpp` /
`oss_file_system_factory.cpp` nor `common/fs/object_store_file_system.cpp`
includes `paimon/common/utils/http_client.h`.
`src/paimon/fs/oss/CMakeLists.txt` already lists `CURL::libcurl` in its own
`DEPENDENCIES`, and `alibabacloud_oss_v2::oss` links it via INTERFACE, so
`PAIMON_CURL_LINK_LIBS` is not required here either.
Could you revert this hunk and keep only the `object_store_file_system.cpp`
gate below? The top-level `CMakeLists.txt` change that makes CURL discoverable
for the SDK build should of course stay. If you would rather keep it, then the
test gate needs the same update.
##########
cmake_modules/ThirdpartyToolchain.cmake:
##########
@@ -1967,6 +2081,9 @@ if(PAIMON_ENABLE_JINDO)
build_jindosdk_c()
build_jindosdk_nextarch()
endif()
+if(PAIMON_ENABLE_OSS)
+ resolve_dependency(OSS_SDK_V2)
Review Comment:
`resolve_dependency` takes its SYSTEM branch through
`find_package(${DEPENDENCY_NAME}Alt REQUIRED MODULE)`, and all 16 existing
dependencies routed through this macro ship a matching
`cmake_modules/Find<Name>Alt.cmake`. This PR does not add
`FindOSS_SDK_V2Alt.cmake`, and `OSS_SDK_V2_SOURCE` is not declared in
`DefineOptions.cmake` either (every other dependency has one).
The default `AUTO` path uses a QUIET lookup and falls back to bundled, so CI
does not hit this — but `-DPAIMON_DEPENDENCY_SOURCE=SYSTEM`, which is a
documented option, fails at configure time with a "Could not find module
FindOSS_SDK_V2Alt.cmake" message that gives no hint it is about OSS.
Would you prefer to add the Find module plus the `OSS_SDK_V2_SOURCE` option,
or to declare the SDK bundled-only? For the latter,
`paimon_set_dependency_source_default(OSS_SDK_V2 BUNDLED "...")` is already
used for Arrow and ORC and would shield this from the global setting.
##########
src/paimon/fs/oss/oss_file_system_factory.cpp:
##########
@@ -0,0 +1,156 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system_factory.h"
+
+#include <algorithm>
+#include <cctype>
+#include <memory>
+#include <string>
+
+#include "alibabacloud/oss2/ClientConfiguration.h"
+#include "alibabacloud/oss2/OSSClient.h"
+#include "alibabacloud/oss2/credentials/CredentialsProvider.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/factories/factory.h"
+#include "paimon/fs/oss/oss_file_system.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+std::string GetOption(const std::map<std::string, std::string>& options, const
std::string& bucket,
+ const std::string& suffix) {
+ auto bucket_option = options.find("fs.oss.bucket." + bucket + "." +
suffix);
+ if (bucket_option != options.end()) {
+ return bucket_option->second;
+ }
+ auto global_option = options.find("fs.oss." + suffix);
+ return global_option == options.end() ? "" : global_option->second;
+}
+
+Result<std::string> GetRequiredOption(const std::map<std::string,
std::string>& options,
+ const std::string& bucket, const
std::string& suffix) {
+ std::string value = GetOption(options, bucket, suffix);
+ if (value.empty()) {
+ return Status::Invalid(fmt::format("OSS option 'fs.oss.{}' must not be
empty", suffix));
+ }
+ return value;
+}
+
+Result<bool> ParseBool(std::string value, const std::string& option) {
+ std::transform(value.begin(), value.end(), value.begin(),
+ [](unsigned char c) { return
static_cast<char>(std::tolower(c)); });
+ if (value == "true" || value == "1" || value == "yes" || value == "on") {
+ return true;
+ }
+ if (value == "false" || value == "0" || value == "no" || value == "off") {
+ return false;
+ }
+ return Status::Invalid(
+ fmt::format("invalid boolean value '{}' for OSS option '{}'", value,
option));
+}
+
+std::string InferRegion(std::string endpoint) {
+ constexpr std::string_view kPrefix = "oss-";
+ constexpr std::string_view kSuffix = ".aliyuncs.com";
+ size_t scheme = endpoint.find("://");
+ if (scheme != std::string::npos) {
+ endpoint.erase(0, scheme + 3);
+ }
+ size_t slash = endpoint.find('/');
+ if (slash != std::string::npos) {
+ endpoint.erase(slash);
+ }
+ if (endpoint.rfind(kPrefix, 0) != 0) {
+ return "";
+ }
+ size_t suffix = endpoint.find(kSuffix);
+ if (suffix == std::string::npos) {
+ return "";
+ }
+ std::string region = endpoint.substr(kPrefix.size(), suffix -
kPrefix.size());
+ constexpr std::string_view kInternal = "-internal";
+ if (region.size() > kInternal.size() &&
+ region.compare(region.size() - kInternal.size(), kInternal.size(),
kInternal) == 0) {
+ region.erase(region.size() - kInternal.size());
+ }
+ return region;
+}
+
+std::string NormalizeEndpoint(std::string endpoint) {
+ if (!endpoint.empty() && endpoint.find("://") == std::string::npos) {
+ endpoint = "https://"; + endpoint;
+ }
+ return endpoint;
+}
+
+} // namespace
+
+const char OssFileSystemFactory::IDENTIFIER[] = "oss";
+
+Result<std::unique_ptr<FileSystem>> OssFileSystemFactory::Create(
+ const std::string& path, const std::map<std::string, std::string>&
options) const {
+ PAIMON_ASSIGN_OR_RAISE(Path parsed_path, PathUtil::ToPath(path));
+ if (parsed_path.scheme != "oss" || parsed_path.authority.empty()) {
+ return Status::Invalid(fmt::format("invalid OSS path '{}'", path));
+ }
+ const std::string& bucket = parsed_path.authority;
+ PAIMON_ASSIGN_OR_RAISE(std::string access_key_id,
+ GetRequiredOption(options, bucket, "accessKeyId"));
+ PAIMON_ASSIGN_OR_RAISE(std::string access_key_secret,
+ GetRequiredOption(options, bucket,
"accessKeySecret"));
+ std::string endpoint = GetOption(options, bucket, "endpoint");
+ std::string region = GetOption(options, bucket, "region");
+ if (region.empty()) {
+ region = InferRegion(endpoint);
+ }
+ if (endpoint.empty() && region.empty()) {
+ return Status::Invalid("OSS endpoint or region must be configured");
+ }
+ std::string security_token = GetOption(options, bucket, "securityToken");
+ if (security_token.empty()) {
+ security_token = GetOption(options, bucket, "sessionToken");
+ }
+
+ oss2::ClientConfiguration config =
oss2::ClientConfiguration::loadDefault();
+ if (!endpoint.empty()) {
+ config.endpoint = NormalizeEndpoint(endpoint);
+ }
+ config.region = region;
Review Comment:
Thanks for wiring up the region handling here. I think there is a gap worth
closing before this lands.
`InferRegion` only recognizes `oss-<region>[-internal].aliyuncs.com` and
returns `""` for anything else (custom CNAME domains,
`oss-accelerate.aliyuncs.com`, transfer-acceleration and dual-stack endpoints).
Since the guard above only rejects the case where *both* endpoint and region
are empty, configuring just `fs.oss.endpoint` with one of those forms passes
validation and leaves `config.region` as an empty string.
The SDK defaults to v4 signing (`ClientImplBase::resolveSigner`: `if ("v1"
== config.signatureVersion.value_or("v4"))`) and passes the region through
unvalidated (`options_.region = config.region.value_or("")`), so `SignerV4`
ends up building a credential scope of `<date>//oss/aliyun_v4_request`. Every
request then fails server-side with `SignatureDoesNotMatch`, and because
`Create()` succeeds the user only sees it at the first I/O rather than at
configuration time.
Could you make the missing-region case explicit? Either returning
`Status::Invalid` asking for `fs.oss.region` when inference fails and no
explicit `signatureVersion` was given, or falling back to v1 signing (which
needs no region, and is closer to what the Hadoop OSS connector does on the
Java side) would work. It would also be a little safer to assign
`config.region` only when it is non-empty, so the SDK's own `has_value()`
checks stay meaningful.
##########
src/paimon/fs/oss/oss_file_system_factory.cpp:
##########
@@ -0,0 +1,156 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system_factory.h"
+
+#include <algorithm>
+#include <cctype>
+#include <memory>
+#include <string>
+
+#include "alibabacloud/oss2/ClientConfiguration.h"
+#include "alibabacloud/oss2/OSSClient.h"
+#include "alibabacloud/oss2/credentials/CredentialsProvider.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/factories/factory.h"
+#include "paimon/fs/oss/oss_file_system.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+std::string GetOption(const std::map<std::string, std::string>& options, const
std::string& bucket,
+ const std::string& suffix) {
+ auto bucket_option = options.find("fs.oss.bucket." + bucket + "." +
suffix);
+ if (bucket_option != options.end()) {
+ return bucket_option->second;
+ }
+ auto global_option = options.find("fs.oss." + suffix);
+ return global_option == options.end() ? "" : global_option->second;
+}
+
+Result<std::string> GetRequiredOption(const std::map<std::string,
std::string>& options,
+ const std::string& bucket, const
std::string& suffix) {
+ std::string value = GetOption(options, bucket, suffix);
+ if (value.empty()) {
+ return Status::Invalid(fmt::format("OSS option 'fs.oss.{}' must not be
empty", suffix));
+ }
+ return value;
+}
+
+Result<bool> ParseBool(std::string value, const std::string& option) {
Review Comment:
There is already a shared helper for this:
`StringUtils::StringToValue<bool>` (`common/utils/string_utils.h`), and
`OptionsUtils::GetValueFromMap<bool>` wraps it into the `Result<T>` form that
the S3 factory uses (`s3_file_system.cpp:646`). Could you reuse one of those
instead of a local `ParseBool`?
One thing to be aware of: the accepted spellings differ slightly. The shared
helper takes `t/true/y/yes/1` and `f/false/n/no/0`, so `on` and `off` would no
longer be accepted. Aligning with the shared helper still seems preferable to
keeping a separate dialect just for OSS options.
##########
src/paimon/fs/oss/oss_file_system_test.cpp:
##########
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system.h"
+
+#include <map>
+#include <string>
+
+#include "gtest/gtest.h"
+#include "paimon/fs/oss/oss_file_system_factory.h"
+
+namespace paimon::oss {
+
+TEST(OssFileSystemFactoryTest, TestOptionValidation) {
+ OssFileSystemFactory factory;
+ std::map<std::string, std::string> options;
+ ASSERT_FALSE(factory.Create("s3://bucket/key", options).ok());
Review Comment:
Could you use `ASSERT_NOK` / `ASSERT_OK` from
`paimon/testing/utils/testharness.h` instead of `ASSERT_FALSE(...ok())` /
`ASSERT_TRUE(...ok())`? They accept `Result<T>` directly and print the actual
status on failure, which makes diagnosing a broken test much easier. This
applies to the other assertions in this file as well (lines 31, 37, 41, 44, 48
and 61); `s3_file_system_test.cpp` already uses these macros.
##########
src/paimon/fs/oss/oss_file_system.cpp:
##########
@@ -0,0 +1,188 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system.h"
+
+#include <curl/curl.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <ctime>
+#include <memory>
+#include <utility>
+
+#include "alibabacloud/oss2/OSSClient.h"
+#include "alibabacloud/oss2/Operation.h"
+#include "alibabacloud/oss2/Types.h"
+#include "alibabacloud/oss2/io/ByteWriter.h"
+#include "alibabacloud/oss2/models/BucketBasic.h"
+#include "alibabacloud/oss2/models/ObjectBasic.h"
+#include "fmt/format.h"
+#include "paimon/executor.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+bool IsNotFoundError(const oss2::OperationError& error) {
+ return error.getStatusCode() == 404 || error.getCode() == "NoSuchKey" ||
+ error.getCode() == "NoSuchBucket" || error.getCode() == "NotFound";
+}
+
+Status ToPaimonStatus(const oss2::OperationError& error, const std::string&
operation,
+ const ObjectStorePath& path) {
+ std::string message = fmt::format("OSS {} 'oss://{}/{}' failed: code={},
status={}, message={}",
+ operation, path.bucket, path.key,
error.getCode(),
+ error.getStatusCode(),
error.getMessage());
+ if (!error.getRequestId().empty()) {
+ message += fmt::format(", request_id={}", error.getRequestId());
+ }
+ if (IsNotFoundError(error)) {
+ return Status::NotExist(message);
+ }
+ if (error.getCode() == "RequestCanceled") {
+ return Status::Cancelled(message);
+ }
+ return Status::IOError(message);
+}
+
+int64_t ParseTimeMillis(const std::string& value) {
+ time_t seconds = curl_getdate(value.c_str(), nullptr);
+ return seconds < 0 ? 0 : static_cast<int64_t>(seconds) * 1000;
+}
+
+class OssObjectStoreClient : public ObjectStoreClient,
+ public
std::enable_shared_from_this<OssObjectStoreClient> {
+ public:
+ OssObjectStoreClient(std::string bucket, std::shared_ptr<oss2::OSSClient>
client)
+ : bucket_(std::move(bucket)), client_(std::move(client)) {}
+
+ Result<ObjectMetadata> HeadObject(const ObjectStorePath& path) const
override {
+ PAIMON_RETURN_NOT_OK(ValidateBucket(path));
+ oss2::models::HeadObjectRequest request;
+ request.setBucket(path.bucket).setKey(path.key);
+ oss2::HeadObjectOutcome outcome = client_->headObject(request);
+ if (!outcome.has_value()) {
+ return ToPaimonStatus(outcome.error(), "HeadObject", path);
+ }
+ const oss2::models::HeadObjectResult& result = outcome.value();
+ if (result.getContentLength() < 0) {
+ return Status::IOError("OSS HeadObject response is missing
Content-Length");
+ }
+ return ObjectMetadata{path.key, result.getContentLength(),
+ ParseTimeMillis(result.getLastModified())};
+ }
+
+ Result<ListObjectsResult> ListObjects(const ObjectStorePath& path,
+ const std::string&
continuation_token,
+ int max_keys) const override {
+ PAIMON_RETURN_NOT_OK(ValidateBucket(path));
+ oss2::models::ListObjectsV2Request request;
+ request.setBucket(path.bucket).setPrefix(path.key).setDelimiter("/");
+ if (!continuation_token.empty()) {
+ request.setContinuationToken(continuation_token);
+ }
+ if (max_keys > 0) {
+ request.setMaxKeys(std::min(max_keys, 999));
+ }
+ oss2::ListObjectsV2Outcome outcome = client_->listObjectsV2(request);
+ if (!outcome.has_value()) {
+ return ToPaimonStatus(outcome.error(), "ListObjectsV2", path);
+ }
+ const oss2::models::ListObjectsV2Result& value = outcome.value();
+ ListObjectsResult result;
+ result.objects.reserve(value.getContents().size());
+ for (const oss2::models::ObjectSummary& object : value.getContents()) {
+ result.objects.push_back(
+ ObjectMetadata{object.key, object.size,
ParseTimeMillis(object.lastModified)});
+ }
+ result.common_prefixes.reserve(value.getCommonPrefixes().size());
+ for (const oss2::models::CommonPrefix& prefix :
value.getCommonPrefixes()) {
+ result.common_prefixes.push_back(prefix.prefix);
+ }
+ result.is_truncated = value.getIsTruncated();
+ result.continuation_token = value.getNextContinuationToken();
+ return result;
+ }
+
+ Result<int64_t> GetObjectRange(const ObjectStorePath& path, int64_t
offset, int64_t size,
+ char* buffer) const override {
+ PAIMON_RETURN_NOT_OK(ValidateBucket(path));
+ if (size == 0) {
+ return 0;
+ }
+ auto writer = std::make_shared<std::shared_ptr<oss2::MemoryWriter>>();
+ oss2::SinkFactory sink;
+ sink.isOneShot = false;
+ sink.supplier = [buffer, size, writer](int64_t, const
oss2::HeaderCollection&) {
+ auto memory_writer = std::make_shared<oss2::MemoryWriter>(
+ reinterpret_cast<uint8_t*>(buffer), static_cast<size_t>(size));
+ *writer = memory_writer;
+ return memory_writer;
+ };
+ oss2::models::GetObjectRequest request;
+ request.setBucket(path.bucket)
+ .setKey(path.key)
+ .setRange(fmt::format("bytes={}-{}", offset, offset + size - 1))
+ .setRangeBehavior("standard")
+ .setSinkFactory(std::move(sink));
+ oss2::GetObjectOutcome outcome = client_->getObject(request);
+ if (!outcome.has_value()) {
+ return ToPaimonStatus(outcome.error(), "GetObject", path);
+ }
+ int64_t written =
+ *writer ? static_cast<int64_t>((*writer)->written()) :
static_cast<int64_t>(0);
+ if (written != size) {
+ return Status::IOError(
+ fmt::format("OSS GetObject read {} bytes for oss://{}/{},
expected {}", written,
+ path.bucket, path.key, size));
+ }
+ return written;
+ }
+
+ void GetObjectRangeAsync(const ObjectStorePath& path, int64_t offset,
int64_t size,
+ char* buffer, std::function<void(Status)>&&
callback) const override {
+ std::shared_ptr<const OssObjectStoreClient> self = shared_from_this();
+ GetGlobalDefaultExecutor()->Add([self = std::move(self), path, offset,
size, buffer,
Review Comment:
This reuses the process-wide singleton pool. `GetGlobalDefaultExecutor()`
(`common/executor/executor.cpp:139`) is a function-local static sized to
`hardware_concurrency()`, and it currently has no production callers at all —
only `default_executor_test.cpp` — so this would be the first one.
The S3 client instead owns a dedicated `std::unique_ptr<Executor>` built by
`CreateDefaultExecutor()` (`s3_file_system.cpp:904` and `:960`). Since
`GetObjectRangeAsync` is the only implementation path behind
`ObjectStoreFileSystem::ReadAsync`, sharing the global pool means that once
callers also submit work to it and block on OSS reads inside those tasks, the
fixed-size pool can starve itself. It also makes the thread count
non-configurable per file system.
Could you follow the S3 pattern and hold a dedicated `Executor` in
`OssObjectStoreClient`, optionally exposing the thread count through an
`fs.oss.*` option?
##########
src/paimon/fs/oss/oss_file_system_test.cpp:
##########
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system.h"
+
+#include <map>
+#include <string>
+
+#include "gtest/gtest.h"
+#include "paimon/fs/oss/oss_file_system_factory.h"
+
+namespace paimon::oss {
+
+TEST(OssFileSystemFactoryTest, TestOptionValidation) {
+ OssFileSystemFactory factory;
+ std::map<std::string, std::string> options;
+ ASSERT_FALSE(factory.Create("s3://bucket/key", options).ok());
+ ASSERT_FALSE(factory.Create("oss://bucket/key", options).ok());
+
+ options[kOssAccessKeyIdOption] = "access-key";
+ options[kOssAccessKeySecretOption] = "secret-key";
+ options[kOssEndpointOption] = "oss-cn-hangzhou.aliyuncs.com";
+ options[kOssUsePathStyleOption] = "treu";
+ ASSERT_FALSE(factory.Create("oss://bucket/key", options).ok());
+
+ options[kOssUsePathStyleOption] = "false";
+ options[kOssSignatureVersionOption] = "v2";
+ ASSERT_FALSE(factory.Create("oss://bucket/key", options).ok());
+
+ options[kOssSignatureVersionOption] = "v4";
+ ASSERT_TRUE(factory.Create("oss://bucket/key", options).ok());
+
+ options[kOssEndpointOption] = "";
+ options[kOssRegionOption] = "cn-hangzhou";
+ ASSERT_TRUE(factory.Create("oss://bucket/key", options).ok());
+}
+
+TEST(OssFileSystemFactoryTest, TestBucketOptionsOverrideGlobalOptions) {
+ OssFileSystemFactory factory;
+ std::map<std::string, std::string> options = {
+ {kOssAccessKeyIdOption, ""},
+ {kOssAccessKeySecretOption, ""},
+ {kOssEndpointOption, ""},
+ {"fs.oss.bucket.bucket.accessKeyId", "access-key"},
+ {"fs.oss.bucket.bucket.accessKeySecret", "secret-key"},
+ {"fs.oss.bucket.bucket.endpoint", "oss-cn-hangzhou.aliyuncs.com"},
+ };
+ ASSERT_TRUE(factory.Create("oss://bucket/key", options).ok());
+}
Review Comment:
Thanks for adding the option-validation tests. The client layer is still
uncovered though: `HeadObject`, `ListObjects`, `GetObjectRange` and
`GetObjectRangeAsync` have no tests, and neither do `InferRegion` or
`ParseTimeMillis`. For comparison, `s3_file_system_test.cpp` is around 500
lines and covers response parsing plus error paths through a `MockHttpClient`.
The equivalent seam exists for OSS:
`alibabacloud/oss2/transport/HttpTransport.h` exposes a public abstract
`HttpTransport` (with `NopHttpTransport` as a starting point) and
`ClientConfiguration::httpTransport` accepts an injected instance, so this can
be tested without any network access.
Would it be possible to add at least table-driven cases for `InferRegion`
(standard / internal / accelerate / custom domain), a malformed-input case for
`ParseTimeMillis`, and success plus error paths for `HeadObject`, `ListObjects`
and `GetObjectRange`? The error mapping in `IsNotFoundError`, the range-header
construction and the short-read check are hard to verify any other way.
##########
src/paimon/fs/oss/oss_file_system.cpp:
##########
@@ -0,0 +1,188 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system.h"
+
+#include <curl/curl.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <ctime>
+#include <memory>
+#include <utility>
+
+#include "alibabacloud/oss2/OSSClient.h"
+#include "alibabacloud/oss2/Operation.h"
+#include "alibabacloud/oss2/Types.h"
+#include "alibabacloud/oss2/io/ByteWriter.h"
+#include "alibabacloud/oss2/models/BucketBasic.h"
+#include "alibabacloud/oss2/models/ObjectBasic.h"
+#include "fmt/format.h"
+#include "paimon/executor.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+bool IsNotFoundError(const oss2::OperationError& error) {
+ return error.getStatusCode() == 404 || error.getCode() == "NoSuchKey" ||
+ error.getCode() == "NoSuchBucket" || error.getCode() == "NotFound";
+}
+
+Status ToPaimonStatus(const oss2::OperationError& error, const std::string&
operation,
+ const ObjectStorePath& path) {
+ std::string message = fmt::format("OSS {} 'oss://{}/{}' failed: code={},
status={}, message={}",
+ operation, path.bucket, path.key,
error.getCode(),
+ error.getStatusCode(),
error.getMessage());
+ if (!error.getRequestId().empty()) {
+ message += fmt::format(", request_id={}", error.getRequestId());
+ }
+ if (IsNotFoundError(error)) {
+ return Status::NotExist(message);
+ }
+ if (error.getCode() == "RequestCanceled") {
+ return Status::Cancelled(message);
+ }
+ return Status::IOError(message);
+}
+
+int64_t ParseTimeMillis(const std::string& value) {
+ time_t seconds = curl_getdate(value.c_str(), nullptr);
+ return seconds < 0 ? 0 : static_cast<int64_t>(seconds) * 1000;
+}
+
+class OssObjectStoreClient : public ObjectStoreClient,
+ public
std::enable_shared_from_this<OssObjectStoreClient> {
+ public:
+ OssObjectStoreClient(std::string bucket, std::shared_ptr<oss2::OSSClient>
client)
+ : bucket_(std::move(bucket)), client_(std::move(client)) {}
+
+ Result<ObjectMetadata> HeadObject(const ObjectStorePath& path) const
override {
+ PAIMON_RETURN_NOT_OK(ValidateBucket(path));
+ oss2::models::HeadObjectRequest request;
+ request.setBucket(path.bucket).setKey(path.key);
+ oss2::HeadObjectOutcome outcome = client_->headObject(request);
+ if (!outcome.has_value()) {
+ return ToPaimonStatus(outcome.error(), "HeadObject", path);
+ }
+ const oss2::models::HeadObjectResult& result = outcome.value();
+ if (result.getContentLength() < 0) {
+ return Status::IOError("OSS HeadObject response is missing
Content-Length");
+ }
+ return ObjectMetadata{path.key, result.getContentLength(),
+ ParseTimeMillis(result.getLastModified())};
+ }
+
+ Result<ListObjectsResult> ListObjects(const ObjectStorePath& path,
+ const std::string&
continuation_token,
+ int max_keys) const override {
Review Comment:
`docs/code-style.md` asks for fixed-width integer types, and the base
declaration is `int32_t max_keys` (`common/fs/object_store_file_system.h:59`);
the S3 client uses `int32_t` as well. Could you change this to `int32_t`?
##########
src/paimon/fs/oss/oss_file_system.h:
##########
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "paimon/common/fs/object_store_file_system.h"
+
+namespace alibabacloud::oss2 {
+class OSSClient;
+}
+
+namespace paimon::oss {
+
+inline constexpr char kOssAccessKeyIdOption[] = "fs.oss.accessKeyId";
Review Comment:
These constants and the keys actually being looked up are two separate
sources of truth: the factory builds its keys from literal suffixes
(`GetOption(options, bucket, "accessKeyId")`), so only `kOssUsePathStyleOption`
is referenced from production code, and `kOssSecurityTokenOption` /
`kOssSessionTokenOption` are not referenced anywhere. Renaming or adding an
option means remembering to update both places.
Could you either have `GetOption` take the full constant and derive the
bucket-scoped key from it, or drop the ones that are unused? Relatedly, the
error message in `GetRequiredOption` ("OSS option 'fs.oss.{}' must not be
empty") always reports the global key, even when the bucket-scoped
`fs.oss.bucket.<bucket>.<suffix>` is the one that was being looked up.
##########
src/paimon/fs/oss/oss_file_system.cpp:
##########
@@ -0,0 +1,188 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/fs/oss/oss_file_system.h"
+
+#include <curl/curl.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <ctime>
+#include <memory>
+#include <utility>
+
+#include "alibabacloud/oss2/OSSClient.h"
+#include "alibabacloud/oss2/Operation.h"
+#include "alibabacloud/oss2/Types.h"
+#include "alibabacloud/oss2/io/ByteWriter.h"
+#include "alibabacloud/oss2/models/BucketBasic.h"
+#include "alibabacloud/oss2/models/ObjectBasic.h"
+#include "fmt/format.h"
+#include "paimon/executor.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+bool IsNotFoundError(const oss2::OperationError& error) {
+ return error.getStatusCode() == 404 || error.getCode() == "NoSuchKey" ||
+ error.getCode() == "NoSuchBucket" || error.getCode() == "NotFound";
+}
+
+Status ToPaimonStatus(const oss2::OperationError& error, const std::string&
operation,
+ const ObjectStorePath& path) {
+ std::string message = fmt::format("OSS {} 'oss://{}/{}' failed: code={},
status={}, message={}",
+ operation, path.bucket, path.key,
error.getCode(),
+ error.getStatusCode(),
error.getMessage());
+ if (!error.getRequestId().empty()) {
+ message += fmt::format(", request_id={}", error.getRequestId());
+ }
+ if (IsNotFoundError(error)) {
+ return Status::NotExist(message);
+ }
+ if (error.getCode() == "RequestCanceled") {
+ return Status::Cancelled(message);
+ }
+ return Status::IOError(message);
+}
+
+int64_t ParseTimeMillis(const std::string& value) {
+ time_t seconds = curl_getdate(value.c_str(), nullptr);
+ return seconds < 0 ? 0 : static_cast<int64_t>(seconds) * 1000;
+}
+
+class OssObjectStoreClient : public ObjectStoreClient,
+ public
std::enable_shared_from_this<OssObjectStoreClient> {
+ public:
+ OssObjectStoreClient(std::string bucket, std::shared_ptr<oss2::OSSClient>
client)
+ : bucket_(std::move(bucket)), client_(std::move(client)) {}
+
+ Result<ObjectMetadata> HeadObject(const ObjectStorePath& path) const
override {
+ PAIMON_RETURN_NOT_OK(ValidateBucket(path));
+ oss2::models::HeadObjectRequest request;
+ request.setBucket(path.bucket).setKey(path.key);
+ oss2::HeadObjectOutcome outcome = client_->headObject(request);
+ if (!outcome.has_value()) {
+ return ToPaimonStatus(outcome.error(), "HeadObject", path);
+ }
+ const oss2::models::HeadObjectResult& result = outcome.value();
+ if (result.getContentLength() < 0) {
+ return Status::IOError("OSS HeadObject response is missing
Content-Length");
+ }
+ return ObjectMetadata{path.key, result.getContentLength(),
+ ParseTimeMillis(result.getLastModified())};
+ }
+
+ Result<ListObjectsResult> ListObjects(const ObjectStorePath& path,
+ const std::string&
continuation_token,
+ int max_keys) const override {
+ PAIMON_RETURN_NOT_OK(ValidateBucket(path));
+ oss2::models::ListObjectsV2Request request;
+ request.setBucket(path.bucket).setPrefix(path.key).setDelimiter("/");
+ if (!continuation_token.empty()) {
+ request.setContinuationToken(continuation_token);
+ }
+ if (max_keys > 0) {
+ request.setMaxKeys(std::min(max_keys, 999));
Review Comment:
Minor: the OSS `max-keys` limit is 1000, so 999 reads as an unexplained
magic number. The base class also only ever calls this with 1 or 0 today
(`object_store_file_system.cpp:367` and `:434`), so the clamp never actually
kicks in. Could you use 1000 with a short comment, or drop the clamp?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
