HaHaJeff commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3873131418
##########
src/paimon/common/table/special_fields.h:
##########
@@ -35,6 +36,10 @@ struct SpecialFields {
static constexpr char KEY_FIELD_PREFIX[] = "_KEY_";
static constexpr int32_t KEY_VALUE_SPECIAL_FIELD_COUNT = 2;
+ static constexpr int32_t kPreparedKeyValueValueKindIndex = 0;
+ static constexpr int32_t kPreparedKeyValueSequenceNumberIndex = 1;
+ static constexpr int32_t kPreparedKeyValueRealtimeOffsetIndex = 2;
+ static constexpr int32_t kPreparedKeyValueValueStartIndex = 3;
Review Comment:
These indexes are specific to the PK realtime transport layout, so I moved
them out of `SpecialFields` and kept them with the corresponding schema
construction and validation in `RealtimePrimaryKeyLayout`. Fixed in
`982eb59256ebf14d943e202950b35ad30b8fc671`.
##########
src/paimon/common/table/special_fields.h:
##########
@@ -85,6 +97,16 @@ struct SpecialFields {
target_fields.insert(target_fields.end(), schema->fields().begin(),
schema->fields().end());
return arrow::schema(target_fields);
}
+
+ static std::shared_ptr<arrow::Schema> PreparedKeyValueSchema(
+ const arrow::FieldVector& value_fields) {
Review Comment:
The previous name was too generic. The helper is now
`RealtimePrimaryKeyLayout::CreateSchema`, which makes the intended scope
explicit. Updated in `982eb59256ebf14d943e202950b35ad30b8fc671`.
##########
src/paimon/core/realtime/prepared_key_value_reader.cpp:
##########
@@ -0,0 +1,577 @@
+/*
+ * 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 <optional>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "arrow/array/array_base.h"
+#include "arrow/array/array_primitive.h"
+#include "arrow/c/bridge.h"
+#include "arrow/type.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/status_utils.h"
+#include "paimon/common/utils/checked_cast.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"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+
+namespace {
+
+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();
+ }
+ }
+}
+
+class RealtimeOffsetCoverage {
+ public:
+ static Result<std::shared_ptr<RealtimeOffsetCoverage>> Create(const
OffsetRange& offsets,
+ size_t
reader_count,
+ bool
allow_committed_prefix) {
+ if (offsets.begin < 0 || offsets.end < offsets.begin) {
+ return Status::Invalid("PK real-time store returned an invalid
offset range");
+ }
+ return std::shared_ptr<RealtimeOffsetCoverage>(
+ new RealtimeOffsetCoverage(offsets, reader_count,
allow_committed_prefix));
+ }
+
+ Status Add(const arrow::Int64Array& offsets) {
+ for (int64_t row = 0; row < offsets.length(); ++row) {
+ const int64_t offset = offsets.Value(row);
+ if (allow_committed_prefix_ && offset < 0) {
+ return Status::Invalid("PK real-time store reader offset must
be non-negative");
+ }
+ if (allow_committed_prefix_ && offset < offsets_.begin) {
+ continue;
+ }
+ if (offset < offsets_.begin || offset >= offsets_.end) {
+ return Status::Invalid(
+ allow_committed_prefix_
+ ? "PK real-time store query reader offset is outside
the visible range"
+ : "PK real-time store commit reader offset is outside
the sealed range");
+ }
+ if (!seen_offsets_.CheckedAdd(offset)) {
+ return CoverageError();
+ }
+ }
+ return Status::OK();
+ }
+
+ Status FinishReader() {
+ ++finished_reader_count_;
+ if (finished_reader_count_ == reader_count_ &&
+ seen_offsets_.Cardinality() != offsets_.Count()) {
+ return CoverageError();
+ }
+ return Status::OK();
+ }
+
+ private:
+ RealtimeOffsetCoverage(const OffsetRange& offsets, size_t reader_count,
+ bool allow_committed_prefix)
+ : offsets_(offsets),
+ reader_count_(reader_count),
+ allow_committed_prefix_(allow_committed_prefix) {}
+
+ Status CoverageError() const {
+ return Status::Invalid(
+ allow_committed_prefix_
+ ? "PK real-time store query readers did not cover the visible
range"
+ : "PK real-time store commit readers did not cover the sealed
range");
+ }
+
+ OffsetRange offsets_;
+ size_t reader_count_;
+ bool allow_committed_prefix_;
+ RoaringBitmap64 seen_offsets_;
+ size_t finished_reader_count_ = 0;
+};
+
+Status CheckPreparedField(const std::shared_ptr<arrow::Schema>& schema,
int32_t field_idx,
+ const DataField& expected_field) {
+ if (schema->num_fields() <= field_idx) {
+ return Status::Invalid(fmt::format("prepared schema missing transport
field {} at index {}",
+ expected_field.Name(), field_idx));
+ }
+ const std::shared_ptr<arrow::Field>& field = schema->field(field_idx);
+ PAIMON_ASSIGN_OR_RAISE(int32_t field_id,
NestedProjectionUtils::GetPaimonFieldId(field));
+ if (field->name() != expected_field.Name() ||
!field->type()->Equals(*expected_field.Type()) ||
+ field->nullable() || field_id != expected_field.Id()) {
+ return Status::Invalid(fmt::format(
+ "prepared schema field {} must be non-null {}:{} with field id {},
got {}:{} "
+ "nullable={} field id {}",
+ field_idx, expected_field.Name(),
expected_field.Type()->ToString(),
+ expected_field.Id(), field->name(), field->type()->ToString(),
field->nullable(),
+ field_id));
+ }
+ return Status::OK();
+}
+
+Result<std::vector<int32_t>> ResolveFieldIndexes(
+ const std::shared_ptr<arrow::Schema>& prepared_schema,
+ const std::unordered_map<int32_t, int32_t>& field_indexes,
+ const std::shared_ptr<arrow::Schema>& row_schema) {
+ std::vector<int32_t> result;
+ result.reserve(row_schema->num_fields());
+ for (const std::shared_ptr<arrow::Field>& row_field :
row_schema->fields()) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t field_id,
+
NestedProjectionUtils::GetPaimonFieldId(row_field));
+ auto field_index = field_indexes.find(field_id);
+ if (field_index == field_indexes.end()) {
+ return Status::Invalid(
+ fmt::format("cannot find field id {} in prepared schema",
field_id));
+ }
+ const std::shared_ptr<arrow::Field>& prepared_field =
+ prepared_schema->field(field_index->second);
+ if (!prepared_field->type()->Equals(row_field->type())) {
+ return Status::Invalid(fmt::format(
+ "prepared field id {} type {} does not match row "
+ "type {}",
+ field_id, prepared_field->type()->ToString(),
row_field->type()->ToString()));
+ }
+ result.push_back(field_index->second);
+ }
+ return result;
+}
+
+Status ValidateReaderParameters(const std::shared_ptr<arrow::Schema>&
prepared_schema,
+ const std::shared_ptr<arrow::Schema>&
key_schema,
+ const std::shared_ptr<arrow::Schema>&
value_schema,
+ const std::shared_ptr<MemoryPool>&
memory_pool) {
+
PAIMON_RETURN_NOT_OK(PreparedKeyValueReaderFactory::ValidateTransportSchema(prepared_schema));
+ if (!key_schema) {
+ return Status::Invalid("prepared key schema cannot be null");
+ }
+ if (!value_schema) {
+ return Status::Invalid("prepared value schema cannot be null");
+ }
+ if (!memory_pool) {
+ return Status::Invalid("prepared reader memory pool cannot be null");
+ }
+ return Status::OK();
+}
+
+Status ValidateExactCommitSchema(const std::shared_ptr<arrow::Schema>&
prepared_schema,
+ const std::shared_ptr<arrow::Schema>&
value_schema) {
+ if (prepared_schema->num_fields() !=
+ value_schema->num_fields() +
SpecialFields::kPreparedKeyValueValueStartIndex) {
+ return Status::Invalid("commit requires the exact prepared writer
schema");
+ }
+ for (int32_t i = 0; i < value_schema->num_fields(); ++i) {
+ if (!prepared_schema->field(i +
SpecialFields::kPreparedKeyValueValueStartIndex)
+ ->Equals(value_schema->field(i), true)) {
+ return Status::Invalid("commit requires the exact prepared writer
schema");
+ }
+ }
+ return Status::OK();
+}
+
+class PreparedReaderPlan {
+ public:
+ static Result<std::shared_ptr<const PreparedReaderPlan>> Create(
+ const std::shared_ptr<arrow::Schema>& prepared_schema,
+ const std::shared_ptr<arrow::Schema>& key_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema) {
+ std::unordered_map<int32_t, int32_t> field_indexes;
+ field_indexes.reserve(prepared_schema->num_fields() -
+ SpecialFields::kPreparedKeyValueValueStartIndex);
+ for (int32_t i = SpecialFields::kPreparedKeyValueValueStartIndex;
+ i < prepared_schema->num_fields(); ++i) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t field_id,
NestedProjectionUtils::GetPaimonFieldId(
+
prepared_schema->field(i)));
+ if (!field_indexes.emplace(field_id, i).second) {
+ return Status::Invalid(
+ fmt::format("duplicate field id {} in prepared schema",
field_id));
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> key_field_indexes,
+ ResolveFieldIndexes(prepared_schema,
field_indexes, key_schema));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> value_field_indexes,
+ ResolveFieldIndexes(prepared_schema,
field_indexes, value_schema));
+ return std::shared_ptr<const PreparedReaderPlan>(new
PreparedReaderPlan(
+ prepared_schema, std::move(key_field_indexes),
std::move(value_field_indexes)));
+ }
+
+ const std::shared_ptr<arrow::Schema>& PreparedSchema() const {
+ return prepared_schema_;
+ }
+
+ const std::vector<int32_t>& KeyFieldIndexes() const {
+ return key_field_indexes_;
+ }
+
+ const std::vector<int32_t>& ValueFieldIndexes() const {
+ return value_field_indexes_;
+ }
+
+ private:
+ PreparedReaderPlan(const std::shared_ptr<arrow::Schema>& schema,
+ std::vector<int32_t>&& key_indexes,
std::vector<int32_t>&& value_indexes)
+ : prepared_schema_(schema),
+ key_field_indexes_(std::move(key_indexes)),
+ value_field_indexes_(std::move(value_indexes)) {}
+
+ const std::shared_ptr<arrow::Schema> prepared_schema_;
+ const std::vector<int32_t> key_field_indexes_;
+ const std::vector<int32_t> value_field_indexes_;
+};
+
+class PreparedKeyValueReader final : public KeyValueRecordReader {
+ public:
+ PreparedKeyValueReader(std::unique_ptr<BatchReader>&& reader,
+ const std::shared_ptr<const PreparedReaderPlan>&
plan,
+ const std::optional<OffsetRange>& visible_offsets,
+ const std::shared_ptr<MemoryPool>& pool,
+ const std::shared_ptr<RealtimeOffsetCoverage>&
offset_coverage)
+ : reader_(std::move(reader)),
+ plan_(plan),
+ visible_offsets_(visible_offsets),
+ pool_(pool),
+ offset_coverage_(offset_coverage) {}
+
+ ~PreparedKeyValueReader() override {
+ Close();
+ }
+
+ class Iterator final : public KeyValueRecordReader::Iterator {
+ public:
+ explicit Iterator(PreparedKeyValueReader* reader) : reader_(reader) {}
+
+ Result<bool> HasNext() const override {
+ return cursor_ < reader_->RowCount();
+ }
+
+ Result<KeyValue> Next() override {
+ if (cursor_ >= reader_->RowCount()) {
+ return Status::Invalid("No more prepared key values in current
iterator");
+ }
+ const int64_t row = reader_->RowAt(cursor_);
+ std::shared_ptr<InternalRow> key =
+ std::make_shared<ColumnarRowRef>(reader_->key_ctx_, row);
+ auto value = std::make_unique<ColumnarRowRef>(reader_->value_ctx_,
row);
+ PAIMON_ASSIGN_OR_RAISE(const RowKind* row_kind,
+
RowKind::FromByteValue(reader_->row_kind_array_->Value(row)));
+ int64_t sequence_number =
reader_->sequence_number_array_->Value(row);
+ ++cursor_;
+ return KeyValue(row_kind, sequence_number,
KeyValue::UNKNOWN_LEVEL, std::move(key),
+ std::move(value));
+ }
+
+ private:
+ PreparedKeyValueReader* reader_;
+ int64_t cursor_ = 0;
+ };
+
+ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch()
override {
+ if (first_error_.has_value()) {
+ return first_error_.value();
+ }
+ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> result =
NextBatchImpl();
+ if (!result.ok()) {
+ first_error_ = result.status();
+ Close();
+ }
+ return result;
+ }
+
+ std::shared_ptr<Metrics> GetReaderMetrics() const override {
+ return reader_->GetReaderMetrics();
+ }
+
+ void Close() override {
+ if (closed_) {
+ return;
+ }
+ closed_ = true;
+ ResetBatchState();
+ reader_->Close();
+ }
+
+ private:
+ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatchImpl() {
+ while (true) {
+ ResetBatchState();
+ BatchReader::ReadBatchWithBitmap batch_with_bitmap;
+ if (visible_offsets_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(batch_with_bitmap,
reader_->NextBatchWithBitmap());
+ } else {
+ PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch,
reader_->NextBatch());
+ batch_with_bitmap.first = std::move(batch);
+ }
Review Comment:
`NextBatchWithBitmap()` is now used consistently, regardless of whether a
visible offset range is present. `_REALTIME_OFFSET` filtering is applied
separately afterward. Fixed in `982eb59256ebf14d943e202950b35ad30b8fc671`.
--
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]
