feat(io): restore the oss scheme with a round trip test and docs

mkdocs/docs/file-io.md

@@ -29,7 +29,7 @@ implementations:
 | Registry name | Schemes |
 |---|---|
 | `arrow-fs-local` | paths without a scheme, `file` |
-| `arrow-fs-s3` | `s3`, `s3a`, `s3n` |
+| `arrow-fs-s3` | `s3`, `s3a`, `s3n`, `oss` |
 
 The S3 implementation requires Arrow S3 support.
 
@@ -56,6 +56,48 @@ For a REST catalog, set `io-impl` to the registry name. If it is omitted, the
 REST catalog uses `ResolvingFileIO` and selects a registered implementation for
 each file location's scheme.
 
+## Configure S3
+
+| Key | Example | Description |
+|---|---|---|
+| `s3.access-key-id` | `admin` | Static access key ID; must be set together with the secret key |
+| `s3.secret-access-key` | `password` | Static secret access key |
+| `s3.session-token` | `AQoDYXdzEJr...` | Session token, for temporary credentials. Ignored unless both static keys are set |
+| `client.region` | `us-east-1` | Region to sign requests for |
+| `s3.endpoint` | `https://127.0.0.1:9000` | Endpoint to use instead of the AWS one |
+| `s3.path-style-access` | `true` | Address buckets as a path (`endpoint/bucket`) instead of a virtual host (`bucket.endpoint`). Only takes effect together with `s3.endpoint` |
+| `s3.ssl.enabled` | `true` | Scheme to use for the endpoint, overriding the one it carries |
+| `s3.connect-timeout-ms` | `1000` | Connection timeout |
+| `s3.socket-timeout-ms` | `5000` | Request timeout. Ignored outside Windows and macOS |
+
+Without credentials, the AWS default credential chain is used, which covers
+environment variables, the shared configuration file, and the various role and
+identity providers.
+
+### S3-compatible storage
+
+Stores that speak the S3 API are served by the same implementation. The scheme
+selects it; `s3.endpoint` decides where requests actually go. A location keeps
+its own scheme and is canonicalized internally, so a credential vended for the
+`s3` prefix applies to it.
+
+For Alibaba Cloud OSS, point `s3.endpoint` at the S3-compatible endpoint of the
+bucket's region and set `s3.path-style-access` to `false`: with a custom
+endpoint, buckets are addressed as a path unless told otherwise, and the
+service rejects that with
+`SecondLevelDomainForbidden: Please use virtual hosted style to access`:
+
+```cpp
+auto file_io = iceberg::FileIORegistry::Load(
+    iceberg::FileIORegistry::kArrowS3FileIO,
+    {{std::string(iceberg::arrow::S3Properties::kEndpoint),
+      "https://s3.oss-cn-hangzhou.aliyuncs.com"},
+     {std::string(iceberg::arrow::S3Properties::kClientRegion), "cn-hangzhou"},
+     {std::string(iceberg::arrow::S3Properties::kPathStyleAccess), "false"}});
+
+file_io.value()->NewInputFile("oss://bucket/path/to/file.parquet");
+```
+
 ## Register a custom FileIO
 
 Register the factory before creating the catalog or resolver:

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -177,10 +177,17 @@ Result<std::shared_ptr<::arrow::fs::FileSystem>> BuildArrowS3FileSystem(
   return std::shared_ptr<::arrow::fs::FileSystem>(std::move(fs));
 }
 
+// Rewrites a foreign alias to `s3://` so locations and credential prefixes
+// compare equal. Derived from kS3Schemes: an alias missing here would not
+// fail, it would silently stop matching its credential.
 std::string CanonicalizeS3Scheme(std::string_view location) {
-  for (std::string_view scheme : {"s3a://", "s3n://"}) {
-    if (location.starts_with(scheme)) {
-      return std::string("s3://").append(location.substr(scheme.size()));
+  for (std::string_view scheme : kS3Schemes) {
+    if (scheme == S3Properties::kS3Schema) {
+      continue;
+    }
+    if (location.starts_with(scheme) &&
+        location.substr(scheme.size()).starts_with("://")) {
+      return std::string("s3://").append(location.substr(scheme.size() + 3));
     }
   }
   return std::string(location);

src/iceberg/arrow/s3/s3_properties.h

@@ -58,9 +58,11 @@ struct S3Properties {
 
 /// \brief URI schemes served by the Arrow S3 FileIO, lower-case.
 ///
-/// Single source of truth: both the registry registration and IsS3Scheme derive
-/// from this list, so a new alias only has to be added here.
-inline constexpr std::array<std::string_view, 3> kS3Schemes = {"s3", "s3a", "s3n"};
+/// Single source of truth: registration, IsS3Scheme and alias canonicalization
+/// all derive from this list, so a new alias only has to be added here.
+///
+/// `oss` is served because the store is S3-compatible; see the FileIO docs.
+inline constexpr std::array<std::string_view, 4> kS3Schemes = {"s3", "s3a", "s3n", "oss"};
 
 /// \brief Return whether a normalized URI scheme is S3-compatible.
 inline constexpr bool IsS3Scheme(std::string_view scheme) {

src/iceberg/test/CMakeLists.txt

@@ -280,10 +280,11 @@ endif()
 
 if(ICEBERG_BUILD_REST)
   function(add_rest_iceberg_test test_name)
+    set(options USE_BUNDLE)
     set(oneValueArgs)
     set(multiValueArgs SOURCES)
     cmake_parse_arguments(ARG
-                          ""
+                          "${options}"
                           "${oneValueArgs}"
                           "${multiValueArgs}"
                           ${ARGN})
@@ -292,12 +293,22 @@ if(ICEBERG_BUILD_REST)
     target_include_directories(${test_name} PRIVATE "${CMAKE_BINARY_DIR}/iceberg/test/")
     target_sources(${test_name} PRIVATE ${ARG_SOURCES})
     target_link_libraries(${test_name} PRIVATE GTest::gmock_main iceberg_rest_static)
+    if(ARG_USE_BUNDLE)
+      target_link_libraries(${test_name}
+                            PRIVATE "$<IF:$<TARGET_EXISTS:iceberg_bundle_static>,iceberg_bundle_static,iceberg_bundle_shared>"
+      )
+    endif()
     if(MSVC_TOOLCHAIN)
       target_compile_options(${test_name} PRIVATE /bigobj)
     endif()
     add_test(NAME ${test_name} COMMAND ${test_name})
   endfunction()
 
+  if(ICEBERG_BUILD_BUNDLE)
+    add_rest_iceberg_test(rest_arrow_file_io_test USE_BUNDLE SOURCES
+                          rest_arrow_file_io_test.cc)
+  endif()
+
   add_rest_iceberg_test(rest_catalog_test
                         SOURCES
                         auth_manager_test.cc

src/iceberg/test/arrow_s3_file_io_test.cc

@@ -191,8 +191,8 @@ TEST_F(ArrowS3FileIOTest, SkipsNonS3CredentialPrefix) {
 // credential that is silently skipped leaves S3 access on the default
 // credentials, which only surfaces much later as an auth error.
 TEST_F(ArrowS3FileIOTest, AppliesEveryS3CompatibleCredentialPrefix) {
-  for (std::string_view prefix :
-       {"s3", "s3://bucket/table", "s3a://bucket/table", "s3n://bucket/table"}) {
+  for (std::string_view prefix : {"s3", "s3://bucket/table", "s3a://bucket/table",
+                                  "s3n://bucket/table", "oss://bucket/table"}) {
     SCOPED_TRACE(prefix);
     auto result = MakeS3FileIO({});
     ASSERT_THAT(result, IsOk());

src/iceberg/test/location_util_test.cc

@@ -70,6 +70,19 @@ TEST(LocationUtilTest, ParseScheme) {
 
   auto empty_scheme = LocationUtil::ParseScheme("://bucket/path");
   EXPECT_TRUE(empty_scheme.empty());
+
+  // Not syntactically a scheme -> a path; the extended-length Windows form.
+  EXPECT_TRUE(LocationUtil::ParseScheme("\\\\?\\C:\\long\\file.parquet").empty());
+  EXPECT_TRUE(LocationUtil::ParseScheme("1:/file.parquet").empty());
+
+#ifdef _WIN32
+  // Drive letters are drives; both slash directions occur.
+  EXPECT_TRUE(LocationUtil::ParseScheme("C:/tmp/file.parquet").empty());
+  EXPECT_TRUE(LocationUtil::ParseScheme("D:\\a\\file.parquet").empty());
+#else
+  // Elsewhere a single letter stays a scheme for registered implementations.
+  EXPECT_EQ(LocationUtil::ParseScheme("C:/tmp/file.parquet"), "C");
+#endif
 }
 
 }  // namespace iceberg

src/iceberg/test/rest_arrow_file_io_test.cc

@@ -0,0 +1,198 @@
+/*
+ * 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.
+ */
+
+/// \file
+/// \brief Covers REST -> ResolvingFileIO -> registry -> Arrow FileIO against the
+/// real registered implementations, which mock delegates cannot exercise.
+
+#include <algorithm>
+#include <cstdlib>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <tuple>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_io_util.h"
+#include "iceberg/arrow/arrow_register.h"
+#include "iceberg/catalog/rest/rest_file_io.h"
+#include "iceberg/logging/logger.h"
+#include "iceberg/storage_credential.h"
+#include "iceberg/test/logging_test_helpers.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/temp_file_test_base.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+class RestArrowFileIOTest : public TempFileTestBase {
+ protected:
+  static void SetUpTestSuite() { iceberg::arrow::RegisterAll(); }
+  static void TearDownTestSuite() { std::ignore = iceberg::arrow::FinalizeS3(); }
+};
+
+TEST_F(RestArrowFileIOTest, ReadsBackWhatItWroteThroughRealLocalFileIO) {
+  auto io = MakeTableFileIO({{"warehouse", "logical_warehouse_name"}},
+                            /*table_config=*/{}, /*storage_credentials=*/{});
+  ASSERT_THAT(io, IsOk());
+
+  const auto path = CreateNewTempFilePathWithSuffix(".txt");
+  constexpr std::string_view kContent = "resolved through the real local FileIO";
+
+  ASSERT_THAT(io.value()->WriteFile(path, kContent), IsOk());
+  EXPECT_THAT(io.value()->ReadFile(path, std::nullopt),
+              HasValue(::testing::Eq(std::string(kContent))));
+  EXPECT_THAT(io.value()->DeleteFile(path), IsOk());
+}
+
+#if ICEBERG_S3_ENABLED
+
+bool HasWarning(const CapturingLogger& logger) {
+  const auto records = logger.records();
+  return std::ranges::any_of(
+      records, [](const LogMessage& record) { return record.level == LogLevel::kWarn; });
+}
+
+std::optional<std::string> GetEnvIfSet(const char* key) {
+  const char* value = std::getenv(key);
+  if (value == nullptr || std::string_view(value).empty()) {
+    return std::nullopt;
+  }
+  return std::string(value);
+}
+
+/// Addresses the store an S3 test reaches as `s3://` the way a catalog vending
+/// `oss://` locations would.
+/// Temporarily removes AWS credential variables, so nothing in the process
+/// environment can stand in for the vended credential under test.
+class ScopedScrubbedAwsCredentialEnv {
+ public:
+  ScopedScrubbedAwsCredentialEnv() {
+    for (const char* name : kNames) {
+      const char* value = std::getenv(name);
+      saved_.emplace_back(
+          name, value != nullptr ? std::optional<std::string>(value) : std::nullopt);
+      Unset(name);
+    }
+  }
+
+  ~ScopedScrubbedAwsCredentialEnv() {
+    for (const auto& [name, value] : saved_) {
+      if (value.has_value()) {
+        Set(name.c_str(), value->c_str());
+      } else {
+        Unset(name.c_str());
+      }
+    }
+  }
+
+ private:
+  static constexpr const char* kNames[] = {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY",
+                                           "AWS_SESSION_TOKEN"};
+
+  static void Set(const char* name, const char* value) {
+#  ifdef _WIN32
+    _putenv_s(name, value);
+#  else
+    ::setenv(name, value, /*overwrite=*/1);
+#  endif
+  }
+
+  static void Unset(const char* name) {
+#  ifdef _WIN32
+    _putenv_s(name, "");
+#  else
+    ::unsetenv(name);
+#  endif
+  }
+
+  std::vector<std::pair<std::string, std::optional<std::string>>> saved_;
+};
+
+std::string AsOssUri(std::string_view uri) {
+  const auto pos = uri.find("://");
+  const auto authority = pos == std::string_view::npos ? uri : uri.substr(pos + 3);
+  return std::string("oss://").append(authority);
+}
+
+// Resolution, credential matching and real I/O for `oss://`. The credential
+// env vars are scrubbed, so only the vended `s3`-scoped credential matching
+// the canonicalized location can authenticate.
+TEST_F(RestArrowFileIOTest, ReadsBackWhatItWroteThroughAnOssLocation) {
+  const auto base_uri = GetEnvIfSet("ICEBERG_TEST_S3_URI");
+  if (!base_uri.has_value()) {
+    GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable the oss:// round trip";
+  }
+
+  const auto access_key = GetEnvIfSet("AWS_ACCESS_KEY_ID");
+  const auto secret_key = GetEnvIfSet("AWS_SECRET_ACCESS_KEY");
+  ASSERT_TRUE(access_key.has_value() && secret_key.has_value())
+      << "Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY alongside "
+         "ICEBERG_TEST_S3_URI";
+  std::unordered_map<std::string, std::string> credential_config = {
+      {"s3.access-key-id", *access_key}, {"s3.secret-access-key", *secret_key}};
+  if (const auto session_token = GetEnvIfSet("AWS_SESSION_TOKEN")) {
+    credential_config["s3.session-token"] = *session_token;
+  }
+
+  ScopedScrubbedAwsCredentialEnv scrubbed;
+
+  // Scoped to `s3`, while the data it grants access to is addressed as `oss://`.
+  auto io = MakeTableFileIO({{"warehouse", "logical_warehouse_name"}},
+                            /*table_config=*/{},
+                            {{.prefix = "s3", .config = std::move(credential_config)}});
+  ASSERT_THAT(io, IsOk());
+
+  const auto object_uri = AsOssUri(*base_uri) + "/iceberg_oss_scheme_round_trip.txt";
+  constexpr std::string_view kContent = "resolved and written through an oss:// location";
+
+  ASSERT_THAT(io.value()->WriteFile(object_uri, kContent), IsOk());
+  EXPECT_THAT(io.value()->ReadFile(object_uri, std::nullopt),
+              HasValue(::testing::Eq(std::string(kContent))));
+  EXPECT_THAT(io.value()->DeleteFile(object_uri), IsOk());
+}
+
+TEST_F(RestArrowFileIOTest, AppliesOssCredentialThroughRealArrowS3FileIO) {
+  auto logger = std::make_shared<CapturingLogger>();
+  ScopedDefaultLogger scoped(logger);
+
+  auto io =
+      MakeTableFileIO({{"warehouse", "logical_warehouse_name"}}, /*table_config=*/{},
+                      {{.prefix = "oss://bucket/table", .config = {{"k", "v"}}}});
+  ASSERT_THAT(io, IsOk());
+
+  // Opening only builds the delegate, so just the pre-network failure modes
+  // are asserted: kNotSupported for a routing break, the warning for a drop.
+  auto input = io.value()->NewInputFile("oss://bucket/table/data/file.parquet");
+  EXPECT_THAT(input, ::testing::Not(IsError(ErrorKind::kNotSupported)));
+  EXPECT_FALSE(HasWarning(*logger));
+}
+
+#endif  // ICEBERG_S3_ENABLED
+
+}  // namespace
+
+}  // namespace iceberg::rest