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..97797e345 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -271,17 +271,26 @@ 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"); + // TODO(yonghao.fyh): refactor warehouse path in the future + 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 9e0dc5fb6..086d56d14 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..1779d7300 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -18,23 +18,73 @@ #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" #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" +#include "rapidjson/stringbuffer.h" +#include "rapidjson/writer.h" namespace paimon { class Predicate; +namespace { + +/// 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. + 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); + // 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. + 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, @@ -377,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"); } @@ -404,6 +454,24 @@ 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"); } + + const std::string read_via_option = + std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader; + if (impl_->catalog_ != nullptr && impl_->identifier_ && + 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_, 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 b820d245e..176df8b0e 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,93 @@ 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()); + catalog->SetOptions({{CatalogOptions::METASTORE, metastore}}); + + ReadContextBuilder builder("table_root_path"); + 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 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; + } +} + +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()); + catalog->SetOptions({{CatalogOptions::METASTORE, "rest"}}); + + ReadContextBuilder builder("table_root_path"); + 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()); + catalog->SetOptions({{CatalogOptions::METASTORE, metastore}}); + + ReadContextBuilder builder("table_root_path"); + 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()); + catalog->SetOptions({{CatalogOptions::METASTORE, "REST"}}); + + ReadContextBuilder builder("table_root_path"); + 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()); @@ -225,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(); 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/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; 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_; 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(); 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") {