Merge branch 'main' into main

Author: dalingmeng

src/paimon/common/data/blob.cpp

@@ -33,19 +33,6 @@ namespace paimon {
 
 class MemoryPool;
 
-Result<std::unique_ptr<Blob>> Blob::FromPath(const std::string& path) {
-    return FromPath(path, /*offset=*/0, /*length=*/-1);
-}
-
-Result<std::unique_ptr<Blob>> Blob::FromPath(const std::string& path, int64_t offset,
-                                             int64_t length) {
-    PAIMON_ASSIGN_OR_RAISE(std::string normalized_path, PathUtil::NormalizePath(path));
-    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BlobDescriptor> descriptor,
-                           BlobDescriptor::Create(normalized_path, offset, length));
-    auto impl = std::make_unique<Blob::Impl>(std::move(descriptor), descriptor->Uri());
-    return std::unique_ptr<Blob>(new Blob(std::move(impl)));
-}
-
 class Blob::Impl {
  public:
     Impl(std::unique_ptr<BlobDescriptor>&& descriptor, const std::string& uri)
@@ -68,6 +55,19 @@ class Blob::Impl {
     std::string uri_;
 };
 
+Result<std::unique_ptr<Blob>> Blob::FromPath(const std::string& path) {
+    return FromPath(path, /*offset=*/0, /*length=*/-1);
+}
+
+Result<std::unique_ptr<Blob>> Blob::FromPath(const std::string& path, int64_t offset,
+                                             int64_t length) {
+    PAIMON_ASSIGN_OR_RAISE(std::string normalized_path, PathUtil::NormalizePath(path));
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BlobDescriptor> descriptor,
+                           BlobDescriptor::Create(normalized_path, offset, length));
+    auto impl = std::make_unique<Blob::Impl>(std::move(descriptor), descriptor->Uri());
+    return std::unique_ptr<Blob>(new Blob(std::move(impl)));
+}
+
 Blob::Blob(std::unique_ptr<Impl>&& impl) : impl_(std::move(impl)) {}
 Blob::~Blob() = default;
 

src/paimon/common/memory/memory_pool.cpp

@@ -55,7 +55,7 @@ void* MemoryPoolImpl::Malloc(uint64_t size, uint64_t alignment) {
     return memptr;
 }
 
-void* MemoryPoolImpl::Realloc(void* p, size_t old_size, size_t new_size, size_t alignment) {
+void* MemoryPoolImpl::Realloc(void* p, size_t old_size, size_t new_size, uint64_t alignment) {
     if (alignment == 0) {
         void* memptr = ::realloc(p, new_size);
         total_allocated_size.fetch_add(new_size - old_size);

src/paimon/core/deletionvectors/deletion_vectors_index_file_test.cpp

@@ -36,7 +36,8 @@ TEST(DeletionVectorsIndexFileTest, Basic) {
                          FileSystemFactory::Get("local", dir->Str(), {}));
     auto path_factory = std::make_shared<MockIndexPathFactory>(dir->Str());
     auto pool = GetDefaultPool();
-    DeletionVectorsIndexFile index_file(fs, path_factory, /*bitmap64=*/false, pool);
+    auto index_file =
+        std::make_shared<DeletionVectorsIndexFile>(fs, path_factory, /*bitmap64=*/false, pool);
 
     std::map<std::string, std::shared_ptr<DeletionVector>> input;
     RoaringBitmap32 roaring_1;
@@ -50,15 +51,18 @@ TEST(DeletionVectorsIndexFileTest, Basic) {
     }
     input["dv2"] = std::make_shared<BitmapDeletionVector>(roaring_2);
 
-    ASSERT_FALSE(index_file.Bitmap64());
-    ASSERT_OK_AND_ASSIGN(auto meta, index_file.WriteSingleFile(input));
+    ASSERT_FALSE(index_file->Bitmap64());
+    ASSERT_OK_AND_ASSIGN(auto meta, index_file->WriteSingleFile(input));
     ASSERT_GT(meta->FileSize(), 0);
+    ASSERT_OK_AND_ASSIGN(auto size, index_file->FileSize(meta));
+    ASSERT_EQ(meta->FileSize(), size);
     ASSERT_EQ(meta->IndexType(), DeletionVectorsIndexFile::DELETION_VECTORS_INDEX);
     ASSERT_EQ(meta->FileName(), "index-0");
+    ASSERT_FALSE(index_file->IsExternalPath());
     ASSERT_EQ(meta->ExternalPath(), std::nullopt);
 
     // Round trip: write then read all deletion vectors from index file.
-    ASSERT_OK_AND_ASSIGN(auto read_back, index_file.ReadAllDeletionVectors(meta));
+    ASSERT_OK_AND_ASSIGN(auto read_back, index_file->ReadAllDeletionVectors(meta));
     ASSERT_EQ(read_back.size(), input.size());
     ASSERT_EQ(read_back.at("dv1")->GetCardinality(), 10);
     ASSERT_EQ(read_back.at("dv2")->GetCardinality(), 10);

src/paimon/core/global_index/global_index_write_task.cpp

@@ -79,6 +79,7 @@ Result<std::unique_ptr<BatchReader>> CreateBatchReader(
     const std::shared_ptr<MemoryPool>& pool) {
     ReadContextBuilder read_context_builder(table_path);
     read_context_builder.SetOptions(core_options.ToMap())
+        .WithFileSystem(core_options.GetFileSystem())
         .EnablePrefetch(true)
         .WithMemoryPool(pool)
         .SetReadSchema({field_name});

src/paimon/core/mergetree/compact/aggregate/field_listagg_agg.h

@@ -79,7 +79,7 @@ class FieldListaggAgg : public FieldAggregator {
             new_result.append(in_str);
             result_ = std::move(new_result);
         }
-        return std::string_view(result_);
+        return std::string_view{result_};
     }
 
  private:

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

@@ -65,21 +65,23 @@ LookupMergeTreeCompactRewriter<T>::Create(
     auto write_schema = SpecialFields::CompleteSequenceAndValueKindField(data_schema);
 
     // TODO(xinyu.lxy): set executor
-    // TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
-    // usage during compaction. Will fix via parquet format refactor.
     ReadContextBuilder read_context_builder(path_factory_cache->RootPath());
     read_context_builder.SetOptions(options.ToMap())
+        .WithFileSystem(options.GetFileSystem())
         .EnablePrefetch(true)
         .SetPrefetchMaxParallelNum(1)
         .SetPrefetchBatchCount(3)
-        .WithMemoryPool(pool)
-        .AddOption("parquet.read.enable-pre-buffer", "false");
+        .WithMemoryPool(pool);
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ReadContext> read_context,
                            read_context_builder.Finish());
-
-    PAIMON_ASSIGN_OR_RAISE(
-        std::shared_ptr<InternalReadContext> internal_context,
-        InternalReadContext::Create(read_context, table_schema, options.ToMap()));
+    // TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
+    // usage during compaction. Will fix via parquet format refactor.
+    auto new_options = options.ToMap();
+    if (new_options.find("parquet.read.enable-pre-buffer") == new_options.end()) {
+        new_options["parquet.read.enable-pre-buffer"] = "false";
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalReadContext> internal_context,
+                           InternalReadContext::Create(read_context, table_schema, new_options));
     PAIMON_ASSIGN_OR_RAISE(
         std::shared_ptr<FileStorePathFactory> path_factory,
         path_factory_cache->GetOrCreatePathFactory(options.GetFileFormat()->Identifier()));

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

@@ -71,21 +71,23 @@ Result<std::unique_ptr<MergeTreeCompactRewriter>> MergeTreeCompactRewriter::Crea
     auto write_schema = SpecialFields::CompleteSequenceAndValueKindField(data_schema);
 
     // TODO(xinyu.lxy): set executor
-    // TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
-    // usage during compaction. Will fix via parquet format refactor.
     ReadContextBuilder read_context_builder(path_factory_cache->RootPath());
     read_context_builder.SetOptions(options.ToMap())
+        .WithFileSystem(options.GetFileSystem())
         .EnablePrefetch(true)
         .SetPrefetchMaxParallelNum(1)
         .SetPrefetchBatchCount(3)
-        .WithMemoryPool(pool)
-        .AddOption("parquet.read.enable-pre-buffer", "false");
+        .WithMemoryPool(pool);
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ReadContext> read_context,
                            read_context_builder.Finish());
-
-    PAIMON_ASSIGN_OR_RAISE(
-        std::shared_ptr<InternalReadContext> internal_context,
-        InternalReadContext::Create(read_context, table_schema, options.ToMap()));
+    // TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
+    // usage during compaction. Will fix via parquet format refactor.
+    auto new_options = options.ToMap();
+    if (new_options.find("parquet.read.enable-pre-buffer") == new_options.end()) {
+        new_options["parquet.read.enable-pre-buffer"] = "false";
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalReadContext> internal_context,
+                           InternalReadContext::Create(read_context, table_schema, new_options));
     PAIMON_ASSIGN_OR_RAISE(
         std::shared_ptr<FileStorePathFactory> path_factory,
         path_factory_cache->GetOrCreatePathFactory(options.GetFileFormat()->Identifier()));

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

@@ -165,6 +165,38 @@ TEST_F(MergeTreeCompactRewriterTest, TestSimple) {
     CheckResult(compact_file_name, fs, table_schema, expected_array);
 }
 
+TEST_F(MergeTreeCompactRewriterTest, TestCancel) {
+    std::string origin_table_path = GetDataDir() + "/orc/pk_table_scan_and_read_mor.db/";
+    auto table_dir = UniqueTestDirectory::Create("local");
+    ASSERT_TRUE(TestUtil::CopyDirectory(origin_table_path, table_dir->Str()));
+    std::string table_path = table_dir->Str() + "/pk_table_scan_and_read_mor";
+    auto fs = table_dir->GetFileSystem();
+
+    SchemaManager schema_manager(fs, table_path);
+    ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+    ASSERT_OK_AND_ASSIGN(auto options, CoreOptions::FromMap(table_schema->Options()));
+    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+    auto dv_factory = [](const std::string&) -> Result<std::shared_ptr<DeletionVector>> {
+        return std::shared_ptr<DeletionVector>();
+    };
+    auto cancellation_controller = std::make_shared<CancellationController>();
+    auto path_factory_cache =
+        std::make_shared<FileStorePathFactoryCache>(table_path, table_schema, options, pool_);
+    ASSERT_OK_AND_ASSIGN(
+        auto rewriter,
+        MergeTreeCompactRewriter::Create(
+            /*bucket=*/1, /*partition=*/BinaryRowGenerator::GenerateRow({10}, pool_.get()),
+            table_schema, dv_factory, path_factory_cache, options, cancellation_controller, pool_));
+
+    ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns(table_path, table_schema, /*bucket=*/1,
+                                                       /*partition=*/{{"f1", "10"}}));
+
+    // cancel compaction here
+    cancellation_controller->Cancel();
+    ASSERT_NOK_WITH_MSG(rewriter->Rewrite(/*output_level=*/5, /*drop_delete=*/true, runs),
+                        "Compaction is cancelled");
+}
+
 TEST_F(MergeTreeCompactRewriterTest, TestNotDropDelete) {
     std::string origin_table_path = GetDataDir() + "/orc/pk_table_scan_and_read_mor.db/";
     auto table_dir = UniqueTestDirectory::Create("local");

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

@@ -92,7 +92,7 @@ Status MergeTreeCompactTask::Rewrite(std::vector<std::vector<SortedRun>>* candid
     if (candidate->size() == 1) {
         const auto& section = (*candidate)[0];
         if (section.empty()) {
-            return Status::OK();
+            return Status::Invalid("invalid section, section cannot be empty in candidate");
         }
         if (section.size() == 1) {
             for (const auto& file : section[0].Files()) {

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

@@ -265,6 +265,18 @@ PartialUpdateMergeFunction::CreateFieldAggregators(
     return aggregators;
 }
 
+void PartialUpdateMergeFunction::Reset() {
+    current_key_.reset();
+    meet_insert_ = false;
+    not_null_column_filled_ = false;
+    row_ = std::make_unique<GenericRow>(getters_.size());
+    last_seq_num_ = 0;
+    for (auto& [_, agg] : field_aggregators_) {
+        assert(agg);
+        agg->Reset();
+    }
+}
+
 Status PartialUpdateMergeFunction::Add(KeyValue&& moved_kv) {
     // refresh key object to avoid reference overwritten
     KeyValue kv = std::move(moved_kv);
@@ -275,33 +287,50 @@ Status PartialUpdateMergeFunction::Add(KeyValue&& moved_kv) {
         // In 0.7- versions, the delete records might be written into data file even when
         // ignore-delete configured, so ignoreDelete still needs to be checked
         if (ignore_delete_) {
+            if (!not_null_column_filled_) {
+                InitRowAndHoldData(std::move(kv.value));
+                not_null_column_filled_ = true;
+            }
             return Status::OK();
         }
+
+        last_seq_num_ = kv.sequence_number;
+
         if (field_sequence_enabled_) {
+            // RetractWithSequenceGroup handles InitRow and AddDataHolder internally
             PAIMON_RETURN_NOT_OK(RetractWithSequenceGroup(std::move(kv)));
             return Status::OK();
         }
         if (remove_record_on_delete_) {
             if (kv.value_kind == RowKind::Delete()) {
                 current_delete_row_ = true;
                 row_ = std::make_unique<GenericRow>(getters_.size());
+                InitRowAndHoldData(std::move(kv.value));
+            } else if (!not_null_column_filled_) {
+                InitRowAndHoldData(std::move(kv.value));
+                not_null_column_filled_ = true;
             }
             return Status::OK();
         }
 
         return Status::Invalid(
             "By default, Partial update can not accept delete records, you can choose one of "
-            "the following solutions:1. Configure ignore-delete to ignore delete records; 2. "
-            "Configure partial-update.remove-record-on-delete to remove the whole row when "
-            "receiving delete records; 3. Configure sequence-group to retract partial "
-            "columns.");
+            "the following solutions:\n"
+            "1. Configure 'ignore-delete' to ignore delete records.\n"
+            "2. Configure 'partial-update.remove-record-on-delete' to remove the whole row "
+            "when receiving delete records.\n"
+            "3. Configure 'sequence-group's to retract partial columns. Also configure "
+            "'partial-update.remove-record-on-sequence-group' to remove the whole row when "
+            "receiving deleted records of specified sequence group.");
     }
     last_seq_num_ = kv.sequence_number;
     if (field_comparators_.empty()) {
         UpdateNonNullFields(std::move(kv));
     } else {
         UpdateWithSequenceGroup(std::move(kv));
     }
+    meet_insert_ = true;
+    not_null_column_filled_ = true;
     return Status::OK();
 }
 
@@ -356,6 +385,15 @@ void PartialUpdateMergeFunction::UpdateWithSequenceGroup(KeyValue&& kv) {
 }
 
 Status PartialUpdateMergeFunction::RetractWithSequenceGroup(KeyValue&& kv) {
+    // Initialize row with all field values if this is the first record.
+    // InitRow only reads field values (string_view etc.) from kv.value without taking ownership.
+    // kv.value remains alive throughout this method, so the views are safe until AddDataHolder
+    // at the end transfers ownership to row_.
+    if (!not_null_column_filled_) {
+        InitRow(*(kv.value));
+        not_null_column_filled_ = true;
+    }
+
     std::set<int32_t> updated_sequence_fields;
     for (size_t i = 0; i < getters_.size(); ++i) {
         auto cmp_iter = field_comparators_.find(i);
@@ -382,6 +420,7 @@ Status PartialUpdateMergeFunction::RetractWithSequenceGroup(KeyValue&& kv) {
                                     sequence_group_partial_delete_.end()) {
                                 current_delete_row_ = true;
                                 row_ = std::make_unique<GenericRow>(getters_.size());
+                                InitRowAndHoldData(std::move(kv.value));
                                 return Status::OK();
                             } else {
                                 row_->SetField(field_idx, getters_[field_idx](*(kv.value)));
@@ -413,6 +452,18 @@ Status PartialUpdateMergeFunction::RetractWithSequenceGroup(KeyValue&& kv) {
     return Status::OK();
 }
 
+void PartialUpdateMergeFunction::InitRow(const InternalRow& value) {
+    for (size_t i = 0; i < getters_.size(); ++i) {
+        VariantType field = getters_[i](value);
+        row_->SetField(i, field);
+    }
+}
+
+void PartialUpdateMergeFunction::InitRowAndHoldData(std::unique_ptr<InternalRow>&& value) {
+    InitRow(*value);
+    row_->AddDataHolder(std::move(value));
+}
+
 bool PartialUpdateMergeFunction::IsEmptySequenceGroup(
     const KeyValue& kv, const std::shared_ptr<FieldsComparator>& comparator) const {
     for (const auto& field_idx : comparator->CompareFields()) {