wangyong9999 commented on code in PR #227:
URL: https://github.com/apache/paimon-cpp/pull/227#discussion_r3859646100
##########
src/paimon/core/operation/file_store_commit_impl.cpp:
##########
@@ -107,6 +111,26 @@ bool MatchPartitionSpec(const std::map<std::string,
std::string>& partition,
return true;
}
+/// The (partition, bucket) pairs whose deletion vectors this commit
materializes: it replaced
+/// normal files and wrote every output without a row id, so the commit
assigns fresh ones and
+/// the global indexes over the old ids no longer address anything.
+MaterializedBuckets CollectMaterializedBuckets(
+ const std::vector<std::shared_ptr<CommitMessage>>& commit_messages) {
+ MaterializedBuckets result;
+ for (const std::shared_ptr<CommitMessage>& message : commit_messages) {
+ auto message_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(message);
+ if (message_impl == nullptr) {
+ continue;
+ }
+ const CompactIncrement& compact_increment =
message_impl->GetCompactIncrement();
+ if
(DataEvolutionUtils::IsMaterializedCompaction(compact_increment.CompactBefore(),
Review Comment:
`IsMaterializedCompaction` only asks whether every output lacks a
`first_row_id`, and `KeyValueDataFileWriter::CreateResult` always builds its
`DataFileMeta` with `/*first_row_id=*/std::nullopt`
(key_value_data_file_writer.cpp:152). So every ordinary primary-key compaction
— which replaces normal files and produces a merged file without a row id —
satisfies the predicate and lands in `materialized_buckets`.
The commit is then routed through `MaterializedIndexChangesProvider`, which
re-scans the index manifest and emits `FileKind::Delete()` for *every*
global-index entry of that (partition, bucket). A plain compaction of a
primary-key table carrying a pk-sorted index therefore drops an index it never
touched. Tables without a global index still pay an unfiltered index-manifest
read on every compaction attempt.
The other two `IsMaterializedCompaction` call sites are already inside the
data-evolution path; this one is on the generic commit path and needs its own
gate — e.g. only collecting when `options_.DataEvolutionEnabled()`, or
requiring the replaced `CompactBefore` files to actually carry a `first_row_id`.
##########
src/paimon/core/append/data_evolution_compact_planner.cpp:
##########
@@ -0,0 +1,319 @@
+/*
+ * 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/append/data_evolution_compact_planner.h"
+
+#include <algorithm>
+#include <iterator>
+#include <limits>
+#include <string>
+#include <utility>
+
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/predicate/predicate_filter.h"
+#include "paimon/common/utils/range_helper.h"
+#include "paimon/common/utils/vector_store_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/manifest/manifest_file_meta.h"
+#include "paimon/core/manifest/manifest_list.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/logging.h"
+
+namespace paimon {
+
+namespace {
+
+/// A bin that packs evolved field groups of one contiguous row id run into
one compact task.
+class CompactBin {
+ public:
+ Status Add(std::vector<std::shared_ptr<DataFileMeta>>&& file_group,
int64_t weight) {
+ if (weight_ > std::numeric_limits<int64_t>::max() - weight) {
+ return Status::Invalid("Data evolution compaction bin weight
overflows.");
+ }
+ files_.insert(files_.end(),
std::make_move_iterator(file_group.begin()),
+ std::make_move_iterator(file_group.end()));
+ weight_ += weight;
+ return Status::OK();
+ }
+
+ int64_t Weight() const {
+ return weight_;
+ }
+
+ std::vector<std::shared_ptr<DataFileMeta>> Drain() {
+ std::vector<std::shared_ptr<DataFileMeta>> result = std::move(files_);
+ files_.clear();
+ weight_ = 0;
+ return result;
+ }
+
+ private:
+ std::vector<std::shared_ptr<DataFileMeta>> files_;
+ int64_t weight_ = 0;
+};
+
+/// Emits a task for the bin's files when there are enough of them; too few
files are simply
+/// left as they are (they will be reconsidered by a later coordinator run
once neighbors
+/// appear).
+Status TriggerTask(std::vector<std::shared_ptr<DataFileMeta>>&& files, const
BinaryRow& partition,
+ int32_t compact_min_file_num,
+ std::vector<DataEvolutionNormalCompactTask>* tasks) {
+ if (static_cast<int32_t>(files.size()) < compact_min_file_num) {
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(DataEvolutionNormalCompactTask task,
+ DataEvolutionNormalCompactTask::Create(partition,
files));
+ tasks->push_back(std::move(task));
+ return Status::OK();
+}
+
+int64_t FileWeight(const std::shared_ptr<DataFileMeta>& file, int64_t
open_file_cost) {
+ return std::max(file->file_size, open_file_cost);
+}
+
+Status PlanPartition(const BinaryRow& partition,
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ int64_t target_file_size, int64_t open_file_cost, int32_t
compact_min_file_num,
+ std::vector<DataEvolutionNormalCompactTask>* tasks) {
+ // Blob files are dedicated storage: they are not rewritten and never
enter a task, and
+ // their row ranges stay covered by the rewritten data files. Vector-store
files were
+ // already rejected by PlanCompactTasks, so everything else is a normal
file.
+ std::vector<std::shared_ptr<DataFileMeta>> data_files;
+ data_files.reserve(files.size());
+ for (const auto& file : files) {
+ if (BlobUtils::IsBlobFile(file->file_name)) {
+ continue;
+ }
+ data_files.push_back(file);
+ }
+ if (data_files.empty()) {
+ return Status::OK();
+ }
+
+ // Contiguous runs extend each file's range to [first, first + row_count],
one past its
+ // last row: adjacent files share an endpoint and overlap, so a run only
breaks on a real
+ // row id gap.
+ RangeHelper<std::shared_ptr<DataFileMeta>> adjacency_helper(
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<int64_t> {
+ return file->NonNullFirstRowId();
+ },
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<int64_t> {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id,
file->NonNullFirstRowId());
+ return first_row_id + file->row_count;
+ });
+ // Field groups use the closed range [first, first + row_count - 1]: only
files holding the
+ // same rows overlap here, and a group must cover the exact same range in
every file.
+ RangeHelper<std::shared_ptr<DataFileMeta>> field_group_helper(
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<int64_t> {
+ return file->NonNullFirstRowId();
+ },
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<int64_t> {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id,
file->NonNullFirstRowId());
+ return first_row_id + file->row_count - 1;
+ });
+
+
PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>>
components,
+
adjacency_helper.MergeOverlappingRanges(std::move(data_files)));
+ for (auto& component : components) {
+
PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>>
field_groups,
+
field_group_helper.MergeOverlappingRanges(std::move(component)));
+ CompactBin bin;
+ for (auto& field_group : field_groups) {
+ PAIMON_ASSIGN_OR_RAISE(bool same_range,
+
field_group_helper.AreAllRangesSame(field_group));
+ if (!same_range) {
+ std::vector<std::string> file_names;
+ file_names.reserve(field_group.size());
+ for (const auto& file : field_group) {
+ file_names.push_back(file->file_name);
+ }
+ return Status::Invalid(fmt::format(
+ "Files of one data evolution field group should share the
same row range, "
+ "but got [{}].",
+ fmt::join(file_names, ", ")));
+ }
+ int64_t weight = 0;
+ for (const auto& file : field_group) {
+ int64_t file_weight = FileWeight(file, open_file_cost);
+ if (weight > std::numeric_limits<int64_t>::max() -
file_weight) {
+ return Status::Invalid(
+ "Data evolution compaction field group weight
overflows.");
+ }
+ weight += file_weight;
+ }
+ if (weight > target_file_size) {
+ // A heavy group cuts the current bin and is considered on its
own (still
+ // subject to the min-file-num gate): merging its files is
worthwhile, but
+ // nothing else is packed on top of it.
+ PAIMON_RETURN_NOT_OK(
+ TriggerTask(bin.Drain(), partition, compact_min_file_num,
tasks));
+ CompactBin single_group_bin;
+
PAIMON_RETURN_NOT_OK(single_group_bin.Add(std::move(field_group), weight));
+ PAIMON_RETURN_NOT_OK(
+ TriggerTask(single_group_bin.Drain(), partition,
compact_min_file_num, tasks));
+ continue;
+ }
+ PAIMON_RETURN_NOT_OK(bin.Add(std::move(field_group), weight));
Review Comment:
Blob files are dropped before packing (line 100-105), and a bin then packs
several adjacent field groups, so the rewritten file spans a wider row range
than the dedicated `.blob` files it left behind. The read side needs those
ranges to match exactly, not just to be covered: `MergeRangesAndSort` groups
files by *overlapping* closed ranges, and `CreateUnionReader` then requires
every bunch of a group to report the same row count and the same first row id.
Concretely, on a data-evolution table with a blob column: commit 1 writes
`{id}` for rows [0,2] (data file D1, no blob file); commit 2 writes `{id,
payload}` for rows [3,5] (data file D2 + blob file B2 [3,5]). Both read fine
today. A compaction with `compaction.min.file-num=2` packs the two field groups
into one bin and rewrites them into a single file O [0,5]. The live set is then
{O[0,5], B2[3,5]}: they overlap, land in one group, and `CreateUnionReader`
fails with "All files in a field merge split should have the same row count."
Every scan of that range fails from then on, and the commit is already durable.
`TestBlobFilesExcluded` only covers one field group whose blob file shares
its exact range, which is why this shape is not caught.
Blob coverage needs to be part of the packing decision: pass the partition's
blob files into `PlanPartition`, compute for each field group the set of blob
field ids whose files cover its exact range, and drain the bin whenever the
next group's coverage set differs — so a rewritten file never spans a change in
dedicated-blob coverage.
##########
src/paimon/core/io/managed_blob_reference_file.cpp:
##########
@@ -0,0 +1,399 @@
+/*
+ * 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/io/managed_blob_reference_file.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <string_view>
+#include <utility>
+
+#include "arrow/util/crc32.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+namespace {
+
+constexpr int32_t kMagic = 0x50424C52; // "PBLR"
+constexpr int8_t kVersion = 1;
+
+void AppendBigEndianInt32(int32_t value, std::string* out) {
+ auto bits = static_cast<uint32_t>(value);
+ out->push_back(static_cast<char>((bits >> 24) & 0xFF));
+ out->push_back(static_cast<char>((bits >> 16) & 0xFF));
+ out->push_back(static_cast<char>((bits >> 8) & 0xFF));
+ out->push_back(static_cast<char>(bits & 0xFF));
+}
+
+/// A sequential big-endian reader over an in-memory buffer.
+class BigEndianBufferReader {
+ public:
+ BigEndianBufferReader(const char* data, size_t size) : data_(data),
size_(size) {}
+
+ Result<int32_t> ReadInt32() {
+ PAIMON_RETURN_NOT_OK(Require(4));
+ uint32_t bits =
(static_cast<uint32_t>(static_cast<uint8_t>(data_[pos_])) << 24) |
+ (static_cast<uint32_t>(static_cast<uint8_t>(data_[pos_
+ 1])) << 16) |
+ (static_cast<uint32_t>(static_cast<uint8_t>(data_[pos_
+ 2])) << 8) |
+ static_cast<uint32_t>(static_cast<uint8_t>(data_[pos_
+ 3]));
+ pos_ += 4;
+ return static_cast<int32_t>(bits);
+ }
+
+ Result<int8_t> ReadInt8() {
+ PAIMON_RETURN_NOT_OK(Require(1));
+ return static_cast<int8_t>(data_[pos_++]);
+ }
+
+ Result<uint16_t> ReadUInt16() {
+ PAIMON_RETURN_NOT_OK(Require(2));
+ auto bits = static_cast<uint16_t>((static_cast<uint8_t>(data_[pos_])
<< 8) |
+ static_cast<uint8_t>(data_[pos_ +
1]));
+ pos_ += 2;
+ return bits;
+ }
+
+ Result<std::string_view> ReadBytes(size_t length) {
+ PAIMON_RETURN_NOT_OK(Require(length));
+ std::string_view view(data_ + pos_, length);
+ pos_ += length;
+ return view;
+ }
+
+ size_t Position() const {
+ return pos_;
+ }
+
+ size_t Remaining() const {
+ return size_ - pos_;
+ }
+
+ private:
+ Status Require(size_t length) const {
+ if (pos_ + length > size_) {
+ return Status::Invalid("Managed blob reference file is
truncated.");
+ }
+ return Status::OK();
+ }
+
+ const char* data_;
+ size_t size_;
+ size_t pos_ = 0;
+};
+
+/// Encodes a UTF-8 string the way the sidecar format stores it: a uint16
big-endian byte
+/// length followed by modified UTF-8 bytes. Modified UTF-8 equals standard
UTF-8 except that
+/// U+0000 becomes the two-byte form 0xC0 0x80 and supplementary characters
are encoded as a
+/// CESU-8 surrogate pair (two three-byte groups).
+Status AppendJavaUtf(const std::string& value, std::string* out) {
+ std::string encoded;
+ encoded.reserve(value.size());
+ size_t i = 0;
+ const auto* bytes = reinterpret_cast<const uint8_t*>(value.data());
+ while (i < value.size()) {
+ uint8_t lead = bytes[i];
+ uint32_t code_point = 0;
+ size_t sequence_length = 0;
+ if (lead < 0x80) {
+ code_point = lead;
+ sequence_length = 1;
+ } else if ((lead >> 5) == 0x6) {
+ code_point = lead & 0x1F;
+ sequence_length = 2;
+ } else if ((lead >> 4) == 0xE) {
+ code_point = lead & 0x0F;
+ sequence_length = 3;
+ } else if ((lead >> 3) == 0x1E) {
+ code_point = lead & 0x07;
+ sequence_length = 4;
+ } else {
+ return Status::Invalid("Managed blob reference contains invalid
UTF-8.");
+ }
+ if (i + sequence_length > value.size()) {
+ return Status::Invalid("Managed blob reference contains truncated
UTF-8.");
+ }
+ for (size_t j = 1; j < sequence_length; j++) {
+ uint8_t continuation = bytes[i + j];
+ if ((continuation >> 6) != 0x2) {
+ return Status::Invalid("Managed blob reference contains
invalid UTF-8.");
+ }
+ code_point = (code_point << 6) | (continuation & 0x3F);
+ }
+ i += sequence_length;
Review Comment:
`AppendJavaUtf` classifies a lead byte by its bit pattern but never
range-checks the decoded code point, so it accepts two inputs it cannot
round-trip — even though every other malformed lead is rejected with "Managed
blob reference contains invalid UTF-8.":
- `(lead >> 3) == 0x1E` matches `F0`–`F7`, so code points above U+10FFFF
pass. For `F5 80 80 80` (U+140000), `offset >> 10` gives `high = 0xDCC0`, which
is a *low* surrogate, and the writer emits `ED B3 80 ED B0 80`. `ReadJavaUtf`
sees no high surrogate, copies both groups verbatim, and returns six bytes
where four went in — the sidecar records a pack path that does not exist.
- A path already holding a CESU-8 lone surrogate (`ED A0 80`) takes the
`code_point < 0x10000` branch and is copied through unchanged. `Write` reports
success, and `ReadJavaUtf` then rejects that sidecar forever with "holds an
unpaired surrogate".
One check after the continuation loop makes the writer refuse exactly what
the reader refuses to decode:
```cpp
i += sequence_length;
if (code_point > 0x10FFFF || (code_point >= 0xD800 && code_point <= 0xDFFF))
{
return Status::Invalid("Managed blob reference contains invalid UTF-8.");
}
```
##########
src/paimon/core/operation/commit/uncommitted_file_cleaner.cpp:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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/uncommitted_file_cleaner.h"
+
+#include <set>
+#include <string>
+
+#include "paimon/commit_message.h"
+#include "paimon/core/index/index_file_meta.h"
+#include "paimon/core/index/index_path_factory.h"
+#include "paimon/core/io/compact_increment.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/data_increment.h"
+#include "paimon/core/table/sink/commit_message_impl.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/logging.h"
+
+namespace paimon {
+
+Status UncommittedFileCleaner::Delete(
+ const std::shared_ptr<FileStorePathFactory>& path_factory,
+ const std::shared_ptr<FileSystem>& fs,
+ const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
Logger* logger) {
+ // A message this cleaner cannot handle stops that message, never the ones
after it: giving
+ // up on the whole list would strand every file the remaining messages
describe, which is
+ // the leak this class exists to prevent. The first such failure is
reported once the list
+ // has been walked.
+ Status first_error = Status::OK();
+ for (const auto& message : commit_messages) {
+ auto* msg = dynamic_cast<CommitMessageImpl*>(message.get());
+ if (msg == nullptr) {
+ if (first_error.ok()) {
+ first_error = Status::Invalid("fail to cast commit message to
impl");
+ }
+ continue;
+ }
+ Result<std::shared_ptr<DataFilePathFactory>>
data_file_path_factory_result =
+ path_factory->CreateDataFilePathFactory(msg->Partition(),
msg->Bucket());
+ Result<std::unique_ptr<IndexPathFactory>>
index_file_path_factory_result =
+ path_factory->CreateIndexFileFactory(msg->Partition(),
msg->Bucket());
+ if (!data_file_path_factory_result.ok() ||
!index_file_path_factory_result.ok()) {
+ const Status& status = data_file_path_factory_result.ok()
+ ?
index_file_path_factory_result.status()
+ :
data_file_path_factory_result.status();
+ if (first_error.ok()) {
+ first_error = status;
+ }
+ PAIMON_LOG_WARN(logger,
+ "Cannot resolve the paths of an uncommitted
message in bucket %d: %s. "
+ "Its files are left behind.",
+ msg->Bucket(), status.ToString().c_str());
+ continue;
+ }
+ std::shared_ptr<DataFilePathFactory> data_file_path_factory =
+ std::move(data_file_path_factory_result).value();
+ std::unique_ptr<IndexPathFactory> index_file_path_factory =
+ std::move(index_file_path_factory_result).value();
+
+ const DataIncrement& new_files_increment = msg->GetNewFilesIncrement();
+ const CompactIncrement& compact_increment = msg->GetCompactIncrement();
+
+ std::vector<std::shared_ptr<DataFileMeta>> data_files_to_delete;
+ auto append_data_files =
+ [&data_files_to_delete](const
std::vector<std::shared_ptr<DataFileMeta>>& files) {
+ data_files_to_delete.insert(data_files_to_delete.end(),
files.begin(), files.end());
+ };
+ append_data_files(new_files_increment.NewFiles());
+ append_data_files(new_files_increment.ChangelogFiles());
+ append_data_files(compact_increment.CompactAfter());
Review Comment:
An LSM upgrade puts the *same* file name in both `CompactBefore` and
`CompactAfter`: `MergeTreeCompactRewriter::Upgrade` returns
`CompactResult({file}, {file->Upgrade(level)})`, and `DataFileMeta::Upgrade`
copies `file_name` through and only changes the level. Deleting
`CompactAfter()` unconditionally therefore deletes a file the current snapshot
still references.
`MergeTreeWriter::DoClose` skips those on purpose ("Upgrade file is required
by previous snapshot"), and so does `UpdateCompactResult`. This cleaner does
not, and the new `prepare_guard` in `AbstractFileStoreWrite::PrepareCommit`
makes it reachable: one writer drains an upgrade-only increment, a later writer
in the same loop fails, and the guard removes a live data file. The rows are
gone from the committed snapshot.
Same guard as `DoClose`:
```cpp
std::unordered_set<std::string> before_names;
for (const auto& file : compact_increment.CompactBefore()) {
before_names.insert(file->file_name);
}
for (const auto& file : compact_increment.CompactAfter()) {
if (before_names.count(file->file_name) == 0) {
data_files_to_delete.push_back(file);
}
}
```
##########
src/paimon/core/append/append_compact_coordinator.cpp:
##########
@@ -281,6 +395,75 @@ std::vector<AppendCompactTask> GenerateCompactTasks(
return tasks;
}
+/// Cleans up the rewritten output files of already finished tasks, best
effort. Used when a
+/// later task fails: the collected commit messages are discarded, so their
outputs would
+/// otherwise linger until an orphan clean. Failures are logged and never mask
the original
+/// compaction error the caller returns.
+void CleanupCompactOutputs(const std::vector<std::shared_ptr<CommitMessage>>&
commit_messages,
+ const std::shared_ptr<FileStorePathFactory>&
path_factory,
+ const CoreOptions& core_options) {
+ auto logger = Logger::GetLogger("AppendCompactCoordinator");
+ for (const auto& message : commit_messages) {
+ auto message_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(message);
+ if (!message_impl) {
+ // The log macros require at least one format argument.
+ PAIMON_LOG_WARN(logger, "%s",
+ "Skipping cleanup of a compact output: unexpected
commit message "
+ "type");
+ continue;
+ }
+ Result<std::shared_ptr<DataFilePathFactory>> data_file_path_factory =
+ path_factory->CreateDataFilePathFactory(message_impl->Partition(),
+ message_impl->Bucket());
+ if (!data_file_path_factory.ok()) {
+ PAIMON_LOG_WARN(logger,
+ "Skipping cleanup of compact outputs in partition
%s bucket %d: %s",
+ message_impl->Partition().ToString().c_str(),
message_impl->Bucket(),
+
data_file_path_factory.status().ToString().c_str());
+ continue;
+ }
+ for (const auto& file :
message_impl->GetCompactIncrement().CompactAfter()) {
Review Comment:
This loop cleans up `CompactAfter()` data files only, so the deletion-vector
index files `DataEvolutionCompactDeletionVectorRewriter` already wrote
(`CompactIncrement::NewIndexFiles()`) stay on disk when the round's commit
fails.
The comment on the failure path says they are "left to orphan cleaning,
which is what collects unreferenced index files", but
`OrphanFilesCleanerImpl::SupportToClean` accepts only `manifest-*`,
`manifest-list-*`, `*.tmp`, and `data-*` with a data-format or `.blobref`
suffix. An index file is named `index-<uuid>`
(`IndexPathFactory::INDEX_PREFIX`), so it matches nothing and is never
reclaimed.
Losing the optimistic-commit race is a normal outcome here —
`CheckDeletionVectorMigrationIsComplete` explicitly tells the caller to "give
up committing and plan again" — so every retry on a contended table leaves more
index files behind for good. `RunMaterializeRound` has the same gap and does
not even pass `index_messages` to the cleanup when `DropGlobalIndexes` fails
after the rewrite succeeded.
`UncommittedFileCleaner::Delete` in this same PR already deletes data files,
sidecars and index files for a message; calling it here (and passing the index
messages on the materialize path) would cover this without new code.
##########
src/paimon/core/append/data_evolution_compact_deletion_vector_rewriter.cpp:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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/append/data_evolution_compact_deletion_vector_rewriter.h"
+
+#include <map>
+#include <optional>
+#include <string>
+#include <unordered_set>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/utils/linked_hash_map.h"
+#include "paimon/common/utils/range_helper.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/deletionvectors/deletion_vector.h"
+#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
+#include "paimon/core/index/deletion_vector_meta.h"
+#include "paimon/core/index/index_file_handler.h"
+#include "paimon/core/index/index_file_meta.h"
+#include "paimon/core/io/compact_increment.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_increment.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/core/table/sink/commit_message_impl.h"
+#include "paimon/core/table/source/deletion_file.h"
+#include "paimon/core/utils/data_evolution_utils.h"
+#include "paimon/logging.h"
+#include "paimon/utils/range.h"
+
+namespace paimon {
+
+namespace {
+
+/// Bucket every data-evolution compact message writes to. Unaware-bucket
tables keep the
+/// legacy single bucket, and the rewritten index files have to land in the
same one.
+constexpr int32_t kUnawareBucket = 0;
+
+/// One deletion-vector index file of a partition, described by metadata
alone: which data
+/// files it stores a vector for and where in the file each vector sits.
+///
+/// Nothing is deserialized to build this. A vector is read only when a move
actually takes it
+/// away, and the vectors that stay are read only for the index files a move
touched, so an
+/// untouched index file is never opened.
+struct IndexFileState {
+ std::shared_ptr<IndexFileMeta> meta;
+ std::string path;
+ /// The vectors this index file still owns, in write order.
+ LinkedHashMap<std::string, DeletionVectorMeta> dv_metas;
+ bool dirty;
+};
+
+/// The recorded position of one vector inside its index file.
+DeletionFile ToDeletionFile(const std::string& index_file_path, const
DeletionVectorMeta& dv_meta) {
+ return DeletionFile(index_file_path, dv_meta.GetOffset(),
dv_meta.GetLength(),
+ dv_meta.GetCardinality());
+}
+
+/// The files a rewritten deletion vector has to be keyed by after one compact
task: the
+/// task's inputs, whose per-group anchors hold the deletions today, and the
single output
+/// file they were merged into.
+///
+/// A materialized task has no such output. It applied the deletions while
rewriting, so its
+/// rows carry fresh row ids the commit assigns and the old vectors are
dropped rather than
+/// moved; `after` is null and `after_range` unused for one of those.
+struct CompactedGroup {
+ std::vector<std::shared_ptr<DataFileMeta>> before;
+ std::shared_ptr<DataFileMeta> after;
+ Range after_range;
+ bool materialized = false;
+};
+
+std::vector<std::shared_ptr<DataFileMeta>> NormalFiles(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files) {
+ std::vector<std::shared_ptr<DataFileMeta>> result;
+ result.reserve(files.size());
+ for (const auto& file : files) {
+ if (DataEvolutionUtils::IsNormalFile(file->file_name)) {
+ result.push_back(file);
+ }
+ }
+ return result;
+}
+
+Result<Range> RowRangeOf(const std::shared_ptr<DataFileMeta>& file) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, file->NonNullFirstRowId());
+ return Range(first_row_id, first_row_id + file->row_count - 1);
+}
+
+/// Merges the files of one compact task into row range groups: files covering
the same rows
+/// belong to the same group and share one deletion vector, keyed by the
group's anchor.
+Result<std::vector<std::vector<std::shared_ptr<DataFileMeta>>>> RowRangeGroups(
+ std::vector<std::shared_ptr<DataFileMeta>>&& files) {
+ RangeHelper<std::shared_ptr<DataFileMeta>> helper(
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<int64_t> {
+ return file->NonNullFirstRowId();
+ },
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<int64_t> {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id,
file->NonNullFirstRowId());
+ return first_row_id + file->row_count - 1;
+ });
+ return helper.MergeOverlappingRanges(std::move(files));
+}
+
+/// Collects the compacted groups of each partition, validating the shape
data-evolution
+/// compaction is expected to produce.
+Result<LinkedHashMap<BinaryRow, std::vector<CompactedGroup>>>
CollectCompactedGroups(
+ const std::vector<std::shared_ptr<CommitMessage>>& compact_messages) {
+ LinkedHashMap<BinaryRow, std::vector<CompactedGroup>> result;
+ for (const auto& message : compact_messages) {
+ auto message_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(message);
+ if (message_impl == nullptr) {
+ return Status::Invalid(
+ "Data evolution compaction produced an unexpected commit
message type.");
+ }
+ const CompactIncrement& compact_increment =
message_impl->GetCompactIncrement();
+ // The rewriter owns every index change of this round; a task that
already produced
+ // one would be silently dropped by the index-only messages built
below.
+ if (!compact_increment.NewIndexFiles().empty() ||
+ !compact_increment.DeletedIndexFiles().empty()) {
+ return Status::Invalid(
+ "Data evolution compaction should not produce index changes
before the "
+ "deletion vector rewrite.");
+ }
+ if (message_impl->Bucket() != kUnawareBucket ||
+ message_impl->TotalBuckets() != std::nullopt) {
+ return Status::Invalid(fmt::format(
+ "Data evolution compaction should only produce unaware-bucket
commit messages, "
+ "but got bucket {}.",
+ message_impl->Bucket()));
+ }
+
+ std::vector<std::shared_ptr<DataFileMeta>> before =
+ NormalFiles(compact_increment.CompactBefore());
+ std::vector<std::shared_ptr<DataFileMeta>> after =
+ NormalFiles(compact_increment.CompactAfter());
+ // A blob-only or index-only message carries no rows whose deletions
could move.
+ if (before.empty() && after.empty()) {
+ continue;
+ }
+ // A materialized task wrote its rows without row ids, so the commit
assigns fresh ones
+ // and the deletions it applied must not follow them: the old vectors
are dropped
+ // rather than moved. The global index dropper and the commit's
conflict check decide
+ // the same shape through the same helper.
+ bool materialized = DataEvolutionUtils::IsMaterializedCompaction(
+ compact_increment.CompactBefore(),
compact_increment.CompactAfter());
+ if (materialized) {
+ result[message_impl->Partition()].push_back(
+ CompactedGroup{std::move(before), /*after=*/nullptr, Range(0,
0),
+ /*materialized=*/true});
+ continue;
+ }
+ if (after.size() != 1) {
+ return Status::Invalid(
+ "One data evolution compact task should produce exactly one
normal file.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(Range after_range, RowRangeOf(after[0]));
+ // A task that keeps its rows in place must cover exactly the rows it
replaced, or the
+ // deletions would move onto the wrong ones.
+ PAIMON_ASSIGN_OR_RAISE(Range before_range,
+
DataEvolutionUtils::CheckContiguousRowRange(before));
+ if (!(before_range == after_range)) {
+ return Status::Invalid(fmt::format(
+ "Data evolution compaction must keep the same row id range,
but compacted {} "
+ "into {}.",
+ before_range.ToString(), after_range.ToString()));
+ }
+ result[message_impl->Partition()].push_back(
+ CompactedGroup{std::move(before), after[0], after_range,
/*materialized=*/false});
+ }
+ return result;
+}
+
+/// Moves one row range group's deletion vector from its anchor file onto
`merged`, whose
+/// positions are relative to `after_range`.
+Status MoveDeletionVector(const std::shared_ptr<DeletionVector>& old_vector,
+ const Range& anchor_range, const Range& after_range,
+ const std::shared_ptr<DeletionVector>& merged) {
+ if (anchor_range == after_range) {
+ // The group already spans the whole compacted file: the vector only
changes its key.
+ return merged->Merge(old_vector);
Review Comment:
`merged` is built from the table option
(`DeletionVector::Create(core_options.DeletionVectorsBitmap64())`) while
`old_vector` is whatever kind is stored on disk, and both `Merge`
implementations reject a mismatched dynamic type. The general path just below —
`ForEachDeletedPosition` + `Delete` — has no such requirement, so the same data
succeeds when a task packs two groups and fails when it packs one.
A table that has bitmap32 vectors on disk and later turns on
`deletion-vectors.bitmap64` reaches exactly that state: Java supports it
(`Bitmap64DeletionVector.fromBitmapDeletionVector` exists for this conversion)
and nothing marks the option immutable — `ValidateImmutableOptions` only
compares the caller's override against the schema, so both read `true` here.
Since the planner deterministically produces the same task shape, such a group
can never be compacted again, while a task that happens to pack two groups over
the same data silently converts.
`shift` is 0 when the ranges are equal, so dropping the special case and
always taking the `ForEachDeletedPosition` path handles both kinds uniformly
and keeps one behaviour for both shapes.
--
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]
