From 7f56b8ba1d27ae4a5ea6ae12277b800a2141dfc2 Mon Sep 17 00:00:00 2001 From: anlowee Date: Thu, 31 Jul 2025 18:31:21 +0000 Subject: [PATCH 01/15] WIP --- velox/connectors/clp/CMakeLists.txt | 1 - velox/connectors/clp/ClpConfig.cpp | 30 +++++++++++ velox/connectors/clp/ClpConfig.h | 12 +++-- .../connectors/clp/search_lib/CMakeLists.txt | 3 ++ .../search_lib/ClpPackageS3AuthProvider.cpp | 42 ++++++++++++++++ .../clp/search_lib/ClpPackageS3AuthProvider.h | 32 ++++++++++++ .../clp/search_lib/ClpS3AuthProviderBase.cpp | 49 ++++++++++++++++++ .../clp/search_lib/ClpS3AuthProviderBase.h | 50 +++++++++++++++++++ 8 files changed, 214 insertions(+), 5 deletions(-) create mode 100644 velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp create mode 100644 velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h create mode 100644 velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp create mode 100644 velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h diff --git a/velox/connectors/clp/CMakeLists.txt b/velox/connectors/clp/CMakeLists.txt index 5cf5366aa9e..6f5d8633a7f 100644 --- a/velox/connectors/clp/CMakeLists.txt +++ b/velox/connectors/clp/CMakeLists.txt @@ -11,7 +11,6 @@ # 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. -#set(CLP_SRC_DIR ${clp_SOURCE_DIR}/components/core/src) add_subdirectory(search_lib) velox_add_library( diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index adbb8f0fc47..0404c7d958d 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -17,11 +17,21 @@ #include #include "velox/connectors/clp/ClpConfig.h" +#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" namespace facebook::velox::connector::clp { namespace { +ClpConfig::S3AuthProvider stringToS3AuthProvider(const std::string& strValue) { + auto upperValue = boost::algorithm::to_upper_copy(strValue); + VELOX_CHECK(!upperValue.empty()); + if (upperValue == "CLP_PACKAGE") { + return ClpConfig::S3AuthProvider::kClpPackage; + } + VELOX_UNSUPPORTED("Unsupported s3 auth provider type: {}.", strValue); +} + ClpConfig::StorageType stringToStorageType(const std::string& strValue) { auto upperValue = boost::algorithm::to_upper_copy(strValue); if (upperValue == "FS") { @@ -35,6 +45,26 @@ ClpConfig::StorageType stringToStorageType(const std::string& strValue) { } // namespace +ClpConfig::ClpConfig(std::shared_ptr config) { + VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); + + // Setup S3 environment variables needed by CLP by user-specific ways + switch (s3AuthProvider()) { + case ClpConfig::S3AuthProvider::kClpPackage: + VELOX_CHECK(std::make_unique()->exportAuthEnvironmentVariables()); + break; + default: + VELOX_FAIL(); + } + + config_ = std::move(config); +} + + +ClpConfig::S3AuthProvider ClpConfig::s3AuthProvider() const { + return stringToS3AuthProvider(config_->get(kAuthProvider, "")); +} + ClpConfig::StorageType ClpConfig::storageType() const { return stringToStorageType(config_->get(kStorageType, "FS")); } diff --git a/velox/connectors/clp/ClpConfig.h b/velox/connectors/clp/ClpConfig.h index ce4808eef62..531ef8845da 100644 --- a/velox/connectors/clp/ClpConfig.h +++ b/velox/connectors/clp/ClpConfig.h @@ -26,17 +26,19 @@ namespace facebook::velox::connector::clp { class ClpConfig { public: + enum class S3AuthProvider { + kClpPackage, + }; + enum class StorageType { kFs, kS3, }; + static constexpr const char* kAuthProvider = "clp.s3-auth-provider"; static constexpr const char* kStorageType = "clp.storage-type"; - explicit ClpConfig(std::shared_ptr config) { - VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); - config_ = std::move(config); - } + explicit ClpConfig(std::shared_ptr config); [[nodiscard]] const std::shared_ptr& config() const { @@ -47,6 +49,8 @@ class ClpConfig { private: std::shared_ptr config_; + + S3AuthProvider s3AuthProvider() const; }; } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index 55e68060fc4..d85a93eacc5 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -16,8 +16,11 @@ velox_add_library( STATIC ClpCursor.cpp ClpCursor.h + ClpPackageS3AuthProvider.cpp ClpQueryRunner.cpp ClpQueryRunner.h + ClpS3AuthProviderBase.cpp + ClpS3AuthProviderBase.h ClpVectorLoader.cpp ClpVectorLoader.h) diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp new file mode 100644 index 00000000000..a4f371a62cc --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -0,0 +1,42 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * 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 +#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" + +namespace facebook::velox::connector::clp { + +bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables( + std::shared_ptr config) const { + auto accessKeyId = config->get(kAccessKeyId, ""); + auto secretAccessKey = config->get(kSecretAccessKey, ""); + auto sessionToken = config->get(kSessionToken, ""); + VELOX_CHECK(!accessKeyId.empty()); + VELOX_CHECK(!secretAccessKey.empty()); + LOG(INFO) << "Setting AWS_ENV_VAR_ACCESS_KEY_ID environment variable: " << accessKeyId; + setupEnvironmentVariables("AWS_ENV_VAR_ACCESS_KEY_ID", accessKeyId); + LOG(INFO) << "Setting AWS_ENV_VAR_SECRET_ACCESS_KEY environment variable: " << secretAccessKey; + setupEnvironmentVariables("AWS_ENV_VAR_SECRET_ACCESS_KEY", secretAccessKey); + if (!sessionToken.empty()) { + LOG(INFO) << "Setting AWS_ENV_VAR_SESSION_TOKEN environment variable: " << sessionToken; + setupEnvironmentVariables("AWS_ENV_VAR_SESSION_TOKEN", sessionToken); + } + + return true; +} + + +} // namespace facebook::velox::connector::clp \ No newline at end of file diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h new file mode 100644 index 00000000000..139de2e9271 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h @@ -0,0 +1,32 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * 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 "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" + +namespace facebook::velox::connector::clp { + +class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { +public: + static constexpr const char* kAccessKeyId = "clp.s3-access-key-id"; + static constexpr const char* kSecretAccessKey = "clp.s3-secret-access-key"; + static constexpr const char* kSessionToken = "clp.s3-session-token"; + + bool exportAuthEnvironmentVariables(std::shared_ptr config) const override; +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp new file mode 100644 index 00000000000..5863fda62d3 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -0,0 +1,49 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * 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 +#include + +#include "velox/common/base/Exceptions.h" +#include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" + +namespace facebook::velox::config { +class ConfigBase; +}; + +namespace facebook::velox::connector::clp { + +void ClpS3AuthProviderBase::setupEnvironmentVariables(std::string_view key, std::string_view value) { + int err{0}; +#ifdef _WIN32 + // Windows version + err = _putenv_s(std::string(key).c_str(), std::string(value).c_str()); +#elif defined(__unix__) || defined(__APPLE__) + // Unix/macOS version + err = setenv(std::string(key).c_str(), std::string(value).c_str(), 1); +#else + VELOX_UNSUPPORTED("Unsuppported OS"); +#endif + VELOX_CHECK_EQ(0, err); + + // Sanity check + auto valueForCheck = std::getenv(std::string(key).c_str()); + VELOX_CHECK_EQ(0, std::strcmp(std::string(value).c_str(), valueForCheck)); +} + +} // namespace facebook::velox::connector::clp + +} // namespace facebook::velox::config diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h new file mode 100644 index 00000000000..f6a58d190ae --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -0,0 +1,50 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * 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 + +#include "velox/common/config/Config.h" + +namespace facebook::velox::connector::clp { + +class ClpS3AuthProviderBase { + public: + virtual ~ClpS3AuthProviderBase() = default; + + /// Export the three environment variables needed by CLP-s to system: + /// AWS_ENV_VAR_ACCESS_KEY_ID + /// AWS_ENV_VAR_SECRET_ACCESS_KEY + /// AWS_ENV_VAR_SESSION_TOKEN (optional) + /// So that in runtime, CLP-s’s code can execute S3-related logic correctly. + /// + /// @param config The config options parsed from clp.properties. User can + /// define customized config options and use then here. + /// @return Did exportation succeed or not. + virtual bool exportAuthEnvironmentVariables(std::shared_ptr config) const = 0; + + protected: + /// Set the environment variables for different OS, then get the environment + /// variables to do the sanity check. + /// + /// @param key The environment variable name. + /// @param value The environment variable value. + /// @return Did exportation succeed or not. + static void setupEnvironmentVariables(std::string_view key, std::string_view value); +}; + +} // namespace facebook::velox::connector::clp \ No newline at end of file From e1f1933466469ed83f16312b1d40aea929bf0496 Mon Sep 17 00:00:00 2001 From: anlowee Date: Thu, 31 Jul 2025 21:20:05 +0000 Subject: [PATCH 02/15] Init --- velox/connectors/clp/ClpConfig.cpp | 14 ++++---- velox/connectors/clp/ClpConfig.h | 10 +++--- velox/connectors/clp/ClpDataSource.cpp | 5 ++- velox/connectors/clp/ClpDataSource.h | 1 + .../connectors/clp/search_lib/CMakeLists.txt | 1 + .../search_lib/ClpPackageS3AuthProvider.cpp | 34 +++++++++++++------ .../clp/search_lib/ClpPackageS3AuthProvider.h | 14 +++++++- .../clp/search_lib/ClpS3AuthProviderBase.cpp | 4 +-- .../clp/search_lib/ClpS3AuthProviderBase.h | 18 +++++++--- 9 files changed, 70 insertions(+), 31 deletions(-) diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index 0404c7d958d..6fe5b61717f 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -16,6 +16,8 @@ #include +#include "velox/common/base/Exceptions.h" +#include "velox/common/config/Config.h" #include "velox/connectors/clp/ClpConfig.h" #include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" @@ -47,22 +49,22 @@ ClpConfig::StorageType stringToStorageType(const std::string& strValue) { ClpConfig::ClpConfig(std::shared_ptr config) { VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); + config_ = std::move(config); // Setup S3 environment variables needed by CLP by user-specific ways - switch (s3AuthProvider()) { + switch (stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { case ClpConfig::S3AuthProvider::kClpPackage: - VELOX_CHECK(std::make_unique()->exportAuthEnvironmentVariables()); + s3AuthProvider_ = std::make_shared(config_); break; default: VELOX_FAIL(); } - - config_ = std::move(config); + VELOX_CHECK(s3AuthProvider_->exportAuthEnvironmentVariables()); } -ClpConfig::S3AuthProvider ClpConfig::s3AuthProvider() const { - return stringToS3AuthProvider(config_->get(kAuthProvider, "")); +std::shared_ptr ClpConfig::s3AuthProvider() const { + return s3AuthProvider_; } ClpConfig::StorageType ClpConfig::storageType() const { diff --git a/velox/connectors/clp/ClpConfig.h b/velox/connectors/clp/ClpConfig.h index 531ef8845da..c4a9a020690 100644 --- a/velox/connectors/clp/ClpConfig.h +++ b/velox/connectors/clp/ClpConfig.h @@ -16,14 +16,14 @@ #pragma once -#include "velox/common/config/Config.h" - namespace facebook::velox::config { class ConfigBase; -} +} // facebook::velox::config namespace facebook::velox::connector::clp { +class ClpS3AuthProviderBase; + class ClpConfig { public: enum class S3AuthProvider { @@ -46,11 +46,11 @@ class ClpConfig { } StorageType storageType() const; + std::shared_ptr s3AuthProvider() const; private: std::shared_ptr config_; - - S3AuthProvider s3AuthProvider() const; + std::shared_ptr s3AuthProvider_; }; } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index a5a574eb318..40aeb8ab570 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -19,6 +19,8 @@ #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" + +#include "search_lib/ClpS3AuthProviderBase.h" #include "velox/connectors/clp/ClpTableHandle.h" #include "velox/connectors/clp/search_lib/ClpCursor.h" #include "velox/connectors/clp/search_lib/ClpVectorLoader.h" @@ -37,6 +39,7 @@ ClpDataSource::ClpDataSource( : pool_(pool), outputType_(outputType) { auto clpTableHandle = std::dynamic_pointer_cast(tableHandle); storageType_ = clpConfig->storageType(); + s3AuthProvider_ = clpConfig->s3AuthProvider(); for (const auto& outputName : outputType->names()) { auto columnHandle = columnHandles.find(outputName); @@ -106,7 +109,7 @@ void ClpDataSource::addSplit(std::shared_ptr split) { clp_s::InputSource::Filesystem, clpSplit->path_); } else if (storageType_ == ClpConfig::StorageType::kS3) { cursor_ = std::make_unique( - clp_s::InputSource::Network, clpSplit->path_); + clp_s::InputSource::Network, s3AuthProvider_->constructS3Url(clpSplit->path_)); } auto pushDownQuery = clpSplit->kqlQuery_; diff --git a/velox/connectors/clp/ClpDataSource.h b/velox/connectors/clp/ClpDataSource.h index 441386199bf..8a3fe3d6e3f 100644 --- a/velox/connectors/clp/ClpDataSource.h +++ b/velox/connectors/clp/ClpDataSource.h @@ -101,6 +101,7 @@ class ClpDataSource : public DataSource { std::vector fields_; std::unique_ptr cursor_; + std::shared_ptr s3AuthProvider_; }; } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index d85a93eacc5..8cd5a414d7a 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -17,6 +17,7 @@ velox_add_library( ClpCursor.cpp ClpCursor.h ClpPackageS3AuthProvider.cpp + ClpPackageS3AuthProvider.h ClpQueryRunner.cpp ClpQueryRunner.h ClpS3AuthProviderBase.cpp diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp index a4f371a62cc..4e3df2fe20d 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -15,24 +15,36 @@ */ #include + +#include "velox/common/base/Exceptions.h" +#include "velox/common/config/Config.h" #include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" namespace facebook::velox::connector::clp { -bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables( - std::shared_ptr config) const { - auto accessKeyId = config->get(kAccessKeyId, ""); - auto secretAccessKey = config->get(kSecretAccessKey, ""); - auto sessionToken = config->get(kSessionToken, ""); +const std::string ClpPackageS3AuthProvider::constructS3Url(std::string_view splitPath) { + if (this->endPoint_.empty()) { + this->endPoint_ = config_->get(kEndPoint, ""); + } + if ('/' == this->endPoint_.back()) { + this->endPoint_.pop_back(); + } + return fmt::format("{}/{}", this->endPoint_, splitPath); +} + +bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() const { + auto accessKeyId = config_->get(kAccessKeyId, ""); + auto secretAccessKey = config_->get(kSecretAccessKey, ""); + auto sessionToken = config_->get(kSessionToken, ""); VELOX_CHECK(!accessKeyId.empty()); VELOX_CHECK(!secretAccessKey.empty()); - LOG(INFO) << "Setting AWS_ENV_VAR_ACCESS_KEY_ID environment variable: " << accessKeyId; - setupEnvironmentVariables("AWS_ENV_VAR_ACCESS_KEY_ID", accessKeyId); - LOG(INFO) << "Setting AWS_ENV_VAR_SECRET_ACCESS_KEY environment variable: " << secretAccessKey; - setupEnvironmentVariables("AWS_ENV_VAR_SECRET_ACCESS_KEY", secretAccessKey); + LOG(INFO) << "Setting AWS_ACCESS_KEY_ID environment variable: " << accessKeyId; + setupEnvironmentVariables("AWS_ACCESS_KEY_ID", accessKeyId); + LOG(INFO) << "Setting AWS_SECRET_ACCESS_KEY environment variable: " << secretAccessKey; + setupEnvironmentVariables("AWS_SECRET_ACCESS_KEY", secretAccessKey); if (!sessionToken.empty()) { - LOG(INFO) << "Setting AWS_ENV_VAR_SESSION_TOKEN environment variable: " << sessionToken; - setupEnvironmentVariables("AWS_ENV_VAR_SESSION_TOKEN", sessionToken); + LOG(INFO) << "Setting AWS_SESSION_TOKEN environment variable: " << sessionToken; + setupEnvironmentVariables("AWS_SESSION_TOKEN", sessionToken); } return true; diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h index 139de2e9271..39002795034 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h @@ -18,15 +18,27 @@ #include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" +namespace facebook::velox::config { +class ConfigBase; +} // facebook::velox::config + namespace facebook::velox::connector::clp { class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { public: + explicit ClpPackageS3AuthProvider(std::shared_ptr config) : ClpS3AuthProviderBase(config) {} + static constexpr const char* kAccessKeyId = "clp.s3-access-key-id"; + static constexpr const char* kEndPoint = "clp.s3-end-point"; static constexpr const char* kSecretAccessKey = "clp.s3-secret-access-key"; static constexpr const char* kSessionToken = "clp.s3-session-token"; - bool exportAuthEnvironmentVariables(std::shared_ptr config) const override; + const std::string constructS3Url(std::string_view splitPath) override; + + bool exportAuthEnvironmentVariables() const override; + +private: + std::string endPoint_; }; } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp index 5863fda62d3..602f4b39b19 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -22,7 +22,7 @@ namespace facebook::velox::config { class ConfigBase; -}; +} // namespace facebook::velox::config namespace facebook::velox::connector::clp { @@ -45,5 +45,3 @@ void ClpS3AuthProviderBase::setupEnvironmentVariables(std::string_view key, std: } } // namespace facebook::velox::connector::clp - -} // namespace facebook::velox::config diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h index f6a58d190ae..e91c1f3c414 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -16,26 +16,34 @@ #pragma once +#include #include -#include "velox/common/config/Config.h" +namespace facebook::velox::config { +class ConfigBase; +} // facebook::velox::config namespace facebook::velox::connector::clp { class ClpS3AuthProviderBase { public: + explicit ClpS3AuthProviderBase(std::shared_ptr config) : config_(config) {} virtual ~ClpS3AuthProviderBase() = default; + /// Construct the actual S3 URL so that CLP-s can access the split. + /// + /// @param splitPath The current path stored in the split to add. + /// @return The constructed S3 URL. + virtual const std::string constructS3Url(std::string_view splitPath) = 0; + /// Export the three environment variables needed by CLP-s to system: /// AWS_ENV_VAR_ACCESS_KEY_ID /// AWS_ENV_VAR_SECRET_ACCESS_KEY /// AWS_ENV_VAR_SESSION_TOKEN (optional) /// So that in runtime, CLP-s’s code can execute S3-related logic correctly. /// - /// @param config The config options parsed from clp.properties. User can - /// define customized config options and use then here. /// @return Did exportation succeed or not. - virtual bool exportAuthEnvironmentVariables(std::shared_ptr config) const = 0; + virtual bool exportAuthEnvironmentVariables() const = 0; protected: /// Set the environment variables for different OS, then get the environment @@ -45,6 +53,8 @@ class ClpS3AuthProviderBase { /// @param value The environment variable value. /// @return Did exportation succeed or not. static void setupEnvironmentVariables(std::string_view key, std::string_view value); + + std::shared_ptr config_; }; } // namespace facebook::velox::connector::clp \ No newline at end of file From 0e1aded01ae86fd84fe97f4a14b1774bd921c441 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 1 Aug 2025 13:53:29 +0000 Subject: [PATCH 03/15] Format --- CMake/resolve_dependency_modules/clp.cmake | 2 +- velox/connectors/clp/ClpConfig.cpp | 4 ++-- velox/connectors/clp/ClpConfig.h | 2 +- .../clp/search_lib/ClpPackageS3AuthProvider.cpp | 15 +++++++++------ .../clp/search_lib/ClpPackageS3AuthProvider.h | 10 ++++++---- .../clp/search_lib/ClpS3AuthProviderBase.cpp | 4 +++- .../clp/search_lib/ClpS3AuthProviderBase.h | 12 ++++++++---- 7 files changed, 30 insertions(+), 19 deletions(-) diff --git a/CMake/resolve_dependency_modules/clp.cmake b/CMake/resolve_dependency_modules/clp.cmake index 840db9f26e5..5cb4c731de2 100644 --- a/CMake/resolve_dependency_modules/clp.cmake +++ b/CMake/resolve_dependency_modules/clp.cmake @@ -16,7 +16,7 @@ include_guard(GLOBAL) FetchContent_Declare( clp GIT_REPOSITORY https://github.com/y-scope/clp.git - GIT_TAG 0bfaf5caec80d17527a2e91ba21fdede88b44f19) + GIT_TAG 581bd46198a97a89b174849851cc3f1f2100e466) set(CLP_BUILD_CLP_REGEX_UTILS OFF diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index 6fe5b61717f..c9a4bb54fad 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -52,7 +52,8 @@ ClpConfig::ClpConfig(std::shared_ptr config) { config_ = std::move(config); // Setup S3 environment variables needed by CLP by user-specific ways - switch (stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { + switch ( + stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { case ClpConfig::S3AuthProvider::kClpPackage: s3AuthProvider_ = std::make_shared(config_); break; @@ -62,7 +63,6 @@ ClpConfig::ClpConfig(std::shared_ptr config) { VELOX_CHECK(s3AuthProvider_->exportAuthEnvironmentVariables()); } - std::shared_ptr ClpConfig::s3AuthProvider() const { return s3AuthProvider_; } diff --git a/velox/connectors/clp/ClpConfig.h b/velox/connectors/clp/ClpConfig.h index c4a9a020690..f9e7c845be6 100644 --- a/velox/connectors/clp/ClpConfig.h +++ b/velox/connectors/clp/ClpConfig.h @@ -18,7 +18,7 @@ namespace facebook::velox::config { class ConfigBase; -} // facebook::velox::config +} // namespace facebook::velox::config namespace facebook::velox::connector::clp { diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp index 4e3df2fe20d..5472610ab33 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -22,7 +22,8 @@ namespace facebook::velox::connector::clp { -const std::string ClpPackageS3AuthProvider::constructS3Url(std::string_view splitPath) { +const std::string ClpPackageS3AuthProvider::constructS3Url( + std::string_view splitPath) { if (this->endPoint_.empty()) { this->endPoint_ = config_->get(kEndPoint, ""); } @@ -38,17 +39,19 @@ bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() const { auto sessionToken = config_->get(kSessionToken, ""); VELOX_CHECK(!accessKeyId.empty()); VELOX_CHECK(!secretAccessKey.empty()); - LOG(INFO) << "Setting AWS_ACCESS_KEY_ID environment variable: " << accessKeyId; + LOG(INFO) << "Setting AWS_ACCESS_KEY_ID environment variable: " + << accessKeyId; setupEnvironmentVariables("AWS_ACCESS_KEY_ID", accessKeyId); - LOG(INFO) << "Setting AWS_SECRET_ACCESS_KEY environment variable: " << secretAccessKey; + LOG(INFO) << "Setting AWS_SECRET_ACCESS_KEY environment variable: " + << secretAccessKey; setupEnvironmentVariables("AWS_SECRET_ACCESS_KEY", secretAccessKey); if (!sessionToken.empty()) { - LOG(INFO) << "Setting AWS_SESSION_TOKEN environment variable: " << sessionToken; + LOG(INFO) << "Setting AWS_SESSION_TOKEN environment variable: " + << sessionToken; setupEnvironmentVariables("AWS_SESSION_TOKEN", sessionToken); } return true; } - -} // namespace facebook::velox::connector::clp \ No newline at end of file +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h index 39002795034..9e8dfa72a6c 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h @@ -20,13 +20,15 @@ namespace facebook::velox::config { class ConfigBase; -} // facebook::velox::config +} // namespace facebook::velox::config namespace facebook::velox::connector::clp { class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { -public: - explicit ClpPackageS3AuthProvider(std::shared_ptr config) : ClpS3AuthProviderBase(config) {} + public: + explicit ClpPackageS3AuthProvider( + std::shared_ptr config) + : ClpS3AuthProviderBase(config) {} static constexpr const char* kAccessKeyId = "clp.s3-access-key-id"; static constexpr const char* kEndPoint = "clp.s3-end-point"; @@ -37,7 +39,7 @@ class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { bool exportAuthEnvironmentVariables() const override; -private: + private: std::string endPoint_; }; diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp index 602f4b39b19..1feb49eeeee 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -26,7 +26,9 @@ class ConfigBase; namespace facebook::velox::connector::clp { -void ClpS3AuthProviderBase::setupEnvironmentVariables(std::string_view key, std::string_view value) { +void ClpS3AuthProviderBase::setupEnvironmentVariables( + std::string_view key, + std::string_view value) { int err{0}; #ifdef _WIN32 // Windows version diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h index e91c1f3c414..176a318a0f4 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -21,13 +21,15 @@ namespace facebook::velox::config { class ConfigBase; -} // facebook::velox::config +} // namespace facebook::velox::config namespace facebook::velox::connector::clp { class ClpS3AuthProviderBase { public: - explicit ClpS3AuthProviderBase(std::shared_ptr config) : config_(config) {} + explicit ClpS3AuthProviderBase( + std::shared_ptr config) + : config_(config) {} virtual ~ClpS3AuthProviderBase() = default; /// Construct the actual S3 URL so that CLP-s can access the split. @@ -52,9 +54,11 @@ class ClpS3AuthProviderBase { /// @param key The environment variable name. /// @param value The environment variable value. /// @return Did exportation succeed or not. - static void setupEnvironmentVariables(std::string_view key, std::string_view value); + static void setupEnvironmentVariables( + std::string_view key, + std::string_view value); std::shared_ptr config_; }; -} // namespace facebook::velox::connector::clp \ No newline at end of file +} // namespace facebook::velox::connector::clp From 07b1488c708ed81041b7ceadc192d90343785b99 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 1 Aug 2025 17:24:14 +0000 Subject: [PATCH 04/15] Add config parse unit test --- .../search_lib/ClpPackageS3AuthProvider.cpp | 22 ++-- .../clp/search_lib/ClpPackageS3AuthProvider.h | 4 + .../clp/search_lib/ClpS3AuthProviderBase.cpp | 20 ++- .../clp/search_lib/ClpS3AuthProviderBase.h | 16 ++- velox/connectors/clp/tests/CMakeLists.txt | 2 +- velox/connectors/clp/tests/ClpConfigTest.cpp | 121 ++++++++++++++++++ 6 files changed, 169 insertions(+), 16 deletions(-) create mode 100644 velox/connectors/clp/tests/ClpConfigTest.cpp diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp index 5472610ab33..e19fdeb830b 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -39,16 +39,20 @@ bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() const { auto sessionToken = config_->get(kSessionToken, ""); VELOX_CHECK(!accessKeyId.empty()); VELOX_CHECK(!secretAccessKey.empty()); - LOG(INFO) << "Setting AWS_ACCESS_KEY_ID environment variable: " - << accessKeyId; - setupEnvironmentVariables("AWS_ACCESS_KEY_ID", accessKeyId); - LOG(INFO) << "Setting AWS_SECRET_ACCESS_KEY environment variable: " - << secretAccessKey; - setupEnvironmentVariables("AWS_SECRET_ACCESS_KEY", secretAccessKey); + LOG(INFO) << "Setting " << kEnvAwsAccessKeyId + << " environment variable: " << accessKeyId; + setupEnvironmentVariable(kEnvAwsAccessKeyId, accessKeyId); + LOG(INFO) << "Setting " << kEnvAwsSecretAccessKey + << " environment variable: " << secretAccessKey; + setupEnvironmentVariable(kEnvAwsSecretAccessKey, secretAccessKey); if (!sessionToken.empty()) { - LOG(INFO) << "Setting AWS_SESSION_TOKEN environment variable: " - << sessionToken; - setupEnvironmentVariables("AWS_SESSION_TOKEN", sessionToken); + LOG(INFO) << "Setting " << kEnvAwsSessionToken + << " environment variable: " << sessionToken; + setupEnvironmentVariable(kEnvAwsSessionToken, sessionToken); + } else { + LOG(INFO) << "Unsetting " << kEnvAwsSessionToken + << " environment variable."; + unsetEnvironmentVariable(kEnvAwsSessionToken); } return true; diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h index 9e8dfa72a6c..4a0b5682917 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h @@ -35,6 +35,10 @@ class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { static constexpr const char* kSecretAccessKey = "clp.s3-secret-access-key"; static constexpr const char* kSessionToken = "clp.s3-session-token"; + static constexpr const char* kEnvAwsAccessKeyId = "AWS_ACCESS_KEY_ID"; + static constexpr const char* kEnvAwsSecretAccessKey = "AWS_SECRET_ACCESS_KEY"; + static constexpr const char* kEnvAwsSessionToken = "AWS_SESSION_TOKEN"; + const std::string constructS3Url(std::string_view splitPath) override; bool exportAuthEnvironmentVariables() const override; diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp index 1feb49eeeee..43928e4691d 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -26,7 +26,7 @@ class ConfigBase; namespace facebook::velox::connector::clp { -void ClpS3AuthProviderBase::setupEnvironmentVariables( +void ClpS3AuthProviderBase::setupEnvironmentVariable( std::string_view key, std::string_view value) { int err{0}; @@ -46,4 +46,22 @@ void ClpS3AuthProviderBase::setupEnvironmentVariables( VELOX_CHECK_EQ(0, std::strcmp(std::string(value).c_str(), valueForCheck)); } +void ClpS3AuthProviderBase::unsetEnvironmentVariable(std::string_view key) { + int err{0}; +#ifdef _WIN32 + // Windows version + err = _putenv(fmt::format("{}=", std::string(key).c_str())); +#elif defined(__unix__) || defined(__APPLE__) + // Unix/macOS version + err = unsetenv(std::string(key).c_str()); +#else + VELOX_UNSUPPORTED("Unsuppported OS"); +#endif + VELOX_CHECK_EQ(0, err); + + // Sanity check + auto valueForCheck = std::getenv(std::string(key).c_str()); + VELOX_CHECK_NULL(valueForCheck); +} + } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h index 176a318a0f4..434bbf933c8 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -48,16 +48,22 @@ class ClpS3AuthProviderBase { virtual bool exportAuthEnvironmentVariables() const = 0; protected: - /// Set the environment variables for different OS, then get the environment - /// variables to do the sanity check. + /// Set the environment variable for different OS, then get the environment + /// variable to do the sanity check. /// - /// @param key The environment variable name. - /// @param value The environment variable value. + /// @param key The environment variable name to set. + /// @param value The environment variable value to set. /// @return Did exportation succeed or not. - static void setupEnvironmentVariables( + static void setupEnvironmentVariable( std::string_view key, std::string_view value); + /// Unset the environment variable for different OS, then get the environment + /// variable to do the sanity check. + /// + /// @param key The environment variable name to unset. + static void unsetEnvironmentVariable(std::string_view key); + std::shared_ptr config_; }; diff --git a/velox/connectors/clp/tests/CMakeLists.txt b/velox/connectors/clp/tests/CMakeLists.txt index 4e53803374b..5f7fd1f57c8 100644 --- a/velox/connectors/clp/tests/CMakeLists.txt +++ b/velox/connectors/clp/tests/CMakeLists.txt @@ -11,7 +11,7 @@ # 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. -add_executable(velox_clp_connector_test ClpConnectorTest.cpp) +add_executable(velox_clp_connector_test ClpConfigTest.cpp ClpConnectorTest.cpp) add_test(velox_clp_connector_test velox_clp_connector_test) diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp new file mode 100644 index 00000000000..482e65b8e7d --- /dev/null +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -0,0 +1,121 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * 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 +#include + +#include "_deps/clp-src/components/core/src/clp/time_types.hpp" +#include "gtest/gtest.h" +#include "velox/common/base/tests/GTestUtils.h" +#include "velox/common/config/Config.h" +#include "velox/connectors/clp/ClpConfig.h" +#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" + +namespace facebook::velox::connector::clp { + +namespace { + +class ClpConfigTest : public testing::Test { + public: + std::unique_ptr buildClpConfig( + std::unordered_map configMap) { + auto config = + std::make_shared(std::move(configMap)); + return std::make_unique(config); + } +}; + +class ClpS3AuthProviderBaseTest : public ClpConfigTest { + public: + bool checkEnvironmentVariableEquals( + std::string_view key, + std::string_view value) { + auto* actualValue = std::getenv(std::string(key).c_str()); + return 0 == std::strcmp(std::string(value).c_str(), actualValue); + } + + bool checkEnvironmentVariableExists(std::string_view key) { + auto* value = std::getenv(std::string(key).c_str()); + return value != nullptr; + } +}; + +class ClpPackageS3AuthProviderTest : public ClpS3AuthProviderBaseTest { + public: + std::unique_ptr buildClpPackageS3AuthProvider( + std::unordered_map configMap) { + auto config = + std::make_shared(std::move(configMap)); + return std::make_unique(config); + } +}; + +} // namespace + +TEST_F(ClpConfigTest, invalidAuthProvider) { + const std::unordered_map configMap( + {{ClpConfig::kAuthProvider, "dummy-provider"}}); + // Both access/secret keys and iam-role cannot be specified + VELOX_ASSERT_UNSUPPORTED_THROW( + buildClpConfig(configMap), + "Unsupported s3 auth provider type: dummy-provider."); +} + +TEST_F(ClpS3AuthProviderBaseTest, caseInsensitiveAuthProvider) { + const std::unordered_map configMap( + {{ClpConfig::kAuthProvider, "ClP_PaCkAgE"}, + {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, + {ClpPackageS3AuthProvider::kSecretAccessKey, "bbbbbb"}}); + VELOX_CHECK_NOT_NULL(buildClpConfig(configMap)); +} + +TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { + const std::string cTestAccessKeyId{"aaaaaa"}; + const std::string cTestEndPoint{"http://aaaaaa"}; + const std::string cTestSecretAccessKey{"bbbbbb"}; + const std::string cTestSessionToken{"cccccc"}; + + // Test all properties + std::unordered_map configMap( + {{ClpConfig::kAuthProvider, "clp_package"}, + {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId}, + {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, + {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}, + {ClpPackageS3AuthProvider::kSessionToken, cTestSessionToken}}); + auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); + VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); + VELOX_CHECK(checkEnvironmentVariableEquals( + ClpPackageS3AuthProvider::kEnvAwsAccessKeyId, cTestAccessKeyId)); + VELOX_CHECK(checkEnvironmentVariableEquals( + ClpPackageS3AuthProvider::kEnvAwsSecretAccessKey, cTestSecretAccessKey)); + VELOX_CHECK(checkEnvironmentVariableEquals( + ClpPackageS3AuthProvider::kEnvAwsSessionToken, cTestSessionToken)); + + // Test auth without the session token + configMap = { + {ClpConfig::kAuthProvider, "clp_package"}, + {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId}, + {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, + {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}}; + clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); + VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); + VELOX_CHECK( + false == + checkEnvironmentVariableExists( + ClpPackageS3AuthProvider::kEnvAwsSessionToken)); +} + +} // namespace facebook::velox::connector::clp From 32c31fbcfffabe2729310ba606af78366d28a011 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 1 Aug 2025 19:01:12 +0000 Subject: [PATCH 05/15] Format --- velox/connectors/clp/ClpDataSource.cpp | 3 ++- velox/connectors/clp/tests/ClpConfigTest.cpp | 1 - 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 40aeb8ab570..d3ad369c5b5 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -109,7 +109,8 @@ void ClpDataSource::addSplit(std::shared_ptr split) { clp_s::InputSource::Filesystem, clpSplit->path_); } else if (storageType_ == ClpConfig::StorageType::kS3) { cursor_ = std::make_unique( - clp_s::InputSource::Network, s3AuthProvider_->constructS3Url(clpSplit->path_)); + clp_s::InputSource::Network, + s3AuthProvider_->constructS3Url(clpSplit->path_)); } auto pushDownQuery = clpSplit->kqlQuery_; diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index 482e65b8e7d..8523990c75f 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -17,7 +17,6 @@ #include #include -#include "_deps/clp-src/components/core/src/clp/time_types.hpp" #include "gtest/gtest.h" #include "velox/common/base/tests/GTestUtils.h" #include "velox/common/config/Config.h" From 2a27297c144fbd2193823eb7ae7a3e52a2cdc209 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 1 Aug 2025 20:09:23 +0000 Subject: [PATCH 06/15] Address coderabbitai comments --- velox/connectors/clp/ClpConfig.cpp | 2 +- velox/connectors/clp/ClpDataSource.h | 2 + .../search_lib/ClpPackageS3AuthProvider.cpp | 35 ++++++++---------- .../clp/search_lib/ClpPackageS3AuthProvider.h | 4 +- .../clp/search_lib/ClpS3AuthProviderBase.cpp | 24 +++++++----- .../clp/search_lib/ClpS3AuthProviderBase.h | 10 +++-- velox/connectors/clp/tests/ClpConfigTest.cpp | 37 ++++++++++++------- 7 files changed, 64 insertions(+), 50 deletions(-) diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index c9a4bb54fad..7eed5548851 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -60,7 +60,7 @@ ClpConfig::ClpConfig(std::shared_ptr config) { default: VELOX_FAIL(); } - VELOX_CHECK(s3AuthProvider_->exportAuthEnvironmentVariables()); + VELOX_CHECK(s3AuthProvider_->parseConfigAndExportAuthEnvironmentVariables()); } std::shared_ptr ClpConfig::s3AuthProvider() const { diff --git a/velox/connectors/clp/ClpDataSource.h b/velox/connectors/clp/ClpDataSource.h index 8a3fe3d6e3f..70ec4564354 100644 --- a/velox/connectors/clp/ClpDataSource.h +++ b/velox/connectors/clp/ClpDataSource.h @@ -28,6 +28,8 @@ class BaseColumnReader; namespace facebook::velox::connector::clp { +class ClpS3AuthProviderBase; + class ClpDataSource : public DataSource { public: ClpDataSource( diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp index e19fdeb830b..b897998c5cb 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -14,44 +14,39 @@ * limitations under the License. */ -#include - +#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" #include "velox/common/base/Exceptions.h" #include "velox/common/config/Config.h" -#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" namespace facebook::velox::connector::clp { -const std::string ClpPackageS3AuthProvider::constructS3Url( +std::string ClpPackageS3AuthProvider::constructS3Url( std::string_view splitPath) { - if (this->endPoint_.empty()) { - this->endPoint_ = config_->get(kEndPoint, ""); - } + VELOX_CHECK(!splitPath.empty(), "splitPath cannot be empty"); + return fmt::format("{}/{}", this->endPoint_, splitPath); +} + +bool ClpPackageS3AuthProvider::parseConfigAndExportAuthEnvironmentVariables() { + this->endPoint_ = config_->get(kEndPoint, ""); + VELOX_CHECK( + !this->endPoint_.empty(), fmt::format("{} cannot be empty", kEndPoint)); if ('/' == this->endPoint_.back()) { this->endPoint_.pop_back(); } - return fmt::format("{}/{}", this->endPoint_, splitPath); -} -bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() const { auto accessKeyId = config_->get(kAccessKeyId, ""); auto secretAccessKey = config_->get(kSecretAccessKey, ""); auto sessionToken = config_->get(kSessionToken, ""); - VELOX_CHECK(!accessKeyId.empty()); - VELOX_CHECK(!secretAccessKey.empty()); - LOG(INFO) << "Setting " << kEnvAwsAccessKeyId - << " environment variable: " << accessKeyId; + VELOX_CHECK( + !accessKeyId.empty(), fmt::format("{} cannot be empty", kAccessKeyId)); + VELOX_CHECK( + !secretAccessKey.empty(), + fmt::format("{} cannot be empty", kSecretAccessKey)); setupEnvironmentVariable(kEnvAwsAccessKeyId, accessKeyId); - LOG(INFO) << "Setting " << kEnvAwsSecretAccessKey - << " environment variable: " << secretAccessKey; setupEnvironmentVariable(kEnvAwsSecretAccessKey, secretAccessKey); if (!sessionToken.empty()) { - LOG(INFO) << "Setting " << kEnvAwsSessionToken - << " environment variable: " << sessionToken; setupEnvironmentVariable(kEnvAwsSessionToken, sessionToken); } else { - LOG(INFO) << "Unsetting " << kEnvAwsSessionToken - << " environment variable."; unsetEnvironmentVariable(kEnvAwsSessionToken); } diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h index 4a0b5682917..d34c2e12466 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h @@ -39,9 +39,9 @@ class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { static constexpr const char* kEnvAwsSecretAccessKey = "AWS_SECRET_ACCESS_KEY"; static constexpr const char* kEnvAwsSessionToken = "AWS_SESSION_TOKEN"; - const std::string constructS3Url(std::string_view splitPath) override; + std::string constructS3Url(std::string_view splitPath) override; - bool exportAuthEnvironmentVariables() const override; + bool parseConfigAndExportAuthEnvironmentVariables() override; private: std::string endPoint_; diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp index 43928e4691d..77b5e82d4c9 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -30,37 +30,43 @@ void ClpS3AuthProviderBase::setupEnvironmentVariable( std::string_view key, std::string_view value) { int err{0}; + const auto keyStr = std::string(key); + const auto* keyCStr = keyStr.c_str(); + const auto valueStr = std::string(value); + const auto* valueCStr = valueStr.c_str(); #ifdef _WIN32 // Windows version - err = _putenv_s(std::string(key).c_str(), std::string(value).c_str()); + err = _putenv_s(keyCStr, valueCStr); #elif defined(__unix__) || defined(__APPLE__) // Unix/macOS version - err = setenv(std::string(key).c_str(), std::string(value).c_str(), 1); + err = setenv(keyCStr, valueCStr, 1); #else - VELOX_UNSUPPORTED("Unsuppported OS"); + VELOX_UNSUPPORTED("Unsupported OS"); #endif VELOX_CHECK_EQ(0, err); // Sanity check - auto valueForCheck = std::getenv(std::string(key).c_str()); - VELOX_CHECK_EQ(0, std::strcmp(std::string(value).c_str(), valueForCheck)); + auto valueForCheck = std::getenv(keyCStr); + VELOX_CHECK_EQ(0, std::strcmp(valueCStr, valueForCheck)); } void ClpS3AuthProviderBase::unsetEnvironmentVariable(std::string_view key) { int err{0}; + const auto keyStr = std::string(key); + const auto* keyCStr = keyStr.c_str(); #ifdef _WIN32 // Windows version - err = _putenv(fmt::format("{}=", std::string(key).c_str())); + err = _putenv(fmt::format("{}=", keyCStr)); #elif defined(__unix__) || defined(__APPLE__) // Unix/macOS version - err = unsetenv(std::string(key).c_str()); + err = unsetenv(keyCStr); #else - VELOX_UNSUPPORTED("Unsuppported OS"); + VELOX_UNSUPPORTED("Unsupported OS"); #endif VELOX_CHECK_EQ(0, err); // Sanity check - auto valueForCheck = std::getenv(std::string(key).c_str()); + auto valueForCheck = std::getenv(keyCStr); VELOX_CHECK_NULL(valueForCheck); } diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h index 434bbf933c8..f3a43b28ca8 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -32,20 +32,22 @@ class ClpS3AuthProviderBase { : config_(config) {} virtual ~ClpS3AuthProviderBase() = default; - /// Construct the actual S3 URL so that CLP-s can access the split. + /// Constructs the actual S3 URL so that CLP-s can access the split. /// /// @param splitPath The current path stored in the split to add. /// @return The constructed S3 URL. - virtual const std::string constructS3Url(std::string_view splitPath) = 0; + virtual std::string constructS3Url(std::string_view splitPath) = 0; - /// Export the three environment variables needed by CLP-s to system: + /// Retrieves some customized config option value and save them to the class + /// property. Also exports the three environment variables needed by CLP-s + /// to system: /// AWS_ENV_VAR_ACCESS_KEY_ID /// AWS_ENV_VAR_SECRET_ACCESS_KEY /// AWS_ENV_VAR_SESSION_TOKEN (optional) /// So that in runtime, CLP-s’s code can execute S3-related logic correctly. /// /// @return Did exportation succeed or not. - virtual bool exportAuthEnvironmentVariables() const = 0; + virtual bool parseConfigAndExportAuthEnvironmentVariables() = 0; protected: /// Set the environment variable for different OS, then get the environment diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index 8523990c75f..a5f7e9064c2 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -39,16 +39,26 @@ class ClpConfigTest : public testing::Test { class ClpS3AuthProviderBaseTest : public ClpConfigTest { public: + /// Checks whether an environment variable is undefined or equals a given + /// value. + /// + /// @param key The name of the environment variable to check. + /// @param expectedValue Optional expected value to compare against. If + /// std::nullopt, the function returns true if the variable is not + /// defined. + /// @return True if: + /// - expectedValue is std::nullopt and the variable is not defined, + /// or + /// - expectedValue is set and matches the variable's current value. bool checkEnvironmentVariableEquals( std::string_view key, - std::string_view value) { - auto* actualValue = std::getenv(std::string(key).c_str()); - return 0 == std::strcmp(std::string(value).c_str(), actualValue); - } - - bool checkEnvironmentVariableExists(std::string_view key) { - auto* value = std::getenv(std::string(key).c_str()); - return value != nullptr; + std::optional expectedValue) { + const char* actualValue = std::getenv(std::string(key).c_str()); + if (nullptr == actualValue) { + return !expectedValue.has_value(); + } + return expectedValue.has_value() && + std::string_view(actualValue) == expectedValue.value(); } }; @@ -67,7 +77,6 @@ class ClpPackageS3AuthProviderTest : public ClpS3AuthProviderBaseTest { TEST_F(ClpConfigTest, invalidAuthProvider) { const std::unordered_map configMap( {{ClpConfig::kAuthProvider, "dummy-provider"}}); - // Both access/secret keys and iam-role cannot be specified VELOX_ASSERT_UNSUPPORTED_THROW( buildClpConfig(configMap), "Unsupported s3 auth provider type: dummy-provider."); @@ -95,7 +104,8 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}, {ClpPackageS3AuthProvider::kSessionToken, cTestSessionToken}}); auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); - VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); + VELOX_CHECK( + clpPackageS3AuthProvider->parseConfigAndExportAuthEnvironmentVariables()); VELOX_CHECK(checkEnvironmentVariableEquals( ClpPackageS3AuthProvider::kEnvAwsAccessKeyId, cTestAccessKeyId)); VELOX_CHECK(checkEnvironmentVariableEquals( @@ -110,11 +120,10 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}}; clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); - VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); VELOX_CHECK( - false == - checkEnvironmentVariableExists( - ClpPackageS3AuthProvider::kEnvAwsSessionToken)); + clpPackageS3AuthProvider->parseConfigAndExportAuthEnvironmentVariables()); + VELOX_CHECK(checkEnvironmentVariableEquals( + ClpPackageS3AuthProvider::kEnvAwsSessionToken, std::nullopt)); } } // namespace facebook::velox::connector::clp From ccce2d6b431e0cba74f50c5ece0ec499ed60e1a9 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 1 Aug 2025 20:55:56 +0000 Subject: [PATCH 07/15] Address coderabbitai comments to add a environment guard in unit test --- velox/connectors/clp/tests/ClpConfigTest.cpp | 111 +++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index a5f7e9064c2..346bfb25607 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -27,14 +27,125 @@ namespace facebook::velox::connector::clp { namespace { +class ClpEnvironmentVariableGuard { + public: + void saveOriginalEnvironmentVariables() { + originalEnv_.clear(); + originalKeys_.clear(); + +#if defined(_WIN32) + LPCH envStrings = GetEnvironmentStringsA(); + if (!envStrings) + return; + + LPCH var = envStrings; + while (*var) { + std::string entry(var); + auto pos = entry.find('='); + if (pos != std::string::npos) { + std::string key = entry.substr(0, pos); + std::string value = entry.substr(pos + 1); + originalEnv_[key] = value; + originalKeys_.insert(key); + } + var += entry.size() + 1; + } + FreeEnvironmentStringsA(envStrings); +#elif defined(__unix__) || defined(__APPLE__) + for (char** current = environ; *current; ++current) { + std::string entry(*current); + auto pos = entry.find('='); + if (pos != std::string::npos) { + std::string key = entry.substr(0, pos); + std::string value = entry.substr(pos + 1); + originalEnv_[key] = value; + originalKeys_.insert(key); + } + } +#else + VELOX_UNSUPPORTED("Unsupported OS"); +#endif + } + + void restoreEnvironmentVariables() { +#if defined(_WIN32) + // Remove added vars and restore originals + for (auto const& [key, value] : originalEnv_) { + _putenv_s(key.c_str(), value.c_str()); // restore original + } + + // Collect variables to unset + std::vector keysToUnset; + for (char** env = _environ; *env; ++env) { + std::string entry(*env); + auto pos = entry.find('='); + if (pos != std::string::npos) { + std::string key = entry.substr(0, pos); + if (originalKeys_.find(key) == originalKeys_.end()) { + keysToUnset.push_back(key); + } + } + } + + // Unset them + for (const auto& key : keysToUnset) { + _putenv((key + "=").c_str()); + } +#elif defined(__unix__) || defined(__APPLE__) + // Restore original variables + for (auto const& [key, value] : originalEnv_) { + setenv(key.c_str(), value.c_str(), 1); + } + + // Collect keys to unset + std::vector keysToUnset; + for (char** env = environ; *env; ++env) { + std::string entry(*env); + auto pos = entry.find('='); + if (pos != std::string::npos) { + std::string key = entry.substr(0, pos); + if (originalKeys_.find(key) == originalKeys_.end()) { + keysToUnset.push_back(key); + } + } + } + + // Unset them + for (const auto& key : keysToUnset) { + unsetenv(key.c_str()); + } +#else + VELOX_UNSUPPORTED("Unsupported OS"); +#endif + } + + private: + std::map originalEnv_; + std::set originalKeys_; +}; + class ClpConfigTest : public testing::Test { public: + explicit ClpConfigTest() + : envVarGuard_(std::make_unique()) {} + + void SetUp() override { + envVarGuard_->saveOriginalEnvironmentVariables(); + } + + void TearDown() override { + envVarGuard_->restoreEnvironmentVariables(); + } + std::unique_ptr buildClpConfig( std::unordered_map configMap) { auto config = std::make_shared(std::move(configMap)); return std::make_unique(config); } + + private: + std::unique_ptr envVarGuard_; }; class ClpS3AuthProviderBaseTest : public ClpConfigTest { From 1f10099d7f8d1bc78b67b04e6130f9995f41fda5 Mon Sep 17 00:00:00 2001 From: anlowee Date: Mon, 4 Aug 2025 09:39:21 -0400 Subject: [PATCH 08/15] Fix unit tests --- velox/connectors/clp/tests/ClpConfigTest.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index 346bfb25607..66d8b64dafc 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -197,7 +197,8 @@ TEST_F(ClpS3AuthProviderBaseTest, caseInsensitiveAuthProvider) { const std::unordered_map configMap( {{ClpConfig::kAuthProvider, "ClP_PaCkAgE"}, {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, - {ClpPackageS3AuthProvider::kSecretAccessKey, "bbbbbb"}}); + {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, + {ClpPackageS3AuthProvider::kSecretAccessKey, "cccccc"}}); VELOX_CHECK_NOT_NULL(buildClpConfig(configMap)); } From 2cc2e5271c712c4feefb8f8b2338c0ee0e089901 Mon Sep 17 00:00:00 2001 From: "Xiaochong(Eddy) Wei" <40865608+anlowee@users.noreply.github.com> Date: Tue, 5 Aug 2025 12:05:03 -0400 Subject: [PATCH 09/15] Apply suggestions from code review Co-authored-by: Devin Gibson --- velox/connectors/clp/ClpConfig.cpp | 2 +- velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h | 8 ++++---- velox/connectors/clp/tests/ClpConfigTest.cpp | 3 +-- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index 7eed5548851..c12f47f6654 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -51,7 +51,7 @@ ClpConfig::ClpConfig(std::shared_ptr config) { VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); config_ = std::move(config); - // Setup S3 environment variables needed by CLP by user-specific ways + // Setup S3 environment variables needed by CLP using configured auth provider switch ( stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { case ClpConfig::S3AuthProvider::kClpPackage: diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h index f3a43b28ca8..29769d2b885 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -50,17 +50,17 @@ class ClpS3AuthProviderBase { virtual bool parseConfigAndExportAuthEnvironmentVariables() = 0; protected: - /// Set the environment variable for different OS, then get the environment - /// variable to do the sanity check. + /// Sets an environment variable for different OS, then gets the environment + /// variable to do a sanity check. /// /// @param key The environment variable name to set. /// @param value The environment variable value to set. - /// @return Did exportation succeed or not. + /// @return Whether environment variable export succeeded or not. static void setupEnvironmentVariable( std::string_view key, std::string_view value); - /// Unset the environment variable for different OS, then get the environment + /// Unsets an environment variable for different OS, then gets the environment /// variable to do the sanity check. /// /// @param key The environment variable name to unset. diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index 66d8b64dafc..63c8220b602 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -150,8 +150,7 @@ class ClpConfigTest : public testing::Test { class ClpS3AuthProviderBaseTest : public ClpConfigTest { public: - /// Checks whether an environment variable is undefined or equals a given - /// value. + /// Checks whether an environment variable matches a given value. /// /// @param key The name of the environment variable to check. /// @param expectedValue Optional expected value to compare against. If From a23915b4a587fab8a84e7f004e455ed600cfa963 Mon Sep 17 00:00:00 2001 From: anlowee Date: Tue, 5 Aug 2025 16:05:54 +0000 Subject: [PATCH 10/15] Fix unit tests --- velox/connectors/clp/tests/ClpConnectorTest.cpp | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index 89aceaaf980..a234fa4dc0d 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -18,11 +18,11 @@ #include #include "velox/common/base/Fs.h" -#include "velox/common/base/tests/GTestUtils.h" #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnector.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpTableHandle.h" +#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" #include "velox/exec/tests/utils/AssertQueryBuilder.h" #include "velox/exec/tests/utils/OperatorTestBase.h" #include "velox/exec/tests/utils/PlanBuilder.h" @@ -55,7 +55,12 @@ class ClpConnectorTest : public exec::test::OperatorTestBase { kClpConnectorId, std::make_shared( std::unordered_map{ - {"clp.split-source", "local"}})); + {"clp.split-source", "local"}, + {ClpConfig::kAuthProvider, "clp_package"}, + {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, + {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, + {ClpPackageS3AuthProvider::kSecretAccessKey, + "cccccc"}})); connector::registerConnector(clpConnector); } From a3f155d86cf1082845c4e6620b692ec851f1c29d Mon Sep 17 00:00:00 2001 From: anlowee Date: Tue, 5 Aug 2025 16:12:36 +0000 Subject: [PATCH 11/15] Address comments --- velox/connectors/clp/ClpConfig.cpp | 2 +- .../clp/search_lib/ClpPackageS3AuthProvider.cpp | 8 ++++---- .../clp/search_lib/ClpPackageS3AuthProvider.h | 2 +- .../clp/search_lib/ClpS3AuthProviderBase.cpp | 2 +- .../clp/search_lib/ClpS3AuthProviderBase.h | 17 ++++++++--------- velox/connectors/clp/tests/ClpConfigTest.cpp | 8 +++----- 6 files changed, 18 insertions(+), 21 deletions(-) diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index c12f47f6654..16bdf2b9b19 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -60,7 +60,7 @@ ClpConfig::ClpConfig(std::shared_ptr config) { default: VELOX_FAIL(); } - VELOX_CHECK(s3AuthProvider_->parseConfigAndExportAuthEnvironmentVariables()); + VELOX_CHECK(s3AuthProvider_->exportAuthEnvironmentVariables()); } std::shared_ptr ClpConfig::s3AuthProvider() const { diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp index b897998c5cb..f76b09c556a 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -26,7 +26,7 @@ std::string ClpPackageS3AuthProvider::constructS3Url( return fmt::format("{}/{}", this->endPoint_, splitPath); } -bool ClpPackageS3AuthProvider::parseConfigAndExportAuthEnvironmentVariables() { +bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() { this->endPoint_ = config_->get(kEndPoint, ""); VELOX_CHECK( !this->endPoint_.empty(), fmt::format("{} cannot be empty", kEndPoint)); @@ -42,10 +42,10 @@ bool ClpPackageS3AuthProvider::parseConfigAndExportAuthEnvironmentVariables() { VELOX_CHECK( !secretAccessKey.empty(), fmt::format("{} cannot be empty", kSecretAccessKey)); - setupEnvironmentVariable(kEnvAwsAccessKeyId, accessKeyId); - setupEnvironmentVariable(kEnvAwsSecretAccessKey, secretAccessKey); + setEnvironmentVariable(kEnvAwsAccessKeyId, accessKeyId); + setEnvironmentVariable(kEnvAwsSecretAccessKey, secretAccessKey); if (!sessionToken.empty()) { - setupEnvironmentVariable(kEnvAwsSessionToken, sessionToken); + setEnvironmentVariable(kEnvAwsSessionToken, sessionToken); } else { unsetEnvironmentVariable(kEnvAwsSessionToken); } diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h index d34c2e12466..ead4df0cf3b 100644 --- a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h @@ -41,7 +41,7 @@ class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase { std::string constructS3Url(std::string_view splitPath) override; - bool parseConfigAndExportAuthEnvironmentVariables() override; + bool exportAuthEnvironmentVariables() override; private: std::string endPoint_; diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp index 77b5e82d4c9..17de4759f66 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -26,7 +26,7 @@ class ConfigBase; namespace facebook::velox::connector::clp { -void ClpS3AuthProviderBase::setupEnvironmentVariable( +void ClpS3AuthProviderBase::setEnvironmentVariable( std::string_view key, std::string_view value) { int err{0}; diff --git a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h index 29769d2b885..c4a0803d06d 100644 --- a/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -38,16 +38,15 @@ class ClpS3AuthProviderBase { /// @return The constructed S3 URL. virtual std::string constructS3Url(std::string_view splitPath) = 0; - /// Retrieves some customized config option value and save them to the class - /// property. Also exports the three environment variables needed by CLP-s + /// Exports the three environment variables needed by CLP-s /// to system: - /// AWS_ENV_VAR_ACCESS_KEY_ID - /// AWS_ENV_VAR_SECRET_ACCESS_KEY - /// AWS_ENV_VAR_SESSION_TOKEN (optional) - /// So that in runtime, CLP-s’s code can execute S3-related logic correctly. + /// AWS_ACCESS_KEY_ID + /// AWS_SECRET_ACCESS_KEY + /// AWS_SESSION_TOKEN (optional) + /// So that at runtime CLP-S's code can authenticate with S3. /// - /// @return Did exportation succeed or not. - virtual bool parseConfigAndExportAuthEnvironmentVariables() = 0; + /// @return Whether environment variable export succeeded or not. + virtual bool exportAuthEnvironmentVariables() = 0; protected: /// Sets an environment variable for different OS, then gets the environment @@ -56,7 +55,7 @@ class ClpS3AuthProviderBase { /// @param key The environment variable name to set. /// @param value The environment variable value to set. /// @return Whether environment variable export succeeded or not. - static void setupEnvironmentVariable( + static void setEnvironmentVariable( std::string_view key, std::string_view value); diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index 63c8220b602..486f9cf8352 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -196,7 +196,7 @@ TEST_F(ClpS3AuthProviderBaseTest, caseInsensitiveAuthProvider) { const std::unordered_map configMap( {{ClpConfig::kAuthProvider, "ClP_PaCkAgE"}, {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, - {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, + {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, {ClpPackageS3AuthProvider::kSecretAccessKey, "cccccc"}}); VELOX_CHECK_NOT_NULL(buildClpConfig(configMap)); } @@ -215,8 +215,7 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}, {ClpPackageS3AuthProvider::kSessionToken, cTestSessionToken}}); auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); - VELOX_CHECK( - clpPackageS3AuthProvider->parseConfigAndExportAuthEnvironmentVariables()); + VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); VELOX_CHECK(checkEnvironmentVariableEquals( ClpPackageS3AuthProvider::kEnvAwsAccessKeyId, cTestAccessKeyId)); VELOX_CHECK(checkEnvironmentVariableEquals( @@ -231,8 +230,7 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}}; clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); - VELOX_CHECK( - clpPackageS3AuthProvider->parseConfigAndExportAuthEnvironmentVariables()); + VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); VELOX_CHECK(checkEnvironmentVariableEquals( ClpPackageS3AuthProvider::kEnvAwsSessionToken, std::nullopt)); } From 44caed741e1085a52a20479ee4901f3c4074cc88 Mon Sep 17 00:00:00 2001 From: anlowee Date: Tue, 5 Aug 2025 18:25:30 +0000 Subject: [PATCH 12/15] Address coderabbitai comment --- velox/connectors/clp/ClpConfig.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index 16bdf2b9b19..55912f84eb3 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -51,7 +51,8 @@ ClpConfig::ClpConfig(std::shared_ptr config) { VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); config_ = std::move(config); - // Setup S3 environment variables needed by CLP using configured auth provider + // Set up S3 environment variables needed by CLP using configured auth + // provider switch ( stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { case ClpConfig::S3AuthProvider::kClpPackage: From bc60f10d6984800e9b4b87b653ead054517ceb9f Mon Sep 17 00:00:00 2001 From: anlowee Date: Mon, 11 Aug 2025 16:06:32 +0000 Subject: [PATCH 13/15] Add docs --- velox/docs/configs.rst | 33 +++++++++++++++++++++++++++++++ velox/docs/develop/connectors.rst | 33 ++++++++++++++++++++++++------- 2 files changed, 59 insertions(+), 7 deletions(-) diff --git a/velox/docs/configs.rst b/velox/docs/configs.rst index e7864ac2fd2..aee9aa32798 100644 --- a/velox/docs/configs.rst +++ b/velox/docs/configs.rst @@ -910,6 +910,39 @@ CLP Connector - Type - Default Value - Description + * - clp.s3-auth-provider + - string + - + - Specifies the S3 auth provider used by CLP to obtain authentication information (typically + the three environment variables ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``, and the + optional ``AWS_SESSION_TOKEN``). This option is required when the ``clp.storage-type`` + option is set to ``S3``. The CLP package includes a default provider, but users can + implement their own by extending the ``ClpS3AuthProviderBase`` interface. Custom providers + may also introduce their own config options to help determine how these values are + retrieved. The ``clp.s3-``-prefixed options described below apply only when ``CLP_PACKAGE`` + is set as the value for this option, allowing the auth information to be directly specified + in the config file. **Allowed values:** ``CLP_PACKAGE``. + * - clp.s3-access-key-id + - string + - + - The S3 access key ID, exported as the ``AWS_ACCESS_KEY_ID`` environment variable and used + for authentication. + * - clp.s3-end-point + - string + - + - The S3 endpoint. For the CLP package, the complete split URL is formatted as + ``/``, where ```` is obtained from the CLP + package's metadata database and stored in the ``ClpConnectorSplit`` class. + * - clp.s3-secret-access-key + - string + - + - The S3 secret access key, exported as the ``AWS_SECRET_ACCESS_KEY`` environment variable + and used for authentication. + * - clp.s3-session-token + - string + - + - The S3 session token, exported as the ``AWS_SESSION_TOKEN`` environment variable and used + for authentication. This option is optional. * - clp.storage-type - string - FS diff --git a/velox/docs/develop/connectors.rst b/velox/docs/develop/connectors.rst index 21a486e84a5..f0e9ce3eec2 100644 --- a/velox/docs/develop/connectors.rst +++ b/velox/docs/develop/connectors.rst @@ -124,26 +124,45 @@ This is the behavior when the proxy settings are enabled: CLP Connector ------------- -The CLP Connector is used to read CLP archives stored on a local file system or S3. It implements similar -interfaces as the Hive Connector except for the ``DataSink`` interface. Here we only describe the ``DataSource`` -interface and the ``ConnectorSplit`` interface implementation since `Connector` and ``ConnectorFactory`` are -similar to the Hive Connector. +The CLP Connector is used to read CLP splits stored on a local file system or S3. It implements similar interfaces as +the Hive Connector except for the ``DataSink`` interface. Here we only describe the ``DataSource`` interface and the +``ConnectorSplit`` interface implementation since ``Connector`` and ``ConnectorFactory`` are similar to the Hive +Connector. We also describe ``ClpS3AuthProviderBase``, an interface that allows users to customize S3 authentication. ClpConnectorSplit ~~~~~~~~~~~~~~~~~ -``ClpConnectorSplit`` describes a data chunk using ``path``, which is the path to the archive file. +``ClpConnectorSplit`` describes a data chunk using ``path``. This path may be the absolute file path to the split file +if it is stored on a local file system, or the complete (or partial) URL of the split if it is stored on S3. In the +latter case, when only a partial URL is provided, ``ClpS3AuthProviderBase`` provides a hook in ``ClpDataSource`` to +assist in constructing the full URL. Refer to :ref:`ClpS3AuthProviderBase` for details. ClpDataSource ~~~~~~~~~~~~~ ``ClpDataSource`` implements the ``addSplit`` API that consumes a ``ClpConnectorSplit`` and ``next`` API that processes the split and returns a batch of rows. -During initialization, it records the KQL query and archive source (S3 or local). It then iterates through +During initialization, it records the KQL query and split source (S3 or local). It then iterates through each output column, accessing its handle to get its type and original name. For row types, it recursively traverses the nested structure to process each field; for non-row types, it directly maps the Velox column type to a CLP column type. -When a split is added, a ``ClpCursor`` is created with the archive path and input source. The query is parsed +When a split is added, a ``ClpCursor`` is created with the split path and input source. The query is parsed and simplified into an AST. On ``next``, the cursor finds matching row indices and, if any exist, ``ClpDataSource`` recursively creates a row vector composed of lazy vectors, which use CLP column readers to decode and load data as needed during execution. + +.. _ClpS3AuthProviderBase + +ClpS3AuthProviderBase +~~~~~~~~~~~~~~~~~~~~~ +``ClpS3AuthProviderBase`` defines an interface for obtaining S3 authentication information and constructing the +complete split URL from the current URL stored in ``ClpConnectorSplit``. It provides the following two functions: + +1. ``exportAuthEnvironmentVariables()`` – Parses user-defined configuration options and exports the three environment + variables required by CLP-s to the system. This ensures that, at runtime, CLP-s can execute S3-related operations + correctly. This function is invoked immediately after the configuration is parsed. +2. ``constructS3Url()`` – Builds the full S3 URL for a split. It takes the ``path`` property of ``ClpConnectorSplit`` + as its argument, allowing customization in URL construction. + +Additionally, this interface maintains a reference to ``config_``, enabling users to define custom configuration +options for passing any required information. From 7e62d41a4b6fab7814f8bf77c12df8e1cd0103c0 Mon Sep 17 00:00:00 2001 From: anlowee Date: Mon, 11 Aug 2025 17:22:56 +0000 Subject: [PATCH 14/15] Fix the bug that when the storage-type is FS the s3-auth-provider is still required --- velox/connectors/clp/ClpConfig.cpp | 27 +++++++++++-------- velox/connectors/clp/ClpConfig.h | 1 + velox/connectors/clp/tests/ClpConfigTest.cpp | 10 ++++--- .../connectors/clp/tests/ClpConnectorTest.cpp | 8 +----- 4 files changed, 25 insertions(+), 21 deletions(-) diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp index 55912f84eb3..67727fc314c 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -51,17 +51,22 @@ ClpConfig::ClpConfig(std::shared_ptr config) { VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); config_ = std::move(config); - // Set up S3 environment variables needed by CLP using configured auth - // provider - switch ( - stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { - case ClpConfig::S3AuthProvider::kClpPackage: - s3AuthProvider_ = std::make_shared(config_); - break; - default: - VELOX_FAIL(); + storageType_ = + stringToStorageType(config_->get(kStorageType, "FS")); + + if (StorageType::kS3 == storageType_) { + // Set up S3 environment variables needed by CLP using configured auth + // provider + switch ( + stringToS3AuthProvider(config_->get(kAuthProvider, ""))) { + case S3AuthProvider::kClpPackage: + s3AuthProvider_ = std::make_shared(config_); + break; + default: + VELOX_FAIL(); + } + VELOX_CHECK(s3AuthProvider_->exportAuthEnvironmentVariables()); } - VELOX_CHECK(s3AuthProvider_->exportAuthEnvironmentVariables()); } std::shared_ptr ClpConfig::s3AuthProvider() const { @@ -69,7 +74,7 @@ std::shared_ptr ClpConfig::s3AuthProvider() const { } ClpConfig::StorageType ClpConfig::storageType() const { - return stringToStorageType(config_->get(kStorageType, "FS")); + return storageType_; } } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpConfig.h b/velox/connectors/clp/ClpConfig.h index f9e7c845be6..393cc113149 100644 --- a/velox/connectors/clp/ClpConfig.h +++ b/velox/connectors/clp/ClpConfig.h @@ -51,6 +51,7 @@ class ClpConfig { private: std::shared_ptr config_; std::shared_ptr s3AuthProvider_; + StorageType storageType_; }; } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/tests/ClpConfigTest.cpp b/velox/connectors/clp/tests/ClpConfigTest.cpp index 486f9cf8352..4e53fbf8745 100644 --- a/velox/connectors/clp/tests/ClpConfigTest.cpp +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -186,7 +186,8 @@ class ClpPackageS3AuthProviderTest : public ClpS3AuthProviderBaseTest { TEST_F(ClpConfigTest, invalidAuthProvider) { const std::unordered_map configMap( - {{ClpConfig::kAuthProvider, "dummy-provider"}}); + {{"clp.storage-type", "s3"}, + {ClpConfig::kAuthProvider, "dummy-provider"}}); VELOX_ASSERT_UNSUPPORTED_THROW( buildClpConfig(configMap), "Unsupported s3 auth provider type: dummy-provider."); @@ -194,7 +195,8 @@ TEST_F(ClpConfigTest, invalidAuthProvider) { TEST_F(ClpS3AuthProviderBaseTest, caseInsensitiveAuthProvider) { const std::unordered_map configMap( - {{ClpConfig::kAuthProvider, "ClP_PaCkAgE"}, + {{"clp.storage-type", "s3"}, + {ClpConfig::kAuthProvider, "ClP_PaCkAgE"}, {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, {ClpPackageS3AuthProvider::kSecretAccessKey, "cccccc"}}); @@ -209,7 +211,8 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { // Test all properties std::unordered_map configMap( - {{ClpConfig::kAuthProvider, "clp_package"}, + {{"clp.storage-type", "s3"}, + {ClpConfig::kAuthProvider, "clp_package"}, {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId}, {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}, @@ -225,6 +228,7 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) { // Test auth without the session token configMap = { + {"clp.storage-type", "s3"}, {ClpConfig::kAuthProvider, "clp_package"}, {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId}, {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index a234fa4dc0d..b87d72cce4c 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -54,13 +54,7 @@ class ClpConnectorTest : public exec::test::OperatorTestBase { ->newConnector( kClpConnectorId, std::make_shared( - std::unordered_map{ - {"clp.split-source", "local"}, - {ClpConfig::kAuthProvider, "clp_package"}, - {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, - {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, - {ClpPackageS3AuthProvider::kSecretAccessKey, - "cccccc"}})); + std::unordered_map{})); connector::registerConnector(clpConnector); } From 838ab7f7d156ed4e548a086b822ad135155a593c Mon Sep 17 00:00:00 2001 From: anlowee Date: Mon, 11 Aug 2025 17:41:29 +0000 Subject: [PATCH 15/15] Address some of coderabbitai comments --- velox/connectors/clp/tests/ClpConnectorTest.cpp | 1 - velox/docs/develop/connectors.rst | 3 ++- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index b87d72cce4c..0e20dbb23b9 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -22,7 +22,6 @@ #include "velox/connectors/clp/ClpConnector.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpTableHandle.h" -#include "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" #include "velox/exec/tests/utils/AssertQueryBuilder.h" #include "velox/exec/tests/utils/OperatorTestBase.h" #include "velox/exec/tests/utils/PlanBuilder.h" diff --git a/velox/docs/develop/connectors.rst b/velox/docs/develop/connectors.rst index f0e9ce3eec2..ec9946d6ab0 100644 --- a/velox/docs/develop/connectors.rst +++ b/velox/docs/develop/connectors.rst @@ -162,7 +162,8 @@ complete split URL from the current URL stored in ``ClpConnectorSplit``. It prov variables required by CLP-s to the system. This ensures that, at runtime, CLP-s can execute S3-related operations correctly. This function is invoked immediately after the configuration is parsed. 2. ``constructS3Url()`` – Builds the full S3 URL for a split. It takes the ``path`` property of ``ClpConnectorSplit`` - as its argument, allowing customization in URL construction. + as its argument, allowing customization in URL construction. For example, the ``path`` could be ``"prefix/split"``, + which must be prefixed with ``"https://bucket.s3.region.amazonaws.com/"`` to form the complete URL. Additionally, this interface maintains a reference to ``config_``, enabling users to define custom configuration options for passing any required information.