Skip to content

Commit ee3fbc0

Browse files
gripleafgripleaf
authored
feat: add read-optimized system table (#403)
1 parent 8cebd65 commit ee3fbc0

14 files changed

Lines changed: 723 additions & 13 deletions

‎docs/source/user_guide.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ User Guide
3030
user_guide/data_types
3131
user_guide/primary_key_table
3232
user_guide/append_only_table
33+
user_guide/system_tables
3334
user_guide/write
3435
user_guide/commit
3536
user_guide/compaction

‎docs/source/user_guide/system_tables.rst‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
.. Copyright 2026-present Alibaba Inc.
2+
3+
.. Licensed under the Apache License, Version 2.0 (the "License");
4+
.. you may not use this file except in compliance with the License.
5+
.. You may obtain a copy of the License at
6+
7+
.. http://www.apache.org/licenses/LICENSE-2.0
8+
9+
.. Unless required by applicable law or agreed to in writing, software
10+
.. distributed under the License is distributed on an "AS IS" BASIS,
11+
.. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
.. See the License for the specific language governing permissions and
13+
.. limitations under the License.
14+
15+
System Tables
16+
=============
17+
18+
Paimon C++ supports reading system tables by appending a system table suffix to
19+
the data table path. For example, ``/warehouse/db.db/orders$snapshots`` reads
20+
the snapshots system table for ``orders``.
21+
22+
Branch-qualified paths are also supported. For example,
23+
``/warehouse/db.db/orders$branch_audit$files`` reads the files system table from
24+
the ``audit`` branch.
25+
26+
Read-Optimized System Table
27+
---------------------------
28+
29+
The read-optimized system table is addressed by the ``$ro`` suffix:
30+
31+
.. code-block:: text
32+
33+
/warehouse/db.db/orders$ro
34+
35+
For primary-key tables, ``$ro`` only plans data files from the highest LSM
36+
level, which is the level produced by full compaction. This avoids merging data
37+
from multiple LSM levels during query planning and reading. The tradeoff is that
38+
the result may lag behind the latest committed data until a full compaction
39+
publishes the newest records into the highest level.
40+
41+
This stale view is not guaranteed to correspond to any single historical table
42+
snapshot. Full compaction may finish independently for different buckets. When
43+
``$ro`` scans the latest snapshot, the selected highest-level files can
44+
therefore combine bucket states produced by full-compaction commits at different
45+
snapshot IDs.
46+
47+
For primary-key tables, ``$ro`` also enables value-stats filtering, so file-level
48+
pruning of the reader predicate can be more aggressive than the base table.
49+
50+
For append-only tables, ``$ro`` has the same read behavior as the base table,
51+
including streaming reads.
52+
53+
Limitations
54+
~~~~~~~~~~~
55+
56+
- Primary-key ``$ro`` scans are batch-only. Streaming scans are not supported.
57+
- Freshness depends on full compaction frequency, and the result may not match
58+
any single historical snapshot across buckets.
59+
- Primary-key tables in bucket-unaware mode are not supported by the current C++
60+
scan path.
61+
62+
Typical Usage
63+
~~~~~~~~~~~~~
64+
65+
Use ``$ro`` for OLAP or batch workloads that can tolerate stale results and
66+
prefer reading compacted files directly. Use the base table path when the query
67+
must see the latest committed data.

‎src/paimon/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -356,6 +356,7 @@ set(PAIMON_CORE_SRCS
356356
core/table/system/binlog_system_table.cpp
357357
core/table/system/in_memory_system_table.cpp
358358
core/table/system/metadata_system_tables.cpp
359+
core/table/system/read_optimized_system_table.cpp
359360
core/table/system/system_table.cpp
360361
core/table/system/system_table_scan.cpp
361362
core/table/system/system_table_schema.cpp

‎src/paimon/core/table/source/data_table_batch_scan.cpp‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,15 @@ class DataSplit;
3333

3434
DataTableBatchScan::DataTableBatchScan(bool pk_table, const CoreOptions& core_options,
3535
const std::shared_ptr<SnapshotReader>& snapshot_reader,
36-
std::optional<int32_t> push_down_limit)
36+
bool read_optimized, std::optional<int32_t> push_down_limit)
3737
: AbstractTableScan(core_options, snapshot_reader), push_down_limit_(push_down_limit) {
38-
if (pk_table && (core_options.DeletionVectorsEnabled() ||
39-
core_options.GetMergeEngine() == MergeEngine::FIRST_ROW)) {
38+
if (pk_table && read_optimized) {
39+
int32_t top_level = core_options.GetNumLevels() - 1;
40+
snapshot_reader_->WithLevelFilter(
41+
[top_level](int32_t level) -> bool { return level == top_level; });
42+
snapshot_reader_->EnableValueFilter();
43+
} else if (pk_table && (core_options.DeletionVectorsEnabled() ||
44+
core_options.GetMergeEngine() == MergeEngine::FIRST_ROW)) {
4045
auto level_filter = [](int32_t level) -> bool { return level > 0; };
4146
snapshot_reader_->WithLevelFilter(level_filter);
4247
snapshot_reader_->EnableValueFilter();

‎src/paimon/core/table/source/data_table_batch_scan.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ class SnapshotReader;
3232
class DataTableBatchScan : public AbstractTableScan {
3333
public:
3434
DataTableBatchScan(bool pk_table, const CoreOptions& core_options,
35-
const std::shared_ptr<SnapshotReader>& snapshot_reader,
35+
const std::shared_ptr<SnapshotReader>& snapshot_reader, bool read_optimized,
3636
std::optional<int32_t> push_down_limit);
3737

3838
Result<std::shared_ptr<Plan>> CreatePlan() override;

‎src/paimon/core/table/source/read_optimized_scan_options.h‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
namespace paimon {
20+
21+
/// Internal scan option used by `T$ro` to request top-level-only planning.
22+
inline constexpr char kReadOptimizedScanOption[] = "__paimon.internal.read-optimized";
23+
24+
} // namespace paimon

‎src/paimon/core/table/source/table_scan.cpp‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include "paimon/common/predicate/predicate_validator.h"
2828
#include "paimon/common/types/data_field.h"
2929
#include "paimon/common/utils/fields_comparator.h"
30+
#include "paimon/common/utils/options_utils.h"
3031
#include "paimon/core/core_options.h"
3132
#include "paimon/core/index/index_file_handler.h"
3233
#include "paimon/core/manifest/index_manifest_file.h"
@@ -47,6 +48,7 @@
4748
#include "paimon/core/table/source/data_table_batch_scan.h"
4849
#include "paimon/core/table/source/data_table_stream_scan.h"
4950
#include "paimon/core/table/source/merge_tree_split_generator.h"
51+
#include "paimon/core/table/source/read_optimized_scan_options.h"
5052
#include "paimon/core/table/source/snapshot/snapshot_reader.h"
5153
#include "paimon/core/table/source/split_generator.h"
5254
#include "paimon/core/table/system/system_table.h"
@@ -225,6 +227,9 @@ Result<std::unique_ptr<TableScan>> NewDataTableScan(const std::shared_ptr<ScanCo
225227
}
226228
table_schema = latest_table_schema.value();
227229
}
230+
PAIMON_ASSIGN_OR_RAISE(bool read_optimized,
231+
OptionsUtils::GetValueFromMap<bool>(context->GetOptions(),
232+
kReadOptimizedScanOption, false));
228233
// merge options
229234
auto options = table_schema->Options();
230235
for (const auto& [key, value] : context->GetOptions()) {
@@ -280,12 +285,16 @@ Result<std::unique_ptr<TableScan>> NewDataTableScan(const std::shared_ptr<ScanCo
280285
context->GetMemoryPool()));
281286
auto snapshot_reader = std::make_shared<SnapshotReader>(
282287
file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler));
288+
const bool pk_table = !table_schema->PrimaryKeys().empty();
289+
if (read_optimized && pk_table && context->IsStreamingMode()) {
290+
return Status::NotImplemented(
291+
"read-optimized system table does not support streaming scan for primary key table");
292+
}
283293
if (context->IsStreamingMode()) {
284294
return std::make_unique<DataTableStreamScan>(core_options, snapshot_reader);
285295
}
286-
auto batch_scan =
287-
std::make_unique<DataTableBatchScan>(/*pk_table=*/!table_schema->PrimaryKeys().empty(),
288-
core_options, snapshot_reader, context->GetLimit());
296+
auto batch_scan = std::make_unique<DataTableBatchScan>(
297+
/*pk_table=*/pk_table, core_options, snapshot_reader, read_optimized, context->GetLimit());
289298
if (!core_options.DataEvolutionEnabled()) {
290299
return batch_scan;
291300
}

‎src/paimon/core/table/source/table_scan_test.cpp‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
#include "paimon/table/source/table_scan.h"
1818

19+
#include <memory>
1920
#include <string>
2021
#include <utility>
2122
#include <vector>
@@ -59,4 +60,17 @@ TEST(TableScanTest, TestPkSchemaEvolutionScan) {
5960
ASSERT_FALSE(plan->Splits().empty());
6061
}
6162

63+
TEST(TableScanTest, TestReadOptimizedPrimaryKeyStreamingScanUnsupported) {
64+
std::string path = paimon::test::GetDataDir() +
65+
"/orc/pk_table_with_alter_table.db/pk_table_with_alter_table$ro";
66+
ScanContextBuilder builder(path);
67+
builder.AddOption(Options::FILE_FORMAT, "orc");
68+
builder.WithStreamingMode(true);
69+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> context, builder.Finish());
70+
71+
ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(context)),
72+
"read-optimized system table does not support streaming scan for primary "
73+
"key table");
74+
}
75+
6276
} // namespace paimon::test

‎src/paimon/core/table/system/read_optimized_system_table.cpp‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/core/table/system/read_optimized_system_table.h"
18+
19+
#include <memory>
20+
#include <string>
21+
#include <utility>
22+
23+
#include "arrow/c/bridge.h"
24+
#include "paimon/common/types/data_field.h"
25+
#include "paimon/core/schema/table_schema.h"
26+
#include "paimon/core/table/source/read_optimized_scan_options.h"
27+
#include "paimon/defs.h"
28+
#include "paimon/read_context.h"
29+
#include "paimon/scan_context.h"
30+
#include "paimon/table/source/table_read.h"
31+
#include "paimon/table/source/table_scan.h"
32+
33+
namespace paimon {
34+
35+
ReadOptimizedSystemTable::ReadOptimizedSystemTable(std::string table_path,
36+
std::shared_ptr<TableSchema> table_schema,
37+
std::map<std::string, std::string> options)
38+
: table_path_(std::move(table_path)),
39+
table_schema_(std::move(table_schema)),
40+
options_(std::move(options)) {}
41+
42+
std::string ReadOptimizedSystemTable::Name() const {
43+
return kName;
44+
}
45+
46+
Result<std::shared_ptr<arrow::Schema>> ReadOptimizedSystemTable::ArrowSchema() const {
47+
return DataField::ConvertDataFieldsToArrowSchema(table_schema_->Fields());
48+
}
49+
50+
std::map<std::string, std::string> ReadOptimizedSystemTable::ReadOptimizedOptions() const {
51+
auto options = options_;
52+
options[kReadOptimizedScanOption] = "true";
53+
return options;
54+
}
55+
56+
Result<std::unique_ptr<TableScan>> ReadOptimizedSystemTable::NewScan(
57+
const std::shared_ptr<ScanContext>& context) const {
58+
auto options = ReadOptimizedOptions();
59+
ScanContextBuilder builder(table_path_);
60+
builder.SetOptions(options)
61+
.WithStreamingMode(context->IsStreamingMode())
62+
.WithMemoryPool(context->GetMemoryPool())
63+
.WithExecutor(context->GetExecutor())
64+
.WithFileSystem(context->GetSpecificFileSystem())
65+
.WithCache(context->GetCache());
66+
if (context->GetLimit().has_value()) {
67+
builder.SetLimit(context->GetLimit().value());
68+
}
69+
if (context->GetScanFilters()) {
70+
if (context->GetScanFilters()->GetBucketFilter().has_value()) {
71+
builder.SetBucketFilter(context->GetScanFilters()->GetBucketFilter().value());
72+
}
73+
builder.SetPartitionFilter(context->GetScanFilters()->GetPartitionFilters());
74+
builder.SetPredicate(context->GetScanFilters()->GetPredicate());
75+
}
76+
if (context->GetGlobalIndexResult()) {
77+
builder.SetGlobalIndexResult(context->GetGlobalIndexResult());
78+
}
79+
if (context->GetSpecificTableSchema().has_value()) {
80+
builder.SetTableSchema(context->GetSpecificTableSchema().value());
81+
}
82+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> base_context, builder.Finish());
83+
return TableScan::Create(std::move(base_context));
84+
}
85+
86+
Result<std::unique_ptr<TableRead>> ReadOptimizedSystemTable::NewRead(
87+
const std::shared_ptr<ReadContext>& context) const {
88+
auto options = options_;
89+
std::string branch = context->GetBranch();
90+
// SystemTableLoader injects Options::BRANCH when parsing paths such as `T$branch_dev$ro`.
91+
auto branch_iter = options.find(Options::BRANCH);
92+
if (branch_iter != options.end()) {
93+
branch = branch_iter->second;
94+
}
95+
ReadContextBuilder builder(table_path_);
96+
builder.SetOptions(options)
97+
.WithBranch(branch)
98+
.SetPredicate(context->GetPredicate())
99+
.EnablePredicateFilter(context->EnablePredicateFilter())
100+
.EnablePrefetch(context->EnablePrefetch())
101+
.SetPrefetchBatchCount(context->GetPrefetchBatchCount())
102+
.SetPrefetchMaxParallelNum(context->GetPrefetchMaxParallelNum())
103+
.EnableMultiThreadRowToBatch(context->EnableMultiThreadRowToBatch())
104+
.SetRowToBatchThreadNumber(context->GetRowToBatchThreadNumber())
105+
.WithMemoryPool(context->GetMemoryPool())
106+
.WithExecutor(context->GetExecutor())
107+
.WithFileSystem(context->GetSpecificFileSystem())
108+
.WithFileSystemSchemeToIdentifierMap(context->GetFileSystemSchemeToIdentifierMap())
109+
.SetPrefetchCacheMode(context->GetPrefetchCacheMode())
110+
.WithCacheConfig(context->GetCacheConfig())
111+
.WithCache(context->GetCache())
112+
.SetReadFieldNames(context->GetReadFieldNames())
113+
.SetReadFieldIds(context->GetReadFieldIds());
114+
if (context->HasReadSchema()) {
115+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> read_schema,
116+
arrow::ImportSchema(context->GetReadSchema()));
117+
auto c_read_schema = std::make_unique<::ArrowSchema>();
118+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema, c_read_schema.get()));
119+
builder.SetReadSchema(std::move(c_read_schema));
120+
}
121+
if (context->GetSpecificTableSchema().has_value()) {
122+
builder.SetTableSchema(context->GetSpecificTableSchema().value());
123+
}
124+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> base_context, builder.Finish());
125+
return TableRead::Create(std::move(base_context));
126+
}
127+
128+
} // namespace paimon

‎src/paimon/core/table/system/read_optimized_system_table.h‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
#include <map>
20+
#include <memory>
21+
#include <string>
22+
23+
#include "paimon/core/table/system/system_table.h"
24+
25+
namespace paimon {
26+
class TableSchema;
27+
28+
/// System table for `T$ro`, exposing read-optimized data.
29+
class ReadOptimizedSystemTable : public SystemTable {
30+
public:
31+
static constexpr const char* kName = "ro";
32+
33+
ReadOptimizedSystemTable(std::string table_path, std::shared_ptr<TableSchema> table_schema,
34+
std::map<std::string, std::string> options);
35+
36+
std::string Name() const override;
37+
Result<std::shared_ptr<arrow::Schema>> ArrowSchema() const override;
38+
Result<std::unique_ptr<TableScan>> NewScan(
39+
const std::shared_ptr<ScanContext>& context) const override;
40+
Result<std::unique_ptr<TableRead>> NewRead(
41+
const std::shared_ptr<ReadContext>& context) const override;
42+
43+
private:
44+
std::map<std::string, std::string> ReadOptimizedOptions() const;
45+
46+
std::string table_path_;
47+
std::shared_ptr<TableSchema> table_schema_;
48+
std::map<std::string, std::string> options_;
49+
};
50+
51+
} // namespace paimon

0 commit comments

Comments
 (0)