SteNicholas commented on code in PR #449: URL: https://github.com/apache/paimon-cpp/pull/449#discussion_r4361381188
########## src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.cpp: ########## @@ -0,0 +1,312 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h" + +#include <cstdint> +#include <limits> +#include <map> +#include <string> +#include <unordered_map> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/casting/casting_utils.h" +#include "paimon/core/deletionvectors/deletion_vectors_index_file.h" +#include "paimon/core/operation/internal_read_context.h" +#include "paimon/core/operation/raw_file_split_read.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/read_context.h" +#include "paimon/utils/range.h" + +namespace paimon { +namespace { + +Status CollectDeletedOffsets(const ManifestEntry& source, const std::shared_ptr<DeletionVector>& dv, + const std::shared_ptr<arrow::Schema>& offset_schema, + const std::shared_ptr<MemoryPool>& pool, RawFileSplitRead* read, + RoaringBitmap64* deleted_offsets) { + std::vector<Range> ranges; + bool invalid_position = false; + PAIMON_RETURN_NOT_OK(dv->ForEachDeletedPosition([&](int64_t position) { + if (position < 0 || position >= source.File()->row_count || + position >= std::numeric_limits<int32_t>::max()) { + invalid_position = true; + } else if (!ranges.empty() && position == ranges.back().to + 1) { + ranges.back().to = position; + } else { + ranges.emplace_back(position, position); + } + })); + if (invalid_position) { + return Status::Invalid("compaction input DV contains an invalid row position"); + } + if (ranges.empty()) { + return Status::OK(); + } + ranges = Range::SortAndMergeOverlap(ranges, /*adjacent=*/true); + int64_t expected_rows = 0; + for (const Range& range : ranges) { + expected_rows += range.Count(); + } + // Read deleted physical rows to recover their stable offsets. Applying the input DV here + // would hide precisely the rows whose deletions need to be transferred to the output. + DeletionVector::Factory no_deletions = + [](const std::string&) -> Result<std::shared_ptr<DeletionVector>> { + return std::shared_ptr<DeletionVector>(); + }; + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<BatchReader> reader, + read->CreateReader(source.Partition(), source.Bucket(), {source.File()}, no_deletions, + ranges, offset_schema, /*predicate=*/nullptr)); + ScopeGuard reader_guard([&]() { reader->Close(); }); + const std::shared_ptr<arrow::MemoryPool> arrow_pool = GetArrowPool(pool); + int64_t emitted_rows = 0; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap, + reader->NextBatchWithBitmap()); + if (BatchReader::IsEofBatch(batch_with_bitmap)) { + break; + } + auto& [batch, selected_rows] = batch_with_bitmap; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Array> array, + arrow::ImportArray(batch.first.get(), batch.second.get())); + if (array->type_id() != arrow::Type::STRUCT) { + return Status::Invalid("compaction offset reader returned a non-struct array"); + } + std::shared_ptr<arrow::Array> raw_offsets = + checked_pointer_cast<arrow::StructArray>(array)->GetFieldByName( + SpecialFields::RealtimeOffset().Name()); + if (raw_offsets && raw_offsets->type_id() == arrow::Type::DICTIONARY) { + PAIMON_ASSIGN_OR_RAISE( + raw_offsets, + CastingUtils::Cast(raw_offsets, arrow::int64(), arrow::compute::CastOptions::Safe(), + arrow_pool.get())); + } + if (!raw_offsets || raw_offsets->type_id() != arrow::Type::INT64) { + return Status::Invalid("compaction input has no valid realtime offsets"); + } + const auto offsets = checked_pointer_cast<arrow::Int64Array>(raw_offsets); + // The raw batch may contain unselected rows. Only selected offsets participate in + // migration and row-count validation; do not materialize a filtered Arrow array. + emitted_rows += selected_rows.Cardinality(); + for (auto selected = selected_rows.Begin(); selected != selected_rows.End(); ++selected) { + const int64_t row = *selected; + if (row >= offsets->length()) { + return Status::Invalid("compaction selected row is out of bounds"); + } + if (offsets->IsNull(row)) { + return Status::Invalid("compaction input contains a null realtime offset"); + } + if (offsets->Value(row) < 0) { + return Status::Invalid("compaction input contains a negative realtime offset"); + } + deleted_offsets->Add(offsets->Value(row)); + } + } + if (emitted_rows != expected_rows) { + return Status::Invalid("compaction offset reader did not return all deleted rows"); + } + return Status::OK(); +} + +} // namespace + +Result<RealtimeCompactionPlan> RealtimeCompactionPlan::Create( + const std::vector<ManifestEntry>& entries) { + std::unordered_map<FileEntry::Identifier, const ManifestEntry*> inputs; + std::unordered_map<FileEntry::Identifier, const ManifestEntry*> outputs; + for (const ManifestEntry& entry : entries) { + if (!entry.File() || entry.File()->row_count < 0) { + return Status::Invalid("invalid realtime compaction file"); + } + if (!(entry.Kind() == FileKind::Add()) && !(entry.Kind() == FileKind::Delete())) { + return Status::Invalid("invalid realtime compaction entry kind"); + } + auto& files = entry.Kind() == FileKind::Add() ? outputs : inputs; + if (!files.emplace(entry.CreateIdentifier(), &entry).second) { + return Status::Invalid("duplicate realtime compaction input or output: ", + entry.FileName()); + } + } + // For F1/F2 -> F3 -> F4, delta_files contains DELETE F1/F2 and ADD F4. + // intermediate_files contains ADD F3, whose matching DELETE cancels it from the final changes. + RealtimeCompactionPlan plan; + for (const ManifestEntry& entry : entries) { + const auto id = entry.CreateIdentifier(); + if (entry.Kind() == FileKind::Delete()) { + if (outputs.count(id) == 0) { + plan.delta_files.push_back(entry); + } else { + const auto& output = outputs.at(id)->File(); + if (entry.File()->row_count != output->row_count || + entry.File()->schema_id != output->schema_id || + entry.File()->file_size != output->file_size) { + return Status::Invalid("inconsistent intermediate compaction file: ", + entry.FileName()); + } + } + } else if (inputs.count(id) == 0) { + plan.delta_files.push_back(entry); + } else { + plan.intermediate_files.push_back(entry); Review Comment: @lxy-9602, clean up intermediate compaction outputs. For a chain `F1/F2 -> F3 -> F4`, normalization removes F3 from the published manifest entries, but `intermediate_files` is only used for validation and never cleaned up. `PrepareCommit()` has already cleared the writer's `compact_after_` through `DrainIncrement()`, so `Close()` cannot delete F3 either, and snapshot expiration cannot discover it because no snapshot references it. Each chained compaction therefore leaves the intermediate data file and its `.offset`/index sidecars orphaned. Please add cleanup after successful publication and extend the chained-compaction test to verify that these intermediate files are removed while final outputs remain intact. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
