From b125cf81e0a73e361411028cd4cb5a71ba03f888 Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Wed, 8 Jul 2026 16:08:11 +0800 Subject: [PATCH 1/2] feat: add read-optimized system table --- docs/source/user_guide.rst | 1 + docs/source/user_guide/system_tables.rst | 60 +++++ src/paimon/CMakeLists.txt | 1 + .../table/source/data_table_batch_scan.cpp | 11 +- .../core/table/source/data_table_batch_scan.h | 2 +- .../source/read_optimized_scan_options.h | 24 ++ src/paimon/core/table/source/table_scan.cpp | 17 +- .../core/table/source/table_scan_test.cpp | 14 ++ .../system/read_optimized_system_table.cpp | 123 +++++++++++ .../system/read_optimized_system_table.h | 51 +++++ src/paimon/core/table/system/system_table.cpp | 9 + .../core/table/system/system_table_test.cpp | 34 +++ src/paimon/testing/utils/test_helper.h | 9 +- test/inte/read_inte_test.cpp | 208 +++++++++++++++++- 14 files changed, 552 insertions(+), 12 deletions(-) create mode 100644 docs/source/user_guide/system_tables.rst create mode 100644 src/paimon/core/table/source/read_optimized_scan_options.h create mode 100644 src/paimon/core/table/system/read_optimized_system_table.cpp create mode 100644 src/paimon/core/table/system/read_optimized_system_table.h diff --git a/docs/source/user_guide.rst b/docs/source/user_guide.rst index e9d109cea..8234e2a7a 100644 --- a/docs/source/user_guide.rst +++ b/docs/source/user_guide.rst @@ -30,6 +30,7 @@ User Guide user_guide/data_types user_guide/primary_key_table user_guide/append_only_table + user_guide/system_tables user_guide/write user_guide/commit user_guide/compaction diff --git a/docs/source/user_guide/system_tables.rst b/docs/source/user_guide/system_tables.rst new file mode 100644 index 000000000..9d2d3a1e2 --- /dev/null +++ b/docs/source/user_guide/system_tables.rst @@ -0,0 +1,60 @@ +.. Copyright 2026-present Alibaba Inc. + +.. Licensed under the Apache License, Version 2.0 (the "License"); +.. you may not use this file except in compliance with the License. +.. You may obtain a copy of the License at + +.. http://www.apache.org/licenses/LICENSE-2.0 + +.. Unless required by applicable law or agreed to in writing, software +.. distributed under the License is distributed on an "AS IS" BASIS, +.. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +.. See the License for the specific language governing permissions and +.. limitations under the License. + +System Tables +============= + +Paimon C++ supports reading system tables by appending a system table suffix to +the data table path. For example, ``/warehouse/db.db/orders$snapshots`` reads +the snapshots system table for ``orders``. + +Branch-qualified paths are also supported. For example, +``/warehouse/db.db/orders$branch_audit$files`` reads the files system table from +the ``audit`` branch. + +Read-Optimized System Table +--------------------------- + +The read-optimized system table is addressed by the ``$ro`` suffix: + +.. code-block:: text + + /warehouse/db.db/orders$ro + +For primary-key tables, ``$ro`` only plans data files from the highest LSM +level, which is the level produced by full compaction. This avoids merging data +from multiple LSM levels during query planning and reading. The tradeoff is that +the result may lag behind the latest committed data until a full compaction +publishes the newest records into the highest level. + +For primary-key tables, ``$ro`` also enables value-stats filtering, so file-level +pruning of the reader predicate can be more aggressive than the base table. + +For append-only tables, ``$ro`` has the same read behavior as the base table, +including streaming reads. + +Limitations +~~~~~~~~~~~ + +- Primary-key ``$ro`` scans are batch-only. Streaming scans are not supported. +- Freshness depends on full compaction frequency. +- Primary-key tables in bucket-unaware mode are not supported by the current C++ + scan path. + +Typical Usage +~~~~~~~~~~~~~ + +Use ``$ro`` for OLAP or batch workloads that can tolerate stale results and +prefer reading compacted files directly. Use the base table path when the query +must see the latest committed data. diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index b044badea..6dc95f208 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -343,6 +343,7 @@ set(PAIMON_CORE_SRCS core/table/system/binlog_system_table.cpp core/table/system/in_memory_system_table.cpp core/table/system/metadata_system_tables.cpp + core/table/system/read_optimized_system_table.cpp core/table/system/system_table.cpp core/table/system/system_table_scan.cpp core/table/system/system_table_schema.cpp diff --git a/src/paimon/core/table/source/data_table_batch_scan.cpp b/src/paimon/core/table/source/data_table_batch_scan.cpp index 0a441f63b..54367745c 100644 --- a/src/paimon/core/table/source/data_table_batch_scan.cpp +++ b/src/paimon/core/table/source/data_table_batch_scan.cpp @@ -33,10 +33,15 @@ class DataSplit; DataTableBatchScan::DataTableBatchScan(bool pk_table, const CoreOptions& core_options, const std::shared_ptr& snapshot_reader, - std::optional push_down_limit) + bool read_optimized, std::optional push_down_limit) : AbstractTableScan(core_options, snapshot_reader), push_down_limit_(push_down_limit) { - if (pk_table && (core_options.DeletionVectorsEnabled() || - core_options.GetMergeEngine() == MergeEngine::FIRST_ROW)) { + if (pk_table && read_optimized) { + int32_t top_level = core_options.GetNumLevels() - 1; + snapshot_reader_->WithLevelFilter( + [top_level](int32_t level) -> bool { return level == top_level; }); + snapshot_reader_->EnableValueFilter(); + } else if (pk_table && (core_options.DeletionVectorsEnabled() || + core_options.GetMergeEngine() == MergeEngine::FIRST_ROW)) { auto level_filter = [](int32_t level) -> bool { return level > 0; }; snapshot_reader_->WithLevelFilter(level_filter); snapshot_reader_->EnableValueFilter(); diff --git a/src/paimon/core/table/source/data_table_batch_scan.h b/src/paimon/core/table/source/data_table_batch_scan.h index d5a1d44e6..582d685a7 100644 --- a/src/paimon/core/table/source/data_table_batch_scan.h +++ b/src/paimon/core/table/source/data_table_batch_scan.h @@ -32,7 +32,7 @@ class SnapshotReader; class DataTableBatchScan : public AbstractTableScan { public: DataTableBatchScan(bool pk_table, const CoreOptions& core_options, - const std::shared_ptr& snapshot_reader, + const std::shared_ptr& snapshot_reader, bool read_optimized, std::optional push_down_limit); Result> CreatePlan() override; diff --git a/src/paimon/core/table/source/read_optimized_scan_options.h b/src/paimon/core/table/source/read_optimized_scan_options.h new file mode 100644 index 000000000..fd07f4f7a --- /dev/null +++ b/src/paimon/core/table/source/read_optimized_scan_options.h @@ -0,0 +1,24 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +namespace paimon { + +/// Internal scan option used by `T$ro` to request top-level-only planning. +inline constexpr char kReadOptimizedScanOption[] = "__paimon.internal.read-optimized"; + +} // namespace paimon diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp index cff55bd21..5cf915da9 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/src/paimon/core/table/source/table_scan.cpp @@ -47,6 +47,7 @@ #include "paimon/core/table/source/data_table_batch_scan.h" #include "paimon/core/table/source/data_table_stream_scan.h" #include "paimon/core/table/source/merge_tree_split_generator.h" +#include "paimon/core/table/source/read_optimized_scan_options.h" #include "paimon/core/table/source/snapshot/snapshot_reader.h" #include "paimon/core/table/source/split_generator.h" #include "paimon/core/table/system/system_table.h" @@ -225,9 +226,15 @@ Result> NewDataTableScan(const std::shared_ptrGetOptions().find(kReadOptimizedScanOption); + const bool read_optimized = + read_optimized_iter != context->GetOptions().end() && read_optimized_iter->second == "true"; // merge options auto options = table_schema->Options(); for (const auto& [key, value] : context->GetOptions()) { + if (key == kReadOptimizedScanOption) { + continue; + } options[key] = value; } PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, @@ -280,12 +287,16 @@ Result> NewDataTableScan(const std::shared_ptrGetMemoryPool())); auto snapshot_reader = std::make_shared( file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler)); + const bool pk_table = !table_schema->PrimaryKeys().empty(); + if (read_optimized && pk_table && context->IsStreamingMode()) { + return Status::NotImplemented( + "read-optimized system table does not support streaming scan for primary key table"); + } if (context->IsStreamingMode()) { return std::make_unique(core_options, snapshot_reader); } - auto batch_scan = - std::make_unique(/*pk_table=*/!table_schema->PrimaryKeys().empty(), - core_options, snapshot_reader, context->GetLimit()); + auto batch_scan = std::make_unique( + /*pk_table=*/pk_table, core_options, snapshot_reader, read_optimized, context->GetLimit()); if (!core_options.DataEvolutionEnabled()) { return batch_scan; } diff --git a/src/paimon/core/table/source/table_scan_test.cpp b/src/paimon/core/table/source/table_scan_test.cpp index 33c5ddd7d..777c8780f 100644 --- a/src/paimon/core/table/source/table_scan_test.cpp +++ b/src/paimon/core/table/source/table_scan_test.cpp @@ -16,6 +16,7 @@ #include "paimon/table/source/table_scan.h" +#include #include #include #include @@ -59,4 +60,17 @@ TEST(TableScanTest, TestPkSchemaEvolutionScan) { ASSERT_FALSE(plan->Splits().empty()); } +TEST(TableScanTest, TestReadOptimizedPrimaryKeyStreamingScanUnsupported) { + std::string path = paimon::test::GetDataDir() + + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table$ro"; + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithStreamingMode(true); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + + ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(context)), + "read-optimized system table does not support streaming scan for primary " + "key table"); +} + } // namespace paimon::test diff --git a/src/paimon/core/table/system/read_optimized_system_table.cpp b/src/paimon/core/table/system/read_optimized_system_table.cpp new file mode 100644 index 000000000..c4c633707 --- /dev/null +++ b/src/paimon/core/table/system/read_optimized_system_table.cpp @@ -0,0 +1,123 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/core/table/system/read_optimized_system_table.h" + +#include +#include +#include + +#include "paimon/common/types/data_field.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/source/read_optimized_scan_options.h" +#include "paimon/defs.h" +#include "paimon/read_context.h" +#include "paimon/scan_context.h" +#include "paimon/table/source/table_read.h" +#include "paimon/table/source/table_scan.h" + +namespace paimon { + +ReadOptimizedSystemTable::ReadOptimizedSystemTable(std::string table_path, + std::shared_ptr table_schema, + std::map options) + : table_path_(std::move(table_path)), + table_schema_(std::move(table_schema)), + options_(std::move(options)) {} + +std::string ReadOptimizedSystemTable::Name() const { + return kName; +} + +Result> ReadOptimizedSystemTable::ArrowSchema() const { + return DataField::ConvertDataFieldsToArrowSchema(table_schema_->Fields()); +} + +std::map ReadOptimizedSystemTable::ReadOptimizedOptions() const { + auto options = options_; + options[kReadOptimizedScanOption] = "true"; + return options; +} + +Result> ReadOptimizedSystemTable::NewScan( + const std::shared_ptr& context) const { + auto options = ReadOptimizedOptions(); + ScanContextBuilder builder(table_path_); + builder.SetOptions(options) + .WithStreamingMode(context->IsStreamingMode()) + .WithMemoryPool(context->GetMemoryPool()) + .WithExecutor(context->GetExecutor()) + .WithFileSystem(context->GetSpecificFileSystem()) + .WithCache(context->GetCache()); + if (context->GetLimit().has_value()) { + builder.SetLimit(context->GetLimit().value()); + } + if (context->GetScanFilters()) { + if (context->GetScanFilters()->GetBucketFilter().has_value()) { + builder.SetBucketFilter(context->GetScanFilters()->GetBucketFilter().value()); + } + builder.SetPartitionFilter(context->GetScanFilters()->GetPartitionFilters()); + builder.SetPredicate(context->GetScanFilters()->GetPredicate()); + } + if (context->GetGlobalIndexResult()) { + builder.SetGlobalIndexResult(context->GetGlobalIndexResult()); + } + if (context->GetSpecificTableSchema().has_value()) { + builder.SetTableSchema(context->GetSpecificTableSchema().value()); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr base_context, builder.Finish()); + return TableScan::Create(std::move(base_context)); +} + +Result> ReadOptimizedSystemTable::NewRead( + const std::shared_ptr& context) const { + auto options = options_; + std::string branch = context->GetBranch(); + // SystemTableLoader injects Options::BRANCH when parsing paths such as `T$branch_dev$ro`. + auto branch_iter = options.find(Options::BRANCH); + if (branch_iter != options.end()) { + branch = branch_iter->second; + } + ReadContextBuilder builder(table_path_); + builder.SetOptions(options) + .WithBranch(branch) + .SetPredicate(context->GetPredicate()) + .EnablePredicateFilter(context->EnablePredicateFilter()) + .EnablePrefetch(context->EnablePrefetch()) + .SetPrefetchBatchCount(context->GetPrefetchBatchCount()) + .SetPrefetchMaxParallelNum(context->GetPrefetchMaxParallelNum()) + .EnableMultiThreadRowToBatch(context->EnableMultiThreadRowToBatch()) + .SetRowToBatchThreadNumber(context->GetRowToBatchThreadNumber()) + .WithMemoryPool(context->GetMemoryPool()) + .WithExecutor(context->GetExecutor()) + .WithFileSystem(context->GetSpecificFileSystem()) + .WithFileSystemSchemeToIdentifierMap(context->GetFileSystemSchemeToIdentifierMap()) + .SetPrefetchCacheMode(context->GetPrefetchCacheMode()) + .WithCacheConfig(context->GetCacheConfig()) + .WithCache(context->GetCache()); + if (!context->GetReadFieldIds().empty()) { + builder.SetReadFieldIds(context->GetReadFieldIds()); + } else { + builder.SetReadFieldNames(context->GetReadFieldNames()); + } + if (context->GetSpecificTableSchema().has_value()) { + builder.SetTableSchema(context->GetSpecificTableSchema().value()); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr base_context, builder.Finish()); + return TableRead::Create(std::move(base_context)); +} + +} // namespace paimon diff --git a/src/paimon/core/table/system/read_optimized_system_table.h b/src/paimon/core/table/system/read_optimized_system_table.h new file mode 100644 index 000000000..4f2e9d2b3 --- /dev/null +++ b/src/paimon/core/table/system/read_optimized_system_table.h @@ -0,0 +1,51 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include + +#include "paimon/core/table/system/system_table.h" + +namespace paimon { +class TableSchema; + +/// System table for `T$ro`, exposing read-optimized data. +class ReadOptimizedSystemTable : public SystemTable { + public: + static constexpr const char* kName = "ro"; + + ReadOptimizedSystemTable(std::string table_path, std::shared_ptr table_schema, + std::map options); + + std::string Name() const override; + Result> ArrowSchema() const override; + Result> NewScan( + const std::shared_ptr& context) const override; + Result> NewRead( + const std::shared_ptr& context) const override; + + private: + std::map ReadOptimizedOptions() const; + + std::string table_path_; + std::shared_ptr table_schema_; + std::map options_; +}; + +} // namespace paimon diff --git a/src/paimon/core/table/system/system_table.cpp b/src/paimon/core/table/system/system_table.cpp index 87e1c3b81..f3a0a7f6b 100644 --- a/src/paimon/core/table/system/system_table.cpp +++ b/src/paimon/core/table/system/system_table.cpp @@ -31,6 +31,7 @@ #include "paimon/core/table/system/audit_log_system_table.h" #include "paimon/core/table/system/binlog_system_table.h" #include "paimon/core/table/system/metadata_system_tables.h" +#include "paimon/core/table/system/read_optimized_system_table.h" #include "paimon/core/utils/branch_manager.h" #include "paimon/status.h" @@ -86,6 +87,14 @@ const std::vector& SystemTableRegistry() { return std::make_shared( fs, table_path, table_schema, MergeOptions(table_schema, dynamic_options)); }}, + {ReadOptimizedSystemTable::kName, + [](const std::shared_ptr& /*fs*/, const std::string& table_path, + const std::shared_ptr& table_schema, + const std::map& dynamic_options) + -> Result> { + return std::make_shared( + table_path, table_schema, MergeOptions(table_schema, dynamic_options)); + }}, {SnapshotsSystemTable::kName, [](const std::shared_ptr& fs, const std::string& table_path, const std::shared_ptr& table_schema, diff --git a/src/paimon/core/table/system/system_table_test.cpp b/src/paimon/core/table/system/system_table_test.cpp index af316935a..ec61439e6 100644 --- a/src/paimon/core/table/system/system_table_test.cpp +++ b/src/paimon/core/table/system/system_table_test.cpp @@ -14,15 +14,20 @@ * limitations under the License. */ +#include "paimon/core/table/system/system_table.h" + #include #include +#include #include +#include #include "arrow/api.h" #include "gtest/gtest.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/system/audit_log_system_table.h" #include "paimon/core/table/system/binlog_system_table.h" +#include "paimon/core/table/system/read_optimized_system_table.h" #include "paimon/defs.h" #include "paimon/fs/file_system.h" #include "paimon/result.h" @@ -62,4 +67,33 @@ TEST(SystemTableTest, TestChangelogArrowSchemaReturnsInvalidOptions) { "Invalid Config [table-read.sequence-number.enabled: invalid]"); } +TEST(SystemTableTest, TestReadOptimizedSystemTableRegistration) { + ASSERT_TRUE(SystemTableLoader::IsSupported(ReadOptimizedSystemTable::kName)); + + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, + CreateTableSchemaForTest(options)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr system_table, + SystemTableLoader::Load(ReadOptimizedSystemTable::kName, /*fs=*/nullptr, + "/tmp/table", table_schema, + /*dynamic_options=*/{})); + ASSERT_EQ(system_table->Name(), ReadOptimizedSystemTable::kName); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr arrow_schema, system_table->ArrowSchema()); + ASSERT_EQ(arrow_schema->field_names(), (std::vector{"pk", "v"})); + ASSERT_EQ(arrow_schema->field(0)->type()->id(), arrow::Type::STRING); + ASSERT_EQ(arrow_schema->field(1)->type()->id(), arrow::Type::INT32); +} + +TEST(SystemTableTest, TestReadOptimizedSystemTablePathParsing) { + ASSERT_OK_AND_ASSIGN(std::optional parsed, + SystemTableLoader::TryParsePath("/tmp/db.db/t$branch_audit$ro")); + ASSERT_TRUE(parsed.has_value()); + ASSERT_EQ(parsed->table_path, "/tmp/db.db/t"); + ASSERT_TRUE(parsed->branch.has_value()); + ASSERT_EQ(parsed->branch.value(), "audit"); + ASSERT_EQ(parsed->system_table_name, ReadOptimizedSystemTable::kName); +} + } // namespace paimon::test diff --git a/src/paimon/testing/utils/test_helper.h b/src/paimon/testing/utils/test_helper.h index 832055a51..d3348a5c7 100644 --- a/src/paimon/testing/utils/test_helper.h +++ b/src/paimon/testing/utils/test_helper.h @@ -57,7 +57,7 @@ class TestHelper { const std::vector& partition_keys, const std::vector& primary_keys, const std::map& options, bool is_streaming_mode, - bool ignore_if_exists = false) { + bool ignore_if_exists = false, const std::string& temp_directory = "") { // only for test && only check the key auto new_options = options; new_options["enable-object-store-catalog-in-inte-test"] = ""; @@ -70,12 +70,12 @@ class TestHelper { partition_keys, primary_keys, new_options, ignore_if_exists)); std::string table_path = PathUtil::JoinPath(root_path, "foo.db/bar"); - return Create(table_path, new_options, is_streaming_mode); + return Create(table_path, new_options, is_streaming_mode, temp_directory); } static Result> Create( const std::string& table_path, const std::map& options, - bool is_streaming_mode) { + bool is_streaming_mode, const std::string& temp_directory = "") { std::string file_system_identifier = "local"; auto fs_iter = options.find(Options::FILE_SYSTEM); if (fs_iter != options.end()) { @@ -85,6 +85,9 @@ class TestHelper { FileSystemFactory::Get(file_system_identifier, table_path, options)); std::string commit_user = "commit_user"; WriteContextBuilder context_builder(table_path, commit_user); + if (!temp_directory.empty()) { + context_builder.WithTempDirectory(temp_directory); + } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr write_context, context_builder.SetOptions(options) .WithStreamingMode(is_streaming_mode) diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index a267bfca2..6298130dd 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -269,9 +269,10 @@ std::vector StructFieldNames(const std::shared_ptr ReadSystemTable(const std::string& system_table_path, - const std::map& options) { + const std::map& options, + bool streaming_mode = false) { ScanContextBuilder scan_context_builder(system_table_path); - scan_context_builder.SetOptions(options); + scan_context_builder.SetOptions(options).WithStreamingMode(streaming_mode); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan_context, scan_context_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_scan, @@ -291,6 +292,17 @@ Result ReadSystemTable(const std::string& system_table_pa return SystemTableReadResult(std::move(batch_reader), result); } +Status WriteAndFullCompact(TestHelper* helper, std::unique_ptr&& batch, + int64_t commit_identifier) { + PAIMON_RETURN_NOT_OK(helper->write_->Write(std::move(batch))); + PAIMON_RETURN_NOT_OK( + helper->write_->Compact(/*partition=*/{}, /*bucket=*/0, /*full_compaction=*/true)); + PAIMON_ASSIGN_OR_RAISE( + std::vector> commit_messages, + helper->write_->PrepareCommit(/*wait_compaction=*/true, commit_identifier)); + return helper->commit_->Commit(commit_messages, commit_identifier); +} + void AssertStructArrayEqualsJson(const std::shared_ptr& actual, const std::string& expected_json) { ASSERT_TRUE(actual); @@ -783,6 +795,198 @@ TEST(SystemTableReadInteTest, TestReadMetadataSystemTables) { ASSERT_FALSE(creation_time_array->IsNull(0)); } +TEST(SystemTableReadInteTest, TestReadOptimizedSystemTable) { + arrow::FieldVector fields = { + arrow::field("k", arrow::int32()), + arrow::field("v", arrow::int32()), + }; + auto schema = arrow::schema(fields); + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "parquet"}, + {Options::MANIFEST_FORMAT, "avro"}, + {Options::BUCKET, "1"}, + {Options::NUM_LEVELS, "3"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN(std::unique_ptr helper, + TestHelper::Create(dir->Str(), schema, + /*partition_keys=*/{}, + /*primary_keys=*/{"k"}, options, + /*is_streaming_mode=*/true)); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + auto row_type = arrow::struct_(fields); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_1, + TestHelper::MakeRecordBatch(row_type, R"([[1, 10], [2, 20]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(WriteAndFullCompact(helper.get(), std::move(batch_1), /*commit_identifier=*/0)); + + ASSERT_OK_AND_ASSIGN(SystemTableReadResult compacted_result, + ReadSystemTable(table_path + "$ro", options)); + std::shared_ptr expected_type = + arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), + arrow::field("k", arrow::int32()), arrow::field("v", arrow::int32())}); + std::shared_ptr expected_compacted; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 1, 10], [0, 2, 20]])"}, &expected_compacted) + .ok()); + ASSERT_TRUE(compacted_result.array->Equals(expected_compacted)) + << compacted_result.array->ToString(); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_2, + TestHelper::MakeRecordBatch(row_type, R"([[1, 11], [3, 30]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch_2), /*commit_identifier=*/1, + /*expected_commit_messages=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(SystemTableReadResult stale_result, + ReadSystemTable(table_path + "$ro", options)); + ASSERT_TRUE(stale_result.array->Equals(expected_compacted)) << stale_result.array->ToString(); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_3, + TestHelper::MakeRecordBatch(row_type, R"([[2, 21], [3, 31]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(WriteAndFullCompact(helper.get(), std::move(batch_3), /*commit_identifier=*/2)); + ASSERT_OK_AND_ASSIGN(SystemTableReadResult refreshed_result, + ReadSystemTable(table_path + "$ro", options)); + std::shared_ptr expected_refreshed; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 1, 11], [0, 2, 21], [0, 3, 31]])"}, &expected_refreshed) + .ok()); + ASSERT_TRUE(refreshed_result.array->Equals(expected_refreshed)) + << refreshed_result.array->ToString(); +} + +TEST(SystemTableReadInteTest, TestReadOptimizedAppendOnlySystemTableWithStreamingScan) { + arrow::FieldVector fields = { + arrow::field("k", arrow::int32()), + arrow::field("v", arrow::int32()), + }; + auto schema = arrow::schema(fields); + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "parquet"}, + {Options::MANIFEST_FORMAT, "avro"}, + {Options::BUCKET, "1"}, + {Options::BUCKET_KEY, "k"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN(std::unique_ptr helper, + TestHelper::Create(dir->Str(), schema, + /*partition_keys=*/{}, + /*primary_keys=*/{}, options, + /*is_streaming_mode=*/true)); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + + ASSERT_OK_AND_ASSIGN( + std::unique_ptr batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([[1, 10], [2, 20]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); + + ASSERT_OK_AND_ASSIGN(SystemTableReadResult result, + ReadSystemTable(table_path + "$ro", options, /*streaming_mode=*/true)); + std::shared_ptr expected_type = + arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), + arrow::field("k", arrow::int32()), arrow::field("v", arrow::int32())}); + std::shared_ptr expected; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 1, 10], [0, 2, 20]])"}, &expected) + .ok()); + ASSERT_TRUE(result.array->Equals(expected)) << result.array->ToString(); +} + +TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableWithBranch) { + arrow::FieldVector fields = { + arrow::field("k", arrow::int32()), + arrow::field("v", arrow::int32()), + }; + auto schema = arrow::schema(fields); + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "parquet"}, + {Options::MANIFEST_FORMAT, "avro"}, + {Options::BUCKET, "1"}, + {Options::NUM_LEVELS, "3"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN(std::unique_ptr helper, + TestHelper::Create(dir->Str(), schema, + /*partition_keys=*/{}, + /*primary_keys=*/{"k"}, options, + /*is_streaming_mode=*/true)); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + auto row_type = arrow::struct_(fields); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr branch_batch, + TestHelper::MakeRecordBatch(row_type, R"([[1, 10], [2, 20]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(WriteAndFullCompact(helper.get(), std::move(branch_batch), + /*commit_identifier=*/0)); + + std::string branch_path = PathUtil::JoinPath(table_path, "branch/branch-rt"); + std::filesystem::create_directories(branch_path); + ASSERT_TRUE(TestUtil::CopyDirectory(PathUtil::JoinPath(table_path, "schema"), + PathUtil::JoinPath(branch_path, "schema"))); + ASSERT_TRUE(TestUtil::CopyDirectory(PathUtil::JoinPath(table_path, "snapshot"), + PathUtil::JoinPath(branch_path, "snapshot"))); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr main_batch, + TestHelper::MakeRecordBatch(row_type, R"([[1, 11], [3, 30]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(WriteAndFullCompact(helper.get(), std::move(main_batch), + /*commit_identifier=*/1)); + + ASSERT_OK_AND_ASSIGN(SystemTableReadResult result, + ReadSystemTable(table_path + "$branch_rt$ro", options)); + std::shared_ptr expected_type = + arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), + arrow::field("k", arrow::int32()), arrow::field("v", arrow::int32())}); + std::shared_ptr expected; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 1, 10], [0, 2, 20]])"}, &expected) + .ok()); + ASSERT_TRUE(result.array->Equals(expected)) << result.array->ToString(); +} + +TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableWithFirstRowMergeEngine) { + arrow::FieldVector fields = { + arrow::field("k", arrow::int32()), + arrow::field("v", arrow::int32()), + }; + auto schema = arrow::schema(fields); + std::map options = { + {Options::FILE_SYSTEM, "local"}, {Options::FILE_FORMAT, "parquet"}, + {Options::MANIFEST_FORMAT, "avro"}, {Options::BUCKET, "1"}, + {Options::BUCKET_KEY, "k"}, {Options::NUM_LEVELS, "5"}, + {Options::MERGE_ENGINE, "first-row"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr helper, + TestHelper::Create(dir->Str(), schema, + /*partition_keys=*/{}, + /*primary_keys=*/{"k"}, options, + /*is_streaming_mode=*/true, + /*ignore_if_exists=*/false, PathUtil::JoinPath(dir->Str(), "tmp"))); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + + ASSERT_OK_AND_ASSIGN( + std::unique_ptr batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([[1, 10], [2, 20]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(WriteAndFullCompact(helper.get(), std::move(batch), /*commit_identifier=*/0)); + + ASSERT_OK_AND_ASSIGN(SystemTableReadResult result, + ReadSystemTable(table_path + "$ro", options)); + std::shared_ptr expected_type = + arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), + arrow::field("k", arrow::int32()), arrow::field("v", arrow::int32())}); + std::shared_ptr expected; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 1, 10], [0, 2, 20]])"}, &expected) + .ok()); + ASSERT_TRUE(result.array->Equals(expected)) << result.array->ToString(); +} + TEST(SystemTableReadInteTest, TestReadFilesSystemTableForPartitionedTable) { arrow::FieldVector fields = { arrow::field("dt", arrow::utf8()), From f83744b94e2d453cd04a13db3e15b72a0bdd3d5f Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Fri, 10 Jul 2026 10:56:21 +0800 Subject: [PATCH 2/2] test: address read-optimized review comments --- include/paimon/read_context.h | 4 +- src/paimon/core/table/source/table_scan.cpp | 7 +- .../system/read_optimized_system_table.cpp | 9 +- test/inte/read_inte_test.cpp | 104 +++++++++++++++++- 4 files changed, 110 insertions(+), 14 deletions(-) diff --git a/include/paimon/read_context.h b/include/paimon/read_context.h index 708dd9a66..faa034e90 100644 --- a/include/paimon/read_context.h +++ b/include/paimon/read_context.h @@ -204,8 +204,8 @@ class PAIMON_EXPORT ReadContextBuilder { /// @param read_field_ids Vector of field ids to read from the table. /// @return Reference to this builder for method chaining. /// @note Currently supports top-level field selection. - /// @note SetReadFieldIds() and SetReadFieldNames() are mutually exclusive. - /// Calling both will ignore the read schema set by SetReadFieldNames(). + /// @note If both SetReadFieldIds() and SetReadFieldNames() are set, both values are + /// preserved in ReadContext. InternalReadContext decides precedence. ReadContextBuilder& SetReadFieldIds(const std::vector& read_field_ids); /// Set the read Arrow Schema for nested column pruning. diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp index 5cf915da9..3df9c0058 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/src/paimon/core/table/source/table_scan.cpp @@ -27,6 +27,7 @@ #include "paimon/common/predicate/predicate_validator.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/fields_comparator.h" +#include "paimon/common/utils/options_utils.h" #include "paimon/core/core_options.h" #include "paimon/core/index/index_file_handler.h" #include "paimon/core/manifest/index_manifest_file.h" @@ -226,9 +227,9 @@ Result> NewDataTableScan(const std::shared_ptrGetOptions().find(kReadOptimizedScanOption); - const bool read_optimized = - read_optimized_iter != context->GetOptions().end() && read_optimized_iter->second == "true"; + PAIMON_ASSIGN_OR_RAISE(bool read_optimized, + OptionsUtils::GetValueFromMap(context->GetOptions(), + kReadOptimizedScanOption, false)); // merge options auto options = table_schema->Options(); for (const auto& [key, value] : context->GetOptions()) { diff --git a/src/paimon/core/table/system/read_optimized_system_table.cpp b/src/paimon/core/table/system/read_optimized_system_table.cpp index c4c633707..2b8443d22 100644 --- a/src/paimon/core/table/system/read_optimized_system_table.cpp +++ b/src/paimon/core/table/system/read_optimized_system_table.cpp @@ -107,12 +107,9 @@ Result> ReadOptimizedSystemTable::NewRead( .WithFileSystemSchemeToIdentifierMap(context->GetFileSystemSchemeToIdentifierMap()) .SetPrefetchCacheMode(context->GetPrefetchCacheMode()) .WithCacheConfig(context->GetCacheConfig()) - .WithCache(context->GetCache()); - if (!context->GetReadFieldIds().empty()) { - builder.SetReadFieldIds(context->GetReadFieldIds()); - } else { - builder.SetReadFieldNames(context->GetReadFieldNames()); - } + .WithCache(context->GetCache()) + .SetReadFieldNames(context->GetReadFieldNames()) + .SetReadFieldIds(context->GetReadFieldIds()); if (context->GetSpecificTableSchema().has_value()) { builder.SetTableSchema(context->GetSpecificTableSchema().value()); } diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 6298130dd..9fd72ebce 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -268,11 +268,15 @@ std::vector StructFieldNames(const std::shared_ptrtype()->fields())->field_names(); } -Result ReadSystemTable(const std::string& system_table_path, - const std::map& options, - bool streaming_mode = false) { +Result ReadSystemTable( + const std::string& system_table_path, const std::map& options, + bool streaming_mode = false, const std::shared_ptr& predicate = nullptr, + const std::vector& read_field_names = {}) { ScanContextBuilder scan_context_builder(system_table_path); scan_context_builder.SetOptions(options).WithStreamingMode(streaming_mode); + if (predicate) { + scan_context_builder.SetPredicate(predicate); + } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan_context, scan_context_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_scan, @@ -281,6 +285,12 @@ Result ReadSystemTable(const std::string& system_table_pa ReadContextBuilder read_context_builder(system_table_path); read_context_builder.SetOptions(options); + if (predicate) { + read_context_builder.SetPredicate(predicate); + } + if (!read_field_names.empty()) { + read_context_builder.SetReadFieldNames(read_field_names); + } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr read_context, read_context_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_read, @@ -312,6 +322,18 @@ void AssertStructArrayEqualsJson(const std::shared_ptr& actu << "expected: " << expected->ToString() << "\nactual: " << actual->ToString(); } +Result CountDataFiles(const std::vector>& splits) { + int64_t file_count = 0; + for (const auto& split : splits) { + auto data_split = std::dynamic_pointer_cast(split); + if (!data_split) { + return Status::Invalid("expected data split"); + } + file_count += data_split->GetFileList().size(); + } + return file_count; +} + } // namespace std::vector PrepareTestParam() { @@ -895,6 +917,74 @@ TEST(SystemTableReadInteTest, TestReadOptimizedAppendOnlySystemTableWithStreamin ASSERT_TRUE(result.array->Equals(expected)) << result.array->ToString(); } +TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableProjectionAndPredicatePushdown) { + arrow::FieldVector fields = { + arrow::field("k", arrow::int32()), + arrow::field("v", arrow::int32()), + arrow::field("extra", arrow::int32()), + }; + auto schema = arrow::schema(fields); + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "parquet"}, + {Options::MANIFEST_FORMAT, "avro"}, + {Options::BUCKET, "1"}, + {Options::BUCKET_KEY, "k"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN(std::unique_ptr helper, + TestHelper::Create(dir->Str(), schema, + /*partition_keys=*/{}, + /*primary_keys=*/{}, options, + /*is_streaming_mode=*/true)); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + auto row_type = arrow::struct_(fields); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_1, + TestHelper::MakeRecordBatch(row_type, R"([[1, 10, 100]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch_1), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_2, + TestHelper::MakeRecordBatch(row_type, R"([[2, 20, 200]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch_2), /*commit_identifier=*/1, + /*expected_commit_messages=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_3, + TestHelper::MakeRecordBatch(row_type, R"([[3, 30, 300]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch_3), /*commit_identifier=*/2, + /*expected_commit_messages=*/std::nullopt)); + + std::shared_ptr predicate = + PredicateBuilder::GreaterThan(0, "k", FieldType::INT, Literal(1)); + + ScanContextBuilder scan_context_builder(table_path + "$ro"); + scan_context_builder.SetOptions(options).SetPredicate(predicate); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scan_context, scan_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(scan_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_OK_AND_ASSIGN(int64_t file_count, CountDataFiles(plan->Splits())); + ASSERT_EQ(file_count, 2); + + ReadContextBuilder read_context_builder(table_path + "$ro"); + read_context_builder.SetOptions(options).SetReadFieldNames({"v"}); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, read_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_read, + TableRead::Create(std::move(read_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_reader, + table_read->CreateReader(plan->Splits())); + ASSERT_OK_AND_ASSIGN(std::shared_ptr result, + ReadResultCollector::CollectResult(batch_reader.get())); + std::shared_ptr expected_type = arrow::struct_( + {arrow::field("_VALUE_KIND", arrow::int8()), arrow::field("v", arrow::int32())}); + std::shared_ptr expected; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 20], [0, 30]])"}, &expected) + .ok()); + ASSERT_TRUE(result->Equals(expected)) << result->ToString(); +} + TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableWithBranch) { arrow::FieldVector fields = { arrow::field("k", arrow::int32()), @@ -945,6 +1035,14 @@ TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableWithBranch) { expected_type, {R"([[0, 1, 10], [0, 2, 20]])"}, &expected) .ok()); ASSERT_TRUE(result.array->Equals(expected)) << result.array->ToString(); + + ASSERT_OK_AND_ASSIGN(SystemTableReadResult main_result, + ReadSystemTable(table_path + "$ro", options)); + std::shared_ptr expected_main; + ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON( + expected_type, {R"([[0, 1, 11], [0, 2, 20], [0, 3, 30]])"}, &expected_main) + .ok()); + ASSERT_TRUE(main_result.array->Equals(expected_main)) << main_result.array->ToString(); } TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableWithFirstRowMergeEngine) {