diff --git a/include/paimon/defs.h b/include/paimon/defs.h index 564854908..5efdf07dd 100644 --- a/include/paimon/defs.h +++ b/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[]; diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index 66fdb0c92..27c21efdc 100644 --- a/src/paimon/common/defs.cpp +++ b/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"; diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index 26519d06f..c1ca0a45a 100644 --- a/src/paimon/core/core_options.cpp +++ b/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 blob_split_by_file_size; bool legacy_partition_name_enabled = true; bool global_index_enabled = true; std::optional 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(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(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 diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h index 03b733cfe..ee6f0f0f5 100644 --- a/src/paimon/core/core_options.h +++ b/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; diff --git a/src/paimon/core/core_options_test.cpp b/src/paimon/core/core_options_test.cpp index 0a7f02242..c2044004f 100644 --- a/src/paimon/core/core_options_test.cpp +++ b/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({"blob1", "blob2"})); ASSERT_EQ(core_options.GetBlobDescriptorFields(), std::vector({"blob3", "blob4"})); + ASSERT_FALSE(core_options.BlobSplitByFileSize()); ASSERT_EQ(core_options.GetBlobViewFields(), std::vector({"blob5"})); ASSERT_EQ(core_options.GetBlobInlineFields(), std::vector({"blob3", "blob4", "blob5"})); @@ -923,6 +926,12 @@ TEST(CoreOptionsTest, TestFallback) { ASSERT_EQ(options.GetBlobDescriptorFields(), std::vector({"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) { diff --git a/src/paimon/core/table/source/data_evolution_split_generator.cpp b/src/paimon/core/table/source/data_evolution_split_generator.cpp index df571508a..b600c75d4 100644 --- a/src/paimon/core/table/source/data_evolution_split_generator.cpp +++ b/src/paimon/core/table/source/data_evolution_split_generator.cpp @@ -21,6 +21,7 @@ #include #include +#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> DataEvolutionSplitGenerator::Spl PAIMON_ASSIGN_OR_RAISE(std::vector>> 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>& 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); }; diff --git a/src/paimon/core/table/source/data_evolution_split_generator.h b/src/paimon/core/table/source/data_evolution_split_generator.h index bd3701114..fb919bbfb 100644 --- a/src/paimon/core/table/source/data_evolution_split_generator.h +++ b/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> SplitForBatch( std::vector>&& 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 diff --git a/src/paimon/core/table/source/split_generator_test.cpp b/src/paimon/core/table/source/split_generator_test.cpp index c7b5807af..5ec1e1859 100644 --- a/src/paimon/core/table/source/split_generator_test.cpp +++ b/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 CreateDataFileMetaWithRowId(const std::string& file_name, + int64_t file_size, int64_t row_count, + int64_t first_row_id) { + return std::make_shared( + 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>(), + /*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& result_groups, const std::vector>& expected_file_names, const std::vector& 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>{ + 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 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 split_groups, + split_generator.SplitForBatch(create_files())); + CheckResult(split_groups, {{"f1", "blob1.blob", "f2", "blob2.blob"}}, {false}); + } +} + TEST_F(SplitGeneratorTest, TestMergeTree) { std::vector> files = { CreateDataFileMeta("1", 0, 10), CreateDataFileMeta("2", 0, 12), diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp index cff55bd21..1c8a015f9 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/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(source_split_target_size, - source_split_open_file_cost); + return std::make_unique( + 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);