This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 8f83fe1 feat: Migrate merge split read modules (#77)
8f83fe1 is described below
commit 8f83fe1b2867609a31a5791faa26236285343426
Author: lxy <[email protected]>
AuthorDate: Tue Jun 16 09:11:35 2026 +0800
feat: Migrate merge split read modules (#77)
---
.../core/operation/merge_file_split_read.cpp | 537 ++++++++
src/paimon/core/operation/merge_file_split_read.h | 191 +++
.../core/operation/merge_file_split_read_test.cpp | 1296 ++++++++++++++++++++
3 files changed, 2024 insertions(+)
diff --git a/src/paimon/core/operation/merge_file_split_read.cpp
b/src/paimon/core/operation/merge_file_split_read.cpp
new file mode 100644
index 0000000..ca6c1d5
--- /dev/null
+++ b/src/paimon/core/operation/merge_file_split_read.cpp
@@ -0,0 +1,537 @@
+/*
+ * 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/merge_file_split_read.h"
+
+#include <algorithm>
+#include <cassert>
+#include <cstddef>
+#include <map>
+#include <optional>
+#include <set>
+#include <utility>
+
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/reader/complete_row_kind_batch_reader.h"
+#include "paimon/common/reader/concat_batch_reader.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
+#include "paimon/core/deletionvectors/deletion_vector.h"
+#include "paimon/core/io/async_key_value_projection_reader.h"
+#include "paimon/core/io/concat_key_value_record_reader.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/key_value_data_file_record_reader.h"
+#include "paimon/core/io/key_value_projection_consumer.h"
+#include "paimon/core/io/key_value_projection_reader.h"
+#include "paimon/core/mergetree/compact/interval_partition.h"
+#include "paimon/core/mergetree/compact/lookup_merge_function.h"
+#include "paimon/core/mergetree/compact/merge_function.h"
+#include "paimon/core/mergetree/compact/partial_update_merge_function.h"
+#include "paimon/core/mergetree/compact/reducer_merge_function_wrapper.h"
+#include "paimon/core/mergetree/compact/sort_merge_reader_with_loser_tree.h"
+#include "paimon/core/mergetree/compact/sort_merge_reader_with_min_heap.h"
+#include "paimon/core/mergetree/drop_delete_reader.h"
+#include "paimon/core/mergetree/sorted_run.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/options/sort_engine.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/core/utils/primary_key_table_utils.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/predicate_utils.h"
+#include "paimon/reader/file_batch_reader.h"
+#include "paimon/table/source/data_split.h"
+#include "paimon/utils/roaring_bitmap32.h"
+
+namespace paimon {
+class BinaryRow;
+class DataFilePathFactory;
+class Executor;
+struct KeyValue;
+template <typename T>
+class MergeFunctionWrapper;
+
+Result<std::unique_ptr<MergeFileSplitRead>> MergeFileSplitRead::Create(
+ const std::shared_ptr<FileStorePathFactory>& path_factory,
+ const std::shared_ptr<InternalReadContext>& context,
+ const std::shared_ptr<MemoryPool>& memory_pool, const
std::shared_ptr<Executor>& executor) {
+ const auto& core_options = context->GetCoreOptions();
+ const auto& table_schema = context->GetTableSchema();
+ assert(table_schema);
+ // value_schema is the schema of member value in KeyValue Object
+ std::shared_ptr<arrow::Schema> value_schema;
+ // read_schema is the read schema for format file reader (e.g., includes
_SEQUENCE_NUMBER)
+ std::shared_ptr<arrow::Schema> read_schema;
+ // comparator of member key in KeyValue object
+ std::shared_ptr<FieldsComparator> key_comparator;
+ // comparator of user-defined sequence fields in member value of KeyValue
object
+ std::shared_ptr<FieldsComparator> user_defined_seq_comparator;
+
+ PAIMON_RETURN_NOT_OK(GenerateKeyValueReadSchema(
+ *table_schema, core_options, context->GetReadSchema(), &value_schema,
&read_schema,
+ &key_comparator, &user_defined_seq_comparator));
+
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Predicate> predicate_for_keys,
+ GenerateKeyPredicates(context->GetPredicate(),
*table_schema));
+
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> key_schema,
+ table_schema->TrimmedPrimaryKeySchema());
+
+ // projection is the mapping from value_schema in KeyValue object to
raw_read_schema
+ std::vector<int32_t> projection;
+ projection.reserve(context->GetReadSchema()->num_fields());
+ bool project_sequence_number =
+ core_options.RowTrackingEnabled() ||
core_options.KeyValueSequenceNumberEnabled();
+ for (const auto& field : context->GetReadSchema()->fields()) {
+ if (field->name() == SpecialFields::SequenceNumber().Name() &&
project_sequence_number) {
+
projection.push_back(KeyValueProjectionConsumer::kSequenceNumberProjection);
+ continue;
+ }
+ if (field->name() == SpecialFields::ValueKind().Name()) {
+
projection.push_back(KeyValueProjectionConsumer::kValueKindProjection);
+ continue;
+ }
+ auto src_field_idx = value_schema->GetFieldIndex(field->name());
+ if (src_field_idx < 0) {
+ return Status::Invalid(
+ fmt::format("Field '{}' not found or duplicate in value
schema", field->name()));
+ }
+ projection.push_back(src_field_idx);
+ }
+
+ return std::unique_ptr<MergeFileSplitRead>(new MergeFileSplitRead(
+ path_factory, context,
+ std::make_unique<SchemaManager>(core_options.GetFileSystem(),
context->GetPath(),
+ context->GetCoreOptions().GetBranch()),
+ key_schema, value_schema, read_schema, projection, key_comparator,
+ user_defined_seq_comparator, predicate_for_keys, memory_pool,
executor));
+}
+
+Result<std::unique_ptr<BatchReader>> MergeFileSplitRead::CreateReader(
+ const std::shared_ptr<Split>& split) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ if (!data_split) {
+ return Status::Invalid("cannot cast split to data_split in
MergeFileSplitRead");
+ }
+ if (!data_split->BeforeFiles().empty()) {
+ return Status::Invalid("this read cannot accept split with before
files.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<DataFilePathFactory> data_file_path_factory,
+ path_factory_->CreateDataFilePathFactory(data_split->Partition(),
data_split->Bucket()));
+ std::unique_ptr<BatchReader> batch_reader;
+ if (data_split->IsStreaming() || data_split->Bucket() ==
BucketModeDefine::POSTPONE_BUCKET) {
+ PAIMON_ASSIGN_OR_RAISE(
+ batch_reader,
+ CreateNoMergeReader(data_split,
/*only_filter_key=*/data_split->IsStreaming(),
+ data_file_path_factory));
+ } else {
+ PAIMON_ASSIGN_OR_RAISE(batch_reader, CreateMergeReader(data_split,
data_file_path_factory));
+ }
+ return
std::make_unique<CompleteRowKindBatchReader>(std::move(batch_reader), pool_);
+}
+
+void MergeFileSplitRead::SetMergeFunctionWrapper(
+ const std::shared_ptr<MergeFunctionWrapper<KeyValue>>&
merge_function_wrapper) {
+ merge_function_wrapper_ = merge_function_wrapper;
+}
+
+Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>>
+MergeFileSplitRead::GetMergeFunctionWrapper() {
+ if (!merge_function_wrapper_) {
+ // In deletion vector mode, streaming data split or postpone bucket
mode, we don't need
+ // to use merge function. Even if the merge function in CoreOptions is
not supported, it
+ // should not affect data reading. So we create
merge_function_wrapper_ lazily, to avoid
+ // raise errors when creating MergeFileSplitRead at the beginning.
+ PAIMON_ASSIGN_OR_RAISE(
+ merge_function_wrapper_,
+ CreateMergeFunctionWrapper(options_, context_->GetTableSchema(),
value_schema_));
+ }
+ return merge_function_wrapper_;
+}
+
+Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>>
+MergeFileSplitRead::CreateMergeFunctionWrapper(const CoreOptions& core_options,
+ const
std::shared_ptr<TableSchema>& table_schema,
+ const
std::shared_ptr<arrow::Schema>& value_schema) {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<MergeFunction> merge_function,
+ PrimaryKeyTableUtils::CreateMergeFunction(
+ value_schema, table_schema->PrimaryKeys(),
core_options));
+ if (core_options.NeedLookup() && core_options.GetMergeEngine() !=
MergeEngine::FIRST_ROW) {
+ // don't wrap first row, it is already OK
+ merge_function =
std::make_unique<LookupMergeFunction>(std::move(merge_function));
+ }
+ return
std::make_shared<ReducerMergeFunctionWrapper>(std::move(merge_function));
+}
+
+Result<std::unique_ptr<FileBatchReader>>
MergeFileSplitRead::ApplyIndexAndDvReaderIfNeeded(
+ std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
+ const std::shared_ptr<arrow::Schema>& data_schema,
+ const std::shared_ptr<arrow::Schema>& read_schema, const
std::shared_ptr<Predicate>& predicate,
+ DeletionVector::Factory dv_factory, const
std::optional<std::vector<Range>>& ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ // merge read does not use index
+ std::shared_ptr<DeletionVector> deletion_vector;
+ if (dv_factory) {
+ PAIMON_ASSIGN_OR_RAISE(deletion_vector, dv_factory(file->file_name));
+ }
+
+ const RoaringBitmap32* deletion = nullptr;
+ if (auto* bitmap_dv =
dynamic_cast<BitmapDeletionVector*>(deletion_vector.get())) {
+ deletion = bitmap_dv->GetBitmap();
+ }
+
+ std::optional<RoaringBitmap32> actual_selection;
+ if (deletion) {
+ actual_selection = *deletion;
+ PAIMON_ASSIGN_OR_RAISE(uint64_t num_rows,
file_reader->GetNumberOfRows());
+ actual_selection.value().Flip(0, num_rows);
+ }
+
+ ::ArrowSchema c_read_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema,
&c_read_schema));
+
+ PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, predicate,
actual_selection));
+
+ if (!file_reader->SupportPreciseBitmapSelection() && actual_selection) {
+ return
std::make_unique<ApplyDeletionVectorBatchReader>(std::move(file_reader),
+
deletion_vector);
+ }
+ if (deletion_vector && !deletion && !deletion_vector->IsEmpty()) {
+ // TODO(xinyu.lxy): if deletion vector is bitmap64, use
ApplyBitmapIndexBatchReader to
+ // filter result
+ return Status::NotImplemented("Only support BitmapDeletionVector");
+ }
+ return std::move(file_reader);
+}
+
+Result<std::unique_ptr<BatchReader>> MergeFileSplitRead::CreateMergeReader(
+ const std::shared_ptr<DataSplitImpl>& data_split,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) {
+ auto dv_factory = DeletionVector::CreateFactory(options_.GetFileSystem(),
+
CreateDeletionFileMap(*data_split), pool_);
+
+ std::vector<std::vector<SortedRun>> sections =
+ IntervalPartition(data_split->DataFiles(),
key_comparator_).Partition();
+ std::vector<std::unique_ptr<BatchReader>> batch_readers;
+ batch_readers.reserve(sections.size());
+ // no overlap through multiple sections
+ for (const auto& section : sections) {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> projection_reader,
+ CreateReaderForSection(section,
data_split->Partition(), dv_factory,
+ data_file_path_factory));
+ batch_readers.push_back(std::move(projection_reader));
+ }
+ auto concat_batch_reader =
std::make_unique<ConcatBatchReader>(std::move(batch_readers), pool_);
+ return
AbstractSplitRead::ApplyPredicateFilterIfNeeded(std::move(concat_batch_reader),
+
context_->GetPredicate());
+}
+
+Result<std::unique_ptr<BatchReader>> MergeFileSplitRead::CreateNoMergeReader(
+ const std::shared_ptr<DataSplitImpl>& data_split, bool only_filter_key,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ auto dv_factory = DeletionVector::CreateFactory(options_.GetFileSystem(),
+
CreateDeletionFileMap(*data_split), pool_);
+
+ // create read schema without extra fields (e.g., completed key, sequence
fields)
+ auto row_kind_field =
DataField::ConvertDataFieldToArrowField(SpecialFields::ValueKind());
+
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema>
read_schema,
+ raw_read_schema_->AddField(0,
row_kind_field));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers,
+ CreateRawFileReaders(data_split->Partition(), data_split->DataFiles(),
read_schema,
+ only_filter_key ? predicate_for_keys_ :
context_->GetPredicate(),
+ dv_factory, /*row_ranges=*/{},
data_file_path_factory));
+
+ auto raw_readers =
+
ObjectUtils::MoveVector<std::unique_ptr<BatchReader>>(std::move(raw_file_readers));
+ auto concat_batch_reader =
std::make_unique<ConcatBatchReader>(std::move(raw_readers), pool_);
+ return
AbstractSplitRead::ApplyPredicateFilterIfNeeded(std::move(concat_batch_reader),
+
context_->GetPredicate());
+}
+
+MergeFileSplitRead::MergeFileSplitRead(
+ const std::shared_ptr<FileStorePathFactory>& path_factory,
+ const std::shared_ptr<InternalReadContext>& context,
+ std::unique_ptr<SchemaManager>&& schema_manager,
+ const std::shared_ptr<arrow::Schema>& key_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema,
+ const std::shared_ptr<arrow::Schema>& read_schema, const
std::vector<int32_t>& projection,
+ const std::shared_ptr<FieldsComparator>& key_comparator,
+ const std::shared_ptr<FieldsComparator>& user_defined_seq_comparator,
+ const std::shared_ptr<Predicate>& predicate_for_keys,
+ const std::shared_ptr<MemoryPool>& memory_pool, const
std::shared_ptr<Executor>& executor)
+ : AbstractSplitRead(path_factory, context, std::move(schema_manager),
memory_pool, executor),
+ key_schema_(key_schema),
+ value_schema_(value_schema),
+ read_schema_(read_schema),
+ projection_(projection),
+ key_comparator_(key_comparator),
+ user_defined_seq_comparator_(user_defined_seq_comparator),
+ predicate_for_keys_(predicate_for_keys) {}
+
+Status MergeFileSplitRead::GenerateKeyValueReadSchema(
+ const TableSchema& table_schema, const CoreOptions& options,
+ const std::shared_ptr<arrow::Schema>& raw_read_schema,
+ std::shared_ptr<arrow::Schema>* value_schema,
std::shared_ptr<arrow::Schema>* read_schema,
+ std::shared_ptr<FieldsComparator>* key_comparator,
+ std::shared_ptr<FieldsComparator>* sequence_fields_comparator) {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> trimmed_key_fields,
+ table_schema.TrimmedPrimaryKeyFields());
+ PAIMON_ASSIGN_OR_RAISE(*key_comparator,
FieldsComparator::Create(trimmed_key_fields,
+
/*is_ascending_order=*/true));
+ const auto& table_fields = table_schema.Fields();
+ auto table_fields_schema =
DataField::ConvertDataFieldsToArrowSchema(table_fields);
+ if (table_fields_schema->Equals(raw_read_schema)) {
+ // Short-circuit: if raw_read_schema is the same as the table schema,
+ // use the table schema field order directly (for compact process).
+ *value_schema = table_fields_schema;
+ // sequence_fields_comparator
+ PAIMON_ASSIGN_OR_RAISE(
+ *sequence_fields_comparator,
+ PrimaryKeyTableUtils::CreateSequenceFieldsComparator(table_fields,
options));
+ *read_schema =
SpecialFields::CompleteSequenceAndValueKindField(*value_schema);
+ return Status::OK();
+ }
+
+ // 1. add user raw read schema to need_fields
+ PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> need_fields,
+
DataField::ConvertArrowSchemaToDataFields(raw_read_schema));
+ if (options.RowTrackingEnabled() ||
options.KeyValueSequenceNumberEnabled()) {
+ // _SEQUENCE_NUMBER is carried by KeyValue metadata, not by
KeyValue.value. Remove it before
+ // splitting key/value fields so the value projection can inject it
from KeyValue directly.
+ need_fields.erase(std::remove_if(need_fields.begin(),
need_fields.end(),
+ [](const DataField& field) {
+ return field.Name() ==
+
SpecialFields::SequenceNumber().Name();
+ }),
+ need_fields.end());
+ }
+ // _VALUE_KIND is also carried by KeyValue metadata. Keep it out of
KeyValue.value so the
+ // projection can inject the actual row kind instead of resolving it as a
table field.
+ need_fields.erase(std::remove_if(need_fields.begin(), need_fields.end(),
+ [](const DataField& field) {
+ return field.Name() ==
SpecialFields::ValueKind().Name();
+ }),
+ need_fields.end());
+ // 2. add user defined sequence field to need_fields
+ PAIMON_RETURN_NOT_OK(CompleteSequenceField(table_schema, options,
&need_fields));
+ if (options.GetMergeEngine() == MergeEngine::PARTIAL_UPDATE) {
+ // add sequence group fields for partial update
+ std::map<std::string, std::vector<std::string>>
value_field_to_seq_group_field;
+ std::set<std::string> seq_group_key_set;
+
PAIMON_RETURN_NOT_OK(PartialUpdateMergeFunction::ParseSequenceGroupFields(
+ options, &value_field_to_seq_group_field, &seq_group_key_set));
+
PAIMON_RETURN_NOT_OK(PartialUpdateMergeFunction::CompleteSequenceGroupFields(
+ table_schema, value_field_to_seq_group_field, &need_fields));
+ }
+ // 3. split need_fields to key and non-key fields
+ std::vector<DataField> key_fields;
+ std::vector<DataField> non_key_fields;
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> trimmed_key_names,
+ table_schema.TrimmedPrimaryKeys());
+ PAIMON_RETURN_NOT_OK(
+ SplitKeyAndNonKeyField(trimmed_key_names, need_fields, &key_fields,
&non_key_fields));
+
+ // 4. construct value fields: key fields are put before non-key fields
+ std::vector<DataField> value_fields;
+ value_fields.insert(value_fields.end(), key_fields.begin(),
key_fields.end());
+ value_fields.insert(value_fields.end(), non_key_fields.begin(),
non_key_fields.end());
+ *value_schema = DataField::ConvertDataFieldsToArrowSchema(value_fields);
+ // 5. create sequence field comparator
+ PAIMON_ASSIGN_OR_RAISE(
+ *sequence_fields_comparator,
+ PrimaryKeyTableUtils::CreateSequenceFieldsComparator(value_fields,
options));
+ // 6. construct actual read fields: special + key + non-key value
+ std::vector<DataField> read_fields = {SpecialFields::SequenceNumber(),
+ SpecialFields::ValueKind()};
+ read_fields.insert(read_fields.end(), trimmed_key_fields.begin(),
trimmed_key_fields.end());
+ read_fields.insert(read_fields.end(), non_key_fields.begin(),
non_key_fields.end());
+ *read_schema = DataField::ConvertDataFieldsToArrowSchema(read_fields);
+ return Status::OK();
+}
+
+Status MergeFileSplitRead::SplitKeyAndNonKeyField(
+ const std::vector<std::string>& trimmed_key_fields, const
std::vector<DataField>& read_fields,
+ std::vector<DataField>* key_fields, std::vector<DataField>*
non_key_fields) {
+ for (const auto& field : read_fields) {
+ auto iter = std::find(trimmed_key_fields.begin(),
trimmed_key_fields.end(), field.Name());
+ if (iter == trimmed_key_fields.end()) {
+ non_key_fields->push_back(field);
+ } else {
+ key_fields->push_back(field);
+ }
+ }
+ return Status::OK();
+}
+
+Status MergeFileSplitRead::CompleteSequenceField(const TableSchema&
table_schema,
+ const CoreOptions& options,
+ std::vector<DataField>*
non_key_fields) {
+ auto sequence_field_names = options.GetSequenceField();
+ if (sequence_field_names.empty()) {
+ return Status::OK();
+ }
+
+ std::set<std::string> non_key_field_names;
+ for (const auto& field : *non_key_fields) {
+ non_key_field_names.insert(field.Name());
+ }
+
+ for (const auto& seq_field_name : sequence_field_names) {
+ auto iter = non_key_field_names.find(seq_field_name);
+ if (iter == non_key_field_names.end()) {
+ // force add sequence fields
+ PAIMON_ASSIGN_OR_RAISE(DataField seq_field,
table_schema.GetField(seq_field_name));
+ non_key_fields->push_back(seq_field);
+ }
+ }
+ return Status::OK();
+}
+
+Result<std::shared_ptr<Predicate>> MergeFileSplitRead::GenerateKeyPredicates(
+ const std::shared_ptr<Predicate>& predicate, const TableSchema&
table_schema) {
+ // extract predicates only contain trimmed key fields
+ if (!predicate) {
+ return std::shared_ptr<Predicate>();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> trimmed_key_fields,
+ table_schema.TrimmedPrimaryKeys());
+ std::set<std::string> non_primary_keys;
+ for (const auto& field_name : table_schema.FieldNames()) {
+ auto iter = std::find(trimmed_key_fields.begin(),
trimmed_key_fields.end(), field_name);
+ if (iter == trimmed_key_fields.end()) {
+ non_primary_keys.insert(field_name);
+ }
+ }
+ return PredicateUtils::ExcludePredicateWithFields(predicate,
non_primary_keys);
+}
+
+Result<std::unique_ptr<BatchReader>>
MergeFileSplitRead::CreateReaderForSection(
+ const std::vector<SortedRun>& section, const BinaryRow& partition,
+ DeletionVector::Factory dv_factory,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) {
+ // with overlap in one section
+ std::shared_ptr<Predicate> predicate;
+ if (section.size() > 1) {
+ predicate = predicate_for_keys_;
+ } else {
+ predicate = context_->GetPredicate();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<SortMergeReader> sort_merge_reader,
+ CreateSortMergeReaderForSection(section, partition,
dv_factory,
+ predicate,
data_file_path_factory,
+
/*drop_delete=*/!force_keep_delete_));
+ // KeyValueProjectionReader converts KeyValue objects to arrow array
according to projection
+ if (!context_->EnableMultiThreadRowToBatch()) {
+ return KeyValueProjectionReader::Create(std::move(sort_merge_reader),
raw_read_schema_,
+ projection_,
options_.GetReadBatchSize(), pool_);
+ }
+ int32_t thread_number = context_->GetRowToBatchThreadNumber();
+ assert(thread_number > 0);
+ return std::make_unique<AsyncKeyValueProjectionReader>(
+ std::move(sort_merge_reader), raw_read_schema_, projection_,
options_.GetReadBatchSize(),
+ thread_number, pool_);
+}
+
+Result<std::unique_ptr<SortMergeReader>>
MergeFileSplitRead::CreateSortMergeReaderForSection(
+ const std::vector<SortedRun>& section, const BinaryRow& partition,
+ DeletionVector::Factory dv_factory, const std::shared_ptr<Predicate>&
predicate,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory, bool
drop_delete) {
+ // with overlap in one section
+ std::vector<std::unique_ptr<KeyValueRecordReader>> record_readers;
+ record_readers.reserve(section.size());
+ for (const auto& run : section) {
+ // no overlap in a run
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<KeyValueRecordReader> run_reader,
+ CreateReaderForRun(partition, run, dv_factory, predicate,
data_file_path_factory));
+ record_readers.emplace_back(std::move(run_reader));
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<SortMergeReader> sort_merge_reader,
+ CreateSortMergeReader(std::move(record_readers)));
+ if (drop_delete) {
+ sort_merge_reader =
std::make_unique<DropDeleteReader>(std::move(sort_merge_reader));
+ }
+ return sort_merge_reader;
+}
+
+Result<std::unique_ptr<KeyValueRecordReader>>
MergeFileSplitRead::CreateReaderForRun(
+ const BinaryRow& partition, const SortedRun& sorted_run,
DeletionVector::Factory dv_factory,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ // no overlap in a run
+ const auto& data_files = sorted_run.Files();
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers,
+ CreateRawFileReaders(partition, data_files, read_schema_, predicate,
dv_factory,
+ /*row_ranges=*/{}, data_file_path_factory));
+
+ assert(data_files.size() == raw_file_readers.size());
+ // KeyValueDataFileRecordReader converts arrow array from format reader to
KeyValue objects
+ std::vector<std::unique_ptr<KeyValueRecordReader>> file_record_readers;
+ file_record_readers.reserve(data_files.size());
+ for (size_t i = 0; i < data_files.size(); i++) {
+
file_record_readers.push_back(std::make_unique<KeyValueDataFileRecordReader>(
+ std::move(raw_file_readers[i]), key_schema_, value_schema_,
data_files[i]->level,
+ pool_));
+ }
+ return
std::make_unique<ConcatKeyValueRecordReader>(std::move(file_record_readers));
+}
+
+Result<std::unique_ptr<SortMergeReader>>
MergeFileSplitRead::CreateSortMergeReader(
+ std::vector<std::unique_ptr<KeyValueRecordReader>>&& record_readers) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<MergeFunctionWrapper<KeyValue>>
merge_function_wrapper,
+ GetMergeFunctionWrapper());
+ auto sort_engine = options_.GetSortEngine();
+ if (sort_engine == SortEngine::MIN_HEAP) {
+ return std::make_unique<SortMergeReaderWithMinHeap>(
+ std::move(record_readers), key_comparator_,
user_defined_seq_comparator_,
+ merge_function_wrapper);
+ } else if (sort_engine == SortEngine::LOSER_TREE) {
+ return std::make_unique<SortMergeReaderWithLoserTree>(
+ std::move(record_readers), key_comparator_,
user_defined_seq_comparator_,
+ merge_function_wrapper);
+ }
+ return Status::Invalid("only support loser-tree or min-heap sort engine");
+}
+
+Result<bool> MergeFileSplitRead::Match(const std::shared_ptr<Split>& split,
+ bool force_keep_delete) const {
+ // TODO(yonghao.fyh): just pass split impl
+ auto split_impl = dynamic_cast<DataSplitImpl*>(split.get());
+ if (split_impl == nullptr) {
+ return Status::Invalid("unexpected error, split cast to impl failed");
+ }
+ return split_impl->BeforeFiles().empty();
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/operation/merge_file_split_read.h
b/src/paimon/core/operation/merge_file_split_read.h
new file mode 100644
index 0000000..f00883b
--- /dev/null
+++ b/src/paimon/core/operation/merge_file_split_read.h
@@ -0,0 +1,191 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "paimon/common/utils/fields_comparator.h"
+#include "paimon/core/io/concat_key_value_record_reader.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/key_value_record_reader.h"
+#include "paimon/core/mergetree/compact/interval_partition.h"
+#include "paimon/core/mergetree/compact/merge_function_wrapper.h"
+#include "paimon/core/mergetree/compact/sort_merge_reader.h"
+#include "paimon/core/mergetree/sorted_run.h"
+#include "paimon/core/operation/abstract_split_read.h"
+#include "paimon/core/table/source/deletion_file.h"
+#include "paimon/predicate/predicate.h"
+#include "paimon/reader/batch_reader.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class Schema;
+} // namespace arrow
+
+namespace paimon {
+class BinaryRow;
+class CoreOptions;
+class DataField;
+class DataFilePathFactory;
+class DataSplit;
+class DataSplitImpl;
+class Executor;
+class FieldsComparator;
+class FileBatchReader;
+class FileStorePathFactory;
+class InternalReadContext;
+class MemoryPool;
+class SchemaManager;
+class SortedRun;
+class TableSchema;
+struct DataFileMeta;
+struct DeletionFile;
+struct KeyValue;
+template <typename T>
+class MergeFunctionWrapper;
+
+/// If the class name below is enclosed in parentheses, it might be present in
the read path;
+/// otherwise, it must be present in the read path.
+///
+/// Readers Overview: (ConcatBatchReader across
+/// splits)->CompleteRowKindBatchReader->(PredicateBatchReader)
+/// ->ConcatBatchReader across no overlapped
+/// files->KeyValueProjectionReader/AsyncKeyValueProjectionReader
+///
->DropDeleteReader->SortMergeReader->ConcatKeyValueRecordReader->KeyValueDataFileRecordReader
+///
->FieldMappingReader->(ApplyDeletionVectorBatchReader)->(DelegatingPrefetchReader)
+/// ->(PrefetchFileBatchReader)->FormatReader
+class MergeFileSplitRead : public AbstractSplitRead {
+ public:
+ static Result<std::unique_ptr<MergeFileSplitRead>> Create(
+ const std::shared_ptr<FileStorePathFactory>& path_factory,
+ const std::shared_ptr<InternalReadContext>& context,
+ const std::shared_ptr<MemoryPool>& memory_pool, const
std::shared_ptr<Executor>& executor);
+
+ Result<std::unique_ptr<BatchReader>> CreateReader(const
std::shared_ptr<Split>& split) override;
+
+ Result<bool> Match(const std::shared_ptr<Split>& split, bool
force_keep_delete) const override;
+
+ void ForceKeepDelete(bool force_keep_delete) {
+ force_keep_delete_ = force_keep_delete;
+ }
+
+ Result<std::unique_ptr<FileBatchReader>> ApplyIndexAndDvReaderIfNeeded(
+ std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
+ const std::shared_ptr<arrow::Schema>& data_schema,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory
dv_factory,
+ const std::optional<std::vector<Range>>& ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const override;
+
+ Result<std::unique_ptr<SortMergeReader>> CreateSortMergeReaderForSection(
+ const std::vector<SortedRun>& section, const BinaryRow& partition,
+ DeletionVector::Factory dv_factory, const std::shared_ptr<Predicate>&
predicate,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
bool drop_delete);
+
+ std::shared_ptr<FileStorePathFactory> GetPathFactory() const {
+ return path_factory_;
+ }
+
+ std::shared_ptr<arrow::Schema> GetValueSchema() const {
+ return value_schema_;
+ }
+
+ void SetMergeFunctionWrapper(
+ const std::shared_ptr<MergeFunctionWrapper<KeyValue>>&
merge_function_wrapper);
+
+ private:
+ Result<std::unique_ptr<BatchReader>> CreateMergeReader(
+ const std::shared_ptr<DataSplitImpl>& data_split,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory);
+
+ Result<std::unique_ptr<BatchReader>> CreateNoMergeReader(
+ const std::shared_ptr<DataSplitImpl>& data_split, bool only_filter_key,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const;
+
+ Result<std::unique_ptr<BatchReader>> CreateReaderForSection(
+ const std::vector<SortedRun>& section, const BinaryRow& partition,
+ DeletionVector::Factory dv_factory,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory);
+
+ Result<std::unique_ptr<KeyValueRecordReader>> CreateReaderForRun(
+ const BinaryRow& partition, const SortedRun& sorted_run,
DeletionVector::Factory dv_factory,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const;
+
+ Result<std::unique_ptr<SortMergeReader>> CreateSortMergeReader(
+ std::vector<std::unique_ptr<KeyValueRecordReader>>&& record_readers);
+
+ Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>>
GetMergeFunctionWrapper();
+
+ MergeFileSplitRead(const std::shared_ptr<FileStorePathFactory>&
path_factory,
+ const std::shared_ptr<InternalReadContext>& context,
+ std::unique_ptr<SchemaManager>&& schema_manager,
+ const std::shared_ptr<arrow::Schema>& key_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::vector<int32_t>& projection,
+ const std::shared_ptr<FieldsComparator>& key_comparator,
+ const std::shared_ptr<FieldsComparator>&
user_defined_seq_comparator,
+ const std::shared_ptr<Predicate>& predicate_for_keys,
+ const std::shared_ptr<MemoryPool>& memory_pool,
+ const std::shared_ptr<Executor>& executor);
+
+ static Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>>
CreateMergeFunctionWrapper(
+ const CoreOptions& core_options, const std::shared_ptr<TableSchema>&
table_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema);
+
+ static Status GenerateKeyValueReadSchema(
+ const TableSchema& table_schema, const CoreOptions& options,
+ const std::shared_ptr<arrow::Schema>& raw_read_schema,
+ std::shared_ptr<arrow::Schema>* value_schema,
std::shared_ptr<arrow::Schema>* read_schema,
+ std::shared_ptr<FieldsComparator>* key_comparator,
+ std::shared_ptr<FieldsComparator>* sequence_fields_comparator);
+
+ static Status SplitKeyAndNonKeyField(const std::vector<std::string>&
trimmed_key_fields,
+ const std::vector<DataField>&
read_fields,
+ std::vector<DataField>* key_fields,
+ std::vector<DataField>*
non_key_fields);
+
+ static Status CompleteSequenceField(const TableSchema& table_schema, const
CoreOptions& options,
+ std::vector<DataField>*
non_key_fields);
+
+ static Result<std::shared_ptr<Predicate>> GenerateKeyPredicates(
+ const std::shared_ptr<Predicate>& predicate, const TableSchema&
table_schema);
+
+ private:
+ // schema of key member in KeyValue object (trimmed pk)
+ std::shared_ptr<arrow::Schema> key_schema_;
+ // schema of value member in KeyValue object
+ std::shared_ptr<arrow::Schema> value_schema_;
+ // actual read schema, e.g., complete all key fields, user defined
sequence fields
+ std::shared_ptr<arrow::Schema> read_schema_;
+ std::vector<int32_t> projection_;
+ // merge_function_wrapper is lazy created, use through
GetMergeFunctionWrapper()
+ std::shared_ptr<MergeFunctionWrapper<KeyValue>> merge_function_wrapper_;
+ std::shared_ptr<FieldsComparator> key_comparator_;
+ std::shared_ptr<FieldsComparator> user_defined_seq_comparator_;
+ std::shared_ptr<Predicate> predicate_for_keys_;
+ bool force_keep_delete_ = false;
+};
+} // namespace paimon
diff --git a/src/paimon/core/operation/merge_file_split_read_test.cpp
b/src/paimon/core/operation/merge_file_split_read_test.cpp
new file mode 100644
index 0000000..70383b2
--- /dev/null
+++ b/src/paimon/core/operation/merge_file_split_read_test.cpp
@@ -0,0 +1,1296 @@
+/*
+ * 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/merge_file_split_read.h"
+
+#include <cstddef>
+#include <map>
+#include <optional>
+#include <ostream>
+#include <tuple>
+#include <utility>
+#include <variant>
+
+#include "arrow/api.h"
+#include "arrow/array/array_base.h"
+#include "arrow/c/abi.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/factories/io_hook.h"
+#include "paimon/common/reader/concat_batch_reader.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/fields_comparator.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/defs.h"
+#include "paimon/executor.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/read_context.h"
+#include "paimon/table/source/data_split.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/io_exception_helper.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon {
+class FileSystem;
+} // namespace paimon
+
+namespace paimon::test {
+// Parameter: min_heap/loser_tree; enable/disable IO prefetch; enable/disable
multi thread row to
+// batch
+class MergeFileSplitReadTest : public ::testing::Test,
+ public
::testing::WithParamInterface<std::tuple<bool, bool, bool>> {
+ void SetUp() override {}
+ void TearDown() override {}
+
+ void CheckResult(const std::shared_ptr<arrow::ChunkedArray>& result,
+ const std::shared_ptr<arrow::ChunkedArray>& expected,
+ const std::shared_ptr<arrow::Schema>& schema) const {
+ if (!std::get<2>(GetParam())) {
+ ASSERT_TRUE(result->ApproxEquals(*expected)) << result->ToString();
+ return;
+ }
+ ASSERT_OK_AND_ASSIGN(auto sorted_result,
ReadResultCollector::SortArray(result, schema));
+ ASSERT_OK_AND_ASSIGN(auto sorted_expected,
+ ReadResultCollector::SortArray(expected, schema));
+ ASSERT_TRUE(sorted_result->ApproxEquals(*sorted_expected))
+ << sorted_result->ToString() << std::endl
+ << sorted_expected->ToString();
+ }
+
+ std::shared_ptr<InternalReadContext> CreateInternalReadContext(
+ const std::shared_ptr<ReadContext>& read_context, int32_t schema_id =
0) {
+ SchemaManager schema_manager(fs_, read_context->GetPath());
+ EXPECT_OK_AND_ASSIGN(auto table_schema,
schema_manager.ReadSchema(schema_id));
+ EXPECT_OK_AND_ASSIGN(auto context,
InternalReadContext::Create(read_context, table_schema,
+
read_context->GetOptions()));
+ return context;
+ }
+
+ void AddOptions(ReadContextBuilder* context_builder) const {
+ auto [use_min_heap, enable_io_prefetch,
enable_multi_thread_row_to_batch] = GetParam();
+ if (use_min_heap) {
+ context_builder->AddOption(Options::SORT_ENGINE, "min-heap");
+ } else {
+ context_builder->AddOption(Options::SORT_ENGINE, "loser-tree");
+ }
+ if (enable_io_prefetch) {
+
context_builder->AddOption("test.enable-adaptive-prefetch-strategy", "false");
+ context_builder->EnablePrefetch(true);
+ context_builder->SetPrefetchBatchCount(/*batch_count=*/3);
+ } else {
+ context_builder->EnablePrefetch(false);
+ }
+ if (enable_multi_thread_row_to_batch) {
+ context_builder->EnableMultiThreadRowToBatch(true);
+ context_builder->SetRowToBatchThreadNumber(4);
+ } else {
+ context_builder->EnableMultiThreadRowToBatch(false);
+ }
+ }
+
+ // for table pk_table_with_mor
+ std::vector<std::shared_ptr<DataSplit>> PrepareDataSplit() const {
+ auto meta1_1 = std::make_shared<DataFileMeta>(
+ "data-c80ccf0f-6387-4cbc-8889-ade8cef54c43-0.parquet",
/*file_size=*/3346,
+ /*row_count=*/4,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 0}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({1, 1}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 0}, {1, 1},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {0, 0, 0, 0, "apple", "!", static_cast<double>(10), false},
+ {1, 1, 0, 0, "driver", "you", 13.3, true}, {0, 0, 0, 0, 0, 0,
0, 0}, pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/3,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735149279565ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ auto meta1_2 = std::make_shared<DataFileMeta>(
+ "data-c80ccf0f-6387-4cbc-8889-ade8cef54c43-1.parquet",
/*file_size=*/3370,
+ /*row_count=*/4,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 0}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({1, 2}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 0}, {1, 2},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {0, 0, 0, 0, "abandon", "!", static_cast<double>(110), false},
+ {1, 2, 0, 0, "driver", "see", 112.2, true}, {0, 0, 0, 0, 0, 0,
0, 0}, pool_.get()),
+ /*min_sequence_number=*/4, /*max_sequence_number=*/7,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735149279917ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ auto meta1_3 = std::make_shared<DataFileMeta>(
+ "data-c80ccf0f-6387-4cbc-8889-ade8cef54c43-2.parquet",
/*file_size=*/3252,
+ /*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({100, 200},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({100, 200},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({100, 200}, {100, 200}, {0, 0},
pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {100, 200, 0, 0, std::string("max"), std::string("number"),
140.4, false},
+ {100, 200, 0, 0, std::string("max"), std::string("number"),
140.4, false},
+ {0, 0, 0, 0, 0, 0, 0, 0}, pool_.get()),
+ /*min_sequence_number=*/8, /*max_sequence_number=*/8,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735230606999ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ DataSplitImpl::Builder builder1(BinaryRowGenerator::GenerateRow({0,
0}, pool_.get()),
+ /*bucket=*/0, /*bucket_path=*/
+ paimon::test::GetDataDir() +
+ "/parquet/pk_table_with_mor.db/"
+
"pk_table_with_mor/p0=0/p1=0/bucket-0",
+ {meta1_1, meta1_2, meta1_3});
+ EXPECT_OK_AND_ASSIGN(
+ auto data_split1,
+
builder1.WithSnapshot(3).IsStreaming(false).RawConvertible(false).Build());
+ auto meta2_1 = std::make_shared<DataFileMeta>(
+ "data-24f8588c-d950-4e44-9d99-a023ea65a136-0.parquet",
/*file_size=*/3245,
+ /*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 1}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({0, 1}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 1}, {0, 1},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {0, 1, 0, 1, "mouse", "you", static_cast<double>(30), false},
+ {0, 1, 0, 1, "mouse", "you", static_cast<double>(30), false},
+ {0, 0, 0, 0, 0, 0, 0, 0}, pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/0,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735149279951ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ auto meta2_2 = std::make_shared<DataFileMeta>(
+ "data-24f8588c-d950-4e44-9d99-a023ea65a136-1.parquet",
/*file_size=*/3229,
+ /*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 1}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({0, 1}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 1}, {0, 1},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {0, 1, 0, 1, "zoo", "you", static_cast<double>(130), false},
+ {0, 1, 0, 1, "zoo", "you", static_cast<double>(130), false},
+ {0, 0, 0, 0, 0, 0, 0, 0}, pool_.get()),
+ /*min_sequence_number=*/1, /*max_sequence_number=*/1,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735149279612ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ DataSplitImpl::Builder builder2(BinaryRowGenerator::GenerateRow({0,
1}, pool_.get()),
+ /*bucket=*/0, /*bucket_path=*/
+ paimon::test::GetDataDir() +
+ "/parquet/pk_table_with_mor.db/"
+
"pk_table_with_mor/p0=0/p1=1/bucket-0",
+ {meta2_1, meta2_2});
+ EXPECT_OK_AND_ASSIGN(
+ auto data_split2,
+
builder2.WithSnapshot(3).IsStreaming(false).RawConvertible(false).Build());
+
+ auto meta3_1 = std::make_shared<DataFileMeta>(
+ "data-184f2304-49fd-4916-ba07-037757e904eb-0.parquet",
/*file_size=*/3259,
+ /*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 0}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({0, 0}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 0}, {0, 0},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {0, 0, 1, 0, "elephant", "hi", static_cast<double>(120),
false},
+ {0, 0, 1, 0, "elephant", "hi", static_cast<double>(120),
false},
+ {0, 0, 0, 0, 0, 0, 0, 0}, pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/0,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735149271981ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ auto meta3_2 = std::make_shared<DataFileMeta>(
+ "data-184f2304-49fd-4916-ba07-037757e904eb-1.parquet",
/*file_size=*/3259,
+ /*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 0}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({0, 0}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 0}, {0, 0},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats(
+ {0, 0, 1, 0, "elephant", "hi", static_cast<double>(20), true},
+ {0, 0, 1, 0, "elephant", "hi", static_cast<double>(20), true},
+ {0, 0, 0, 0, 0, 0, 0, 0}, pool_.get()),
+ /*min_sequence_number=*/1, /*max_sequence_number=*/1,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1735149279651ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ DataSplitImpl::Builder builder3(BinaryRowGenerator::GenerateRow({1,
0}, pool_.get()),
+ /*bucket=*/0, /*bucket_path=*/
+ paimon::test::GetDataDir() +
+ "/parquet/pk_table_with_mor.db/"
+
"pk_table_with_mor/p0=1/p1=0/bucket-0",
+ {meta3_1, meta3_2});
+ EXPECT_OK_AND_ASSIGN(
+ auto data_split3,
+
builder3.WithSnapshot(3).IsStreaming(false).RawConvertible(false).Build());
+ return {data_split1, data_split2, data_split3};
+ }
+
+ // for table pk_table_partial_update
+ std::vector<std::shared_ptr<DataSplit>> PrepareDataSplit2() const {
+ auto meta1_1 = std::make_shared<DataFileMeta>(
+ "data-d03e13e5-5e2e-463a-b53a-8d44e4dc9141-0.parquet",
+ /*file_size=*/2554, /*row_count=*/
+ 3,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 1}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({1, 1}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 0}, {1, 1},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({0, 0, 2.0, false,
std::string("apple")},
+ {1, 1, 2.0, true,
std::string("banana")},
+ {0, 0, 2, 0, 1}, pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/2,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1736793059256ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ auto meta1_2 = std::make_shared<DataFileMeta>(
+ "data-d03e13e5-5e2e-463a-b53a-8d44e4dc9141-1.parquet",
+ /*file_size=*/2623, /*row_count=*/
+ 5,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({0, 1}, pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({2, 2}, pool_.get()),
+ /*key_stats=*/BinaryRowGenerator::GenerateStats({0, 0}, {2, 2},
{0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({0, 0, 100.0, false,
std::string("new_apple")},
+ {2, 2, 144.4, true,
std::string("orange")},
+ {0, 0, 0, 0, 3}, pool_.get()),
+ /*min_sequence_number=*/3, /*max_sequence_number=*/7,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1736793059526ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ DataSplitImpl::Builder builder1(
+ /*partition=*/BinaryRow::EmptyRow(),
+ /*bucket=*/0, /*bucket_path=*/
+ paimon::test::GetDataDir() +
+
"/parquet/pk_table_partial_update.db/pk_table_partial_update/bucket-0",
+ {meta1_1, meta1_2});
+ EXPECT_OK_AND_ASSIGN(
+ auto data_split1,
+
builder1.WithSnapshot(2).IsStreaming(false).RawConvertible(false).Build());
+
+ return {data_split1};
+ }
+
+ Result<std::unique_ptr<BatchReader>> CreateReader(
+ const std::shared_ptr<InternalReadContext>& internal_context,
+ const std::vector<std::shared_ptr<DataSplit>>& data_splits) {
+ const auto& core_options = internal_context->GetCoreOptions();
+ const auto& table_schema = internal_context->GetTableSchema();
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ EXPECT_OK_AND_ASSIGN(std::vector<std::string> external_paths,
+ core_options.CreateExternalPaths());
+ EXPECT_OK_AND_ASSIGN(std::optional<std::string>
global_index_external_path,
+ core_options.CreateGlobalIndexExternalPath());
+
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ FileStorePathFactory::Create(
+ internal_context->GetPath(), arrow_schema,
table_schema->PartitionKeys(),
+ core_options.GetPartitionDefaultName(),
core_options.GetFileFormat()->Identifier(),
+ core_options.DataFilePrefix(),
core_options.LegacyPartitionNameEnabled(),
+ external_paths, global_index_external_path,
core_options.IndexFileInDataFileDir(),
+ pool_));
+ PAIMON_ASSIGN_OR_RAISE(auto split_read,
+ MergeFileSplitRead::Create(path_factory,
std::move(internal_context),
+ pool_, executor_));
+ std::vector<std::unique_ptr<BatchReader>> batch_readers;
+ batch_readers.reserve(data_splits.size());
+ for (const auto& split : data_splits) {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> reader,
+ split_read->CreateReader(split));
+ batch_readers.emplace_back(std::move(reader));
+ }
+ return std::make_unique<ConcatBatchReader>(std::move(batch_readers),
pool_);
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
+ std::shared_ptr<FileSystem> fs_ = std::make_shared<LocalFileSystem>();
+ std::shared_ptr<Executor> executor_ =
CreateDefaultExecutor(/*thread_count=*/4);
+};
+
+// test GenerateKeyValueReadSchema with user define fields
+TEST_F(MergeFileSplitReadTest, TestGenerateKeyValueReadSchema) {
+ std::string table_path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ auto schema_manager = std::make_unique<SchemaManager>(fs_, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
schema_manager->ReadSchema(/*schema_id=*/0));
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::SEQUENCE_FIELD,
"s0,s1"},
+ {Options::MERGE_ENGINE,
"deduplicate"},
+ {Options::SORT_ENGINE,
"min-heap"},
+ {Options::IGNORE_DELETE,
"true"}}));
+ std::vector<DataField> raw_read_fields = {
+ DataField(1, arrow::field("k1", arrow::int32(), /*nullable=*/false)),
+ DataField(3, arrow::field("p1", arrow::int32(), /*nullable=*/false)),
+ DataField(5, arrow::field("s1", arrow::utf8())),
+ DataField(6, arrow::field("v0", arrow::float64())),
+ DataField(7, arrow::field("v1", arrow::boolean()))};
+ auto raw_read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(raw_read_schema);
+
+ std::shared_ptr<arrow::Schema> value_schema;
+ std::shared_ptr<arrow::Schema> read_schema;
+ std::shared_ptr<FieldsComparator> key_comparator;
+ std::shared_ptr<FieldsComparator> sequence_fields_comparator;
+ ASSERT_OK(MergeFileSplitRead::GenerateKeyValueReadSchema(
+ *table_schema, options, raw_read_schema, &value_schema, &read_schema,
&key_comparator,
+ &sequence_fields_comparator));
+
+ // check result
+ std::vector<DataField> expected_value_fields = {
+ DataField(1, arrow::field("k1", arrow::int32(), /*nullable=*/false)),
+ DataField(3, arrow::field("p1", arrow::int32(), /*nullable=*/false)),
+ DataField(5, arrow::field("s1", arrow::utf8())),
+ DataField(6, arrow::field("v0", arrow::float64())),
+ DataField(7, arrow::field("v1", arrow::boolean())),
+ DataField(4, arrow::field("s0", arrow::utf8()))};
+ auto expected_value_schema =
DataField::ConvertDataFieldsToArrowSchema(expected_value_fields);
+
+ ASSERT_OK_AND_ASSIGN(auto result_fields,
+
DataField::ConvertArrowSchemaToDataFields(value_schema));
+ ASSERT_OK_AND_ASSIGN(auto expected_fields,
+
DataField::ConvertArrowSchemaToDataFields(expected_value_schema));
+ for (size_t i = 0; i < expected_fields.size(); i++) {
+ EXPECT_EQ(result_fields[i], expected_fields[i]);
+ EXPECT_OK_AND_ASSIGN(std::string result_str,
result_fields[i].ToJsonString());
+ EXPECT_OK_AND_ASSIGN(std::string expected_str,
expected_fields[i].ToJsonString());
+ EXPECT_EQ(result_str, expected_str);
+ }
+ ASSERT_TRUE(value_schema->Equals(*expected_value_schema));
+
+ std::vector<DataField> expected_read_data_fields = {
+ SpecialFields::SequenceNumber(),
+ SpecialFields::ValueKind(),
+ DataField(0, arrow::field("k0", arrow::int32(), /*nullable=*/false)),
+ DataField(1, arrow::field("k1", arrow::int32(), /*nullable=*/false)),
+ DataField(3, arrow::field("p1", arrow::int32(), /*nullable=*/false)),
+ DataField(5, arrow::field("s1", arrow::utf8())),
+ DataField(6, arrow::field("v0", arrow::float64())),
+ DataField(7, arrow::field("v1", arrow::boolean())),
+ DataField(4, arrow::field("s0", arrow::utf8()))};
+ auto expected_read_schema =
+ DataField::ConvertDataFieldsToArrowSchema(expected_read_data_fields);
+ ASSERT_TRUE(read_schema->Equals(*expected_read_schema));
+
+ std::vector<int32_t> expected_sort_key_fields = {0, 1};
+ ASSERT_EQ(key_comparator->sort_fields_, expected_sort_key_fields);
+ ASSERT_EQ(key_comparator->is_ascending_order_, true);
+
+ std::vector<int32_t> expected_sort_seq_fields = {5, 2};
+ ASSERT_EQ(sequence_fields_comparator->sort_fields_,
expected_sort_seq_fields);
+ ASSERT_EQ(sequence_fields_comparator->is_ascending_order_, true);
+}
+
+// test GenerateKeyValueReadSchema without user define fields
+TEST_F(MergeFileSplitReadTest, TestGenerateKeyValueReadSchema1) {
+ std::string table_path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ auto schema_manager = std::make_unique<SchemaManager>(fs_, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
schema_manager->ReadSchema(/*schema_id=*/0));
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::MERGE_ENGINE,
"deduplicate"},
+ {Options::SORT_ENGINE,
"min-heap"},
+ {Options::IGNORE_DELETE,
"true"}}));
+ std::vector<DataField> raw_read_fields = {
+ DataField(1, arrow::field("k1", arrow::int32(), /*nullable=*/false)),
+ DataField(3, arrow::field("p1", arrow::int32(), /*nullable=*/false)),
+ DataField(5, arrow::field("s1", arrow::utf8())),
+ DataField(6, arrow::field("v0", arrow::float64())),
+ DataField(7, arrow::field("v1", arrow::boolean()))};
+ auto raw_read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(raw_read_schema);
+
+ std::shared_ptr<arrow::Schema> value_schema;
+ std::shared_ptr<arrow::Schema> read_schema;
+ std::shared_ptr<FieldsComparator> key_comparator;
+ std::shared_ptr<FieldsComparator> sequence_fields_comparator;
+ ASSERT_OK(MergeFileSplitRead::GenerateKeyValueReadSchema(
+ *table_schema, options, raw_read_schema, &value_schema, &read_schema,
&key_comparator,
+ &sequence_fields_comparator));
+
+ // check result
+ std::vector<DataField> expected_value_fields = {
+ DataField(1, arrow::field("k1", arrow::int32(), /*nullable=*/false)),
+ DataField(3, arrow::field("p1", arrow::int32(), /*nullable=*/false)),
+ DataField(5, arrow::field("s1", arrow::utf8())),
+ DataField(6, arrow::field("v0", arrow::float64())),
+ DataField(7, arrow::field("v1", arrow::boolean()))};
+ auto expected_value_schema =
DataField::ConvertDataFieldsToArrowSchema(expected_value_fields);
+ ASSERT_TRUE(value_schema->Equals(*expected_value_schema));
+
+ std::vector<DataField> expected_read_data_fields = {
+ SpecialFields::SequenceNumber(),
+ SpecialFields::ValueKind(),
+ DataField(0, arrow::field("k0", arrow::int32(), /*nullable=*/false)),
+ DataField(1, arrow::field("k1", arrow::int32(), /*nullable=*/false)),
+ DataField(3, arrow::field("p1", arrow::int32(), /*nullable=*/false)),
+ DataField(5, arrow::field("s1", arrow::utf8())),
+ DataField(6, arrow::field("v0", arrow::float64())),
+ DataField(7, arrow::field("v1", arrow::boolean()))};
+ auto expected_read_schema =
+ DataField::ConvertDataFieldsToArrowSchema(expected_read_data_fields);
+ ASSERT_TRUE(read_schema->Equals(*expected_read_schema));
+
+ std::vector<int32_t> expected_sort_key_fields = {0, 1};
+ ASSERT_EQ(key_comparator->sort_fields_, expected_sort_key_fields);
+ ASSERT_EQ(key_comparator->is_ascending_order_, true);
+
+ ASSERT_FALSE(sequence_fields_comparator);
+}
+
+// test GenerateKeyValueReadSchema with compaction mode that raw_read_schema
equals table_schema
+TEST_F(MergeFileSplitReadTest, TestGenerateKeyValueReadSchema2) {
+ std::string table_path =
+ paimon::test::GetDataDir() +
"parquet/pk_table_with_mor.db/pk_table_with_mor";
+ auto schema_manager = std::make_unique<SchemaManager>(fs_, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
schema_manager->ReadSchema(/*schema_id=*/0));
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::SEQUENCE_FIELD,
"s0,s1"},
+ {Options::MERGE_ENGINE,
"deduplicate"},
+ {Options::SORT_ENGINE,
"min-heap"},
+ {Options::IGNORE_DELETE,
"true"}}));
+ auto raw_read_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ ASSERT_TRUE(raw_read_schema);
+
+ std::shared_ptr<arrow::Schema> value_schema;
+ std::shared_ptr<arrow::Schema> read_schema;
+ std::shared_ptr<FieldsComparator> key_comparator;
+ std::shared_ptr<FieldsComparator> sequence_fields_comparator;
+ ASSERT_OK(MergeFileSplitRead::GenerateKeyValueReadSchema(
+ *table_schema, options, raw_read_schema, &value_schema, &read_schema,
&key_comparator,
+ &sequence_fields_comparator));
+
+ // check result
+ auto expected_value_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ ASSERT_OK_AND_ASSIGN(auto result_fields,
+
DataField::ConvertArrowSchemaToDataFields(value_schema));
+ ASSERT_OK_AND_ASSIGN(auto expected_fields,
+
DataField::ConvertArrowSchemaToDataFields(expected_value_schema));
+ for (size_t i = 0; i < expected_fields.size(); i++) {
+ EXPECT_EQ(result_fields[i], expected_fields[i]);
+ EXPECT_OK_AND_ASSIGN(std::string result_str,
result_fields[i].ToJsonString());
+ EXPECT_OK_AND_ASSIGN(std::string expected_str,
expected_fields[i].ToJsonString());
+ EXPECT_EQ(result_str, expected_str);
+ }
+ ASSERT_TRUE(value_schema->Equals(*expected_value_schema));
+
+ auto expected_read_schema =
+
SpecialFields::CompleteSequenceAndValueKindField(expected_value_schema);
+ ASSERT_TRUE(read_schema->Equals(*expected_read_schema));
+
+ std::vector<int32_t> expected_sort_key_fields = {0, 1};
+ ASSERT_EQ(key_comparator->sort_fields_, expected_sort_key_fields);
+ ASSERT_EQ(key_comparator->is_ascending_order_, true);
+
+ std::vector<int32_t> expected_sort_seq_fields = {4, 5};
+ ASSERT_EQ(sequence_fields_comparator->sort_fields_,
expected_sort_seq_fields);
+ ASSERT_EQ(sequence_fields_comparator->is_ascending_order_, true);
+}
+
+// test no predicate
+TEST_F(MergeFileSplitReadTest, TestGenerateKeyPredicates) {
+ std::string table_path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ auto schema_manager = std::make_unique<SchemaManager>(fs_, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
schema_manager->ReadSchema(/*schema_id=*/0));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> predicate_result,
+ MergeFileSplitRead::GenerateKeyPredicates(/*predicate=*/nullptr,
*table_schema));
+ ASSERT_FALSE(predicate_result);
+}
+
+// test exist primary predicate
+TEST_F(MergeFileSplitReadTest, TestGenerateKeyPredicates1) {
+ std::string table_path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ auto schema_manager = std::make_unique<SchemaManager>(fs_, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
schema_manager->ReadSchema(/*schema_id=*/0));
+
+ ASSERT_OK_AND_ASSIGN(
+ auto predicate_result,
+ PredicateBuilder::And({PredicateBuilder::Equal(/*field_index=*/0,
/*field_name=*/"k0",
+ FieldType::INT,
Literal(3)),
+ PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"p0",
+ FieldType::INT,
Literal(5))}));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> result,
+ MergeFileSplitRead::GenerateKeyPredicates(
+ /*predicate=*/predicate_result, *table_schema));
+ ASSERT_OK_AND_ASSIGN(auto expected_key_predicate,
+ PredicateBuilder::And({PredicateBuilder::Equal(
+ /*field_index=*/0, /*field_name=*/"k0",
FieldType::INT, Literal(3))}));
+ ASSERT_EQ(*result, *expected_key_predicate);
+}
+
+// test non-primary predicate
+TEST_F(MergeFileSplitReadTest, TestGenerateKeyPredicates2) {
+ std::string table_path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ auto schema_manager = std::make_unique<SchemaManager>(fs_, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
schema_manager->ReadSchema(/*schema_id=*/0));
+
+ ASSERT_OK_AND_ASSIGN(
+ auto predicate,
+ PredicateBuilder::And({PredicateBuilder::Equal(/*field_index=*/3,
/*field_name=*/"p1",
+ FieldType::INT,
Literal(3)),
+ PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"p0",
+ FieldType::INT,
Literal(5))}));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> result,
+ MergeFileSplitRead::GenerateKeyPredicates(
+ /*predicate=*/predicate, *table_schema));
+ ASSERT_FALSE(result);
+}
+
+TEST_P(MergeFileSplitReadTest, TestSimple) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(5, arrow::field("s1",
arrow::utf8())),
+ DataField(6, arrow::field("v0",
arrow::float64())),
+ DataField(7, arrow::field("v1",
arrow::boolean()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "p1", "s1", "v0", "v1"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<MergeFileSplitRead> split_read,
+ MergeFileSplitRead::Create(/*path_factory=*/nullptr, internal_context,
pool_, executor_));
+ auto data_splits = PrepareDataSplit();
+
+ // test split read match
+ {
+ ASSERT_OK_AND_ASSIGN(bool matched,
+ split_read->Match(data_splits[0],
/*force_keep_delete=*/false));
+ ASSERT_TRUE(matched);
+ }
+ {
+ auto fake_data_split = PrepareDataSplit()[0];
+ auto split_impl = dynamic_cast<DataSplitImpl*>(fake_data_split.get());
+ split_impl->before_files_ = split_impl->data_files_;
+ ASSERT_OK_AND_ASSIGN(bool matched,
+ split_read->Match(fake_data_split,
/*force_keep_delete=*/false));
+ ASSERT_FALSE(matched);
+ }
+
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
data_splits));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, "see", 110.0, false],
+ [0, 1, 0, "you", 11.1, false],
+ [0, 0, 0, "later", 12.2, true],
+ [0, 1, 0, "!", 13.3, false],
+ [0, 2, 0, "!", 13.3, false],
+ [0, 200, 0, "number",140.4, false],
+ [0, 1, 1, "you", 130.0, false],
+ [0, 0, 0, "hi", 120.0, false]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestLookUp) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(5, arrow::field("s1",
arrow::utf8())),
+ DataField(6, arrow::field("v0",
arrow::float64())),
+ DataField(7, arrow::field("v1",
arrow::boolean()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "p1", "s1", "v0", "v1"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"},
+ {Options::FORCE_LOOKUP, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context);
+
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, "see", 110.0, false],
+ [0, 1, 0, "you", 11.1, false],
+ [0, 0, 0, "later", 12.2, true],
+ [0, 1, 0, "!", 13.3, false],
+ [0, 2, 0, "!", 13.3, false],
+ [0, 200, 0, "number",140.4, false],
+ [0, 1, 1, "you", 130.0, false],
+ [0, 0, 0, "hi", 120.0, false]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestReadWithLimits) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"},
+ {Options::READ_BATCH_SIZE, "1"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context);
+
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+
+ // simulate read limits, only read 4 batches
+ for (int32_t i = 0; i < 4; i++) {
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch,
batch_reader->NextBatch());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> array,
+ ReadResultCollector::GetArray(std::move(batch)));
+ ASSERT_TRUE(array);
+ ASSERT_EQ(array->length(), 1);
+ }
+ batch_reader->Close();
+}
+
+TEST_P(MergeFileSplitReadTest, TestDeduplicateMergeEngineWithDeleteMsg) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_partial_update.db/pk_table_partial_update";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(0, arrow::field("k0",
arrow::int32())),
+ DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(2, arrow::field("v0",
arrow::float64())),
+ DataField(3, arrow::field("v1",
arrow::boolean())),
+ DataField(4, arrow::field("v2",
arrow::utf8()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k0", "k1", "v0", "v1", "v2"});
+ context_builder.SetOptions({{Options::MERGE_ENGINE, "deduplicate"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit2()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 1, 100.0, true, "new_apple"],
+ [0, 1, 1, 133.3, false, null],
+ [0, 2, 1, 144.4, true, "orange"]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestReadWithPredicate) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(5, arrow::field("s1",
arrow::utf8())),
+ DataField(4, arrow::field("s0",
arrow::utf8())),
+ DataField(6, arrow::field("v0",
arrow::float64())),
+ DataField(7, arrow::field("v1",
arrow::boolean()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "p1", "s1", "s0", "v0", "v1"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+
+ // less_than will be ignore as it is partition predicate
+ auto less_than = PredicateBuilder::GreaterThan(/*field_index=*/1,
/*field_name=*/"p1",
+ FieldType::INT,
Literal(-1));
+ // greater_or_equal is key predicate, will always be pushed down
+ auto greater_or_equal =
PredicateBuilder::GreaterOrEqual(/*field_index=*/0, /*field_name=*/"k1",
+ FieldType::INT,
Literal(1));
+ // greater_than is value predicate, will be pushed down while the number
of sorted run in
+ // section equals 1
+ auto greater_than = PredicateBuilder::GreaterThan(/*field_index=*/4,
/*field_name=*/"v0",
+ FieldType::DOUBLE,
Literal(150.0));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate_result,
+ PredicateBuilder::And({less_than, greater_or_equal,
greater_than}));
+ context_builder.SetPredicate(predicate_result);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, "see", "apple", 110.0, false],
+ [0, 1, 0, "you", "banana", 11.1, false],
+ [0, 0, 0, "later", "car", 12.2, true],
+ [0, 1, 0, "!", "driver", 13.3, false],
+ [0, 2, 0, "!", "driver", 13.3, false],
+ [0, 1, 1, "you", "zoo", 130.0, false]
+
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestReadWithAlterTable) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(0, arrow::field("k1",
arrow::int32())),
+ DataField(1, arrow::field("k0",
arrow::int32())),
+ DataField(2, arrow::field("p0",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(4, arrow::field("s1",
arrow::utf8())),
+ DataField(5, arrow::field("s0",
arrow::binary())),
+ DataField(6, arrow::field("v0",
arrow::int32())),
+ DataField(7, arrow::field("v1",
arrow::utf8())),
+ DataField(8, arrow::field("v2",
arrow::int32()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "k0", "p0", "p1", "s1", "s0", "v0",
"v1", "v2"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context,
/*schema_id=*/7);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, 0, 0, "apple", "see", 110, "false", null],
+ [0, 0, 1, 0, 0, "banana", "you", 11, "false", null],
+ [0, 1, 0, 0, 0, "car", "later", 12, "true", null],
+ [0, 1, 1, 0, 0, "driver", "!", 13, "false", null],
+ [0, 1, 2, 0, 0, "driver", "!", 13, "false", null],
+ [0, 100, 200, 0, 0, "max", "number", 140, "false", null],
+ [0, 0, 1, 0, 1, "zoo", "you", 130, "false", null],
+ [0, 0, 0, 1, 0, "elephant", "hi", 120, "false", null]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestReadWithAlterTableWithReverseSequence) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(8, arrow::field("v2",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(1, arrow::field("k0",
arrow::int32())),
+ DataField(2, arrow::field("p0",
arrow::int32())),
+ DataField(5, arrow::field("s0",
arrow::binary())),
+ DataField(6, arrow::field("v0",
arrow::int32()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"v2", "p1", "k0", "p0", "s0", "v0"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context,
/*schema_id=*/7);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, null, 0, 0, 0, "see", 110],
+ [0, null, 0, 1, 0, "you", 11],
+ [0, null, 0, 0, 0, "later", 12],
+ [0, null, 0, 1, 0, "!", 13],
+ [0, null, 0, 2, 0, "!", 13],
+ [0, null, 0, 200, 0, "number", 140],
+ [0, null, 1, 1, 0, "you", 130],
+ [0, null, 0, 0, 1, "hi", 120]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestAggregateMergeEngine) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(5, arrow::field("s1",
arrow::utf8())),
+ DataField(6, arrow::field("v0",
arrow::float64())),
+ DataField(7, arrow::field("v1",
arrow::boolean()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "p1", "s1", "v0", "v1"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {"fields.v1.aggregate-function", "bool_and"},
+ {"fields.v0.aggregate-function", "sum"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, "see", 120.0, false],
+ [0, 1, 0, "you", 122.2, false],
+ [0, 0, 0, "later", 124.4, true],
+ [0, 1, 0, "!", 13.3, false],
+ [0, 2, 0, "!", 13.3, false],
+ [0, 200, 0, "number",140.4, false],
+ [0, 1, 1, "you", 160.0, false],
+ [0, 0, 0, "hi", 140.0, false]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestPartialUpdateMergeEngine) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(5, arrow::field("s1",
arrow::utf8())),
+ DataField(6, arrow::field("v0",
arrow::float64()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "p1", "s1", "v0"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "partial-update"},
+ {"fields.v1.sequence-group", "v0"},
+ {"fields.v0.aggregate-function",
"first_value"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, "see", 110.0],
+ [0, 1, 0, "you", 11.1],
+ [0, 0, 0, "later", 112.2],
+ [0, 1, 0, "!", 13.3],
+ [0, 2, 0, "!", 13.3],
+ [0, 200, 0, "number",140.4],
+ [0, 1, 1, "you", 30.0],
+ [0, 0, 0, "hi", 120.0]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestPartialUpdateMergeEngineWithIgnoreDelete) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_partial_update.db/pk_table_partial_update";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(0, arrow::field("k0",
arrow::int32())),
+ DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(2, arrow::field("v0",
arrow::float64())),
+ DataField(3, arrow::field("v1",
arrow::boolean())),
+ DataField(4, arrow::field("v2",
arrow::utf8()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k0", "k1", "v0", "v1", "v2"});
+ context_builder.SetOptions(
+ {{Options::MERGE_ENGINE, "partial-update"}, {Options::IGNORE_DELETE,
"true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit2()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON(
+ arrow::struct_(fields_with_row_kind),
+ {R"([ [0, 0, 1, 100.0, true, "new_apple"], [0, 1, 0, 2.0, false,
null], [0, 1, 1, 133.3,
+ false, "banana"], [0, 2, 1, 144.4, true, "orange"]
+
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest,
TestPartialUpdateMergeEngineWithRemoveRecordOnDelete) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_partial_update.db/pk_table_partial_update";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(0, arrow::field("k0",
arrow::int32())),
+ DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(2, arrow::field("v0",
arrow::float64())),
+ DataField(3, arrow::field("v1",
arrow::boolean())),
+ DataField(4, arrow::field("v2",
arrow::utf8()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k0", "k1", "v0", "v1", "v2"});
+ context_builder.SetOptions({{Options::MERGE_ENGINE, "partial-update"},
+
{Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
PrepareDataSplit2()));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON(
+ arrow::struct_(fields_with_row_kind),
+ {R"([ [0, 0, 1, 100.0, true, "new_apple"], [0, 1, 1, 133.3, false,
"banana"], [0, 2, 1,
+ 144.4, true, "orange"]
+
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+}
+
+TEST_P(MergeFileSplitReadTest, TestEmptyPlan) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_partial_update.db/pk_table_partial_update";
+ ReadContextBuilder context_builder(path);
+
+ std::vector<DataField> raw_read_fields = {DataField(0, arrow::field("k0",
arrow::int32())),
+ DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(2, arrow::field("v0",
arrow::float64())),
+ DataField(3, arrow::field("v1",
arrow::boolean())),
+ DataField(4, arrow::field("v2",
arrow::utf8()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k0", "k1", "v0", "v1", "v2"});
+ context_builder.SetOptions({{Options::MERGE_ENGINE, "partial-update"},
+
{Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+ auto internal_context = CreateInternalReadContext(read_context);
+ std::vector<std::shared_ptr<DataSplit>> empty_data_split;
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
empty_data_split));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> read_result,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+ // empty result with null pointer batch
+ ASSERT_FALSE(read_result);
+}
+
+TEST_P(MergeFileSplitReadTest, TestIOException) {
+ std::string path =
+ paimon::test::GetDataDir() +
"/parquet/pk_table_with_mor.db/pk_table_with_mor";
+ std::vector<DataField> raw_read_fields = {DataField(1, arrow::field("k1",
arrow::int32())),
+ DataField(3, arrow::field("p1",
arrow::int32())),
+ DataField(5, arrow::field("s1",
arrow::utf8())),
+ DataField(6, arrow::field("v0",
arrow::float64())),
+ DataField(7, arrow::field("v1",
arrow::boolean()))};
+
+ ReadContextBuilder context_builder(path);
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+
+ context_builder.SetReadSchema({"k1", "p1", "s1", "v0", "v1"});
+ context_builder.SetOptions({{Options::SEQUENCE_FIELD, "s0,s1"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::IGNORE_DELETE, "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+
+ auto internal_context = CreateInternalReadContext(read_context);
+
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 0, 0, "see", 110.0, false],
+ [0, 1, 0, "you", 11.1, false],
+ [0, 0, 0, "later", 12.2, true],
+ [0, 1, 0, "!", 13.3, false],
+ [0, 2, 0, "!", 13.3, false],
+ [0, 200, 0, "number",140.4, false],
+ [0, 1, 1, "you", 130.0, false],
+ [0, 0, 0, "hi", 120.0, false]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+
+ bool run_complete = false;
+ auto io_hook = IOHook::GetInstance();
+ for (size_t i = 0; i < 300; i++) {
+ ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
+ io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
+ auto batch_reader = CreateReader(internal_context, PrepareDataSplit());
+ CHECK_HOOK_STATUS(batch_reader.status(), i);
+ auto read_result =
ReadResultCollector::CollectResult(batch_reader.value().get());
+ CHECK_HOOK_STATUS(read_result.status(), i);
+ auto result_array = read_result.value();
+ CheckResult(result_array, expected_array, read_schema);
+ run_complete = true;
+ break;
+ }
+ ASSERT_TRUE(run_complete);
+}
+
+TEST_P(MergeFileSplitReadTest, Test09VersionWithoutInlineFieldId) {
+ std::string path = paimon::test::GetDataDir() + "/orc/pk_09.db/pk_09";
+ ReadContextBuilder context_builder(path);
+ std::vector<DataField> raw_read_fields = {DataField(3, arrow::field("f3",
arrow::float64())),
+ DataField(2, arrow::field("f2",
arrow::int32())),
+ DataField(0, arrow::field("f0",
arrow::utf8())),
+ DataField(1, arrow::field("f1",
arrow::int32()))};
+ auto read_schema =
DataField::ConvertDataFieldsToArrowSchema(raw_read_fields);
+ ASSERT_TRUE(read_schema);
+ context_builder.SetReadSchema({"f3", "f2", "f0", "f1"});
+ context_builder.SetOptions({{Options::FILE_FORMAT, "orc"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {"orc.read.enable-metrics", "true"}});
+ AddOptions(&context_builder);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
+ auto meta1 = std::make_shared<DataFileMeta>(
+ "data-00e3ed53-16ba-4537-9264-b7dc03fefc65-0.orc", /*file_size=*/803,
/*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({std::string("Tony"), 0},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({std::string("Tony"), 0},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Tony"), 0},
{std::string("Tony"), 0},
+ {0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Tony"), 10, 0, 14.1},
+ {std::string("Tony"), 10, 0, 14.1},
{0, 0, 0, 0},
+ pool_.get()),
+ /*min_sequence_number=*/5, /*max_sequence_number=*/5, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0ll, 0),
+ /*delete_row_count=*/1, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ auto meta2 = std::make_shared<DataFileMeta>(
+ "data-6871b960-edd9-40fc-9859-aaca9ea205cf-0.orc", /*file_size=*/887,
/*row_count=*/5,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({std::string("Alex"), 0},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({std::string("Tony"), 0},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alex"), 0},
{std::string("Tony"), 0},
+ {0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alex"), 10, 0, 12.1},
+ {std::string("Tony"), 10, 0, 17.1},
{0, 0, 0, 0},
+ pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ DataSplitImpl::Builder builder(
+ BinaryRowGenerator::GenerateRow({10}, pool_.get()),
+ /*bucket=*/1, /*bucket_path=*/
+ paimon::test::GetDataDir() + "/orc/pk_09.db/pk_09/f1=10/bucket-1/",
{meta1, meta2});
+ ASSERT_OK_AND_ASSIGN(auto data_split,
+
builder.WithSnapshot(5).IsStreaming(false).RawConvertible(false).Build());
+ auto internal_context = CreateInternalReadContext(read_context);
+ ASSERT_OK_AND_ASSIGN(auto batch_reader, CreateReader(internal_context,
{data_split}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result_array,
+
ReadResultCollector::CollectResult(batch_reader.get()));
+ auto fields_with_row_kind = read_schema->fields();
+ fields_with_row_kind.insert(fields_with_row_kind.begin(),
+ arrow::field("_VALUE_KIND", arrow::int8()));
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(fields_with_row_kind),
{R"([
+ [0, 16.1, 0, "Alex", 10],
+ [0, 12.1, 0, "Bob", 10],
+ [0, 17.1, 0, "David", 10],
+ [0, 13.1, 0, "Emily", 10]
+ ])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(result_array, expected_array, read_schema);
+
+ batch_reader->Close();
+ auto read_metrics = batch_reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
read_metrics->GetCounter("orc.read.io.count"));
+ ASSERT_GT(io_count, 0);
+ ASSERT_OK_AND_ASSIGN(uint64_t latency,
+
read_metrics->GetCounter("orc.read.inclusive.latency.us"));
+ ASSERT_GT(latency, 0);
+}
+
+INSTANTIATE_TEST_SUITE_P(UseMinHeapAndEnablePrefetchAndEnableMultiThreadProject,
+ MergeFileSplitReadTest,
+ ::testing::Combine(::testing::Bool(),
::testing::Bool(),
+ ::testing::Bool()));
+
+} // namespace paimon::test
