mrdrivingduck commented on code in PR #205:
URL: https://github.com/apache/paimon-cpp/pull/205#discussion_r3837978999
##########
cmake_modules/ThirdpartyToolchain.cmake:
##########
@@ -1359,6 +1373,106 @@ macro(build_jindosdk_nextarch)
add_dependencies(jindosdk::nextarch jindosdk-nextarch_ep)
endmacro()
+macro(build_oss_sdk_v2)
+ message(STATUS "Building Alibaba Cloud OSS C++ SDK v2 from source")
+ find_package(CURL REQUIRED)
+ find_package(Threads REQUIRED)
+
+ set(OSS_SDK_V2_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/oss_sdk_v2_ep-install")
+ set(OSS_SDK_V2_INCLUDE_DIR "${OSS_SDK_V2_PREFIX}/include")
+ set(OSS_SDK_V2_INSTALL_LIBDIR "${CMAKE_INSTALL_LIBDIR}")
+ set(OSS_SDK_V2_LIB_DIR "${OSS_SDK_V2_PREFIX}/${OSS_SDK_V2_INSTALL_LIBDIR}")
+ set(OSS_SDK_V2_STATIC_LIB
+
"${OSS_SDK_V2_LIB_DIR}/${CMAKE_STATIC_LIBRARY_PREFIX}alibabacloud-oss-cpp-sdk-v2${CMAKE_STATIC_LIBRARY_SUFFIX}"
+ )
+
+ set(OSS_SDK_V2_CMAKE_CXX_FLAGS "${EP_CXX_FLAGS} -Wno-error")
+ set(OSS_SDK_V2_CMAKE_C_FLAGS "${EP_C_FLAGS} -Wno-error")
+ string(REPLACE "-Werror" "" OSS_SDK_V2_CMAKE_CXX_FLAGS
${OSS_SDK_V2_CMAKE_CXX_FLAGS})
+
+ set(OSS_SDK_V2_CMAKE_ARGS
+ ${EP_COMMON_CMAKE_ARGS}
+ "-DCMAKE_INSTALL_PREFIX=${OSS_SDK_V2_PREFIX}"
+ "-DCMAKE_INSTALL_LIBDIR=${OSS_SDK_V2_INSTALL_LIBDIR}"
+ "-DCMAKE_CXX_FLAGS=${OSS_SDK_V2_CMAKE_CXX_FLAGS}"
+ "-DCMAKE_C_FLAGS=${OSS_SDK_V2_CMAKE_C_FLAGS}"
+ -DCMAKE_PARENT_CXX_STANDARD=17
+ -DBUILD_SHARED_LIBS=OFF
+ -DBUILD_TESTS=OFF
+ -DBUILD_SAMPLES=OFF
+ -DENABLE_RTTI=OFF
+ -DUSE_CURL_TRANSPORT=ON
+ -DUSE_SYSTEM_CURL=ON
+ -DUSE_SYSTEM_OPENSSL=OFF
+ -DUSE_SYSTEM_MBEDTLS=OFF
+ -DUSE_SYSTEM_TINYXML2=OFF
+ -DUSE_STD_EXPECTED=OFF
+ -DENABLE_ENCRYPTION=OFF)
+ set(OSS_SDK_V2_CURL_INCLUDE_DIR "${CURL_INCLUDE_DIR}")
+ if(NOT OSS_SDK_V2_CURL_INCLUDE_DIR AND CURL_INCLUDE_DIRS)
+ list(GET CURL_INCLUDE_DIRS 0 OSS_SDK_V2_CURL_INCLUDE_DIR)
+ endif()
+ set(OSS_SDK_V2_CURL_LIBRARY "${CURL_LIBRARY_RELEASE}")
+ if(NOT OSS_SDK_V2_CURL_LIBRARY)
+ set(OSS_SDK_V2_CURL_LIBRARY "${CURL_LIBRARY}")
+ endif()
+ if(TARGET CURL::libcurl)
+ if(NOT OSS_SDK_V2_CURL_INCLUDE_DIR)
+ get_target_property(OSS_SDK_V2_CURL_INCLUDE_DIR CURL::libcurl
+ INTERFACE_INCLUDE_DIRECTORIES)
+ endif()
+ if(NOT OSS_SDK_V2_CURL_LIBRARY)
+ foreach(CURL_CONFIG RELEASE RELWITHDEBINFO DEBUG NOCONFIG)
+ get_target_property(OSS_SDK_V2_CURL_LIBRARY CURL::libcurl
+ "IMPORTED_LOCATION_${CURL_CONFIG}")
+ if(OSS_SDK_V2_CURL_LIBRARY)
+ break()
+ endif()
+ endforeach()
+ endif()
+ if(NOT OSS_SDK_V2_CURL_LIBRARY)
+ get_target_property(OSS_SDK_V2_CURL_LIBRARY CURL::libcurl
IMPORTED_LOCATION)
+ endif()
+ endif()
+ if(OSS_SDK_V2_CURL_INCLUDE_DIR AND OSS_SDK_V2_CURL_LIBRARY)
+ list(APPEND
+ OSS_SDK_V2_CMAKE_ARGS
+ "-DCURL_INCLUDE_DIR=${OSS_SDK_V2_CURL_INCLUDE_DIR}"
+ "-DCURL_LIBRARY=${OSS_SDK_V2_CURL_LIBRARY}"
+ "-DCURL_LIBRARY_RELEASE=${OSS_SDK_V2_CURL_LIBRARY}")
+ endif()
+
+ externalproject_add(oss_sdk_v2_ep
+ ${EP_COMMON_OPTIONS}
+ URL ${OSS_SDK_V2_SOURCE_URL}
+ URL_HASH
"SHA256=${PAIMON_OSS_SDK_V2_BUILD_SHA256_CHECKSUM}"
+ CMAKE_ARGS ${OSS_SDK_V2_CMAKE_ARGS}
+ PATCH_COMMAND bash -c
+ "perl -0pi -e
's/\"-Wall\"\\s+\"-Werror\"\\s+\"-pedantic\"\\s+\"-Wextra\"/\"-Wall\"
\"-pedantic\" \"-Wextra\"/g' <SOURCE_DIR>/sdk/CMakeLists.txt"
+ COMMAND bash -c
+ "perl -0pi -e 's/#if LIBCURL_VERSION_NUM >=
0x074400/#if 0/g' <SOURCE_DIR>/sdk/src/transport/curl/CurlMultiTransport.cpp"
+ COMMAND bash -c
+ "perl -0pi -e 's/#if LIBCURL_VERSION_NUM >=
0x074200/#if 0/g' <SOURCE_DIR>/sdk/src/transport/curl/CurlMultiTransport.cpp"
+ COMMAND bash -c
+ "perl -0pi -e
's/request->headers\\.emplace\\(\"Authorization\",
credentialHeader\\)\\x3b/request->headers.insert_or_assign(\"Authorization\",
credentialHeader)\\x3b/g' <SOURCE_DIR>/sdk/src/signer/SignerV1.cpp"
+ COMMAND bash -c
+ "perl -0pi -e
's/request->headers\\.emplace\\(\"Authorization\",
ss\\.str\\(\\)\\)\\x3b/request->headers.insert_or_assign(\"Authorization\",
ss.str())\\x3b/g' <SOURCE_DIR>/sdk/src/signer/SignerV4.cpp"
+ ${THIRDPARTY_LOG_OPTIONS}
Review Comment:
Done.
##########
src/paimon/fs/oss/oss_file_system.cpp:
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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);
Review Comment:
Done.
##########
src/paimon/fs/oss/oss_file_system_factory.cpp:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <utility>
+
+#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/common/utils/string_utils.h"
+#include "paimon/factories/factory.h"
+#include "paimon/fs/oss/oss_file_system.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+constexpr std::string_view kOssOptionPrefix = "fs.oss.";
+
+std::string GetBucketOptionKey(const std::string& bucket, std::string_view
option) {
+ return fmt::format("fs.oss.bucket.{}.{}", bucket,
option.substr(kOssOptionPrefix.size()));
+}
+
+const std::string* FindOption(const std::map<std::string, std::string>&
options,
+ const std::string& bucket, std::string_view
option,
+ std::string* option_key) {
+ std::string bucket_option_key = GetBucketOptionKey(bucket, option);
+ auto bucket_option = options.find(bucket_option_key);
+ if (bucket_option != options.end()) {
+ *option_key = std::move(bucket_option_key);
+ return &bucket_option->second;
+ }
+ option_key->assign(option);
+ auto global_option = options.find(*option_key);
+ return global_option == options.end() ? nullptr : &global_option->second;
+}
+
+std::string GetOption(const std::map<std::string, std::string>& options, const
std::string& bucket,
+ std::string_view option) {
+ std::string option_key;
+ const std::string* value = FindOption(options, bucket, option,
&option_key);
+ return value == nullptr ? "" : *value;
+}
+
+Result<std::string> GetRequiredOption(const std::map<std::string,
std::string>& options,
+ const std::string& bucket,
std::string_view option) {
+ std::string option_key;
+ const std::string* value = FindOption(options, bucket, option,
&option_key);
+ if (value == nullptr || value->empty()) {
+ return Status::Invalid(fmt::format("OSS option '{}' must not be
empty", option_key));
+ }
+ return *value;
+}
+
+Result<std::unique_ptr<Executor>> CreateExecutor(const std::map<std::string,
std::string>& options,
+ const std::string& bucket) {
+ std::string option_key;
+ const std::string* value =
+ FindOption(options, bucket, kOssExecutorThreadCountOption,
&option_key);
+ if (value == nullptr) {
+ return CreateDefaultExecutor();
+ }
+ std::optional<uint32_t> thread_count =
StringUtils::StringToValue<uint32_t>(*value);
+ if (!thread_count.has_value() || *thread_count == 0) {
+ return Status::Invalid(fmt::format(
+ "OSS executor thread count for option '{}' must be greater than
0", option_key));
+ }
+ return CreateDefaultExecutor(*thread_count);
+}
+
+std::string NormalizeEndpoint(std::string endpoint) {
+ if (!endpoint.empty() && endpoint.find("://") == std::string::npos) {
+ endpoint = "https://"; + endpoint;
+ }
+ return endpoint;
+}
+
+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 || suffix + kSuffix.size() !=
endpoint.size()) {
+ 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());
+ }
+ if (region == "accelerate" ||
Review Comment:
Done.
##########
src/paimon/fs/oss/oss_file_system_factory.cpp:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <utility>
+
+#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/common/utils/string_utils.h"
+#include "paimon/factories/factory.h"
+#include "paimon/fs/oss/oss_file_system.h"
+
+namespace paimon::oss {
+namespace {
+
+namespace oss2 = alibabacloud::oss2;
+
+constexpr std::string_view kOssOptionPrefix = "fs.oss.";
+
+std::string GetBucketOptionKey(const std::string& bucket, std::string_view
option) {
+ return fmt::format("fs.oss.bucket.{}.{}", bucket,
option.substr(kOssOptionPrefix.size()));
+}
+
+const std::string* FindOption(const std::map<std::string, std::string>&
options,
+ const std::string& bucket, std::string_view
option,
+ std::string* option_key) {
+ std::string bucket_option_key = GetBucketOptionKey(bucket, option);
+ auto bucket_option = options.find(bucket_option_key);
+ if (bucket_option != options.end()) {
+ *option_key = std::move(bucket_option_key);
+ return &bucket_option->second;
+ }
+ option_key->assign(option);
+ auto global_option = options.find(*option_key);
+ return global_option == options.end() ? nullptr : &global_option->second;
+}
+
+std::string GetOption(const std::map<std::string, std::string>& options, const
std::string& bucket,
+ std::string_view option) {
+ std::string option_key;
+ const std::string* value = FindOption(options, bucket, option,
&option_key);
+ return value == nullptr ? "" : *value;
+}
+
+Result<std::string> GetRequiredOption(const std::map<std::string,
std::string>& options,
+ const std::string& bucket,
std::string_view option) {
+ std::string option_key;
+ const std::string* value = FindOption(options, bucket, option,
&option_key);
+ if (value == nullptr || value->empty()) {
+ return Status::Invalid(fmt::format("OSS option '{}' must not be
empty", option_key));
+ }
+ return *value;
+}
+
+Result<std::unique_ptr<Executor>> CreateExecutor(const std::map<std::string,
std::string>& options,
+ const std::string& bucket) {
+ std::string option_key;
+ const std::string* value =
+ FindOption(options, bucket, kOssExecutorThreadCountOption,
&option_key);
+ if (value == nullptr) {
+ return CreateDefaultExecutor();
+ }
+ std::optional<uint32_t> thread_count =
StringUtils::StringToValue<uint32_t>(*value);
+ if (!thread_count.has_value() || *thread_count == 0) {
+ return Status::Invalid(fmt::format(
+ "OSS executor thread count for option '{}' must be greater than
0", option_key));
+ }
+ return CreateDefaultExecutor(*thread_count);
+}
+
+std::string NormalizeEndpoint(std::string endpoint) {
+ if (!endpoint.empty() && endpoint.find("://") == std::string::npos) {
+ endpoint = "https://"; + endpoint;
+ }
+ return endpoint;
+}
+
+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) {
Review Comment:
Done.
--
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]
