address some issues

Author: wgtmac

ci/scripts/build_iceberg.sh

@@ -36,6 +36,7 @@ CMAKE_ARGS=(
     "-DCMAKE_INSTALL_PREFIX=${CMAKE_INSTALL_PREFIX:-${ICEBERG_HOME}}"
     "-DICEBERG_BUILD_STATIC=ON"
     "-DICEBERG_BUILD_SHARED=ON"
+    "-DICEBERG_S3=ON"
     "-DICEBERG_BUILD_REST_INTEGRATION_TESTS=${build_rest_integration_test}"
 )
 

ci/scripts/start_minio.sh

@@ -35,6 +35,11 @@ wait_for_minio() {
     fi
     sleep 1
   done
+  echo "MinIO did not become ready after 30 seconds." >&2
+  echo "Endpoint: ${MINIO_ENDPOINT}" >&2
+  if command -v docker >/dev/null 2>&1; then
+    docker logs "${MINIO_CONTAINER_NAME}" 2>&1 || true
+  fi
   return 1
 }
 

src/iceberg/CMakeLists.txt

@@ -107,7 +107,6 @@ 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
@@ -251,14 +250,17 @@ if(ICEBERG_BUILD_BUNDLE)
                   OUTPUTS
                   ICEBERG_BUNDLE_LIBRARIES)
 
-  if(ICEBERG_S3)
-    foreach(target iceberg_bundle_static iceberg_bundle_shared)
-      if(TARGET ${target})
+  foreach(target iceberg_bundle_static iceberg_bundle_shared)
+    if(TARGET ${target})
+      if(ICEBERG_S3)
         target_compile_definitions(${target}
                                    PUBLIC "$<BUILD_INTERFACE:ICEBERG_S3_ENABLED=1>")
+      else()
+        target_compile_definitions(${target}
+                                   PUBLIC "$<BUILD_INTERFACE:ICEBERG_S3_ENABLED=0>")
       endif()
-    endforeach()
-  endif()
+    endif()
+  endforeach()
 
   add_subdirectory(arrow)
   add_subdirectory(avro)

src/iceberg/arrow/arrow_file_io.h

@@ -36,7 +36,7 @@ ICEBERG_BUNDLE_EXPORT std::unique_ptr<FileIO> MakeLocalFileIO();
 /// \brief Create an S3 FileIO backed by Arrow's S3FileSystem.
 ///
 /// This function initializes the S3 subsystem if not already initialized (thread-safe).
-/// The S3 initialization is done once per process using std::call_once.
+/// The S3 initialization is cached once per process.
 ///
 /// \param properties Configuration properties for S3 access. See S3Properties
 ///        for available keys (credentials, region, endpoint, timeouts, etc.).

src/iceberg/arrow/file_io_register.cc

@@ -18,32 +18,44 @@
 #include "iceberg/arrow/file_io_register.h"
 
 #include <mutex>
+#include <string>
 
 #include "iceberg/arrow/arrow_file_io.h"
-#include "iceberg/arrow/arrow_fs_file_io_internal.h"
 #include "iceberg/file_io_registry.h"
-#include "iceberg/util/macros.h"
 
 namespace iceberg::arrow {
 
+namespace {
+
 void RegisterLocalFileIO() {
-  static std::once_flag flag;
-  std::call_once(flag, []() {
-    FileIORegistry::Register(
-        std::string(FileIORegistry::kArrowLocalFileIO),
-        [](const std::unordered_map<std::string, std::string>& /*properties*/)
-            -> Result<std::unique_ptr<FileIO>> { return MakeLocalFileIO(); });
-  });
+  FileIORegistry::Register(
+      std::string(FileIORegistry::kArrowLocalFileIO),
+      [](const std::unordered_map<std::string, std::string>& /*properties*/)
+          -> Result<std::unique_ptr<FileIO>> { return MakeLocalFileIO(); });
 }
 
 void RegisterS3FileIO() {
+#if ICEBERG_S3_ENABLED
+  FileIORegistry::Register(
+      std::string(FileIORegistry::kArrowS3FileIO),
+      [](const std::unordered_map<std::string, std::string>& properties)
+          -> Result<std::unique_ptr<FileIO>> { return MakeS3FileIO(properties); });
+#endif
+}
+
+}  // namespace
+
+void EnsureArrowFileIOsRegistered() {
   static std::once_flag flag;
   std::call_once(flag, []() {
-    FileIORegistry::Register(
-        std::string(FileIORegistry::kArrowS3FileIO),
-        [](const std::unordered_map<std::string, std::string>& properties)
-            -> Result<std::unique_ptr<FileIO>> { return MakeS3FileIO(properties); });
+    RegisterLocalFileIO();
+    RegisterS3FileIO();
   });
 }
 
+[[maybe_unused]] const bool kArrowFileIOsRegistered = []() {
+  EnsureArrowFileIOsRegistered();
+  return true;
+}();
+
 }  // namespace iceberg::arrow

src/iceberg/arrow/file_io_register.h

@@ -26,10 +26,9 @@
 
 namespace iceberg::arrow {
 
-/// \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();
+/// \brief Register built-in Arrow FileIO implementations into the FileIORegistry.
+///
+/// This operation is idempotent and safe to call multiple times.
+ICEBERG_BUNDLE_EXPORT void EnsureArrowFileIOsRegistered();
 
 }  // namespace iceberg::arrow

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -18,15 +18,13 @@
  */
 
 #include <cstdlib>
-#include <mutex>
+#include <optional>
 #include <string>
+#include <string_view>
 
 #include <arrow/filesystem/filesystem.h>
-#ifdef ICEBERG_S3_ENABLED
+#if ICEBERG_S3_ENABLED
 #  include <arrow/filesystem/s3fs.h>
-#  define ICEBERG_ARROW_HAS_S3 1
-#else
-#  define ICEBERG_ARROW_HAS_S3 0
 #endif
 
 #include "iceberg/arrow/arrow_file_io.h"
@@ -40,55 +38,77 @@ namespace iceberg::arrow {
 
 namespace {
 
-#if ICEBERG_ARROW_HAS_S3
+#if ICEBERG_S3_ENABLED
+const std::string* FindProperty(
+    const std::unordered_map<std::string, std::string>& properties,
+    std::string_view key) {
+  auto it = properties.find(std::string(key));
+  return it == properties.end() ? nullptr : &it->second;
+}
+
+Result<std::optional<bool>> ParseOptionalBool(
+    const std::unordered_map<std::string, std::string>& properties,
+    std::string_view key) {
+  const auto* value = FindProperty(properties, key);
+  if (value == nullptr) {
+    return std::nullopt;
+  }
+  if (*value == "true") {
+    return true;
+  }
+  if (*value == "false") {
+    return false;
+  }
+  return InvalidArgument(R"("{}" must be "true" or "false")", key);
+}
+
 Status EnsureS3Initialized() {
-  static std::once_flag init_flag;
-  static ::arrow::Status init_status = ::arrow::Status::OK();
-  std::call_once(init_flag, []() {
+  static const ::arrow::Status init_status = []() {
     auto options = ::arrow::fs::S3GlobalOptions::Defaults();
-    init_status = ::arrow::fs::InitializeS3(options);
-  });
+    return ::arrow::fs::InitializeS3(options);
+  }();
   if (!init_status.ok()) {
     return std::unexpected(Error{.kind = ::iceberg::arrow::ToErrorKind(init_status),
                                  .message = init_status.ToString()});
   }
   return {};
 }
+
 /// \brief Configure S3Options from a properties map.
 ///
 /// \param properties The configuration properties map.
 /// \return Configured S3Options.
 Result<::arrow::fs::S3Options> ConfigureS3Options(
     const std::unordered_map<std::string, std::string>& properties) {
-  ::arrow::fs::S3Options options;
+  auto options = ::arrow::fs::S3Options::Defaults();
 
   // Configure credentials
-  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()) {
-      options.ConfigureAccessKey(access_key_it->second, secret_key_it->second,
-                                 session_token_it->second);
+  const auto* access_key = FindProperty(properties, S3Properties::kAccessKeyId);
+  const auto* secret_key = FindProperty(properties, S3Properties::kSecretAccessKey);
+  const auto* session_token = FindProperty(properties, S3Properties::kSessionToken);
+
+  if ((access_key == nullptr) != (secret_key == nullptr)) {
+    return InvalidArgument(
+        "S3 client access key ID and secret access key must be set at the same time");
+  }
+  if (access_key != nullptr) {
+    if (session_token != nullptr) {
+      options.ConfigureAccessKey(*access_key, *secret_key, *session_token);
     } else {
-      options.ConfigureAccessKey(access_key_it->second, secret_key_it->second);
+      options.ConfigureAccessKey(*access_key, *secret_key);
     }
-  } else {
-    // Use default credential chain (environment, instance profile, etc.)
-    options.ConfigureDefaultCredentials();
   }
 
   // Configure region
-  auto region_it = properties.find(std::string(S3Properties::kRegion));
-  if (region_it != properties.end()) {
-    options.region = region_it->second;
+  if (const auto* region = FindProperty(properties, S3Properties::kRegion);
+      region != nullptr) {
+    options.region = *region;
   }
 
   // Configure endpoint (for MinIO, LocalStack, etc.)
-  auto endpoint_it = properties.find(std::string(S3Properties::kEndpoint));
-  if (endpoint_it != properties.end()) {
-    options.endpoint_override = endpoint_it->second;
+  if (const auto* endpoint = FindProperty(properties, S3Properties::kEndpoint);
+      endpoint != nullptr) {
+    options.endpoint_override = *endpoint;
   } else {
     // Fall back to AWS standard environment variables for endpoint override
     const char* s3_endpoint_env = std::getenv("AWS_ENDPOINT_URL_S3");
@@ -102,14 +122,16 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
     }
   }
 
-  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;
+  ICEBERG_ASSIGN_OR_RAISE(const auto path_style_access,
+                          ParseOptionalBool(properties, S3Properties::kPathStyleAccess));
+  if (path_style_access.has_value()) {
+    options.force_virtual_addressing = !*path_style_access;
   }
 
   // Configure SSL
-  auto ssl_it = properties.find(std::string(S3Properties::kSslEnabled));
-  if (ssl_it != properties.end() && ssl_it->second == "false") {
+  ICEBERG_ASSIGN_OR_RAISE(const auto ssl_enabled,
+                          ParseOptionalBool(properties, S3Properties::kSslEnabled));
+  if (ssl_enabled.has_value() && !*ssl_enabled) {
     options.scheme = "http";
   }
 
@@ -136,21 +158,21 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
 
 Result<std::unique_ptr<FileIO>> MakeS3FileIO(
     const std::unordered_map<std::string, std::string>& properties) {
-#if !ICEBERG_ARROW_HAS_S3
-  return NotSupported("Arrow S3 support is not enabled");
-#else
+#if ICEBERG_S3_ENABLED
   ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
 
   // Configure S3 options from properties (uses default credentials if empty)
   ICEBERG_ASSIGN_OR_RAISE(auto options, ConfigureS3Options(properties));
   ICEBERG_ARROW_ASSIGN_OR_RETURN(auto fs, ::arrow::fs::S3FileSystem::Make(options));
 
   return std::make_unique<ArrowFileSystemFileIO>(std::move(fs));
+#else
+  return NotSupported("Arrow S3 support is not enabled");
 #endif
 }
 
 Status FinalizeS3() {
-#if ICEBERG_ARROW_HAS_S3
+#if ICEBERG_S3_ENABLED
   auto status = ::arrow::fs::FinalizeS3();
   ICEBERG_ARROW_RETURN_NOT_OK(status);
   return {};

src/iceberg/catalog/rest/CMakeLists.txt

@@ -30,6 +30,7 @@ set(ICEBERG_REST_SOURCES
     json_serde.cc
     resource_paths.cc
     rest_catalog.cc
+    rest_file_io.cc
     rest_util.cc
     types.cc)
 

src/iceberg/catalog/rest/catalog_properties.cc

@@ -61,12 +61,4 @@ 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

src/iceberg/catalog/rest/catalog_properties.h

@@ -19,12 +19,10 @@
 
 #pragma once
 
-#include <memory>
 #include <string>
 #include <unordered_map>
 
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
-#include "iceberg/file_io.h"
 #include "iceberg/result.h"
 #include "iceberg/util/config.h"
 
@@ -49,6 +47,8 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   inline static Entry<std::string> kName{"name", ""};
   /// \brief The warehouse path.
   inline static Entry<std::string> kWarehouse{"warehouse", ""};
+  /// \brief The FileIO implementation name.
+  inline static Entry<std::string> kIOImpl{"io-impl", ""};
   /// \brief The optional prefix for REST API paths.
   inline static Entry<std::string> kPrefix{"prefix", ""};
   /// \brief The encoded separator used to join namespace levels in REST paths.
@@ -77,16 +77,6 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   ///         "REFS", or an error if the value is invalid. Parsing is
   ///         case-insensitive to match Java behavior.
   Result<SnapshotMode> SnapshotLoadingMode() const;
-
-  /// \brief Set a custom FileIO instance. If not set, FileIO will be
-  /// auto-detected from warehouse URI scheme or "io-impl" property.
-  void SetFileIO(std::shared_ptr<FileIO> file_io);
-
-  /// \brief Get the configured FileIO instance (may be nullptr).
-  const std::shared_ptr<FileIO>& GetFileIO() const;
-
- private:
-  std::shared_ptr<FileIO> file_io_;
 };
 
 }  // namespace iceberg::rest