feat(blob): support blob.split-by-file-size to weigh blob files in scan splitting

Align scan split weighting with CoreOptions.BLOB_SPLIT_BY_FILE_SIZE in Java
Paimon:

- Add the 'blob.split-by-file-size' option (no default value) to control
  whether blob file size is considered as a factor when performing scan
  splitting, and parse it into CoreOptions together with 'blob-as-descriptor'.
- Expose CoreOptions::BlobSplitByFileSize(), which returns the explicit value
  or falls back to the negation of 'blob-as-descriptor': content reads count
  blob file sizes, while descriptor reads, which materialize only small
  descriptors instead of blob payloads, should not let huge blob files
  dominate split weights and shatter the scan into tiny splits.
- Add a count_blob_size flag to DataEvolutionSplitGenerator: the split weight
  function counts a blob file as its file_size when enabled, otherwise as the
  open file cost; non-blob files always count as file_size, and each row-id
  range still weighs at least the open file cost.
- Pass CoreOptions::BlobSplitByFileSize() when constructing
  DataEvolutionSplitGenerator in the data-evolution scan path.
- Tests: CoreOptionsTest covers the default, the value derived from
  'blob-as-descriptor', and explicit-value precedence; SplitGeneratorTest
  verifies that the same data/blob file layout packs into different splits
  depending on the flag.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: 子懿

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. Default is 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/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

@@ -443,6 +443,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;
@@ -567,6 +569,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();
     }
 
@@ -944,6 +951,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

@@ -68,6 +68,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));
 
@@ -227,6 +228,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"},
@@ -365,6 +367,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"}));
@@ -923,6 +926,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

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

@@ -36,6 +36,7 @@
 #include "paimon/core/stats/simple_stats.h"
 #include "paimon/core/table/bucket_mode.h"
 #include "paimon/core/table/source/append_only_split_generator.h"
+#include "paimon/core/table/source/data_evolution_split_generator.h"
 #include "paimon/core/table/source/merge_tree_split_generator.h"
 #include "paimon/data/timestamp.h"
 #include "paimon/memory/memory_pool.h"
@@ -116,6 +117,21 @@ class SplitGeneratorTest : public testing::Test {
             /*write_cols=*/std::nullopt);
     }
 
+    std::shared_ptr<DataFileMeta> CreateDataFileMetaWithRowId(const std::string& file_name,
+                                                              int64_t file_size, int64_t row_count,
+                                                              int64_t first_row_id) {
+        return std::make_shared<DataFileMeta>(
+            file_name, file_size, row_count, /*min_key=*/BinaryRow::EmptyRow(),
+            /*max_key=*/BinaryRow::EmptyRow(), /*key_stats=*/SimpleStats::EmptyStats(),
+            /*value_stats=*/SimpleStats::EmptyStats(), /*min_sequence_number=*/0,
+            /*max_sequence_number=*/0, /*schema_id=*/0,
+            /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+            /*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/0,
+            /*embedded_index=*/nullptr, FileSource::Append(), /*value_stats_cols=*/std::nullopt,
+            /*external_path=*/std::nullopt, first_row_id,
+            /*write_cols=*/std::nullopt);
+    }
+
     static void CheckResult(const std::vector<SplitGenerator::SplitGroup>& result_groups,
                             const std::vector<std::vector<std::string>>& expected_file_names,
                             const std::vector<bool>& expected_raw_convertible) {
@@ -206,6 +222,41 @@ TEST_F(SplitGeneratorTest, TestAppend) {
     }
 }
 
+TEST_F(SplitGeneratorTest, TestDataEvolutionBlobSplitByFileSize) {
+    // two row id ranges, each with one data file and one much larger blob file
+    auto create_files = [&]() {
+        return std::vector<std::shared_ptr<DataFileMeta>>{
+            CreateDataFileMetaWithRowId("f1", /*file_size=*/100, /*row_count=*/100,
+                                        /*first_row_id=*/0),
+            CreateDataFileMetaWithRowId("blob1.blob", /*file_size=*/1000, /*row_count=*/100,
+                                        /*first_row_id=*/0),
+            CreateDataFileMetaWithRowId("f2", /*file_size=*/100, /*row_count=*/100,
+                                        /*first_row_id=*/100),
+            CreateDataFileMetaWithRowId("blob2.blob", /*file_size=*/1000, /*row_count=*/100,
+                                        /*first_row_id=*/100)};
+    };
+    {
+        // blob file size counts in splitting: each range weighs 1100, so the two ranges cannot
+        // be packed into one split of target size 1200
+        DataEvolutionSplitGenerator split_generator(/*target_split_size=*/1200,
+                                                    /*open_file_cost=*/10,
+                                                    /*count_blob_size=*/true);
+        ASSERT_OK_AND_ASSIGN(std::vector<SplitGenerator::SplitGroup> split_groups,
+                             split_generator.SplitForBatch(create_files()));
+        CheckResult(split_groups, {{"f1", "blob1.blob"}, {"f2", "blob2.blob"}}, {false, false});
+    }
+    {
+        // blob file only weighs the open file cost: each range weighs 110, both ranges fit in
+        // one split of target size 1200
+        DataEvolutionSplitGenerator split_generator(/*target_split_size=*/1200,
+                                                    /*open_file_cost=*/10,
+                                                    /*count_blob_size=*/false);
+        ASSERT_OK_AND_ASSIGN(std::vector<SplitGenerator::SplitGroup> split_groups,
+                             split_generator.SplitForBatch(create_files()));
+        CheckResult(split_groups, {{"f1", "blob1.blob", "f2", "blob2.blob"}}, {false});
+    }
+}
+
 TEST_F(SplitGeneratorTest, TestMergeTree) {
     std::vector<std::shared_ptr<DataFileMeta>> files = {
         CreateDataFileMeta("1", 0, 10),  CreateDataFileMeta("2", 0, 12),

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

@@ -134,8 +134,9 @@ class TableScanImpl {
         auto source_split_open_file_cost = core_options.GetSourceSplitOpenFileCost();
         if (table_schema->PrimaryKeys().empty()) {
             if (core_options.DataEvolutionEnabled()) {
-                return std::make_unique<DataEvolutionSplitGenerator>(source_split_target_size,
-                                                                     source_split_open_file_cost);
+                return std::make_unique<DataEvolutionSplitGenerator>(
+                    source_split_target_size, source_split_open_file_cost,
+                    core_options.BlobSplitByFileSize());
             }
             BucketMode bucket_mode = (core_options.GetBucket() == -1 ? BucketMode::BUCKET_UNAWARE
                                                                      : BucketMode::HASH_FIXED);