This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 758d225c refactor(blob-view): resolve blob view dependencies through
catalog context (#455)
758d225c is described below
commit 758d225c2a364cd8da976c1a70b7837232e41d77
Author: lszskye <[email protected]>
AuthorDate: Fri Oct 23 13:13:39 2026 +0800
refactor(blob-view): resolve blob view dependencies through catalog context
(#455)
---
include/paimon/defs.h | 6 -
src/paimon/common/defs.cpp | 1 -
src/paimon/core/core_options.cpp | 8 -
src/paimon/core/core_options.h | 1 -
src/paimon/core/core_options_test.cpp | 4 -
.../core/operation/data_evolution_split_read.cpp | 21 +--
.../core/operation/internal_read_context.cpp | 63 +++++++
src/paimon/core/operation/internal_read_context.h | 4 +
.../core/operation/internal_read_context_test.cpp | 195 +++++++++++++++++++--
src/paimon/core/operation/read_context.cpp | 63 -------
src/paimon/core/operation/read_context_test.cpp | 96 ++--------
src/paimon/core/table/source/table_read_test.cpp | 35 ++++
test/inte/blob_table_inte_test.cpp | 76 +++-----
13 files changed, 326 insertions(+), 247 deletions(-)
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 709d4490..2c2ef191 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -577,12 +577,6 @@ 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" - 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<BLOB> element or a MAP<..., BLOB> value, when the
referenced file does
/// not exist at write time.
diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp
index d709d6f8..9ddefb16 100644
--- a/src/paimon/common/defs.cpp
+++ b/src/paimon/common/defs.cpp
@@ -154,7 +154,6 @@ const char Options::BLOB_DESCRIPTOR_FIELD[] =
"blob-descriptor-field";
const char Options::FALLBACK_BLOB_DESCRIPTOR_FIELD[] =
"blob.stored-descriptor-fields";
const char Options::BLOB_VIEW_FIELD[] = "blob-view-field";
const char Options::BLOB_VIEW_RESOLVE_ENABLED[] = "blob-view.resolve.enabled";
-const char Options::BLOB_VIEW_UPSTREAM_WAREHOUSE[] =
"blob-view-upstream-warehouse";
const char Options::BLOB_WRITE_NULL_ON_MISSING_FILE[] =
"blob-write-null-on-missing-file";
const char Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE[] =
"blob-write-null-on-fetch-failure";
const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp
index 666d7685..b02930ff 100644
--- a/src/paimon/core/core_options.cpp
+++ b/src/paimon/core/core_options.cpp
@@ -390,7 +390,6 @@ struct CoreOptions::Impl {
std::optional<std::string> field_default_func;
std::optional<std::string> scan_fallback_branch;
std::optional<std::string> data_file_external_paths;
- std::optional<std::string> blob_view_upstream_warehouse;
std::optional<std::string> changelog_file_compression;
std::optional<std::string> global_index_external_path;
std::optional<std::string> scan_tag_name;
@@ -582,9 +581,6 @@ struct CoreOptions::Impl {
PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(Options::BLOB_VIEW_FIELD,
Options::FIELDS_SEPARATOR,
&blob_view_fields,
/*need_trim=*/true));
- // Parse blob-view-upstream-warehouse - warehouse path for configured
blob view fields
- PAIMON_RETURN_NOT_OK(
- parser.Parse(Options::BLOB_VIEW_UPSTREAM_WAREHOUSE,
&blob_view_upstream_warehouse));
// Parse blob-view.resolve.enabled - whether to resolve blob view
fields at read time
PAIMON_RETURN_NOT_OK(
parser.Parse<bool>(Options::BLOB_VIEW_RESOLVE_ENABLED,
&blob_view_resolve_enabled));
@@ -1966,10 +1962,6 @@ const std::vector<std::string>&
CoreOptions::GetBlobViewFields() const {
return impl_->blob_view_fields;
}
-std::optional<std::string> CoreOptions::GetBlobViewUpstreamWarehouse() const {
- return impl_->blob_view_upstream_warehouse;
-}
-
bool CoreOptions::BlobViewResolveEnabled() const {
return impl_->blob_view_resolve_enabled;
}
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index 8bf79426..34aaad53 100644
--- a/src/paimon/core/core_options.h
+++ b/src/paimon/core/core_options.h
@@ -304,7 +304,6 @@ class PAIMON_EXPORT CoreOptions {
const std::vector<std::string>& GetBlobFields() const;
const std::vector<std::string>& GetBlobDescriptorFields() const;
const std::vector<std::string>& GetBlobViewFields() const;
- std::optional<std::string> GetBlobViewUpstreamWarehouse() const;
bool BlobViewResolveEnabled() const;
std::vector<std::string> GetBlobInlineFields() const;
diff --git a/src/paimon/core/core_options_test.cpp
b/src/paimon/core/core_options_test.cpp
index e3625db8..fa2c7780 100644
--- a/src/paimon/core/core_options_test.cpp
+++ b/src/paimon/core/core_options_test.cpp
@@ -175,7 +175,6 @@ TEST(CoreOptionsTest, TestDefaultValue) {
ASSERT_TRUE(core_options.GetBlobDescriptorFields().empty());
ASSERT_TRUE(core_options.GetBlobViewFields().empty());
ASSERT_TRUE(core_options.GetBlobInlineFields().empty());
- ASSERT_EQ(std::nullopt, core_options.GetBlobViewUpstreamWarehouse());
ASSERT_TRUE(core_options.BlobViewResolveEnabled());
ASSERT_TRUE(core_options.LegacyPartitionNameEnabled());
ASSERT_TRUE(core_options.GlobalIndexEnabled());
@@ -303,7 +302,6 @@ TEST(CoreOptionsTest, TestFromMap) {
{Options::BLOB_DESCRIPTOR_FIELD, "blob3,blob4"},
{Options::BLOB_AS_DESCRIPTOR, "true"},
{Options::BLOB_VIEW_FIELD, "blob5"},
- {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE,
"FILE:///tmp/blob_view_upstream_warehouse/"},
{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"},
{Options::PARTITION_GENERATE_LEGACY_NAME, "false"},
{Options::GLOBAL_INDEX_ENABLED, "false"},
@@ -476,8 +474,6 @@ TEST(CoreOptionsTest, TestFromMap) {
ASSERT_EQ(core_options.GetBlobViewFields(),
std::vector<std::string>({"blob5"}));
ASSERT_EQ(core_options.GetBlobInlineFields(),
std::vector<std::string>({"blob3", "blob4", "blob5"}));
- ASSERT_EQ(core_options.GetBlobViewUpstreamWarehouse(),
-
std::optional<std::string>("FILE:///tmp/blob_view_upstream_warehouse/"));
ASSERT_FALSE(core_options.BlobViewResolveEnabled());
ASSERT_FALSE(core_options.LegacyPartitionNameEnabled());
ASSERT_FALSE(core_options.GlobalIndexEnabled());
diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp
b/src/paimon/core/operation/data_evolution_split_read.cpp
index e3a52c6f..6d065b46 100644
--- a/src/paimon/core/operation/data_evolution_split_read.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read.cpp
@@ -34,7 +34,6 @@
#include "arrow/array/array_binary.h"
#include "arrow/array/array_nested.h"
#include "arrow/c/bridge.h"
-#include "paimon/common/catalog/catalog_context.h"
#include "paimon/common/data/blob_defs.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/data/blob_view_struct.h"
@@ -49,7 +48,6 @@
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/object_utils.h"
-#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/range_helper.h"
#include "paimon/common/utils/vector_store_utils.h"
#include "paimon/core/core_options.h"
@@ -282,33 +280,18 @@ Result<std::unique_ptr<BatchReader>>
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
`<database>.db/<table>` (or
- // `<database>/<table>`) components off this table's path yields that
warehouse.
- std::optional<std::string> warehouse_path =
options_.GetBlobViewUpstreamWarehouse();
- if (!warehouse_path) {
- // TODO(yonghao.fyh): refactor warehouse path in the future
- warehouse_path =
-
PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(context_->GetPath()));
- }
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> pre_reader,
CreateBlobViewReader(data_split,
read_blob_view_fields, row_ranges));
PAIMON_ASSIGN_OR_RAISE(std::unordered_set<BlobViewStruct>
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<CatalogContext>(
- 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;
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Executor> executor,
CreateDefaultExecutor(thread_num));
PAIMON_ASSIGN_OR_RAISE(
BlobViewResolver resolver,
- BlobViewLookup::CreateResolver(blob_view_structs, catalog_context,
pool_, executor));
+ BlobViewLookup::CreateResolver(
+ blob_view_structs, context_->CreateDependencyCatalogContext(),
pool_, executor));
return
std::make_unique<BlobViewResolvingBatchReader>(std::move(inner_reader),
std::move(read_blob_view_fields),
std::move(resolver),
arrow_pool_);
diff --git a/src/paimon/core/operation/internal_read_context.cpp
b/src/paimon/core/operation/internal_read_context.cpp
index a89eb0ae..f36a15d7 100644
--- a/src/paimon/core/operation/internal_read_context.cpp
+++ b/src/paimon/core/operation/internal_read_context.cpp
@@ -27,6 +27,9 @@
#include "arrow/c/abi.h"
#include "arrow/c/bridge.h"
#include "fmt/format.h"
+#include "paimon/catalog/catalog.h"
+#include "paimon/catalog_options.h"
+#include "paimon/common/catalog/catalog_context.h"
#include "paimon/common/data/variant/variant_access_utils.h"
#include "paimon/common/data/variant/variant_type_utils.h"
#include "paimon/common/predicate/full_text_match_predicate.h"
@@ -36,6 +39,9 @@
#include "paimon/common/utils/arrow/mem_utils.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/common/utils/string_utils.h"
+#include "paimon/common/utils/url_utils.h"
#include "paimon/core/options/map_storage_layout.h"
#include "paimon/core/schema/arrow_schema_validator.h"
#include "paimon/core/utils/nested_projection_utils.h"
@@ -43,9 +49,66 @@
#include "paimon/predicate/full_text_search.h"
#include "paimon/predicate/predicate_utils.h"
#include "paimon/predicate/vector_search.h"
+#include "paimon/rest/rest_api.h"
#include "paimon/status.h"
+#include "rapidjson/document.h"
+#include "rapidjson/stringbuffer.h"
+#include "rapidjson/writer.h"
namespace paimon {
+namespace {
+
+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);
+ // Keep branch and system-table suffixes in the reported object name.
+ document.AddMember(rapidjson::StringRef("object"),
+ rapidjson::Value(identifier.GetObjectName(),
allocator), allocator);
+ rapidjson::StringBuffer buffer;
+ rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+ document.Accept(writer);
+ return buffer.GetString();
+}
+
+} // namespace
+
+std::shared_ptr<CatalogContext>
InternalReadContext::CreateDependencyCatalogContext() const {
+ const auto& catalog = read_context_->GetCatalog();
+ if (!catalog) {
+ // Preserve path-only filesystem reads using
<warehouse>/<database>.db/<table>.
+ return std::make_shared<CatalogContext>(
+ PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(GetPath())),
options_.ToMap(),
+ GetFileSystemSchemeToIdentifierMap());
+ }
+
+ // Preserve the dependency catalog's effective settings: table options,
then catalog
+ // options, then explicit read overrides. The upstream reader loads its
own table options;
+ // ReadContextBuilder must not copy these catalog settings back into that
read.
+ auto dependency_options = options_.ToMap();
+ const auto& catalog_options = catalog->GetOptions();
+ for (const auto& [key, value] : catalog_options) {
+ dependency_options[key] = value;
+ }
+ const auto& read_options = read_context_->GetOptions();
+ for (const auto& [key, value] : read_options) {
+ dependency_options[key] = value;
+ }
+ const std::string read_via_option =
+ std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader;
+ auto metastore = dependency_options.find(CatalogOptions::METASTORE);
+ if (metastore != dependency_options.end() &&
+ StringUtils::ToLowerCase(metastore->second) == "rest" &&
read_context_->GetIdentifier() &&
+ dependency_options.count(read_via_option) == 0) {
+ // Only fill in a missing header when the merged options select a REST
dependency catalog.
+ dependency_options[read_via_option] =
+
UrlUtils::EncodeString(IdentifierToFlatJson(*read_context_->GetIdentifier()));
+ }
+ // A REST warehouse identifies the catalog, independent of the table's
storage location.
+ return std::make_shared<CatalogContext>(catalog->GetRootPath(),
dependency_options,
+
GetFileSystemSchemeToIdentifierMap());
+}
Result<std::shared_ptr<arrow::Field>>
InternalReadContext::AlignReadFieldWithTableFieldIds(
const std::shared_ptr<arrow::Field>& read_field,
diff --git a/src/paimon/core/operation/internal_read_context.h
b/src/paimon/core/operation/internal_read_context.h
index d3601fdf..9de415d5 100644
--- a/src/paimon/core/operation/internal_read_context.h
+++ b/src/paimon/core/operation/internal_read_context.h
@@ -35,6 +35,7 @@ class Schema;
} // namespace arrow
namespace paimon {
+struct CatalogContext;
class Executor;
class MemoryPool;
class Predicate;
@@ -80,6 +81,9 @@ class InternalReadContext {
const std::string& GetPath() const {
return read_context_->GetPath();
}
+
+ std::shared_ptr<CatalogContext> CreateDependencyCatalogContext() const;
+
const std::vector<std::string>& GetPartitionKeys() const {
return table_schema_->PartitionKeys();
}
diff --git a/src/paimon/core/operation/internal_read_context_test.cpp
b/src/paimon/core/operation/internal_read_context_test.cpp
index ca7b79dc..6e4957f4 100644
--- a/src/paimon/core/operation/internal_read_context_test.cpp
+++ b/src/paimon/core/operation/internal_read_context_test.cpp
@@ -29,6 +29,8 @@
#include "arrow/type.h"
#include "fmt/format.h"
#include "gtest/gtest.h"
+#include "paimon/catalog_options.h"
+#include "paimon/common/catalog/catalog_context.h"
#include "paimon/common/table/special_fields.h"
#include "paimon/common/types/data_field.h"
#include "paimon/core/schema/schema_manager.h"
@@ -40,6 +42,8 @@
#include "paimon/predicate/predicate_builder.h"
#include "paimon/predicate/vector_search.h"
#include "paimon/status.h"
+#include "paimon/testing/mock/mock_catalog.h"
+#include "paimon/testing/mock/mock_file_system.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -78,25 +82,186 @@ 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.
+TEST(InternalReadContext, TestDependencyReadMergesTableCatalogAndReadOptions) {
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<LocalFileSystem>(),
read_context->GetPath());
+ SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), path);
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";
+ auto catalog = std::make_shared<MockVersionManagedCatalog>();
+ catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
+ const std::map<std::string, std::string> catalog_options = {
+ {CatalogOptions::METASTORE, "rest"},
+ {CatalogOptions::URI, "http://catalog"},
+ {CatalogOptions::TOKEN, "catalog-token"},
+ {CatalogOptions::DATA_TOKEN_ENABLED, "true"}};
+ catalog->SetOptions(catalog_options);
+ const std::map<std::string, std::string> read_options = {
+ {CatalogOptions::METASTORE, "filesystem"},
+ {CatalogOptions::URI, "http://read"},
+ {CatalogOptions::TOKEN, "read-token"},
+ {Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}};
+ ReadContextBuilder builder(path);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<ReadContext> read_context,
+ builder.WithCatalog(catalog, Identifier("db1",
"t1")).SetOptions(read_options).Finish());
std::map<std::string, std::string> options = table_schema->Options();
- options["header.X-Paimon-Read-Via"] = read_via;
+ options[Options::READ_BATCH_SIZE] = "64";
+ options[CatalogOptions::DATA_TOKEN_ENABLED] = "false";
+ options["header.X-Paimon-Read-Via"] = "table-view";
+ for (const auto& [key, value] : read_options) {
+ options[key] = value;
+ }
+ ASSERT_OK_AND_ASSIGN(auto internal_context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+
+ auto dependency_context =
internal_context->CreateDependencyCatalogContext();
+ auto expected_options = options;
+ expected_options[CatalogOptions::DATA_TOKEN_ENABLED] = "true";
+ ASSERT_EQ(dependency_context->root_path, catalog->GetRootPath());
+ ASSERT_EQ(dependency_context->options, expected_options);
+ ASSERT_EQ(dependency_context->options.at("header.X-Paimon-Read-Via"),
"table-view");
+ ASSERT_EQ(read_context->GetOptions(), read_options);
+ ASSERT_EQ(catalog->GetOptions(), catalog_options);
+ ASSERT_EQ(internal_context->GetCoreOptions().ToMap(), options);
+}
+
+TEST(InternalReadContext, TestDependencyReadKeepsTheOutermostView) {
+ std::string path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
+ SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ auto catalog = std::make_shared<MockVersionManagedCatalog>();
+ catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
+ struct HeaderCase {
+ const char* catalog_header;
+ const char* read_header;
+ const char* expected_header;
+ };
+ const std::vector<HeaderCase> cases = {{nullptr, nullptr, "table-view"},
+ {"catalog-view", nullptr,
"catalog-view"},
+ {"catalog-view", "explicit-view",
"explicit-view"}};
+ for (const auto& test_case : cases) {
+ std::map<std::string, std::string> catalog_options =
{{CatalogOptions::METASTORE, "rest"}};
+ if (test_case.catalog_header) {
+ catalog_options["header.X-Paimon-Read-Via"] =
test_case.catalog_header;
+ }
+ catalog->SetOptions(catalog_options);
+ std::map<std::string, std::string> read_options;
+ if (test_case.read_header) {
+ read_options["header.X-Paimon-Read-Via"] = test_case.read_header;
+ }
+ ReadContextBuilder builder(path);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
+ builder.WithCatalog(catalog, Identifier("db1",
"nested-view"))
+ .SetOptions(read_options)
+ .Finish());
+ auto options = table_schema->Options();
+ options["header.X-Paimon-Read-Via"] =
+ test_case.read_header ? test_case.read_header : "table-view";
+ ASSERT_OK_AND_ASSIGN(auto internal_context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+ auto dependency_context =
internal_context->CreateDependencyCatalogContext();
+ ASSERT_EQ(dependency_context->options.at("header.X-Paimon-Read-Via"),
+ test_case.expected_header);
+ ASSERT_EQ(read_context->GetOptions(), read_options);
+ ASSERT_EQ(catalog->GetOptions(), catalog_options);
+ ASSERT_EQ(internal_context->GetCoreOptions().ToMap(), options);
+ }
+}
+
+TEST(InternalReadContext, TestDependencyReadHeaderUsesFinalMetastore) {
+ std::string path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
+ SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ struct MetastoreCase {
+ const char* name;
+ const char* table_metastore;
+ const char* catalog_metastore;
+ const char* read_metastore;
+ const char* expected_metastore;
+ bool generates_header;
+ };
+ const std::vector<MetastoreCase> cases = {
+ {"read disables REST", nullptr, "rest", "filesystem", "filesystem",
false},
+ {"read enables REST", nullptr, "filesystem", "rest", "rest", true},
+ {"read enables uppercase REST", nullptr, "hive", "REST", "REST", true},
+ {"table enables REST", "rest", nullptr, nullptr, "rest", true},
+ {"catalog overrides table REST", "rest", "filesystem", nullptr,
"filesystem", false},
+ {"missing metastore", nullptr, nullptr, nullptr, nullptr, false}};
+ const std::string expected_header =
+ "%7B%22database%22%3A%22db1%22%2C%22object%22%3A%22t1%22%7D";
+ for (const auto& test_case : cases) {
+ auto catalog = std::make_shared<MockVersionManagedCatalog>();
+ catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
+ std::map<std::string, std::string> catalog_options;
+ if (test_case.catalog_metastore) {
+ catalog_options[CatalogOptions::METASTORE] =
test_case.catalog_metastore;
+ }
+ catalog->SetOptions(catalog_options);
+ auto options = table_schema->Options();
+ if (test_case.table_metastore) {
+ options[CatalogOptions::METASTORE] = test_case.table_metastore;
+ }
+ std::map<std::string, std::string> read_options;
+ if (test_case.read_metastore) {
+ read_options[CatalogOptions::METASTORE] = test_case.read_metastore;
+ options[CatalogOptions::METASTORE] = test_case.read_metastore;
+ }
+ ReadContextBuilder builder(path);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
+ builder.WithCatalog(catalog, Identifier("db1",
"t1"))
+ .SetOptions(read_options)
+ .Finish());
+ ASSERT_OK_AND_ASSIGN(auto internal_context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+ auto dependency_context =
internal_context->CreateDependencyCatalogContext();
+ if (test_case.expected_metastore) {
+
ASSERT_EQ(dependency_context->options.at(CatalogOptions::METASTORE),
+ test_case.expected_metastore);
+ } else {
+
ASSERT_EQ(dependency_context->options.count(CatalogOptions::METASTORE), 0u);
+ }
+ if (test_case.generates_header) {
+
ASSERT_EQ(dependency_context->options.at("header.X-Paimon-Read-Via"),
expected_header);
+ } else {
+
ASSERT_EQ(dependency_context->options.count("header.X-Paimon-Read-Via"), 0u);
+ }
+ ASSERT_EQ(read_context->GetOptions(), read_options);
+ ASSERT_EQ(catalog->GetOptions(), catalog_options);
+ ASSERT_EQ(internal_context->GetCoreOptions().ToMap(), options);
+ }
+}
+
+TEST(InternalReadContext, TestDependencyReadHeaderNeedsAnIdentifier) {
+ std::string path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
+ SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ ReadContextBuilder builder(path);
+ ASSERT_OK_AND_ASSIGN(auto read_context,
+ builder.AddOption(CatalogOptions::METASTORE,
"REST").Finish());
+ ASSERT_FALSE(read_context->GetIdentifier());
+ auto options = table_schema->Options();
+ options[CatalogOptions::METASTORE] = "REST";
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);
+ auto dependency_context =
internal_context->CreateDependencyCatalogContext();
+ ASSERT_EQ(dependency_context->options.count("header.X-Paimon-Read-Via"),
0u);
+}
+
+TEST(InternalReadContext,
TestDependencyReadHeaderReportsTheBranchOfTheStartingTable) {
+ std::string path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
+ SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ auto catalog = std::make_shared<MockVersionManagedCatalog>();
+ catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
+ catalog->SetOptions({{CatalogOptions::METASTORE, "REST"}});
+ ReadContextBuilder builder(path);
+ ASSERT_OK_AND_ASSIGN(auto read_context,
+ builder.WithCatalog(catalog, Identifier("db1", "t1",
"dev")).Finish());
+ ASSERT_OK_AND_ASSIGN(auto internal_context,
+ InternalReadContext::Create(std::move(read_context),
table_schema,
+ table_schema->Options()));
+ auto dependency_context =
internal_context->CreateDependencyCatalogContext();
+ ASSERT_EQ(dependency_context->options.at("header.X-Paimon-Read-Via"),
+
"%7B%22database%22%3A%22db1%22%2C%22object%22%3A%22t1%24branch_dev%22%7D");
}
TEST(InternalReadContext, TestFileSystemSchemeMapReachesTheDependencyRead) {
@@ -116,6 +281,8 @@ TEST(InternalReadContext,
TestFileSystemSchemeMapReachesTheDependencyRead) {
InternalReadContext::Create(std::move(read_context),
table_schema,
table_schema->Options()));
ASSERT_EQ(internal_context->GetFileSystemSchemeToIdentifierMap(),
scheme_map);
+
ASSERT_EQ(internal_context->CreateDependencyCatalogContext()->fs_scheme_to_identifier_map,
+ scheme_map);
}
TEST(InternalReadContext, TestReadWithSpecifiedFieldId) {
diff --git a/src/paimon/core/operation/read_context.cpp
b/src/paimon/core/operation/read_context.cpp
index d731c4d3..bc65b466 100644
--- a/src/paimon/core/operation/read_context.cpp
+++ b/src/paimon/core/operation/read_context.cpp
@@ -25,64 +25,18 @@
#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<std::string, std::string>&
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);
- // `GetObjectName()` keeps any branch and system-table suffix, matching
the Java
- // `getObjectName()` this field mirrors. So a branch read reports the
branch.
- document.AddMember(rapidjson::StringRef("object"),
- rapidjson::Value(identifier.GetObjectName(),
allocator), allocator);
- // Compact, not the pretty form `RapidJsonUtil::ToJson` writes: the value
goes into a header.
- rapidjson::StringBuffer buffer;
- rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
- document.Accept(writer);
- return buffer.GetString();
-}
-
-} // namespace
-
ReadContext::ReadContext(
const std::string& path, const std::string& branch,
const std::vector<std::string>& read_field_names, const
std::vector<int32_t>& read_field_ids,
@@ -474,23 +428,6 @@ Result<std::unique_ptr<ReadContext>>
ReadContextBuilder::Finish() {
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<ReadContext>(
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 d9079e70..4fb5e00b 100644
--- a/src/paimon/core/operation/read_context_test.cpp
+++ b/src/paimon/core/operation/read_context_test.cpp
@@ -85,91 +85,31 @@ 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<MockVersionManagedCatalog>();
- catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
- 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<std::string, std::string>& 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()`.
+TEST(ReadContextTest, TestWithCatalogKeepsReadOptionsSeparate) {
auto catalog = std::make_shared<MockVersionManagedCatalog>();
catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
- catalog->SetOptions({{CatalogOptions::METASTORE, "rest"}});
+ catalog->SetOptions({{CatalogOptions::METASTORE, "rest"},
+ {CatalogOptions::URI, "http://catalog"},
+ {Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}});
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<MockVersionManagedCatalog>();
- catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
- 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;
- }
+ // Finishing a read neither imports catalog defaults nor generates a
dependency header.
+ ASSERT_TRUE(ctx->GetOptions().empty());
- // 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<MockVersionManagedCatalog>();
- catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
- ReadContextBuilder no_metastore_builder("table_root_path");
+ catalog->SetOptions({{CatalogOptions::METASTORE, "rest"},
+ {CatalogOptions::URI, "http://catalog"},
+ {"header.X-Paimon-Read-Via", "catalog-view"}});
+ ASSERT_OK_AND_ASSIGN(auto nested_ctx,
+ builder.WithCatalog(catalog, Identifier("db1",
"t1")).Finish());
+ ASSERT_TRUE(nested_ctx->GetOptions().empty());
+ const std::map<std::string, std::string> read_options = {
+ {Options::BLOB_VIEW_RESOLVE_ENABLED, "true"},
+ {"header.X-Paimon-Read-Via", "explicit-view"}};
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<MockVersionManagedCatalog>();
- catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
- 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");
+ auto explicit_ctx,
+ builder.WithCatalog(catalog, Identifier("db1",
"t1")).SetOptions(read_options).Finish());
+ ASSERT_EQ(explicit_ctx->GetOptions(), read_options);
}
TEST(ReadContextTest, TestDefaultValue) {
diff --git a/src/paimon/core/table/source/table_read_test.cpp
b/src/paimon/core/table/source/table_read_test.cpp
index 31e3f74f..ead31b17 100644
--- a/src/paimon/core/table/source/table_read_test.cpp
+++ b/src/paimon/core/table/source/table_read_test.cpp
@@ -26,8 +26,10 @@
#include <utility>
#include <vector>
+#include "arrow/type.h"
#include "gtest/gtest.h"
#include "paimon/catalog/table.h"
+#include "paimon/common/catalog/catalog_context.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/core/core_options.h"
#include "paimon/core/operation/abstract_split_read.h"
@@ -269,4 +271,37 @@ TEST(TableReadTest, TestMergeOptions) {
ASSERT_EQ(expected_options, core_options.ToMap());
}
+TEST(TableReadTest, TestDependencyReadWithoutCatalogUsesReadOptions) {
+ const std::string read_via_option = "header.X-Paimon-Read-Via";
+ const std::map<std::string, std::string> table_options = {
+ {Options::FILE_FORMAT, "orc"}, {Options::BUCKET, "-1"},
{read_via_option, "table-view"}};
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TableSchema> table_schema,
+ TableSchema::Create(0, arrow::schema({arrow::field("id",
arrow::int32())}),
+ /*partition_keys=*/{}, /*primary_keys=*/{},
table_options));
+ ASSERT_OK_AND_ASSIGN(std::string schema_json,
table_schema->ToJsonString());
+
+ for (bool explicit_header : {false, true}) {
+ ReadContextBuilder builder("warehouse/db.db/table");
+ builder.SetTableSchema(schema_json);
+ if (explicit_header) {
+ builder.AddOption(read_via_option, "read-view");
+ }
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context,
builder.Finish());
+ // Exercise the actual table/read merge before creating the dependency
context.
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read,
+ TableRead::Create(std::move(read_context)));
+ auto* append_only_table_read =
dynamic_cast<AppendOnlyTableRead*>(table_read.get());
+ ASSERT_NE(append_only_table_read, nullptr);
+ const auto& internal_context = append_only_table_read->context_;
+ auto dependency_context =
internal_context->CreateDependencyCatalogContext();
+
+ auto expected_options = table_options;
+ expected_options[read_via_option] = explicit_header ? "read-view" :
"table-view";
+ ASSERT_EQ(dependency_context->root_path, "warehouse");
+ ASSERT_EQ(dependency_context->options, expected_options);
+ ASSERT_EQ(internal_context->GetTableSchema()->Options(),
table_options);
+ }
+}
+
} // namespace paimon::test
diff --git a/test/inte/blob_table_inte_test.cpp
b/test/inte/blob_table_inte_test.cpp
index eabb28a7..4cbfda4d 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -200,20 +200,15 @@ class BlobTableInteTest : public testing::Test, public
::testing::WithParamInter
}
/// Creates the (f0 INT, view) blob-view table the dangling-reference
tests share: unaware
- /// bucket, data evolution on, and `view` resolved against
`upstream_warehouse`. Returns the
- /// table's fields, which the callers reuse to build their expected arrays.
- arrow::FieldVector CreateBlobViewTable(const std::string&
upstream_warehouse,
- bool deletion_vectors_enabled)
const {
+ /// bucket and data evolution on. Returns the table's fields, which the
callers reuse to
+ /// build their expected arrays.
+ arrow::FieldVector CreateBlobViewTable(bool deletion_vectors_enabled)
const {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> options = {
- {Options::FILE_FORMAT, GetParam()},
- {Options::BUCKET, "-1"},
- {Options::ROW_TRACKING_ENABLED, "true"},
- {Options::DATA_EVOLUTION_ENABLED, "true"},
- {Options::BLOB_VIEW_FIELD, "view"},
- {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, upstream_warehouse},
- {Options::FILE_SYSTEM, "local"}};
+ {Options::FILE_FORMAT, GetParam()}, {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM,
"local"}};
if (deletion_vectors_enabled) {
options.emplace(Options::DELETION_VECTORS_ENABLED, "true");
}
@@ -4013,13 +4008,9 @@ TEST_P(BlobTableInteTest,
TestBlobViewFieldWithUpstreamTable) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> 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, dir_->Str()},
- {Options::FILE_SYSTEM, "local"}};
+ {Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM,
"local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
@@ -4177,8 +4168,7 @@ 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 configures no explicit upstream warehouse; a default
read falls back to the
- // warehouse it lives in (dir_), where the copied upstream table sits.
+ // Both blob-view tables resolve the upstream table from the warehouse
they share.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> source_options = {
@@ -4188,15 +4178,12 @@ TEST_P(BlobTableInteTest, TestForwardBlobViewReference)
{
CreateTable(fields, /*partition_keys=*/{}, source_options);
std::string source_table_path = PathUtil::JoinPath(dir_->Str(),
"foo.db/bar");
- // The target table configures the upstream warehouse for resolving
forwarded references.
- std::map<std::string, std::string> target_options = source_options;
- target_options[Options::BLOB_VIEW_UPSTREAM_WAREHOUSE] = dir_->Str();
auto schema = arrow::schema(fields);
::ArrowSchema c_target_schema;
ASSERT_TRUE(arrow::ExportSchema(*schema, &c_target_schema).ok());
ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir_->Str(), {}));
ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar_forward"),
&c_target_schema,
- /*partition_keys=*/{}, /*primary_keys=*/{},
target_options,
+ /*partition_keys=*/{}, /*primary_keys=*/{},
source_options,
/*ignore_if_exists=*/false));
std::string target_table_path = PathUtil::JoinPath(dir_->Str(),
"foo.db/bar_forward");
@@ -4231,9 +4218,8 @@ TEST_P(BlobTableInteTest, TestForwardBlobViewReference) {
WriteArray(source_table_path, {},
schema->field_names(), {write_struct}));
ASSERT_OK(Commit(source_table_path, commit_msgs));
- // 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.
+ // A default read resolves the copied upstream table in the source table's
warehouse. The
+ // pass-through read below uses the dynamic option to preserve the raw
BlobViewStruct bytes.
ASSERT_OK_AND_ASSIGN(auto source_plan, ScanTable(source_table_path));
ASSERT_OK_AND_ASSIGN(
auto source_default_result,
@@ -4370,13 +4356,9 @@ TEST_P(BlobTableInteTest,
TestBlobViewFieldWithUpstreamDescriptorBlob) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> 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, dir_->Str()},
- {Options::FILE_SYSTEM, "local"}};
+ {Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM,
"local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
@@ -4475,13 +4457,9 @@ TEST_P(BlobTableInteTest,
TestBlobViewFieldWithMultipleUpstreamTables) {
BlobUtils::ToArrowField("view1", true),
BlobUtils::ToArrowField("view2", true)};
std::map<std::string, std::string> options = {
- {Options::FILE_FORMAT, file_format},
- {Options::BUCKET, "-1"},
- {Options::ROW_TRACKING_ENABLED, "true"},
- {Options::DATA_EVOLUTION_ENABLED, "true"},
- {Options::BLOB_VIEW_FIELD, "view1,view2"},
- {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, dir_->Str()},
- {Options::FILE_SYSTEM, "local"}};
+ {Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::BLOB_VIEW_FIELD, "view1,view2"}, {Options::FILE_SYSTEM,
"local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
@@ -4605,9 +4583,7 @@ TEST_P(BlobTableInteTest,
TestBlobViewFailsWhenUpstreamTableAbsent) {
if (GetParam() == "mosaic") {
return;
}
- auto upstream_dir = UniqueTestDirectory::Create("local");
- arrow::FieldVector fields =
- CreateBlobViewTable(upstream_dir->Str(),
/*deletion_vectors_enabled=*/false);
+ arrow::FieldVector fields =
CreateBlobViewTable(/*deletion_vectors_enabled=*/false);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
@@ -4623,9 +4599,7 @@ TEST_P(BlobTableInteTest,
TestBlobViewSkipsDanglingReferenceOfDeletedRow) {
if (GetParam() == "mosaic") {
return;
}
- auto upstream_dir = UniqueTestDirectory::Create("local");
- arrow::FieldVector fields =
- CreateBlobViewTable(upstream_dir->Str(),
/*deletion_vectors_enabled=*/true);
+ arrow::FieldVector fields =
CreateBlobViewTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
@@ -4666,9 +4640,7 @@ TEST_P(BlobTableInteTest,
TestBlobViewSkipsDanglingReferenceInEveryRowRangeGroup
if (GetParam() == "mosaic") {
return;
}
- auto upstream_dir = UniqueTestDirectory::Create("local");
- arrow::FieldVector fields =
- CreateBlobViewTable(upstream_dir->Str(),
/*deletion_vectors_enabled=*/true);
+ arrow::FieldVector fields =
CreateBlobViewTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
@@ -4710,9 +4682,7 @@ TEST_P(BlobTableInteTest,
TestBlobViewPreReadHonorsRowRangesNotPredicate) {
if (GetParam() == "mosaic") {
return;
}
- auto upstream_dir = UniqueTestDirectory::Create("local");
- arrow::FieldVector fields =
- CreateBlobViewTable(upstream_dir->Str(),
/*deletion_vectors_enabled=*/false);
+ arrow::FieldVector fields =
CreateBlobViewTable(/*deletion_vectors_enabled=*/false);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
