feat: add configs for compaction and default target file size for different table (#151)

Author: lucasfang

include/paimon/defs.h

@@ -105,8 +105,8 @@ struct PAIMON_EXPORT Options {
     /// Default value is local.
     static const char FILE_SYSTEM[];
 
-    /// "target-file-size" - Target size of a file. Default value is 256MB.
-    // TODO(xinyu.lxy): change the default value to 128MB for primary key table.
+    /// "target-file-size" - Target size of a file. primary key table: the default value is 128 MB.
+    /// append table: the default value is 256 MB.
     static const char TARGET_FILE_SIZE[];
 
     /// "blob.target-file-size" - Target size of a blob file. Default is TARGET_FILE_SIZE.
@@ -193,6 +193,10 @@ struct PAIMON_EXPORT Options {
     /// cause performance issue. Default value is false.
     static const char SNAPSHOT_CLEAN_EMPTY_DIRECTORIES[];
 
+    /// "commit.force-compact" - Whether to force a compaction before commit. Default value is
+    /// "false".
+    static const char COMMIT_FORCE_COMPACT[];
+
     /// "commit.timeout" - Timeout duration of retry when commit failed. No default value.
     static const char COMMIT_TIMEOUT[];
 
@@ -289,6 +293,16 @@ struct PAIMON_EXPORT Options {
     static const char GLOBAL_INDEX_EXTERNAL_PATH[];
     /// "scan.tag-name" - Optional tag name used in case of "from-snapshot" scan mode.
     static const char SCAN_TAG_NAME[];
+    /// "write-only" - If set to "true", compactions and snapshot expiration will be skipped. This
+    /// option is used along with dedicated compact jobs. Default value is "false".
+    /// @note: This option will be ignore until compaction is supported.
+    static const char WRITE_ONLY[];
+    /// "compaction.min.file-num" - For file set [f_0,...,f_N], the minimum file number to trigger a
+    /// compaction for append-only table. Default value is 5.
+    static const char COMPACTION_MIN_FILE_NUM[];
+    /// "compaction.force-rewrite-all-files" - Whether to force pick all files for a full
+    /// compaction. Usually seen in a compaction task to external paths. Default value is "false".
+    static const char COMPACTION_FORCE_REWRITE_ALL_FILES[];
 };
 
 static constexpr int64_t BATCH_WRITE_COMMIT_IDENTIFIER = std::numeric_limits<int64_t>::max();

src/paimon/common/defs.cpp

@@ -53,6 +53,7 @@ const char Options::SNAPSHOT_NUM_RETAINED_MAX[] = "snapshot.num-retained.max";
 const char Options::SNAPSHOT_TIME_RETAINED[] = "snapshot.time-retained";
 const char Options::SNAPSHOT_EXPIRE_LIMIT[] = "snapshot.expire.limit";
 const char Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES[] = "snapshot.clean-empty-directories";
+const char Options::COMMIT_FORCE_COMPACT[] = "commit.force-compact";
 const char Options::COMMIT_TIMEOUT[] = "commit.timeout";
 const char Options::COMMIT_MAX_RETRIES[] = "commit.max-retries";
 const char Options::SEQUENCE_FIELD[] = "sequence.field";
@@ -82,4 +83,8 @@ const char Options::BLOB_AS_DESCRIPTOR[] = "blob-as-descriptor";
 const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
 const char Options::GLOBAL_INDEX_EXTERNAL_PATH[] = "global-index.external-path";
 const char Options::SCAN_TAG_NAME[] = "scan.tag-name";
+const char Options::WRITE_ONLY[] = "write-only";
+const char Options::COMPACTION_MIN_FILE_NUM[] = "compaction.min.file-num";
+const char Options::COMPACTION_FORCE_REWRITE_ALL_FILES[] = "compaction.force-rewrite-all-files";
+
 }  // namespace paimon

src/paimon/core/append/append_only_writer.cpp

@@ -113,7 +113,8 @@ AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingRowWrit
         return CreateRollingBlobWriter(schemas);
     } else {
         return std::make_unique<RollingFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>(
-            options_.GetTargetFileSize(), GetDataFileWriterCreator(write_schema_, write_cols_));
+            options_.GetTargetFileSize(/*has_primary_key=*/false),
+            GetDataFileWriterCreator(write_schema_, write_cols_));
     }
 }
 
@@ -192,7 +193,7 @@ AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingBlobWri
             options_.GetBlobTargetFileSize(), single_blob_file_writer_creator);
     };
     return std::make_unique<RollingBlobFileWriter>(
-        options_.GetTargetFileSize(),
+        options_.GetTargetFileSize(/*has_primary_key=*/false),
         GetDataFileWriterCreator(schemas.main_schema, schemas.main_schema->field_names()),
         rolling_blob_file_writer_creator, arrow::struct_(write_schema_->fields()));
 }

src/paimon/core/core_options.cpp

@@ -19,6 +19,7 @@
 #include <cstring>
 #include <limits>
 #include <memory>
+#include <optional>
 #include <utility>
 
 #include "fmt/format.h"
@@ -246,6 +247,10 @@ class ConfigParser {
         return Status::OK();
     }
 
+    bool ContainsKey(const std::string& key) const {
+        return config_map_.find(key) != config_map_.end();
+    }
+
  private:
     const std::map<std::string, std::string> config_map_;
 };
@@ -254,9 +259,8 @@ class ConfigParser {
 // storing various configurable fields and their default values.
 struct CoreOptions::Impl {
     int64_t page_size = 64 * 1024;
-    int64_t target_file_size = 256 * 1024 * 1024;  // TODO(xinyu.lxy): target_file_size of primary
-                                                   // key table is 128 MB and append table is 256 MB
-    int64_t blob_target_file_size = 256 * 1024 * 1024;
+    std::optional<int64_t> target_file_size;
+    std::optional<int64_t> blob_target_file_size;
     int64_t source_split_target_size = 128 * 1024 * 1024;
     int64_t source_split_open_file_cost = 4 * 1024 * 1024;
     int64_t manifest_target_file_size = 8 * 1024 * 1024;
@@ -293,6 +297,7 @@ struct CoreOptions::Impl {
     int32_t read_batch_size = 1024;
     int32_t write_batch_size = 1024;
     int32_t commit_max_retries = 10;
+    int32_t compaction_min_file_num = 5;
 
     SortOrder sequence_field_sort_order = SortOrder::ASCENDING;
     MergeEngine merge_engine = MergeEngine::DEDUPLICATE;
@@ -303,6 +308,7 @@ struct CoreOptions::Impl {
     int32_t file_compression_zstd_level = 1;
 
     bool ignore_delete = false;
+    bool write_only = false;
     bool deletion_vectors_enabled = false;
     bool force_lookup = false;
     bool partial_update_remove_record_on_delete = false;
@@ -313,6 +319,8 @@ struct CoreOptions::Impl {
     bool data_evolution_enabled = false;
     bool legacy_partition_name_enabled = true;
     bool global_index_enabled = true;
+    bool commit_force_compact = false;
+    bool compaction_force_rewrite_all_files = false;
     std::optional<std::string> global_index_external_path;
 
     std::optional<std::string> scan_tag_name;
@@ -348,11 +356,17 @@ Result<CoreOptions> CoreOptions::FromMap(
 
     // Parse memory size configurations
     PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::PAGE_SIZE, &impl->page_size));
-    PAIMON_RETURN_NOT_OK(
-        parser.ParseMemorySize(Options::TARGET_FILE_SIZE, &impl->target_file_size));
-    impl->blob_target_file_size = impl->target_file_size;
-    PAIMON_RETURN_NOT_OK(
-        parser.ParseMemorySize(Options::BLOB_TARGET_FILE_SIZE, &impl->blob_target_file_size));
+    if (parser.ContainsKey(Options::TARGET_FILE_SIZE)) {
+        int64_t target_file_size;
+        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::TARGET_FILE_SIZE, &target_file_size));
+        impl->target_file_size = target_file_size;
+    }
+    if (parser.ContainsKey(Options::BLOB_TARGET_FILE_SIZE)) {
+        int64_t blob_target_file_size;
+        PAIMON_RETURN_NOT_OK(
+            parser.ParseMemorySize(Options::BLOB_TARGET_FILE_SIZE, &blob_target_file_size));
+        impl->blob_target_file_size = blob_target_file_size;
+    }
     PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::MANIFEST_TARGET_FILE_SIZE,
                                                 &impl->manifest_target_file_size));
     PAIMON_RETURN_NOT_OK(
@@ -411,6 +425,9 @@ Result<CoreOptions> CoreOptions::FromMap(
     // Parse ignore delete
     PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::IGNORE_DELETE, &impl->ignore_delete));
 
+    // Parse write-only
+    PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::WRITE_ONLY, &impl->write_only));
+
     // Parse default agg function
     std::string field_default_func;
     PAIMON_RETURN_NOT_OK(parser.ParseString(Options::FIELDS_DEFAULT_AGG_FUNC, &field_default_func));
@@ -490,6 +507,18 @@ Result<CoreOptions> CoreOptions::FromMap(
         impl->scan_tag_name = scan_tag_name;
     }
 
+    // Parse commit.force-compact
+    PAIMON_RETURN_NOT_OK(
+        parser.Parse<bool>(Options::COMMIT_FORCE_COMPACT, &impl->commit_force_compact));
+
+    // Parse compaction.min.file-num
+    PAIMON_RETURN_NOT_OK(
+        parser.Parse(Options::COMPACTION_MIN_FILE_NUM, &impl->compaction_min_file_num));
+
+    // Parse compaction.force-rewrite-all-files
+    PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::COMPACTION_FORCE_REWRITE_ALL_FILES,
+                                            &impl->compaction_force_rewrite_all_files));
+
     return options;
 }
 
@@ -531,12 +560,25 @@ int64_t CoreOptions::GetPageSize() const {
     return impl_->page_size;
 }
 
-int64_t CoreOptions::GetTargetFileSize() const {
-    return impl_->target_file_size;
+int64_t CoreOptions::GetTargetFileSize(bool has_primary_key) const {
+    if (impl_->target_file_size == std::nullopt) {
+        return has_primary_key ? 128 * 1024 * 1024 : 256 * 1024 * 1024;
+    }
+    return impl_->target_file_size.value();
 }
 
 int64_t CoreOptions::GetBlobTargetFileSize() const {
-    return impl_->blob_target_file_size;
+    if (impl_->blob_target_file_size == std::nullopt) {
+        return GetTargetFileSize(/*has_primary_key=*/false);
+    }
+    return impl_->blob_target_file_size.value();
+}
+
+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
+    // file maybe be less than target file generated by rolling file write).
+    return GetTargetFileSize(has_primary_key) / 10 * 7;
 }
 
 std::string CoreOptions::GetPartitionDefaultName() const {
@@ -594,6 +636,10 @@ int64_t CoreOptions::GetWriteBufferSize() const {
     return impl_->write_buffer_size;
 }
 
+bool CoreOptions::CommitForceCompact() const {
+    return impl_->commit_force_compact;
+}
+
 int64_t CoreOptions::GetCommitTimeout() const {
     return impl_->commit_timeout;
 }
@@ -602,6 +648,10 @@ int32_t CoreOptions::GetCommitMaxRetries() const {
     return impl_->commit_max_retries;
 }
 
+int32_t CoreOptions::GetCompactionMinFileNum() const {
+    return impl_->compaction_min_file_num;
+}
+
 const ExpireConfig& CoreOptions::GetExpireConfig() const {
     return impl_->expire_config;
 }
@@ -626,6 +676,10 @@ bool CoreOptions::IgnoreDelete() const {
     return impl_->ignore_delete;
 }
 
+bool CoreOptions::WriteOnly() const {
+    return impl_->write_only;
+}
+
 std::optional<std::string> CoreOptions::GetFieldsDefaultFunc() const {
     return impl_->field_default_func;
 }
@@ -674,6 +728,10 @@ bool CoreOptions::NeedLookup() const {
            impl_->force_lookup;
 }
 
+bool CoreOptions::CompactionForceRewriteAllFiles() const {
+    return impl_->compaction_force_rewrite_all_files;
+}
+
 std::map<std::string, std::string> CoreOptions::GetFieldsSequenceGroups() const {
     auto raw_options = impl_->raw_options;
     std::map<std::string, std::string> sequence_groups;

src/paimon/core/core_options.h

@@ -57,8 +57,9 @@ class PAIMON_EXPORT CoreOptions {
     const std::string& GetFileCompression() const;
     int32_t GetFileCompressionZstdLevel() const;
     int64_t GetPageSize() const;
-    int64_t GetTargetFileSize() const;
+    int64_t GetTargetFileSize(bool has_primary_key) const;
     int64_t GetBlobTargetFileSize() const;
+    int64_t GetCompactionFileSize(bool has_primary_key) const;
     std::string GetPartitionDefaultName() const;
 
     std::shared_ptr<FileFormat> GetManifestFormat() const;
@@ -78,14 +79,18 @@ class PAIMON_EXPORT CoreOptions {
 
     const ExpireConfig& GetExpireConfig() const;
 
+    bool CommitForceCompact() const;
+    bool CompactionForceRewriteAllFiles() const;
     int64_t GetCommitTimeout() const;
     int32_t GetCommitMaxRetries() const;
+    int32_t GetCompactionMinFileNum() const;
 
     const std::vector<std::string>& GetSequenceField() const;
     bool SequenceFieldSortOrderIsAscending() const;
     MergeEngine GetMergeEngine() const;
     SortEngine GetSortEngine() const;
     bool IgnoreDelete() const;
+    bool WriteOnly() const;
 
     std::optional<std::string> GetFieldsDefaultFunc() const;
     Result<std::optional<std::string>> GetFieldAggFunc(const std::string& field_name) const;

src/paimon/core/core_options_test.cpp

@@ -37,8 +37,12 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_TRUE(core_options.GetFileSystem());
     ASSERT_EQ(-1, core_options.GetBucket());
     ASSERT_EQ(64 * 1024L, core_options.GetPageSize());
-    ASSERT_EQ(256 * 1024 * 1024L, core_options.GetTargetFileSize());
+    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_EQ(187904815, core_options.GetCompactionFileSize(/*has_primary_key=*/false));
+    ASSERT_EQ(93952404, core_options.GetCompactionFileSize(/*has_primary_key=*/true));
+
     ASSERT_EQ("__DEFAULT_PARTITION__", core_options.GetPartitionDefaultName());
     ASSERT_EQ(std::nullopt, core_options.GetScanSnapshotId());
     ASSERT_EQ("zstd", core_options.GetFileCompression());
@@ -53,6 +57,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(1024, core_options.GetReadBatchSize());
     ASSERT_EQ(1024, core_options.GetWriteBatchSize());
     ASSERT_EQ(256 * 1024 * 1024, core_options.GetWriteBufferSize());
+    ASSERT_FALSE(core_options.CommitForceCompact());
     ASSERT_EQ(std::numeric_limits<int64_t>::max(), core_options.GetCommitTimeout());
     ASSERT_EQ(10, core_options.GetCommitMaxRetries());
     ExpireConfig expire_config = core_options.GetExpireConfig();
@@ -66,6 +71,9 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(MergeEngine::DEDUPLICATE, core_options.GetMergeEngine());
     ASSERT_EQ(SortEngine::LOSER_TREE, core_options.GetSortEngine());
     ASSERT_FALSE(core_options.IgnoreDelete());
+    ASSERT_FALSE(core_options.WriteOnly());
+    ASSERT_EQ(5, core_options.GetCompactionMinFileNum());
+    ASSERT_FALSE(core_options.CompactionForceRewriteAllFiles());
     ASSERT_EQ(std::nullopt, core_options.GetFieldsDefaultFunc());
     ASSERT_EQ(std::nullopt, core_options.GetFieldAggFunc("f0").value());
     ASSERT_FALSE(core_options.FieldAggIgnoreRetract("f1").value());
@@ -109,6 +117,7 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::READ_BATCH_SIZE, "2048"},
         {Options::WRITE_BUFFER_SIZE, "16MB"},
         {Options::WRITE_BATCH_SIZE, "1234"},
+        {Options::COMMIT_FORCE_COMPACT, "true"},
         {Options::COMMIT_TIMEOUT, "120s"},
         {Options::COMMIT_MAX_RETRIES, "20"},
         {Options::SCAN_SNAPSHOT_ID, "5"},
@@ -148,7 +157,9 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::GLOBAL_INDEX_ENABLED, "false"},
         {Options::GLOBAL_INDEX_EXTERNAL_PATH, "FILE:///tmp/global_index/"},
         {Options::SCAN_TAG_NAME, "test-tag"},
-    };
+        {Options::WRITE_ONLY, "true"},
+        {Options::COMPACTION_MIN_FILE_NUM, "10"},
+        {Options::COMPACTION_FORCE_REWRITE_ALL_FILES, "true"}};
 
     ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
     auto fs = core_options.GetFileSystem();
@@ -162,7 +173,8 @@ TEST(CoreOptionsTest, TestFromMap) {
 
     ASSERT_EQ(3, core_options.GetBucket());
     ASSERT_EQ(128 * 1024L, core_options.GetPageSize());
-    ASSERT_EQ(512 * 1024 * 1024L, core_options.GetTargetFileSize());
+    ASSERT_EQ(512 * 1024 * 1024L, core_options.GetTargetFileSize(/*has_primary_key=*/true));
+    ASSERT_EQ(512 * 1024 * 1024L, core_options.GetTargetFileSize(/*has_primary_key=*/false));
     ASSERT_EQ(1024 * 1024 * 1024L, core_options.GetBlobTargetFileSize());
     ASSERT_EQ("foo", core_options.GetPartitionDefaultName());
     ASSERT_EQ(16 * 1024 * 1024L, core_options.GetManifestTargetFileSize());
@@ -173,6 +185,7 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_EQ(2048, core_options.GetReadBatchSize());
     ASSERT_EQ(1234, core_options.GetWriteBatchSize());
     ASSERT_EQ(16 * 1024 * 1024, core_options.GetWriteBufferSize());
+    ASSERT_TRUE(core_options.CommitForceCompact());
     ASSERT_EQ(120 * 1000, core_options.GetCommitTimeout());
     ASSERT_EQ(20, core_options.GetCommitMaxRetries());
     ASSERT_EQ(5, core_options.GetScanSnapshotId().value_or(-1));
@@ -220,7 +233,12 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_EQ(core_options.GetGlobalIndexExternalPath().value(), "FILE:///tmp/global_index/");
     ASSERT_EQ("test-tag", core_options.GetScanTagName().value());
     ASSERT_EQ(StartupMode::FromSnapshot(), core_options.GetStartupMode());
-}
+    ASSERT_EQ(375809637, core_options.GetCompactionFileSize(/*has_primary_key=*/true));
+    ASSERT_EQ(375809637, core_options.GetCompactionFileSize(/*has_primary_key=*/false));
+    ASSERT_TRUE(core_options.WriteOnly());
+    ASSERT_EQ(10, core_options.GetCompactionMinFileNum());
+    ASSERT_TRUE(core_options.CompactionForceRewriteAllFiles());
+}  // namespace paimon::test
 
 TEST(CoreOptionsTest, TestInvalidCase) {
     ASSERT_NOK_WITH_MSG(CoreOptions::FromMap({{Options::BUCKET, "3.5"}}),

src/paimon/core/io/data_file_meta.h

@@ -136,8 +136,8 @@ struct DataFileMeta {
     std::vector<std::optional<std::string>> extra_files;
     Timestamp creation_time;
 
-    // row_count = addRowCount + deleteRowCount
-    // Why don't we keep addRowCount and deleteRowCount?
+    // row_count = add_row_count + delete_row_count
+    // Why don't we keep add_row_count and delete_row_count?
     // Because in previous versions of DataFileMeta, we only keep row_count.
     // We have to keep the compatibility.
     std::optional<int64_t> delete_row_count;

src/paimon/core/mergetree/merge_tree_writer.cpp

@@ -204,7 +204,7 @@ MergeTreeWriter::CreateRollingRowWriter() const {
         return writer;
     };
     return std::make_unique<RollingFileWriter<KeyValueBatch, std::shared_ptr<DataFileMeta>>>(
-        options_.GetTargetFileSize(), create_file_writer);
+        options_.GetTargetFileSize(/*has_primary_key=*/true), create_file_writer);
 }
 
 Result<int64_t> MergeTreeWriter::EstimateMemoryUse(const std::shared_ptr<arrow::Array>& array) {

src/paimon/core/postpone/postpone_bucket_writer.cpp

@@ -261,7 +261,7 @@ PostponeBucketWriter::CreateRollingRowWriter() const {
         return writer;
     };
     return std::make_unique<RollingFileWriter<KeyValueBatch, std::shared_ptr<DataFileMeta>>>(
-        options_.GetTargetFileSize(), create_file_writer);
+        options_.GetTargetFileSize(/*has_primary_key=*/true), create_file_writer);
 }
 
 Status PostponeBucketWriter::Flush() {