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

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);