lxy-9602 commented on code in PR #438:
URL: https://github.com/apache/paimon-cpp/pull/438#discussion_r4312972867
##########
src/paimon/core/core_options.cpp:
##########
@@ -599,6 +601,15 @@ struct CoreOptions::Impl {
// Parse file.format - data file format, default "parquet"
PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
Options::FILE_FORMAT, /*default_identifier=*/"parquet",
&file_format));
+ if (parser.ContainsKey(Options::VECTOR_FILE_FORMAT)) {
+ PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
+ Options::VECTOR_FILE_FORMAT, /*default_identifier=*/"vortex",
&vector_file_format));
+ }
+ PAIMON_RETURN_NOT_OK(
+ parser.ParseMemorySize(Options::VECTOR_TARGET_FILE_SIZE,
&vector_target_file_size));
+ if (vector_target_file_size && vector_target_file_size.value() <= 0) {
+ return Status::Invalid("vector.target-file-size must be positive");
+ }
Review Comment:
Please add tests for these two interfaces in `core_options_test.cpp`. May
only need to add two lines for `TestDefaultValue` and `TestFromMap`.
##########
test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/README.md:
##########
@@ -0,0 +1,128 @@
+<!--
+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.
+-->
+
Review Comment:
Please refer to the `README` of other test data and align the format.
##########
test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/GenerateDedicatedVectorRollingFixture.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.
+ */
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BinaryVector;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.CloseableIterator;
+
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** Generates one Java main file with independently rolled dedicated VECTOR
files. */
Review Comment:
Please remove this Java file and the pom file. We want the data construction
pattern to be understandable via the README alone.
##########
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:
This behavior isn’t consistent with Java. In Java, the VECTOR writer itself
is a `RollingFileWriterImpl`: when the VECTOR size reaches its target, it only
rolls the VECTOR files; the VECTOR writer is only closed when the main file
rolls. However, the current C++ implementation closes both sides together
whenever either side hits its size/line targets.
##########
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:
Can we just use the `ReadTable` function here?
##########
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:
For a single main file covered by multiple independently rolled VECTOR
files, this implementation appears to create a separate main-file reader for
each planned range when projecting both regular and vector columns. Java keeps
the sequential read path for contiguous, non-overlapping VECTOR files, avoiding
these repeated opens.
Could we at least add a code comment documenting this difference and a TODO
to preserve the sequential path for this layout? That would make the current
trade-off explicit and help track the follow-up optimization.
##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -1261,6 +1263,58 @@ Status SchemaValidation::ValidateMapStorageLayout(const
TableSchema& schema,
return Status::OK();
}
+Status SchemaValidation::ValidateVectorStore(const TableSchema& schema,
+ const CoreOptions& options) {
+ const std::shared_ptr<FileFormat> vector_format =
options.GetVectorFileFormat();
+ if (!vector_format) {
+ return Status::OK();
+ }
+ if (vector_format->Identifier() != "vortex") {
+ return Status::NotImplemented("Dedicated VECTOR storage currently only
supports vortex");
Review Comment:
It seems Java’s independently configured dedicated VECTOR storage isn’t
limited to Vortex—Lance and even Parquet support it as well. Why do we want to
enforce a specific format here?
--
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]
