Merge branch 'main' into blob-write-null-options

Author: lxy-9602

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.

include/paimon/defs.h

@@ -118,6 +118,10 @@ struct PAIMON_EXPORT Options {
     /// "blob.target-file-size" - Target size of a blob file. Default is TARGET_FILE_SIZE.
     static const char BLOB_TARGET_FILE_SIZE[];
 
+    /// "blob.split-by-file-size" - Whether to consider blob file size as a factor when performing
+    /// scan splitting. When unset, defaults to the negation of BLOB_AS_DESCRIPTOR.
+    static const char BLOB_SPLIT_BY_FILE_SIZE[];
+
     /// "partition.default-name" - The default partition name in case the dynamic partition column
     /// value is null/empty string. Default is "__DEFAULT_PARTITION__".
     static const char PARTITION_DEFAULT_NAME[];

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/common/defs.cpp

@@ -33,6 +33,7 @@ const char Options::FILE_FORMAT[] = "file.format";
 const char Options::FILE_SYSTEM[] = "file-system";
 const char Options::TARGET_FILE_SIZE[] = "target-file-size";
 const char Options::BLOB_TARGET_FILE_SIZE[] = "blob.target-file-size";
+const char Options::BLOB_SPLIT_BY_FILE_SIZE[] = "blob.split-by-file-size";
 const char Options::PAGE_SIZE[] = "page-size";
 const char Options::PARTITION_DEFAULT_NAME[] = "partition.default-name";
 const char Options::FILE_COMPRESSION[] = "file.compression";

src/paimon/core/core_options.cpp

@@ -448,6 +448,8 @@ struct CoreOptions::Impl {
     bool row_tracking_partition_group_on_commit = true;
     bool data_evolution_enabled = false;
     bool blob_view_resolve_enabled = true;
+    bool blob_as_descriptor = false;
+    std::optional<bool> blob_split_by_file_size;
     bool legacy_partition_name_enabled = true;
     bool global_index_enabled = true;
     std::optional<int32_t> global_index_thread_num;
@@ -578,6 +580,11 @@ struct CoreOptions::Impl {
         // Parse blob-view.resolve.enabled - whether to resolve blob view fields at read time
         PAIMON_RETURN_NOT_OK(
             parser.Parse<bool>(Options::BLOB_VIEW_RESOLVE_ENABLED, &blob_view_resolve_enabled));
+        // Parse blob-as-descriptor - read blob field as descriptor rather than blob bytes
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::BLOB_AS_DESCRIPTOR, &blob_as_descriptor));
+        // Parse blob.split-by-file-size - whether blob file size counts in scan splitting
+        PAIMON_RETURN_NOT_OK(
+            parser.Parse(Options::BLOB_SPLIT_BY_FILE_SIZE, &blob_split_by_file_size));
         return Status::OK();
     }
 
@@ -984,6 +991,10 @@ int64_t CoreOptions::GetBlobTargetFileSize() const {
     return impl_->blob_target_file_size.value();
 }
 
+bool CoreOptions::BlobSplitByFileSize() const {
+    return impl_->blob_split_by_file_size.value_or(!impl_->blob_as_descriptor);
+}
+
 int64_t CoreOptions::GetCompactionFileSize(bool has_primary_key) const {
     // file size to join the compaction, we don't process on middle file size to avoid
     // compact a same file twice (the compression is not calculate so accurately. the output

src/paimon/core/core_options.h

@@ -77,6 +77,7 @@ class PAIMON_EXPORT CoreOptions {
     int64_t GetPageSize() const;
     int64_t GetTargetFileSize(bool has_primary_key) const;
     int64_t GetBlobTargetFileSize() const;
+    bool BlobSplitByFileSize() const;
     int64_t GetCompactionFileSize(bool has_primary_key) const;
     std::string GetPartitionDefaultName() const;
 

src/paimon/core/core_options_test.cpp

@@ -41,6 +41,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(256 * 1024 * 1024L, core_options.GetTargetFileSize(/*has_primary_key=*/false));
     ASSERT_EQ(128 * 1024 * 1024L, core_options.GetTargetFileSize(/*has_primary_key=*/true));
     ASSERT_EQ(256 * 1024 * 1024L, core_options.GetBlobTargetFileSize());
+    ASSERT_TRUE(core_options.BlobSplitByFileSize());
     ASSERT_EQ(187904815, core_options.GetCompactionFileSize(/*has_primary_key=*/false));
     ASSERT_EQ(93952404, core_options.GetCompactionFileSize(/*has_primary_key=*/true));
 
@@ -240,6 +241,7 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::DATA_EVOLUTION_ENABLED, "true"},
         {Options::BLOB_FIELD, "blob1,blob2"},
         {Options::BLOB_DESCRIPTOR_FIELD, "blob3,blob4"},
+        {Options::BLOB_AS_DESCRIPTOR, "true"},
         {Options::BLOB_VIEW_FIELD, "blob5"},
         {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, "FILE:///tmp/blob_view_upstream_warehouse/"},
         {Options::BLOB_VIEW_RESOLVE_ENABLED, "false"},
@@ -385,6 +387,7 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_TRUE(core_options.DataEvolutionEnabled());
     ASSERT_EQ(core_options.GetBlobFields(), std::vector<std::string>({"blob1", "blob2"}));
     ASSERT_EQ(core_options.GetBlobDescriptorFields(), std::vector<std::string>({"blob3", "blob4"}));
+    ASSERT_FALSE(core_options.BlobSplitByFileSize());
     ASSERT_EQ(core_options.GetBlobViewFields(), std::vector<std::string>({"blob5"}));
     ASSERT_EQ(core_options.GetBlobInlineFields(),
               std::vector<std::string>({"blob3", "blob4", "blob5"}));
@@ -970,6 +973,12 @@ TEST(CoreOptionsTest, TestFallback) {
         ASSERT_EQ(options.GetBlobDescriptorFields(),
                   std::vector<std::string>({"new_b1", "new_b2"}));
     }
+    {
+        ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                             CoreOptions::FromMap({{Options::BLOB_AS_DESCRIPTOR, "true"},
+                                                   {Options::BLOB_SPLIT_BY_FILE_SIZE, "true"}}));
+        ASSERT_TRUE(options.BlobSplitByFileSize());
+    }
 }
 
 TEST(CoreOptionsTest, TestMapStorageLayout) {

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

@@ -21,6 +21,7 @@
 #include <functional>
 #include <iterator>
 
+#include "paimon/common/data/blob_utils.h"
 #include "paimon/common/utils/bin_packing.h"
 #include "paimon/common/utils/range_helper.h"
 #include "paimon/core/io/data_file_meta.h"
@@ -40,11 +41,15 @@ Result<std::vector<SplitGenerator::SplitGroup>> DataEvolutionSplitGenerator::Spl
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> ranges,
                            range_helper.MergeOverlappingRanges(std::move(input)));
 
-    auto weight_func = [open_file_cost = open_file_cost_](
+    auto weight_func = [open_file_cost = open_file_cost_, count_blob_size = count_blob_size_](
                            const std::vector<std::shared_ptr<DataFileMeta>>& metas) -> int64_t {
         int64_t file_size_sum = 0;
         for (const auto& meta : metas) {
-            file_size_sum += meta->file_size;
+            if (BlobUtils::IsBlobFile(meta->file_name)) {
+                file_size_sum += count_blob_size ? meta->file_size : open_file_cost;
+            } else {
+                file_size_sum += meta->file_size;
+            }
         }
         return std::max(file_size_sum, open_file_cost);
     };

src/paimon/core/table/source/data_evolution_split_generator.h

@@ -30,8 +30,11 @@ struct DataFileMeta;
 /// Append data evolution table split generator, which implementation of `SplitGenerator`.
 class DataEvolutionSplitGenerator : public SplitGenerator {
  public:
-    DataEvolutionSplitGenerator(int64_t target_split_size, int64_t open_file_cost)
-        : target_split_size_(target_split_size), open_file_cost_(open_file_cost) {}
+    DataEvolutionSplitGenerator(int64_t target_split_size, int64_t open_file_cost,
+                                bool count_blob_size)
+        : target_split_size_(target_split_size),
+          open_file_cost_(open_file_cost),
+          count_blob_size_(count_blob_size) {}
 
     Result<std::vector<SplitGroup>> SplitForBatch(
         std::vector<std::shared_ptr<DataFileMeta>>&& input) const override;
@@ -44,6 +47,7 @@ class DataEvolutionSplitGenerator : public SplitGenerator {
  private:
     int64_t target_split_size_;
     int64_t open_file_cost_;
+    bool count_blob_size_;
 };
 
 }  // namespace paimon