wangyong9999 commented on code in PR #227: URL: https://github.com/apache/paimon-cpp/pull/227#discussion_r3859646108
########## 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/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]
