HaHaJeff commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3864076506
##########
src/paimon/core/table/source/key_value_table_read.cpp:
##########
@@ -34,6 +51,63 @@ class Executor;
class FileStorePathFactory;
class InternalReadContext;
class MemoryPool;
+struct ColumnarBatchContext;
+
+namespace {
+
+Result<std::vector<std::unique_ptr<KeyValueRecordReader>>> CreateMemoryReaders(
+ const std::shared_ptr<RealtimeSplit>& split, const
RealtimePartitionBucketView& memory,
+ const std::shared_ptr<arrow::Schema>& key_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema,
+ const std::shared_ptr<FieldsComparator>& key_comparator,
+ const std::shared_ptr<InternalReadContext>& context,
+ const std::shared_ptr<MemoryPool>& memory_pool) {
+ arrow::FieldVector prepared_fields = {
+
DataField::ConvertDataFieldToArrowField(SpecialFields::ValueKind())->WithNullable(false),
+
DataField::ConvertDataFieldToArrowField(SpecialFields::SequenceNumber())
+ ->WithNullable(false),
+
DataField::ConvertDataFieldToArrowField(SpecialFields::RealtimeOffset())};
+ prepared_fields.insert(prepared_fields.end(),
value_schema->fields().begin(),
Review Comment:
Fixed in 178ca71c8cb4b473f304e4b4aa9a6999a38a2d19; reader setup was then
simplified in 3c5c8e633d311ff05b421d47b799bb4a32194b33. The read schema now
uses a field-ID-deduplicated union of the full trimmed primary key and
projected value fields. Covered by TestPkKeylessProjection and
TestCompositePkKeylessProjection.
##########
src/paimon/core/operation/merge_file_split_read.cpp:
##########
@@ -78,6 +78,102 @@ struct KeyValue;
template <typename T>
class MergeFunctionWrapper;
+class MergeFileSplitRead::RealtimeReaderBuilder {
+ public:
+ static Result<std::unique_ptr<BatchReader>> Create(
+ const std::vector<std::shared_ptr<Split>>& disk_splits,
+ std::vector<std::unique_ptr<KeyValueRecordReader>>&&
additional_readers,
+ MergeFileSplitRead* owner) {
+ RealtimeReaderBuilder builder(owner);
+ std::vector<std::unique_ptr<KeyValueRecordReader>> readers;
+ if (!disk_splits.empty()) {
+ PAIMON_RETURN_NOT_OK(builder.CollectDiskReaders(disk_splits,
&readers));
+ }
+ readers.reserve(readers.size() + additional_readers.size());
+ for (std::unique_ptr<KeyValueRecordReader>& additional_reader :
additional_readers) {
+ readers.push_back(std::move(additional_reader));
+ }
+ return builder.CreateMergedReader(std::move(readers));
+ }
+
+ private:
+ explicit RealtimeReaderBuilder(MergeFileSplitRead* owner) : owner_(owner)
{}
+
+ Status CollectDiskReaders(const std::vector<std::shared_ptr<Split>>&
disk_splits,
+
std::vector<std::unique_ptr<KeyValueRecordReader>>* readers) {
+ std::shared_ptr<DataSplitImpl> first_split =
+ std::dynamic_pointer_cast<DataSplitImpl>(disk_splits.front());
+ if (!first_split) {
+ return Status::Invalid("merge input disk split is not a data
split");
+ }
+ const BinaryRow& partition = first_split->Partition();
+ const int32_t bucket = first_split->Bucket();
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory>
data_file_path_factory,
+
owner_->path_factory_->CreateDataFilePathFactory(partition, bucket));
+
+ std::vector<std::shared_ptr<DataFileMeta>> data_files;
+ std::vector<std::optional<DeletionFile>> deletion_files;
+ for (const std::shared_ptr<Split>& disk_split : disk_splits) {
+ std::shared_ptr<DataSplitImpl> data_split =
+ std::dynamic_pointer_cast<DataSplitImpl>(disk_split);
+ if (!data_split || !(data_split->Partition() == partition) ||
+ data_split->Bucket() != bucket) {
+ return Status::Invalid("merge input disk splits do not share a
partition-bucket");
+ }
+ if (!data_split->BeforeFiles().empty() ||
data_split->IsStreaming() ||
+ data_split->Bucket() == BucketModeDefine::POSTPONE_BUCKET) {
+ return Status::Invalid("additional merge input requires
fixed-bucket batch splits");
+ }
+ const std::vector<std::shared_ptr<DataFileMeta>>& split_files =
data_split->DataFiles();
+ const std::vector<std::optional<DeletionFile>>&
split_deletion_files =
+ data_split->DeletionFiles();
+ if (!split_deletion_files.empty() &&
+ split_deletion_files.size() != split_files.size()) {
+ return Status::Invalid(
+ "merge input disk split deletion files must be empty or
match data files");
+ }
+ data_files.insert(data_files.end(), split_files.begin(),
split_files.end());
+ if (split_deletion_files.empty()) {
+ deletion_files.insert(deletion_files.end(),
split_files.size(), std::nullopt);
+ } else {
+ deletion_files.insert(deletion_files.end(),
split_deletion_files.begin(),
+ split_deletion_files.end());
+ }
+ }
+
+ DeletionVector::Factory dv_factory;
+ std::vector<std::vector<SortedRun>> disk_sections;
+ PAIMON_RETURN_NOT_OK(
+ owner_->CreateDiskSections(data_files, deletion_files,
&dv_factory, &disk_sections));
+ for (const std::vector<SortedRun>& section : disk_sections) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<KeyValueRecordReader>>
section_readers,
+ owner_->CreateRecordReadersForSection(section, partition,
dv_factory,
+
owner_->predicate_for_keys_,
+ data_file_path_factory));
+ for (std::unique_ptr<KeyValueRecordReader>& reader :
section_readers) {
+ readers->push_back(std::move(reader));
Review Comment:
The disk side is fixed in 08a695c1934b533af516941254dccec2d86bc1ac by
composing the disk sections into one sorted disk reader before the final merge.
The memory side is not fully fixed: fan-in remains unbounded by prepared-batch
count, so many small writes increase final merge fan-in and retained first
batches. This is a current performance/resource limitation rather than a
correctness issue, and I am keeping this discussion open.
##########
src/paimon/core/operation/key_value_file_store_write.cpp:
##########
@@ -109,19 +124,74 @@ Result<std::shared_ptr<BatchWriter>>
KeyValueFileStoreWrite::CreateWriter(
PAIMON_ASSIGN_OR_RAISE(
Review Comment:
Fixed in 478138342e953adb35b443c7f65f84338fb96cfa by moving Levels::Create
into the non-realtime compact-manager branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
