diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md index 394948c3c..3573e746e 100644 --- a/mkdocs/docs/file-io.md +++ b/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: diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index 7c2799f7d..d2d31846a 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -177,10 +177,17 @@ Result> 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); diff --git a/src/iceberg/arrow/s3/s3_properties.h b/src/iceberg/arrow/s3/s3_properties.h index 35180537e..5f3f197f4 100644 --- a/src/iceberg/arrow/s3/s3_properties.h +++ b/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 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 kS3Schemes = {"s3", "s3a", "s3n", "oss"}; /// \brief Return whether a normalized URI scheme is S3-compatible. inline constexpr bool IsS3Scheme(std::string_view scheme) { diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index a5d902c75..2f12b2d05 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/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 "$,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 diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index 40719827a..ac6caf39f 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/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()); diff --git a/src/iceberg/test/location_util_test.cc b/src/iceberg/test/location_util_test.cc index 1aabe0ee0..0b806345c 100644 --- a/src/iceberg/test/location_util_test.cc +++ b/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 diff --git a/src/iceberg/test/rest_arrow_file_io_test.cc b/src/iceberg/test/rest_arrow_file_io_test.cc new file mode 100644 index 000000000..ce5a4a38c --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#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 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(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>> 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 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(); + 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 diff --git a/src/iceberg/util/location_util.cc b/src/iceberg/util/location_util.cc index 3ae2decc1..a7be56bc0 100644 --- a/src/iceberg/util/location_util.cc +++ b/src/iceberg/util/location_util.cc @@ -19,14 +19,45 @@ #include "iceberg/util/location_util.h" +#include +#include + namespace iceberg { +namespace { + +/// Whether `candidate` is a syntactically valid URI scheme (RFC 3986 section +/// 3.1): a letter followed by letters, digits, `+`, `-` or `.`. +bool IsValidScheme(std::string_view candidate) { + if (candidate.empty() || !std::isalpha(static_cast(candidate.front()))) { + return false; + } + return std::ranges::all_of(candidate, [](char c) { + const auto uc = static_cast(c); + return std::isalnum(uc) || c == '+' || c == '-' || c == '.'; + }); +} + +} // namespace + std::string_view LocationUtil::ParseScheme(std::string_view location) { const auto colon = location.find(':'); if (colon == std::string_view::npos || colon == 0) { return {}; } - return location.substr(0, colon); + const auto candidate = location.substr(0, colon); + // Cannot be a scheme -> a path whose first segment has a colon, such as the + // extended-length Windows form `\\?\C:\...`. + if (!IsValidScheme(candidate)) { + return {}; + } +#ifdef _WIN32 + // A single letter before the colon is a drive, not a scheme. + if (candidate.size() == 1) { + return {}; + } +#endif + return candidate; } } // namespace iceberg diff --git a/src/iceberg/util/location_util.h b/src/iceberg/util/location_util.h index 176ea1230..2a6147eed 100644 --- a/src/iceberg/util/location_util.h +++ b/src/iceberg/util/location_util.h @@ -33,8 +33,10 @@ class ICEBERG_EXPORT LocationUtil { /// \brief Extract the URI scheme from a location. /// /// This follows Java's ResolvingFileIO: the text before the first colon is - /// the scheme; an empty result means that no scheme was found. It does not - /// validate the rest of the location. + /// the scheme; an empty result means that no scheme was found. Text that is + /// not syntactically a scheme (RFC 3986) yields no scheme, and on Windows a + /// single letter is a drive rather than a scheme. The rest of the location is + /// not validated. static std::string_view ParseScheme(std::string_view location); static std::string_view StripTrailingSlash(std::string_view path) {