feat(compact): support remote lookup file & add DropFileCallback in Levels (#214)

Author: lxy-9602

include/paimon/defs.h

@@ -384,6 +384,13 @@ struct PAIMON_EXPORT Options {
     /// "lookup.cache.bloom.filter.fpp" - Define the default false positive probability for lookup
     /// cache bloom filters. Default value is 0.05.
     static const char LOOKUP_CACHE_BLOOM_FILTER_FPP[];
+    /// "lookup.remote-file.enabled" - Whether to enable the remote file for lookup.
+    /// Default value is false.
+    static const char LOOKUP_REMOTE_FILE_ENABLED[];
+    /// "lookup.remote-file.level-threshold" - Level threshold of lookup to generate remote lookup
+    /// files. Level files below this threshold will not generate remote lookup files.
+    /// Default value is INT32_MIN.
+    static const char LOOKUP_REMOTE_LEVEL_THRESHOLD[];
     /// "lookup.cache-spill-compression" - Spill compression for lookup cache, currently zstd, none,
     /// lz4 are supported. Default value is zstd.
     /// Noted that java paimon also supports lzo which paimon-cpp does not support for now.

src/paimon/CMakeLists.txt

@@ -244,6 +244,7 @@ set(PAIMON_CORE_SRCS
     core/mergetree/write_buffer.cpp
     core/mergetree/levels.cpp
     core/mergetree/lookup_levels.cpp
+    core/mergetree/lookup/remote_lookup_file_manager.cpp
     core/migrate/file_meta_utils.cpp
     core/operation/data_evolution_file_store_scan.cpp
     core/operation/data_evolution_split_read.cpp
@@ -596,6 +597,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/compact/merge_tree_compact_manager_factory_test.cpp
                     core/mergetree/compact/merge_tree_compact_rewriter_test.cpp
                     core/mergetree/lookup/persist_processor_test.cpp
+                    core/mergetree/lookup/remote_lookup_file_manager_test.cpp
                     core/mergetree/drop_delete_reader_test.cpp
                     core/mergetree/merge_tree_writer_test.cpp
                     core/mergetree/write_buffer_test.cpp

src/paimon/common/defs.cpp

@@ -98,6 +98,8 @@ const char Options::COMPACT_OFFPEAK_END_HOUR[] = "compaction.offpeak.end.hour";
 const char Options::COMPACTION_OFFPEAK_RATIO[] = "compaction.offpeak-ratio";
 const char Options::LOOKUP_CACHE_BLOOM_FILTER_ENABLED[] = "lookup.cache.bloom.filter.enabled";
 const char Options::LOOKUP_CACHE_BLOOM_FILTER_FPP[] = "lookup.cache.bloom.filter.fpp";
+const char Options::LOOKUP_REMOTE_FILE_ENABLED[] = "lookup.remote-file.enabled";
+const char Options::LOOKUP_REMOTE_LEVEL_THRESHOLD[] = "lookup.remote-file.level-threshold";
 const char Options::LOOKUP_CACHE_SPILL_COMPRESSION[] = "lookup.cache-spill-compression";
 const char Options::SPILL_COMPRESSION_ZSTD_LEVEL[] = "spill-compression.zstd-level";
 const char Options::CACHE_PAGE_SIZE[] = "cache-page-size";

src/paimon/core/core_options.cpp

@@ -420,6 +420,8 @@ struct CoreOptions::Impl {
     int32_t compact_off_peak_ratio = 0;
     bool lookup_cache_bloom_filter = true;
     double lookup_cache_bloom_filter_fpp = 0.05;
+    bool lookup_remote_file_enabled = false;
+    int32_t lookup_remote_level_threshold = INT32_MIN;
     CompressOptions lookup_compress_options{"zstd", 1};
     int64_t cache_page_size = 64 * 1024;  // 64KB
     std::map<int32_t, std::shared_ptr<FileFormat>> file_format_per_level;
@@ -676,6 +678,14 @@ Result<CoreOptions> CoreOptions::FromMap(
     PAIMON_RETURN_NOT_OK(parser.Parse<double>(Options::LOOKUP_CACHE_BLOOM_FILTER_FPP,
                                               &impl->lookup_cache_bloom_filter_fpp));
 
+    // Parse lookup.remote-file.enabled
+    PAIMON_RETURN_NOT_OK(
+        parser.Parse<bool>(Options::LOOKUP_REMOTE_FILE_ENABLED, &impl->lookup_remote_file_enabled));
+
+    // Parse lookup.remote-file.level-threshold
+    PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::LOOKUP_REMOTE_LEVEL_THRESHOLD,
+                                               &impl->lookup_remote_level_threshold));
+
     // Parse lookup.cache-spill-compression
     std::string lookup_compress_options_compression_str;
     PAIMON_RETURN_NOT_OK(parser.ParseString(Options::LOOKUP_CACHE_SPILL_COMPRESSION,
@@ -1187,6 +1197,14 @@ const CompressOptions& CoreOptions::GetLookupCompressOptions() const {
     return impl_->lookup_compress_options;
 }
 
+bool CoreOptions::LookupRemoteFileEnabled() const {
+    return impl_->lookup_remote_file_enabled;
+}
+
+int32_t CoreOptions::GetLookupRemoteLevelThreshold() const {
+    return impl_->lookup_remote_level_threshold;
+}
+
 int32_t CoreOptions::GetCachePageSize() const {
     return static_cast<int32_t>(impl_->cache_page_size);
 }

src/paimon/core/core_options.h

@@ -154,6 +154,9 @@ class PAIMON_EXPORT CoreOptions {
     bool LookupCacheBloomFilterEnabled() const;
     double GetLookupCacheBloomFilterFpp() const;
 
+    bool LookupRemoteFileEnabled() const;
+    int32_t GetLookupRemoteLevelThreshold() const;
+
     const CompressOptions& GetLookupCompressOptions() const;
     int32_t GetCachePageSize() const;
 

src/paimon/core/core_options_test.cpp

@@ -128,6 +128,8 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(10, core_options.GetLookupCompactMaxInterval());
     ASSERT_EQ(256 * 1024 * 1024, core_options.GetLookupCacheMaxMemory());
     ASSERT_EQ(0.25, core_options.GetLookupCacheHighPrioPoolRatio());
+    ASSERT_FALSE(core_options.LookupRemoteFileEnabled());
+    ASSERT_EQ(core_options.GetLookupRemoteLevelThreshold(), INT32_MIN);
 }
 
 TEST(CoreOptionsTest, TestFromMap) {
@@ -214,7 +216,9 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::FILE_FORMAT_PER_LEVEL, "0:AVRO,3:parquet"},
         {Options::FILE_COMPRESSION_PER_LEVEL, "0:lz4,3:none"},
         {Options::LOOKUP_CACHE_MAX_MEMORY_SIZE, "1MB"},
-        {Options::LOOKUP_CACHE_HIGH_PRIO_POOL_RATIO, "0.35"}};
+        {Options::LOOKUP_CACHE_HIGH_PRIO_POOL_RATIO, "0.35"},
+        {Options::LOOKUP_REMOTE_FILE_ENABLED, "True"},
+        {Options::LOOKUP_REMOTE_LEVEL_THRESHOLD, "2"}};
 
     ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
     auto fs = core_options.GetFileSystem();
@@ -327,6 +331,8 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_EQ(6 * 1024 * 1024, core_options.GetCachePageSize());
     ASSERT_EQ(1024 * 1024, core_options.GetLookupCacheMaxMemory());
     ASSERT_EQ(0.35, core_options.GetLookupCacheHighPrioPoolRatio());
+    ASSERT_TRUE(core_options.LookupRemoteFileEnabled());
+    ASSERT_EQ(core_options.GetLookupRemoteLevelThreshold(), 2);
 }
 
 TEST(CoreOptionsTest, TestInvalidCase) {

src/paimon/core/io/data_file_meta.cpp

@@ -85,6 +85,15 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta::Upgrade(int32_t new_level) c
         first_row_id, write_cols);
 }
 
+std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithExtraFiles(
+    const std::vector<std::optional<std::string>>& new_extra_files) const {
+    return std::make_shared<DataFileMeta>(
+        file_name, file_size, row_count, min_key, max_key, key_stats, value_stats,
+        min_sequence_number, max_sequence_number, schema_id, level, new_extra_files, creation_time,
+        delete_row_count, embedded_index, file_source, value_stats_cols, external_path,
+        first_row_id, write_cols);
+}
+
 DataFileMeta::DataFileMeta(
     const std::string& _file_name, int64_t _file_size, int64_t _row_count,
     const BinaryRow& _min_key, const BinaryRow& _max_key, const SimpleStats& _key_stats,

src/paimon/core/io/data_file_meta.h

@@ -77,6 +77,10 @@ struct DataFileMeta {
 
     Result<std::shared_ptr<DataFileMeta>> Upgrade(int32_t new_level) const;
 
+    /// Create a copy of this DataFileMeta with the given extra files.
+    std::shared_ptr<DataFileMeta> CopyWithExtraFiles(
+        const std::vector<std::optional<std::string>>& new_extra_files) const;
+
     std::optional<int64_t> AddRowCount() const {
         return delete_row_count == std::nullopt ? std::optional<int64_t>()
                                                 : row_count - delete_row_count.value();

src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.cpp

@@ -115,9 +115,8 @@ Result<CompactResult> ChangelogMergeTreeRewriter::RewriteOrProduceChangelog(
     if (rewrite_compact_file) {
         NotifyRewriteCompactBefore(before);
     }
+    PAIMON_ASSIGN_OR_RAISE(after, NotifyRewriteCompactAfter(after));
     write_guard.Release();
-
-    after = NotifyRewriteCompactAfter(after);
     return CompactResult(before, after);
 }
 

src/paimon/core/mergetree/compact/lookup_merge_tree_compact_rewriter.cpp

@@ -35,15 +35,17 @@ LookupMergeTreeCompactRewriter<T>::LookupMergeTreeCompactRewriter(
     std::unique_ptr<MergeFileSplitRead>&& merge_file_split_read,
     MergeFunctionWrapperFactory merge_function_wrapper_factory,
     const std::shared_ptr<MemoryPool>& pool,
-    const std::shared_ptr<CancellationController>& cancellation_controller)
+    const std::shared_ptr<CancellationController>& cancellation_controller,
+    std::unique_ptr<RemoteLookupFileManager<T>>&& remote_lookup_file_manager)
     : ChangelogMergeTreeRewriter(
           max_level, /*force_drop_delete=*/dv_maintainer != nullptr, partition, bucket, schema_id,
           trimmed_primary_keys, options, data_schema, write_schema,
           DeletionVector::CreateFactory(dv_maintainer), path_factory_cache,
           std::move(merge_file_split_read), std::move(merge_function_wrapper_factory), pool,
           cancellation_controller),
       lookup_levels_(std::move(lookup_levels)),
-      dv_maintainer_(dv_maintainer) {}
+      dv_maintainer_(dv_maintainer),
+      remote_lookup_file_manager_(std::move(remote_lookup_file_manager)) {}
 
 template <typename T>
 Result<std::unique_ptr<LookupMergeTreeCompactRewriter<T>>>
@@ -54,7 +56,8 @@ LookupMergeTreeCompactRewriter<T>::Create(
     const BinaryRow& partition, const std::shared_ptr<TableSchema>& table_schema,
     const std::shared_ptr<FileStorePathFactoryCache>& path_factory_cache,
     const CoreOptions& options, const std::shared_ptr<MemoryPool>& pool,
-    const std::shared_ptr<CancellationController>& cancellation_controller) {
+    const std::shared_ptr<CancellationController>& cancellation_controller,
+    std::unique_ptr<RemoteLookupFileManager<T>>&& remote_lookup_file_manager) {
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> trimmed_primary_keys,
                            table_schema->TrimmedPrimaryKeys());
     auto data_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
@@ -86,7 +89,7 @@ LookupMergeTreeCompactRewriter<T>::Create(
         std::move(lookup_levels), dv_maintainer, max_level, partition, bucket, table_schema->Id(),
         trimmed_primary_keys, options, data_schema, write_schema, path_factory_cache,
         std::move(merge_file_split_read), std::move(merge_function_wrapper_factory), pool,
-        cancellation_controller));
+        cancellation_controller, std::move(remote_lookup_file_manager)));
 }
 
 template <typename T>
@@ -169,11 +172,21 @@ void LookupMergeTreeCompactRewriter<T>::NotifyRewriteCompactBefore(
 }
 
 template <typename T>
-std::vector<std::shared_ptr<DataFileMeta>>
+Result<std::vector<std::shared_ptr<DataFileMeta>>>
 LookupMergeTreeCompactRewriter<T>::NotifyRewriteCompactAfter(
     const std::vector<std::shared_ptr<DataFileMeta>>& files) {
-    // TODO(xinyu.lxy): support remoteLookupFileManager
-    return files;
+    if (!remote_lookup_file_manager_) {
+        return files;
+    }
+
+    std::vector<std::shared_ptr<DataFileMeta>> result;
+    result.reserve(files.size());
+    for (const auto& file : files) {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> new_meta,
+                               remote_lookup_file_manager_->GenRemoteLookupFile(file));
+        result.push_back(std::move(new_meta));
+    }
+    return result;
 }
 
 template class LookupMergeTreeCompactRewriter<KeyValue>;