From aa4f24376fc2007be8ae89f20418a148d157913d Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Sat, 10 Oct 2026 16:07:10 +0800 Subject: [PATCH 1/8] feat(blob): read blob-view dependency tables through a catalog --- include/paimon/defs.h | 8 +- src/paimon/common/catalog/catalog_context.h | 21 ++- .../operation/data_evolution_split_read.cpp | 14 +- .../core/operation/internal_read_context.h | 5 + .../operation/internal_read_context_test.cpp | 40 +++++ src/paimon/core/operation/read_context.cpp | 63 ++++++++ .../core/operation/read_context_test.cpp | 86 +++++++++++ src/paimon/core/utils/blob_view_lookup.cpp | 40 +---- src/paimon/core/utils/blob_view_lookup.h | 4 +- test/inte/blob_table_inte_test.cpp | 140 +++--------------- 10 files changed, 251 insertions(+), 170 deletions(-) diff --git a/include/paimon/defs.h b/include/paimon/defs.h index d6abd047d..b9dcb64ec 100644 --- a/include/paimon/defs.h +++ b/include/paimon/defs.h @@ -569,9 +569,11 @@ struct PAIMON_EXPORT Options { /// tables at read time. Set to false to preserve serialized BlobViewStruct bytes when /// forwarding blob view values to another blob-view table. Default value is "true". static const char BLOB_VIEW_RESOLVE_ENABLED[]; - /// "blob-view-upstream-warehouse" - Since the catalog capabilities are partially missing, when - /// Blob View is enabled, cpp paimon cannot automatically obtain the upstream table warehouse - /// path and requires manual configuration by the user. No default value. + /// "blob-view-upstream-warehouse" - Warehouse root the upstream table a blob-view field points + /// at is resolved from. Since the catalog capabilities are partially missing, cpp paimon cannot + /// always obtain it automatically, so it may be configured explicitly. When it is not set, the + /// read falls back to the warehouse the downstream table itself lives in, which resolves a + /// co-located upstream table. No default value. static const char BLOB_VIEW_UPSTREAM_WAREHOUSE[]; /// "blob-write-null-on-missing-file" - Whether to write NULL for a descriptor BLOB value, /// including an ARRAY element or a MAP<..., BLOB> value, when the referenced file does diff --git a/src/paimon/common/catalog/catalog_context.h b/src/paimon/common/catalog/catalog_context.h index fecc3e012..7a9f8b873 100644 --- a/src/paimon/common/catalog/catalog_context.h +++ b/src/paimon/common/catalog/catalog_context.h @@ -21,18 +21,29 @@ #include #include -#include "paimon/fs/file_system.h" - namespace paimon { + +/// What a dependency table - the blob table behind a BlobView - is read through. Like the Java +/// `CatalogContext` that `CatalogEnvironment.dependencyReadContext()` returns, it carries no file +/// system. The catalog built from these options issues credentials for the table about to be read, +/// whereas a file system handed down from the read above authenticates as a different table - and +/// the `X-Paimon-Read-Via` header travelling with it cannot excuse that, being request context a +/// server must not treat as authorization proof. A caller supplying its own file system therefore +/// gets no blob view delegation. struct CatalogContext { CatalogContext(const std::string& _root_path, const std::map& _options, - const std::shared_ptr& _file_system) - : root_path(_root_path), options(_options), file_system(_file_system) {} + const std::map& _fs_scheme_to_identifier_map) + : root_path(_root_path), + options(_options), + fs_scheme_to_identifier_map(_fs_scheme_to_identifier_map) {} std::string root_path; std::map options; - std::shared_ptr file_system; + /// Maps a URI scheme to the registered file system serving it, the map the file system of the + /// read this descends from was resolved with. The catalog builds its own file system out of + /// these options and needs the map to route a scheme that is not their default. + std::map fs_scheme_to_identifier_map; }; } // namespace paimon diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 7a1a7f181..f6ede5fa9 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -271,17 +271,25 @@ Result> DataEvolutionSplitRead::WrapWithBlobViewRes if (read_blob_view_fields.empty()) { return std::move(inner_reader); } + // The explicit upstream warehouse is optional. Absent it, fall back to the warehouse this + // (downstream) table itself lives in, which is where a co-located upstream table is resolved + // from; a metastore-backed catalog resolves the upstream location by identifier anyway, so the + // root is only a hint there. Stripping the trailing `.db/` (or + // `/
`) components off this table's path yields that warehouse. std::optional warehouse_path = options_.GetBlobViewUpstreamWarehouse(); if (!warehouse_path) { - return Status::Invalid( - "invalid config for blob view, supposed to set BLOB_VIEW_UPSTREAM_WAREHOUSE"); + warehouse_path = + PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(context_->GetPath())); } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr pre_reader, CreateBlobViewReader(data_split, read_blob_view_fields, row_ranges)); PAIMON_ASSIGN_OR_RAISE(std::unordered_set blob_view_structs, ExtractBlobViewStructs(pre_reader.get())); + // The dependency table is read through a catalog of its own, told how to resolve a scheme but + // not handed this read's file system: that one authenticates as this table, and the table about + // to be read is a different one whose credentials the metastore issues only for it. auto catalog_context = std::make_shared( - warehouse_path.value(), options_.ToMap(), options_.GetFileSystem()); + warehouse_path.value(), options_.ToMap(), context_->GetFileSystemSchemeToIdentifierMap()); // use global thread number uint32_t cpu_count = std::thread::hardware_concurrency(); uint32_t thread_num = cpu_count > 0 ? cpu_count : 1; diff --git a/src/paimon/core/operation/internal_read_context.h b/src/paimon/core/operation/internal_read_context.h index e3c2326fc..8e34a85bd 100644 --- a/src/paimon/core/operation/internal_read_context.h +++ b/src/paimon/core/operation/internal_read_context.h @@ -111,6 +111,11 @@ class InternalReadContext { std::shared_ptr GetRealtimeContext() const { return read_context_->GetRealtimeContext(); } + /// The scheme map the file system of this read was resolved with, which a catalog reading a + /// dependency table on its behalf needs to resolve that table's location the same way. + const std::map& GetFileSystemSchemeToIdentifierMap() const { + return read_context_->GetFileSystemSchemeToIdentifierMap(); + } bool ReadAheadCacheEnabled() const { return read_context_->ReadAheadCacheEnabled(); diff --git a/src/paimon/core/operation/internal_read_context_test.cpp b/src/paimon/core/operation/internal_read_context_test.cpp index 8878a27a1..fceeb2965 100644 --- a/src/paimon/core/operation/internal_read_context_test.cpp +++ b/src/paimon/core/operation/internal_read_context_test.cpp @@ -75,6 +75,46 @@ TEST(InternalReadContext, TestReadWithSpecifiedSchema) { ASSERT_TRUE(internal_context->GetReadSchema()->Equals(expected_schema)); } +TEST(InternalReadContext, TestHeaderOptionSurvivesIntoCoreOptions) { + // The `header.` option a read through a REST catalog reports its starting table with has to + // still be among the options a dependency table is read through: `DataEvolutionSplitRead` + // builds that catalog context out of `CoreOptions::ToMap()`, so an option dropped on the way + // here would quietly drop the `X-Paimon-Read-Via` delegation header along with it. + std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09"; + ReadContextBuilder context_builder(path); + ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish()); + SchemaManager schema_manager(std::make_shared(), read_context->GetPath()); + ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0)); + + // The merge `TableRead::Create` performs: the table's own options first, then the ones the read + // context carries over them, which is where a header option arrives from. + const std::string read_via = "%7B%22database%22%3A%22db1%22%2C%22object%22%3A%22t1%22%7D"; + std::map options = table_schema->Options(); + options["header.X-Paimon-Read-Via"] = read_via; + ASSERT_OK_AND_ASSIGN(auto internal_context, InternalReadContext::Create(std::move(read_context), + table_schema, options)); + ASSERT_EQ(internal_context->GetCoreOptions().ToMap().at("header.X-Paimon-Read-Via"), read_via); +} + +TEST(InternalReadContext, TestFileSystemSchemeMapReachesTheDependencyRead) { + // A blob view reads its dependency table through a catalog of its own, which builds a file + // system out of the options and this map rather than borrowing the one of the read it descends + // from. Nothing reading a local path would notice the map going missing: a location is + // resolved through it only when its scheme is not the default of the options, so dropping it + // surfaces as a `dfs://` dependency table that cannot be opened, far away from the cause. + std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09"; + const std::map scheme_map = {{"file", "local"}}; + ReadContextBuilder context_builder(path); + context_builder.WithFileSystemSchemeToIdentifierMap(scheme_map); + ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish()); + SchemaManager schema_manager(std::make_shared(), read_context->GetPath()); + ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0)); + ASSERT_OK_AND_ASSIGN(auto internal_context, + InternalReadContext::Create(std::move(read_context), table_schema, + table_schema->Options())); + ASSERT_EQ(internal_context->GetFileSystemSchemeToIdentifierMap(), scheme_map); +} + TEST(InternalReadContext, TestReadWithSpecifiedFieldId) { std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09"; ReadContextBuilder context_builder(path); diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index 5cbbb5dba..336b8c036 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -18,12 +18,17 @@ #include "paimon/read_context.h" +#include +#include #include #include "arrow/c/abi.h" #include "arrow/c/bridge.h" #include "paimon/catalog/catalog.h" +#include "paimon/catalog_options.h" #include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/common/utils/url_utils.h" #include "paimon/core/utils/branch_manager.h" #include "paimon/executor.h" #include "paimon/memory/memory_pool.h" @@ -31,10 +36,55 @@ #include "paimon/predicate/vector_search.h" #include "paimon/status.h" #include "paimon/table/format/format_table.h" +#include "rapidjson/document.h" +#include "rapidjson/stringbuffer.h" +#include "rapidjson/writer.h" namespace paimon { class Predicate; +namespace { + +/// The catalog option that carries the `X-Paimon-Read-Via` request header, spelled with the prefix +/// a REST catalog strips when it turns an option into a header. That prefix is +/// `RestApi::kHeaderOptionPrefix`, not referenced here because its header pulls in the http client; +/// mirrors the Java `RESTApi.READ_VIA_OPTION`. +constexpr char kReadViaOption[] = "header.X-Paimon-Read-Via"; + +/// True when the options select a catalog that speaks the REST protocol, the only kind turning a +/// `header.`-prefixed option into a request header. These same options decide which catalog a +/// dependency table is read through, so when they select none there is nobody to send the header +/// to. The Java counterpart answers this from the catalog loader it was built with. +bool IsRestMetastore(const std::map& options) { + auto metastore = options.find(CatalogOptions::METASTORE); + if (metastore == options.end()) { + return false; + } + // Matched leniently in lower case, the way `Catalog::Create` matches it. + const std::string name = StringUtils::ToLowerCase(metastore->second); + return name == "rest"; +} + +/// `{"database":"db1","object":"t1"}`, the flat JSON the metastore parses a header value back into. +/// The field names are the Java `Identifier.FIELD_DATABASE_NAME` and `FIELD_OBJECT_NAME`, and the +/// object name keeps any branch and system-table suffix, as the `getObjectName()` the Java +/// serializer reads does, so a read of a branch reports that branch. +std::string IdentifierToFlatJson(const Identifier& identifier) { + rapidjson::Document document(rapidjson::kObjectType); + auto& allocator = document.GetAllocator(); + document.AddMember(rapidjson::StringRef("database"), + rapidjson::Value(identifier.GetDatabaseName(), allocator), allocator); + document.AddMember(rapidjson::StringRef("object"), + rapidjson::Value(identifier.GetTableName(), allocator), allocator); + // Compact, not the pretty form `RapidJsonUtil::ToJson` writes: the value goes into a header. + rapidjson::StringBuffer buffer; + rapidjson::Writer writer(buffer); + document.Accept(writer); + return buffer.GetString(); +} + +} // namespace + ReadContext::ReadContext( const std::string& path, const std::string& branch, const std::vector& read_field_names, const std::vector& read_field_ids, @@ -404,6 +454,19 @@ Result> ReadContextBuilder::Finish() { if (impl_->enable_multi_thread_row_to_batch_ && impl_->row_to_batch_thread_number_ <= 0) { return Status::Invalid("row to batch thread number should be greater than 0"); } + // A read that goes through a catalog says which table it started from, which is what lets the + // metastore authorize a dependency table - the blob table behind a BlobView - through the owner + // of that table. Mirrors the Java `CatalogEnvironment.dependencyReadContext()` and both of its + // guards: only a REST metastore understands the header, and a value already there wins, so a + // chain of views reports its outermost one instead of the innermost. It travels in the options, + // which `TableRead::Create` merges into the table options and the blob view read then hands to + // the catalog it reads the dependency table through, so it reaches that request with no second + // channel to keep in step. + if (impl_->identifier_ && IsRestMetastore(impl_->options_) && + impl_->options_.count(kReadViaOption) == 0) { + impl_->options_[kReadViaOption] = + UrlUtils::EncodeString(IdentifierToFlatJson(impl_->identifier_.value())); + } auto ctx = std::make_unique( impl_->path_, branch, impl_->read_field_names_, impl_->read_field_ids_, impl_->predicate_, impl_->vector_search_, impl_->full_text_search_, impl_->enable_predicate_filter_, diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 7ea998a0d..14f70bc68 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -23,6 +23,7 @@ #include "arrow/c/bridge.h" #include "arrow/type.h" #include "gtest/gtest.h" +#include "paimon/catalog_options.h" #include "paimon/common/io/cache/lru_cache.h" #include "paimon/core/utils/branch_manager.h" #include "paimon/defs.h" @@ -77,6 +78,91 @@ TEST(ReadContextTest, TestWithCatalogNullRejected) { "cannot read through a null catalog"); } +TEST(ReadContextTest, TestReadViaHeaderReportsTheStartingTable) { + // A read through a REST catalog carries the table it started from, so that the metastore can + // authorize a dependency table - the blob table behind a BlobView - through that table's owner. + // The value is the identifier's flat JSON in the form-urlencoded flavor the REST server + // expects, byte for byte what the Java client sends. + const std::string expected = "%7B%22database%22%3A%22db1%22%2C%22object%22%3A%22t1%22%7D"; + for (const std::string& metastore : {"rest"}) { + auto catalog = std::make_shared(); + catalog->SetTableFileSystem(std::make_shared()); + + ReadContextBuilder builder("table_root_path"); + builder.AddOption(CatalogOptions::METASTORE, metastore); + ASSERT_OK_AND_ASSIGN(auto ctx, + builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); + + const std::map& options = ctx->GetOptions(); + ASSERT_EQ(options.count("header.X-Paimon-Read-Via"), 1u) << metastore; + ASSERT_EQ(options.at("header.X-Paimon-Read-Via"), expected) << metastore; + // The metastore is the only other option, so the header is added rather than anything + // rewritten around it. + ASSERT_EQ(options.size(), 2u) << metastore; + } +} + +TEST(ReadContextTest, TestReadViaHeaderKeepsTheOutermostView) { + // A view read on behalf of another view arrives with the header already set, and the outermost + // view is the one the metastore hears about: the option in hand wins, as it does in the Java + // `CatalogEnvironment.dependencyReadContext()`. + auto catalog = std::make_shared(); + catalog->SetTableFileSystem(std::make_shared()); + + ReadContextBuilder builder("table_root_path"); + builder.AddOption(CatalogOptions::METASTORE, "rest"); + builder.AddOption("header.X-Paimon-Read-Via", "outer-view"); + ASSERT_OK_AND_ASSIGN(auto ctx, builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); + ASSERT_EQ(ctx->GetOptions().at("header.X-Paimon-Read-Via"), "outer-view"); +} + +TEST(ReadContextTest, TestReadViaHeaderNeedsARestCatalogAndATable) { + // Only a REST metastore turns a `header.` option into a request header, and only a read naming + // a table has one to report: the two guards the Java counterpart checks before copying the + // options over. A catalog that is not one is told nothing. + for (const std::string& metastore : {"filesystem", "hive", "alake"}) { + auto catalog = std::make_shared(); + catalog->SetTableFileSystem(std::make_shared()); + + ReadContextBuilder builder("table_root_path"); + builder.AddOption(CatalogOptions::METASTORE, metastore); + ASSERT_OK_AND_ASSIGN(auto ctx, + builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); + ASSERT_EQ(ctx->GetOptions().count("header.X-Paimon-Read-Via"), 0u) << metastore; + } + + // Nor is one added when the options leave the metastore out, which reads as a filesystem + // catalog, or when the read goes through no catalog and so names no table - however the + // metastore is spelled, since `Catalog::Create` matches it in lower case. + auto catalog = std::make_shared(); + catalog->SetTableFileSystem(std::make_shared()); + ReadContextBuilder no_metastore_builder("table_root_path"); + ASSERT_OK_AND_ASSIGN( + auto no_metastore_ctx, + no_metastore_builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); + ASSERT_EQ(no_metastore_ctx->GetOptions().count("header.X-Paimon-Read-Via"), 0u); + + ReadContextBuilder no_catalog_builder("table_root_path"); + no_catalog_builder.AddOption(CatalogOptions::METASTORE, "REST"); + ASSERT_OK_AND_ASSIGN(auto no_catalog_ctx, no_catalog_builder.Finish()); + ASSERT_EQ(no_catalog_ctx->GetOptions().count("header.X-Paimon-Read-Via"), 0u); +} + +TEST(ReadContextTest, TestReadViaHeaderReportsTheBranchOfTheStartingTable) { + // What is reported is the object name, the branch suffix included, because that is what the + // Java `getObjectName()` its serializer reads returns: a read of a branch says so instead of + // naming the data table. The metastore is matched in upper case here as well. + auto catalog = std::make_shared(); + catalog->SetTableFileSystem(std::make_shared()); + + ReadContextBuilder builder("table_root_path"); + builder.AddOption(CatalogOptions::METASTORE, "REST"); + ASSERT_OK_AND_ASSIGN(auto ctx, + builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")).Finish()); + ASSERT_EQ(ctx->GetOptions().at("header.X-Paimon-Read-Via"), + "%7B%22database%22%3A%22db1%22%2C%22object%22%3A%22t1%24branch_dev%22%7D"); +} + TEST(ReadContextTest, TestDefaultValue) { ReadContextBuilder builder("table_root_path"); ASSERT_OK_AND_ASSIGN(auto ctx, builder.Finish()); diff --git a/src/paimon/core/utils/blob_view_lookup.cpp b/src/paimon/core/utils/blob_view_lookup.cpp index 99e7dc835..30999fe62 100644 --- a/src/paimon/core/utils/blob_view_lookup.cpp +++ b/src/paimon/core/utils/blob_view_lookup.cpp @@ -32,7 +32,6 @@ #include "paimon/common/table/special_fields.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/checked_cast.h" -#include "paimon/common/utils/path_util.h" #include "paimon/data/blob_descriptor.h" #include "paimon/defs.h" #include "paimon/executor.h" @@ -150,13 +149,16 @@ Result BlobViewLookup::LoadTableDescriptorChu if (branch) { return Status::Invalid("do not support upstream table with branch"); } - auto file_system = catalog_context->file_system; - PAIMON_ASSIGN_OR_RAISE(std::string table_path, GetTableLocation(catalog_context, identifier)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr catalog, + Catalog::Create(catalog_context->root_path, catalog_context->options, + /*file_system=*/nullptr, catalog_context->fs_scheme_to_identifier_map)); + PAIMON_ASSIGN_OR_RAISE(std::string table_path, catalog->GetTableLocation(identifier)); ScanContextBuilder scan_builder(table_path); auto global_index_result = BitmapGlobalIndexResult::FromRanges(row_ranges); scan_builder.SetGlobalIndexResult(global_index_result) .WithMemoryPool(pool) - .WithFileSystem(file_system); + .WithCatalog(catalog, identifier); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan_context, scan_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_scan, TableScan::Create(std::move(scan_context))); @@ -169,7 +171,7 @@ Result BlobViewLookup::LoadTableDescriptorChu .AddOption(Options::BLOB_AS_DESCRIPTOR, "true") .EnablePrefetch(true) .WithMemoryPool(pool) - .WithFileSystem(file_system); + .WithCatalog(catalog, identifier); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr read_context, read_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_read, TableRead::Create(std::move(read_context))); @@ -182,34 +184,6 @@ Result BlobViewLookup::LoadTableDescriptorChu return mapping; } -Result BlobViewLookup::GetTableLocation( - const std::shared_ptr& catalog_context, const Identifier& identifier) { - auto file_system = catalog_context->file_system; - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr catalog, - Catalog::Create(catalog_context->root_path, catalog_context->options, file_system)); - // The table path may be either xxx/test_database/test_table or - // xxx/test_database.db/test_table. If neither path exists or both paths exist, it means we - // cannot infer the table path, and an error will be reported. If only one of the paths - // exists, we will use that path. - PAIMON_ASSIGN_OR_RAISE(std::string source_table_path, catalog->GetTableLocation(identifier)); - std::string database_name = identifier.GetDatabaseName(); - PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); - std::string fallback_source_table_path = PathUtil::JoinPath( - PathUtil::JoinPath(catalog_context->root_path, database_name), data_table_name); - - PAIMON_ASSIGN_OR_RAISE(bool exist, catalog_context->file_system->Exists(source_table_path)); - PAIMON_ASSIGN_OR_RAISE(bool fallback_exist, file_system->Exists(fallback_source_table_path)); - if (exist == fallback_exist) { - return Status::Invalid( - fmt::format("Ambiguous table path: both table path {} and fallback table path {} are " - "present or absent", - source_table_path, fallback_source_table_path)); - } - std::string final_table_path = exist ? source_table_path : fallback_source_table_path; - return final_table_path; -} - Status BlobViewLookup::ExtractBlobDescriptors(const Identifier& identifier, const std::vector& field_ids, const std::shared_ptr& pool, diff --git a/src/paimon/core/utils/blob_view_lookup.h b/src/paimon/core/utils/blob_view_lookup.h index 169d9f83b..336863db4 100644 --- a/src/paimon/core/utils/blob_view_lookup.h +++ b/src/paimon/core/utils/blob_view_lookup.h @@ -29,6 +29,7 @@ #include "paimon/catalog/identifier.h" #include "paimon/common/catalog/catalog_context.h" #include "paimon/common/data/blob_view_struct.h" +#include "paimon/fs/file_system.h" #include "paimon/memory/bytes.h" #include "paimon/result.h" #include "paimon/utils/range.h" @@ -93,9 +94,6 @@ class BlobViewLookup { static std::vector> SplitRowRanges(const std::vector& row_ranges, int64_t target_rows_per_task); - - static Result GetTableLocation( - const std::shared_ptr& catalog_context, const Identifier& identifier); }; } // namespace paimon diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 0d566595e..80ea97e52 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -4156,8 +4156,8 @@ TEST_P(BlobTableInteTest, TestForwardBlobViewReference) { PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name); ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path)); - // The source table has no upstream warehouse configured: only a read with resolve - // dynamically disabled can succeed on it. + // The source table configures no explicit upstream warehouse; a default read falls back to the + // warehouse it lives in (dir_), where the copied upstream table sits. arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("view", true)}; std::map source_options = { @@ -4210,12 +4210,15 @@ TEST_P(BlobTableInteTest, TestForwardBlobViewReference) { WriteArray(source_table_path, {}, schema->field_names(), {write_struct})); ASSERT_OK(Commit(source_table_path, commit_msgs)); - // A default read fails on the missing upstream warehouse: the pass-through below is - // enabled by the dynamic option alone. + // With no explicit upstream warehouse, a default read now falls back to the source table's own + // warehouse (dir_) and resolves the copied upstream table instead of failing. The pass-through + // read below still uses the dynamic option to preserve the raw BlobViewStruct bytes. ASSERT_OK_AND_ASSIGN(auto source_plan, ScanTable(source_table_path)); - ASSERT_NOK_WITH_MSG( - ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr), - "BLOB_VIEW_UPSTREAM_WAREHOUSE"); + ASSERT_OK_AND_ASSIGN( + auto source_default_result, + ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr)); + ASSERT_TRUE(source_default_result); + ASSERT_EQ(source_default_result->length(), 8); ASSERT_OK_AND_ASSIGN( auto source_result, @@ -4577,7 +4580,7 @@ TEST_P(BlobTableInteTest, TestBlobViewFieldWithMultipleUpstreamTables) { << "expected:" << expected_with_rk->ToString(); } -TEST_P(BlobTableInteTest, TestBlobViewFailsWhenBothPathsAbsent) { +TEST_P(BlobTableInteTest, TestBlobViewFailsWhenUpstreamTableAbsent) { if (GetParam() == "mosaic") { return; } @@ -4589,10 +4592,10 @@ TEST_P(BlobTableInteTest, TestBlobViewFailsWhenBothPathsAbsent) { ASSERT_OK(WriteDanglingBlobViewRows(table_path)); - // reading fails because neither the table path nor the fallback one exists upstream + // reading fails because the upstream table does not exist at the catalog-issued location ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr), - "Ambiguous table path"); + "not found latest schema"); } TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceOfDeletedRow) { @@ -4613,7 +4616,7 @@ TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceOfDeletedRow) { // without the deletion vector the dangling reference fails the read ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr), - "Ambiguous table path"); + "not found latest schema"); ASSERT_OK_AND_ASSIGN(std::string anchor_file_name, DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs)); @@ -4659,7 +4662,7 @@ TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceInEveryRowRangeGroup // without deletion vectors both groups' dangling references fail the read ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr), - "Ambiguous table path"); + "not found latest schema"); // each group's anchor gets a deletion vector dropping its row 0, the dangling one ASSERT_OK_AND_ASSIGN(std::string anchor_file_name0, @@ -4698,7 +4701,7 @@ TEST_P(BlobTableInteTest, TestBlobViewPreReadHonorsRowRangesNotPredicate) { // reading the whole split fails on the dangling reference of row 0 ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr), - "Ambiguous table path"); + "not found latest schema"); // a predicate keeping only row 1 does not rescue the query: it is evaluated above the split // read, while the pre-read takes none, so row 0's reference is still collected and resolved @@ -4706,7 +4709,7 @@ TEST_P(BlobTableInteTest, TestBlobViewPreReadHonorsRowRangesNotPredicate) { FieldType::INT, Literal(101)); ASSERT_OK_AND_ASSIGN(auto filtered_plan, ScanTable(table_path, keep_row_1)); ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), filtered_plan, keep_row_1), - "Ambiguous table path"); + "not found latest schema"); // dropping the same row through the row ranges does: the pre-read honors them the same way // the main read does, so row 0 is never resolved @@ -4717,115 +4720,6 @@ TEST_P(BlobTableInteTest, TestBlobViewPreReadHonorsRowRangesNotPredicate) { /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 1)})); } -TEST_P(BlobTableInteTest, TestBlobViewWithFallbackPath) { - auto file_format = GetParam(); - if (file_format == "mosaic") { - return; - } - const std::string upstream_db_name = "fallback_db"; - const std::string upstream_table_name = "fallback_table"; - arrow::FieldVector upstream_fields = {arrow::field("f0", arrow::int32()), - BlobUtils::ToArrowField("blob", true)}; - auto upstream_schema = arrow::schema(upstream_fields); - std::map upstream_options = { - {Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"}, - {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"}, - {Options::BLOB_AS_DESCRIPTOR, "true"}, {Options::FILE_SYSTEM, "local"}}; - - // Create the upstream table at the fallback path: /db/table (no .db). - auto upstream_dir = UniqueTestDirectory::Create("local"); - std::string fallback_table_path = - PathUtil::JoinPath(upstream_dir->Str(), upstream_db_name + "/" + upstream_table_name); - - // Manually create schema at fallback path so it can be read as a valid paimon table. - { - // Use a temporary warehouse with Catalog to build the table data, then copy to fallback. - auto temp_dir = UniqueTestDirectory::Create("local"); - ::ArrowSchema c_schema; - ASSERT_TRUE(arrow::ExportSchema(*upstream_schema, &c_schema).ok()); - ASSERT_OK_AND_ASSIGN(auto catalog, - Catalog::Create(temp_dir->Str(), {{Options::FILE_SYSTEM, "local"}})); - ASSERT_OK(catalog->CreateDatabase(upstream_db_name, {}, /*ignore_if_exists=*/true)); - ASSERT_OK(catalog->CreateTable(Identifier(upstream_db_name, upstream_table_name), &c_schema, - /*partition_keys=*/{}, /*primary_keys=*/{}, upstream_options, - /*ignore_if_exists=*/false)); - std::string temp_table_path = - PathUtil::JoinPath(temp_dir->Str(), upstream_db_name + ".db/" + upstream_table_name); - - // Write data to the temp table. - std::string raw_json = R"([[0, "hello"], [1, "world"]])"; - auto raw_array = std::dynamic_pointer_cast( - arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(upstream_fields), raw_json) - .ValueOrDie()); - ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"})); - ASSERT_OK_AND_ASSIGN( - auto upstream_commit_msgs, - WriteArray(temp_table_path, {}, upstream_schema->field_names(), {desc_array})); - ASSERT_OK(Commit(temp_table_path, upstream_commit_msgs)); - - // Copy the temp table to the fallback path (without .db). - ASSERT_TRUE(TestUtil::CopyDirectory(temp_table_path, fallback_table_path)); - } - - // Build the downstream table. - arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), - BlobUtils::ToArrowField("view", true)}; - std::map options = { - {Options::FILE_FORMAT, file_format}, - {Options::BUCKET, "-1"}, - {Options::ROW_TRACKING_ENABLED, "true"}, - {Options::DATA_EVOLUTION_ENABLED, "true"}, - {Options::BLOB_VIEW_FIELD, "view"}, - {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, upstream_dir->Str()}, - {Options::FILE_SYSTEM, "local"}}; - CreateTable(fields, /*partition_keys=*/{}, options); - std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); - - // Write downstream rows referencing the upstream fallback table. - Identifier upstream_identifier(upstream_db_name, upstream_table_name); - arrow::LargeBinaryBuilder view_builder; - for (int64_t row = 0; row < 2; ++row) { - BlobViewStruct view_struct(upstream_identifier, /*field_id=*/1, /*row_id=*/row); - ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR serialized, view_struct.Serialize(pool_)); - ASSERT_TRUE( - view_builder - .Append(reinterpret_cast(serialized->data()), serialized->size()) - .ok()); - } - std::shared_ptr write_view_array; - ASSERT_TRUE(view_builder.Finish(&write_view_array).ok()); - auto write_f0_array = - arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100, 101])").ValueOrDie(); - auto write_struct = std::dynamic_pointer_cast( - arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}), - std::vector({"f0", "view"})) - .ValueOrDie()); - - auto schema = arrow::schema(fields); - ASSERT_OK_AND_ASSIGN(auto commit_msgs, - WriteArray(table_path, {}, schema->field_names(), {write_struct})); - ASSERT_OK(Commit(table_path, commit_msgs)); - - // Read and verify - ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); - ASSERT_OK_AND_ASSIGN(auto result, - ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr)); - ASSERT_TRUE(result); - auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie(); - auto read_struct = std::dynamic_pointer_cast(read_concat); - ASSERT_EQ(read_struct->length(), 2); - ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"})); - - std::string expected_json = R"([[100, "hello"], [101, "world"]])"; - auto expected_struct = std::dynamic_pointer_cast( - arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json) - .ValueOrDie()); - ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct)); - ASSERT_TRUE(result_array->Equals(expected_with_rk)) - << "result_array:" << result_array->ToString() << std::endl - << "expected:" << expected_with_rk->ToString(); -} - TEST_P(BlobTableInteTest, TestReadBlobDescriptorFieldFromJava) { auto file_format = GetParam(); if (file_format == "avro" || file_format == "mosaic") { From c585da8185e3d972ff6ef236c0d724b5947a5f76 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Sat, 10 Oct 2026 17:35:55 +0800 Subject: [PATCH 2/8] fix --- src/paimon/core/operation/read_context_test.cpp | 4 ++-- src/paimon/rest/rest_catalog.h | 2 ++ src/paimon/rest/rest_credential_provider.cpp | 9 +++++++-- src/paimon/rest/rest_credential_provider.h | 7 ++++++- 4 files changed, 17 insertions(+), 5 deletions(-) diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 14f70bc68..915402ea3 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -84,7 +84,7 @@ TEST(ReadContextTest, TestReadViaHeaderReportsTheStartingTable) { // The value is the identifier's flat JSON in the form-urlencoded flavor the REST server // expects, byte for byte what the Java client sends. const std::string expected = "%7B%22database%22%3A%22db1%22%2C%22object%22%3A%22t1%22%7D"; - for (const std::string& metastore : {"rest"}) { + for (const std::string metastore : {"rest"}) { auto catalog = std::make_shared(); catalog->SetTableFileSystem(std::make_shared()); @@ -120,7 +120,7 @@ TEST(ReadContextTest, TestReadViaHeaderNeedsARestCatalogAndATable) { // Only a REST metastore turns a `header.` option into a request header, and only a read naming // a table has one to report: the two guards the Java counterpart checks before copying the // options over. A catalog that is not one is told nothing. - for (const std::string& metastore : {"filesystem", "hive", "alake"}) { + for (const std::string metastore : {"filesystem", "hive", "alake"}) { auto catalog = std::make_shared(); catalog->SetTableFileSystem(std::make_shared()); diff --git a/src/paimon/rest/rest_catalog.h b/src/paimon/rest/rest_catalog.h index c9089d434..2659b67c4 100644 --- a/src/paimon/rest/rest_catalog.h +++ b/src/paimon/rest/rest_catalog.h @@ -131,6 +131,8 @@ class RestCatalog : public Catalog, public VersionManagedCatalog { const Identifier& data_identifier, const std::optional& branch, std::string* table_path, std::string* table_id) const; + /// Converts a `GetTableResponse` into a `TableSchema`, applying the table path, audit + /// and branch options. static Result> ToTableSchema( const GetTableResponse& response, const std::optional& branch); diff --git a/src/paimon/rest/rest_credential_provider.cpp b/src/paimon/rest/rest_credential_provider.cpp index fda2de9cc..5e93b84d4 100644 --- a/src/paimon/rest/rest_credential_provider.cpp +++ b/src/paimon/rest/rest_credential_provider.cpp @@ -65,7 +65,12 @@ size_t RestToken::Hash::operator()(const RestToken& rest_token) const { RestCredentialProvider::RestCredentialProvider(const std::shared_ptr& api, const Identifier& identifier, Clock clock) - : api_(api), + : RestCredentialProvider([api, identifier]() { return api->LoadTableToken(identifier); }, + identifier, std::move(clock)) {} + +RestCredentialProvider::RestCredentialProvider(TokenLoader token_loader, + const Identifier& identifier, Clock clock) + : token_loader_(std::move(token_loader)), identifier_(identifier), clock_(std::move(clock)), logger_(Logger::GetLogger("RestCredentialProvider")) {} @@ -139,7 +144,7 @@ std::map RestCredentialProvider::MergeOptionsWithCrede Status RestCredentialProvider::RefreshToken() const { PAIMON_LOG_INFO(logger_, "begin refresh data token for identifier [%s]", identifier_.ToString().c_str()); - PAIMON_ASSIGN_OR_RAISE(GetTableTokenResponse response, api_->LoadTableToken(identifier_)); + PAIMON_ASSIGN_OR_RAISE(GetTableTokenResponse response, token_loader_()); PAIMON_LOG_INFO(logger_, "end refresh data token for identifier [%s] expiresAtMillis [%ld]", identifier_.ToString().c_str(), static_cast(response.GetExpiresAtMillis())); diff --git a/src/paimon/rest/rest_credential_provider.h b/src/paimon/rest/rest_credential_provider.h index 3860ee671..c46f38d8a 100644 --- a/src/paimon/rest/rest_credential_provider.h +++ b/src/paimon/rest/rest_credential_provider.h @@ -65,6 +65,7 @@ struct RestToken { class RestCredentialProvider : public CredentialProvider { public: using Clock = std::function; + using TokenLoader = std::function()>; /// @param api Client of the catalog that issues the credentials. Shared because a /// provider commonly outlives the catalog it was obtained from. @@ -73,6 +74,10 @@ class RestCredentialProvider : public CredentialProvider { RestCredentialProvider(const std::shared_ptr& api, const Identifier& identifier, Clock clock = std::chrono::system_clock::now); + /// Creates a provider backed by an arbitrary table-token loader. This keeps the refresh and + /// file-system cache semantics shared by REST-compatible catalog implementations. + RestCredentialProvider(TokenLoader token_loader, const Identifier& identifier, Clock clock); + ~RestCredentialProvider() override = default; /// Returns credentials that are not about to expire, reloading them when needed. The @@ -103,7 +108,7 @@ class RestCredentialProvider : public CredentialProvider { /// Whether `token_` is absent or expires within the safe time. bool ShouldRefresh() const; - std::shared_ptr api_; + TokenLoader token_loader_; Identifier identifier_; Clock clock_; std::shared_ptr logger_; From d3b1ab1f9581224491952449b89056ddcc45e47d Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Mon, 12 Oct 2026 18:19:06 +0800 Subject: [PATCH 3/8] fix --- src/paimon/core/operation/read_context.cpp | 12 +++++++----- src/paimon/rest/rest_api.h | 5 +++++ 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index 336b8c036..5287991ff 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -34,6 +34,7 @@ #include "paimon/memory/memory_pool.h" #include "paimon/predicate/full_text_search.h" #include "paimon/predicate/vector_search.h" +#include "paimon/rest/rest_api.h" #include "paimon/status.h" #include "paimon/table/format/format_table.h" #include "rapidjson/document.h" @@ -45,11 +46,12 @@ class Predicate; namespace { -/// The catalog option that carries the `X-Paimon-Read-Via` request header, spelled with the prefix -/// a REST catalog strips when it turns an option into a header. That prefix is -/// `RestApi::kHeaderOptionPrefix`, not referenced here because its header pulls in the http client; -/// mirrors the Java `RESTApi.READ_VIA_OPTION`. -constexpr char kReadViaOption[] = "header.X-Paimon-Read-Via"; +/// The catalog option that carries the `X-Paimon-Read-Via` request header: the header name +/// spelled with the prefix a REST catalog strips when it turns an option into a header. Composed +/// from the `RestApi` constants the way the Java `CatalogEnvironment.READ_VIA_OPTION` is built from +/// `RESTApi.HEADER_PREFIX` and `RESTApi.READ_VIA_HEADER`. +const std::string kReadViaOption = + std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader; /// True when the options select a catalog that speaks the REST protocol, the only kind turning a /// `header.`-prefixed option into a request header. These same options decide which catalog a diff --git a/src/paimon/rest/rest_api.h b/src/paimon/rest/rest_api.h index 6efdc1806..722fc525c 100644 --- a/src/paimon/rest/rest_api.h +++ b/src/paimon/rest/rest_api.h @@ -80,6 +80,11 @@ class RestApi { static constexpr const char* kOptionUrlPrefix = "prefix"; /// Options with this prefix are sent as http headers (with the prefix stripped). static constexpr const char* kHeaderOptionPrefix = "header."; + /// Optional header carrying the URL-encoded `Identifier` JSON of the table which initiated a + /// dependency read. It only provides request context; servers must not treat it as + /// authorization proof. Mirrors the Java `RESTApi.READ_VIA_HEADER`, and a REST catalog turns + /// the `kHeaderOptionPrefix`-prefixed option into this header. + static constexpr const char* kReadViaHeader = "X-Paimon-Read-Via"; /// Credentials are refreshed once they expire in less than this, so that a request /// signed with them cannot expire while it is in flight. static constexpr int64_t kTokenExpirationSafeTimeMillis = 3600000; From 7d0407e2bd651251d83b7ddf639986e652372bd3 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 13 Oct 2026 10:36:21 +0800 Subject: [PATCH 4/8] fix --- src/paimon/core/operation/read_context.cpp | 39 +++++++------------ .../core/operation/read_context_test.cpp | 14 +++---- src/paimon/testing/mock/mock_catalog.h | 5 +++ 3 files changed, 26 insertions(+), 32 deletions(-) diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index 5287991ff..dd971cb49 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -46,20 +46,14 @@ class Predicate; namespace { -/// The catalog option that carries the `X-Paimon-Read-Via` request header: the header name -/// spelled with the prefix a REST catalog strips when it turns an option into a header. Composed -/// from the `RestApi` constants the way the Java `CatalogEnvironment.READ_VIA_OPTION` is built from -/// `RESTApi.HEADER_PREFIX` and `RESTApi.READ_VIA_HEADER`. -const std::string kReadViaOption = - std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader; - -/// True when the options select a catalog that speaks the REST protocol, the only kind turning a -/// `header.`-prefixed option into a request header. These same options decide which catalog a -/// dependency table is read through, so when they select none there is nobody to send the header -/// to. The Java counterpart answers this from the catalog loader it was built with. -bool IsRestMetastore(const std::map& options) { - auto metastore = options.find(CatalogOptions::METASTORE); - if (metastore == options.end()) { +/// True when the catalog-level options - the ones the catalog was created with, not the read's own +/// table options - select the REST protocol, the only kind turning a `header.`-prefixed option into +/// a request header. `Catalog::Create` dispatches on this same `metastore` key, so a catalog that +/// speaks REST is exactly one whose `GetOptions()` names it. The Java counterpart answers this from +/// the catalog loader / `catalogContext.options()` it was built with. +bool IsRestMetastore(const std::map& catalog_options) { + auto metastore = catalog_options.find(CatalogOptions::METASTORE); + if (metastore == catalog_options.end()) { return false; } // Matched leniently in lower case, the way `Catalog::Create` matches it. @@ -456,17 +450,12 @@ Result> ReadContextBuilder::Finish() { if (impl_->enable_multi_thread_row_to_batch_ && impl_->row_to_batch_thread_number_ <= 0) { return Status::Invalid("row to batch thread number should be greater than 0"); } - // A read that goes through a catalog says which table it started from, which is what lets the - // metastore authorize a dependency table - the blob table behind a BlobView - through the owner - // of that table. Mirrors the Java `CatalogEnvironment.dependencyReadContext()` and both of its - // guards: only a REST metastore understands the header, and a value already there wins, so a - // chain of views reports its outermost one instead of the innermost. It travels in the options, - // which `TableRead::Create` merges into the table options and the blob view read then hands to - // the catalog it reads the dependency table through, so it reaches that request with no second - // channel to keep in step. - if (impl_->identifier_ && IsRestMetastore(impl_->options_) && - impl_->options_.count(kReadViaOption) == 0) { - impl_->options_[kReadViaOption] = + + const std::string read_via_option = + std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader; + if (impl_->identifier_ && IsRestMetastore(impl_->catalog_->GetOptions()) && + impl_->options_.count(read_via_option) == 0) { + impl_->options_[read_via_option] = UrlUtils::EncodeString(IdentifierToFlatJson(impl_->identifier_.value())); } auto ctx = std::make_unique( diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 915402ea3..70755f25c 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -87,18 +87,18 @@ TEST(ReadContextTest, TestReadViaHeaderReportsTheStartingTable) { for (const std::string metastore : {"rest"}) { auto catalog = std::make_shared(); catalog->SetTableFileSystem(std::make_shared()); + catalog->SetOptions({{CatalogOptions::METASTORE, metastore}}); ReadContextBuilder builder("table_root_path"); - builder.AddOption(CatalogOptions::METASTORE, metastore); ASSERT_OK_AND_ASSIGN(auto ctx, builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); const std::map& options = ctx->GetOptions(); ASSERT_EQ(options.count("header.X-Paimon-Read-Via"), 1u) << metastore; ASSERT_EQ(options.at("header.X-Paimon-Read-Via"), expected) << metastore; - // The metastore is the only other option, so the header is added rather than anything - // rewritten around it. - ASSERT_EQ(options.size(), 2u) << metastore; + // The metastore is a catalog-level option, not a read option, so the injected header is + // the only entry the read options gain. + ASSERT_EQ(options.size(), 1u) << metastore; } } @@ -108,9 +108,9 @@ TEST(ReadContextTest, TestReadViaHeaderKeepsTheOutermostView) { // `CatalogEnvironment.dependencyReadContext()`. auto catalog = std::make_shared(); catalog->SetTableFileSystem(std::make_shared()); + catalog->SetOptions({{CatalogOptions::METASTORE, "rest"}}); ReadContextBuilder builder("table_root_path"); - builder.AddOption(CatalogOptions::METASTORE, "rest"); builder.AddOption("header.X-Paimon-Read-Via", "outer-view"); ASSERT_OK_AND_ASSIGN(auto ctx, builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); ASSERT_EQ(ctx->GetOptions().at("header.X-Paimon-Read-Via"), "outer-view"); @@ -123,9 +123,9 @@ TEST(ReadContextTest, TestReadViaHeaderNeedsARestCatalogAndATable) { for (const std::string metastore : {"filesystem", "hive", "alake"}) { auto catalog = std::make_shared(); catalog->SetTableFileSystem(std::make_shared()); + catalog->SetOptions({{CatalogOptions::METASTORE, metastore}}); ReadContextBuilder builder("table_root_path"); - builder.AddOption(CatalogOptions::METASTORE, metastore); ASSERT_OK_AND_ASSIGN(auto ctx, builder.WithCatalog(catalog, Identifier("db1", "t1")).Finish()); ASSERT_EQ(ctx->GetOptions().count("header.X-Paimon-Read-Via"), 0u) << metastore; @@ -154,9 +154,9 @@ TEST(ReadContextTest, TestReadViaHeaderReportsTheBranchOfTheStartingTable) { // naming the data table. The metastore is matched in upper case here as well. auto catalog = std::make_shared(); catalog->SetTableFileSystem(std::make_shared()); + catalog->SetOptions({{CatalogOptions::METASTORE, "REST"}}); ReadContextBuilder builder("table_root_path"); - builder.AddOption(CatalogOptions::METASTORE, "REST"); ASSERT_OK_AND_ASSIGN(auto ctx, builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")).Finish()); ASSERT_EQ(ctx->GetOptions().at("header.X-Paimon-Read-Via"), diff --git a/src/paimon/testing/mock/mock_catalog.h b/src/paimon/testing/mock/mock_catalog.h index 88b9b914a..b9954e313 100644 --- a/src/paimon/testing/mock/mock_catalog.h +++ b/src/paimon/testing/mock/mock_catalog.h @@ -251,6 +251,11 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog { const std::map& GetOptions() const override { return options_; } + + /// Sets the catalog-level options `GetOptions()` reports, e.g. `metastore`. + void SetOptions(const std::map& options) { + options_ = options; + } Result> ListSnapshots(const Identifier&, const std::string&) const override { return Unsupported(); From e6c3c0c7cdb6ca9a31fab9f41ad2dd707e155b84 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 13 Oct 2026 10:39:01 +0800 Subject: [PATCH 5/8] fix --- src/paimon/core/operation/read_context.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index dd971cb49..4a5d8c14d 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -453,7 +453,8 @@ Result> ReadContextBuilder::Finish() { const std::string read_via_option = std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader; - if (impl_->identifier_ && IsRestMetastore(impl_->catalog_->GetOptions()) && + if (impl_->catalog_ != nullptr && impl_->identifier_ && + IsRestMetastore(impl_->catalog_->GetOptions()) && impl_->options_.count(read_via_option) == 0) { impl_->options_[read_via_option] = UrlUtils::EncodeString(IdentifierToFlatJson(impl_->identifier_.value())); From 68f3497e219d6fa16b5fbfbda209aa8fd6067175 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 13 Oct 2026 11:16:28 +0800 Subject: [PATCH 6/8] fix --- src/paimon/core/operation/read_context.cpp | 17 +++++++++++++---- src/paimon/core/operation/read_context_test.cpp | 8 +++++--- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index 4a5d8c14d..ed23a180a 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -454,10 +454,19 @@ Result> ReadContextBuilder::Finish() { const std::string read_via_option = std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader; if (impl_->catalog_ != nullptr && impl_->identifier_ && - IsRestMetastore(impl_->catalog_->GetOptions()) && - impl_->options_.count(read_via_option) == 0) { - impl_->options_[read_via_option] = - UrlUtils::EncodeString(IdentifierToFlatJson(impl_->identifier_.value())); + IsRestMetastore(impl_->catalog_->GetOptions())) { + // A blob-view dependency read (BlobViewLookup) rebuilds its own catalog from a + // CatalogContext whose options come from these read options, so they carry none of the + // catalog-level connection settings. Hand over the ones RestApi::Create needs to come back + // as the same REST catalog - metastore, uri and the auth options - instead of letting it + // silently degrade to a filesystem catalog; a key the caller already set wins. + for (const auto& [key, value] : impl_->catalog_->GetOptions()) { + impl_->options_.emplace(key, value); + } + // Reports the outermost starting table; an existing value, set by an outer read, is kept. + impl_->options_.emplace( + read_via_option, + UrlUtils::EncodeString(IdentifierToFlatJson(impl_->identifier_.value()))); } auto ctx = std::make_unique( impl_->path_, branch, impl_->read_field_names_, impl_->read_field_ids_, impl_->predicate_, diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 70755f25c..766049c1c 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -96,9 +96,11 @@ TEST(ReadContextTest, TestReadViaHeaderReportsTheStartingTable) { const std::map& options = ctx->GetOptions(); ASSERT_EQ(options.count("header.X-Paimon-Read-Via"), 1u) << metastore; ASSERT_EQ(options.at("header.X-Paimon-Read-Via"), expected) << metastore; - // The metastore is a catalog-level option, not a read option, so the injected header is - // the only entry the read options gain. - ASSERT_EQ(options.size(), 1u) << metastore; + // The catalog-level connection options are handed to the read options so the blob-view + // dependency catalog can be rebuilt as the same REST catalog; here that is the metastore. + ASSERT_EQ(options.count(CatalogOptions::METASTORE), 1u) << metastore; + ASSERT_EQ(options.at(CatalogOptions::METASTORE), metastore) << metastore; + ASSERT_EQ(options.size(), 2u) << metastore; } } From 021fd2bfa8058ca28d6432e4f84158af8878f252 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 13 Oct 2026 11:44:42 +0800 Subject: [PATCH 7/8] fix --- src/paimon/core/operation/data_evolution_split_read.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index f6ede5fa9..97797e345 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -278,6 +278,7 @@ Result> DataEvolutionSplitRead::WrapWithBlobViewRes // `/
`) components off this table's path yields that warehouse. std::optional warehouse_path = options_.GetBlobViewUpstreamWarehouse(); if (!warehouse_path) { + // TODO(yonghao.fyh): refactor warehouse path in the future warehouse_path = PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(context_->GetPath())); } From 4c194648db7a50d0ef76c83c7a8e1576714a49cd Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 13 Oct 2026 13:59:55 +0800 Subject: [PATCH 8/8] fix --- src/paimon/core/operation/read_context.cpp | 10 +++++--- .../core/operation/read_context_test.cpp | 25 +++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index ed23a180a..1779d7300 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -70,6 +70,10 @@ std::string IdentifierToFlatJson(const Identifier& identifier) { auto& allocator = document.GetAllocator(); document.AddMember(rapidjson::StringRef("database"), rapidjson::Value(identifier.GetDatabaseName(), allocator), allocator); + // C++ `GetTableName()` returns the stored object name - branch and system-table suffix + // included - which is the Java `getObjectName()` this field mirrors, not Java's stripped + // `getTableName()` (that one is `GetDataTableName()` here). So a branch read reports the + // branch. document.AddMember(rapidjson::StringRef("object"), rapidjson::Value(identifier.GetTableName(), allocator), allocator); // Compact, not the pretty form `RapidJsonUtil::ToJson` writes: the value goes into a header. @@ -423,9 +427,9 @@ Result> ReadContextBuilder::Finish() { PAIMON_ASSIGN_OR_RAISE(impl_->specific_file_system_, impl_->catalog_->GetTableFileSystem(impl_->identifier_.value())); } - PAIMON_ASSIGN_OR_RAISE(std::string branch, - BranchManager::ResolveBranch(/*identifier=*/std::nullopt, - impl_->options_, impl_->branch_, "read")); + PAIMON_ASSIGN_OR_RAISE( + std::string branch, + BranchManager::ResolveBranch(impl_->identifier_, impl_->options_, impl_->branch_, "read")); if (impl_->enable_prefetch_ && impl_->prefetch_max_parallel_num_ == 0) { return Status::Invalid("prefetch max parallel num should be greater than 0"); } diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 766049c1c..972f6f21f 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -313,6 +313,31 @@ TEST(ReadContextTest, TestBranch) { ASSERT_NOK_WITH_MSG(escaping_builder.Finish(), "branch name cannot contain path separators"); } +TEST(ReadContextTest, TestBranchAgreesWithIdentifier) { + // A read through a catalog names its table with an Identifier that can itself carry a branch, + // so that branch and the one WithBranch() names must agree, exactly as the write and commit + // builders already require: one read cannot be aimed at two branches at once. + auto catalog = std::make_shared(); + catalog->SetTableFileSystem(std::make_shared()); + + // The identifier's branch is honored when WithBranch() stays silent. + ReadContextBuilder identifier_builder("table_root_path"); + identifier_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")); + ASSERT_OK_AND_ASSIGN(auto identifier_ctx, identifier_builder.Finish()); + ASSERT_EQ("dev", identifier_ctx->GetBranch()); + + // An agreeing WithBranch() is accepted. + ReadContextBuilder agreeing_builder("table_root_path"); + agreeing_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")).WithBranch("dev"); + ASSERT_OK_AND_ASSIGN(auto agreeing_ctx, agreeing_builder.Finish()); + ASSERT_EQ("dev", agreeing_ctx->GetBranch()); + + // A disagreeing WithBranch() is refused, and both branches are named. + ReadContextBuilder mixed_builder("table_root_path"); + mixed_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")).WithBranch("rt"); + ASSERT_NOK_WITH_MSG(mixed_builder.Finish(), "but both 'dev' and 'rt' were named"); +} + TEST(ReadContextTest, TestFileSystemAndSchemeMapConflict) { ReadContextBuilder builder("table_root_path"); auto fs = std::make_shared();