ChaomingZhangCN commented on code in PR #438:
URL: https://github.com/apache/paimon-cpp/pull/438#discussion_r4323974658
##########
test/inte/blob_table_inte_test.cpp:
##########
@@ -1894,12 +1894,37 @@ TEST_P(BlobTableInteTest,
TestDataEvolutionBlobPartialUpdateMultipleLayers) {
const std::vector<std::string> expected_rows = {
R"([[1, "a", "update1_0"]])", R"([[2, "b", "blob_1"]])", R"([[3, "c",
"blob_2"]])",
R"([[4, "d", "update2_3"]])"};
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> unpruned_plan,
ScanTable(table_path));
+ ASSERT_EQ(unpruned_plan->Splits().size(), 1);
+ auto unpruned_split =
std::dynamic_pointer_cast<DataSplitImpl>(unpruned_plan->Splits()[0]);
+ ASSERT_NE(unpruned_split, nullptr);
+ ASSERT_EQ(unpruned_split->DataFiles().size(), 4);
for (int32_t i = 0; i < static_cast<int32_t>(expected_rows.size()); i++) {
+ SCOPED_TRACE(i);
auto expected_single = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
expected_rows[i])
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(),
expected_single,
/*predicate=*/nullptr, /*row_ranges=*/{Range(i,
i)}));
+
+ // Range planning can narrow a read after scanning, leaving unselected
BLOB files
+ // in the split. Their sequence layers must still align with the
selected rows.
+ auto indexed_split =
+ std::make_shared<IndexedSplitImpl>(unpruned_split,
std::vector<Range>{Range(i, i)});
+ ReadContextBuilder read_builder(table_path);
+
read_builder.SetReadFieldNames(schema->field_names()).EnableLateMaterializing(false);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context,
read_builder.Finish());
Review Comment:
Updated the test to reuse ReadTable, preserving the regression coverage.
Fixed in 491279e.
##########
src/paimon/core/io/rolling_vector_file_writer.cpp:
##########
@@ -0,0 +1,158 @@
+/*
+ * 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/io/rolling_vector_file_writer.h"
+
+#include "arrow/array/array_nested.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+
+namespace paimon {
+
+RollingVectorFileWriter::RollingVectorFileWriter(
+ int64_t target_file_size, int64_t target_file_row_num, int64_t
vector_target_file_size,
+ const std::shared_ptr<WriterFactory>& main_factory,
+ const std::shared_ptr<WriterFactory>& vector_factory,
+ const std::shared_ptr<arrow::Schema>& write_schema, int32_t vector_index)
+ : RollingFileWriter(target_file_size, target_file_row_num, main_factory),
+ vector_target_file_size_(vector_target_file_size),
+ vector_factory_(vector_factory),
+ write_schema_(write_schema),
+ vector_index_(vector_index) {}
+
+Status RollingVectorFileWriter::Write(::ArrowArray* record) {
+ if (closed_ || aborted_) {
+ return Status::Invalid("Writer has already closed or aborted!");
+ }
+ ScopeGuard guard([this]() { Abort(); });
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(record, arrow::struct_(write_schema_->fields())));
+ if (!array || array->type_id() != arrow::Type::STRUCT ||
array->null_count() != 0) {
+ return Status::Invalid("Dedicated VECTOR storage expects non-null
struct rows");
+ }
+ if (array->length() == 0) {
+ guard.Release();
+ return Status::OK();
+ }
+ if (!current_writer_) {
+ PAIMON_RETURN_NOT_OK(OpenCurrentWriter());
+ PAIMON_ASSIGN_OR_RAISE(vector_writer_,
vector_factory_->CreateWriter());
+ }
+ auto rows = checked_pointer_cast<arrow::StructArray>(array);
+ arrow::ArrayVector main_arrays;
+ arrow::FieldVector main_fields;
+ for (int32_t i = 0; i < write_schema_->num_fields(); ++i) {
+ if (i != vector_index_) {
+ main_arrays.push_back(rows->field(i));
+ main_fields.push_back(write_schema_->field(i));
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray>
main_array,
+ arrow::StructArray::Make(main_arrays,
main_fields));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> vector_array,
+ arrow::StructArray::Make({rows->field(vector_index_)},
+ {write_schema_->field(vector_index_)}));
+ ::ArrowArray main_batch = {};
+ ::ArrowArray vector_batch = {};
+ ScopeGuard arrays_guard([&]() {
+ ArrowArrayRelease(&main_batch);
+ ArrowArrayRelease(&vector_batch);
+ });
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*main_array,
&main_batch));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*vector_array,
&vector_batch));
+ PAIMON_RETURN_NOT_OK(current_writer_->Write(&main_batch));
+ PAIMON_RETURN_NOT_OK(vector_writer_->Write(&vector_batch));
+ record_count_ += rows->length();
+ current_file_record_count_ += rows->length();
+
+ const bool suggested_check = SuggestCheck();
+ PAIMON_ASSIGN_OR_RAISE(bool main_full,
+ current_writer_->ReachTargetSize(suggested_check,
target_file_size_));
+ PAIMON_ASSIGN_OR_RAISE(bool vector_full, vector_writer_->ReachTargetSize(
+ suggested_check,
vector_target_file_size_));
+ if (main_full || vector_full || current_file_record_count_ >=
target_file_row_num_) {
+ PAIMON_RETURN_NOT_OK(CloseCurrentPair());
Review Comment:
Updated the writer to roll VECTOR files independently. Rolling the main file
also closes the active VECTOR file. Aligned Vortex's size estimate with Java
and added regression tests for file ordering, row ID ranges, and failure
cleanup. Fixed in 491279e.
##########
src/paimon/core/operation/data_evolution_split_read.cpp:
##########
@@ -674,7 +676,200 @@ DataEvolutionSplitRead::SplitFieldBunches(
return fields_files;
}
-Result<std::unique_ptr<DataEvolutionFileReader>>
DataEvolutionSplitRead::CreateUnionReader(
+Result<std::vector<DataEvolutionSplitRead::VectorReadRange>>
+DataEvolutionSplitRead::PlanVectorReadRanges(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::vector<int32_t>& read_field_ids,
+ const std::function<Result<std::vector<int32_t>>(const
std::shared_ptr<DataFileMeta>&)>&
+ file_to_field_ids) {
+ struct Candidate {
+ std::shared_ptr<DataFileMeta> file;
+ std::vector<int32_t> field_ids;
+ };
+ std::vector<Candidate> candidates;
+ std::vector<std::shared_ptr<DataFileMeta>> other_files;
+ std::map<int64_t, std::vector<size_t>> boundaries;
+ std::unordered_set<int32_t> projected_ids(read_field_ids.begin(),
read_field_ids.end());
+ std::optional<Range> logical_range;
+ int64_t first_row_id = std::numeric_limits<int64_t>::max();
+ int64_t end_row_id = std::numeric_limits<int64_t>::min();
+ bool vector_only = std::all_of(files.begin(), files.end(), [](const auto&
file) {
+ return VectorStoreUtils::IsVectorStoreFile(file->file_name);
+ });
+ for (const auto& file : files) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first, file->NonNullFirstRowId());
+ if (first < 0 || file->row_count <= 0 ||
+ first > std::numeric_limits<int64_t>::max() - file->row_count) {
+ return Status::Invalid("Invalid row id range in VECTOR merge
group");
+ }
+ int64_t end = first + file->row_count;
+ first_row_id = std::min(first_row_id, first);
+ end_row_id = std::max(end_row_id, end);
+ if (!VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ other_files.push_back(file);
+ if (DataEvolutionUtils::IsNormalFile(file->file_name)) {
+ logical_range = Range(first, end - 1);
+ }
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> field_ids,
file_to_field_ids(file));
+ field_ids.erase(std::remove_if(field_ids.begin(), field_ids.end(),
+ [&](int32_t id) { return
projected_ids.count(id) == 0; }),
+ field_ids.end());
+ if (!field_ids.empty() || vector_only) {
+ size_t index = candidates.size();
+ candidates.push_back(Candidate{file, std::move(field_ids)});
+ boundaries[first].push_back(index);
+ boundaries[end].push_back(index);
+ }
+ }
+ if (files.empty()) {
+ return std::vector<VectorReadRange>();
+ }
+ if (!logical_range) {
+ logical_range = Range(first_row_id, end_row_id - 1);
+ }
+ boundaries[logical_range->from];
+ boundaries[logical_range->to + 1];
+
+ auto newest_first = [&](size_t left, size_t right) {
+ const auto& lhs = candidates[left].file;
+ const auto& rhs = candidates[right].file;
+ if (lhs->max_sequence_number != rhs->max_sequence_number) {
+ return lhs->max_sequence_number > rhs->max_sequence_number;
+ }
+ if (lhs->file_name != rhs->file_name) {
+ return lhs->file_name < rhs->file_name;
+ }
+ return left < right;
+ };
+ std::set<size_t, decltype(newest_first)> active(newest_first);
+ std::vector<VectorReadRange> result;
+ int64_t start = boundaries.begin()->first;
+ for (const auto& [end, events] : boundaries) {
+ int64_t from = std::max(start, logical_range->from);
+ int64_t to = std::min(end - 1, logical_range->to);
+ if (from <= to) {
+ std::unordered_set<int32_t> assigned;
+ std::vector<std::shared_ptr<DataFileMeta>> providers = other_files;
+ for (size_t index : active) {
+ const auto& candidate = candidates[index];
+ bool provides_field = false;
+ for (int32_t id : candidate.field_ids) {
+ provides_field |= assigned.insert(id).second;
+ }
+ // An explicit NULL in a newer file overrides the older value
as well.
+ if (provides_field) {
+ providers.push_back(candidate.file);
+ }
+ }
+ if (providers.empty() && vector_only && !active.empty()) {
+ providers.push_back(candidates[*active.begin()].file);
+ }
+ // Ordinary inline fields and dedicated fields share the same
sequence precedence.
+ // BLOB files remain last and retain their own per-row fallback
ordering.
+ std::stable_sort(
+ providers.begin(), providers.end(), [](const auto& lhs, const
auto& rhs) {
+ bool lhs_blob = BlobUtils::IsBlobFile(lhs->file_name);
+ bool rhs_blob = BlobUtils::IsBlobFile(rhs->file_name);
+ if (lhs_blob != rhs_blob) {
+ return !lhs_blob;
+ }
+ return !lhs_blob && lhs->max_sequence_number >
rhs->max_sequence_number;
+ });
+ if (!result.empty() && result.back().range.to + 1 == from &&
+ result.back().files == providers) {
+ result.back().range.to = to;
+ } else {
+ result.push_back(VectorReadRange{Range(from, to),
std::move(providers)});
+ }
+ }
+ for (size_t index : events) {
+ if (active.erase(index) == 0) {
+ active.insert(index);
+ }
+ }
+ start = end;
+ }
+ return result;
Review Comment:
Added an explanation and a TODO for a sequential fast path to avoid
reopening the main file across ranges. Fixed in 491279e.
--
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]
