This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 0a9ed795 perf(vector): reuse main readers across sequential vector
files (#453)
0a9ed795 is described below
commit 0a9ed795bed3426f4cba868721c83101fe416116
Author: 小明同学 <[email protected]>
AuthorDate: Fri Oct 23 23:03:30 2026 +0800
perf(vector): reuse main readers across sequential vector files (#453)
---
.../core/operation/data_evolution_split_read.cpp | 149 ++++++++++++-
.../core/operation/data_evolution_split_read.h | 32 +++
.../operation/data_evolution_split_read_test.cpp | 231 +++++++++++++++++++++
test/inte/data_evolution_table_test.cpp | 133 +++++++++++-
4 files changed, 528 insertions(+), 17 deletions(-)
diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp
b/src/paimon/core/operation/data_evolution_split_read.cpp
index 6d065b46..5fea0f21 100644
--- a/src/paimon/core/operation/data_evolution_split_read.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read.cpp
@@ -777,6 +777,65 @@ DataEvolutionSplitRead::PlanVectorReadRanges(
return result;
}
+Result<std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>>
+DataEvolutionSplitRead::TryPlanSequentialVectorRead(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::function<Result<std::vector<int32_t>>(const
std::shared_ptr<DataFileMeta>&)>&
+ file_to_field_ids) {
+ std::vector<std::shared_ptr<FieldBunch>> no_match;
+ std::shared_ptr<DataFileMeta> main_file;
+ std::vector<std::shared_ptr<DataFileMeta>> vector_files;
+ for (const auto& file : files) {
+ if (VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ vector_files.push_back(file);
+ } else if (DataEvolutionUtils::IsNormalFile(file->file_name) &&
!main_file) {
+ main_file = file;
+ } else {
+ return no_match;
+ }
+ }
+ if (!main_file || vector_files.empty()) {
+ return no_match;
+ }
+ 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");
+ }
+ }
+ std::sort(vector_files.begin(), vector_files.end(), [](const auto& left,
const auto& right) {
+ return left->first_row_id.value() < right->first_row_id.value();
+ });
+ const auto& first_vector = vector_files[0];
+ if (!first_vector->write_cols || first_vector->write_cols->size() != 1) {
+ return no_match;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::string format, first_vector->FileFormat());
+ int64_t next_row_id = main_file->first_row_id.value();
+ for (const auto& file : vector_files) {
+ PAIMON_ASSIGN_OR_RAISE(std::string current_format, file->FileFormat());
+ if (file->schema_id != first_vector->schema_id || current_format !=
format ||
+ file->write_cols != first_vector->write_cols || file->first_row_id
!= next_row_id) {
+ return no_match;
+ }
+ next_row_id += file->row_count;
+ }
+ if (next_row_id != main_file->first_row_id.value() + main_file->row_count)
{
+ return no_match;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> vector_ids,
file_to_field_ids(first_vector));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> main_ids,
file_to_field_ids(main_file));
+ // An inline VECTOR in a historical main file can have different
precedence in each range.
+ if (vector_ids.size() != 1 ||
+ std::find(main_ids.begin(), main_ids.end(), vector_ids[0]) !=
main_ids.end()) {
+ return no_match;
+ }
+ return std::vector<std::shared_ptr<FieldBunch>>{
+ std::make_shared<DataBunch>(main_file),
+ std::make_shared<VectorBunch>(std::move(vector_files),
main_file->row_count)};
+}
+
Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateUnionReader(
const BinaryRow& partition, const
std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files,
const std::optional<std::vector<Range>>& row_ranges,
@@ -802,13 +861,16 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateUnionReader(
}
auto file_to_field_ids =
[&](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
- if (!file->write_cols) {
+ if (VectorStoreUtils::IsVectorStoreFile(file->file_name) &&
!file->write_cols) {
return Status::Invalid("VECTOR files must declare their write
columns");
}
auto data_schema = context_->GetTableSchema();
if (file->schema_id != data_schema->Id()) {
PAIMON_ASSIGN_OR_RAISE(data_schema,
schema_manager_->ReadSchema(file->schema_id));
}
+ if (!file->write_cols) {
+ return DataField::GetAllFieldIds(data_schema->Fields());
+ }
std::vector<int32_t> ids;
for (const auto& name : file->write_cols.value()) {
if (SpecialFields::IsSystemField(name)) {
@@ -819,19 +881,31 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateUnionReader(
}
return ids;
};
- PAIMON_ASSIGN_OR_RAISE(
- std::vector<VectorReadRange> ranges,
- PlanVectorReadRanges(need_merge_files, read_field_ids,
file_to_field_ids));
- // Each range creates a separate field-union reader, reopening a main file
that spans ranges.
- // TODO(ChaomingZhangCN): Keep one main reader and concatenate VECTOR
readers for contiguous,
- // non-overlapping files whose field providers can be merged sequentially.
- std::vector<LazyConcatBatchReader::Source> sources;
- auto split_read =
- std::make_shared<DataEvolutionSplitRead>(path_factory_, context_,
pool_, executor_);
std::optional<std::vector<Range>> normalized_ranges;
if (row_ranges) {
normalized_ranges = Range::SortAndMergeOverlap(row_ranges.value(),
/*adjacent=*/true);
}
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FieldBunch>>
sequential_bunches,
+ TryPlanSequentialVectorRead(need_merge_files,
file_to_field_ids));
+ auto split_read =
+ std::make_shared<DataEvolutionSplitRead>(path_factory_, context_,
pool_, executor_);
+ if (!sequential_bunches.empty()) {
+ std::vector<LazyConcatBatchReader::Source> sources;
+ sources.push_back(LazyConcatBatchReader::Source{
+ sequential_bunches[0]->Files()[0]->file_name,
+ [split_read, partition, sequential_bunches =
std::move(sequential_bunches),
+ normalized_ranges, data_file_path_factory, group_dv_factory,
group_dv,
+ read_schema]() -> Result<std::unique_ptr<BatchReader>> {
+ return split_read->CreateFieldUnionReader(partition,
sequential_bunches,
+ normalized_ranges,
data_file_path_factory,
+ group_dv_factory,
group_dv, read_schema);
+ }});
+ return std::make_unique<LazyConcatBatchReader>(std::move(sources),
arrow_pool_);
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<VectorReadRange> ranges,
+ PlanVectorReadRanges(need_merge_files, read_field_ids,
file_to_field_ids));
+ std::vector<LazyConcatBatchReader::Source> sources;
for (const auto& range : ranges) {
std::vector<Range> selected = {range.range};
if (normalized_ranges) {
@@ -884,6 +958,17 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
SplitFieldBunches(need_merge_files, blob_field_to_field_id,
/*has_row_ranges_selection=*/row_ranges.has_value()));
+ return CreateFieldUnionReader(partition, fields_files, row_ranges,
data_file_path_factory,
+ group_dv_factory, group_dv, read_schema);
+}
+
+Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionReader(
+ const BinaryRow& partition, const
std::vector<std::shared_ptr<FieldBunch>>& fields_files,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const DeletionVector::Factory& group_dv_factory,
+ const std::optional<GroupDeletionVector>& group_dv,
+ const std::shared_ptr<arrow::Schema>& read_schema) const {
assert(!fields_files.empty());
int64_t row_count = fields_files[0]->RowCount();
PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id,
fields_files[0]->Files()[0]->NonNullFirstRowId());
@@ -912,6 +997,10 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
std::vector<int32_t> reader_offsets(all_read_fields.size(), -1);
// which field index in the reading row
std::vector<int32_t> field_offsets(all_read_fields.size(), -1);
+ const bool has_normal_file =
+ std::any_of(fields_files.begin(), fields_files.end(), [](const auto&
bunch) {
+ return
DataEvolutionUtils::IsNormalFile(bunch->Files()[0]->file_name);
+ });
for (int32_t file_idx = 0; file_idx <
static_cast<int32_t>(fields_files.size()); file_idx++) {
const auto& bunch = fields_files[file_idx];
@@ -932,6 +1021,11 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
if (reader_offsets[read_field_idx] != -1) {
continue;
}
+ // Java takes sequence numbers from normal files before dedicated
column files.
+ if (has_normal_file &&
!DataEvolutionUtils::IsNormalFile(first_file->file_name) &&
+ read_field_ids[read_field_idx] ==
SpecialFields::SequenceNumber().Id()) {
+ continue;
+ }
for (int32_t data_field_id : data_field_ids) {
// Check if the read field index matches the file field
// index
@@ -950,7 +1044,40 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
// create new FieldMappingReader for read partial fields
auto file_read_schema =
DataField::ConvertDataFieldsToArrowSchema(read_fields_in_file);
auto blob_bunch = std::dynamic_pointer_cast<BlobBunch>(bunch);
- if (blob_bunch && !blob_bunch->SequentialReadOptimize()) {
+ if (std::dynamic_pointer_cast<VectorBunch>(bunch)) {
+ std::vector<LazyConcatBatchReader::Source> sources;
+ auto split_read =
std::make_shared<DataEvolutionSplitRead>(path_factory_, context_,
+
pool_, executor_);
+ for (const auto& file : bunch->Files()) {
+ std::optional<std::vector<Range>> selected = row_ranges;
+ if (row_ranges) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first,
file->NonNullFirstRowId());
+ selected = Range::And({Range(first, first +
file->row_count - 1)},
+ row_ranges.value());
+ if (selected->empty()) {
+ continue;
+ }
+ }
+ sources.push_back(LazyConcatBatchReader::Source{
+ file->file_name,
+ [split_read, partition, file, file_read_schema,
selected,
+ data_file_path_factory,
+ group_dv_factory]() ->
Result<std::unique_ptr<BatchReader>> {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<FileBatchReader>>
readers,
+ split_read->CreateRawFileReaders(
+ partition, {file}, file_read_schema,
/*predicate=*/nullptr,
+ group_dv_factory, selected,
data_file_path_factory,
+ /*extra_format_options=*/{}));
+ return std::make_unique<ConcatBatchReader>(
+
ObjectUtils::MoveVector<std::unique_ptr<BatchReader>>(
+ std::move(readers)),
+ split_read->arrow_pool_);
+ }});
+ }
+ file_batch_readers[file_idx] =
+
std::make_unique<LazyConcatBatchReader>(std::move(sources), arrow_pool_);
+ } else if (blob_bunch && !blob_bunch->SequentialReadOptimize()) {
// blob files span multiple max sequence number layers:
placeholder entries of
// newer layers must fall back row by row to older layers
PAIMON_ASSIGN_OR_RAISE(
diff --git a/src/paimon/core/operation/data_evolution_split_read.h
b/src/paimon/core/operation/data_evolution_split_read.h
index b51dd447..e0406495 100644
--- a/src/paimon/core/operation/data_evolution_split_read.h
+++ b/src/paimon/core/operation/data_evolution_split_read.h
@@ -26,6 +26,7 @@
#include <string>
#include <unordered_map>
#include <unordered_set>
+#include <utility>
#include <vector>
#include "paimon/common/data/blob_view_struct.h"
@@ -123,6 +124,23 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
std::vector<std::shared_ptr<DataFileMeta>> data_files_;
};
+ /// Compatible VECTOR files tiling one main file's row range without gaps
or overlaps.
+ class VectorBunch : public FieldBunch {
+ public:
+ VectorBunch(std::vector<std::shared_ptr<DataFileMeta>> files, int64_t
row_count)
+ : files_(std::move(files)), row_count_(row_count) {}
+ int64_t RowCount() const override {
+ return row_count_;
+ }
+ const std::vector<std::shared_ptr<DataFileMeta>>& Files() const
override {
+ return files_;
+ }
+
+ private:
+ std::vector<std::shared_ptr<DataFileMeta>> files_;
+ int64_t row_count_;
+ };
+
/// All blob files of one blob field in a merge group. Unlike data files,
every file is kept
/// (aligned with Java's BlobFileBunch): files whose max sequence numbers
differ form layers
/// of a data-evolution partial update, where newer layers record
untouched rows as
@@ -227,6 +245,12 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
const std::function<Result<std::vector<int32_t>>(const
std::shared_ptr<DataFileMeta>&)>&
file_to_field_ids);
+ /// Empty means the layout needs range planning. Errors resolving
providers still propagate.
+ static Result<std::vector<std::shared_ptr<FieldBunch>>>
TryPlanSequentialVectorRead(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::function<Result<std::vector<int32_t>>(const
std::shared_ptr<DataFileMeta>&)>&
+ file_to_field_ids);
+
static
Result<std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>>
SplitFieldBunches(const std::vector<std::shared_ptr<DataFileMeta>>&
need_merge_files,
const std::function<Result<int32_t>(const
std::shared_ptr<DataFileMeta>&)>&
@@ -278,6 +302,14 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
const std::optional<GroupDeletionVector>& group_dv,
const std::shared_ptr<arrow::Schema>& read_schema) const;
+ Result<std::unique_ptr<BatchReader>> CreateFieldUnionReader(
+ const BinaryRow& partition, const
std::vector<std::shared_ptr<FieldBunch>>& fields_files,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const DeletionVector::Factory& group_dv_factory,
+ const std::optional<GroupDeletionVector>& group_dv,
+ const std::shared_ptr<arrow::Schema>& read_schema) const;
+
/// Builds the row-level fallback reader for a blob bunch spanning
multiple max sequence
/// number layers: groups the files by max sequence number, pads uncovered
row id ranges of
/// each layer with placeholder gap segments, and resolves each row to the
newest
diff --git a/src/paimon/core/operation/data_evolution_split_read_test.cpp
b/src/paimon/core/operation/data_evolution_split_read_test.cpp
index 0b113725..4bd0e0c2 100644
--- a/src/paimon/core/operation/data_evolution_split_read_test.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp
@@ -19,14 +19,21 @@
#include "paimon/core/operation/data_evolution_split_read.h"
#include <algorithm>
+#include <atomic>
#include <cstdint>
#include <map>
#include <optional>
#include <utility>
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/table/special_fields.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/vector_store_utils.h"
#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
@@ -39,15 +46,42 @@
#include "paimon/core/table/source/data_split_impl.h"
#include "paimon/defs.h"
#include "paimon/executor.h"
+#include "paimon/format/file_format_factory.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/writer_builder.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/read_context.h"
#include "paimon/status.h"
+#include "paimon/testing/utils/read_result_collector.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
+namespace {
+class CountingLocalFileSystem : public LocalFileSystem {
+ public:
+ explicit CountingLocalFileSystem(std::string counted_path)
+ : counted_path_(std::move(counted_path)) {}
+
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ if (path == counted_path_) {
+ ++open_count_;
+ }
+ return LocalFileSystem::Open(path);
+ }
+
+ int32_t OpenCount() const {
+ return open_count_.load();
+ }
+
+ private:
+ std::string counted_path_;
+ mutable std::atomic<int32_t> open_count_{0};
+};
+} // namespace
+
class DataEvolutionSplitReadTest : public ::testing::Test {
public:
std::shared_ptr<DataFileMeta> CreateDataFileMeta(
@@ -751,6 +785,203 @@ TEST_F(DataEvolutionSplitReadTest,
TestPlanIndependentlyRolledVectorReadRanges)
ASSERT_EQ(vector1->row_count, 4);
}
+TEST_F(DataEvolutionSplitReadTest, TestPlanSequentialVectorRead) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 100, 3, 1);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 103, 4, 2);
+ auto vector2 = CreateDataFileMeta("vector2.vector.vortex", 107, 3, 3);
+ for (const auto& file : {vector0, vector1, vector2}) {
+ file->write_cols = std::vector<std::string>{"embedding"};
+ }
+ auto field_ids =
+ [main](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
+ return file == main ? std::vector<int32_t>{0, 2} :
std::vector<int32_t>{1};
+ };
+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
bunches,
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead(
+ {vector2, main, vector0, vector1}, field_ids));
+ ASSERT_EQ(bunches.size(), 2);
+ ASSERT_EQ(bunches[0]->Files(),
std::vector<std::shared_ptr<DataFileMeta>>{main});
+ ASSERT_EQ(bunches[1]->Files(),
+ (std::vector<std::shared_ptr<DataFileMeta>>{vector0, vector1,
vector2}));
+ ASSERT_EQ(bunches[0]->RowCount(), 10);
+ ASSERT_EQ(bunches[1]->RowCount(), 10);
+ ASSERT_EQ(vector1->first_row_id, 103);
+ ASSERT_EQ(vector1->row_count, 4);
+}
+
+TEST_F(DataEvolutionSplitReadTest,
TestSequentialVectorReadFallsBackForUnsupportedLayouts) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 100, 4, 1);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 104, 6, 1);
+ vector0->write_cols = vector1->write_cols =
std::vector<std::string>{"embedding"};
+ auto field_ids =
+ [main](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
+ return file == main ? std::vector<int32_t>{0, 2} :
std::vector<int32_t>{1};
+ };
+ auto check_fallback = [&](const
std::vector<std::shared_ptr<DataFileMeta>>& files) {
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
bunches,
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead(files,
field_ids));
+ ASSERT_TRUE(bunches.empty());
+ };
+ check_fallback({vector0, vector1});
+ check_fallback({main, main, vector0, vector1});
+ check_fallback({main, vector0, vector1, CreateBlobFile("payload", 100, 10,
1, {{"payload"}})});
+ check_fallback({main, vector1});
+ vector1->first_row_id = 103;
+ check_fallback({main, vector0, vector1});
+ vector1->first_row_id = 105;
+ check_fallback({main, vector0, vector1});
+ vector1->first_row_id = 104;
+ vector1->row_count = 7;
+ check_fallback({main, vector0, vector1});
+ vector1->row_count = 6;
+ vector1->schema_id = 1;
+ check_fallback({main, vector0, vector1});
+ vector1->schema_id = 0;
+ vector1->file_name = "vector1.vector.parquet";
+ check_fallback({main, vector0, vector1});
+ vector1->file_name = "vector1.vector.vortex";
+ vector1->write_cols = std::vector<std::string>{"another_embedding"};
+ check_fallback({main, vector0, vector1});
+ vector1->write_cols = vector0->write_cols =
+ std::vector<std::string>{"embedding", "another_embedding"};
+ check_fallback({main, vector0, vector1});
+
+ vector1->write_cols = vector0->write_cols =
std::vector<std::string>{"embedding"};
+ auto inline_field_ids =
+ [main](const std::shared_ptr<DataFileMeta>& file) ->
Result<std::vector<int32_t>> {
+ return file == main ? std::vector<int32_t>{0, 1, 2} :
std::vector<int32_t>{1};
+ };
+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
bunches,
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead(
+ {main, vector0, vector1}, inline_field_ids));
+ ASSERT_TRUE(bunches.empty());
+}
+
+TEST_F(DataEvolutionSplitReadTest,
TestSequentialVectorReadPropagatesMetadataErrors) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector = CreateDataFileMeta("vector.vector.vortex", 100, 10, 1);
+ vector->write_cols = std::vector<std::string>{"embedding"};
+ auto failed_schema = [](const std::shared_ptr<DataFileMeta>&) ->
Result<std::vector<int32_t>> {
+ return Status::IOError("schema unavailable");
+ };
+ auto result =
+ DataEvolutionSplitRead::TryPlanSequentialVectorRead({main, vector},
failed_schema);
+ ASSERT_NOK_WITH_MSG(result, "schema unavailable");
+ ASSERT_TRUE(result.status().IsIOError());
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestSequentialVectorReadReusesMainReader) {
+ auto dir = UniqueTestDirectory::Create("local");
+ auto fs = std::make_shared<LocalFileSystem>();
+ auto schema = DataField::ConvertDataFieldsToArrowSchema(
+ {DataField(0, arrow::field("id", arrow::int32())),
+ DataField(1, arrow::field("embedding",
arrow::fixed_size_list(arrow::float32(), 2)))});
+ const std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"},
+
{Options::READ_BATCH_SIZE, "2"}};
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+ TableSchema::Create(0, schema, {}, {}, options));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ FileStorePathFactory::Create(dir->Str(), schema, {}, "default",
"parquet", "data-",
+ /*legacy_partition_name_enabled=*/true,
{}, std::nullopt,
+ /*index_file_in_data_file_dir=*/false,
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(std::string bucket_path,
+ path_factory->BucketPath(BinaryRow::EmptyRow(), 0));
+ ASSERT_OK(fs->Mkdirs(bucket_path));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileFormat> format,
+ FileFormatFactory::Get("parquet", options));
+ auto write_file = [&](const std::string& file_name, int64_t first_row_id,
+ const std::shared_ptr<arrow::Array>& array,
+ int64_t sequence) ->
Result<std::shared_ptr<DataFileMeta>> {
+ ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportType(*array->type(),
&c_schema));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriterBuilder> builder,
+ format->CreateWriterBuilder(&c_schema, 2));
+ const std::string path = PathUtil::JoinPath(bucket_path, file_name);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> out,
fs->Create(path, false));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatWriter> writer,
builder->Build(out, "zstd"));
+ ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array));
+ PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array));
+ PAIMON_RETURN_NOT_OK(writer->Finish());
+ PAIMON_RETURN_NOT_OK(out->Close());
+ PAIMON_ASSIGN_OR_RAISE(FileStatus file_status,
fs->GetFileStatus(path));
+ std::vector<std::string> write_cols;
+ for (const auto& field : array->type()->fields()) {
+ write_cols.push_back(field->name());
+ }
+ return DataFileMeta::ForAppend(file_name, file_status.GetLen(),
array->length(),
+ SimpleStats::EmptyStats(), sequence,
sequence,
+ /*schema_id=*/0, FileSource::Append(),
std::nullopt,
+ std::nullopt, first_row_id, write_cols);
+ };
+ auto main_data =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({schema->field(0)}),
+ "[[0], [1],
[2], [3], [4], [5]]")
+ .ValueOrDie();
+ auto vector_type = arrow::struct_(
+ {DataField(1, arrow::field("embedding",
arrow::list(arrow::float32()))).ArrowField()});
+ auto vector0_data =
+ arrow::ipc::internal::json::ArrayFromJSON(vector_type, "[[[0, 1]],
[null], [[2, 3]]]")
+ .ValueOrDie();
+ auto vector1_data =
+ arrow::ipc::internal::json::ArrayFromJSON(vector_type, "[[[3, 4]],
[[4, 5]], [[5, 6]]]")
+ .ValueOrDie();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> main,
+ write_file("main.parquet", 0, main_data, 3));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> vector0,
+ write_file("vector0.vector.parquet", 0, vector0_data,
4));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> vector1,
+ write_file("vector1.vector.parquet", 3, vector1_data,
5));
+ DataSplitImpl::Builder split_builder(BinaryRow::EmptyRow(), 0, bucket_path,
+ {vector1, main, vector0});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataSplitImpl> split,
+ split_builder.RawConvertible(false).Build());
+ auto expected_type =
arrow::struct_({SpecialFields::ValueKind().ArrowField(), schema->field(0),
+ schema->field(1),
SpecialFields::RowId().ArrowField(),
+
SpecialFields::SequenceNumber().ArrowField()});
+ auto expected = checked_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(expected_type, R"([
+ [0, 0, [0, 1], 0, 3], [0, 1, null, 1, 3], [0, 2, [2, 3], 2, 3],
+ [0, 3, [3, 4], 3, 3], [0, 4, [4, 5], 4, 3], [0, 5, [5, 6], 5, 3]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor,
CreateDefaultExecutor(2));
+ for (const std::vector<std::string>& projection :
+ {std::vector<std::string>{"id", "embedding"},
+ std::vector<std::string>{"id", "embedding", "_ROW_ID",
"_SEQUENCE_NUMBER"},
+ std::vector<std::string>{"id", "_ROW_ID", "_SEQUENCE_NUMBER"},
+ std::vector<std::string>{"_ROW_ID", "_SEQUENCE_NUMBER"}}) {
+ auto counting_fs = std::make_shared<CountingLocalFileSystem>(
+ PathUtil::JoinPath(bucket_path, main->file_name));
+ ReadContextBuilder builder(dir->Str());
+
builder.WithFileSystem(counting_fs).EnablePrefetch(false).SetReadFieldNames(projection);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+ auto split_read =
std::make_unique<DataEvolutionSplitRead>(path_factory, context,
+
GetDefaultPool(), executor);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader,
split_read->CreateReader(split));
+ split_read.reset();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+
ReadResultCollector::CollectResult(std::move(reader)));
+ ASSERT_NE(result, nullptr);
+ arrow::ArrayVector columns = {expected->field(0)};
+ std::vector<std::string> names = {"_VALUE_KIND"};
+ for (const auto& name : projection) {
+ columns.push_back(expected->GetFieldByName(name));
+ names.push_back(name);
+ }
+ auto projected = arrow::StructArray::Make(columns, names).ValueOrDie();
+ ASSERT_TRUE(result->Equals(arrow::ChunkedArray(projected))) <<
result->ToString();
+ ASSERT_EQ(counting_fs->OpenCount(), 1);
+ }
+}
+
TEST_F(DataEvolutionSplitReadTest,
TestPlanVectorReadRangesPreservesOldPrefixAndSuffix) {
auto main = CreateNormalFile("main.parquet", 0, 10, 0);
auto old_vector = CreateDataFileMeta("old.vector.vortex", 0, 10, 1);
diff --git a/test/inte/data_evolution_table_test.cpp
b/test/inte/data_evolution_table_test.cpp
index 2924e541..395de88d 100644
--- a/test/inte/data_evolution_table_test.cpp
+++ b/test/inte/data_evolution_table_test.cpp
@@ -18,6 +18,7 @@
*/
#include <algorithm>
#include <iterator>
+#include <set>
#include <tuple>
#include "arrow/type.h"
@@ -2027,6 +2028,77 @@ TEST_P(DataEvolutionTableTest,
TestReadJavaIndependentlyRolledVectorStore) {
ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected_rows)))
<< "actual=" << result->ToString() << "\nexpected=" <<
expected_rows->ToString();
}
+
+ // Like Java, the normal file supplies sequence numbers even when VECTOR
files are newer.
+ auto sequence_main = std::make_shared<DataFileMeta>(*main_file);
+ sequence_main->AssignSequenceNumber(3, 3);
+ std::vector<std::shared_ptr<DataFileMeta>> sequence_files =
{sequence_main};
+ int64_t sequence = 2;
+ for (const auto& [first, file] : vector_files) {
+ auto sequence_vector = std::make_shared<DataFileMeta>(*file);
+ sequence_vector->AssignSequenceNumber(sequence, sequence);
+ sequence_files.push_back(sequence_vector);
+ ++sequence;
+ }
+ auto sequence_tail = sequence_files.back();
+ DataSplitImpl::Builder sequence_builder(original_split->Partition(),
original_split->Bucket(),
+ original_split->BucketPath(),
+ std::move(sequence_files));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataSplitImpl> sequence_split,
+ sequence_builder.RawConvertible(false).Build());
+ auto tracking = PrepareBulkData(
+ kRowCount, [](int32_t id) { return fmt::format("0, {}, 3", id); },
+ {SpecialFields::ValueKind().ArrowField(),
SpecialFields::RowId().ArrowField(),
+ SpecialFields::SequenceNumber().ArrowField()});
+ for (bool sequential : {true, false}) {
+ SCOPED_TRACE(sequential);
+ std::shared_ptr<Split> split = sequence_split;
+ std::shared_ptr<arrow::Array> vectors = expected->field(1);
+ if (!sequential) {
+ // A newer VECTOR covering only the tail forces the range-reader
fallback.
+ const auto& tail = sequence_tail;
+ ASSERT_GT(tail->max_sequence_number,
sequence_main->max_sequence_number);
+ DataSplitImpl::Builder tail_builder(
+ original_split->Partition(), original_split->Bucket(),
original_split->BucketPath(),
+ std::vector<std::shared_ptr<DataFileMeta>>{sequence_main,
tail});
+ ASSERT_OK_AND_ASSIGN(split,
tail_builder.RawConvertible(false).Build());
+ const int64_t first = tail->first_row_id.value();
+ auto missing = arrow::MakeArrayOfNull(vector_type,
first).ValueOrDie();
+ vectors = arrow::Concatenate({missing,
vectors->Slice(first)}).ValueOrDie();
+ }
+ auto sequence_expected =
+ arrow::StructArray::Make(
+ {tracking->field(0), expected->field(0), vectors,
tracking->field(1),
+ tracking->field(2)},
+ {"_VALUE_KIND", "id", "embedding", "_ROW_ID",
"_SEQUENCE_NUMBER"})
+ .ValueOrDie();
+ for (const std::vector<std::string>& projection :
+ {std::vector<std::string>{"id", "embedding", "_ROW_ID",
"_SEQUENCE_NUMBER"},
+ std::vector<std::string>{"id", "_ROW_ID", "_SEQUENCE_NUMBER"},
+ std::vector<std::string>{"_ROW_ID", "_SEQUENCE_NUMBER"}}) {
+ ReadContextBuilder sequence_read_builder(table_path);
+ sequence_read_builder.SetReadFieldNames(projection);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> sequence_context,
+ sequence_read_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> sequence_read,
+
TableRead::Create(std::move(sequence_context)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> sequence_reader,
+ sequence_read->CreateReader(split));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray>
sequence_result,
+
ReadResultCollector::CollectResult(std::move(sequence_reader)));
+ ASSERT_NE(sequence_result, nullptr);
+ arrow::ArrayVector projected_arrays = {tracking->field(0)};
+ std::vector<std::string> projected_names = {"_VALUE_KIND"};
+ for (const auto& name : projection) {
+
projected_arrays.push_back(sequence_expected->GetFieldByName(name));
+ projected_names.push_back(name);
+ }
+ auto projected_expected =
+ arrow::StructArray::Make(projected_arrays,
projected_names).ValueOrDie();
+
ASSERT_TRUE(sequence_result->Equals(arrow::ChunkedArray(projected_expected)))
+ << sequence_result->ToString();
+ }
+ }
}
#endif
@@ -2137,12 +2209,19 @@ TEST_P(DedicatedVectorTableTest,
TestIndependentVectorRolling) {
arrow::field("embedding", vector_type),
arrow::field("tag", arrow::utf8())};
const std::map<std::string, std::string> options = {
- {Options::FILE_FORMAT, "parquet"}, {Options::VECTOR_FILE_FORMAT,
FileFormat()},
- {Options::FILE_SYSTEM, "local"}, {Options::BUCKET, "-1"},
- {Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
- {Options::WRITE_ONLY, "true"}, {Options::TARGET_FILE_SIZE,
"128 mb"},
- {Options::TARGET_FILE_ROW_NUM, "2500"},
{Options::VECTOR_TARGET_FILE_SIZE, "1 b"},
- {Options::WRITE_BATCH_SIZE, "100"}, {Options::READ_BATCH_SIZE,
"127"}};
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::VECTOR_FILE_FORMAT, FileFormat()},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
+ {Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::WRITE_ONLY, "true"},
+ {Options::TARGET_FILE_SIZE, "128 mb"},
+ {Options::TARGET_FILE_ROW_NUM, "2500"},
+ {Options::VECTOR_TARGET_FILE_SIZE, "1 b"},
+ {Options::WRITE_BATCH_SIZE, "100"},
+ {Options::READ_BATCH_SIZE, "127"},
+ {Options::DELETION_VECTORS_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
const std::string table_path = PathUtil::JoinPath(dir_->Str(),
"foo.db/bar");
auto input = PrepareBulkData(
@@ -2234,6 +2313,48 @@ TEST_P(DedicatedVectorTableTest,
TestIndependentVectorRolling) {
/*predicate=*/nullptr, /*row_ranges=*/{range}));
}
+ // Delete across main/VECTOR boundaries, including a whole VECTOR file and
the last row.
+ std::set<int64_t> deleted_ids = {0, 999, 1000, 2499, 2500, 2999, 3000,
4999, 6000};
+ if (FileFormat() != "lance") {
+ const auto& deleted_file = std::next(vector_files.begin())->second;
+ for (int64_t i = 0; i < deleted_file->row_count; ++i) {
+ deleted_ids.insert(deleted_file->first_row_id.value() + i);
+ }
+ }
+ std::map<std::string, std::vector<int64_t>> deleted_positions;
+ for (const auto& [first, file] : main_files) {
+ for (int64_t id : deleted_ids) {
+ if (id >= first && id < first + file->row_count) {
+ deleted_positions[file->file_name].push_back(id - first);
+ }
+ }
+ }
+ ASSERT_OK(CommitDeletionVectors(table_path, messages[0],
deleted_positions));
+ for (const std::vector<Range>& selection :
+ {std::vector<Range>{Range(0, 6000)},
+ std::vector<Range>{Range(998, 1002), Range(2498, 2502), Range(5998,
6000)}}) {
+ arrow::ArrayVector surviving_rows;
+ for (const auto& range : selection) {
+ int64_t next = range.from;
+ for (int64_t id : deleted_ids) {
+ if (id < next || id > range.to) {
+ continue;
+ }
+ if (next < id) {
+ surviving_rows.push_back(input->Slice(next, id - next));
+ }
+ next = id + 1;
+ }
+ if (next <= range.to) {
+ surviving_rows.push_back(input->Slice(next, range.to - next +
1));
+ }
+ }
+ auto expected = checked_pointer_cast<arrow::StructArray>(
+ arrow::Concatenate(surviving_rows).ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, expected,
+ /*predicate=*/nullptr, selection));
+ }
+
ASSERT_OK_AND_ASSIGN(messages, WriteArray(table_path, {}, input->Slice(0,
2)));
ASSERT_OK(Commit(table_path, messages));
auto appended = checked_pointer_cast<arrow::StructArray>(input->Slice(0,
2));
