fix

Author: MisterRaindrop

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 "Build with S3 support" ON)
 option(ICEBERG_ENABLE_ASAN "Enable Address Sanitizer" OFF)
 option(ICEBERG_ENABLE_UBSAN "Enable Undefined Behavior Sanitizer" OFF)
 

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -102,7 +102,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

@@ -244,6 +244,15 @@ if(ICEBERG_BUILD_BUNDLE)
                   OUTPUTS
                   ICEBERG_BUNDLE_LIBRARIES)
 
+  if(ICEBERG_S3)
+    foreach(target iceberg_bundle_static iceberg_bundle_shared)
+      if(TARGET ${target})
+        target_compile_definitions(${target} PUBLIC
+            "$<BUILD_INTERFACE:ICEBERG_S3_ENABLED=1>")
+      endif()
+    endforeach()
+  endif()
+
   add_subdirectory(arrow)
   add_subdirectory(avro)
   add_subdirectory(parquet)

src/iceberg/arrow/arrow_fs_file_io.cc

@@ -25,13 +25,24 @@
 #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/util/macros.h"
 
 namespace iceberg::arrow {
 
+Result<std::string> ArrowFileSystemFileIO::ResolvePath(
+    const std::string& file_location) {
+  if (file_location.find("://") != std::string::npos) {
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto path, arrow_fs_->PathFromUri(file_location));
+    return path;
+  }
+  return file_location;
+}
+
 /// \brief Read the content of the file at the given location.
 Result<std::string> ArrowFileSystemFileIO::ReadFile(const std::string& file_location,
                                                     std::optional<size_t> length) {
-  ::arrow::fs::FileInfo file_info(file_location);
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
+  ::arrow::fs::FileInfo file_info(path);
   if (length.has_value()) {
     file_info.set_size(length.value());
   }
@@ -47,6 +58,10 @@ Result<std::string> ArrowFileSystemFileIO::ReadFile(const std::string& file_loca
     ICEBERG_ARROW_ASSIGN_OR_RETURN(
         auto read_bytes,
         file->Read(read_length, reinterpret_cast<uint8_t*>(&content[offset])));
+    if (read_bytes == 0) {
+      return IOError("Unexpected EOF reading {}: got {} of {} bytes", file_location,
+                     offset, file_size);
+    }
     remain -= read_bytes;
     offset += read_bytes;
   }
@@ -57,7 +72,8 @@ Result<std::string> ArrowFileSystemFileIO::ReadFile(const std::string& file_loca
 /// \brief Write the given content to the file at the given location.
 Status ArrowFileSystemFileIO::WriteFile(const std::string& file_location,
                                         std::string_view content) {
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenOutputStream(file_location));
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, arrow_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());
@@ -66,7 +82,8 @@ Status ArrowFileSystemFileIO::WriteFile(const std::string& file_location,
 
 /// \brief Delete a file at the given location.
 Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) {
-  ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFile(file_location));
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
+  ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFile(path));
   return {};
 }
 

src/iceberg/arrow/arrow_fs_file_io_internal.h

@@ -56,6 +56,9 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
 
  private:
+  /// \brief Resolve a file location to a filesystem path.
+  Result<std::string> ResolvePath(const std::string& file_location);
+
   std::shared_ptr<::arrow::fs::FileSystem> arrow_fs_;
 };
 

src/iceberg/arrow/arrow_s3_file_io.cc

@@ -17,13 +17,11 @@
  * under the License.
  */
 
-#include <cstdlib>
 #include <mutex>
-#include <string_view>
+#include <stdexcept>
 
 #include <arrow/filesystem/filesystem.h>
-#include <arrow/filesystem/localfs.h>
-#if __has_include(<arrow/filesystem/s3fs.h>)
+#ifdef ICEBERG_S3_ENABLED
 #include <arrow/filesystem/s3fs.h>
 #define ICEBERG_ARROW_HAS_S3 1
 #else
@@ -40,18 +38,13 @@ namespace iceberg::arrow {
 
 namespace {
 
-bool IsS3Uri(std::string_view uri) { return uri.rfind("s3://", 0) == 0; }
-
 Status EnsureS3Initialized() {
 #if ICEBERG_ARROW_HAS_S3
   static std::once_flag init_flag;
   static ::arrow::Status init_status = ::arrow::Status::OK();
   std::call_once(init_flag, []() {
     ::arrow::fs::S3GlobalOptions options;
     init_status = ::arrow::fs::InitializeS3(options);
-    if (init_status.ok()) {
-      std::atexit([]() { (void)::arrow::fs::FinalizeS3(); });
-    }
   });
   if (!init_status.ok()) {
     return std::unexpected<Error>{
@@ -69,7 +62,7 @@ Status EnsureS3Initialized() {
 ///
 /// \param properties The configuration properties map.
 /// \return Configured S3Options.
-::arrow::fs::S3Options ConfigureS3Options(
+Result<::arrow::fs::S3Options> ConfigureS3Options(
     const std::unordered_map<std::string, std::string>& properties) {
   ::arrow::fs::S3Options options;
 
@@ -102,11 +95,9 @@ ::arrow::fs::S3Options ConfigureS3Options(
     options.endpoint_override = endpoint_it->second;
   }
 
-  // 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,117 +109,45 @@ ::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;
+    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());
+    }
   }
 
   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;
+    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());
+    }
   }
 
   return options;
 }
-
-/// \brief Create an S3 FileSystem with the given options.
-///
-/// \param options The S3Options to use.
-/// \return A shared_ptr to the S3FileSystem, or an error.
-Result<std::shared_ptr<::arrow::fs::FileSystem>> MakeS3FileSystem(
-    const ::arrow::fs::S3Options& options) {
-  ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto fs, ::arrow::fs::S3FileSystem::Make(options));
-  return fs;
-}
 #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,
     const std::unordered_map<std::string, std::string>& properties) {
-  if (!IsS3Uri(uri)) {
+  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");
 #else
-  // If properties are empty, use the simple URI-based resolution
-  if (properties.empty()) {
-    // Validate that S3 can be initialized and the URI is valid
-    std::string path;
-    ICEBERG_ASSIGN_OR_RAISE(auto fs, ResolveFileSystemFromUri(uri, &path));
-    (void)path;
-    (void)fs;
-    return std::make_unique<ArrowUriFileIO>();
-  }
+  ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
 
-  // Create S3FileSystem with explicit configuration
-  auto options = ConfigureS3Options(properties);
-  ICEBERG_ASSIGN_OR_RAISE(auto fs, MakeS3FileSystem(options));
+  // 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 ArrowFileSystemFileIO with the configured S3 filesystem
   return std::make_unique<ArrowFileSystemFileIO>(std::move(fs));
 #endif
 }

src/iceberg/catalog/rest/rest_catalog.cc

@@ -194,7 +194,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
     impl_name = io_impl->second;
   } else {
     // Use default based on warehouse URI scheme
-    if (warehouse.rfind("s3://", 0) == 0) {
+    if (warehouse.starts_with("s3://")) {
       impl_name = FileIORegistry::kArrowS3FileIO;
     } else {
       impl_name = FileIORegistry::kArrowLocalFileIO;

src/iceberg/test/arrow_s3_file_io_test.cc

@@ -21,9 +21,6 @@
 #include <string>
 #include <unordered_map>
 
-#if __has_include(<arrow/filesystem/s3fs.h>)
-#include <arrow/filesystem/s3fs.h>
-#endif
 #include <gtest/gtest.h>
 
 #include "iceberg/arrow/arrow_file_io.h"
@@ -32,22 +29,13 @@
 
 namespace iceberg::arrow {
 
-#if __has_include(<arrow/filesystem/s3fs.h>)
-namespace {
-class ArrowS3Environment final : public ::testing::Environment {
- public:
-  void TearDown() override { (void)::arrow::fs::FinalizeS3(); }
-};
-}  // namespace
-#endif
-
 TEST(ArrowS3FileIOTest, RejectsNonS3Uri) {
   auto result = MakeS3FileIO("file:///tmp/not-s3");
   EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
   EXPECT_THAT(result, HasErrorMessage("s3://"));
 }
 
-#if __has_include(<arrow/filesystem/s3fs.h>)
+#ifdef ICEBERG_S3_ENABLED
 TEST(ArrowS3FileIOTest, RequiresS3SupportAtBuildTime) {
   auto result = MakeS3FileIO("s3://bucket/path");
   if (!result.has_value()) {
@@ -216,11 +204,3 @@ TEST(ArrowS3FileIOTest, MakeS3FileIOWithTimeouts) {
 }
 
 }  // namespace iceberg::arrow
-
-#if __has_include(<arrow/filesystem/s3fs.h>)
-int main(int argc, char** argv) {
-  ::testing::InitGoogleTest(&argc, argv);
-  ::testing::AddGlobalTestEnvironment(new iceberg::arrow::ArrowS3Environment());
-  return RUN_ALL_TESTS();
-}
-#endif