From dbc6a433ed2ef5048489188f55902c1a6cb3b47e Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Mon, 26 Oct 2026 17:55:27 +0800 Subject: [PATCH 1/8] docs: fix drifted option names, version numbers and feature lists --- README.md | 4 +- docs/code-style.md | 24 +++++++++-- docs/source/api/catalog.rst | 9 ++++ docs/source/basic_concepts.rst | 3 +- docs/source/build_system.rst | 11 +++++ docs/source/building.rst | 12 +++++- docs/source/user_guide/catalog.rst | 7 ++++ docs/source/user_guide/compaction.rst | 17 +++++--- .../user_guide/primary_key_global_index.rst | 9 ++-- docs/source/user_guide/system_tables.rst | 41 +++++++++++++++++++ docs/source/user_guide/write.rst | 6 ++- 11 files changed, 124 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index da928e713..e6d4e5a5e 100644 --- a/README.md +++ b/README.md @@ -31,12 +31,12 @@ Background and documentation are available at [paimon.apache.org](https://paimon Paimon C++ currently provides: - **Write**: append table and primary key table write support with compaction. -- **Commit**: append table commit support for simple append-only tables. +- **Commit**: append table and primary key table commit support. Direct file-system commits work on non-object-store paths; object-store paths require a catalog-mediated commit. - **Scan**: batch and stream scan for append tables and primary key tables without changelog. - **Read**: append table read, primary key table read with deletion vector, and primary key table merge-on-read. - **Arrow integration**: batch read and write interfaces based on the [Arrow Columnar In-Memory Format](https://arrow.apache.org). - **File systems**: file system abstraction with built-in local, Jindo, OSS, and S3 file system support. -- **File formats**: file format abstraction with built-in ORC, Parquet, and Avro support. +- **File formats**: file format abstraction with built-in Parquet, Blob, ORC, and Avro support, plus optional Lance and Mosaic formats. - **Runtime utilities**: memory pool and thread pool abstractions with default implementations. - **AI-Oriented Features**: supports RowTracking and DataEvolution mode and provides Global Index capabilities including B-tree index, DiskANN-based vector search with Lumina, and Lucene-based diff --git a/docs/code-style.md b/docs/code-style.md index cb8c648e8..3772f0f33 100644 --- a/docs/code-style.md +++ b/docs/code-style.md @@ -404,6 +404,11 @@ Before writing new helper code, check whether the project already provides what | `UniqueTestDirectory` | RAII helper that creates a unique temp directory and cleans it up on destruction | | `TimezoneGuard` | RAII guard that sets `TZ` environment variable and restores it on destruction | | `IOExceptionHelper` | Inject I/O errors via `IOHook` for fault-injection testing | +| `CountingRoutingCache` | `Cache` routing by `CacheKind` to a per-kind `LruCache`, counting gets and supplier calls overall and per kind | +| `DeletionVectorTestHelper` | Express row-level deletes the way an external engine does: write a deletion vector index file and commit it as an index-only commit message | +| `GatedAsyncInputStream` | `InputStream` wrapper holding `ReadAsync` callbacks until `ReleaseAll()`, so a test can observe a cache while its fetches are in flight | +| `BuildTestSnapshot()` | Build a `Snapshot` from an id and optional uuid, filling in every other field | +| `VariantTestData` | Build variant test data, including an Arrow `StructArray` batch | | `testharness.h` | Test macros: `ASSERT_OK`, `ASSERT_NOK`, `ASSERT_NOK_WITH_MSG`, `ASSERT_RAISES` | ### Mock Objects (`src/paimon/testing/mock/`) @@ -417,6 +422,13 @@ Before writing new helper code, check whether the project already provides what | `MockIndexPathFactory` | Mock index path factory | | `MockKeyValueDataFileRecordReader` | Mock key-value data file reader holding in-memory data | | `MockStatsExtractor` | Mock statistics extractor | +| `MockInputStream` / `MockOutputStream` | No-op I/O doubles: reads and writes report zero bytes and `ReadAsync` completes inline | +| `MockVersionManagedCatalog` | Mock `Catalog` that also implements `VersionManagedCatalog` and records every commit call | +| `MockGlobalIndexReader` | Mock `GlobalIndexReader`, constructed with or without search support | +| `MockSearchFileIndex` | File index backend with configured matches in search order; applies pre-filters before limits and returns vector scores aligned with ascending file-local row positions | +| `MockSearchFileIndexer` / `MockSearchFileIndexReader` | Mock `FileIndexer` and `FileIndexReader` serving those configured scored rows | +| `MockFullTextSearchFileIndexFactory` / `MockVectorSearchFileIndexFactory` | Mock `FileIndexerFactory` for the full-text and vector search index types | +| `OptionsCheckingMockSearchFileIndexFactory` | Reuses the mock search backend while requiring the exact options supplied to its factory | --- @@ -453,6 +465,8 @@ This installs Git hooks that automatically run on every commit: | `trailing-whitespace` | Trailing whitespace | | `end-of-file-fixer` | Missing newline at end of file | | `check-added-large-files` | Files > 5 MB | +| `check-yaml` | YAML syntax | +| `sphinx-lint` (alias `docs`) | reStructuredText syntax under `docs/source/` | To run all hooks on all files manually: @@ -464,10 +478,12 @@ pre-commit run --all-files Static analysis is configured in `.clang-tidy`. Key enabled check groups: -- `bugprone-*` — Common bug patterns -- `google-*` — Google style checks -- `modernize-*` — C++ modernization suggestions -- `clang-analyzer-*` — Clang static analyzer +- `bugprone-*` — Common bug patterns. These are listed one check at a time rather than enabled as a wildcard, so a new upstream `bugprone` check never turns itself on. +- `misc-misleading-identifier`, `misc-homoglyph` — The only two `misc` checks enabled; the rest of `misc-*` is off. +- `google-*` — Google style checks, except `google-default-arguments`. +- `modernize-*` — C++ modernization suggestions, except `modernize-return-braced-init-list`, `modernize-avoid-c-arrays`, `modernize-use-trailing-return-type`, `modernize-use-nodiscard` and `modernize-pass-by-value`. +- `clang-analyzer-*` — Clang static analyzer, except the `clang-analyzer-alpha*` checkers and `clang-analyzer-cplusplus.NewDeleteLeaks`. +- `clang-diagnostic-*` — Compiler diagnostics, except `clang-diagnostic-global-constructors` and `clang-diagnostic-sign-compare`. ### Building & Testing diff --git a/docs/source/api/catalog.rst b/docs/source/api/catalog.rst index c3956d32e..df247f4a8 100644 --- a/docs/source/api/catalog.rst +++ b/docs/source/api/catalog.rst @@ -30,3 +30,12 @@ Interface .. doxygenclass:: paimon::Identifier :members: :undoc-members: + +.. doxygenclass:: paimon::Database + :members: + +.. doxygenclass:: paimon::Table + :members: + +.. doxygenstruct:: paimon::CatalogOptions + :members: diff --git a/docs/source/basic_concepts.rst b/docs/source/basic_concepts.rst index 0af523056..9be8fec40 100644 --- a/docs/source/basic_concepts.rst +++ b/docs/source/basic_concepts.rst @@ -55,7 +55,8 @@ the corresponding snapshot. Data Files --------------------------- Data files are grouped by partitions. Paimon C++ supports Parquet (default), -ORC, and Avro as data file formats, with optional Lance and Mosaic support. +ORC, and Avro as data file formats, with optional Lance, Mosaic, and Vortex +support. Partition --------------------------- diff --git a/docs/source/build_system.rst b/docs/source/build_system.rst index 6687290ce..44495f73c 100644 --- a/docs/source/build_system.rst +++ b/docs/source/build_system.rst @@ -102,6 +102,8 @@ Paimon provides a set of built-in optional plugins that you can link to as neede - ``Paimon::paimon_avro_file_format_shared`` - ``Paimon::paimon_blob_file_format_shared`` - ``Paimon::paimon_lance_file_format_shared`` + - ``Paimon::paimon_mosaic_file_format_shared`` + - ``Paimon::paimon_vortex_file_format_shared`` - File system plugins: @@ -113,8 +115,17 @@ Paimon provides a set of built-in optional plugins that you can link to as neede - Index plugins: - ``Paimon::paimon_file_index_shared`` + - ``Paimon::paimon_global_index_shared`` - ``Paimon::paimon_lumina_index_shared`` - ``Paimon::paimon_lucene_index_shared`` + - ``Paimon::paimon_full_text_index_shared`` + +.. note:: + + A plugin target only exists when its component is enabled. ``parquet``, + ``blob``, ``local``, ``file_index`` and ``global_index`` are always built; the + others require the matching ``PAIMON_ENABLE_*`` flag described in + :ref:`cpp_build_optional_components`. .. note:: diff --git a/docs/source/building.rst b/docs/source/building.rst index 93a4890b3..b0fe95e44 100644 --- a/docs/source/building.rst +++ b/docs/source/building.rst @@ -173,12 +173,20 @@ Optional Components By default, the C++ build system creates a fairly minimal build. We have several optional system components which you can opt into building by passing -boolean flags to ``cmake``. +boolean flags to ``cmake``. ORC and Avro are the exceptions: both already default +to ``ON``, while every other component below defaults to ``OFF``. * ``-DPAIMON_ENABLE_ORC=ON``: Paimon integration with Apache ORC * ``-DPAIMON_ENABLE_AVRO=ON``: Apache Avro libraries and Paimon integration * ``-DPAIMON_ENABLE_LANCE=ON``: Lance 0.39 file format support through a Rust FFI; requires Rust 1.88 or newer, cbindgen, and protoc 3.12 or newer +* ``-DPAIMON_ENABLE_MOSAIC=ON``: Apache Paimon Mosaic file format support through a Rust FFI; + requires cargo on ``PATH`` +* ``-DPAIMON_ENABLE_VORTEX=ON``: Vortex file format support through a Rust FFI; requires cargo, + the stable Rust toolchain (Vortex pins ``rust-version`` 1.95 or newer and is built with + ``RUSTUP_TOOLCHAIN=stable``), and a ``flatc`` whose version matches the flatbuffers crate that + Vortex uses. Distribution ``flatc`` packages are typically too old; set ``FLATC`` to the path of + a suitable binary when it is not on ``PATH``. * ``-DPAIMON_ENABLE_JINDO=ON``: Support for Alibaba Jindo filesystems * ``-DPAIMON_ENABLE_OSS=ON``: Support for Alibaba Cloud OSS through OSS SDK V2 * ``-DPAIMON_ENABLE_S3=ON``: Support for Amazon S3-compatible filesystems @@ -278,6 +286,8 @@ For development builds, you will often want to enable additional targets in enable to exercise your changes, using the following ``cmake`` options. * ``-DPAIMON_BUILD_TESTS=ON``: Build executable unit tests. +* ``-DPAIMON_ENABLE_NETWORK_TESTS=ON``: Enable tests that access real remote services over the + network. These are disabled by default. Optional Checks ~~~~~~~~~~~~~~~ diff --git a/docs/source/user_guide/catalog.rst b/docs/source/user_guide/catalog.rst index d881968c4..aa42c2d77 100644 --- a/docs/source/user_guide/catalog.rst +++ b/docs/source/user_guide/catalog.rst @@ -153,6 +153,13 @@ offsets when a real-time writer is created: PAIMON_ASSIGN_OR_RAISE(std::unique_ptr writer, paimon::FileStoreWrite::Create(std::move(write_context))); +The ``Table`` handle is an equivalent entry point: ``table->NewCommit()``, +``table->NewWrite()``, ``table->NewRead()`` and ``table->NewScan()`` each take the +same finished context and return the same operator as ``FileStoreCommit::Create()``, +``FileStoreWrite::Create()``, ``TableRead::Create()`` and ``TableScan::Create()``. +Pick whichever reads better; the context still carries the catalog when one is +needed. See the :ref:`Table API `. + For real-time writes, also set ``WithStreamingMode(true)`` and ``WithRealtimeContext(...)`` with ``realtime.enabled=true``. After ``CommitWithProgress`` succeeds, pass its returned snapshot id to diff --git a/docs/source/user_guide/compaction.rst b/docs/source/user_guide/compaction.rst index 782fd53fd..b80bfdee4 100644 --- a/docs/source/user_guide/compaction.rst +++ b/docs/source/user_guide/compaction.rst @@ -78,8 +78,8 @@ Full Compaction Full compaction rewrites all eligible files in the bucket. During full compaction: -- Files whose size is already at or above ``compaction.file-size`` (and have no - associated deletion vectors) are skipped to avoid unnecessary rewrites. +- Files whose size is already at or above the compaction file-size threshold (and + have no associated deletion vectors) are skipped to avoid unnecessary rewrites. - When deletion vectors are enabled, all files are always eligible for compaction regardless of size, because deletion vectors must be applied. - When ``compaction.force-rewrite-all-files`` is ``true``, all files are @@ -88,10 +88,17 @@ compaction: small files exceeds the number of large files and the total file count is at least 3. -After compaction, if the last output file is still smaller than -``compaction.file-size``, it is placed back into the compaction queue for future +After compaction, if the last output file is still smaller than the compaction +file-size threshold, it is placed back into the compaction queue for future merging. +The compaction file-size threshold is not a separate option. It is derived from +``target-file-size`` as ``target-file-size / 10 * 7``, that is 70% of the target +size, so that a file already close to the target is not compacted twice. +``target-file-size`` defaults to 256 MiB for append-only tables and 128 MiB for +primary-key tables, which makes the derived threshold about 179 MiB and 89 MiB +respectively. Adjust ``target-file-size`` to move the threshold. + Dictionary Passthrough ~~~~~~~~~~~~~~~~~~~~~~ An append-only compaction rewrite copies rows into the new file without @@ -126,7 +133,7 @@ dictionary-encodes it the same way; this option only selects ``STRING`` columns. A file with a serialized Arrow dictionary schema can independently produce a ``BINARY`` dictionary, which the value accessors also support. The rewrite also overrides the option back to ``false`` when the table writes a format other than Parquet, when -``parquet.enable-dictionary`` is ``false`` because the writer would only expand +``parquet.enable.dictionary`` is ``false`` because the writer would only expand the values again, or when variant/map shredding is configured because those writers reshape each batch against a fixed physical schema. Setting the option therefore never makes a rewrite fail; at worst it has no effect. diff --git a/docs/source/user_guide/primary_key_global_index.rst b/docs/source/user_guide/primary_key_global_index.rst index 9307d938d..e955cd43f 100644 --- a/docs/source/user_guide/primary_key_global_index.rst +++ b/docs/source/user_guide/primary_key_global_index.rst @@ -84,10 +84,11 @@ Current scope definitions are recognized for validation, but their evaluation conservatively falls back to a normal scan until their dedicated payload readers are supported. - The read path targets the Java release-2.0.0 layout and scan semantics (source metadata - v1, ``GlobalIndexMeta`` with ``_SOURCE_META``, commit message v12). Source-file names - currently use the existing C++ length-prefixed UTF-8 streams; ASCII and non-null BMP - names are compatible with Java ``writeUTF``, while complete modified UTF-8 support for - supplementary code points will be handled by a shared stream-level change. + v1, ``GlobalIndexMeta`` with ``_SOURCE_META``). Commit messages are written at version 13 + and remain readable down to version 3. Source-file names currently use the existing C++ + length-prefixed UTF-8 streams; ASCII and non-null BMP names are compatible with Java + ``writeUTF``, while complete modified UTF-8 support for supplementary code points will be + handled by a shared stream-level change. - Automatic maintenance is synchronous during prepare-commit. Java's asynchronous build scheduling, retries, fairness metrics, and manual rebuild actions are not part of the C++ API. Realtime and postpone-bucket writers do not build source-backed payloads. diff --git a/docs/source/user_guide/system_tables.rst b/docs/source/user_guide/system_tables.rst index 6ac018823..90c88c133 100644 --- a/docs/source/user_guide/system_tables.rst +++ b/docs/source/user_guide/system_tables.rst @@ -26,6 +26,47 @@ Branch-qualified paths are also supported. For example, ``/warehouse/db.db/orders$branch_audit$files`` reads the files system table from the ``audit`` branch. +Table-Level System Tables +------------------------- + +These suffixes are recognized on a data table path. Matching is +case-insensitive, and an unrecognized suffix fails with +``unsupported system table: `` instead of falling back to the base table. + +* ``$options``: the latest base table options as key/value rows. +* ``$snapshots``: snapshot commit history. +* ``$schemas``: schema evolution history. +* ``$tags``: tags and the snapshots they reference. +* ``$branches``: table branches, including ``main``. +* ``$consumers``: persisted streaming consumer offsets. +* ``$manifests``: data manifest metadata in the latest snapshot. +* ``$files``: data file metadata in the latest snapshot. +* ``$audit_log``: row-level changelog records with rowkind. +* ``$binlog``: changelog records with list-wrapped data columns. +* ``$ro``: read-optimized data, described in the next section. + +Global System Tables +-------------------- + +Four system tables are not bound to a single data table. They live under the +reserved ``sys`` database and enumerate across all databases in the warehouse: + +* ``sys.catalog_options``: catalog-level configuration as key/value rows. +* ``sys.all_table_options``: all table options as + ``(database_name, table_name, key, value)`` rows. +* ``sys.tables``: metadata for all tables, including record counts and file + statistics. +* ``sys.partitions``: partition-level file statistics for all tables. + +These need a catalog to enumerate databases and tables, so load them through a +catalog as the identifier ``sys.`` rather than from a bare +``/sys/`` path; a path load without catalog context fails with +``global system table requires catalog context: ``. +``sys.all_table_options``, ``sys.tables`` and ``sys.partitions`` are always +available. ``sys.catalog_options`` is disabled by default and has to be turned +on with the catalog option ``catalog-options-table.enabled``; a disabled one is +reported as not existing. + Read-Optimized System Table --------------------------- diff --git a/docs/source/user_guide/write.rst b/docs/source/user_guide/write.rst index 4a3d27dbe..6c7bff16b 100644 --- a/docs/source/user_guide/write.rst +++ b/docs/source/user_guide/write.rst @@ -231,7 +231,9 @@ produce a correct ``Snapshot``, which commonly includes (but is not limited to): The C++ writer supports Append and PK tables and can produce ``CommitMessage`` objects for both. ``FileStoreCommit`` supports direct file-system commits for both table types on non-object-store paths. - Object-store paths require REST catalog commit mode. Changelog is out of + Object-store paths require a catalog-mediated commit, configured with either + ``CommitContextBuilder::WithCatalog()`` or ``UseRESTCatalogCommit()``; + neither needs an atomic rename on the object store. Changelog is out of scope and should not be emitted in ``CommitMessage`` until explicitly supported. @@ -256,7 +258,7 @@ Operational Flow 1. Writer nodes perform data ingestion and produce Arrow ``RecordBatch`` organized by partition and bucket. -2. Writers flush batches into ORC, Parquet, or Avro files via registered +2. Writers flush batches into data files through the registered ``file.format`` and ``file-system`` backends, producing file-level metadata and per-batch commit state. From 58a50cac4802546b622309d5bd315abe415ddbb1 Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Mon, 26 Oct 2026 16:27:44 +0700 Subject: [PATCH 2/8] fix(logging): read logger creator under lock in GetLogger (#464) --- include/paimon/logging.h | 16 ++++++ src/paimon/common/logging/logging.cpp | 28 ++++++---- src/paimon/common/logging/logging_test.cpp | 65 ++++++++++++++++++++++ 3 files changed, 99 insertions(+), 10 deletions(-) diff --git a/include/paimon/logging.h b/include/paimon/logging.h index 0859d4435..95d22b7d2 100644 --- a/include/paimon/logging.h +++ b/include/paimon/logging.h @@ -55,8 +55,24 @@ class PAIMON_EXPORT Logger { public: using LoggerCreator = std::function(const std::string&)>; + /// Register a creator that `GetLogger()` uses instead of the default glog logger. + /// + /// @param creator Creates a logger from the `path` passed to `GetLogger()`. All `GetLogger()` + /// calls share it, so it must be safe to call from multiple threads simultaneously, and its + /// state persists across calls. It may call `RegisterLogger()`. It replaces any previously + /// registered creator, which stays alive until running `GetLogger()` calls finish with it. + /// + /// @note This method is thread-safe and can be called concurrently with `GetLogger()`. static void RegisterLogger(LoggerCreator creator); + /// Get a logger from the registered creator, or a glog logger if no creator is registered. + /// + /// @param path Name of the logger, such as the class that logs. It is passed to the + /// registered creator, and the glog logger ignores it. + /// + /// @return A unique pointer to the logger. + /// + /// @note This method is thread-safe. static std::unique_ptr GetLogger(const std::string& path); virtual void LogV(PaimonLogLevel level, const char* fname, int lineno, const char* function, diff --git a/src/paimon/common/logging/logging.cpp b/src/paimon/common/logging/logging.cpp index c91d8fc96..bfd649f68 100644 --- a/src/paimon/common/logging/logging.cpp +++ b/src/paimon/common/logging/logging.cpp @@ -23,9 +23,10 @@ #include #include #include +#include #include -#include #include +#include #include "glog/log_severity.h" #include "glog/logging.h" @@ -43,8 +44,8 @@ static const char* GetProgramName() { #endif } -static std::optional& getLoggerCreator() { - static std::optional _loggerCreator; +static std::shared_ptr& getLoggerCreator() { + static std::shared_ptr _loggerCreator; return _loggerCreator; } @@ -55,7 +56,7 @@ static std::shared_mutex& getRegistryLock() { void Logger::RegisterLogger(LoggerCreator creator) { std::unique_lock lock(getRegistryLock()); - getLoggerCreator() = creator; + getLoggerCreator() = std::make_shared(std::move(creator)); } static google::LogSeverity ToGlogLevel(PaimonLogLevel level) { @@ -99,15 +100,22 @@ class GlogAdaptor : public Logger { }; std::unique_ptr Logger::GetLogger(const std::string& path) { - auto& creator = getLoggerCreator(); - if (creator) { + // Share the registered creator under the lock and call it after unlocking, so it keeps + // its state, stays alive if replaced, and may call RegisterLogger() without deadlocking. + std::shared_ptr creator; + { std::shared_lock lock(getRegistryLock()); - return creator.value()(path); + creator = getLoggerCreator(); } - std::unique_lock ulock(getRegistryLock()); - if (!google::IsGoogleLoggingInitialized()) { - google::InitGoogleLogging(GetProgramName()); + if (creator) { + return (*creator)(path); } + static std::once_flag glog_init_flag; + std::call_once(glog_init_flag, []() { + if (!google::IsGoogleLoggingInitialized()) { + google::InitGoogleLogging(GetProgramName()); + } + }); return std::make_unique(); } diff --git a/src/paimon/common/logging/logging_test.cpp b/src/paimon/common/logging/logging_test.cpp index 3c391a4bf..d0122b31d 100644 --- a/src/paimon/common/logging/logging_test.cpp +++ b/src/paimon/common/logging/logging_test.cpp @@ -19,6 +19,7 @@ #include "paimon/logging.h" #include +#include #include #include #include @@ -66,6 +67,55 @@ void RegisterCustomLoggerCreatorAndExit() { std::_Exit(used_creator ? 0 : 1); } +void RegisterStatefulLoggerCreatorAndExit() { + g_creator_calls.store(0); + Logger::RegisterLogger([calls = 0](const std::string&) mutable -> std::unique_ptr { + g_creator_calls.store(++calls); + return nullptr; + }); + Logger::GetLogger("first"); + Logger::GetLogger("second"); + std::_Exit(g_creator_calls.load() == 2 ? 0 : 1); +} + +void RegisterLoggerCreatorWhileGettingLoggerAndExit() { + g_creator_calls.store(0); + std::thread getter([]() { + while (g_creator_calls.load() == 0) { + Logger::GetLogger("getter"); + } + }); + Logger::RegisterLogger([](const std::string&) -> std::unique_ptr { + g_creator_calls.fetch_add(1); + return nullptr; + }); + getter.join(); + // std::exit runs ThreadSanitizer's exit check, which fails the process on a race. + std::exit(0); +} + +void RegisterLoggerFromLoggerCreatorAndExit() { + static std::weak_ptr creator_state; + static bool kept_alive = false; + auto state = std::make_shared(0); + creator_state = state; + Logger::RegisterLogger([state](const std::string&) -> std::unique_ptr { + Logger::RegisterLogger( + [](const std::string&) -> std::unique_ptr { return nullptr; }); + kept_alive = !creator_state.expired(); + return nullptr; + }); + state.reset(); + std::future getter = + std::async(std::launch::async, []() { Logger::GetLogger("getter"); }); + if (getter.wait_for(std::chrono::seconds(30)) != std::future_status::ready) { + std::cerr << "GetLogger() did not finish within 30 seconds" << std::endl; + std::_Exit(1); + } + getter.get(); + std::_Exit(kept_alive && creator_state.expired() ? 0 : 1); +} + } // namespace TEST(LoggerTest, TestMultiThreadGetLogger) { ASSERT_OK_AND_ASSIGN(auto executor, CreateDefaultExecutor(/*thread_count=*/4)); @@ -192,4 +242,19 @@ TEST(LoggerTest, TestRegisterCustomLoggerCreator) { ::testing::FLAGS_gtest_death_test_style = "threadsafe"; ASSERT_EXIT(RegisterCustomLoggerCreatorAndExit(), ::testing::ExitedWithCode(0), ""); } + +TEST(LoggerTest, TestRegisterStatefulLoggerCreator) { + ::testing::FLAGS_gtest_death_test_style = "threadsafe"; + ASSERT_EXIT(RegisterStatefulLoggerCreatorAndExit(), ::testing::ExitedWithCode(0), ""); +} + +TEST(LoggerTest, TestRegisterLoggerCreatorWhileGettingLogger) { + ::testing::FLAGS_gtest_death_test_style = "threadsafe"; + ASSERT_EXIT(RegisterLoggerCreatorWhileGettingLoggerAndExit(), ::testing::ExitedWithCode(0), ""); +} + +TEST(LoggerTest, TestRegisterLoggerFromLoggerCreator) { + ::testing::FLAGS_gtest_death_test_style = "threadsafe"; + ASSERT_EXIT(RegisterLoggerFromLoggerCreatorAndExit(), ::testing::ExitedWithCode(0), ""); +} } // namespace paimon::test From 43e2256a146e267f7e23748bfd3695ce02df80ef Mon Sep 17 00:00:00 2001 From: wangyong9999 <81852543+wangyong9999@users.noreply.github.com> Date: Mon, 26 Oct 2026 18:27:04 +0800 Subject: [PATCH 3/8] perf(scan): skip manifest lists on exact snapshot cache hits (#374) --- .../user_guide/manifest_entry_cache.rst | 7 +- .../snapshot_live_manifest_entries.cpp | 13 ++- .../manifest/snapshot_live_manifest_entries.h | 6 +- .../append_only_file_store_scan_test.cpp | 47 ++++++++- src/paimon/core/operation/file_store_scan.cpp | 98 +++++++++++-------- src/paimon/core/operation/file_store_scan.h | 20 ++-- .../core/operation/file_store_scan_test.cpp | 24 +++-- 7 files changed, 147 insertions(+), 68 deletions(-) diff --git a/docs/source/user_guide/manifest_entry_cache.rst b/docs/source/user_guide/manifest_entry_cache.rst index 0f37138c2..995d77e38 100644 --- a/docs/source/user_guide/manifest_entry_cache.rst +++ b/docs/source/user_guide/manifest_entry_cache.rst @@ -28,8 +28,8 @@ scans that target the same bucket. The cache stores decoded and merged live manifest entries by table path, branch, and bucket for ``ScanMode::ALL``. Each cache value can retain several snapshot -results for that bucket. Exact snapshot hits are served from the cache; cache -misses rebuild the target snapshot bucket from the target snapshot's data +results for that bucket. Exact snapshot hits are served from the cache without reading or decoding the +manifest lists again; cache misses rebuild the target snapshot bucket from the target snapshot's data manifests and store the rebuilt live entries. Request-specific filters are not stored in the cache. Partition, level, and @@ -120,3 +120,6 @@ The scan metrics expose existing counters for the last scan: - ``lastScannedManifests``: how many manifest files were loaded during this scan before manifest entry decoding. + +On an exact cache hit, ``lastScannedManifests`` is zero. The cached table file +count preserves the skipped-file metric even when manifest lists are bypassed. diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp index c6d981e10..6f925ba15 100644 --- a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp +++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp @@ -34,7 +34,7 @@ namespace paimon { namespace { -constexpr int32_t kMagic = 0x534d4543; // SMEC +constexpr int32_t kMagic = 0x534d4544; // SMED: includes per-snapshot total file counts size_t NormalizeMaxSnapshots(int32_t max_snapshots) { return static_cast(std::max(0, max_snapshots)); @@ -61,12 +61,14 @@ std::shared_ptr ToBytes(const MemorySegmentOutputStream& out, SnapshotLiveManifestEntries::SnapshotLiveManifestEntries(int32_t max_snapshots) : max_snapshots_(max_snapshots) {} -void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector&& entries) { +void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector&& entries, + int64_t total_data_files) { if (NormalizeMaxSnapshots(max_snapshots_) == 0) { return; } entries_by_snapshot_[snapshot_id] = std::make_shared>(std::move(entries)); + total_data_files_by_snapshot_[snapshot_id] = total_data_files; EvictIfNeeded(); } @@ -83,6 +85,7 @@ Result> SnapshotLiveManifestEntries::Serialize( ManifestEntrySerializer serializer(pool); for (const auto& [snapshot_id, entries] : entries_by_snapshot_) { out.WriteValue(snapshot_id); + out.WriteValue(total_data_files_by_snapshot_.at(snapshot_id)); PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out)); } return ToBytes(out, pool); @@ -111,7 +114,10 @@ Result SnapshotLiveManifestEntries::Deserialize( ManifestEntrySerializer serializer(pool); for (int32_t i = 0; i < snapshot_count; i++) { PAIMON_ASSIGN_OR_RAISE(int64_t snapshot_id, in.ReadValue()); + PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files, in.ReadValue()); PAIMON_ASSIGN_OR_RAISE(std::vector entries, serializer.DeserializeList(&in)); + snapshot_live_manifest_entries.total_data_files_by_snapshot_[snapshot_id] = + total_data_files; snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] = std::make_shared>(std::move(entries)); } @@ -122,6 +128,7 @@ Result SnapshotLiveManifestEntries::Deserialize( void SnapshotLiveManifestEntries::EvictIfNeeded() { size_t max_snapshots = NormalizeMaxSnapshots(max_snapshots_); while (entries_by_snapshot_.size() > max_snapshots) { + total_data_files_by_snapshot_.erase(entries_by_snapshot_.begin()->first); entries_by_snapshot_.erase(entries_by_snapshot_.begin()); } } @@ -151,6 +158,7 @@ SnapshotLiveManifestEntries::ReadSnapshot( std::optional previous_id; for (int32_t i = 0; i < snapshot_count; ++i) { PAIMON_ASSIGN_OR_RAISE(int64_t id, in.ReadValue()); + PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files, in.ReadValue()); // Serialize emits snapshots in ascending order. Validate this before using the position // to honor a retention limit smaller than the one used when the cache was populated. if (previous_id && id <= previous_id.value()) { @@ -165,6 +173,7 @@ SnapshotLiveManifestEntries::ReadSnapshot( if (keep) { selected.emplace(); selected->candidate_entries = entry_count; + selected->total_data_files = total_data_files; if (!row_filter) { selected->entries.reserve(entry_count); } diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.h b/src/paimon/core/manifest/snapshot_live_manifest_entries.h index 784d31b28..f2d4343ad 100644 --- a/src/paimon/core/manifest/snapshot_live_manifest_entries.h +++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.h @@ -45,11 +45,12 @@ class SnapshotLiveManifestEntries { struct Selection { std::vector entries; uint64_t candidate_entries = 0; + int64_t total_data_files = 0; }; explicit SnapshotLiveManifestEntries(int32_t max_snapshots); - void Put(int64_t snapshot_id, std::vector&& entries); + void Put(int64_t snapshot_id, std::vector&& entries, int64_t total_data_files); size_t Size() const; Result> Serialize(const std::shared_ptr& pool) const; @@ -59,7 +60,7 @@ class SnapshotLiveManifestEntries { /// Select an exact retained snapshot, applying a row filter before constructing file metadata. /// The reader validates the entry version before invoking the filter, as in - /// ManifestFile::ReadSelectedEntries. The serialized cache value is unchanged. A Selection is + /// ManifestFile::ReadSelectedEntries. A Selection is /// deliberately not a cache value: query-filtered entries must never replace the complete /// bucket's cached entries. static Result> ReadSnapshot( @@ -71,6 +72,7 @@ class SnapshotLiveManifestEntries { void EvictIfNeeded(); std::map>> entries_by_snapshot_; + std::map total_data_files_by_snapshot_; int32_t max_snapshots_; }; diff --git a/src/paimon/core/operation/append_only_file_store_scan_test.cpp b/src/paimon/core/operation/append_only_file_store_scan_test.cpp index 29d19d7f3..b8dfbfced 100644 --- a/src/paimon/core/operation/append_only_file_store_scan_test.cpp +++ b/src/paimon/core/operation/append_only_file_store_scan_test.cpp @@ -815,7 +815,9 @@ TEST(AppendOnlyFileStoreScanTest, TestDropStatsAfterFiltering) { TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { TimezoneGuard guard("Asia/Shanghai"); std::string table_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/"; - auto cache = std::make_shared(/*max_weight=*/16 * 1024 * 1024); + auto cache = std::make_shared( + std::map{{CacheKind::MANIFEST, 16 * 1024 * 1024}, + {CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 * 1024 * 1024}}); // First scan on snapshot 5: cache miss, entries rebuilt from all manifests. auto scan_first = BuildScan(table_path, cache, /*bucket=*/0); @@ -839,6 +841,11 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { ASSERT_EQ(first_cache_hit, 0); ASSERT_EQ(first_cache_misses, 1); + const int64_t manifest_gets = cache->GetCount(CacheKind::MANIFEST); + ASSERT_GT(manifest_gets, 0); + ASSERT_OK_AND_ASSIGN(uint64_t first_skipped, + first_metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES)); + // Second scan on the same snapshot should read the same bucket live entries from cache. auto scan_second = BuildScan(table_path, cache, /*bucket=*/0); scan_second->WithSnapshot(snapshot_5); @@ -855,9 +862,36 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { uint64_t materialized_rows, second_metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS)); ASSERT_EQ(second_cache_hit, 1); + ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), manifest_gets); + ASSERT_OK_AND_ASSIGN(uint64_t second_scanned, + second_metrics->GetCounter(ScanMetrics::LAST_SCANNED_MANIFESTS)); + ASSERT_OK_AND_ASSIGN(uint64_t second_skipped, + second_metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES)); + ASSERT_EQ(second_scanned, 0); + ASSERT_EQ(second_skipped, first_skipped); ASSERT_EQ(second_cache_hits, 1); ASSERT_GE(scanned_rows, materialized_rows); ASSERT_OK(second_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION)); + + // A different predicate must still filter the cached, unfiltered bucket entries. + std::shared_ptr predicate = + PredicateBuilder::Equal(0, "f0", FieldType::STRING, Literal(FieldType::STRING, "David", 5)); + auto filtered = BuildScan(table_path, cache, /*bucket=*/0, predicate); + filtered->WithSnapshot(snapshot_5); + ASSERT_OK_AND_ASSIGN(auto filtered_plan, filtered->CreatePlan()); + auto expected = BuildScan(table_path, nullptr, /*bucket=*/0, predicate); + expected->WithSnapshot(snapshot_5); + ASSERT_OK_AND_ASSIGN(auto expected_plan, expected->CreatePlan()); + ASSERT_EQ(SortedFileNames(filtered_plan->Files()), SortedFileNames(expected_plan->Files())); + ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), manifest_gets); + + // Eviction must rebuild from source; an exact-hit shortcut cannot survive invalidation. + cache->InvalidateAll(); + auto rebuilt = BuildScan(table_path, cache, /*bucket=*/0); + rebuilt->WithSnapshot(snapshot_5); + ASSERT_OK_AND_ASSIGN(auto rebuilt_plan, rebuilt->CreatePlan()); + ASSERT_EQ(first_file_names, SortedFileNames(rebuilt_plan->Files())); + ASSERT_GT(cache->GetCount(CacheKind::MANIFEST), manifest_gets); } TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) { @@ -998,9 +1032,10 @@ TEST(AppendOnlyFileStoreScanTest, SnapshotCacheRebuildsCorruptValues) { segment.PutValue(offset, -1); } MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); - out.WriteValue(corruption == "magic" ? 0 : 0x534d4543); + out.WriteValue(corruption == "magic" ? 0 : 0x534d4544); out.WriteValue(1); out.WriteValue(snapshot.Id()); + out.WriteValue(1); out.WriteValue(1); ASSERT_OK(row_serializer.Serialize(row, &out)); ASSERT_EQ(out.Segments().size(), 1); @@ -1078,7 +1113,11 @@ TEST(AppendOnlyFileStoreScanTest, PartitionPruningAcrossCacheModes) { scan->partition_filter_ = counting; ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); ASSERT_EQ(counting->row_calls_.load(), 2); - ASSERT_GT(counting->stats_calls_.load(), 0); + if (mode == 2 && scans > 0) { + ASSERT_EQ(counting->stats_calls_.load(), 0); + } else { + ASSERT_GT(counting->stats_calls_.load(), 0); + } ASSERT_EQ(plan->Files().size(), partition == 99 ? 0 : 1); if (partition != 99) { ASSERT_EQ(plan->Files()[0].Partition().GetInt(0), partition); @@ -1156,7 +1195,7 @@ TEST(AppendOnlyFileStoreScanTest, PartitionPruningSkipsExcludedMetadataInCreateP if (mode == 2) { // Seed the complete snapshot directly: rebuilding must decode every partition. SnapshotLiveManifestEntries entries(8); - entries.Put(1, {kept, excluded}); + entries.Put(1, {kept, excluded}, /*total_data_files=*/2); ASSERT_OK(scan->StoreSnapshotLiveManifestEntries(0, entries)); } for (int32_t attempt = 0; attempt < 2; ++attempt) { diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp index 4e508bab2..ca2de308a 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -158,8 +158,8 @@ Result> FileStoreScan::ReadPartitionEntries() const PAIMON_RETURN_NOT_OK( ReadManifests(&snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas)); std::vector manifest_entries; - PAIMON_RETURN_NOT_OK(ReadFileEntries(filtered_manifest_file_metas, &manifest_entries, - /*apply_scan_filter=*/true)); + PAIMON_RETURN_NOT_OK(ReadFileEntries(filtered_manifest_file_metas, /*apply_scan_filter=*/true, + &manifest_entries)); std::unordered_map partitions; PAIMON_RETURN_NOT_OK(PartitionEntry::Merge(manifest_entries, &partitions)); @@ -178,11 +178,10 @@ Result> FileStoreScan::CreatePlan() cons Duration manifest_read_duration; metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION, 0); metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION, 0); - std::optional snapshot; + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, ReadSnapshot()); + int64_t all_data_files = 0; std::vector all_manifest_file_metas; std::vector filtered_manifest_file_metas; - PAIMON_RETURN_NOT_OK( - ReadManifests(&snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas)); std::vector manifest_entries; std::optional cache_bucket = bucket_filter_; @@ -197,11 +196,19 @@ Result> FileStoreScan::CreatePlan() cons uint64_t lazy_decode_scanned_rows = 0; bool snapshot_cache_hit = false; if (use_snapshot_live_manifest_cache) { - PAIMON_ASSIGN_OR_RAISE(lazy_decode_scanned_rows, - ReadManifestEntriesWithCache( - snapshot.value(), all_manifest_file_metas, cache_bucket.value(), - &manifest_entries, &snapshot_cache_hit)); + PAIMON_ASSIGN_OR_RAISE( + lazy_decode_scanned_rows, + ReadManifestEntriesWithCache(snapshot.value(), cache_bucket.value(), + &all_manifest_file_metas, &filtered_manifest_file_metas, + &manifest_entries, &snapshot_cache_hit, &all_data_files)); } else { + PAIMON_RETURN_NOT_OK( + ReadManifestLists(snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas)); + all_data_files = + std::accumulate(all_manifest_file_metas.begin(), all_manifest_file_metas.end(), + int64_t{0}, [](int64_t sum, const ManifestFileMeta& meta) { + return sum + meta.NumAddedFiles() - meta.NumDeletedFiles(); + }); lazy_decode_scanned_rows = std::accumulate( filtered_manifest_file_metas.begin(), filtered_manifest_file_metas.end(), uint64_t{0}, [](uint64_t sum, const ManifestFileMeta& meta) { @@ -240,11 +247,6 @@ Result> FileStoreScan::CreatePlan() cons entry = entry.CopyWithoutStats(); } } - const int64_t all_data_files = std::accumulate( - all_manifest_file_metas.begin(), all_manifest_file_metas.end(), int64_t{0}, - [](const int64_t sum, const ManifestFileMeta& manifest_file_meta) { - return sum + manifest_file_meta.NumAddedFiles() - manifest_file_meta.NumDeletedFiles(); - }); const uint64_t scan_duration_ms = duration.Get(); metrics_->SetCounter(ScanMetrics::LAST_SCAN_DURATION, scan_duration_ms); metrics_->ObserveHistogram(ScanMetrics::SCAN_DURATION, static_cast(scan_duration_ms)); @@ -276,27 +278,31 @@ Result> FileStoreScan::CreatePlan() cons std::move(manifest_entries)); } -Status FileStoreScan::ReadManifests(std::optional* snapshot_ptr, - std::vector* all_manifests_ptr, - std::vector* filter_manifests_ptr) const { - auto& snapshot = *snapshot_ptr; - auto& all_manifests = *all_manifests_ptr; - auto& filtered_manifests = *filter_manifests_ptr; - if (specified_snapshot_ != std::nullopt) { - snapshot = specified_snapshot_; - } else { - PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager_->LatestSnapshot()); +Result> FileStoreScan::ReadSnapshot() const { + if (specified_snapshot_) { + return specified_snapshot_; } - if (snapshot == std::nullopt) { - all_manifests = std::vector(); - filtered_manifests = std::vector(); + return snapshot_manager_->LatestSnapshot(); +} + +Status FileStoreScan::ReadManifests(std::optional* snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const { + PAIMON_ASSIGN_OR_RAISE(*snapshot, ReadSnapshot()); + return ReadManifestLists(*snapshot, all_manifests, filtered_manifests); +} + +Status FileStoreScan::ReadManifestLists(const std::optional& snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const { + if (!snapshot) { return Status::OK(); } - PAIMON_RETURN_NOT_OK(ReadManifestsWithSnapshot(snapshot.value(), &all_manifests)); - for (const auto& meta : all_manifests) { - PAIMON_ASSIGN_OR_RAISE(bool filter_meta_result, FilterManifestFileMeta(meta)); - if (filter_meta_result) { - filtered_manifests.push_back(meta); + PAIMON_RETURN_NOT_OK(ReadManifestsWithSnapshot(snapshot.value(), all_manifests)); + for (const auto& meta : *all_manifests) { + PAIMON_ASSIGN_OR_RAISE(bool keep, FilterManifestFileMeta(meta)); + if (keep) { + filtered_manifests->push_back(meta); } } return Status::OK(); @@ -343,8 +349,8 @@ Status FileStoreScan::ReadManifestsWithSnapshot(const Snapshot& snapshot, } Status FileStoreScan::ReadFileEntries(const std::vector& manifest_metas, - std::vector* manifest_entries, - bool apply_scan_filter) const { + bool apply_scan_filter, + std::vector* manifest_entries) const { std::vector>>> futures; for (const auto& meta : manifest_metas) { auto read_meta_task = [this, meta, @@ -389,8 +395,9 @@ Status FileStoreScan::ReadManifestEntries(const std::vector& m // can be returned directly; cache misses rebuild the target snapshot bucket from the target // snapshot's data manifests. Result FileStoreScan::ReadManifestEntriesWithCache( - const Snapshot& snapshot, const std::vector& all_manifest_metas, - int32_t bucket, std::vector* manifest_entries, bool* cache_hit) const { + const Snapshot& snapshot, int32_t bucket, std::vector* all_manifest_metas, + std::vector* filtered_manifest_metas, + std::vector* manifest_entries, bool* cache_hit, int64_t* all_data_files) const { Duration cache_load_duration; const int32_t max_snapshots = core_options_.GetScanManifestEntryCacheMaxSnapshots(); SnapshotLiveManifestEntries cached_entries(max_snapshots); @@ -413,6 +420,7 @@ Result FileStoreScan::ReadManifestEntriesWithCache( } if (selected.ok() && selected.value()) { *cache_hit = true; + *all_data_files = selected.value()->total_data_files; partition_filtered = static_cast(row_filter); candidate_entries = selected.value()->candidate_entries; *manifest_entries = std::move(selected.value()->entries); @@ -432,8 +440,16 @@ Result FileStoreScan::ReadManifestEntriesWithCache( metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION, static_cast(cache_load_duration_ms)); if (!*cache_hit) { + // The selected snapshot is unchanged. Exact hits skip both lists and their decoding. + PAIMON_RETURN_NOT_OK( + ReadManifestLists(snapshot, all_manifest_metas, filtered_manifest_metas)); + *all_data_files = + std::accumulate(all_manifest_metas->begin(), all_manifest_metas->end(), int64_t{0}, + [](int64_t sum, const ManifestFileMeta& meta) { + return sum + meta.NumAddedFiles() - meta.NumDeletedFiles(); + }); std::vector bucket_manifest_metas; - for (const auto& meta : all_manifest_metas) { + for (const auto& meta : *all_manifest_metas) { if ((!bucket_filter_ && bucket_selector_) || MayContainBucket(meta, bucket)) { bucket_manifest_metas.push_back(meta); } @@ -442,7 +458,7 @@ Result FileStoreScan::ReadManifestEntriesWithCache( ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket, manifest_entries)); candidate_entries = manifest_entries->size(); std::vector cache_entries = *manifest_entries; - cached_entries.Put(snapshot.Id(), std::move(cache_entries)); + cached_entries.Put(snapshot.Id(), std::move(cache_entries), *all_data_files); Duration cache_store_duration; PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket, cached_entries)); const uint64_t cache_store_duration_ms = cache_store_duration.Get(); @@ -534,7 +550,7 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries( std::vector unmerged_entries; std::vector entries; - PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, &entries, /*apply_scan_filter=*/false)); + PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, /*apply_scan_filter=*/false, &entries)); unmerged_entries.reserve(entries.size()); for (auto& entry : entries) { if (entry.Bucket() == bucket || @@ -566,14 +582,14 @@ Status FileStoreScan::ReadAndMergeFileEntries(const std::vector* merged_entries) const { std::vector unmerged_entries; PAIMON_RETURN_NOT_OK( - ReadFileEntries(manifest_metas, &unmerged_entries, /*apply_scan_filter=*/true)); + ReadFileEntries(manifest_metas, /*apply_scan_filter=*/true, &unmerged_entries)); return MergeLiveEntries(unmerged_entries, merged_entries); } Status FileStoreScan::ReadAndNoMergeFileEntries( const std::vector& manifest_metas, std::vector* manifest_entries) const { - return ReadFileEntries(manifest_metas, manifest_entries, /*apply_scan_filter=*/true); + return ReadFileEntries(manifest_metas, /*apply_scan_filter=*/true, manifest_entries); } bool FileStoreScan::MayContainBucket(const ManifestFileMeta& manifest, int32_t bucket) const { diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index 25911d0e5..df8c6d614 100644 --- a/src/paimon/core/operation/file_store_scan.h +++ b/src/paimon/core/operation/file_store_scan.h @@ -276,9 +276,13 @@ class FileStoreScan { const std::shared_ptr& data_schema) const; private: - Status ReadManifests(std::optional* snapshot_ptr, - std::vector* all_manifests_ptr, - std::vector* filtered_manifests_ptr) const; + Status ReadManifests(std::optional* snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const; + Result> ReadSnapshot() const; + Status ReadManifestLists(const std::optional& snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const; Status ReadManifestsWithSnapshot(const Snapshot& snapshot, std::vector* manifests) const; @@ -289,8 +293,10 @@ class FileStoreScan { /// Read request-filtered entries while caching complete bucket snapshots. /// @return Number of entries considered before request-specific filtering. Result ReadManifestEntriesWithCache( - const Snapshot& snapshot, const std::vector& bucket_manifest_metas, - int32_t bucket, std::vector* manifest_entries, bool* cache_hit) const; + const Snapshot& snapshot, int32_t bucket, std::vector* all_manifest_metas, + std::vector* filtered_manifest_metas, + std::vector* manifest_entries, bool* cache_hit, + int64_t* all_data_files) const; std::shared_ptr CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const; Status StoreSnapshotLiveManifestEntries(int32_t bucket, const SnapshotLiveManifestEntries& entries) const; @@ -306,8 +312,8 @@ class FileStoreScan { std::vector* manifest_entries) const; Status ReadFileEntries(const std::vector& manifest_metas, - std::vector* manifest_entries, - bool apply_scan_filter) const; + bool apply_scan_filter, + std::vector* manifest_entries) const; bool MayContainBucket(const ManifestFileMeta& manifest, int32_t bucket) const; diff --git a/src/paimon/core/operation/file_store_scan_test.cpp b/src/paimon/core/operation/file_store_scan_test.cpp index 5c53f138d..eb49f3541 100644 --- a/src/paimon/core/operation/file_store_scan_test.cpp +++ b/src/paimon/core/operation/file_store_scan_test.cpp @@ -232,8 +232,8 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) { manifest_entries.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0, /*total_buckets=*/1, file1); SnapshotLiveManifestEntries entries(/*max_snapshots=*/2); - entries.Put(/*snapshot_id=*/1, std::move(manifest_entries)); - entries.Put(/*snapshot_id=*/3, {}); + entries.Put(/*snapshot_id=*/1, std::move(manifest_entries), /*total_data_files=*/1); + entries.Put(/*snapshot_id=*/3, {}, /*total_data_files=*/0); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(GetDefaultPool())); ASSERT_OK_AND_ASSIGN(auto deserialized, @@ -273,9 +273,9 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { BinaryRowGenerator::GenerateRow({partition}, pool.get()), 0, 1, file); } SnapshotLiveManifestEntries entries(3); - entries.Put(1, {records[0]}); - entries.Put(3, {records[1], records[2]}); - entries.Put(5, {}); + entries.Put(1, {records[0]}, 1); + entries.Put(3, {records[1], records[2]}, 17); + entries.Put(5, {}, 0); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool)); const MemorySegment segment = MemorySegment::Wrap(bytes); ASSERT_OK_AND_ASSIGN(auto first, @@ -300,6 +300,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { pool)); ASSERT_TRUE(selected); ASSERT_EQ(selected->candidate_entries, 2); + ASSERT_EQ(selected->total_data_files, 17); ASSERT_EQ(filter_calls, 2); ASSERT_EQ(selected->entries, std::vector({records[2]})); ASSERT_OK_AND_ASSIGN(auto absent, @@ -326,7 +327,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { pool), "injected filter error"); - entries.Put(7, {}); + entries.Put(7, {}, 0); ASSERT_EQ(entries.Size(), 3); ASSERT_OK_AND_ASSIGN(auto updated, entries.Serialize(pool)); // Put must evict the oldest snapshot even when the reader permits more snapshots. @@ -336,6 +337,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { ASSERT_OK_AND_ASSIGN(auto retained, SnapshotLiveManifestEntries::ReadSnapshot( MemorySegment::Wrap(updated), 3, 4, nullptr, pool)); ASSERT_TRUE(retained); + ASSERT_EQ(retained->total_data_files, 17); ASSERT_EQ(retained->entries, std::vector({records[1], records[2]})); } @@ -349,7 +351,7 @@ TEST_F(FileStoreScanTest, SkipDecodingFilteredManifestEntries) { ManifestEntry excluded(FileKind(static_cast(2)), BinaryRowGenerator::GenerateRow({10}, pool.get()), 0, 1, file); SnapshotLiveManifestEntries entries(1); - entries.Put(1, {excluded}); + entries.Put(1, {excluded}, 1); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool)); const MemorySegment segment = MemorySegment::Wrap(bytes); ASSERT_OK_AND_ASSIGN(auto selected, SnapshotLiveManifestEntries::ReadSnapshot( @@ -373,9 +375,10 @@ TEST_F(FileStoreScanTest, RejectUnsupportedSnapshotEntryVersionBeforeFiltering) BinaryRow row = BinaryRowGenerator::GenerateRow({3}, pool.get()); BinaryRowSerializer serializer(1, pool); MemorySegmentOutputStream out(64, pool); - out.WriteValue(0x534d4543); + out.WriteValue(0x534d4544); out.WriteValue(1); out.WriteValue(1); + out.WriteValue(7); out.WriteValue(1); ASSERT_OK(serializer.Serialize(row, &out)); bool filter_called = false; @@ -400,16 +403,17 @@ TEST_F(FileStoreScanTest, RejectInvalidSnapshotSelectionHeader) { out.WriteValue(magic); out.WriteValue(snapshots); out.WriteValue(1); + out.WriteValue(7); out.WriteValue(entries); return out.Segments().front(); }; auto segment = header(0, 1, 0); ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool), "invalid snapshot live manifest entries magic"); - segment = header(0x534d4543, -1, 0); + segment = header(0x534d4544, -1, 0); ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool), "snapshot count is negative"); - segment = header(0x534d4543, 1, -1); + segment = header(0x534d4544, 1, -1); ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool), "entry count is negative"); auto truncated = MemorySegment::WrapView(segment.Data(), 7); From b7565008d8f0a62e4dcfcbc271b919eab7fa8e4a Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Mon, 26 Oct 2026 21:42:50 +0700 Subject: [PATCH 4/8] feat(rest): support DLF openapi-v4 signing algorithm (#469) --- docs/source/user_guide/catalog.rst | 9 +- include/paimon/catalog_options.h | 2 +- src/paimon/rest/dlf_auth.cpp | 334 +++++++++++++++++++++++++++-- src/paimon/rest/dlf_auth.h | 51 ++++- src/paimon/rest/dlf_auth_test.cpp | 271 ++++++++++++++++++++++- 5 files changed, 635 insertions(+), 32 deletions(-) diff --git a/docs/source/user_guide/catalog.rst b/docs/source/user_guide/catalog.rst index aa42c2d77..a3f4bcbee 100644 --- a/docs/source/user_guide/catalog.rst +++ b/docs/source/user_guide/catalog.rst @@ -67,8 +67,11 @@ registered on the REST server. The catalog is configured through the the role from the metadata endpoint when it is omitted. * ``dlf.signing-algorithm``: ``default`` selects DLF4-HMAC-SHA256 for DLF VPC endpoints and ``openapi`` selects ROA HMAC-SHA1 for DlfNext OpenAPI endpoints. - When omitted, an endpoint containing ``dlfnext`` selects ``openapi`` and other - endpoints select ``default``. + ``openapi-v4`` selects ACS4-HMAC-SHA256 for DlfNext OpenAPI endpoints. When a + request matches a registered DlfNext operation, it also names that operation's + API in the ``x-acs-action`` header; other requests are signed without it. + Without this option, an endpoint containing ``dlfnext`` selects ``openapi`` and + other endpoints select ``default``; ``openapi-v4`` is only used when configured. * ``table-default.``: table option defaults applied when a created table left ```` unset. * ``header.``: sent as the ```` http header on every request to the @@ -103,6 +106,8 @@ are reloaded when less than one hour of validity remains. {"dlf.access-key-secret", ""}, // Optional for temporary credentials: {"dlf.security-token", ""}, + // Optional for endpoints that require ACS4-HMAC-SHA256 signing: + {"dlf.signing-algorithm", "openapi-v4"}, }; On creation the catalog queries the server's ``/v1/config`` endpoint and merges diff --git a/include/paimon/catalog_options.h b/include/paimon/catalog_options.h index f781fa189..c1862ad9d 100644 --- a/include/paimon/catalog_options.h +++ b/include/paimon/catalog_options.h @@ -62,7 +62,7 @@ struct PAIMON_EXPORT CatalogOptions { /// "dlf.token-ecs-role-name" - Optional ECS RAM role name. static const char DLF_TOKEN_ECS_ROLE_NAME[]; - /// "dlf.signing-algorithm" - DLF signer ("default" or "openapi"). + /// "dlf.signing-algorithm" - DLF signer ("default", "openapi" or "openapi-v4"). static const char DLF_SIGNING_ALGORITHM[]; /// "dlf.oss-endpoint" - OSS endpoint that overrides the "fs.oss.endpoint" of a data diff --git a/src/paimon/rest/dlf_auth.cpp b/src/paimon/rest/dlf_auth.cpp index 6a4906279..ae5b6f0c9 100644 --- a/src/paimon/rest/dlf_auth.cpp +++ b/src/paimon/rest/dlf_auth.cpp @@ -21,6 +21,7 @@ #include +#include #include #include #include @@ -72,6 +73,13 @@ constexpr const char kAcsSignatureNonceHeader[] = "x-acs-signature-nonce"; constexpr const char kAcsSignatureVersionHeader[] = "x-acs-signature-version"; constexpr const char kAcsVersionHeader[] = "x-acs-version"; constexpr const char kAcsSecurityTokenHeader[] = "x-acs-security-token"; +constexpr const char kAcsDateHeader[] = "x-acs-date"; +constexpr const char kAcsContentSha256Header[] = "x-acs-content-sha256"; +constexpr const char kAcsActionHeader[] = "x-acs-action"; +constexpr const char kAcs4Algorithm[] = "ACS4-HMAC-SHA256"; +// SHA-256 of the empty string, the payload hash of a request without a body. +constexpr const char kEmptyBodySha256[] = + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; Result RequiredNonEmptyOption(const std::map& options, const std::string& key) { @@ -119,6 +127,15 @@ Result FormatDlfTime(std::chrono::system_clock::time_point time) { return std::string(buffer.data()); } +Result FormatAcsTime(std::chrono::system_clock::time_point time) { + PAIMON_ASSIGN_OR_RAISE(std::tm utc, ToUtc(time)); + std::array buffer{}; + if (std::strftime(buffer.data(), buffer.size(), "%Y-%m-%dT%H:%M:%SZ", &utc) == 0) { + return Status::Invalid("failed to format DLF OpenAPI V4 signing time"); + } + return std::string(buffer.data()); +} + Result FormatRfc1123Time(std::chrono::system_clock::time_point time) { PAIMON_ASSIGN_OR_RAISE(std::tm utc, ToUtc(time)); static constexpr std::array kWeekdays = {"Sun", "Mon", "Tue", "Wed", @@ -226,6 +243,17 @@ std::string HexEncode(const Bytes& value) { return encoded; } +// The signing key shared by the DLF4 and ACS4 schemes, derived from the secret through the +// date, the region and the product. +Result DeriveSigningKey(const std::string& access_key_secret, const std::string& date, + const std::string& region) { + PAIMON_ASSIGN_OR_RAISE(Bytes date_key, + Hmac(EVP_sha256(), ToBytes("aliyun_v4" + access_key_secret), date)); + PAIMON_ASSIGN_OR_RAISE(Bytes region_key, Hmac(EVP_sha256(), date_key, region)); + PAIMON_ASSIGN_OR_RAISE(Bytes product_key, Hmac(EVP_sha256(), region_key, "DlfNext")); + return Hmac(EVP_sha256(), product_key, "aliyun_v4_request"); +} + Result Base64Encode(const Bytes& value) { if (value.size() > static_cast(INT_MAX)) { return Status::Invalid("DLF digest is too large to encode"); @@ -343,6 +371,85 @@ Result GenerateNonce(std::chrono::system_clock::time_point now) { return fmt::format("{}{}{}", uuid, millis, thread_id.str()); } +struct Acs4CanonicalHeaders { + std::string canonical_headers; + std::string signed_headers; +}; + +// The headers the ACS4 scheme signs: "host", "content-type" and every "x-acs-*" header, named +// in lower case with trimmed values and sorted by name. +Acs4CanonicalHeaders Acs4CanonicalizeHeaders(const DlfRequestSigner::Headers& headers) { + std::map sorted; + for (const auto& [key, value] : headers) { + std::string lower_key = StringUtils::ToLowerCase(key); + if (StringUtils::StartsWith(lower_key, "x-acs-") || lower_key == "host" || + lower_key == "content-type") { + sorted[lower_key] = Trimmed(value); + } + } + Acs4CanonicalHeaders result; + for (const auto& [key, value] : sorted) { + result.canonical_headers += key + ":" + value + "\n"; + if (!result.signed_headers.empty()) { + result.signed_headers += ";"; + } + result.signed_headers += key; + } + return result; +} + +// The query parameters sorted by name and RFC 3986 encoded. `RestAuthParameter` holds the +// values url-encoded once already, so they are decoded first: encoding them again would sign a +// form the gateway cannot rebuild from the wire. +std::string Acs4CanonicalQueryString(const RestAuthParameter& parameter) { + std::string canonical; + bool first = true; + for (const auto& [key, value] : parameter.parameters) { + if (!first) { + canonical += "&"; + } + canonical += UrlUtils::PercentEncode(key) + "="; + if (!value.empty()) { + canonical += UrlUtils::PercentEncode(UrlUtils::DecodeString(value)); + } + first = false; + } + return canonical; +} + +struct SplitOperation { + std::vector template_segments; + std::string action; +}; + +// The registered operations grouped by method, with their path templates split once. +const std::map>& OperationsByMethod() { + static const std::map> kOperationsByMethod = [] { + std::map> operations_by_method; + for (const DlfOpenApiActions::Operation& operation : DlfOpenApiActions::Operations()) { + operations_by_method[operation.method].push_back( + {StringUtils::Split(operation.path_template, "/", /*ignore_empty=*/false), + operation.action}); + } + return operations_by_method; + }(); + return kOperationsByMethod; +} + +bool MatchesPathTemplate(const std::vector& template_segments, + const std::vector& segments) { + if (segments.size() != template_segments.size()) { + return false; + } + for (size_t i = 0; i < template_segments.size(); ++i) { + bool variable = StringUtils::StartsWith(template_segments[i], "{"); + if (variable ? segments[i].empty() : template_segments[i] != segments[i]) { + return false; + } + } + return true; +} + Result> CreateSigner(const std::string& algorithm, const std::string& region) { if (algorithm == DlfDefaultSigner::kIdentifier) { @@ -351,9 +458,13 @@ Result> CreateSigner(const std::string& algori if (algorithm == DlfOpenApiSigner::kIdentifier) { return std::make_unique(); } - return Status::Invalid(fmt::format( - "unsupported DLF signing algorithm '{}', supported values are 'default' and 'openapi'", - algorithm)); + if (algorithm == DlfOpenApiV4Signer::kIdentifier) { + return std::make_unique(region); + } + return Status::Invalid( + fmt::format("unsupported DLF signing algorithm '{}', supported values are 'default', " + "'openapi' and 'openapi-v4'", + algorithm)); } } // namespace @@ -531,24 +642,21 @@ Result DlfDefaultSigner::SignHeaders( Result DlfDefaultSigner::Authorization(const RestAuthParameter& parameter, const DlfToken& token, const std::string& host, - const Headers& sign_headers) const { - PAIMON_ASSIGN_OR_RAISE(std::string date_time, RequiredHeader(sign_headers, kDlfDateHeader)); + Headers* sign_headers) const { + const Headers& headers = *sign_headers; + PAIMON_ASSIGN_OR_RAISE(std::string date_time, RequiredHeader(headers, kDlfDateHeader)); if (date_time.size() < 8) { return Status::Invalid("DLF signing date is invalid"); } std::string date = date_time.substr(0, 8); std::string scope = fmt::format("{}/{}/DlfNext/aliyun_v4_request", date, region_); - std::string canonical_request = DefaultCanonicalRequest(parameter, sign_headers); + std::string canonical_request = DefaultCanonicalRequest(parameter, headers); PAIMON_ASSIGN_OR_RAISE(Bytes canonical_hash, Digest(EVP_sha256(), canonical_request)); std::string string_to_sign = fmt::format("DLF4-HMAC-SHA256\n{}\n{}\n{}", date_time, scope, HexEncode(canonical_hash)); - PAIMON_ASSIGN_OR_RAISE( - Bytes date_key, - Hmac(EVP_sha256(), ToBytes("aliyun_v4" + token.GetAccessKeySecret()), date)); - PAIMON_ASSIGN_OR_RAISE(Bytes region_key, Hmac(EVP_sha256(), date_key, region_)); - PAIMON_ASSIGN_OR_RAISE(Bytes service_key, Hmac(EVP_sha256(), region_key, "DlfNext")); - PAIMON_ASSIGN_OR_RAISE(Bytes signing_key, Hmac(EVP_sha256(), service_key, "aliyun_v4_request")); + PAIMON_ASSIGN_OR_RAISE(Bytes signing_key, + DeriveSigningKey(token.GetAccessKeySecret(), date, region_)); PAIMON_ASSIGN_OR_RAISE(Bytes signature, Hmac(EVP_sha256(), signing_key, string_to_sign)); return fmt::format("DLF4-HMAC-SHA256 Credential={}/{},Signature={}", token.GetAccessKeyId(), scope, HexEncode(signature)); @@ -579,28 +687,212 @@ Result DlfOpenApiSigner::SignHeaders( Result DlfOpenApiSigner::Authorization(const RestAuthParameter& parameter, const DlfToken& token, const std::string& host, - const Headers& sign_headers) const { - PAIMON_ASSIGN_OR_RAISE(std::string accept, RequiredHeader(sign_headers, kOpenApiAcceptHeader)); - PAIMON_ASSIGN_OR_RAISE(std::string date, RequiredHeader(sign_headers, kOpenApiDateHeader)); + Headers* sign_headers) const { + const Headers& headers = *sign_headers; + PAIMON_ASSIGN_OR_RAISE(std::string accept, RequiredHeader(headers, kOpenApiAcceptHeader)); + PAIMON_ASSIGN_OR_RAISE(std::string date, RequiredHeader(headers, kOpenApiDateHeader)); std::string content_md5; - auto md5_iter = sign_headers.find(kContentMd5Header); - if (md5_iter != sign_headers.end()) { + auto md5_iter = headers.find(kContentMd5Header); + if (md5_iter != headers.end()) { content_md5 = md5_iter->second; } std::string content_type; - auto type_iter = sign_headers.find(kContentTypeHeader); - if (type_iter != sign_headers.end()) { + auto type_iter = headers.find(kContentTypeHeader); + if (type_iter != headers.end()) { content_type = type_iter->second; } std::string string_to_sign = parameter.method + "\n" + accept + "\n" + content_md5 + "\n" + content_type + "\n" + date + - "\n" + OpenApiCanonicalizedHeaders(sign_headers) + OpenApiCanonicalizedResource(parameter); + "\n" + OpenApiCanonicalizedHeaders(headers) + OpenApiCanonicalizedResource(parameter); PAIMON_ASSIGN_OR_RAISE(Bytes signature, Hmac(EVP_sha1(), ToBytes(token.GetAccessKeySecret()), string_to_sign)); PAIMON_ASSIGN_OR_RAISE(std::string encoded_signature, Base64Encode(signature)); return fmt::format("acs {}:{}", token.GetAccessKeyId(), encoded_signature); } +DlfOpenApiV4Signer::DlfOpenApiV4Signer(const std::string& region) : region_(region) {} + +Result DlfOpenApiV4Signer::SignHeaders( + const std::string& body, std::chrono::system_clock::time_point now, + const std::optional& security_token, const std::string& host) const { + if (host.empty()) { + return Status::Invalid("DLF OpenAPI signing host is empty"); + } + PAIMON_ASSIGN_OR_RAISE(std::string date, FormatAcsTime(now)); + PAIMON_ASSIGN_OR_RAISE(std::string nonce, GenerateNonce(now)); + // "Host" and "Content-Type" keep the spelling `RestApi` and the other signers use, so that + // merging these headers replaces a header of the same name rather than sending a second + // copy of it; the signature names every header in lower case either way. + Headers headers = {{kAcsDateHeader, date}, + {kOpenApiHostHeader, host}, + {kAcsContentSha256Header, kEmptyBodySha256}, + {kAcsSignatureNonceHeader, nonce}, + {kAcsVersionHeader, "2026-01-18"}}; + if (!body.empty()) { + PAIMON_ASSIGN_OR_RAISE(Bytes body_hash, Digest(EVP_sha256(), body)); + headers[kAcsContentSha256Header] = HexEncode(body_hash); + headers[kContentTypeHeader] = kJsonMediaType; + } + if (security_token) { + headers[kAcsSecurityTokenHeader] = security_token.value(); + } + return headers; +} + +Result DlfOpenApiV4Signer::Authorization(const RestAuthParameter& parameter, + const DlfToken& token, + const std::string& host, + Headers* sign_headers) const { + PAIMON_ASSIGN_OR_RAISE(std::string date_time, RequiredHeader(*sign_headers, kAcsDateHeader)); + if (date_time.size() < 10) { + return Status::Invalid("DLF signing date is invalid"); + } + std::string date = date_time.substr(0, 10); + date.erase(std::remove(date.begin(), date.end(), '-'), date.end()); + + // POP gateways may route by the action, so it is signed and sent with the other headers. + std::optional action = + DlfOpenApiActions::Resolve(parameter.method, parameter.resource_path); + if (action) { + (*sign_headers)[kAcsActionHeader] = action.value(); + } + + Acs4CanonicalHeaders canonical_headers = Acs4CanonicalizeHeaders(*sign_headers); + std::string canonical_uri = StringUtils::IsNullOrWhitespaceOnly(parameter.resource_path) + ? std::string("/") + : parameter.resource_path; + auto payload_iter = sign_headers->find(kAcsContentSha256Header); + std::string hashed_payload = + payload_iter == sign_headers->end() ? std::string(kEmptyBodySha256) : payload_iter->second; + std::string canonical_request = parameter.method + "\n" + canonical_uri + "\n" + + Acs4CanonicalQueryString(parameter) + "\n" + + canonical_headers.canonical_headers + "\n" + + canonical_headers.signed_headers + "\n" + hashed_payload; + PAIMON_ASSIGN_OR_RAISE(Bytes canonical_hash, Digest(EVP_sha256(), canonical_request)); + std::string string_to_sign = std::string(kAcs4Algorithm) + "\n" + HexEncode(canonical_hash); + + PAIMON_ASSIGN_OR_RAISE(Bytes signing_key, + DeriveSigningKey(token.GetAccessKeySecret(), date, region_)); + PAIMON_ASSIGN_OR_RAISE(Bytes signature, Hmac(EVP_sha256(), signing_key, string_to_sign)); + return fmt::format( + "{} Credential={}/{}/{}/DlfNext/aliyun_v4_request,SignedHeaders={},Signature={}", + kAcs4Algorithm, token.GetAccessKeyId(), date, region_, canonical_headers.signed_headers, + HexEncode(signature)); +} + +const std::vector& DlfOpenApiActions::Operations() { + // Mirrors the Java `DLFOpenApiActions`, so both clients name every request the same way. + static const std::vector kOperations = { + {"GET", "/v1/config", "GetConfig"}, + {"GET", "/v1/{prefix}/databases", "ListDatabases"}, + {"POST", "/v1/{prefix}/databases", "CreateDatabase"}, + {"GET", "/v1/{prefix}/databases/{database}", "GetDatabase"}, + {"POST", "/v1/{prefix}/databases/{database}", "AlterDatabase"}, + {"DELETE", "/v1/{prefix}/databases/{database}", "DropDatabase"}, + {"POST", "/v1/{prefix}/databases/{database}/register", "RegisterTable"}, + {"GET", "/v1/{prefix}/databases/{database}/tables", "ListTables"}, + {"POST", "/v1/{prefix}/databases/{database}/tables", "CreateTable"}, + {"GET", "/v1/{prefix}/databases/{database}/table-details", "ListTableDetails"}, + {"GET", "/v1/{prefix}/tables", "ListTablesGlobally"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}", "GetTable"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}", "AlterTable"}, + {"DELETE", "/v1/{prefix}/databases/{database}/tables/{table}", "DropTable"}, + {"GET", "/v1/{prefix}/tables/id/{tableId}", "GetTableById"}, + {"POST", "/v1/{prefix}/tables/rename", "RenameTable"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/auth", "AuthTableQuery"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit", "CommitTable"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit-lock", + "AcquireCommitLock"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit-lock/renew", + "RenewCommitLock"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/replace", "ReplaceTable"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/token", "GetTableToken"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/rollback", "RollbackToSnapshot"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/rollback-schema", + "RollbackSchema"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/snapshot", "GetTableSnapshot"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/snapshots", "ListSnapshots"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/snapshots/{version}", + "GetVersionSnapshot"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/schemas", "ListSchemas"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/schemas/{version}", "GetSchema"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/partitions", "ListPartitions"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/partitions", "CreatePartitions"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/partitions/drop", + "DropPartitions"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/partitions/mark", + "MarkDonePartitions"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/partitions/list-by-names", + "ListPartitionsByNames"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/partitions/list-by-filter", + "ListPartitionsByFilter"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/branches", "ListBranches"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/branches", "CreateBranch"}, + {"DELETE", "/v1/{prefix}/databases/{database}/tables/{table}/branches/{branch}", + "DropBranch"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/branches/{branch}/forward", + "FastForwardBranch"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/tags", "ListTags"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/tags", "CreateTag"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/tags/{tag}", "GetTag"}, + {"DELETE", "/v1/{prefix}/databases/{database}/tables/{table}/tags/{tag}", "DropTag"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/consumers", "ListConsumers"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/consumers/reset", + "ResetConsumer"}, + {"GET", "/v1/{prefix}/databases/{database}/tables/{table}/policies", "ListPolicies"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/policies", "CreatePolicy"}, + {"POST", "/v1/{prefix}/databases/{database}/tables/{table}/policies/drop", "DropPolicy"}, + {"GET", "/v1/{prefix}/databases/{database}/views", "ListViews"}, + {"POST", "/v1/{prefix}/databases/{database}/views", "CreateView"}, + {"GET", "/v1/{prefix}/databases/{database}/view-details", "ListViewDetails"}, + {"GET", "/v1/{prefix}/views", "ListViewsGlobally"}, + {"GET", "/v1/{prefix}/databases/{database}/views/{view}", "GetView"}, + {"POST", "/v1/{prefix}/databases/{database}/views/{view}", "AlterView"}, + {"DELETE", "/v1/{prefix}/databases/{database}/views/{view}", "DropView"}, + {"POST", "/v1/{prefix}/views/rename", "RenameView"}, + {"GET", "/v1/{prefix}/databases/{database}/semantic-views", "ListSemanticViews"}, + {"GET", "/v1/{prefix}/databases/{database}/semantic-views/{semanticView}", + "GetSemanticView"}, + {"POST", "/v1/{prefix}/databases/{database}/semantic-views/{semanticView}", + "UpsertSemanticView"}, + {"DELETE", "/v1/{prefix}/databases/{database}/semantic-views/{semanticView}", + "DeleteSemanticView"}, + {"GET", "/v1/{prefix}/databases/{database}/functions", "ListFunctions"}, + {"POST", "/v1/{prefix}/databases/{database}/functions", "CreateFunction"}, + {"GET", "/v1/{prefix}/databases/{database}/function-details", "ListFunctionDetails"}, + {"GET", "/v1/{prefix}/functions", "ListFunctionsGlobally"}, + {"GET", "/v1/{prefix}/databases/{database}/functions/{function}", "GetFunction"}, + {"POST", "/v1/{prefix}/databases/{database}/functions/{function}", "AlterFunction"}, + {"DELETE", "/v1/{prefix}/databases/{database}/functions/{function}", "DropFunction"}, + {"GET", "/v1/{prefix}/permissions", "ListPermissionAssignments"}, + {"POST", "/v1/{prefix}/permissions/grant", "GrantPermissionAssignment"}, + {"POST", "/v1/{prefix}/permissions/revoke", "RevokePermissionAssignment"}, + {"GET", "/v1/{prefix}/labels", "ListLabels"}, + {"POST", "/v1/{prefix}/labels/get", "GetLabel"}, + {"POST", "/v1/{prefix}/labels/upsert", "UpsertLabel"}, + {"POST", "/v1/{prefix}/labels/delete", "DeleteLabel"}, + }; + return kOperations; +} + +std::optional DlfOpenApiActions::Resolve(const std::string& method, + const std::string& resource_path) { + const std::map>& operations_by_method = + OperationsByMethod(); + auto candidates = operations_by_method.find(StringUtils::ToUpperCase(method)); + if (candidates == operations_by_method.end()) { + return std::nullopt; + } + std::vector segments = + StringUtils::Split(resource_path, "/", /*ignore_empty=*/false); + for (const SplitOperation& operation : candidates->second) { + if (MatchesPathTemplate(operation.template_segments, segments)) { + return operation.action; + } + } + return std::nullopt; +} + DlfAuthProvider::DlfAuthProvider(std::unique_ptr token_loader, const std::optional& token, const std::string& host, std::unique_ptr signer, Clock clock) @@ -744,7 +1036,7 @@ Result> DlfAuthProvider::MergeAuthHeader( DlfRequestSigner::Headers sign_headers, signer_->SignHeaders(parameter.data, signing_time, token.GetSecurityToken(), host_)); PAIMON_ASSIGN_OR_RAISE(std::string authorization, - signer_->Authorization(parameter, token, host_, sign_headers)); + signer_->Authorization(parameter, token, host_, &sign_headers)); std::map headers = base_header; for (const auto& [key, value] : sign_headers) { headers[key] = value; diff --git a/src/paimon/rest/dlf_auth.h b/src/paimon/rest/dlf_auth.h index c10b655b7..2f8238f0f 100644 --- a/src/paimon/rest/dlf_auth.h +++ b/src/paimon/rest/dlf_auth.h @@ -124,9 +124,11 @@ class DlfRequestSigner { const std::optional& security_token, const std::string& host) const = 0; + /// Signs the request. `sign_headers` holds the headers `SignHeaders` returned and must not + /// be null; a signer may add headers to it, which it signs and the request must send too. virtual Result Authorization(const RestAuthParameter& parameter, const DlfToken& token, const std::string& host, - const Headers& sign_headers) const = 0; + Headers* sign_headers) const = 0; }; /// DLF4-HMAC-SHA256 signer used by the default DLF VPC endpoint. @@ -142,7 +144,7 @@ class DlfDefaultSigner : public DlfRequestSigner { Result Authorization(const RestAuthParameter& parameter, const DlfToken& token, const std::string& host, - const Headers& sign_headers) const override; + Headers* sign_headers) const override; private: std::string region_; @@ -159,7 +161,50 @@ class DlfOpenApiSigner : public DlfRequestSigner { Result Authorization(const RestAuthParameter& parameter, const DlfToken& token, const std::string& host, - const Headers& sign_headers) const override; + Headers* sign_headers) const override; +}; + +/// ACS4-HMAC-SHA256 signer for DlfNext OpenAPI endpoints, selected only when configured. +/// Unlike the other signers, it adds an "x-acs-action" header to a request that matches a +/// registered operation, naming that operation's API, and signs it with the other headers. +class DlfOpenApiV4Signer : public DlfRequestSigner { + public: + static constexpr const char* kIdentifier = "openapi-v4"; + + explicit DlfOpenApiV4Signer(const std::string& region); + + Result SignHeaders(const std::string& body, std::chrono::system_clock::time_point now, + const std::optional& security_token, + const std::string& host) const override; + + Result Authorization(const RestAuthParameter& parameter, const DlfToken& token, + const std::string& host, + Headers* sign_headers) const override; + + private: + std::string region_; +}; + +/// Maps a REST request to the POP action it is registered under in DlfNext 2026-01-18. +class DlfOpenApiActions { + public: + /// One registered operation. A "{name}" segment of `path_template` matches any one + /// non-empty path segment. + struct Operation { + const char* method; + const char* path_template; + const char* action; + }; + + DlfOpenApiActions() = delete; + + /// Every registered operation, in the order requests are matched against them. + static const std::vector& Operations(); + + /// Returns the action of the request, or `std::nullopt` when no operation is registered + /// for it. The method is matched ignoring case, the path segment by segment. + static std::optional Resolve(const std::string& method, + const std::string& resource_path); }; /// Generates DLF authentication headers and refreshes expiring credentials. diff --git a/src/paimon/rest/dlf_auth_test.cpp b/src/paimon/rest/dlf_auth_test.cpp index 5e82ff506..a6e3a3883 100644 --- a/src/paimon/rest/dlf_auth_test.cpp +++ b/src/paimon/rest/dlf_auth_test.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -36,7 +37,9 @@ #include "gtest/gtest.h" #include "paimon/catalog_options.h" +#include "paimon/rest/resource_paths.h" #include "paimon/testing/utils/testharness.h" +#include "paimon/testing/utils/timezone_guard.h" namespace paimon::test { @@ -48,6 +51,14 @@ std::chrono::system_clock::time_point FixedTime() { return std::chrono::system_clock::from_time_t(1744775086); } +constexpr const char kEmptyBodySha256[] = + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; + +void AssertAction(const std::optional& expected, const std::string& method, + const std::string& path) { + ASSERT_EQ(expected, DlfOpenApiActions::Resolve(method, path)) << method << " " << path; +} + class SequenceTokenLoader : public DlfTokenLoader { public: explicit SequenceTokenLoader( @@ -159,7 +170,7 @@ TEST(DlfDefaultSignerTest, SignsJavaCompatibleRequest) { ASSERT_EQ("securityToken", headers.at("x-dlf-security-token")); ASSERT_OK_AND_ASSIGN(std::string authorization, - signer.Authorization(parameter, token, "unused", headers)); + signer.Authorization(parameter, token, "unused", &headers)); ASSERT_EQ( "DLF4-HMAC-SHA256 Credential=YourAccessKeyId/20250416/cn-beijing/" "DlfNext/aliyun_v4_request,Signature=" @@ -180,7 +191,7 @@ TEST(DlfDefaultSignerTest, MatchesJavaGoldenAuthorization) { ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, signer.SignHeaders(body, signing_time, token.GetSecurityToken(), "host")); ASSERT_OK_AND_ASSIGN(std::string authorization, - signer.Authorization(parameter, token, "host", headers)); + signer.Authorization(parameter, token, "host", &headers)); ASSERT_EQ( "DLF4-HMAC-SHA256 Credential=access-key-id/20231203/cn-hangzhou/" "DlfNext/aliyun_v4_request,Signature=" @@ -220,7 +231,7 @@ TEST(DlfOpenApiSignerTest, SignsJavaCompatibleRequest) { ASSERT_EQ("securityToken", headers.at("x-acs-security-token")); ASSERT_OK_AND_ASSIGN(std::string authorization, - signer.Authorization(parameter, token, host, headers)); + signer.Authorization(parameter, token, host, &headers)); ASSERT_EQ("acs YourAccessKeyId:wX4CDPSCtfgYkxdK9tJIO3ez5VI=", authorization); } @@ -234,7 +245,7 @@ TEST(DlfOpenApiSignerTest, DecodesPathAndQueryValuesBeforeSigning) { headers["x-acs-signature-nonce"] = "fixed-nonce"; ASSERT_OK_AND_ASSIGN(std::string authorization, - signer.Authorization(parameter, token, "host", headers)); + signer.Authorization(parameter, token, "host", &headers)); ASSERT_EQ("acs ak:vD+M7291KKoOTvt2gQuD6jPDlw8=", authorization); ASSERT_EQ(0, headers.count("Content-MD5")); ASSERT_EQ(0, headers.count("Content-Type")); @@ -487,7 +498,7 @@ TEST(DlfDefaultSignerTest, SignsSpecialCharQueryLikeJava) { ASSERT_EQ("20231203T121212Z", headers.at("x-dlf-date")); ASSERT_EQ("F2vHiexkcYvt4XOY5o4tmQ==", headers.at("Content-MD5")); ASSERT_OK_AND_ASSIGN(std::string authorization, - signer.Authorization(parameter, token, "host", headers)); + signer.Authorization(parameter, token, "host", &headers)); ASSERT_EQ( "DLF4-HMAC-SHA256 Credential=access-key-id/20231203/cn-hangzhou/" "DlfNext/aliyun_v4_request,Signature=" @@ -579,6 +590,7 @@ TEST(DlfOpenApiSignerTest, ConcurrentNonceGenerationStaysUnique) { TEST(DlfSignerTest, ExposesJavaCompatibleIdentifiers) { ASSERT_EQ(std::string("default"), std::string(DlfDefaultSigner::kIdentifier)); ASSERT_EQ(std::string("openapi"), std::string(DlfOpenApiSigner::kIdentifier)); + ASSERT_EQ(std::string("openapi-v4"), std::string(DlfOpenApiV4Signer::kIdentifier)); } TEST(DlfAuthProviderTest, ParsesSigningAlgorithmFromUriBoundaries) { @@ -618,4 +630,253 @@ TEST(DlfAuthProviderTest, ParsesRegionFromUriVariants) { "could not determine DLF region"); } +TEST(DlfOpenApiV4SignerTest, SignsHeadersWithBody) { + DlfOpenApiV4Signer signer("cn-beijing"); + const std::string body = R"({"CategoryName":"test","CategoryType":"UNSTRUCTURED"})"; + const std::string host = "dlfnext.cn-beijing.aliyuncs.com"; + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders(body, FixedTime(), std::nullopt, host)); + ASSERT_EQ("2025-04-16T03:44:46Z", headers.at("x-acs-date")); + ASSERT_EQ(host, headers.at("Host")); + ASSERT_EQ("d8171130c3edc0f7d4e2827c5dab801a603c9a376b214a9ab6d5d12665c8c5df", + headers.at("x-acs-content-sha256")); + ASSERT_EQ("application/json", headers.at("Content-Type")); + ASSERT_FALSE(headers.at("x-acs-signature-nonce").empty()); + ASSERT_EQ("2026-01-18", headers.at("x-acs-version")); + ASSERT_EQ(6, headers.size()); + + ASSERT_NOK_WITH_MSG(signer.SignHeaders(body, FixedTime(), std::nullopt, "").status(), + "host is empty"); +} + +TEST(DlfOpenApiV4SignerTest, OmitsBodyAndSecurityTokenHeadersWhenAbsent) { + DlfOpenApiV4Signer signer("cn-beijing"); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders("", FixedTime(), std::nullopt, "host")); + ASSERT_EQ(kEmptyBodySha256, headers.at("x-acs-content-sha256")); + ASSERT_EQ(0, headers.count("Content-Type")); + ASSERT_EQ(0, headers.count("x-acs-security-token")); + ASSERT_EQ(5, headers.size()); +} + +TEST(DlfOpenApiV4SignerTest, SignsUtcDateInNonUtcTimezone) { + TimezoneGuard timezone_guard("Asia/Shanghai"); + DlfOpenApiV4Signer signer("cn-beijing"); + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + RestAuthParameter parameter = RestAuthParameter::Create("GET", "/test/path", {}, ""); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders("", std::chrono::system_clock::from_time_t(1744747200), + std::nullopt, "host")); + ASSERT_EQ("2025-04-15T20:00:00Z", headers.at("x-acs-date")); + + ASSERT_OK_AND_ASSIGN(std::string authorization, + signer.Authorization(parameter, token, "host", &headers)); + ASSERT_EQ(0, authorization.find("ACS4-HMAC-SHA256 Credential=TestAKId/20250415/cn-beijing/")); +} + +TEST(DlfOpenApiV4SignerTest, MatchesJavaGoldenSignatureForEncodedQueryValues) { + DlfOpenApiV4Signer signer("cn-hangzhou"); + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + const std::string host = "dlfnext.cn-hangzhou.aliyuncs.com"; + RestAuthParameter parameter = + RestAuthParameter::Create("GET", "/v1/prefix/functions", + {{"pageToken", "a b/c"}, {"functionNamePattern", "func%"}}, ""); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders("", FixedTime(), std::nullopt, host)); + headers["x-acs-signature-nonce"] = "fixed-nonce-for-test"; + + ASSERT_OK_AND_ASSIGN(std::string authorization, + signer.Authorization(parameter, token, host, &headers)); + ASSERT_EQ( + "ACS4-HMAC-SHA256 Credential=TestAKId/20250416/cn-hangzhou/DlfNext/aliyun_v4_request," + "SignedHeaders=host;x-acs-action;x-acs-content-sha256;x-acs-date;" + "x-acs-signature-nonce;x-acs-version," + "Signature=2d5fdb5201901a5cdceee97318366666a88504eeae0b1ce534dbf86889cc00fb", + authorization); + ASSERT_EQ("ListFunctionsGlobally", headers.at("x-acs-action")); +} + +TEST(DlfOpenApiV4SignerTest, MatchesJavaGoldenSignatureWithAction) { + DlfOpenApiV4Signer signer("cn-hangzhou"); + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + const std::string host = "dlfnext.cn-hangzhou.aliyuncs.com"; + const std::string body = R"({"identifier":{"database":"db","object":"t"}})"; + RestAuthParameter parameter = + RestAuthParameter::Create("POST", "/v1/clg-paimon-1/databases/db/tables", {}, body); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders(body, FixedTime(), std::nullopt, host)); + headers["x-acs-signature-nonce"] = "fixed-nonce-for-test"; + + ASSERT_OK_AND_ASSIGN(std::string authorization, + signer.Authorization(parameter, token, host, &headers)); + ASSERT_EQ("CreateTable", headers.at("x-acs-action")); + ASSERT_EQ( + "ACS4-HMAC-SHA256 Credential=TestAKId/20250416/cn-hangzhou/DlfNext/aliyun_v4_request," + "SignedHeaders=content-type;host;x-acs-action;x-acs-content-sha256;" + "x-acs-date;x-acs-signature-nonce;x-acs-version," + "Signature=420206b5263536e6bc271a7a32582b4a820e23c8c58ae692b708c69f9d19e51d", + authorization); +} + +TEST(DlfOpenApiV4SignerTest, SignatureDependsOnRegion) { + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + RestAuthParameter parameter = RestAuthParameter::Create("GET", "/test/path", {}, ""); + DlfOpenApiV4Signer beijing("cn-beijing"); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + beijing.SignHeaders("", FixedTime(), std::nullopt, "host")); + headers["x-acs-signature-nonce"] = "fixed-nonce-for-test"; + ASSERT_OK_AND_ASSIGN(std::string beijing_authorization, + beijing.Authorization(parameter, token, "host", &headers)); + + DlfOpenApiV4Signer hangzhou("cn-hangzhou"); + ASSERT_OK_AND_ASSIGN(std::string hangzhou_authorization, + hangzhou.Authorization(parameter, token, "host", &headers)); + size_t beijing_signature_start = beijing_authorization.find(",Signature="); + size_t hangzhou_signature_start = hangzhou_authorization.find(",Signature="); + ASSERT_NE(std::string::npos, beijing_signature_start); + ASSERT_NE(std::string::npos, hangzhou_signature_start); + ASSERT_NE(beijing_authorization.substr(beijing_signature_start), + hangzhou_authorization.substr(hangzhou_signature_start)); + ASSERT_NE(std::string::npos, beijing_authorization.find("/cn-beijing/DlfNext/")); + ASSERT_NE(std::string::npos, hangzhou_authorization.find("/cn-hangzhou/DlfNext/")); +} + +TEST(DlfOpenApiV4SignerTest, SignsHeaderNamesInLowerCaseAndValuesTrimmed) { + DlfOpenApiV4Signer signer("cn-beijing"); + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + RestAuthParameter parameter = RestAuthParameter::Create("GET", "/test/path", {}, ""); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders("", FixedTime(), std::nullopt, "host")); + headers["x-acs-signature-nonce"] = "fixed-nonce-for-test"; + DlfRequestSigner::Headers renamed_headers = headers; + renamed_headers.erase("x-acs-signature-nonce"); + renamed_headers["X-ACS-SIGNATURE-NONCE"] = " fixed-nonce-for-test "; + + ASSERT_OK_AND_ASSIGN(std::string expected, + signer.Authorization(parameter, token, "host", &headers)); + ASSERT_OK_AND_ASSIGN(std::string actual, + signer.Authorization(parameter, token, "host", &renamed_headers)); + ASSERT_EQ(expected, actual); +} + +TEST(DlfOpenApiV4SignerTest, SignsUnregisteredPathWithoutAction) { + DlfOpenApiV4Signer signer("cn-beijing"); + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + RestAuthParameter parameter = + RestAuthParameter::Create("GET", "/v1/clg-paimon-1/databases/db/tables/t/unknown", {}, ""); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders("", FixedTime(), std::nullopt, "host")); + + ASSERT_OK_AND_ASSIGN(std::string authorization, + signer.Authorization(parameter, token, "host", &headers)); + ASSERT_EQ(0, headers.count("x-acs-action")); + ASSERT_EQ(std::string::npos, authorization.find("x-acs-action")); +} + +TEST(DlfOpenApiV4SignerTest, RequiresSigningDate) { + DlfOpenApiV4Signer signer("cn-beijing"); + DlfToken token("TestAKId", "TestAKSecret", std::nullopt, std::nullopt); + RestAuthParameter parameter = RestAuthParameter::Create("GET", "/test/path", {}, ""); + ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers, + signer.SignHeaders("", FixedTime(), std::nullopt, "host")); + headers["x-acs-date"] = "2025"; + ASSERT_NOK_WITH_MSG(signer.Authorization(parameter, token, "host", &headers).status(), + "signing date is invalid"); + headers.erase("x-acs-date"); + ASSERT_NOK_WITH_MSG(signer.Authorization(parameter, token, "host", &headers).status(), + "x-acs-date"); +} + +TEST(DlfAuthProviderTest, SignsWithOpenApiV4OnlyWhenConfigured) { + std::map options = { + {CatalogOptions::URI, "https://dlfnext.cn-hangzhou.aliyuncs.com"}, + {CatalogOptions::TOKEN_PROVIDER, "dlf"}, + {CatalogOptions::DLF_ACCESS_KEY_ID, "ak"}, + {CatalogOptions::DLF_ACCESS_KEY_SECRET, "sk"}, + {CatalogOptions::DLF_SECURITY_TOKEN, "sts"}}; + ASSERT_EQ("openapi", + DlfAuthProvider::ParseSigningAlgorithmFromUri(options.at(CatalogOptions::URI))); + const std::string body = R"({"name":"db"})"; + RestAuthParameter parameter = + RestAuthParameter::Create("POST", "/v1/clg-paimon-1/databases", {}, body); + const StringMap base_headers = {{"Content-Type", "application/json"}}; + for (const char* algorithm : {DlfDefaultSigner::kIdentifier, DlfOpenApiSigner::kIdentifier}) { + options[CatalogOptions::DLF_SIGNING_ALGORITHM] = algorithm; + ASSERT_OK_AND_ASSIGN(std::unique_ptr provider, AuthProvider::Create(options)); + ASSERT_OK_AND_ASSIGN(StringMap headers, provider->MergeAuthHeader(base_headers, parameter)); + ASSERT_EQ(0, headers.count("x-acs-action")) << algorithm; + } + + options[CatalogOptions::DLF_SIGNING_ALGORITHM] = DlfOpenApiV4Signer::kIdentifier; + ASSERT_OK_AND_ASSIGN(std::unique_ptr provider, AuthProvider::Create(options)); + ASSERT_OK_AND_ASSIGN(StringMap headers, provider->MergeAuthHeader(base_headers, parameter)); + const std::string& authorization = headers.at("Authorization"); + ASSERT_EQ(0, authorization.find("ACS4-HMAC-SHA256 Credential=ak/")); + ASSERT_NE(std::string::npos, + authorization.find("/cn-hangzhou/DlfNext/aliyun_v4_request,SignedHeaders=" + "content-type;host;x-acs-action;x-acs-content-sha256;x-acs-date;" + "x-acs-security-token;x-acs-signature-nonce;x-acs-version,")); + ASSERT_EQ("CreateDatabase", headers.at("x-acs-action")); + ASSERT_EQ("sts", headers.at("x-acs-security-token")); + ASSERT_EQ("application/json", headers.at("Content-Type")); + ASSERT_EQ(0, headers.count("content-type")); + ASSERT_EQ("dlfnext.cn-hangzhou.aliyuncs.com", headers.at("Host")); + ASSERT_EQ(0, headers.count("host")); +} + +TEST(DlfOpenApiActionsTest, EveryOperationResolvesToItself) { + static const std::regex kVariable("\\{[^}]+\\}"); + std::set actions; + for (const DlfOpenApiActions::Operation& operation : DlfOpenApiActions::Operations()) { + AssertAction(operation.action, operation.method, + std::regex_replace(operation.path_template, kVariable, "x1")); + actions.insert(operation.action); + } + ASSERT_EQ(DlfOpenApiActions::Operations().size(), actions.size()); + ASSERT_EQ(74, actions.size()); +} + +TEST(DlfOpenApiActionsTest, ClientPathsResolveToRegisteredActions) { + ResourcePaths paths("clg-paimon-1"); + AssertAction("GetConfig", "GET", ResourcePaths::Config()); + AssertAction("ListDatabases", "GET", paths.Databases()); + AssertAction("CreateDatabase", "POST", paths.Databases()); + AssertAction("GetDatabase", "GET", paths.Database("db")); + AssertAction("DropDatabase", "DELETE", paths.Database("db")); + AssertAction("ListTables", "GET", paths.Tables("db")); + AssertAction("CreateTable", "POST", paths.Tables("db")); + AssertAction("GetTable", "GET", paths.Table("db", "t")); + AssertAction("DropTable", "DELETE", paths.Table("db", "t")); + AssertAction("RenameTable", "POST", paths.RenameTable()); + AssertAction("ListSnapshots", "GET", paths.Snapshots("db", "t")); + AssertAction("CommitTable", "POST", paths.CommitTable("db", "t")); + AssertAction("GetTableSnapshot", "GET", paths.TableSnapshot("db", "t")); + AssertAction("GetTableToken", "GET", paths.TableToken("db", "t")); +} + +TEST(DlfOpenApiActionsTest, MatchesWholeSegmentsAndMethodIgnoringCase) { + ResourcePaths paths("clg-paimon-1"); + AssertAction("GetConfig", "get", ResourcePaths::Config()); + AssertAction("GetDatabase", "GET", paths.Database("tables")); + AssertAction("GetTable", "GET", paths.Table("db", "token")); + AssertAction("GetTableToken", "GET", paths.TableToken("config", "t$snapshots")); + AssertAction("GetTable", "GET", paths.Table("a/b", "c d")); + AssertAction("ListDatabases", "GET", ResourcePaths("rename").Databases()); + AssertAction("DropBranch", "DELETE", "/v1/clg-paimon-1/databases/db/tables/t/branches/forward"); + AssertAction("GetTag", "GET", "/v1/clg-paimon-1/databases/db/tables/t/tags/tags"); +} + +TEST(DlfOpenApiActionsTest, UnregisteredRequestsHaveNoAction) { + ResourcePaths paths("clg-paimon-1"); + AssertAction(std::nullopt, "GET", paths.RenameTable()); + AssertAction(std::nullopt, "DELETE", "/v1/clg-paimon-1/tables/id/tbl-1"); + AssertAction(std::nullopt, "GET", paths.CommitTable("db", "t")); + AssertAction(std::nullopt, "PUT", paths.Table("db", "t")); + AssertAction(std::nullopt, "GET", ResourcePaths::Config() + "/"); + AssertAction(std::nullopt, "GET", "/v1//databases"); + AssertAction(std::nullopt, "GET", ResourcePaths("").Databases()); + AssertAction(std::nullopt, "", ResourcePaths::Config()); + AssertAction(std::nullopt, "GET", ""); +} + } // namespace paimon::test From 575ef4d1294b7065a26a7390be08bf75f4260238 Mon Sep 17 00:00:00 2001 From: lszskye <57179283+lszskye@users.noreply.github.com> Date: Tue, 27 Oct 2026 08:51:53 +0800 Subject: [PATCH 5/8] feat(global-index): add checkpoint cleanup by task id (#465) --- docs/source/user_guide/global_index.rst | 16 ++ .../global_index/global_index_write_task.h | 20 ++- .../global_index_file_manager_test.cpp | 149 ++++++++++++++++-- .../global_index/global_index_write_task.cpp | 60 +++++-- .../index/index_checkpoint_path_factory.h | 11 ++ .../core/utils/file_store_path_factory.cpp | 8 +- .../utils/file_store_path_factory_test.cpp | 24 ++- .../lumina/lumina_global_index_test.cpp | 9 +- test/inte/global_index_test.cpp | 23 ++- 9 files changed, 269 insertions(+), 51 deletions(-) diff --git a/docs/source/user_guide/global_index.rst b/docs/source/user_guide/global_index.rst index f3dca7445..897cfebbb 100644 --- a/docs/source/user_guide/global_index.rst +++ b/docs/source/user_guide/global_index.rst @@ -38,6 +38,22 @@ Global indexes work on top of Data Evolution tables. To use global indexes, your - ``'row-tracking.enabled' = 'true'`` - ``'data-evolution.enabled' = 'true'`` +Cleaning Up Build Checkpoints +----------------------------- + +Checkpoint-enabled index builds remove their checkpoints after a successful build. If a process +crashes, use ``GlobalIndexWriteTask::DeleteCheckpoint`` to remove all checkpoints left by that +task once it is no longer running or being retried: + +.. code-block:: cpp + + PAIMON_RETURN_NOT_OK(paimon::GlobalIndexWriteTask::DeleteCheckpoint( + table_path, "build-task-1", options, file_system)); + +Pass the same table path and task id used in ``WriteIndex``. Task ids must be unique within the +table. The cleanup recursively removes the whole task directory, other tasks' directories and completed +index files are preserved. + Bitmap Index ------------ diff --git a/include/paimon/global_index/global_index_write_task.h b/include/paimon/global_index/global_index_write_task.h index 391c02caf..8c89a9c18 100644 --- a/include/paimon/global_index/global_index_write_task.h +++ b/include/paimon/global_index/global_index_write_task.h @@ -27,6 +27,7 @@ #include "paimon/global_index/indexed_split.h" #include "paimon/memory/memory_pool.h" #include "paimon/result.h" +#include "paimon/status.h" #include "paimon/utils/range.h" #include "paimon/visibility.h" @@ -48,9 +49,9 @@ class PAIMON_EXPORT GlobalIndexWriteTask { /// @param options Index-specific configuration (e.g., false positive rate for bloom /// filters). /// @param task_id When checkpoints are enabled, the caller must provide a non-empty task - /// identifier that uniquely identifies an index build task. Reuse it when - /// retrying the same build. If the source data, build configuration, or - /// build source code changes, the caller must use a new identifier; + /// identifier that uniquely identifies an index build task within the + /// table. Reuse it when retrying the same build. If the source data, build + /// configuration, or build source code changes, use a new identifier; /// otherwise, the index build may fail. Pass nullopt when checkpoints are /// disabled. Index types without checkpoint support ignore this value. /// @param pool Memory pool for temporary allocations during index construction. @@ -65,6 +66,19 @@ class PAIMON_EXPORT GlobalIndexWriteTask { const std::map& options, const std::optional& task_id, const std::shared_ptr& pool, const std::shared_ptr& file_system = nullptr); + + /// Deletes all checkpoints left by a global index build task. The caller must ensure that this + /// task is no longer building or being retried. + /// + /// @param table_path Path to the table root directory used for the build. + /// @param task_id Non-empty, unique task identifier passed to `WriteIndex`. + /// @param options Configuration overrides. Used to load the schema, then merged with + /// schema options, with these overrides taking precedence. + /// @param file_system Specifies the file system. If `nullptr`, use default file system. + /// @return OK if the task directory was deleted or does not exist. + static Status DeleteCheckpoint(const std::string& table_path, const std::string& task_id, + const std::map& options, + const std::shared_ptr& file_system = nullptr); }; } // namespace paimon diff --git a/src/paimon/core/global_index/global_index_file_manager_test.cpp b/src/paimon/core/global_index/global_index_file_manager_test.cpp index ab27ad2e0..d5cce4adb 100644 --- a/src/paimon/core/global_index/global_index_file_manager_test.cpp +++ b/src/paimon/core/global_index/global_index_file_manager_test.cpp @@ -20,6 +20,7 @@ #include "paimon/core/global_index/global_index_file_manager.h" #include +#include #include #include @@ -27,8 +28,11 @@ #include "fmt/format.h" #include "gtest/gtest.h" #include "paimon/common/utils/string_utils.h" +#include "paimon/core/schema/table_schema.h" #include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/defs.h" #include "paimon/fs/local/local_file_system.h" +#include "paimon/global_index/global_index_write_task.h" #include "paimon/testing/utils/testharness.h" #include "paimon/utils/range.h" @@ -46,8 +50,24 @@ class GlobalIndexFileManagerTest : public ::testing::Test { } return LocalFileSystem::ListDir(directory, status_list); } + Status Delete(const std::string& path, bool recursive = true) const override { + delete_paths_.push_back(path); + if (path == fail_delete_path_) { + return Status::IOError("checkpoint delete failed"); + } + return LocalFileSystem::Delete(path, recursive); + } + Result Exists(const std::string& path) const override { + if (path == fail_exists_path_) { + return Status::IOError("checkpoint exists failed"); + } + return LocalFileSystem::Exists(path); + } bool fail_list_ = false; + std::string fail_delete_path_; + std::string fail_exists_path_; mutable int32_t list_count_ = 0; + mutable std::vector delete_paths_; }; void SetUp() override { @@ -80,6 +100,17 @@ class GlobalIndexFileManagerTest : public ::testing::Test { return PathUtil::GetName(uri); } + Status CreateSchema(const std::map& options = {}) const { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr schema, + TableSchema::Create(/*schema_id=*/0, + arrow::schema({arrow::field("vector", arrow::int32())}), + /*partition_keys=*/{}, /*primary_keys=*/{}, options)); + PAIMON_ASSIGN_OR_RAISE(std::string content, schema->ToJsonString()); + return fs_->WriteFile(PathUtil::JoinPath(dir_->Str(), "schema/schema-0"), content, + /*overwrite=*/false); + } + std::shared_ptr fs_ = std::make_shared(); std::unique_ptr dir_; std::shared_ptr path_factory_; @@ -140,8 +171,8 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointNotConfigured) { TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); ASSERT_EQ(fs_->list_count_, 0); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint"); - std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + std::string prefix = "lumina-global-index-vector-10-20-"; std::string previous_path = PathUtil::JoinPath(directory, prefix + "9.index.ckpt"); // The latest sequence must be discovered on first file name allocation, not at construction. ASSERT_OK(fs_->WriteFile(previous_path, "old", /*overwrite=*/false)); @@ -167,8 +198,8 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) { ASSERT_OK_AND_ASSIGN(name, CreateCheckpointFile(restarted)); ASSERT_TRUE(StringUtils::EndsWith(name, "-12.index.ckpt")); - std::string other_task_path = - PathUtil::JoinPath(directory, "lumina-global-index-vector-10-20-task-2-99.index.ckpt"); + std::string other_task_path = PathUtil::JoinPath( + dir_->Str(), "index/checkpoint/task-2/lumina-global-index-vector-10-20-99.index.ckpt"); std::string index_path = manager->ToPath("retained.index"); ASSERT_OK(fs_->WriteFile(other_task_path, "other", /*overwrite=*/false)); ASSERT_OK(fs_->WriteFile(index_path, "index", /*overwrite=*/false)); @@ -205,8 +236,8 @@ TEST_F(GlobalIndexFileManagerTest, TestInitializationFailureCanRetry) { TEST_F(GlobalIndexFileManagerTest, TestLatestCheckpointUsesNumericId) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint"); - std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + std::string prefix = "lumina-global-index-vector-10-20-"; ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix + "9.index.ckpt"), "older", /*overwrite=*/false)); ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix + "10.index.ckpt"), "newer", @@ -223,8 +254,8 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists()); ASSERT_FALSE(exists); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint"); - std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + std::string prefix = "lumina-global-index-vector-10-20-"; ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix + "9.index.ckpt"), "old", /*overwrite=*/false)); ASSERT_OK_AND_ASSIGN(std::unique_ptr input, manager->OpenCheckpointInputStream()); @@ -243,8 +274,8 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) { TEST_F(GlobalIndexFileManagerTest, TestFileIdOverflow) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint"); - std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + std::string prefix = "lumina-global-index-vector-10-20-"; int64_t max_id = std::numeric_limits::max(); ASSERT_OK(fs_->WriteFile( PathUtil::JoinPath(directory, fmt::format("{}{}.index.ckpt", prefix, max_id - 1)), "old", @@ -291,4 +322,102 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteDoesNotResetFileId) { ASSERT_EQ(fs_->list_count_, 2); } +TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpoint) { + ASSERT_OK(CreateSchema()); + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + std::vector matching_names = {"lumina-global-index-vector-10-20-9.index.ckpt", + "lumina-global-index-vector-10-20-10.index.ckpt", + "lumina-global-index-other-10-20-0.index.ckpt", + "other-global-index-vector-10-20-0.index.ckpt", + "lumina-global-index-vector-10-21-0.index.ckpt", + "lumina-global-index-vector-11-20-0.index.ckpt", + "lumina-global-index-vector-10-20-0.index.ckpt.bak", + "incomplete-file", + "nested/unfinished-file"}; + for (const std::string& name : matching_names) { + ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, name), "incomplete checkpoint", + /*overwrite=*/false)); + } + std::vector retained_tasks = {"task-2", "task-1-extended", "task-1-9", "task-1-"}; + std::vector retained_paths; + for (const std::string& task_id : retained_tasks) { + std::string path = PathUtil::JoinPath( + dir_->Str(), "index/checkpoint/" + task_id + "/" + matching_names.front()); + ASSERT_OK(fs_->WriteFile(path, "retained", /*overwrite=*/false)); + retained_paths.push_back(path); + } + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_)); + ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(directory)); + ASSERT_FALSE(exists); + ASSERT_EQ(fs_->delete_paths_, std::vector({directory})); + for (const std::string& path : retained_paths) { + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(path)); + ASSERT_TRUE(exists) << path; + } + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_)); +} + +TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSystem) { + ASSERT_OK(CreateSchema()); + ASSERT_NOK_WITH_MSG(GlobalIndexWriteTask::DeleteCheckpoint( + dir_->Str(), "task-1", {{Options::FILE_SYSTEM, "unregistered-fs"}}), + "Create factory failed with identifier 'unregistered-fs'"); + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{})); + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(directory)); + ASSERT_FALSE(exists); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); + ASSERT_OK(CreateCheckpointFile(manager)); + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{})); + ASSERT_OK_AND_ASSIGN(exists, manager->CheckpointExists()); + ASSERT_FALSE(exists); + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{})); +} + +TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointRejectsInvalidTaskId) { + for (const std::string& invalid : std::vector{"", ".", "..", " ", "a/b", "a\\b", + "a\nb", std::string("a\0b", 3)}) { + ASSERT_NOK( + GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), invalid, /*options=*/{}, fs_)); + } + ASSERT_EQ(fs_->list_count_, 0); + ASSERT_TRUE(fs_->delete_paths_.empty()); +} + +TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointPropagatesFileSystemErrors) { + ASSERT_OK(CreateSchema()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); + ASSERT_OK_AND_ASSIGN(std::string first, CreateCheckpointFile(manager)); + ASSERT_OK_AND_ASSIGN(std::string second, CreateCheckpointFile(manager)); + fs_->fail_list_ = true; + ASSERT_NOK_WITH_MSG( + GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_), + "checkpoint list failed"); + ASSERT_TRUE(fs_->delete_paths_.empty()); + + fs_->fail_list_ = false; + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + fs_->fail_exists_path_ = directory; + ASSERT_NOK_WITH_MSG( + GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_), + "checkpoint exists failed"); + ASSERT_TRUE(fs_->delete_paths_.empty()); + fs_->fail_exists_path_.clear(); + fs_->fail_delete_path_ = directory; + ASSERT_NOK_WITH_MSG( + GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_), + "checkpoint delete failed"); + ASSERT_EQ(fs_->delete_paths_, std::vector({directory})); + for (const std::string& name : {first, second}) { + ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(PathUtil::JoinPath(directory, name))); + ASSERT_TRUE(exists); + } + + fs_->fail_delete_path_.clear(); + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_)); + ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(directory)); + ASSERT_FALSE(exists); +} + } // namespace paimon::test diff --git a/src/paimon/core/global_index/global_index_write_task.cpp b/src/paimon/core/global_index/global_index_write_task.cpp index d5bd13e99..f04d96bb0 100644 --- a/src/paimon/core/global_index/global_index_write_task.cpp +++ b/src/paimon/core/global_index/global_index_write_task.cpp @@ -20,6 +20,7 @@ #include "paimon/global_index/global_index_write_task.h" #include +#include #include "arrow/array/array_nested.h" #include "arrow/c/bridge.h" @@ -34,6 +35,7 @@ #include "paimon/core/casting/casting_utils.h" #include "paimon/core/core_options.h" #include "paimon/core/global_index/global_index_file_manager.h" +#include "paimon/core/index/index_checkpoint_path_factory.h" #include "paimon/core/io/data_increment.h" #include "paimon/core/schema/schema_manager.h" #include "paimon/core/schema/table_schema.h" @@ -47,6 +49,26 @@ #include "paimon/table/source/table_read.h" namespace paimon { namespace { +Result, CoreOptions>> LoadSchemaAndOptions( + const std::string& table_path, const std::map& options, + const std::shared_ptr& file_system) { + PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options, CoreOptions::FromMap(options, file_system)); + SchemaManager schema_manager(tmp_options.GetFileSystem(), table_path); + PAIMON_ASSIGN_OR_RAISE(std::optional> latest_table_schema, + schema_manager.Latest()); + if (latest_table_schema == std::nullopt) { + return Status::Invalid("not found latest schema"); + } + const auto& table_schema = latest_table_schema.value(); + auto final_options = table_schema->Options(); + for (const auto& [key, value] : options) { + final_options[key] = value; + } + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, + CoreOptions::FromMap(final_options, file_system)); + return std::make_pair(table_schema, std::move(core_options)); +} + Result> CreateGlobalIndexer(const std::string& index_type, const CoreOptions& core_options) { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr indexer, @@ -342,22 +364,10 @@ Result> GlobalIndexWriteTask::WriteIndex( std::shared_ptr pool = memory_pool ? memory_pool : GetDefaultPool(); std::shared_ptr arrow_pool = GetArrowPool(pool); - // load schema - PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options, CoreOptions::FromMap(options, file_system)); - SchemaManager schema_manager(tmp_options.GetFileSystem(), table_path); - PAIMON_ASSIGN_OR_RAISE(std::optional> latest_table_schema, - schema_manager.Latest()); - if (latest_table_schema == std::nullopt) { - return Status::Invalid("not found latest schema"); - } - // merge options - const auto& table_schema = latest_table_schema.value(); - auto final_options = table_schema->Options(); - for (const auto& [key, value] : options) { - final_options[key] = value; - } - PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, - CoreOptions::FromMap(final_options, file_system)); + std::pair, CoreOptions> schema_and_options; + PAIMON_ASSIGN_OR_RAISE(schema_and_options, + LoadSchemaAndOptions(table_path, options, file_system)); + const auto& [table_schema, core_options] = schema_and_options; PAIMON_ASSIGN_OR_RAISE(std::unique_ptr indexer, CreateGlobalIndexer(index_type, core_options)); PAIMON_ASSIGN_OR_RAISE(std::optional> extra_field_names, @@ -410,4 +420,22 @@ Result> GlobalIndexWriteTask::WriteIndex( extra_field_ids); } +Status GlobalIndexWriteTask::DeleteCheckpoint(const std::string& table_path, + const std::string& task_id, + const std::map& options, + const std::shared_ptr& file_system) { + PAIMON_ASSIGN_OR_RAISE(std::string task_directory, + IndexCheckpointPathFactory::GetTaskDirectory(table_path, task_id)); + std::pair, CoreOptions> schema_and_options; + PAIMON_ASSIGN_OR_RAISE(schema_and_options, + LoadSchemaAndOptions(table_path, options, file_system)); + std::shared_ptr fs = schema_and_options.second.GetFileSystem(); + PAIMON_ASSIGN_OR_RAISE(bool exists, fs->Exists(task_directory)); + if (!exists) { + return Status::OK(); + } + Status status = fs->Delete(task_directory, /*recursive=*/true); + return status.IsNotExist() ? Status::OK() : status; +} + } // namespace paimon diff --git a/src/paimon/core/index/index_checkpoint_path_factory.h b/src/paimon/core/index/index_checkpoint_path_factory.h index 0c1b13e51..0bfa56aa7 100644 --- a/src/paimon/core/index/index_checkpoint_path_factory.h +++ b/src/paimon/core/index/index_checkpoint_path_factory.h @@ -22,6 +22,10 @@ #include #include +#include "paimon/common/utils/path_util.h" +#include "paimon/result.h" +#include "paimon/status.h" + namespace paimon { /// Path factory for global index checkpoints scoped to one task. @@ -38,6 +42,13 @@ class IndexCheckpointPathFactory { /// Returns the checkpoint id when the file name belongs to this factory. virtual std::optional GetCheckpointId(const std::string& file_name) const = 0; + + /// Returns the task's checkpoint directory after validating its identifier. + static Result GetTaskDirectory(const std::string& table_path, + const std::string& task_id) { + PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint task id", task_id)); + return PathUtil::JoinPath(PathUtil::JoinPath(table_path, "index/checkpoint"), task_id); + } }; } // namespace paimon diff --git a/src/paimon/core/utils/file_store_path_factory.cpp b/src/paimon/core/utils/file_store_path_factory.cpp index ac15ad638..9acb6d4e3 100644 --- a/src/paimon/core/utils/file_store_path_factory.cpp +++ b/src/paimon/core/utils/file_store_path_factory.cpp @@ -248,10 +248,10 @@ FileStorePathFactory::CreateGlobalIndexCheckpointPathFactory(const std::string& }; PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint index type", index_type)); PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint field", field_name)); - PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint task id", task_id)); - std::string directory = PathUtil::JoinPath(IndexPath(root_), "checkpoint"); - std::string file_name_prefix = fmt::format("{}-global-index-{}-{}-{}-{}-", index_type, - field_name, range.from, range.to, task_id); + PAIMON_ASSIGN_OR_RAISE(std::string directory, + IndexCheckpointPathFactory::GetTaskDirectory(root_, task_id)); + std::string file_name_prefix = + fmt::format("{}-global-index-{}-{}-{}-", index_type, field_name, range.from, range.to); return std::unique_ptr( std::make_unique(directory, file_name_prefix)); } diff --git a/src/paimon/core/utils/file_store_path_factory_test.cpp b/src/paimon/core/utils/file_store_path_factory_test.cpp index bfb2b9621..447cb4f47 100644 --- a/src/paimon/core/utils/file_store_path_factory_test.cpp +++ b/src/paimon/core/utils/file_store_path_factory_test.cpp @@ -596,11 +596,14 @@ TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) { std::shared_ptr file_store_path_factory = CreateFactory(dir->Str()); std::shared_ptr file_system = dir->GetFileSystem(); - std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + std::string prefix = "lumina-global-index-vector-10-20-"; ASSERT_OK_AND_ASSIGN(std::unique_ptr checkpoint_path_factory, file_store_path_factory->CreateGlobalIndexCheckpointPathFactory( "lumina", "vector", Range(10, 20), "task-1")); - std::string checkpoint_dir = PathUtil::JoinPath(dir->Str(), "index/checkpoint"); + std::string checkpoint_dir = PathUtil::JoinPath(dir->Str(), "index/checkpoint/task-1"); + ASSERT_OK_AND_ASSIGN(std::string task_directory, + IndexCheckpointPathFactory::GetTaskDirectory(dir->Str(), "task-1")); + ASSERT_EQ(task_directory, checkpoint_dir); ASSERT_EQ(checkpoint_path_factory->GetDirectoryPath(), checkpoint_dir); std::string first_path = checkpoint_path_factory->NewPath(0); ASSERT_EQ(first_path, PathUtil::JoinPath(checkpoint_dir, prefix + "0.index.ckpt")); @@ -609,9 +612,9 @@ TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) { ASSERT_EQ(checkpoint_path_factory->ToPath("checkpoint"), PathUtil::JoinPath(checkpoint_dir, "checkpoint")); ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix + "9.index.ckpt"), 9); - ASSERT_EQ(checkpoint_path_factory->GetCheckpointId( - "lumina-global-index-other-10-20-task-1-9.index.ckpt"), - std::nullopt); + ASSERT_EQ( + checkpoint_path_factory->GetCheckpointId("lumina-global-index-other-10-20-9.index.ckpt"), + std::nullopt); ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix + "invalid-10.index.ckpt"), std::nullopt); ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix + "-1.index.ckpt"), std::nullopt); @@ -659,14 +662,18 @@ TEST_F(FileStorePathFactoryTest, TestCheckpointTaskIsolation) { ASSERT_OK_AND_ASSIGN(std::unique_ptr other_factory, factory->CreateGlobalIndexCheckpointPathFactory("lumina", "vector", Range(10, 20), "task-2")); - ASSERT_EQ(own_factory->GetDirectoryPath(), other_factory->GetDirectoryPath()); + ASSERT_NE(own_factory->GetDirectoryPath(), other_factory->GetDirectoryPath()); ASSERT_OK(fs->Mkdirs(own_factory->GetDirectoryPath())); + ASSERT_OK(fs->Mkdirs(other_factory->GetDirectoryPath())); std::string own_path = own_factory->NewPath(0); ASSERT_OK(fs->WriteFile(own_path, "checkpoint", /*overwrite=*/false)); + std::string same_name_other_path = other_factory->NewPath(0); + ASSERT_EQ(PathUtil::GetName(own_path), PathUtil::GetName(same_name_other_path)); + ASSERT_OK(fs->WriteFile(same_name_other_path, "foreign", /*overwrite=*/false)); std::string other_path = other_factory->NewPath(100); ASSERT_OK(fs->WriteFile(other_path, "foreign", /*overwrite=*/false)); ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(own_path)), 0); - ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(other_path)), std::nullopt); + ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(other_path)), 100); GlobalIndexFileManager manager(fs, factory->CreateGlobalIndexFileFactory(), std::move(own_factory)); @@ -690,6 +697,8 @@ TEST_F(FileStorePathFactoryTest, TestCheckpointTaskIsolation) { ASSERT_OK(manager.DeleteCheckpoint()); ASSERT_OK_AND_ASSIGN(exists, fs->Exists(other_path)); ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, fs->Exists(same_name_other_path)); + ASSERT_TRUE(exists); } TEST_F(FileStorePathFactoryTest, TestCheckpointRejectsInvalidPathComponents) { @@ -698,6 +707,7 @@ TEST_F(FileStorePathFactoryTest, TestCheckpointRejectsInvalidPathComponents) { auto factory = CreateFactory(dir->Str()); for (const std::string& invalid : std::vector{"", ".", "..", " ", "a/b", "a\\b", "a\nb", std::string("a\0b", 3)}) { + ASSERT_NOK(IndexCheckpointPathFactory::GetTaskDirectory(dir->Str(), invalid)); ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory(invalid, "vector", Range(10, 20), "task")); ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory("lumina", invalid, Range(10, 20), diff --git a/src/paimon/indexer/lumina/lumina_global_index_test.cpp b/src/paimon/indexer/lumina/lumina_global_index_test.cpp index e3451c44b..0d5605f49 100644 --- a/src/paimon/indexer/lumina/lumina_global_index_test.cpp +++ b/src/paimon/indexer/lumina/lumina_global_index_test.cpp @@ -429,7 +429,7 @@ TEST_F(LuminaGlobalIndexTest, TestCheckpointFileManagement) { auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); ASSERT_TRUE(test_root_dir); std::string test_root = test_root_dir->Str(); - std::string prefix = "lumina-global-index-f0-10-20-task-1-"; + std::string prefix = "lumina-global-index-f0-10-20-"; ASSERT_OK_AND_ASSIGN(std::unique_ptr checkpoint_path_factory, CreateCheckpointPathFactory(test_root, "lumina", "f0", Range(10, 20))); ASSERT_OK_AND_ASSIGN(std::shared_ptr path_factory, @@ -457,13 +457,13 @@ TEST_F(LuminaGlobalIndexTest, TestCheckpointFileManagement) { ASSERT_EQ(second_written, 6); ASSERT_OK(second_output->Close()); - std::string checkpoint_dir = PathUtil::JoinPath(test_root, "index/checkpoint"); + std::string checkpoint_dir = PathUtil::JoinPath(test_root, "index/checkpoint/task-1"); ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(checkpoint_dir, prefix + "9.index.ckpt"), "id-nine", /*overwrite=*/false)); ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(checkpoint_dir, prefix + "10.index.ckpt"), "id-ten", /*overwrite=*/false)); std::string unrelated_file = - PathUtil::JoinPath(checkpoint_dir, "lumina-global-index-other-10-20-task-2-99.index.ckpt"); + PathUtil::JoinPath(checkpoint_dir, "lumina-global-index-other-10-20-99.index.ckpt"); ASSERT_OK(fs_->WriteFile(unrelated_file, "unrelated", /*overwrite=*/false)); std::string malformed_file = PathUtil::JoinPath(checkpoint_dir, prefix + "invalid-99.index.ckpt"); @@ -709,7 +709,8 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithTagFilter) { GlobalIndexIOMeta meta, WriteGlobalIndex(test_root, tag_data_type, tag_options, tag_array, Range(0, 3))); std::vector checkpoint_files; - ASSERT_OK(fs_->ListDir(PathUtil::JoinPath(test_root, "index/checkpoint"), &checkpoint_files)); + ASSERT_OK( + fs_->ListDir(PathUtil::JoinPath(test_root, "index/checkpoint/task-1"), &checkpoint_files)); ASSERT_TRUE(checkpoint_files.empty()); ASSERT_OK_AND_ASSIGN(std::shared_ptr reader, CreateGlobalIndexReader(test_root, data_type_, tag_options, meta)); diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 97b169823..c5b421629 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -576,10 +576,12 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithCheckpoint) { std::shared_ptr checkpoint_fs = dir_->GetFileSystem(); std::string task_id = "task-1"; std::string checkpoint_dir = PathUtil::JoinPath(table_path, "index/checkpoint"); - ASSERT_OK(checkpoint_fs->Mkdirs(checkpoint_dir)); + std::string task_checkpoint_dir = PathUtil::JoinPath(checkpoint_dir, task_id); + ASSERT_OK(checkpoint_fs->Mkdirs(task_checkpoint_dir)); + ASSERT_OK(checkpoint_fs->Mkdirs(PathUtil::JoinPath(checkpoint_dir, "another-task"))); auto checkpoint_path = [&](const std::string& task, int64_t id) { - return PathUtil::JoinPath( - checkpoint_dir, fmt::format("lumina-global-index-f1-0-3-{}-{}.index.ckpt", task, id)); + return PathUtil::JoinPath(PathUtil::JoinPath(checkpoint_dir, task), + fmt::format("lumina-global-index-f1-0-3-{}.index.ckpt", id)); }; std::string matching_checkpoint = checkpoint_path(task_id, 9); ASSERT_OK(checkpoint_fs->WriteFile(matching_checkpoint, "invalid checkpoint", @@ -593,10 +595,17 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithCheckpoint) { std::make_shared(split, std::vector{Range(0, 3)}), lumina_options, task_id, pool_)); std::vector checkpoint_files; - ASSERT_OK(checkpoint_fs->ListDir(checkpoint_dir, &checkpoint_files)); - ASSERT_EQ(checkpoint_files.size(), 1); - ASSERT_EQ(PathUtil::GetName(checkpoint_files[0].GetPath()), - PathUtil::GetName(unrelated_checkpoint)); + ASSERT_OK(checkpoint_fs->ListDir(task_checkpoint_dir, &checkpoint_files)); + ASSERT_TRUE(checkpoint_files.empty()); + ASSERT_OK_AND_ASSIGN(bool unrelated_exists, checkpoint_fs->Exists(unrelated_checkpoint)); + ASSERT_TRUE(unrelated_exists); + + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(table_path, "another-task", lumina_options)); + ASSERT_OK_AND_ASSIGN(unrelated_exists, checkpoint_fs->Exists(unrelated_checkpoint)); + ASSERT_FALSE(unrelated_exists); + ASSERT_OK_AND_ASSIGN(bool unrelated_directory_exists, + checkpoint_fs->Exists(PathUtil::JoinPath(checkpoint_dir, "another-task"))); + ASSERT_FALSE(unrelated_directory_exists); } TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithMismatchedDimension) { From ed13f2644f64ace1d1c422c060761834f9a79918 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 27 Oct 2026 14:46:28 +0800 Subject: [PATCH 6/8] fix --- docs/source/user_guide/catalog.rst | 10 ++++++++++ docs/source/user_guide/manifest_entry_cache.rst | 8 ++++++++ .../shredding/map_shared_shredding_schema_utils.h | 12 +++++++----- src/paimon/common/data/blob_defs.h | 2 +- 4 files changed, 26 insertions(+), 6 deletions(-) diff --git a/docs/source/user_guide/catalog.rst b/docs/source/user_guide/catalog.rst index a3f4bcbee..de97e8a6a 100644 --- a/docs/source/user_guide/catalog.rst +++ b/docs/source/user_guide/catalog.rst @@ -386,12 +386,22 @@ file system of your own when the credentials expire. class MyCredentialProvider : public paimon::CredentialProvider { public: + // The factory hands over the path and the file system options of the accesses to + // authenticate, which is where you read your own configuration from. + MyCredentialProvider(std::string path, std::map options) + : path_(std::move(path)), options_(std::move(options)) {} + // Called as each access is signed; reload the credentials before they expire. paimon::Result> GetCredentials() const override { + // Request fresh credentials for path_ from the token service in options_. return std::map{ {"fs.oss.accessKeyId", ...}, {"fs.oss.accessKeySecret", ...}, {"fs.oss.securityToken", ...}}; } + + private: + std::string path_; + std::map options_; }; class MyCredentialProviderFactory : public paimon::CredentialProviderFactory { diff --git a/docs/source/user_guide/manifest_entry_cache.rst b/docs/source/user_guide/manifest_entry_cache.rst index 995d77e38..4d9fed056 100644 --- a/docs/source/user_guide/manifest_entry_cache.rst +++ b/docs/source/user_guide/manifest_entry_cache.rst @@ -86,6 +86,14 @@ under .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "3") .Finish()); +.. note:: + + ``LruCache`` is the built-in weight-bounded implementation, but it is declared in + ``src/paimon/common/io/cache/lru_cache.h`` and only ``include/`` is installed, so it is + not reachable from an installed Paimon. ``WithCache`` takes the public ``Cache`` + interface of ``include/paimon/cache/cache.h``; pass your own implementation of it when + you build against the installed package. + Cache entries are scoped by table path, branch, and bucket, so they can be reused across newly created ``TableScan`` and ``FileStoreScan`` instances as long as they share the same cache object and scan the same bucket. diff --git a/include/paimon/data/shredding/map_shared_shredding_schema_utils.h b/include/paimon/data/shredding/map_shared_shredding_schema_utils.h index bce0b6534..50eb123e1 100644 --- a/include/paimon/data/shredding/map_shared_shredding_schema_utils.h +++ b/include/paimon/data/shredding/map_shared_shredding_schema_utils.h @@ -64,12 +64,14 @@ struct PAIMON_EXPORT MapSharedShreddingFieldMeta { /// /// Example: read keys "age" and "score" from MAP column `attributes`: /// -/// auto builder = MapSharedShreddingAccessBuilder::Create(attributes_field); -/// builder->AddKey("age"); -/// builder->AddKey("score"); -/// auto field = builder->Build(); +/// PAIMON_ASSIGN_OR_RAISE(std::unique_ptr builder, +/// MapSharedShreddingAccessBuilder::Create(attributes_field)); +/// PAIMON_RETURN_NOT_OK(builder->AddKey("age")); +/// PAIMON_RETURN_NOT_OK(builder->AddKey("score")); +/// PAIMON_ASSIGN_OR_RAISE(std::unique_ptr field, builder->Build()); /// -/// Use the returned field in `ReadContextBuilder::SetReadSchema`. +/// Use the resulting field in `ReadContextBuilder::SetReadSchema`. Both macros require the +/// enclosing function to return `Status` or `Result`. class PAIMON_EXPORT MapSharedShreddingAccessBuilder { public: /// Creates a builder bound to the original MAP field. diff --git a/src/paimon/common/data/blob_defs.h b/src/paimon/common/data/blob_defs.h index ce5932b43..8dbb681b1 100644 --- a/src/paimon/common/data/blob_defs.h +++ b/src/paimon/common/data/blob_defs.h @@ -40,7 +40,7 @@ class BlobDefs { /// To create a Blob field: /// @code /// std::unordered_map blob_metadata_map = { - /// {Blob::kExtensionTypeKey, Blob::kExtensionTypeValue} + /// {BlobDefs::kExtensionTypeKey, BlobDefs::kExtensionTypeValue} /// }; /// auto field = arrow::field("my_blob_field", arrow::large_binary(), false, /// std::make_shared(blob_metadata_map)); From 88d1bc11c8e25b361f89fa5307a2274e52badbfc Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 27 Oct 2026 14:58:34 +0800 Subject: [PATCH 7/8] fix --- include/paimon/data/variant.h | 10 ++++++---- include/paimon/read_context.h | 5 ++++- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/include/paimon/data/variant.h b/include/paimon/data/variant.h index 1958d09de..d93195856 100644 --- a/include/paimon/data/variant.h +++ b/include/paimon/data/variant.h @@ -167,11 +167,13 @@ class PAIMON_EXPORT Variant { /// Example: read `$.age` as INT64 and `$.city` as STRING from variant column `v`: /// /// VariantAccessBuilder builder; -/// builder.AddField(age_type, "$.age"); -/// builder.AddField(city_type, "$.city"); -/// auto field = builder.Build("v"); // use in ReadContextBuilder::SetReadSchema +/// PAIMON_RETURN_NOT_OK(builder.AddField(age_type, "$.age")); +/// PAIMON_RETURN_NOT_OK(builder.AddField(city_type, "$.city")); +/// PAIMON_ASSIGN_OR_RAISE(std::unique_ptr field, builder.Build("v")); /// -/// The resulting struct has one child per added field, named by its position ("0", "1", ...). +/// Use the resulting field in `ReadContextBuilder::SetReadSchema`. Both macros require the +/// enclosing function to return `Status` or `Result`. The resulting struct has one child +/// per added field, named by its position ("0", "1", ...). class PAIMON_EXPORT VariantAccessBuilder { public: VariantAccessBuilder(); diff --git a/include/paimon/read_context.h b/include/paimon/read_context.h index e62386c65..81b0f8fa2 100644 --- a/include/paimon/read_context.h +++ b/include/paimon/read_context.h @@ -311,7 +311,10 @@ class PAIMON_EXPORT ReadContextBuilder { /// }); /// /// auto c_schema = std::make_unique(); - /// arrow::ExportSchema(*projected_schema, c_schema.get()); + /// arrow::Status export_status = arrow::ExportSchema(*projected_schema, c_schema.get()); + /// if (!export_status.ok()) { + /// return paimon::Status::Invalid(export_status.message()); + /// } /// /// ReadContextBuilder builder("/path/to/table"); /// builder.SetReadSchema(std::move(c_schema)); From 10cf0a351040a621dc4e8f16f2667cf8897c5b17 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 27 Oct 2026 15:50:53 +0800 Subject: [PATCH 8/8] fix --- docs/source/basic_concepts.rst | 3 +-- docs/source/user_guide/system_tables.rst | 22 +++++++++++++++++----- docs/source/user_guide/write.rst | 14 +++++++++----- 3 files changed, 27 insertions(+), 12 deletions(-) diff --git a/docs/source/basic_concepts.rst b/docs/source/basic_concepts.rst index 9be8fec40..0af523056 100644 --- a/docs/source/basic_concepts.rst +++ b/docs/source/basic_concepts.rst @@ -55,8 +55,7 @@ the corresponding snapshot. Data Files --------------------------- Data files are grouped by partitions. Paimon C++ supports Parquet (default), -ORC, and Avro as data file formats, with optional Lance, Mosaic, and Vortex -support. +ORC, and Avro as data file formats, with optional Lance and Mosaic support. Partition --------------------------- diff --git a/docs/source/user_guide/system_tables.rst b/docs/source/user_guide/system_tables.rst index 90c88c133..843640eef 100644 --- a/docs/source/user_guide/system_tables.rst +++ b/docs/source/user_guide/system_tables.rst @@ -49,7 +49,8 @@ Global System Tables -------------------- Four system tables are not bound to a single data table. They live under the -reserved ``sys`` database and enumerate across all databases in the warehouse: +reserved ``sys`` database, and three of them enumerate across all databases in +the warehouse: * ``sys.catalog_options``: catalog-level configuration as key/value rows. * ``sys.all_table_options``: all table options as @@ -58,15 +59,26 @@ reserved ``sys`` database and enumerate across all databases in the warehouse: statistics. * ``sys.partitions``: partition-level file statistics for all tables. -These need a catalog to enumerate databases and tables, so load them through a -catalog as the identifier ``sys.`` rather than from a bare -``/sys/`` path; a path load without catalog context fails with -``global system table requires catalog context: ``. ``sys.all_table_options``, ``sys.tables`` and ``sys.partitions`` are always available. ``sys.catalog_options`` is disabled by default and has to be turned on with the catalog option ``catalog-options-table.enabled``; a disabled one is reported as not existing. +Those three enumerate databases and tables, so they need a catalog: loading a +bare ``/sys/`` path fails with ``global system table requires +catalog context: ``. ``sys.catalog_options`` enumerates nothing, and is +the one a bare path serves. + +.. note:: + + No public read path serves those three yet. A catalog lists them and + returns their schema, but does not hand back a readable table: + ``Catalog::GetTable(Identifier("sys", "tables"))`` is rejected by a + file-system catalog with ``do not support checking TableSchemaExists for + system table``, and a REST catalog returns one whose path is left unset, + which ``TableScan`` and ``TableRead`` cannot dispatch to the global system + table loader. + Read-Optimized System Table --------------------------- diff --git a/docs/source/user_guide/write.rst b/docs/source/user_guide/write.rst index 6c7bff16b..127889ad1 100644 --- a/docs/source/user_guide/write.rst +++ b/docs/source/user_guide/write.rst @@ -231,11 +231,15 @@ produce a correct ``Snapshot``, which commonly includes (but is not limited to): The C++ writer supports Append and PK tables and can produce ``CommitMessage`` objects for both. ``FileStoreCommit`` supports direct file-system commits for both table types on non-object-store paths. - Object-store paths require a catalog-mediated commit, configured with either - ``CommitContextBuilder::WithCatalog()`` or ``UseRESTCatalogCommit()``; - neither needs an atomic rename on the object store. Changelog is out of - scope and should not be emitted in ``CommitMessage`` until explicitly - supported. + Object-store paths need a commit that does not rename: either a catalog that + manages the versions of its tables, such as the REST catalog, passed to + ``CommitContextBuilder::WithCatalog()``, or ``UseRESTCatalogCommit(true)``, + which builds the commit request for the caller to send instead of publishing + the snapshot itself. The two are mutually exclusive. A catalog without + version management still commits by renaming, so passing one is rejected on + an object-store path. Changelog files are produced on the write path and + committed with their snapshot, except by the overwrite flow, which ignores + them and logs a warning naming the files it ignored. Serialization and Deserialization ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~