fix: address PR apache#548 review feedback for S3 FileIO integration

Key changes based on reviewer (wgtmac) feedback:

1. Add ICEBERG_S3 CMake option to conditionally enable ARROW_S3,
   replacing the unconditional `set(ARROW_S3 ON)`.

2. Replace `#if __has_include(<arrow/filesystem/s3fs.h>)` with
   `#ifdef ICEBERG_HAVE_S3` compile definition controlled by CMake.

3. Remove ArrowUriFileIO class - reuse existing ArrowFileSystemFileIO
   by wrapping Arrow S3FileSystem created via Arrow API.

4. Create missing files:
   - s3_properties.h: S3 configuration property key constants
   - file_io_registry.h/.cc: FileIO factory registry (Register/Load)
   - file_io_register.cc: Arrow FileIO factory registration
   - file_io_util.h/.cc: Reusable FileIO creation utility
   - file_io_registry_test.cc: Unit tests for the registry

5. Extract FileIO creation logic from rest_catalog.cc into
   iceberg/util/file_io_util.h for reusability.

6. Fix code issues:
   - Implement path-style access (force_virtual_addressing = false)
   - Add timeout value validation (non-negative check)
   - Replace rfind("s3://", 0) with starts_with("s3://")
   - Fix format string bug in rest_catalog_test.cc

7. Update CI workflow and build script to pass ICEBERG_S3=ON.

https://claude.ai/code/session_01GzV7A8VoYyWUN8QdtqgiMq

Author: claude

.github/workflows/test.yml

@@ -64,7 +64,7 @@ jobs:
         env:
           CC: gcc-14
           CXX: g++-14
-        run: ci/scripts/build_iceberg.sh $(pwd) ON
+        run: ci/scripts/build_iceberg.sh $(pwd) ON OFF ON
       - name: Build Example
         shell: bash
         env:
@@ -92,7 +92,7 @@ jobs:
         run: bash ci/scripts/start_minio.sh
       - name: Build Iceberg
         shell: bash
-        run: ci/scripts/build_iceberg.sh $(pwd)
+        run: ci/scripts/build_iceberg.sh $(pwd) OFF OFF ON
       - name: Build Example
         shell: bash
         run: ci/scripts/build_example.sh $(pwd)/example
@@ -127,7 +127,7 @@ jobs:
           SCCACHE_GHA_ENABLED: "true"
         run: |
           call "C:\Program Files\Microsoft Visual Studio\2022\Enterprise\VC\Auxiliary\Build\vcvarsall.bat" x64
-          bash -c "ci/scripts/build_iceberg.sh $(pwd) OFF ON"
+          bash -c "ci/scripts/build_iceberg.sh $(pwd) OFF ON ON"
           sccache --show-stats
       - name: Build Example
         shell: cmd

CMakeLists.txt

@@ -45,6 +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 "Enable S3 support via Arrow S3" OFF)
 option(ICEBERG_ENABLE_ASAN "Enable Address Sanitizer" OFF)
 option(ICEBERG_ENABLE_UBSAN "Enable Undefined Behavior Sanitizer" OFF)
 

ci/scripts/build_iceberg.sh

@@ -23,6 +23,7 @@ source_dir=${1}
 build_dir=${1}/build
 build_rest_integration_test=${2:-OFF}
 build_enable_sccache=${3:-OFF}
+build_s3=${4:-OFF}
 
 mkdir ${build_dir}
 pushd ${build_dir}
@@ -37,6 +38,7 @@ CMAKE_ARGS=(
     "-DICEBERG_BUILD_STATIC=ON"
     "-DICEBERG_BUILD_SHARED=ON"
     "-DICEBERG_BUILD_REST_INTEGRATION_TESTS=${build_rest_integration_test}"
+    "-DICEBERG_S3=${build_s3}"
 )
 
 if is_windows; then

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -87,7 +87,7 @@ function(resolve_arrow_dependency)
   # Work around undefined symbol: arrow::ipc::ReadSchema(arrow::io::InputStream*, arrow::ipc::DictionaryMemo*)
   set(ARROW_IPC ON)
   set(ARROW_FILESYSTEM ON)
-  set(ARROW_S3 ON)
+  set(ARROW_S3 ${ICEBERG_S3})
   set(ARROW_JSON ON)
   set(ARROW_PARQUET ON)
   set(ARROW_SIMD_LEVEL "NONE")

src/iceberg/CMakeLists.txt

@@ -101,6 +101,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
@@ -239,6 +240,15 @@ if(ICEBERG_BUILD_BUNDLE)
                   OUTPUTS
                   ICEBERG_BUNDLE_LIBRARIES)
 
+  if(ICEBERG_S3)
+    if(TARGET iceberg_bundle_static)
+      target_compile_definitions(iceberg_bundle_static PUBLIC ICEBERG_HAVE_S3=1)
+    endif()
+    if(TARGET iceberg_bundle_shared)
+      target_compile_definitions(iceberg_bundle_shared PUBLIC ICEBERG_HAVE_S3=1)
+    endif()
+  endif()
+
   add_subdirectory(arrow)
   add_subdirectory(avro)
   add_subdirectory(parquet)

src/iceberg/arrow/arrow_s3_file_io.cc

@@ -23,11 +23,8 @@
 
 #include <arrow/filesystem/filesystem.h>
 #include <arrow/filesystem/localfs.h>
-#if __has_include(<arrow/filesystem/s3fs.h>)
+#ifdef ICEBERG_HAVE_S3
 #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,10 +37,10 @@ namespace iceberg::arrow {
 
 namespace {
 
-bool IsS3Uri(std::string_view uri) { return uri.rfind("s3://", 0) == 0; }
+bool IsS3Uri(std::string_view uri) { return uri.starts_with("s3://"); }
 
 Status EnsureS3Initialized() {
-#if ICEBERG_ARROW_HAS_S3
+#ifdef ICEBERG_HAVE_S3
   static std::once_flag init_flag;
   static ::arrow::Status init_status = ::arrow::Status::OK();
   std::call_once(init_flag, []() {
@@ -64,7 +61,7 @@ Status EnsureS3Initialized() {
 #endif
 }
 
-#if ICEBERG_ARROW_HAS_S3
+#ifdef ICEBERG_HAVE_S3
 /// \brief Configure S3Options from a properties map.
 ///
 /// \param properties The configuration properties map.
@@ -104,9 +101,8 @@ ::arrow::fs::S3Options ConfigureS3Options(
 
   // Configure path-style access (needed for MinIO)
   auto path_style_it = properties.find(S3Properties::kPathStyleAccess);
-  if (path_style_it != properties.end()) {
-    // Arrow's S3 path-style is controlled via endpoint scheme
-    // For path-style access, we need to ensure the endpoint is properly configured
+  if (path_style_it != properties.end() && path_style_it->second == "true") {
+    options.force_virtual_addressing = false;
   }
 
   // Configure SSL
@@ -118,12 +114,18 @@ ::arrow::fs::S3Options ConfigureS3Options(
   // Configure timeouts
   auto connect_timeout_it = properties.find(S3Properties::kConnectTimeoutMs);
   if (connect_timeout_it != properties.end()) {
-    options.connect_timeout = std::stod(connect_timeout_it->second) / 1000.0;
+    double timeout_ms = std::stod(connect_timeout_it->second);
+    if (timeout_ms >= 0) {
+      options.connect_timeout = timeout_ms / 1000.0;
+    }
   }
 
   auto socket_timeout_it = properties.find(S3Properties::kSocketTimeoutMs);
   if (socket_timeout_it != properties.end()) {
-    options.request_timeout = std::stod(socket_timeout_it->second) / 1000.0;
+    double timeout_ms = std::stod(socket_timeout_it->second);
+    if (timeout_ms >= 0) {
+      options.request_timeout = timeout_ms / 1000.0;
+    }
   }
 
   return options;
@@ -141,83 +143,19 @@ Result<std::shared_ptr<::arrow::fs::FileSystem>> MakeS3FileSystem(
 }
 #endif
 
-Result<std::shared_ptr<::arrow::fs::FileSystem>> ResolveFileSystemFromUri(
-    const std::string& uri, std::string* out_path) {
-  if (IsS3Uri(uri)) {
-    ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
-  }
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto fs, ::arrow::fs::FileSystemFromUri(uri, out_path));
-  return fs;
-}
-
-/// \brief ArrowUriFileIO resolves FileSystem from URI for each operation.
-///
-/// This implementation is thread-safe as it creates a new FileSystem instance
-/// for each operation. However, it may be less efficient than caching the
-/// FileSystem. S3 initialization is done once per process.
-class ArrowUriFileIO : public FileIO {
- public:
-  Result<std::string> ReadFile(const std::string& file_location,
-                               std::optional<size_t> length) override {
-    std::string path;
-    ICEBERG_ASSIGN_OR_RAISE(auto fs, ResolveFileSystemFromUri(file_location, &path));
-    ::arrow::fs::FileInfo file_info(path);
-    if (length.has_value()) {
-      file_info.set_size(length.value());
-    }
-    std::string content;
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, fs->OpenInputFile(file_info));
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file_size, file->GetSize());
-
-    content.resize(file_size);
-    size_t remain = file_size;
-    size_t offset = 0;
-    while (remain > 0) {
-      size_t read_length = std::min(remain, static_cast<size_t>(1024 * 1024));
-      ICEBERG_ARROW_ASSIGN_OR_RETURN(
-          auto read_bytes,
-          file->Read(read_length, reinterpret_cast<uint8_t*>(&content[offset])));
-      remain -= read_bytes;
-      offset += read_bytes;
-    }
-
-    return content;
-  }
-
-  Status WriteFile(const std::string& file_location,
-                   std::string_view content) override {
-    std::string path;
-    ICEBERG_ASSIGN_OR_RAISE(auto fs, ResolveFileSystemFromUri(file_location, &path));
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, fs->OpenOutputStream(path));
-    ICEBERG_ARROW_RETURN_NOT_OK(file->Write(content.data(), content.size()));
-    ICEBERG_ARROW_RETURN_NOT_OK(file->Flush());
-    ICEBERG_ARROW_RETURN_NOT_OK(file->Close());
-    return {};
-  }
-
-  Status DeleteFile(const std::string& file_location) override {
-    std::string path;
-    ICEBERG_ASSIGN_OR_RAISE(auto fs, ResolveFileSystemFromUri(file_location, &path));
-    ICEBERG_ARROW_RETURN_NOT_OK(fs->DeleteFile(path));
-    return {};
-  }
-};
-
 }  // namespace
 
 Result<std::unique_ptr<FileIO>> MakeS3FileIO(const std::string& uri) {
   if (!IsS3Uri(uri)) {
     return InvalidArgument("S3 URI must start with s3://");
   }
-#if !ICEBERG_ARROW_HAS_S3
+#ifndef ICEBERG_HAVE_S3
   return NotImplemented("Arrow S3 support is not enabled");
 #else
-  // Validate that S3 can be initialized and the URI is valid
+  ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
   std::string path;
-  ICEBERG_ASSIGN_OR_RAISE(auto fs, ResolveFileSystemFromUri(uri, &path));
-  (void)path;
-  (void)fs;
-  return std::make_unique<ArrowUriFileIO>();
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto fs, ::arrow::fs::FileSystemFromUri(uri, &path));
+  return std::make_unique<ArrowFileSystemFileIO>(std::move(fs));
 #endif
 }
 
@@ -227,7 +165,7 @@ Result<std::unique_ptr<FileIO>> MakeS3FileIO(
   if (!IsS3Uri(uri)) {
     return InvalidArgument("S3 URI must start with s3://");
   }
-#if !ICEBERG_ARROW_HAS_S3
+#ifndef ICEBERG_HAVE_S3
   return NotImplemented("Arrow S3 support is not enabled");
 #else
   // If properties are empty, use the simple URI-based resolution

src/iceberg/arrow/file_io_register.cc

@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/arrow/arrow_file_io.h"
+#include "iceberg/file_io_registry.h"
+
+namespace iceberg::arrow {
+
+namespace {
+
+struct ArrowFileIORegistrar {
+  ArrowFileIORegistrar() {
+    FileIORegistry::Register(
+        std::string(FileIORegistry::kArrowLocalFileIO),
+        [](const std::string& /*warehouse*/,
+           const std::unordered_map<std::string, std::string>& /*properties*/)
+            -> Result<std::shared_ptr<FileIO>> { return MakeLocalFileIO(); });
+
+    FileIORegistry::Register(
+        std::string(FileIORegistry::kArrowS3FileIO),
+        [](const std::string& warehouse,
+           const std::unordered_map<std::string, std::string>& properties)
+            -> Result<std::shared_ptr<FileIO>> {
+          auto result = MakeS3FileIO(warehouse, properties);
+          if (!result.has_value()) {
+            return std::unexpected(result.error());
+          }
+          return std::shared_ptr<FileIO>(std::move(result.value()));
+        });
+  }
+};
+
+// Static initialization triggers registration at program startup
+static ArrowFileIORegistrar registrar;
+
+}  // namespace
+
+}  // namespace iceberg::arrow

src/iceberg/arrow/s3_properties.h

@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <string_view>
+
+namespace iceberg::arrow {
+
+/// \brief Reserved property keys for S3 FileIO configuration.
+///
+/// These properties are used to configure S3 access when creating an S3-backed FileIO
+/// instance. They map to the corresponding Arrow S3Options fields.
+struct S3Properties {
+  static constexpr std::string_view kAccessKeyId = "s3.access-key-id";
+  static constexpr std::string_view kSecretAccessKey = "s3.secret-access-key";
+  static constexpr std::string_view kSessionToken = "s3.session-token";
+  static constexpr std::string_view kRegion = "s3.region";
+  static constexpr std::string_view kEndpoint = "s3.endpoint";
+  static constexpr std::string_view kPathStyleAccess = "s3.path-style-access";
+  static constexpr std::string_view kSslEnabled = "s3.ssl.enabled";
+  static constexpr std::string_view kConnectTimeoutMs = "s3.connect-timeout-ms";
+  static constexpr std::string_view kSocketTimeoutMs = "s3.socket-timeout-ms";
+};
+
+}  // namespace iceberg::arrow

src/iceberg/catalog/rest/rest_catalog.cc

@@ -27,7 +27,6 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/catalog/rest/catalog_properties.h"
-#include "iceberg/file_io_registry.h"
 #include "iceberg/catalog/rest/constant.h"
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/error_handlers.h"
@@ -46,6 +45,7 @@
 #include "iceberg/table_requirement.h"
 #include "iceberg/table_update.h"
 #include "iceberg/transaction.h"
+#include "iceberg/util/file_io_util.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -147,33 +147,11 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
 
 Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
     const RestCatalogProperties& config) {
-  // Get warehouse location to determine the appropriate FileIO type
   auto warehouse = config.Get(RestCatalogProperties::kWarehouse);
-  if (warehouse.empty()) {
-    return InvalidArgument(
-        "Warehouse location is required when FileIO is not explicitly provided. "
-        "Set the 'warehouse' property to an S3 URI (s3://...) or local path.");
-  }
-
-  // Check for user-specified io-impl property
-  auto io_impl = config.configs().find(FileIOProperties::kImpl);
-  std::string impl_name;
-
-  if (io_impl != config.configs().end() && !io_impl->second.empty()) {
-    // User specified a custom io-impl
-    impl_name = io_impl->second;
-  } else {
-    // Use default based on warehouse URI scheme
-    if (warehouse.rfind("s3://", 0) == 0) {
-      impl_name = FileIORegistry::kArrowS3FileIO;
-    } else {
-      impl_name = FileIORegistry::kArrowLocalFileIO;
-    }
-  }
 
-  // Load FileIO from registry
+  // Delegate FileIO creation to the utility function
   ICEBERG_ASSIGN_OR_RAISE(auto file_io,
-                          FileIORegistry::Load(impl_name, warehouse, config.configs()));
+                          CreateFileIOFromProperties(warehouse, config.configs()));
 
   // Call the main Make method with the created FileIO
   return Make(config, std::move(file_io));