fix

Author: MisterRaindrop

CMakeLists.txt

@@ -45,7 +45,7 @@ option(ICEBERG_BUILD_TESTS "Build tests" ON)
 option(ICEBERG_BUILD_BUNDLE "Build the battery included library" ON)
 option(ICEBERG_BUILD_REST "Build rest catalog client" ON)
 option(ICEBERG_BUILD_REST_INTEGRATION_TESTS "Build rest catalog integration tests" OFF)
-option(ICEBERG_S3 "Build with S3 support" ON)
+option(ICEBERG_S3 "Build with S3 support" OFF)
 option(ICEBERG_ENABLE_ASAN "Enable Address Sanitizer" OFF)
 option(ICEBERG_ENABLE_UBSAN "Enable Undefined Behavior Sanitizer" OFF)
 
@@ -69,6 +69,12 @@ if(ICEBERG_BUILD_REST_INTEGRATION_TESTS AND WIN32)
   message(WARNING "Cannot build rest integration test on Windows, turning it off.")
 endif()
 
+# ICEBERG_S3 requires ICEBERG_BUILD_BUNDLE
+if(NOT ICEBERG_BUILD_BUNDLE AND ICEBERG_S3)
+  set(ICEBERG_S3 OFF)
+  message(STATUS "ICEBERG_S3 is disabled because ICEBERG_BUILD_BUNDLE is OFF")
+endif()
+
 include(CMakeParseArguments)
 include(IcebergBuildUtils)
 include(IcebergSanitizer)

src/iceberg/CMakeLists.txt

@@ -104,6 +104,7 @@ set(ICEBERG_SOURCES
     update/update_statistics.cc
     util/bucket_util.cc
     util/content_file_util.cc
+    util/file_io_util.cc
     util/conversions.cc
     util/decimal.cc
     util/gzip_internal.cc
@@ -177,7 +178,7 @@ add_subdirectory(util)
 if(ICEBERG_BUILD_BUNDLE)
   set(ICEBERG_BUNDLE_SOURCES
       arrow/arrow_fs_file_io.cc
-      arrow/arrow_s3_file_io.cc
+      arrow/s3/arrow_s3_file_io.cc
       arrow/file_io_register.cc
       arrow/metadata_column_util.cc
       avro/avro_data_util.cc

src/iceberg/arrow/CMakeLists.txt

@@ -16,3 +16,5 @@
 # under the License.
 
 iceberg_install_all_headers(iceberg/arrow)
+
+add_subdirectory(s3)

src/iceberg/arrow/arrow_file_io.h

@@ -38,13 +38,16 @@ ICEBERG_BUNDLE_EXPORT std::unique_ptr<FileIO> MakeLocalFileIO();
 /// This function initializes the S3 subsystem if not already initialized (thread-safe).
 /// The S3 initialization is done once per process using std::call_once.
 ///
-/// \param uri An S3 URI (must start with "s3://") used to validate the scheme.
-/// \param properties Optional configuration properties for S3 access. See S3Properties
+/// \param properties Configuration properties for S3 access. See S3Properties
 ///        for available keys (credentials, region, endpoint, timeouts, etc.).
-/// \return A FileIO instance for S3 operations, or an error if S3 is not supported
-///         or the URI is invalid.
+/// \return A FileIO instance for S3 operations, or an error if S3 is not supported.
 ICEBERG_BUNDLE_EXPORT Result<std::unique_ptr<FileIO>> MakeS3FileIO(
-    const std::string& uri,
     const std::unordered_map<std::string, std::string>& properties = {});
 
+/// \brief Finalize (clean up) the Arrow S3 subsystem.
+///
+/// Must be called before process exit if S3 was initialized, otherwise Arrow's
+/// static destructors may cause a non-zero exit.
+ICEBERG_BUNDLE_EXPORT Status FinalizeS3();
+
 }  // namespace iceberg::arrow

src/iceberg/arrow/file_io_register.cc

@@ -26,26 +26,24 @@
 
 namespace iceberg::arrow {
 
-void RegisterFileIO() {
+void RegisterLocalFileIO() {
   static std::once_flag flag;
   std::call_once(flag, []() {
-    // Register Arrow local filesystem FileIO
     FileIORegistry::Register(
-        FileIORegistry::kArrowLocalFileIO,
-        [](const std::string& /*warehouse*/,
-           const std::unordered_map<std::string, std::string>& /*properties*/)
-            -> Result<std::shared_ptr<FileIO>> {
-          return std::shared_ptr<FileIO>(MakeLocalFileIO());
-        });
+        std::string(FileIORegistry::kArrowLocalFileIO),
+        [](const std::unordered_map<std::string, std::string>& /*properties*/)
+            -> Result<std::unique_ptr<FileIO>> { return MakeLocalFileIO(); });
+  });
+}
 
-    // Register Arrow S3 FileIO
+void RegisterS3FileIO() {
+  static std::once_flag flag;
+  std::call_once(flag, []() {
     FileIORegistry::Register(
-        FileIORegistry::kArrowS3FileIO,
-        [](const std::string& warehouse,
-           const std::unordered_map<std::string, std::string>& properties)
-            -> Result<std::shared_ptr<FileIO>> {
-          ICEBERG_ASSIGN_OR_RAISE(auto file_io, MakeS3FileIO(warehouse, properties));
-          return std::shared_ptr<FileIO>(std::move(file_io));
+        std::string(FileIORegistry::kArrowS3FileIO),
+        [](const std::unordered_map<std::string, std::string>& properties)
+            -> Result<std::unique_ptr<FileIO>> {
+          return MakeS3FileIO(properties);
         });
   });
 }

src/iceberg/arrow/file_io_register.h

@@ -26,15 +26,10 @@
 
 namespace iceberg::arrow {
 
-/// \brief Register Arrow FileIO implementations (local and S3) into the
-/// FileIORegistry.
-///
-/// This function is idempotent and thread-safe. It registers:
-///   - ArrowFileIO (local filesystem)
-///   - ArrowS3FileIO (S3 filesystem)
-///
-/// Must be called before using FileIORegistry::Load() with the built-in
-/// implementation names (e.g., from RestCatalog::Make(config)).
-ICEBERG_BUNDLE_EXPORT void RegisterFileIO();
+/// \brief Register the Arrow local filesystem FileIO into the FileIORegistry.
+ICEBERG_BUNDLE_EXPORT void RegisterLocalFileIO();
+
+/// \brief Register the Arrow S3 FileIO into the FileIORegistry.
+ICEBERG_BUNDLE_EXPORT void RegisterS3FileIO();
 
 }  // namespace iceberg::arrow

src/iceberg/arrow/s3/CMakeLists.txt

@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+iceberg_install_all_headers(iceberg/arrow/s3)

src/iceberg/arrow/arrow_s3_file_io.cc src/iceberg/arrow/s3/arrow_s3_file_io.ccsrc/iceberg/arrow/arrow_s3_file_io.cc renamed to src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -19,7 +19,7 @@
 
 #include <cstdlib>
 #include <mutex>
-#include <stdexcept>
+#include <string>
 
 #include <arrow/filesystem/filesystem.h>
 #ifdef ICEBERG_S3_ENABLED
@@ -32,8 +32,9 @@
 #include "iceberg/arrow/arrow_file_io.h"
 #include "iceberg/arrow/arrow_fs_file_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
-#include "iceberg/arrow/s3_properties.h"
+#include "iceberg/arrow/s3/s3_properties.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 
 namespace iceberg::arrow {
 
@@ -44,7 +45,7 @@ Status EnsureS3Initialized() {
   static std::once_flag init_flag;
   static ::arrow::Status init_status = ::arrow::Status::OK();
   std::call_once(init_flag, []() {
-    ::arrow::fs::S3GlobalOptions options;
+    auto options = ::arrow::fs::S3GlobalOptions::Defaults();
     init_status = ::arrow::fs::InitializeS3(options);
   });
   if (!init_status.ok()) {
@@ -53,7 +54,7 @@ Status EnsureS3Initialized() {
   }
   return {};
 #else
-  return NotImplemented("Arrow S3 support is not enabled");
+  return NotSupported("Arrow S3 support is not enabled");
 #endif
 }
 
@@ -67,9 +68,9 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
   ::arrow::fs::S3Options options;
 
   // Configure credentials
-  auto access_key_it = properties.find(S3Properties::kAccessKeyId);
-  auto secret_key_it = properties.find(S3Properties::kSecretAccessKey);
-  auto session_token_it = properties.find(S3Properties::kSessionToken);
+  auto access_key_it = properties.find(std::string(S3Properties::kAccessKeyId));
+  auto secret_key_it = properties.find(std::string(S3Properties::kSecretAccessKey));
+  auto session_token_it = properties.find(std::string(S3Properties::kSessionToken));
 
   if (access_key_it != properties.end() && secret_key_it != properties.end()) {
     if (session_token_it != properties.end()) {
@@ -84,13 +85,13 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
   }
 
   // Configure region
-  auto region_it = properties.find(S3Properties::kRegion);
+  auto region_it = properties.find(std::string(S3Properties::kRegion));
   if (region_it != properties.end()) {
     options.region = region_it->second;
   }
 
   // Configure endpoint (for MinIO, LocalStack, etc.)
-  auto endpoint_it = properties.find(S3Properties::kEndpoint);
+  auto endpoint_it = properties.find(std::string(S3Properties::kEndpoint));
   if (endpoint_it != properties.end()) {
     options.endpoint_override = endpoint_it->second;
   } else {
@@ -106,36 +107,30 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
     }
   }
 
-  auto path_style_it = properties.find(S3Properties::kPathStyleAccess);
+  auto path_style_it = properties.find(std::string(S3Properties::kPathStyleAccess));
   if (path_style_it != properties.end() && path_style_it->second == "true") {
     options.force_virtual_addressing = false;
   }
 
   // Configure SSL
-  auto ssl_it = properties.find(S3Properties::kSslEnabled);
+  auto ssl_it = properties.find(std::string(S3Properties::kSslEnabled));
   if (ssl_it != properties.end() && ssl_it->second == "false") {
     options.scheme = "http";
   }
 
   // Configure timeouts
-  auto connect_timeout_it = properties.find(S3Properties::kConnectTimeoutMs);
+  auto connect_timeout_it = properties.find(std::string(S3Properties::kConnectTimeoutMs));
   if (connect_timeout_it != properties.end()) {
-    try {
-      options.connect_timeout = std::stod(connect_timeout_it->second) / 1000.0;
-    } catch (const std::exception& e) {
-      return InvalidArgument("Invalid {}: '{}' ({})", S3Properties::kConnectTimeoutMs,
-                             connect_timeout_it->second, e.what());
-    }
+    ICEBERG_ASSIGN_OR_RAISE(auto timeout_ms,
+                            StringUtils::ParseNumber<double>(connect_timeout_it->second));
+    options.connect_timeout = timeout_ms / 1000.0;
   }
 
-  auto socket_timeout_it = properties.find(S3Properties::kSocketTimeoutMs);
+  auto socket_timeout_it = properties.find(std::string(S3Properties::kSocketTimeoutMs));
   if (socket_timeout_it != properties.end()) {
-    try {
-      options.request_timeout = std::stod(socket_timeout_it->second) / 1000.0;
-    } catch (const std::exception& e) {
-      return InvalidArgument("Invalid {}: '{}' ({})", S3Properties::kSocketTimeoutMs,
-                             socket_timeout_it->second, e.what());
-    }
+    ICEBERG_ASSIGN_OR_RAISE(auto timeout_ms,
+                            StringUtils::ParseNumber<double>(socket_timeout_it->second));
+    options.request_timeout = timeout_ms / 1000.0;
   }
 
   return options;
@@ -145,13 +140,9 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
 }  // namespace
 
 Result<std::unique_ptr<FileIO>> MakeS3FileIO(
-    const std::string& uri,
     const std::unordered_map<std::string, std::string>& properties) {
-  if (!uri.starts_with("s3://")) {
-    return InvalidArgument("S3 URI must start with s3://");
-  }
 #if !ICEBERG_ARROW_HAS_S3
-  return NotImplemented("Arrow S3 support is not enabled");
+  return NotSupported("Arrow S3 support is not enabled");
 #else
   ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
 
@@ -163,4 +154,14 @@ Result<std::unique_ptr<FileIO>> MakeS3FileIO(
 #endif
 }
 
+Status FinalizeS3() {
+#if ICEBERG_ARROW_HAS_S3
+  auto status = ::arrow::fs::FinalizeS3();
+  ICEBERG_ARROW_RETURN_NOT_OK(status);
+  return {};
+#else
+  return NotSupported("Arrow S3 support is not enabled");
+#endif
+}
+
 }  // namespace iceberg::arrow

src/iceberg/arrow/s3_properties.h src/iceberg/arrow/s3/s3_properties.hsrc/iceberg/arrow/s3_properties.h renamed to src/iceberg/arrow/s3/s3_properties.h

@@ -19,7 +19,7 @@
 
 #pragma once
 
-#include <string>
+#include <string_view>
 
 namespace iceberg::arrow {
 
@@ -29,24 +29,26 @@ namespace iceberg::arrow {
 /// via the Arrow filesystem integration, following the Iceberg spec for
 /// S3 configuration properties.
 struct S3Properties {
+  /// S3 URI scheme
+  static constexpr std::string_view kS3Schema = "s3";
   /// AWS access key ID
-  static constexpr const char* kAccessKeyId = "s3.access-key-id";
+  static constexpr std::string_view kAccessKeyId = "s3.access-key-id";
   /// AWS secret access key
-  static constexpr const char* kSecretAccessKey = "s3.secret-access-key";
+  static constexpr std::string_view kSecretAccessKey = "s3.secret-access-key";
   /// AWS session token (for temporary credentials)
-  static constexpr const char* kSessionToken = "s3.session-token";
+  static constexpr std::string_view kSessionToken = "s3.session-token";
   /// AWS region
-  static constexpr const char* kRegion = "s3.region";
+  static constexpr std::string_view kRegion = "s3.region";
   /// Custom endpoint override (for MinIO, LocalStack, etc.)
-  static constexpr const char* kEndpoint = "s3.endpoint";
+  static constexpr std::string_view kEndpoint = "s3.endpoint";
   /// Whether to use path-style access (needed for MinIO)
-  static constexpr const char* kPathStyleAccess = "s3.path-style-access";
+  static constexpr std::string_view kPathStyleAccess = "s3.path-style-access";
   /// Whether SSL is enabled
-  static constexpr const char* kSslEnabled = "s3.ssl.enabled";
+  static constexpr std::string_view kSslEnabled = "s3.ssl.enabled";
   /// Connection timeout in milliseconds
-  static constexpr const char* kConnectTimeoutMs = "s3.connect-timeout-ms";
+  static constexpr std::string_view kConnectTimeoutMs = "s3.connect-timeout-ms";
   /// Socket timeout in milliseconds
-  static constexpr const char* kSocketTimeoutMs = "s3.socket-timeout-ms";
+  static constexpr std::string_view kSocketTimeoutMs = "s3.socket-timeout-ms";
 };
 
 }  // namespace iceberg::arrow

src/iceberg/catalog/rest/catalog_properties.cc

@@ -61,4 +61,12 @@ Result<SnapshotMode> RestCatalogProperties::SnapshotLoadingMode() const {
   }
 }
 
+void RestCatalogProperties::SetFileIO(std::shared_ptr<FileIO> file_io) {
+  file_io_ = std::move(file_io);
+}
+
+const std::shared_ptr<FileIO>& RestCatalogProperties::GetFileIO() const {
+  return file_io_;
+}
+
 }  // namespace iceberg::rest