wangyong9999 commented on code in PR #224: URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3850131700
########## 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: Each output batch copies its rows three times: one `Take` per contributing source, then `Concatenate`, then a second `Take` to restore merge order. The store keeps one `StoredBatch` per `Write` call and `CreateQueryReaders` merges every batch in the view, so a CDC-style writer with many small batches lands in this multi-source path nearly every time. `Take` accepts a ChunkedArray with indices spanning chunks, so keeping the sources as one ChunkedArray and taking `chunk_base[source] + row` in a single call yields the same output with one copy and less code. ########## 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: `ValidatePrimaryKeyRealtimeOptions` rejects every other unsupported option up front, but compaction options are silently dropped here instead: `commit.force-compact` becomes a no-op (`NoopCompactManager::GetCompactionResult` returns empty even when blocking), `num-sorted-run.stop-trigger` stops applying backpressure (`ShouldWaitFor*` always false), and level-0 runs grow unbounded until an external compactor runs. The read side pays for that growth directly, since PK realtime folds every disk split of a bucket into a single merge. Rejecting explicitly-set compaction options in the validator would make this behave like the other unsupported options instead of quietly ignoring user configuration. -- 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]
