feat: Add S3 bucket configuration for path-style URL support.

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 9e991ab49681faff0aea259a37b489c3f416e06e)
+  GIT_TAG v0.8.0)
 
 set(CLP_BUILD_CLP_REGEX_UTILS
     OFF

velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.cpp

@@ -20,20 +20,50 @@
 
 namespace facebook::velox::connector::clp {
 
+namespace {
+// Detects AWS S3 virtual-hosted style endpoints where bucket is in the
+// hostname. Virtual-hosted: https://<bucket>.s3.<region>.amazonaws.com (returns
+// true) Path-style: https://s3.<region>.amazonaws.com (returns false)
+bool isAwsVirtualHostedStyleEndpoint(const std::string& endpoint) {
+  if (endpoint.find("amazonaws.com") == std::string::npos) {
+    return false;
+  }
+
+  auto schemeEnd = endpoint.find("://");
+  if (schemeEnd == std::string::npos) {
+    return false;
+  }
+
+  // Virtual-hosted has ".s3." in hostname (bucket.s3.region.amazonaws.com)
+  // Path-style starts with "s3." (s3.region.amazonaws.com)
+  auto hostPortion = endpoint.substr(schemeEnd + 3);
+  return hostPortion.find(".s3.") != std::string::npos;
+}
+} // namespace
+
 std::string ClpPackageS3AuthProvider::constructS3Url(
     std::string_view splitPath) {
   VELOX_CHECK(!splitPath.empty(), "splitPath cannot be empty");
-  return fmt::format("{}/{}", this->endPoint_, splitPath);
+  if (bucket_.empty()) {
+    return fmt::format("{}/{}", endPoint_, splitPath);
+  }
+  return fmt::format("{}/{}/{}", endPoint_, bucket_, splitPath);
 }
 
 bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() {
-  this->endPoint_ = config_->get<std::string>(kEndPoint, "");
-  VELOX_CHECK(
-      !this->endPoint_.empty(), fmt::format("{} cannot be empty", kEndPoint));
-  if ('/' == this->endPoint_.back()) {
-    this->endPoint_.pop_back();
+  endPoint_ = config_->get<std::string>(kEndPoint, "");
+  VELOX_CHECK(!endPoint_.empty(), fmt::format("{} cannot be empty", kEndPoint));
+  if ('/' == endPoint_.back()) {
+    endPoint_.pop_back();
   }
 
+  bucket_ = config_->get<std::string>(kBucket, "");
+  VELOX_CHECK(
+      bucket_.empty() || !isAwsVirtualHostedStyleEndpoint(endPoint_),
+      "{} should not be set when using AWS S3 virtual-hosted style URLs. "
+      "The bucket is already part of the endpoint hostname.",
+      kBucket);
+
   auto accessKeyId = config_->get<std::string>(kAccessKeyId, "");
   auto secretAccessKey = config_->get<std::string>(kSecretAccessKey, "");
   auto sessionToken = config_->get<std::string>(kSessionToken, "");
@@ -42,8 +72,10 @@ bool ClpPackageS3AuthProvider::exportAuthEnvironmentVariables() {
   VELOX_CHECK(
       !secretAccessKey.empty(),
       fmt::format("{} cannot be empty", kSecretAccessKey));
+
   setEnvironmentVariable(kEnvAwsAccessKeyId, accessKeyId);
   setEnvironmentVariable(kEnvAwsSecretAccessKey, secretAccessKey);
+
   if (!sessionToken.empty()) {
     setEnvironmentVariable(kEnvAwsSessionToken, sessionToken);
   } else {

velox/connectors/clp/search_lib/ClpPackageS3AuthProvider.h

@@ -31,6 +31,7 @@ class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase {
       : ClpS3AuthProviderBase(config) {}
 
   static constexpr const char* kAccessKeyId = "clp.s3-access-key-id";
+  static constexpr const char* kBucket = "clp.s3-bucket";
   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";
@@ -44,6 +45,7 @@ class ClpPackageS3AuthProvider : public ClpS3AuthProviderBase {
   bool exportAuthEnvironmentVariables() override;
 
  private:
+  std::string bucket_;
   std::string endPoint_;
 };
 

velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp

@@ -173,7 +173,7 @@ ErrorCode ClpArchiveCursor::loadSplit() {
         case ColumnType::String:
           columnDescriptor->set_matching_types(
               LiteralType::ClpStringT | LiteralType::VarStringT |
-              LiteralType::EpochDateT);
+              LiteralType::TimestampT);
           break;
         case ColumnType::Integer:
           columnDescriptor->set_matching_types(LiteralType::IntegerT);
@@ -189,7 +189,7 @@ ErrorCode ClpArchiveCursor::loadSplit() {
           break;
         case ColumnType::Timestamp:
           columnDescriptor->set_matching_types(
-              LiteralType::EpochDateT | LiteralType::IntegerT |
+              LiteralType::TimestampT | LiteralType::IntegerT |
               LiteralType::FloatT);
           break;
         default:

velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp

@@ -55,18 +55,18 @@ void ClpIrVectorLoader::loadInternal(
         if (value->is<std::string>()) {
           auto stringValue = value->get_immutable_view<std::string>();
           stringVector->set(vectorIndex, StringView(stringValue));
-        } else if (value->is<::clp::ir::EightByteEncodedTextAst>()) {
+        } else if (value->is<::clp::ffi::EightByteEncodedTextAst>()) {
           auto decodeResult =
-              value->get_immutable_view<::clp::ir::EightByteEncodedTextAst>()
-                  .decode_and_unparse();
+              value->get_immutable_view<::clp::ffi::EightByteEncodedTextAst>()
+                  .to_string();
           if (!decodeResult.has_value()) {
             continue;
           }
           stringVector->set(vectorIndex, StringView(decodeResult.value()));
-        } else if (value->is<::clp::ir::FourByteEncodedTextAst>()) {
+        } else if (value->is<::clp::ffi::FourByteEncodedTextAst>()) {
           auto decodeResult =
-              value->get_immutable_view<::clp::ir::FourByteEncodedTextAst>()
-                  .decode_and_unparse();
+              value->get_immutable_view<::clp::ffi::FourByteEncodedTextAst>()
+                  .to_string();
           if (!decodeResult.has_value()) {
             continue;
           }
@@ -117,18 +117,18 @@ void ClpIrVectorLoader::loadInternal(
       case ColumnType::Array: {
         auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector);
         std::string jsonString;
-        if (value->is<::clp::ir::EightByteEncodedTextAst>()) {
+        if (value->is<::clp::ffi::EightByteEncodedTextAst>()) {
           auto decodeResult =
-              value->get_immutable_view<::clp::ir::EightByteEncodedTextAst>()
-                  .decode_and_unparse();
+              value->get_immutable_view<::clp::ffi::EightByteEncodedTextAst>()
+                  .to_string();
           if (!decodeResult.has_value()) {
             continue;
           }
           jsonString = std::move(decodeResult.value());
         } else {
           auto decodeResult =
-              value->get_immutable_view<::clp::ir::FourByteEncodedTextAst>()
-                  .decode_and_unparse();
+              value->get_immutable_view<::clp::ffi::FourByteEncodedTextAst>()
+                  .to_string();
           if (!decodeResult.has_value()) {
             continue;
           }

velox/connectors/clp/tests/ClpConfigTest.cpp

@@ -205,7 +205,8 @@ TEST_F(ClpS3AuthProviderBaseTest, caseInsensitiveAuthProvider) {
 
 TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) {
   const std::string cTestAccessKeyId{"aaaaaa"};
-  const std::string cTestEndPoint{"http://aaaaaa"};
+  const std::string cTestBucket{"test-bucket"};
+  const std::string cTestEndPoint{"http://localhost:9000"};
   const std::string cTestSecretAccessKey{"bbbbbb"};
   const std::string cTestSessionToken{"cccccc"};
 
@@ -214,6 +215,7 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) {
       {{"clp.storage-type", "s3"},
        {ClpConfig::kAuthProvider, "clp_package"},
        {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId},
+       {ClpPackageS3AuthProvider::kBucket, cTestBucket},
        {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint},
        {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey},
        {ClpPackageS3AuthProvider::kSessionToken, cTestSessionToken}});
@@ -239,4 +241,93 @@ TEST_F(ClpPackageS3AuthProviderTest, readAndExportAwsAuthEnvironmentVariables) {
       ClpPackageS3AuthProvider::kEnvAwsSessionToken, std::nullopt));
 }
 
+// Tests URL construction for S3-compatible storage (e.g., MinIO) using
+// path-style URLs with bucket.
+TEST_F(ClpPackageS3AuthProviderTest, constructS3UrlForPathStyleWithBucket) {
+  const std::string cTestAccessKeyId{"aaaaaa"};
+  const std::string cTestBucket{"logs"};
+  const std::string cTestEndPoint{"http://172.26.105.44:9000"};
+  const std::string cTestSecretAccessKey{"bbbbbb"};
+  const std::string cTestSplitPath{"archives/default/abc123"};
+
+  std::unordered_map<std::string, std::string> configMap(
+      {{"clp.storage-type", "s3"},
+       {ClpConfig::kAuthProvider, "clp_package"},
+       {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId},
+       {ClpPackageS3AuthProvider::kBucket, cTestBucket},
+       {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint},
+       {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}});
+  auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap);
+  VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables());
+
+  auto url = clpPackageS3AuthProvider->constructS3Url(cTestSplitPath);
+  VELOX_CHECK_EQ(url, "http://172.26.105.44:9000/logs/archives/default/abc123");
+}
+
+// Tests URL construction for AWS S3 using virtual-hosted style URLs without
+// bucket config.
+TEST_F(ClpPackageS3AuthProviderTest, constructS3UrlForAwsVirtualHostedStyle) {
+  const std::string cTestAccessKeyId{"aaaaaa"};
+  const std::string cTestEndPoint{"https://logs.s3.us-east-1.amazonaws.com"};
+  const std::string cTestSecretAccessKey{"bbbbbb"};
+  const std::string cTestSplitPath{"archives/default/abc123"};
+
+  std::unordered_map<std::string, std::string> configMap(
+      {{"clp.storage-type", "s3"},
+       {ClpConfig::kAuthProvider, "clp_package"},
+       {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId},
+       {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint},
+       {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}});
+  auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap);
+  VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables());
+
+  auto url = clpPackageS3AuthProvider->constructS3Url(cTestSplitPath);
+  VELOX_CHECK_EQ(
+      url, "https://logs.s3.us-east-1.amazonaws.com/archives/default/abc123");
+}
+
+// Tests URL construction for AWS S3 using path-style URLs with bucket config.
+TEST_F(ClpPackageS3AuthProviderTest, constructS3UrlForAwsPathStyleWithBucket) {
+  const std::string cTestAccessKeyId{"aaaaaa"};
+  const std::string cTestBucket{"logs"};
+  const std::string cTestEndPoint{"https://s3.us-east-1.amazonaws.com"};
+  const std::string cTestSecretAccessKey{"bbbbbb"};
+  const std::string cTestSplitPath{"archives/default/abc123"};
+
+  std::unordered_map<std::string, std::string> configMap(
+      {{"clp.storage-type", "s3"},
+       {ClpConfig::kAuthProvider, "clp_package"},
+       {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId},
+       {ClpPackageS3AuthProvider::kBucket, cTestBucket},
+       {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint},
+       {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}});
+  auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap);
+  VELOX_CHECK(clpPackageS3AuthProvider->exportAuthEnvironmentVariables());
+
+  auto url = clpPackageS3AuthProvider->constructS3Url(cTestSplitPath);
+  VELOX_CHECK_EQ(
+      url, "https://s3.us-east-1.amazonaws.com/logs/archives/default/abc123");
+}
+
+// Tests that setting bucket with AWS virtual-hosted style endpoint is rejected.
+TEST_F(ClpPackageS3AuthProviderTest, rejectBucketWithAwsVirtualHostedEndpoint) {
+  const std::string cTestAccessKeyId{"aaaaaa"};
+  const std::string cTestBucket{"logs"};
+  const std::string cTestEndPoint{"https://logs.s3.us-east-1.amazonaws.com"};
+  const std::string cTestSecretAccessKey{"bbbbbb"};
+
+  std::unordered_map<std::string, std::string> configMap(
+      {{"clp.storage-type", "s3"},
+       {ClpConfig::kAuthProvider, "clp_package"},
+       {ClpPackageS3AuthProvider::kAccessKeyId, cTestAccessKeyId},
+       {ClpPackageS3AuthProvider::kBucket, cTestBucket},
+       {ClpPackageS3AuthProvider::kEndPoint, cTestEndPoint},
+       {ClpPackageS3AuthProvider::kSecretAccessKey, cTestSecretAccessKey}});
+  auto clpPackageS3AuthProvider = buildClpPackageS3AuthProvider(configMap);
+  VELOX_ASSERT_THROW(
+      clpPackageS3AuthProvider->exportAuthEnvironmentVariables(),
+      "clp.s3-bucket should not be set when using AWS S3 virtual-hosted style "
+      "URLs");
+}
+
 } // namespace facebook::velox::connector::clp