zjw1111 commented on code in PR #163: URL: https://github.com/apache/paimon-cpp/pull/163#discussion_r3748924738
########## src/paimon/core/realtime/realtime_context.cpp: ########## @@ -0,0 +1,194 @@ +/* + * 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/realtime/realtime_context.h" + +#include <limits> +#include <map> +#include <mutex> +#include <optional> +#include <tuple> +#include <utility> +#include <vector> + +#include "arrow/c/helpers.h" +#include "paimon/arrow/abi.h" +#include "paimon/macros.h" +#include "paimon/realtime/arrow_mem_indexer_factory.h" +#include "paimon/realtime/mem_indexer.h" +#include "paimon/status.h" + +namespace paimon { + +class RealtimeContext::Impl { + public: + explicit Impl(const std::shared_ptr<MemIndexerFactory>& factory) : factory_(factory) {} + + Result<RealtimeMemIndexerState> GetOrCreateMemIndexer( + const std::map<std::string, std::string>& partition, int32_t bucket, + std::unique_ptr<ArrowSchema> write_schema, + const std::map<std::string, std::string>& options, + const std::shared_ptr<MemoryPool>& memory_pool) { + std::lock_guard<std::mutex> progress_lock(progress_mutex_); + std::lock_guard<std::mutex> registry_lock(mutex_); + const RealtimePartitionBucket key(partition, bucket); + int64_t initial_offset = 0; + auto offset_iter = committed_offsets_.find(key); + if (offset_iter != committed_offsets_.end()) { + if (offset_iter->second == std::numeric_limits<int64_t>::max()) { + if (write_schema) { + ArrowSchemaRelease(write_schema.get()); + } + return Status::Invalid("real-time offset has reached INT64_MAX"); + } + initial_offset = offset_iter->second + 1; Review Comment: The initial offset handed to a writer is derived only from the committed offsets. It ignores segments that a reused indexer has already sealed but not yet committed. `SealForCommit()` resets `building_range_` (`arrow_mem_indexer.cpp:296`), and `Write()` only checks continuity against `building_range_` (`arrow_mem_indexer.cpp:269-270`), so there is no continuity check across a seal boundary. `MemIndexer` also exposes no accessor for the highest offset it currently holds, and the context does not track it. Consequence: when a new writer is created against a context that still holds a sealed-but-uncommitted segment for the same partition-bucket — the cross-writer-instance context reuse that `realtime_context.h:87-91` describes — the writer restarts at `committed + 1` and the overlapping range is accepted silently. The indexer then holds both the sealed `[x..y]` segment and a new building range starting at `x`; `AcquireReadView()` merges the two, so memory reads return duplicated rows. At commit time the second range fails the continuity check in `realtime_commit_properties.cpp:215-220`, far from the root cause; if the first commit message was abandoned, the already-flushed data files are left uncommitted. For scope: this is not reachable inside a single `FileStoreWrite` instance, since realtime writers are never evicted (the legacy `PrepareCommit` is rejected at `abstract_file_store_write.cpp:150-153` and the memory manager is a no-op). Two live writers sharing one context are also caught by the `building_range_` check. The seal boundary is the only unguarded case. ########## include/paimon/file_store_commit.h: ########## @@ -71,6 +72,21 @@ class PAIMON_EXPORT FileStoreCommit { int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER, std::optional<int64_t> watermark = std::nullopt) = 0; + /// Commit sealed real-time segments and persist their partition-bucket offset progress. + /// + /// Entries for each partition-bucket must form a contiguous range beginning after the offset + /// recorded by the latest committed snapshot. Input entries may be unordered; this method + /// orders them by partition, bucket, and offset before validating continuity. The resulting + /// snapshot atomically publishes the data files and the updated offset map. + /// + /// @param realtime_commits Commit messages and inclusive offset ranges to commit. + /// @param commit_identifier Identifier of the streaming commit operation. + /// @param watermark Optional event-time watermark. + /// @return Status indicating the success or failure of the commit operation. + virtual Status CommitWithProgress(const std::vector<RealtimeCommitProgress>& realtime_commits, Review Comment: `CommitWithProgress` returns only `Status`, but `FileStoreWrite::RefreshCommittedSnapshot(int64_t snapshot_id)` needs the id of the snapshot this call produced. The API offers no way to obtain it, so the caller has to query `LatestSnapshot()` again and cannot tell whether the result is its own snapshot — `test/inte/realtime_write_inte_test.cpp` does exactly this. There is no reliable link between a commit and the refresh that is supposed to follow it. In the current implementation the offsets map is carried forward by every commit path (`realtime_commit_properties.cpp:191-201`), so refreshing with a later snapshot is usually harmless. `RollbackTo` is the exception: the new snapshot is built from the target snapshot's properties (`file_store_commit_impl.cpp:345-354`), so the offsets map can move backwards, and `AdvanceCommittedProgress` then rejects every later refresh with `committed snapshot cannot move backwards`. ########## src/paimon/core/realtime/realtime_append_only_writer.cpp: ########## @@ -0,0 +1,214 @@ +/* + * 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/realtime/realtime_append_only_writer.h" + +#include <cstdint> +#include <limits> +#include <optional> +#include <utility> +#include <vector> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "paimon/common/reader/concat_batch_reader.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/types/row_kind.h" +#include "paimon/common/utils/arrow/arrow_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/append/append_only_writer.h" +#include "paimon/core/utils/commit_increment.h" +#include "paimon/macros.h" +#include "paimon/realtime/realtime_context.h" + +namespace paimon { + +Result<std::shared_ptr<RealtimeAppendOnlyWriter>> RealtimeAppendOnlyWriter::Create( + const std::map<std::string, std::string>& partition, int32_t bucket, + std::unique_ptr<::ArrowSchema> write_schema, + const std::shared_ptr<RealtimeContext>& realtime_context, + const std::shared_ptr<AppendOnlyWriter>& file_writer, + const std::shared_ptr<arrow::Schema>& input_schema, + const std::map<std::string, std::string>& options, + const std::shared_ptr<MemoryPool>& memory_pool) { + if (!realtime_context) { + return Status::Invalid("real-time context is null"); + } + PAIMON_ASSIGN_OR_RAISE(RealtimeMemIndexerState indexer_state, + realtime_context->GetOrCreateMemIndexer( + partition, bucket, std::move(write_schema), options, memory_pool)); + return std::shared_ptr<RealtimeAppendOnlyWriter>( + new RealtimeAppendOnlyWriter(indexer_state.indexer, file_writer, input_schema, + indexer_state.initial_offset, memory_pool)); +} + +RealtimeAppendOnlyWriter::RealtimeAppendOnlyWriter( + const std::shared_ptr<MemIndexer>& mem_indexer, + const std::shared_ptr<AppendOnlyWriter>& file_writer, + const std::shared_ptr<arrow::Schema>& input_schema, int64_t next_offset, + const std::shared_ptr<MemoryPool>& memory_pool) + : memory_pool_(memory_pool), + mem_indexer_(mem_indexer), + file_writer_(file_writer), + input_schema_(input_schema), + next_offset_(next_offset) {} + +Status RealtimeAppendOnlyWriter::Write(std::unique_ptr<RecordBatch>&& batch) { + for (RecordBatch::RowKind row_kind : batch->GetRowKind()) { + if (row_kind != RecordBatch::RowKind::INSERT) { + PAIMON_ASSIGN_OR_RAISE(const RowKind* kind, + RowKind::FromByteValue(static_cast<int8_t>(row_kind))); + return Status::Invalid("Append only writer can not accept record batch with RowKind ", + kind->Name()); + } + } + + int64_t row_count = batch->GetData()->length; + if (row_count == 0) { + return Status::OK(); + } + std::lock_guard<std::mutex> lock(mem_indexer_mutex_); + // Reserve INT64_MAX as the exhausted next-offset sentinel. + if (row_count > std::numeric_limits<int64_t>::max() - next_offset_) { + return Status::Invalid("real-time offset range exceeds INT64_MAX"); + } + Range range(next_offset_, next_offset_ + row_count - 1); + PAIMON_RETURN_NOT_OK(mem_indexer_->Write(RealtimeWriteBatch{std::move(batch), range})); + next_offset_ += row_count; + return Status::OK(); +} + +Result<CommitIncrement> RealtimeAppendOnlyWriter::PrepareCommit(bool wait_compaction) { + std::lock_guard<std::mutex> lock(prepare_mutex_); + std::optional<std::shared_ptr<RealtimeSegmentHandle>> segment; + { + std::lock_guard<std::mutex> mem_indexer_lock(mem_indexer_mutex_); + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<RealtimeSegmentHandle>> sealed_segment, + mem_indexer_->SealForCommit()); + segment = std::move(sealed_segment); + } + if (segment) { + PAIMON_RETURN_NOT_OK(FlushSegment(segment.value())); + } + PAIMON_ASSIGN_OR_RAISE(CommitIncrement increment, file_writer_->PrepareCommit(wait_compaction)); + if (segment) { + increment.SetRealtimeOffsetRange(segment.value()->GetOffsetRange()); + } + return increment; +} + +Status RealtimeAppendOnlyWriter::FlushSegment( + const std::shared_ptr<RealtimeSegmentHandle>& segment) { + PAIMON_ASSIGN_OR_RAISE(std::vector<std::unique_ptr<BatchReader>> readers, + mem_indexer_->CreateCommitReaders(segment)); + ConcatBatchReader reader(std::move(readers), memory_pool_); + ScopeGuard reader_guard([&reader]() { reader.Close(); }); + const Range offset_range = segment->GetOffsetRange(); + int64_t emitted_rows = 0; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader.NextBatch()); + if (BatchReader::IsEofBatch(batch)) { + break; + } + auto& [c_array, c_schema] = batch; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> imported, + arrow::ImportArray(c_array.get(), c_schema.get())); + std::shared_ptr<arrow::StructArray> struct_array = + std::dynamic_pointer_cast<arrow::StructArray>(imported); + if (!struct_array) { + return Status::Invalid("mem indexer commit reader returned a non-StructArray"); + } + std::shared_ptr<arrow::Array> value_kind = + struct_array->GetFieldByName(SpecialFields::ValueKind().Name()); + if (!value_kind || value_kind->type_id() != arrow::Type::INT8) { + return Status::Invalid( + "mem indexer commit reader must return an INT8 _VALUE_KIND field"); + } + std::shared_ptr<arrow::Int8Array> row_kinds = + std::static_pointer_cast<arrow::Int8Array>(value_kind); + for (int64_t i = 0; i < row_kinds->length(); ++i) { + if (row_kinds->IsNull(i) || + row_kinds->Value(i) != static_cast<int8_t>(RecordBatch::RowKind::INSERT)) { + return Status::Invalid( + "append mem indexer commit reader returned a non-INSERT row"); + } + } + PAIMON_ASSIGN_OR_RAISE(struct_array, ArrowUtils::RemoveFieldFromStructArray( + struct_array, SpecialFields::ValueKind().Name())); + if (!struct_array->type()->Equals(arrow::struct_(input_schema_->fields()))) { + return Status::Invalid( + "mem indexer commit reader schema does not match table write schema"); + } + + int64_t row_count = struct_array->length(); + if (row_count > offset_range.Count() - emitted_rows) { + return Status::Invalid( + "mem indexer commit readers returned more rows than the sealed offset range"); + } + emitted_rows += row_count; + + auto output = std::make_unique<ArrowArray>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*struct_array, output.get())); + RecordBatchBuilder builder(output.get()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<RecordBatch> record_batch, builder.Finish()); + PAIMON_RETURN_NOT_OK(file_writer_->Write(std::move(record_batch))); + } + if (emitted_rows != offset_range.Count()) { + return Status::Invalid( + "mem indexer commit readers returned fewer rows than the sealed offset range"); + } + return Status::OK(); +} + +Status RealtimeAppendOnlyWriter::Compact(bool) { + return Status::Invalid("real-time append write does not support explicit compaction"); +} + +uint64_t RealtimeAppendOnlyWriter::GetMemoryUsage() const { + // The first implementation does not spill sealed or building segments through + // WriterMemoryManager. + return 0; +} + +Status RealtimeAppendOnlyWriter::FlushMemory() { + return Status::OK(); +} + +Result<bool> RealtimeAppendOnlyWriter::CompactNotCompleted() { + return file_writer_->CompactNotCompleted(); +} + +Status RealtimeAppendOnlyWriter::Sync() { + return file_writer_->Sync(); +} + +Status RealtimeAppendOnlyWriter::Close() { + { + std::lock_guard<std::mutex> lock(mem_indexer_mutex_); + PAIMON_RETURN_NOT_OK(mem_indexer_->Close()); Review Comment: The writer closes a `MemIndexer` that `RealtimeContext` owns and shares with scans and with later writers. `FileStoreWrite::Close()` closes every writer and clears the writer map (`abstract_file_store_write.cpp:337-348`) — a normal lifecycle call, not an error path. `ArrowMemIndexer::Close()` sets `closed_` and drops all batches (`arrow_mem_indexer.cpp:377-385`), after which `Write`, `SealForCommit`, `AcquireReadView` and `AdvanceCommittedOffset` all fail with `mem indexer is closed` (`:266`, `:287`, `:314`, `:353`). The indexer stays in `RealtimeContext::indexers_`, which is never erased, and `GetOrCreateMemIndexer` returns the cached entry without checking `closed_` (`realtime_context.cpp:62-68`). So after a writer is closed, the shared context is permanently degraded for that partition-bucket: scans through the same context fail in `AcquireReadViews`, `RefreshCommittedSnapshot` fails inside `AdvanceCommittedOffset`, and a writer newly created from the same context can never write, with no way to recover. This is uncovered: every case in `test/inte/realtime_write_inte_test.cpp` calls `writer->Close()` as its last statement, and both restore tests create a fresh `RealtimeContext` for the second writer, so "close a writer, then keep using the same context" is never exercised. ########## include/paimon/realtime/mem_indexer.h: ########## @@ -0,0 +1,165 @@ +/* + * 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. + */ + +#pragma once + +#include <cstdint> +#include <map> +#include <memory> +#include <optional> +#include <string> +#include <vector> + +#include "paimon/reader/batch_reader.h" +#include "paimon/record_batch.h" +#include "paimon/result.h" +#include "paimon/utils/range.h" +#include "paimon/visibility.h" + +struct ArrowSchema; + +namespace paimon { + +class MemoryPool; +class Predicate; + +/// A table record batch and its framework-assigned contiguous offset range. +/// +/// The batch contains only table write fields. Row `i` is associated with +/// `offset_range.from + i`; the offset is progress metadata and is not a table field. +struct PAIMON_EXPORT RealtimeWriteBatch { + /// Input batch whose ownership is transferred to `MemIndexer::Write`. + std::unique_ptr<RecordBatch> batch; + /// Inclusive `[from, to]` offset range covered by `batch`. + Range offset_range; +}; + +/// Opaque handle to an immutable segment returned by `MemIndexer::SealForCommit`. +/// +/// A plugin may store the segment in memory or in spill files. Callers use this handle only to +/// request commit readers and inspect its offset range. +class PAIMON_EXPORT RealtimeSegmentHandle { + public: + virtual ~RealtimeSegmentHandle() = default; + + /// Returns the inclusive offset range covered by this segment. + virtual Range GetOffsetRange() const = 0; +}; + +/// Opaque immutable view of the rows visible from one `MemIndexer`. +/// +/// A view pins all referenced resources until the readers created from it are closed. Later +/// writes, seals, and committed-offset reclamation do not change the contents of an existing view. +class PAIMON_EXPORT MemReadView { + public: + virtual ~MemReadView() = default; + + /// Returns the inclusive offset range visible in this view, or no range when it is empty. + virtual std::optional<Range> GetOffsetRange() const = 0; +}; + +/// Parameters used by a `MemIndexer` to create readers for a query. +struct PAIMON_EXPORT MemQueryContext { + /// Requested output fields before the mandatory leading `_VALUE_KIND` field is added. + ::ArrowSchema* read_schema; + /// Predicate using field indexes from `read_schema`. + std::shared_ptr<Predicate> predicate; + /// Whether the plugin may use `predicate` to prune candidate rows. + /// + /// Keep this disabled for primary-key merge-on-read. Pruning memory before PK merge may remove + /// the newest row and incorrectly expose an older disk row. Exact predicate filtering, when + /// requested, is applied by the Paimon read framework after plugin reader creation. + bool enable_predicate_pushdown; +}; + +/// Plugin interface for buffering real-time writes before Paimon data-file generation. +/// +/// Paimon serializes calls to `Write` and `SealForCommit` for the same indexer. After sealing, +/// `CreateCommitReaders` may read the immutable sealed segment while later `Write` calls append to +/// a new building segment. Paimon retains control of file format, rolling, indexes, and +/// commit-message generation. +class PAIMON_EXPORT MemIndexer { + public: + virtual ~MemIndexer() = default; + + /// Adds a batch to the current building segment. + /// + /// The row count matches the framework-assigned `offset_range`. + virtual Status Write(RealtimeWriteBatch&& batch) = 0; + + /// Seals the current building data and opens a new building segment. + /// + /// Returns an immutable segment handle, or `std::nullopt` when there is no data to seal. + virtual Result<std::optional<std::shared_ptr<RealtimeSegmentHandle>>> SealForCommit() = 0; + + /// Creates readers that expose all rows in a sealed segment for Paimon file writing. + /// + /// Concatenating the returned readers must produce every sealed row exactly once and in write + /// order. Each output batch contains `_VALUE_KIND` followed by all fields from the factory's + /// `write_schema`. + virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders( + const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0; + + /// Acquires an immutable view containing the current sealed and building rows. + /// + /// This method may be called concurrently with query-reader creation and reclamation. It must + /// also provide a consistent snapshot when a write or seal is in progress. + virtual Result<std::shared_ptr<MemReadView>> AcquireReadView() = 0; + + /// Creates readers over rows in `view` whose offsets are greater than + /// `offset_lower_exclusive`. + /// + /// Each output batch contains `_VALUE_KIND` first, followed by the fields requested by + /// `context.read_schema` except a duplicate `_VALUE_KIND`. Concatenating all returned readers + /// must produce every matching row once. + virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateQueryReaders( + const std::shared_ptr<MemReadView>& view, int64_t offset_lower_exclusive, + const MemQueryContext& context) = 0; + + /// Notifies the indexer that its partition-bucket committed offset has advanced. + /// + /// Calls are monotonic and may repeat the same offset after a previous call reports an error, + /// so implementations must apply this notification idempotently. + /// + /// An implementation may reclaim covered segments immediately, defer destruction, spill them, + /// or retain them. Existing read views continue to keep referenced resources alive. + virtual Status AdvanceCommittedOffset(int64_t committed_offset) = 0; + + /// Returns the number of bytes currently retained by building and sealed segments. + virtual uint64_t GetMemoryUsage() const = 0; + + /// Releases resources owned by this indexer and rejects subsequent writes or seals. + virtual Status Close() = 0; +}; + +/// Factory for application-provided `MemIndexer` implementations. +class PAIMON_EXPORT MemIndexerFactory { + public: + virtual ~MemIndexerFactory() = default; + + /// Creates an indexer configured with the supplied schema, options, and memory pool. + /// @param write_schema Complete table write schema. + /// @param options Effective table options available to the indexer. + /// @param memory_pool Memory pool provided by the write context. + virtual Result<std::shared_ptr<MemIndexer>> Create( + ::ArrowSchema* write_schema, const std::map<std::string, std::string>& options, Review Comment: The ownership and lifetime contract of `write_schema` is unspecified: whether ownership transfers, whether the implementation must consume or release it, whether it may retain the pointer, and who releases it when `Create` fails. The two sides currently disagree in a way that only works by accident. The default factory consumes the schema through `arrow::ImportSchema` (`src/paimon/core/realtime/arrow_mem_indexer_factory.cpp`), while `RealtimeContext` calls `ArrowSchemaRelease` unconditionally on every path, including the success path (`src/paimon/core/realtime/realtime_context.cpp:55-57`, `:64-66`, `:71-73`). That is only safe because `ArrowSchemaRelease` is a no-op once `release` is null, so the effective contract is "released unless already released" and an implementation cannot tell whether the schema is still live after `Create` returns. An implementation that retains the raw pointer gets a use-after-release, since the context releases it as soon as `Create` returns. One level up the ownership is explicit — `RealtimeContext::GetOrCreateMemIndexer` takes `std::unique_ptr<ArrowSchema>` — and that information is discarded at the plugin boundary. ########## src/paimon/core/operation/commit/realtime_commit_properties.cpp: ########## @@ -0,0 +1,255 @@ +/* + * 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_commit_properties.h" + +#include <algorithm> +#include <limits> +#include <map> +#include <stdexcept> +#include <string> +#include <utility> +#include <vector> + +#include "fmt/format.h" +#include "paimon/common/utils/rapidjson_util.h" +#include "paimon/common/utils/uuid.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/fs/file_system.h" +#include "paimon/macros.h" + +namespace paimon { +namespace { + +class OffsetEntryJson { + public: + OffsetEntryJson() = default; + + OffsetEntryJson(std::map<std::string, std::string> partition, int32_t bucket, int64_t offset) + : partition_(std::move(partition)), bucket_(bucket), offset_(offset) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const { + rapidjson::Value value(rapidjson::kObjectType); + value.AddMember("partition", RapidJsonUtil::SerializeValue(partition_, allocator), + *allocator); + value.AddMember("bucket", RapidJsonUtil::SerializeValue(bucket_, allocator), *allocator); + value.AddMember("offset", RapidJsonUtil::SerializeValue(offset_, allocator), *allocator); + return value; + } + + void FromJson(const rapidjson::Value& value) { + partition_ = RapidJsonUtil::DeserializeKeyValue<std::map<std::string, std::string>>( + value, "partition"); + bucket_ = RapidJsonUtil::DeserializeKeyValue<int32_t>(value, "bucket"); + offset_ = RapidJsonUtil::DeserializeKeyValue<int64_t>(value, "offset"); + } + + const std::map<std::string, std::string>& Partition() const { + return partition_; + } + + int32_t Bucket() const { + return bucket_; + } + + int64_t Offset() const { + return offset_; + } + + private: + std::map<std::string, std::string> partition_; + int32_t bucket_ = -1; + int64_t offset_ = -1; +}; + +class OffsetsJson { + public: + OffsetsJson() = default; + + explicit OffsetsJson(const RealtimeOffsetMap& offsets) : offsets_(offsets) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const { + rapidjson::Value value(rapidjson::kObjectType); + value.AddMember( + "version", + RapidJsonUtil::SerializeValue(RealtimeCommitProperties::kOffsetsVersion, allocator), + *allocator); + std::vector<OffsetEntryJson> entries; + entries.reserve(offsets_.size()); + for (const auto& [partition_bucket, offset] : offsets_) { + if (partition_bucket.bucket < 0) { + throw std::invalid_argument( + fmt::format("invalid bucket {} in offsets", partition_bucket.bucket)); + } + if (offset < 0) { + throw std::invalid_argument(fmt::format("invalid offset {} for bucket {}", offset, + partition_bucket.bucket)); + } + entries.emplace_back(partition_bucket.partition, partition_bucket.bucket, offset); + } + value.AddMember("offsets", RapidJsonUtil::SerializeValue(entries, allocator), *allocator); + return value; + } + + void FromJson(const rapidjson::Value& value) { + auto version = RapidJsonUtil::DeserializeKeyValue<int32_t>(value, "version"); + if (version != RealtimeCommitProperties::kOffsetsVersion) { + throw std::invalid_argument(fmt::format("unsupported offsets version {}", version)); + } + auto entries = + RapidJsonUtil::DeserializeKeyValue<std::vector<OffsetEntryJson>>(value, "offsets"); + offsets_.clear(); + for (const OffsetEntryJson& entry : entries) { + if (entry.Bucket() < 0) { + throw std::invalid_argument( + fmt::format("invalid bucket {} in offsets", entry.Bucket())); + } + if (entry.Offset() < 0) { + throw std::invalid_argument( + fmt::format("invalid offset {} in offsets", entry.Offset())); + } + RealtimePartitionBucket partition_bucket(entry.Partition(), entry.Bucket()); + if (!offsets_.emplace(std::move(partition_bucket), entry.Offset()).second) { + throw std::invalid_argument( + fmt::format("duplicate partition-bucket {} in offsets", entry.Bucket())); + } + } + } + + const RealtimeOffsetMap& Offsets() const { + return offsets_; + } + + private: + RealtimeOffsetMap offsets_; +}; + +} // namespace + +void RealtimeCommitProperties::Sort(std::vector<RealtimeCommitProgress>* commits) { + std::stable_sort(commits->begin(), commits->end(), + [](const RealtimeCommitProgress& lhs, const RealtimeCommitProgress& rhs) { + if (!(lhs.partition_bucket == rhs.partition_bucket)) { + return lhs.partition_bucket < rhs.partition_bucket; + } + return lhs.offset_range.from < rhs.offset_range.from; + }); +} + +std::string RealtimeCommitProperties::OffsetsDirectory(const std::string& table_root, + const std::string& branch) { + return PathUtil::JoinPath(BranchManager::BranchPath(table_root, branch), "metadata"); +} + +Result<RealtimeOffsetMap> RealtimeCommitProperties::ReadOffsets( + const std::optional<Snapshot>& snapshot, const std::shared_ptr<FileSystem>& file_system) { + if (!snapshot || !snapshot->Properties()) { + return RealtimeOffsetMap{}; + } + const std::map<std::string, std::string>& properties = snapshot->Properties().value(); + auto iter = properties.find(kOffsetsKey); + if (iter == properties.end()) { + return RealtimeOffsetMap{}; + } + if (file_system == nullptr) { + return Status::Invalid("file system is null when reading real-time offsets"); + } + std::string content; + PAIMON_RETURN_NOT_OK(file_system->ReadFile(iter->second, &content)); + return ParseOffsets(content); +} + +Result<std::string> RealtimeCommitProperties::SerializeOffsets(const RealtimeOffsetMap& offsets) { + std::string result; + PAIMON_RETURN_NOT_OK(RapidJsonUtil::ToJsonString(OffsetsJson(offsets), &result)); + return result; +} + +Result<std::map<std::string, std::string>> RealtimeCommitProperties::Build( + const std::map<std::string, std::string>& properties, + const std::optional<Snapshot>& latest_snapshot, + const std::map<RealtimePartitionBucket, Range>& realtime_ranges, + const std::shared_ptr<FileSystem>& file_system, const std::string& table_root, + const std::string& branch) { + std::map<std::string, std::string> merged_properties = properties; + if (realtime_ranges.empty()) { + if (latest_snapshot && latest_snapshot->Properties()) { + const std::map<std::string, std::string>& latest_properties = + latest_snapshot->Properties().value(); + auto offsets_iter = latest_properties.find(kOffsetsKey); + if (offsets_iter != latest_properties.end()) { + merged_properties[kOffsetsKey] = offsets_iter->second; + } + } + return merged_properties; + } + + PAIMON_ASSIGN_OR_RAISE(RealtimeOffsetMap merged_offsets, + ReadOffsets(latest_snapshot, file_system)); + for (const auto& [partition_bucket, offset_range] : realtime_ranges) { + if (partition_bucket.bucket < 0) { + return Status::Invalid( + fmt::format("real-time commit bucket {} is invalid", partition_bucket.bucket)); + } + if (offset_range.from > offset_range.to) { + return Status::Invalid("real-time commit offset range is invalid"); + } + auto offset_iter = merged_offsets.find(partition_bucket); + int64_t previous_offset = offset_iter == merged_offsets.end() ? -1 : offset_iter->second; + if (previous_offset == std::numeric_limits<int64_t>::max() || + offset_range.from != previous_offset + 1) { + return Status::Invalid( + fmt::format("real-time commit offsets for bucket {} are not contiguous", + partition_bucket.bucket)); + } + merged_offsets[partition_bucket] = offset_range.to; + } + PAIMON_ASSIGN_OR_RAISE( + merged_properties[kOffsetsKey], + WriteOffsets(merged_offsets, file_system, OffsetsDirectory(table_root, branch))); + return merged_properties; +} + +Result<std::string> RealtimeCommitProperties::WriteOffsets( + const RealtimeOffsetMap& offsets, const std::shared_ptr<FileSystem>& file_system, + const std::string& offsets_directory) { + if (file_system == nullptr) { + return Status::Invalid("file system is null when writing real-time offsets"); + } + if (offsets_directory.empty()) { + return Status::Invalid("real-time offsets directory is empty"); + } + std::string uuid; + if (!UUID::Generate(&uuid)) { + return Status::Invalid("fail to generate uuid for real-time offsets file"); + } + PAIMON_RETURN_NOT_OK(file_system->Mkdirs(offsets_directory)); + std::string path = PathUtil::JoinPath(offsets_directory, uuid + ".offsets"); Review Comment: The offsets file reference written into snapshot properties is an absolute path, and `ReadOffsets` reads it back directly (`:174`). This couples snapshot metadata to the current table location: after the table directory is relocated, restored elsewhere, or the mount point changes, old snapshots can no longer resolve their offsets file. Every other snapshot metadata reference in this repo stores a bare file name and resolves it through `FileStorePathFactory` — see `ToManifestListPath` in `src/paimon/core/utils/file_store_path_factory.h:134`. This is the only reference that does not follow that convention, and it is persisted format, so the compatibility cost grows once it ships. -- 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]
