HaHaJeff commented on code in PR #224: URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3852189318
########## src/paimon/core/realtime/prepared_key_value_reader.cpp: ########## @@ -0,0 +1,771 @@ +/* + * 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/prepared_key_value_reader.h" + +#include <cstdint> +#include <memory> +#include <mutex> +#include <optional> +#include <unordered_map> +#include <utility> +#include <vector> + +#include "arrow/array/array_base.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/array_primitive.h" +#include "arrow/array/builder_primitive.h" +#include "arrow/buffer.h" +#include "arrow/c/bridge.h" +#include "arrow/compute/api.h" +#include "arrow/type.h" +#include "arrow/util/bit_util.h" +#include "fmt/format.h" +#include "paimon/common/data/columnar/columnar_batch_context.h" +#include "paimon/common/data/columnar/columnar_row_ref.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/types/row_kind.h" +#include "paimon/common/utils/arrow/arrow_utils.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/fields_comparator.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/utils/nested_projection_utils.h" +#include "paimon/macros.h" +#include "paimon/reader/batch_reader.h" +#include "paimon/status.h" + +namespace paimon { + +namespace { + +constexpr int32_t kValueKindIndex = 0; +constexpr int32_t kSequenceNumberIndex = 1; +constexpr int32_t kRealtimeOffsetIndex = 2; +constexpr int32_t kPreparedValueStartIndex = 3; + +template <typename Reader> +void CloseReaders(const std::vector<std::unique_ptr<Reader>>& readers) { + for (const std::unique_ptr<Reader>& reader : readers) { + if (reader) { + reader->Close(); + } + } +} + +Result<std::shared_ptr<arrow::Array>> AlignArrayByPaimonIds( + const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::DataType>& read_type, + arrow::MemoryPool* arrow_pool); + +class RealtimeOffsetCoverage { + public: + static Result<std::shared_ptr<RealtimeOffsetCoverage>> Create( + const OffsetRange& sealed_offsets, size_t reader_count, + const std::shared_ptr<arrow::MemoryPool>& arrow_pool) { + if (sealed_offsets.begin < 0 || sealed_offsets.end < sealed_offsets.begin) { + return Status::Invalid("PK real-time store returned an invalid sealed offset range"); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Buffer> seen_offsets, + arrow::AllocateEmptyBitmap(sealed_offsets.Count(), arrow_pool.get())); + return std::shared_ptr<RealtimeOffsetCoverage>(new RealtimeOffsetCoverage( + sealed_offsets, reader_count, std::move(seen_offsets), arrow_pool)); + } + + Status Add(const arrow::Int64Array& offsets) { Review Comment: The bitmap was originally introduced to prove that multiple commit readers returned every sealed offset exactly once. As noted, it added allocation, locking, and per-row bookkeeping while providing no result when a merge was abandoned. Commit 66e0b03b81ed400b7366125ab13830b10a460b31 removes the bitmap and keeps range bounds plus a lightweight count/min/max sanity check. Exact reader output remains part of the RealtimeStore contract; this check is intentionally not described as exact duplicate detection. ########## src/paimon/core/realtime/primary_key_realtime_store.cpp: ########## @@ -0,0 +1,509 @@ +/* + * 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/primary_key_realtime_store.h" + +#include <cstddef> +#include <mutex> +#include <optional> +#include <queue> +#include <unordered_map> +#include <utility> +#include <vector> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "arrow/compute/api.h" +#include "paimon/common/data/columnar/columnar_batch_context.h" +#include "paimon/common/data/columnar/columnar_row_ref.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/fields_comparator.h" +#include "paimon/core/core_options.h" +#include "paimon/core/index/pk/primary_key_index_definitions.h" +#include "paimon/core/realtime/prepared_key_value_reader.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/macros.h" + +namespace paimon { + +Status ValidatePrimaryKeyRealtimeOptions(const CoreOptions& options, const TableSchema& schema) { + if (options.GetBucket() <= 0) { + return Status::NotImplemented("PK realtime v1 requires fixed buckets"); + } + if (options.GetMergeEngine() != MergeEngine::DEDUPLICATE) { + return Status::NotImplemented("PK realtime v1 supports only the DEDUPLICATE merge engine"); + } + if (options.DataEvolutionEnabled()) { + return Status::NotImplemented("PK realtime v1 does not support data evolution"); + } + if (!options.GetFieldsSequenceGroups().empty()) { + return Status::NotImplemented("PK realtime v1 does not support sequence groups"); + } + if (options.IgnoreDelete() || options.PartialUpdateRemoveRecordOnDelete() || + options.AggregationRemoveRecordOnDelete() || + !options.GetPartialUpdateRemoveRecordOnSequenceGroup().empty()) { + return Status::NotImplemented("PK realtime v1 requires default delete behavior"); + } + if (!options.GetSequenceField().empty()) { + return Status::NotImplemented("PK realtime v1 does not support sequence.field"); + } + if (!options.SequenceFieldSortOrderIsAscending()) { + return Status::NotImplemented( + "PK realtime v1 supports only ascending sequence.field.sort-order"); + } + if (options.NeedLookup() || options.DeletionVectorsEnabled() || + options.GetChangelogProducer() != ChangelogProducer::NONE) { + return Status::NotImplemented("PK realtime v1 does not support lookup or early MOR"); + } + PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> primary_key_fields, + schema.TrimmedPrimaryKeyFields()); + for (const DataField& field : primary_key_fields) { + if (field.Type()->id() == arrow::Type::FLOAT || field.Type()->id() == arrow::Type::DOUBLE) { + return Status::NotImplemented( + "PK realtime v1 does not support FLOAT or DOUBLE primary keys"); + } + } + if (options.GlobalIndexEnabled()) { + PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(schema)); + if (!definitions.Definitions().empty()) { + return Status::NotImplemented("PK realtime v1 does not support global indexes"); + } + } + return Status::OK(); +} + +namespace { + +uint64_t GetArrayMemoryUsage(const std::shared_ptr<arrow::ArrayData>& data) { + uint64_t total = 0; + for (const std::shared_ptr<arrow::Buffer>& buffer : data->buffers) { + if (buffer) { + total += static_cast<uint64_t>(buffer->size()); + } + } + for (const std::shared_ptr<arrow::ArrayData>& child : data->child_data) { + total += GetArrayMemoryUsage(child); + } + if (data->dictionary) { + total += GetArrayMemoryUsage(data->dictionary); + } + return total; +} + +struct StoredBatch { + std::shared_ptr<arrow::StructArray> data; + OffsetRange offset_range; + uint64_t memory_usage; +}; + +class Segment final : public RealtimeSegmentHandle { + public: + Segment(const OffsetRange& range, std::vector<StoredBatch>&& batches) + : range_(range), batches_(std::move(batches)) {} + + OffsetRange GetOffsetRange() const override { + return range_; + } + const std::vector<StoredBatch>& Batches() const { + return batches_; + } + + private: + OffsetRange range_; + std::vector<StoredBatch> batches_; +}; + +class ReadView final : public RealtimeReadView { + public: + explicit ReadView(std::vector<std::shared_ptr<Segment>>&& segments) + : segments_(std::move(segments)) { + if (!segments_.empty()) { + range_ = OffsetRange(segments_.front()->GetOffsetRange().begin, + segments_.back()->GetOffsetRange().end); + } + } + + std::optional<OffsetRange> GetOffsetRange() const override { + return range_; + } + const std::vector<std::shared_ptr<Segment>>& Segments() const { + return segments_; + } + + private: + std::vector<std::shared_ptr<Segment>> segments_; + std::optional<OffsetRange> range_; +}; + +class RawBatchReader final : public BatchReader { + public: + RawBatchReader(std::vector<StoredBatch> batches, std::vector<int32_t> key_field_indexes, + const std::shared_ptr<FieldsComparator>& key_comparator, + const std::shared_ptr<MemoryPool>& memory_pool) + : batches_(std::move(batches)), + positions_(batches_.size(), 0), + key_field_indexes_(std::move(key_field_indexes)), + key_comparator_(key_comparator), + memory_pool_(memory_pool), + arrow_pool_(GetArrowPool(memory_pool)), + heap_(SourceGreater{this}), + metrics_(std::make_shared<MetricsImpl>()) { + key_contexts_.reserve(batches_.size()); + sequence_arrays_.reserve(batches_.size()); + for (size_t i = 0; i < batches_.size(); ++i) { + const StoredBatch& batch = batches_[i]; + arrow::ArrayVector key_arrays; + key_arrays.reserve(key_field_indexes_.size()); + for (int32_t field_index : key_field_indexes_) { + key_arrays.push_back(batch.data->field(field_index)); + } + key_contexts_.push_back( + std::make_shared<ColumnarBatchContext>(key_arrays, memory_pool_)); + sequence_arrays_.push_back( + checked_pointer_cast<arrow::Int64Array>(batch.data->field(1))); + if (batch.data->length() > 0) { + heap_.push(i); + } + } + } + + Result<ReadBatch> NextBatch() override { + if (heap_.empty()) { + return MakeEofBatch(); + } + + struct SelectedRow { + size_t selected_source; + int64_t source_ordinal; + }; + struct SelectedSource { + size_t source; + std::vector<int64_t> rows; + int64_t base = -1; + }; + std::vector<SelectedRow> selected_rows; + selected_rows.reserve(kOutputBatchSize); + std::vector<SelectedSource> selected_sources; + std::unordered_map<size_t, size_t> selected_source_indexes; + while (!heap_.empty() && selected_rows.size() < kOutputBatchSize) { + const size_t source = heap_.top(); + heap_.pop(); + auto [source_it, inserted] = + selected_source_indexes.emplace(source, selected_sources.size()); + if (inserted) { + selected_sources.push_back(SelectedSource{source, {}}); + } + SelectedSource& selected_source = selected_sources[source_it->second]; + selected_rows.push_back( + SelectedRow{source_it->second, static_cast<int64_t>(selected_source.rows.size())}); + selected_source.rows.push_back(positions_[source]++); + if (positions_[source] < batches_[source].data->length()) { + heap_.push(source); + } + } + + arrow::compute::ExecContext context(arrow_pool_.get()); + arrow::ArrayVector grouped_batches; + int64_t grouped_row_count = 0; + for (SelectedSource& selected_source : selected_sources) { + arrow::Int64Builder source_index_builder(arrow_pool_.get()); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + source_index_builder.AppendValues(selected_source.rows)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> source_indices, + source_index_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + arrow::Datum source_batch, + arrow::compute::Take(arrow::Datum(batches_[selected_source.source].data), + arrow::Datum(source_indices), + arrow::compute::TakeOptions::NoBoundsCheck(), &context)); + selected_source.base = grouped_row_count; + grouped_row_count += static_cast<int64_t>(selected_source.rows.size()); + grouped_batches.push_back(source_batch.make_array()); + } + + std::shared_ptr<arrow::Array> batch; + if (grouped_batches.size() == 1) { + batch = std::move(grouped_batches[0]); + } else { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Array> grouped, + arrow::Concatenate(grouped_batches, arrow_pool_.get())); Review Comment: The store-side heap merge was originally introduced as a performance optimization: it bounded the number of readers returned for many small stored batches and reduced downstream merge inputs. However, it also required repeated Take/Concatenate operations and coupled the store to PK ordering and merge behavior. Commit ff444191efa75ec4fe292dfb52081f01121135d9 removes that duplicate merge layer and returns one zero-copy reader per prepared sorted batch. This trades higher reader cardinality for fewer copies and a storage-only plugin boundary. If reader cardinality becomes a measured bottleneck, sorted-run composition should be optimized in the framework instead. ########## src/paimon/core/operation/key_value_file_store_write.cpp: ########## @@ -109,19 +124,73 @@ Result<std::shared_ptr<BatchWriter>> KeyValueFileStoreWrite::CreateWriter( PAIMON_ASSIGN_OR_RAISE( std::shared_ptr<Levels> levels, Levels::Create(key_comparator_, restore_data_files, options_.GetNumLevels())); - auto compact_strategy = compact_manager_factory_->CreateCompactStrategy(); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr<CompactManager> compact_manager, - compact_manager_factory_->CreateCompactManager(partition, bucket, compact_strategy, - compact_executor_, levels, dv_maintainer)); + std::map<std::string, std::string> partition_map; + std::shared_ptr<CompactManager> compact_manager; + std::shared_ptr<RealtimeContextImpl> realtime_context_impl; + std::optional<RealtimeStoreState> realtime_store_state; + if (realtime_context_) { + std::vector<std::pair<std::string, std::string>> partition_values; + PAIMON_ASSIGN_OR_RAISE(partition_values, + file_store_path_factory_->GeneratePartitionVector(partition)); + partition_map = + std::map<std::string, std::string>(partition_values.begin(), partition_values.end()); + PAIMON_ASSIGN_OR_RAISE(realtime_context_impl, RealtimeContextImpl::Cast(realtime_context_)); + if (schema_->GetFieldByName(SpecialFields::RealtimeOffset().Name())) { + return Status::Invalid("PK real-time write schema contains reserved transport field " + + SpecialFields::RealtimeOffset().Name()); + } + arrow::FieldVector prepared_fields = { + DataField::ConvertDataFieldToArrowField(SpecialFields::ValueKind()) + ->WithNullable(false), + DataField::ConvertDataFieldToArrowField(SpecialFields::SequenceNumber()) + ->WithNullable(false), + DataField::ConvertDataFieldToArrowField(SpecialFields::RealtimeOffset())}; + prepared_fields.insert(prepared_fields.end(), schema_->fields().begin(), + schema_->fields().end()); + auto c_write_schema = std::make_unique<ArrowSchema>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow::ExportSchema(*arrow::schema(std::move(prepared_fields)), c_write_schema.get())); + PAIMON_ASSIGN_OR_RAISE( + RealtimeStoreState store_state, + realtime_context_impl->GetOrCreateRealtimeStore(RealtimeStoreCreateRequest{ + std::move(c_write_schema), options_.ToMap(), pool_, partition_map, bucket, + PrimaryKeyRealtimeStoreCreateConfig{trimmed_primary_keys}})); + realtime_store_state = std::move(store_state); + compact_manager = std::make_shared<NoopCompactManager>(); Review Comment: Thanks for pointing this out. The behavior is intentional and matches append realtime. With a realtime context, both writer paths use NoopCompactManager, reject explicit Compact(), and do not perform writer-local compaction during PrepareCommit. Compaction-related options are therefore not rejected; external compaction is responsible for rewriting accumulated files. For consistency with append realtime, this PR keeps the PK options unrejected. ########## include/paimon/realtime/realtime_store.h: ########## @@ -41,10 +43,42 @@ namespace paimon { class MemoryPool; class Predicate; -/// A table record batch and its framework-assigned contiguous offset range. +struct PAIMON_EXPORT AppendRealtimeStoreCreateConfig { + StatisticsMode statistics_mode; +}; + +struct PAIMON_EXPORT PrimaryKeyRealtimeStoreCreateConfig { + /// Primary-key fields after removing partition fields, in comparison order. + std::vector<std::string> trimmed_primary_keys; +}; Review Comment: Good point. The PK fields were originally passed to the store because it constructed a comparator and performed store-side sorting and merging. That no longer matches the storage-only plugin boundary. Commit ff444191efa75ec4fe292dfb52081f01121135d9 removes the PK fields and comparator dependency, and b2827df103f32c533b7595205fa9bcd7655b18c6 simplifies the remaining selection to RealtimeStoreMode. -- 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]
