feat: add read-optimized system table

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

docs/source/user_guide/system_tables.rst

@@ -0,0 +1,67 @@
+.. 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.
+
+This stale view is not guaranteed to correspond to any single historical table
+snapshot. Full compaction may finish independently for different buckets. When
+``$ro`` scans the latest snapshot, the selected highest-level files can
+therefore combine bucket states produced by full-compaction commits at different
+snapshot IDs.
+
+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, and the result may not match
+  any single historical snapshot across buckets.
+- 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.

src/paimon/CMakeLists.txt

@@ -356,6 +356,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

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<SnapshotReader>& snapshot_reader,
-                                       std::optional<int32_t> push_down_limit)
+                                       bool read_optimized, std::optional<int32_t> 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();

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<SnapshotReader>& snapshot_reader,
+                       const std::shared_ptr<SnapshotReader>& snapshot_reader, bool read_optimized,
                        std::optional<int32_t> push_down_limit);
 
     Result<std::shared_ptr<Plan>> CreatePlan() override;

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

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"
@@ -47,6 +48,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,6 +227,9 @@ Result<std::unique_ptr<TableScan>> NewDataTableScan(const std::shared_ptr<ScanCo
         }
         table_schema = latest_table_schema.value();
     }
+    PAIMON_ASSIGN_OR_RAISE(bool read_optimized,
+                           OptionsUtils::GetValueFromMap<bool>(context->GetOptions(),
+                                                               kReadOptimizedScanOption, false));
     // merge options
     auto options = table_schema->Options();
     for (const auto& [key, value] : context->GetOptions()) {
@@ -280,12 +285,16 @@ Result<std::unique_ptr<TableScan>> NewDataTableScan(const std::shared_ptr<ScanCo
                                                                  context->GetMemoryPool()));
     auto snapshot_reader = std::make_shared<SnapshotReader>(
         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<DataTableStreamScan>(core_options, snapshot_reader);
     }
-    auto batch_scan =
-        std::make_unique<DataTableBatchScan>(/*pk_table=*/!table_schema->PrimaryKeys().empty(),
-                                             core_options, snapshot_reader, context->GetLimit());
+    auto batch_scan = std::make_unique<DataTableBatchScan>(
+        /*pk_table=*/pk_table, core_options, snapshot_reader, read_optimized, context->GetLimit());
     if (!core_options.DataEvolutionEnabled()) {
         return batch_scan;
     }

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

@@ -16,6 +16,7 @@
 
 #include "paimon/table/source/table_scan.h"
 
+#include <memory>
 #include <string>
 #include <utility>
 #include <vector>
@@ -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<ScanContext> 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

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

@@ -0,0 +1,128 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <utility>
+
+#include "arrow/c/bridge.h"
+#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<TableSchema> table_schema,
+                                                   std::map<std::string, std::string> 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<std::shared_ptr<arrow::Schema>> ReadOptimizedSystemTable::ArrowSchema() const {
+    return DataField::ConvertDataFieldsToArrowSchema(table_schema_->Fields());
+}
+
+std::map<std::string, std::string> ReadOptimizedSystemTable::ReadOptimizedOptions() const {
+    auto options = options_;
+    options[kReadOptimizedScanOption] = "true";
+    return options;
+}
+
+Result<std::unique_ptr<TableScan>> ReadOptimizedSystemTable::NewScan(
+    const std::shared_ptr<ScanContext>& 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<ScanContext> base_context, builder.Finish());
+    return TableScan::Create(std::move(base_context));
+}
+
+Result<std::unique_ptr<TableRead>> ReadOptimizedSystemTable::NewRead(
+    const std::shared_ptr<ReadContext>& 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())
+        .SetReadFieldNames(context->GetReadFieldNames())
+        .SetReadFieldIds(context->GetReadFieldIds());
+    if (context->HasReadSchema()) {
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> read_schema,
+                                          arrow::ImportSchema(context->GetReadSchema()));
+        auto c_read_schema = std::make_unique<::ArrowSchema>();
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema, c_read_schema.get()));
+        builder.SetReadSchema(std::move(c_read_schema));
+    }
+    if (context->GetSpecificTableSchema().has_value()) {
+        builder.SetTableSchema(context->GetSpecificTableSchema().value());
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> base_context, builder.Finish());
+    return TableRead::Create(std::move(base_context));
+}
+
+}  // namespace paimon

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 <map>
+#include <memory>
+#include <string>
+
+#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<TableSchema> table_schema,
+                             std::map<std::string, std::string> options);
+
+    std::string Name() const override;
+    Result<std::shared_ptr<arrow::Schema>> ArrowSchema() const override;
+    Result<std::unique_ptr<TableScan>> NewScan(
+        const std::shared_ptr<ScanContext>& context) const override;
+    Result<std::unique_ptr<TableRead>> NewRead(
+        const std::shared_ptr<ReadContext>& context) const override;
+
+ private:
+    std::map<std::string, std::string> ReadOptimizedOptions() const;
+
+    std::string table_path_;
+    std::shared_ptr<TableSchema> table_schema_;
+    std::map<std::string, std::string> options_;
+};
+
+}  // namespace paimon