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 7c0af00e feat(blob): read blob-view dependency tables through a
catalog (#427)
7c0af00e is described below
commit 7c0af00eeac8ce5e543e1ccd6a6d08cfa6374ca7
Author: Yonghao Fang <[email protected]>
AuthorDate: Tue Oct 13 15:36:20 2026 +0800
feat(blob): read blob-view dependency tables through a catalog (#427)
---
include/paimon/defs.h | 8 +-
src/paimon/common/catalog/catalog_context.h | 21 +++-
.../core/operation/data_evolution_split_read.cpp | 15 ++-
src/paimon/core/operation/internal_read_context.h | 5 +
.../core/operation/internal_read_context_test.cpp | 40 ++++++
src/paimon/core/operation/read_context.cpp | 74 ++++++++++-
src/paimon/core/operation/read_context_test.cpp | 113 +++++++++++++++++
src/paimon/core/utils/blob_view_lookup.cpp | 40 ++----
src/paimon/core/utils/blob_view_lookup.h | 4 +-
src/paimon/rest/rest_api.h | 5 +
src/paimon/rest/rest_catalog.h | 2 +
src/paimon/rest/rest_credential_provider.cpp | 9 +-
src/paimon/rest/rest_credential_provider.h | 7 +-
src/paimon/testing/mock/mock_catalog.h | 5 +
test/inte/blob_table_inte_test.cpp | 140 +++------------------
15 files changed, 312 insertions(+), 176 deletions(-)
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 7b473240..8c88c3de 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -568,9 +568,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<BLOB> 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 fecc3e01..7a9f8b87 100644
--- a/src/paimon/common/catalog/catalog_context.h
+++ b/src/paimon/common/catalog/catalog_context.h
@@ -21,18 +21,29 @@
#include <map>
#include <string>
-#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<std::string, std::string>& _options,
- const std::shared_ptr<FileSystem>& _file_system)
- : root_path(_root_path), options(_options), file_system(_file_system)
{}
+ const std::map<std::string, std::string>&
_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<std::string, std::string> options;
- std::shared_ptr<FileSystem> 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<std::string, std::string> 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 5c26d341..7491ad3e 100644
--- a/src/paimon/core/operation/data_evolution_split_read.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read.cpp
@@ -280,17 +280,26 @@ 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) {
- 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<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(), 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 e3c2326f..8e34a85b 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<RealtimeContext> 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<std::string, std::string>&
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 9e0dc5fb..086d56d1 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<LocalFileSystem>(),
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<std::string, std::string> 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<std::string, std::string> 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<LocalFileSystem>(),
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 5cbbb5db..1779d730 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 <map>
+#include <string>
#include <utility>
#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);
+ // 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<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,
@@ -377,9 +427,9 @@ Result<std::unique_ptr<ReadContext>>
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<std::unique_ptr<ReadContext>>
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<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 b820d245..176df8b0 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<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()`.
+ auto catalog = std::make_shared<MockVersionManagedCatalog>();
+ catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
+ 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<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;
+ }
+
+ // 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");
+ 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");
+}
+
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<MockVersionManagedCatalog>();
+ catalog->SetTableFileSystem(std::make_shared<MockFileSystem>());
+
+ // 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<MockFileSystem>();
diff --git a/src/paimon/core/utils/blob_view_lookup.cpp
b/src/paimon/core/utils/blob_view_lookup.cpp
index 99e7dc83..30999fe6 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::DescriptorMapping>
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,
+ 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<ScanContext> scan_context,
scan_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableScan> table_scan,
TableScan::Create(std::move(scan_context)));
@@ -169,7 +171,7 @@ Result<BlobViewLookup::DescriptorMapping>
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<ReadContext> read_context,
read_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableRead> table_read,
TableRead::Create(std::move(read_context)));
@@ -182,34 +184,6 @@ Result<BlobViewLookup::DescriptorMapping>
BlobViewLookup::LoadTableDescriptorChu
return mapping;
}
-Result<std::string> BlobViewLookup::GetTableLocation(
- const std::shared_ptr<CatalogContext>& catalog_context, const Identifier&
identifier) {
- auto file_system = catalog_context->file_system;
- PAIMON_ASSIGN_OR_RAISE(
- std::unique_ptr<Catalog> 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<int32_t>&
field_ids,
const
std::shared_ptr<MemoryPool>& pool,
diff --git a/src/paimon/core/utils/blob_view_lookup.h
b/src/paimon/core/utils/blob_view_lookup.h
index 169d9f83..336863db 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<std::vector<Range>> SplitRowRanges(const
std::vector<Range>& row_ranges,
int64_t
target_rows_per_task);
-
- static Result<std::string> GetTableLocation(
- const std::shared_ptr<CatalogContext>& catalog_context, const
Identifier& identifier);
};
} // namespace paimon
diff --git a/src/paimon/rest/rest_api.h b/src/paimon/rest/rest_api.h
index 6efdc180..722fc525 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 c9089d43..2659b67c 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<std::string>&
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<std::unique_ptr<TableSchema>> ToTableSchema(
const GetTableResponse& response, const std::optional<std::string>&
branch);
diff --git a/src/paimon/rest/rest_credential_provider.cpp
b/src/paimon/rest/rest_credential_provider.cpp
index fda2de9c..5e93b84d 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<RestApi>&
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<std::string, std::string>
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<int64_t>(response.GetExpiresAtMillis()));
diff --git a/src/paimon/rest/rest_credential_provider.h
b/src/paimon/rest/rest_credential_provider.h
index 3860ee67..c46f38d8 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<std::chrono::system_clock::time_point()>;
+ using TokenLoader = std::function<Result<GetTableTokenResponse>()>;
/// @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<RestApi>& 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<RestApi> api_;
+ TokenLoader token_loader_;
Identifier identifier_;
Clock clock_;
std::shared_ptr<Logger> logger_;
diff --git a/src/paimon/testing/mock/mock_catalog.h
b/src/paimon/testing/mock/mock_catalog.h
index 88b9b914..b9954e31 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<std::string, std::string>& GetOptions() const override {
return options_;
}
+
+ /// Sets the catalog-level options `GetOptions()` reports, e.g.
`metastore`.
+ void SetOptions(const std::map<std::string, std::string>& options) {
+ options_ = options;
+ }
Result<std::vector<SnapshotInfo>> 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 0d566595..80ea97e5 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<std::string, std::string> 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<std::string, std::string> 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: <warehouse>/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::StructArray>(
-
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<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, 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<Bytes> serialized,
view_struct.Serialize(pool_));
- ASSERT_TRUE(
- view_builder
- .Append(reinterpret_cast<const uint8_t*>(serialized->data()),
serialized->size())
- .ok());
- }
- std::shared_ptr<arrow::Array> 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>(
- arrow::StructArray::Make(arrow::ArrayVector({write_f0_array,
write_view_array}),
- std::vector<std::string>({"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<arrow::StructArray>(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::StructArray>(
- 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") {
