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/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..67727fc314c 100644 --- a/velox/connectors/clp/ClpConfig.cpp +++ b/velox/connectors/clp/ClpConfig.cpp @@ -16,12 +16,24 @@ #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" 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,8 +47,34 @@ 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"); + config_ = std::move(config); + + 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()); + } +} + +std::shared_ptr ClpConfig::s3AuthProvider() const { + return s3AuthProvider_; +} + 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 ce4808eef62..393cc113149 100644 --- a/velox/connectors/clp/ClpConfig.h +++ b/velox/connectors/clp/ClpConfig.h @@ -16,27 +16,29 @@ #pragma once -#include "velox/common/config/Config.h" - namespace facebook::velox::config { class ConfigBase; -} +} // namespace facebook::velox::config namespace facebook::velox::connector::clp { +class ClpS3AuthProviderBase; + 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 { @@ -44,9 +46,12 @@ class ClpConfig { } StorageType storageType() const; + std::shared_ptr s3AuthProvider() const; private: std::shared_ptr config_; + std::shared_ptr s3AuthProvider_; + StorageType storageType_; }; } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index a5a574eb318..d3ad369c5b5 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,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, 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..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( @@ -101,6 +103,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 55e68060fc4..8cd5a414d7a 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -16,8 +16,12 @@ velox_add_library( STATIC ClpCursor.cpp ClpCursor.h + ClpPackageS3AuthProvider.cpp + ClpPackageS3AuthProvider.h 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..f76b09c556a --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp @@ -0,0 +1,56 @@ +/* + * 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 "velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h" +#include "velox/common/base/Exceptions.h" +#include "velox/common/config/Config.h" + +namespace facebook::velox::connector::clp { + +std::string ClpPackageS3AuthProvider::constructS3Url( + std::string_view splitPath) { + VELOX_CHECK(!splitPath.empty(), "splitPath cannot be empty"); + return fmt::format("{}/{}", this->endPoint_, splitPath); +} + +bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() { + this->endPoint_ = config_->get(kEndPoint, ""); + VELOX_CHECK( + !this->endPoint_.empty(), fmt::format("{} cannot be empty", kEndPoint)); + if ('/' == this->endPoint_.back()) { + this->endPoint_.pop_back(); + } + + auto accessKeyId = config_->get(kAccessKeyId, ""); + auto secretAccessKey = config_->get(kSecretAccessKey, ""); + auto sessionToken = config_->get(kSessionToken, ""); + VELOX_CHECK( + !accessKeyId.empty(), fmt::format("{} cannot be empty", kAccessKeyId)); + VELOX_CHECK( + !secretAccessKey.empty(), + fmt::format("{} cannot be empty", kSecretAccessKey)); + setEnvironmentVariable(kEnvAwsAccessKeyId, accessKeyId); + setEnvironmentVariable(kEnvAwsSecretAccessKey, secretAccessKey); + if (!sessionToken.empty()) { + setEnvironmentVariable(kEnvAwsSessionToken, sessionToken); + } else { + unsetEnvironmentVariable(kEnvAwsSessionToken); + } + + return true; +} + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h new file mode 100644 index 00000000000..ead4df0cf3b --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.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 "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" + +namespace facebook::velox::config { +class ConfigBase; +} // namespace 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"; + + 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"; + + std::string constructS3Url(std::string_view splitPath) override; + + bool exportAuthEnvironmentVariables() 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 new file mode 100644 index 00000000000..17de4759f66 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.cpp @@ -0,0 +1,73 @@ +/* + * 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::config + +namespace facebook::velox::connector::clp { + +void ClpS3AuthProviderBase::setEnvironmentVariable( + 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(keyCStr, valueCStr); +#elif defined(__unix__) || defined(__APPLE__) + // Unix/macOS version + err = setenv(keyCStr, valueCStr, 1); +#else + VELOX_UNSUPPORTED("Unsupported OS"); +#endif + VELOX_CHECK_EQ(0, err); + + // Sanity check + 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("{}=", keyCStr)); +#elif defined(__unix__) || defined(__APPLE__) + // Unix/macOS version + err = unsetenv(keyCStr); +#else + VELOX_UNSUPPORTED("Unsupported OS"); +#endif + VELOX_CHECK_EQ(0, err); + + // Sanity check + auto valueForCheck = std::getenv(keyCStr); + 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 new file mode 100644 index 00000000000..c4a0803d06d --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h @@ -0,0 +1,71 @@ +/* + * 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 + +namespace facebook::velox::config { +class ConfigBase; +} // namespace facebook::velox::config + +namespace facebook::velox::connector::clp { + +class ClpS3AuthProviderBase { + public: + explicit ClpS3AuthProviderBase( + std::shared_ptr config) + : config_(config) {} + virtual ~ClpS3AuthProviderBase() = default; + + /// 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 std::string constructS3Url(std::string_view splitPath) = 0; + + /// Exports the three environment variables needed by CLP-s + /// to system: + /// 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 Whether environment variable export succeeded or not. + virtual bool exportAuthEnvironmentVariables() = 0; + + protected: + /// 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 Whether environment variable export succeeded or not. + static void setEnvironmentVariable( + std::string_view key, + std::string_view value); + + /// 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. + static void unsetEnvironmentVariable(std::string_view key); + + std::shared_ptr config_; +}; + +} // namespace facebook::velox::connector::clp 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..4e53fbf8745 --- /dev/null +++ b/velox/connectors/clp/tests/ClpConfigTest.cpp @@ -0,0 +1,242 @@ +/* + * 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 "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 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 { + public: + /// 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 + /// 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::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(); + } +}; + +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( + {{"clp.storage-type", "s3"}, + {ClpConfig::kAuthProvider, "dummy-provider"}}); + VELOX_ASSERT_UNSUPPORTED_THROW( + buildClpConfig(configMap), + "Unsupported s3 auth provider type: dummy-provider."); +} + +TEST_F(ClpS3AuthProviderBaseTest, caseInsensitiveAuthProvider) { + const std::unordered_map configMap( + {{"clp.storage-type", "s3"}, + {ClpConfig::kAuthProvider, "ClP_PaCkAgE"}, + {ClpPackageS3AuthProvider::kAccessKeyId, "aaaaaa"}, + {ClpPackageS3AuthProvider::kEndPoint, "http://aaaaaa"}, + {ClpPackageS3AuthProvider::kSecretAccessKey, "cccccc"}}); + 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( + {{"clp.storage-type", "s3"}, + {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 = { + {"clp.storage-type", "s3"}, + {ClpConfig::kAuthProvider, "clp_package"}, + {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId}, + {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint}, + {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}}; + clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap); + VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables()); + VELOX_CHECK(checkEnvironmentVariableEquals( + ClpPackageS3AuthProvider::kEnvAwsSessionToken, std::nullopt)); +} + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index 89aceaaf980..0e20dbb23b9 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -18,7 +18,6 @@ #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" @@ -54,8 +53,7 @@ class ClpConnectorTest : public exec::test::OperatorTestBase { ->newConnector( kClpConnectorId, std::make_shared( - std::unordered_map{ - {"clp.split-source", "local"}})); + std::unordered_map{})); connector::registerConnector(clpConnector); } 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..ec9946d6ab0 100644 --- a/velox/docs/develop/connectors.rst +++ b/velox/docs/develop/connectors.rst @@ -124,26 +124,46 @@ 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. 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.