From 1094b2d15e5f53981811581a18d0d0bb5f7fa75b Mon Sep 17 00:00:00 2001 From: zhangchaoming Date: Thu, 22 Oct 2026 11:59:47 +0800 Subject: [PATCH 1/4] perf(vector): reuse main readers across sequential vector files Generated-by: OpenAI Codex CLI 0.163.0-alpha.10.2 --- .../operation/data_evolution_split_read.cpp | 144 ++++++++++++- .../operation/data_evolution_split_read.h | 32 +++ .../data_evolution_split_read_test.cpp | 89 ++++++++ test/inte/data_evolution_table_test.cpp | 199 +++++++++++++++++- 4 files changed, 447 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 e3a52c6fe..10110c6dc 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -794,6 +794,65 @@ DataEvolutionSplitRead::PlanVectorReadRanges( return result; } +Result>> +DataEvolutionSplitRead::TryPlanSequentialVectorRead( + const std::vector>& files, + const std::function>(const std::shared_ptr&)>& + file_to_field_ids) { + std::vector> no_match; + std::shared_ptr main_file; + std::vector> 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::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 vector_ids, file_to_field_ids(first_vector)); + PAIMON_ASSIGN_OR_RAISE(std::vector 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::make_shared(main_file), + std::make_shared(std::move(vector_files), main_file->row_count)}; +} + Result> DataEvolutionSplitRead::CreateUnionReader( const BinaryRow& partition, const std::vector>& need_merge_files, const std::optional>& row_ranges, @@ -819,13 +878,16 @@ Result> DataEvolutionSplitRead::CreateUnionReader( } auto file_to_field_ids = [&](const std::shared_ptr& file) -> Result> { - 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 ids; for (const auto& name : file->write_cols.value()) { if (SpecialFields::IsSystemField(name)) { @@ -836,19 +898,35 @@ Result> DataEvolutionSplitRead::CreateUnionReader( } return ids; }; - PAIMON_ASSIGN_OR_RAISE( - std::vector 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 sources; - auto split_read = - std::make_shared(path_factory_, context_, pool_, executor_); std::optional> normalized_ranges; if (row_ranges) { normalized_ranges = Range::SortAndMergeOverlap(row_ranges.value(), /*adjacent=*/true); } + std::vector> sequential_bunches; + // The sequence-number provider can change at each boundary even for disjoint columns. + if (read_schema->GetFieldIndex(SpecialFields::SequenceNumber().Name()) == -1) { + PAIMON_ASSIGN_OR_RAISE(sequential_bunches, + TryPlanSequentialVectorRead(need_merge_files, file_to_field_ids)); + } + auto split_read = + std::make_shared(path_factory_, context_, pool_, executor_); + if (!sequential_bunches.empty()) { + std::vector 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> { + return split_read->CreateFieldUnionReader(partition, sequential_bunches, + normalized_ranges, data_file_path_factory, + group_dv_factory, group_dv, read_schema); + }}); + return std::make_unique(std::move(sources), arrow_pool_); + } + PAIMON_ASSIGN_OR_RAISE( + std::vector ranges, + PlanVectorReadRanges(need_merge_files, read_field_ids, file_to_field_ids)); + std::vector sources; for (const auto& range : ranges) { std::vector selected = {range.range}; if (normalized_ranges) { @@ -901,6 +979,17 @@ Result> 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> DataEvolutionSplitRead::CreateFieldUnionReader( + const BinaryRow& partition, const std::vector>& fields_files, + const std::optional>& row_ranges, + const std::shared_ptr& data_file_path_factory, + const DeletionVector::Factory& group_dv_factory, + const std::optional& group_dv, + const std::shared_ptr& 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()); @@ -967,7 +1056,40 @@ Result> 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(bunch); - if (blob_bunch && !blob_bunch->SequentialReadOptimize()) { + if (std::dynamic_pointer_cast(bunch)) { + std::vector sources; + auto split_read = std::make_shared(path_factory_, context_, + pool_, executor_); + for (const auto& file : bunch->Files()) { + std::optional> 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> { + PAIMON_ASSIGN_OR_RAISE( + std::vector> 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( + ObjectUtils::MoveVector>( + std::move(readers)), + split_read->arrow_pool_); + }}); + } + file_batch_readers[file_idx] = + std::make_unique(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 b51dd4475..e0406495c 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 #include #include +#include #include #include "paimon/common/data/blob_view_struct.h" @@ -123,6 +124,23 @@ class DataEvolutionSplitRead : public AbstractSplitRead { std::vector> data_files_; }; + /// Compatible VECTOR files tiling one main file's row range without gaps or overlaps. + class VectorBunch : public FieldBunch { + public: + VectorBunch(std::vector> files, int64_t row_count) + : files_(std::move(files)), row_count_(row_count) {} + int64_t RowCount() const override { + return row_count_; + } + const std::vector>& Files() const override { + return files_; + } + + private: + std::vector> 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>(const std::shared_ptr&)>& file_to_field_ids); + /// Empty means the layout needs range planning. Errors resolving providers still propagate. + static Result>> TryPlanSequentialVectorRead( + const std::vector>& files, + const std::function>(const std::shared_ptr&)>& + file_to_field_ids); + static Result>> SplitFieldBunches(const std::vector>& need_merge_files, const std::function(const std::shared_ptr&)>& @@ -278,6 +302,14 @@ class DataEvolutionSplitRead : public AbstractSplitRead { const std::optional& group_dv, const std::shared_ptr& read_schema) const; + Result> CreateFieldUnionReader( + const BinaryRow& partition, const std::vector>& fields_files, + const std::optional>& row_ranges, + const std::shared_ptr& data_file_path_factory, + const DeletionVector::Factory& group_dv_factory, + const std::optional& group_dv, + const std::shared_ptr& 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 0b113725f..f82049712 100644 --- a/src/paimon/core/operation/data_evolution_split_read_test.cpp +++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp @@ -751,6 +751,95 @@ 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{"embedding"}; + } + auto field_ids = + [main](const std::shared_ptr& file) -> Result> { + return file == main ? std::vector{0, 2} : std::vector{1}; + }; + ASSERT_OK_AND_ASSIGN(std::vector> bunches, + DataEvolutionSplitRead::TryPlanSequentialVectorRead( + {vector2, main, vector0, vector1}, field_ids)); + ASSERT_EQ(bunches.size(), 2); + ASSERT_EQ(bunches[0]->Files(), std::vector>{main}); + ASSERT_EQ(bunches[1]->Files(), + (std::vector>{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{"embedding"}; + auto field_ids = + [main](const std::shared_ptr& file) -> Result> { + return file == main ? std::vector{0, 2} : std::vector{1}; + }; + auto check_fallback = [&](const std::vector>& files) { + ASSERT_OK_AND_ASSIGN( + std::vector> 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{"another_embedding"}; + check_fallback({main, vector0, vector1}); + vector1->write_cols = vector0->write_cols = + std::vector{"embedding", "another_embedding"}; + check_fallback({main, vector0, vector1}); + + vector1->write_cols = vector0->write_cols = std::vector{"embedding"}; + auto inline_field_ids = + [main](const std::shared_ptr& file) -> Result> { + return file == main ? std::vector{0, 1, 2} : std::vector{1}; + }; + ASSERT_OK_AND_ASSIGN(std::vector> 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{"embedding"}; + auto failed_schema = [](const std::shared_ptr&) -> Result> { + return Status::IOError("schema unavailable"); + }; + auto result = + DataEvolutionSplitRead::TryPlanSequentialVectorRead({main, vector}, failed_schema); + ASSERT_FALSE(result.ok()); + ASSERT_TRUE(result.status().IsIOError()); + ASSERT_NE(result.status().ToString().find("schema unavailable"), std::string::npos); +} + 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 c508fd199..e9d788dfe 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -18,6 +18,8 @@ */ #include #include +#include +#include #include #include "arrow/type.h" @@ -37,6 +39,7 @@ #include "paimon/defs.h" #include "paimon/format/mosaic/mosaic_format_defs.h" #include "paimon/fs/file_system.h" +#include "paimon/fs/local/local_file_system.h" #include "paimon/global_index/bitmap_global_index_result.h" #include "paimon/global_index/bitmap_scored_global_index_result.h" #include "paimon/global_index/indexed_split.h" @@ -50,6 +53,36 @@ #include "paimon/testing/utils/test_helper.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { +namespace { +class CountingLocalFileSystem : public LocalFileSystem { + public: + explicit CountingLocalFileSystem(const std::string& failed_path = "") + : failed_path_(failed_path) {} + + Result> Open(const std::string& path) const override { + { + std::scoped_lock lock(mutex_); + ++open_counts_[path]; + } + if (path == failed_path_) { + return Status::IOError("injected open failure: " + path); + } + return LocalFileSystem::Open(path); + } + + int64_t OpenCount(const std::string& path) const { + std::scoped_lock lock(mutex_); + auto iter = open_counts_.find(path); + return iter == open_counts_.end() ? 0 : iter->second; + } + + private: + mutable std::mutex mutex_; + mutable std::map open_counts_; + std::string failed_path_; +}; +} // namespace + using DataEvolutionTableParam = std::tuple; // This is a sdk end-to-end test for data evolution @@ -1938,6 +1971,71 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { ASSERT_EQ(main_file->row_count, kRowCount); ASSERT_GT(vector_files.size(), 1); + // Count physical opens during a mixed projection, independently of manifest scanning. + auto counting_fs = std::make_shared(); + ReadContextBuilder counted_builder(table_path); + counted_builder.WithFileSystem(counting_fs) + .EnablePrefetch(false) + .SetReadFieldNames({"id", "embedding", "tag"}); + ASSERT_OK_AND_ASSIGN(std::unique_ptr counted_context, counted_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr counted_read, + TableRead::Create(std::move(counted_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr counted_reader, + counted_read->CreateReader(splits)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr counted_result, + ReadResultCollector::CollectResult(std::move(counted_reader))); + ASSERT_NE(counted_result, nullptr); + ASSERT_EQ(counted_result->length(), kRowCount); + auto counted_fields = counted_result->Flatten().ValueOrDie(); + ASSERT_EQ(counted_fields.size(), 4); + for (int32_t i = 0; i < 3; ++i) { + ASSERT_TRUE(counted_fields[i + 1]->Equals(arrow::ChunkedArray(expected->field(i)))); + } + auto counted_split = std::dynamic_pointer_cast(splits[0]); + ASSERT_EQ(counting_fs->OpenCount( + PathUtil::JoinPath(counted_split->BucketPath(), main_file->file_name)), + 1); + + // Later files stay unopened on early Close, and an open failure stops at that file. + const std::string main_path = + PathUtil::JoinPath(counted_split->BucketPath(), main_file->file_name); + const std::string failed_path = PathUtil::JoinPath( + counted_split->BucketPath(), std::next(vector_files.begin())->second->file_name); + for (bool fail_next_file : {false, true}) { + auto lazy_fs = std::make_shared(fail_next_file ? failed_path : ""); + ReadContextBuilder lazy_builder(table_path); + lazy_builder.WithFileSystem(lazy_fs).EnablePrefetch(false).SetReadFieldNames( + {"id", "embedding", "tag"}); + ASSERT_OK_AND_ASSIGN(std::unique_ptr lazy_context, lazy_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr lazy_read, + TableRead::Create(std::move(lazy_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr lazy_reader, + lazy_read->CreateReader(splits)); + lazy_read.reset(); + ASSERT_EQ(lazy_fs->OpenCount(main_path), 0); + for (const auto& [first, file] : vector_files) { + ASSERT_EQ(lazy_fs->OpenCount( + PathUtil::JoinPath(counted_split->BucketPath(), file->file_name)), + 0); + } + ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch first_batch, lazy_reader->NextBatch()); + ASSERT_FALSE(BatchReader::IsEofBatch(first_batch)); + if (fail_next_file) { + auto result = ReadResultCollector::CollectResult(std::move(lazy_reader)); + ASSERT_FALSE(result.ok()); + ASSERT_TRUE(result.status().IsIOError()); + ASSERT_NE(result.status().ToString().find(failed_path), std::string::npos); + } else { + lazy_reader->Close(); + } + ASSERT_EQ(lazy_fs->OpenCount(main_path), 1); + for (const auto& [first, file] : vector_files) { + std::string path = PathUtil::JoinPath(counted_split->BucketPath(), file->file_name); + int64_t expected_opens = first == 0 || (fail_next_file && path == failed_path) ? 1 : 0; + ASSERT_EQ(lazy_fs->OpenCount(path), expected_opens); + } + } + int64_t next_row_id = 0; std::vector selected_ranges = {Range(0, 0)}; for (const auto& [first_row_id, file] : vector_files) { @@ -2020,6 +2118,46 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected_rows))) << "actual=" << result->ToString() << "\nexpected=" << expected_rows->ToString(); } + + // Reading sequence numbers must retain the newest provider independently in each range. + auto sequence_main = std::make_shared(*main_file); + sequence_main->AssignSequenceNumber(3, 3); + std::vector> sequence_files = {sequence_main}; + int64_t sequence = 2; + for (const auto& [first, file] : vector_files) { + auto sequence_vector = std::make_shared(*file); + sequence_vector->AssignSequenceNumber(sequence, sequence); + sequence_files.push_back(sequence_vector); + ++sequence; + } + DataSplitImpl::Builder sequence_builder(original_split->Partition(), original_split->Bucket(), + original_split->BucketPath(), + std::move(sequence_files)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr sequence_split, + sequence_builder.RawConvertible(false).Build()); + auto tracking = PrepareBulkData( + kRowCount, + [](int32_t id) { return fmt::format("0, {}, {}", id, std::max(3, 2 + id / 1000)); }, + {SpecialFields::ValueKind().ArrowField(), SpecialFields::RowId().ArrowField(), + SpecialFields::SequenceNumber().ArrowField()}); + auto sequence_expected = + arrow::StructArray::Make({tracking->field(0), expected->field(0), expected->field(1), + tracking->field(1), tracking->field(2)}, + {"_VALUE_KIND", "id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}) + .ValueOrDie(); + ReadContextBuilder sequence_read_builder(table_path); + sequence_read_builder.SetReadFieldNames({"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}); + ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_context, + sequence_read_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_read, + TableRead::Create(std::move(sequence_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_reader, + sequence_read->CreateReader(sequence_split)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr sequence_result, + ReadResultCollector::CollectResult(std::move(sequence_reader))); + ASSERT_NE(sequence_result, nullptr); + ASSERT_TRUE(sequence_result->Equals(arrow::ChunkedArray(sequence_expected))) + << sequence_result->ToString(); } #endif @@ -2130,12 +2268,19 @@ TEST_P(DedicatedVectorTableTest, TestIndependentVectorRolling) { arrow::field("embedding", vector_type), arrow::field("tag", arrow::utf8())}; const std::map 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( @@ -2227,6 +2372,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 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> 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& selection : + {std::vector{Range(0, 6000)}, + std::vector{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::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(input->Slice(0, 2)); From b40b91c55e7a4c779a57fa2e74da2b599a9da799 Mon Sep 17 00:00:00 2001 From: zhangchaoming Date: Thu, 22 Oct 2026 13:52:43 +0800 Subject: [PATCH 2/4] fix(vector): align sequence number providers with Java --- .../operation/data_evolution_split_read.cpp | 17 ++-- test/inte/data_evolution_table_test.cpp | 79 ++++++++++++++----- 2 files changed, 69 insertions(+), 27 deletions(-) diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 10110c6dc..7b9042409 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -902,12 +902,8 @@ Result> DataEvolutionSplitRead::CreateUnionReader( if (row_ranges) { normalized_ranges = Range::SortAndMergeOverlap(row_ranges.value(), /*adjacent=*/true); } - std::vector> sequential_bunches; - // The sequence-number provider can change at each boundary even for disjoint columns. - if (read_schema->GetFieldIndex(SpecialFields::SequenceNumber().Name()) == -1) { - PAIMON_ASSIGN_OR_RAISE(sequential_bunches, - TryPlanSequentialVectorRead(need_merge_files, file_to_field_ids)); - } + PAIMON_ASSIGN_OR_RAISE(std::vector> sequential_bunches, + TryPlanSequentialVectorRead(need_merge_files, file_to_field_ids)); auto split_read = std::make_shared(path_factory_, context_, pool_, executor_); if (!sequential_bunches.empty()) { @@ -1018,6 +1014,10 @@ Result> DataEvolutionSplitRead::CreateFieldUnionRea std::vector reader_offsets(all_read_fields.size(), -1); // which field index in the reading row std::vector 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(fields_files.size()); file_idx++) { const auto& bunch = fields_files[file_idx]; @@ -1038,6 +1038,11 @@ Result> 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 diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index e9d788dfe..b645747ba 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -2119,7 +2119,7 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { << "actual=" << result->ToString() << "\nexpected=" << expected_rows->ToString(); } - // Reading sequence numbers must retain the newest provider independently in each range. + // Like Java, the normal file supplies sequence numbers even when VECTOR files are newer. auto sequence_main = std::make_shared(*main_file); sequence_main->AssignSequenceNumber(3, 3); std::vector> sequence_files = {sequence_main}; @@ -2130,34 +2130,71 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { 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 sequence_split, sequence_builder.RawConvertible(false).Build()); auto tracking = PrepareBulkData( - kRowCount, - [](int32_t id) { return fmt::format("0, {}, {}", id, std::max(3, 2 + id / 1000)); }, + kRowCount, [](int32_t id) { return fmt::format("0, {}, 3", id); }, {SpecialFields::ValueKind().ArrowField(), SpecialFields::RowId().ArrowField(), SpecialFields::SequenceNumber().ArrowField()}); - auto sequence_expected = - arrow::StructArray::Make({tracking->field(0), expected->field(0), expected->field(1), - tracking->field(1), tracking->field(2)}, - {"_VALUE_KIND", "id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}) - .ValueOrDie(); - ReadContextBuilder sequence_read_builder(table_path); - sequence_read_builder.SetReadFieldNames({"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}); - ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_context, - sequence_read_builder.Finish()); - ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_read, - TableRead::Create(std::move(sequence_context))); - ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_reader, - sequence_read->CreateReader(sequence_split)); - ASSERT_OK_AND_ASSIGN(std::shared_ptr sequence_result, - ReadResultCollector::CollectResult(std::move(sequence_reader))); - ASSERT_NE(sequence_result, nullptr); - ASSERT_TRUE(sequence_result->Equals(arrow::ChunkedArray(sequence_expected))) - << sequence_result->ToString(); + for (bool sequential : {true, false}) { + SCOPED_TRACE(sequential); + std::shared_ptr split = sequence_split; + std::shared_ptr 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>{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& projection : + {std::vector{"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}, + std::vector{"id", "_ROW_ID", "_SEQUENCE_NUMBER"}, + std::vector{"_ROW_ID", "_SEQUENCE_NUMBER"}}) { + auto sequence_fs = std::make_shared(); + ReadContextBuilder sequence_read_builder(table_path); + sequence_read_builder.WithFileSystem(sequence_fs) + .EnablePrefetch(false) + .SetReadFieldNames(projection); + ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_context, + sequence_read_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_read, + TableRead::Create(std::move(sequence_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_reader, + sequence_read->CreateReader(split)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr sequence_result, + ReadResultCollector::CollectResult(std::move(sequence_reader))); + ASSERT_NE(sequence_result, nullptr); + arrow::ArrayVector projected_arrays = {tracking->field(0)}; + std::vector 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(); + if (sequential) { + ASSERT_EQ(sequence_fs->OpenCount(main_path), 1); + } + } + } } #endif From 4b82f8fdb4b22eb629b2455b5889776505709b9d Mon Sep 17 00:00:00 2001 From: zhangchaoming Date: Thu, 22 Oct 2026 15:00:01 +0800 Subject: [PATCH 3/4] test(vector): release retained batch in lazy reader regression --- test/inte/data_evolution_table_test.cpp | 3 +++ 1 file changed, 3 insertions(+) diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index b645747ba..2d604d4c1 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -27,6 +27,7 @@ #include "gtest/gtest.h" #include "paimon/common/factories/io_hook.h" #include "paimon/common/io/cache/lru_cache.h" +#include "paimon/common/reader/reader_utils.h" #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/checked_cast.h" @@ -2019,6 +2020,8 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { 0); } ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch first_batch, lazy_reader->NextBatch()); + ScopeGuard release_first_batch( + [&first_batch]() { ReaderUtils::ReleaseReadBatch(std::move(first_batch)); }); ASSERT_FALSE(BatchReader::IsEofBatch(first_batch)); if (fail_next_file) { auto result = ReadResultCollector::CollectResult(std::move(lazy_reader)); From 402673f8019b0f275c39f1be6a1c05262196214e Mon Sep 17 00:00:00 2001 From: zhangchaoming Date: Fri, 23 Oct 2026 16:20:00 +0800 Subject: [PATCH 4/4] test(vector): address sequential reader review feedback --- .../data_evolution_split_read_test.cpp | 146 +++++++++++++++++- test/inte/data_evolution_table_test.cpp | 108 +------------ 2 files changed, 145 insertions(+), 109 deletions(-) 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 f82049712..4bd0e0c29 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 +#include #include #include #include #include +#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> 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 open_count_{0}; +}; +} // namespace + class DataEvolutionSplitReadTest : public ::testing::Test { public: std::shared_ptr CreateDataFileMeta( @@ -835,9 +869,117 @@ TEST_F(DataEvolutionSplitReadTest, TestSequentialVectorReadPropagatesMetadataErr }; auto result = DataEvolutionSplitRead::TryPlanSequentialVectorRead({main, vector}, failed_schema); - ASSERT_FALSE(result.ok()); + ASSERT_NOK_WITH_MSG(result, "schema unavailable"); ASSERT_TRUE(result.status().IsIOError()); - ASSERT_NE(result.status().ToString().find("schema unavailable"), std::string::npos); +} + +TEST_F(DataEvolutionSplitReadTest, TestSequentialVectorReadReusesMainReader) { + auto dir = UniqueTestDirectory::Create("local"); + auto fs = std::make_shared(); + 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 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 table_schema, + TableSchema::Create(0, schema, {}, {}, options)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr 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 format, + FileFormatFactory::Get("parquet", options)); + auto write_file = [&](const std::string& file_name, int64_t first_row_id, + const std::shared_ptr& array, + int64_t sequence) -> Result> { + ArrowSchema c_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportType(*array->type(), &c_schema)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr builder, + format->CreateWriterBuilder(&c_schema, 2)); + const std::string path = PathUtil::JoinPath(bucket_path, file_name); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr out, fs->Create(path, false)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr 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 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 main, + write_file("main.parquet", 0, main_data, 3)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr vector0, + write_file("vector0.vector.parquet", 0, vector0_data, 4)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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::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, CreateDefaultExecutor(2)); + for (const std::vector& projection : + {std::vector{"id", "embedding"}, + std::vector{"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}, + std::vector{"id", "_ROW_ID", "_SEQUENCE_NUMBER"}, + std::vector{"_ROW_ID", "_SEQUENCE_NUMBER"}}) { + auto counting_fs = std::make_shared( + 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 read_context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr context, + InternalReadContext::Create(read_context, table_schema, options)); + auto split_read = std::make_unique(path_factory, context, + GetDefaultPool(), executor); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, split_read->CreateReader(split)); + split_read.reset(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr result, + ReadResultCollector::CollectResult(std::move(reader))); + ASSERT_NE(result, nullptr); + arrow::ArrayVector columns = {expected->field(0)}; + std::vector 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) { diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index 284a34b25..395de88d0 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -18,7 +18,6 @@ */ #include #include -#include #include #include @@ -27,7 +26,6 @@ #include "gtest/gtest.h" #include "paimon/common/factories/io_hook.h" #include "paimon/common/io/cache/lru_cache.h" -#include "paimon/common/reader/reader_utils.h" #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/checked_cast.h" @@ -47,7 +45,6 @@ #include "paimon/executor.h" #include "paimon/format/mosaic/mosaic_format_defs.h" #include "paimon/fs/file_system.h" -#include "paimon/fs/local/local_file_system.h" #include "paimon/global_index/bitmap_global_index_result.h" #include "paimon/global_index/bitmap_scored_global_index_result.h" #include "paimon/global_index/indexed_split.h" @@ -61,36 +58,6 @@ #include "paimon/testing/utils/test_helper.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { -namespace { -class CountingLocalFileSystem : public LocalFileSystem { - public: - explicit CountingLocalFileSystem(const std::string& failed_path = "") - : failed_path_(failed_path) {} - - Result> Open(const std::string& path) const override { - { - std::scoped_lock lock(mutex_); - ++open_counts_[path]; - } - if (path == failed_path_) { - return Status::IOError("injected open failure: " + path); - } - return LocalFileSystem::Open(path); - } - - int64_t OpenCount(const std::string& path) const { - std::scoped_lock lock(mutex_); - auto iter = open_counts_.find(path); - return iter == open_counts_.end() ? 0 : iter->second; - } - - private: - mutable std::mutex mutex_; - mutable std::map open_counts_; - std::string failed_path_; -}; -} // namespace - using DataEvolutionTableParam = std::tuple; // This is a sdk end-to-end test for data evolution @@ -1979,73 +1946,6 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { ASSERT_EQ(main_file->row_count, kRowCount); ASSERT_GT(vector_files.size(), 1); - // Count physical opens during a mixed projection, independently of manifest scanning. - auto counting_fs = std::make_shared(); - ReadContextBuilder counted_builder(table_path); - counted_builder.WithFileSystem(counting_fs) - .EnablePrefetch(false) - .SetReadFieldNames({"id", "embedding", "tag"}); - ASSERT_OK_AND_ASSIGN(std::unique_ptr counted_context, counted_builder.Finish()); - ASSERT_OK_AND_ASSIGN(std::unique_ptr counted_read, - TableRead::Create(std::move(counted_context))); - ASSERT_OK_AND_ASSIGN(std::unique_ptr counted_reader, - counted_read->CreateReader(splits)); - ASSERT_OK_AND_ASSIGN(std::shared_ptr counted_result, - ReadResultCollector::CollectResult(std::move(counted_reader))); - ASSERT_NE(counted_result, nullptr); - ASSERT_EQ(counted_result->length(), kRowCount); - auto counted_fields = counted_result->Flatten().ValueOrDie(); - ASSERT_EQ(counted_fields.size(), 4); - for (int32_t i = 0; i < 3; ++i) { - ASSERT_TRUE(counted_fields[i + 1]->Equals(arrow::ChunkedArray(expected->field(i)))); - } - auto counted_split = std::dynamic_pointer_cast(splits[0]); - ASSERT_EQ(counting_fs->OpenCount( - PathUtil::JoinPath(counted_split->BucketPath(), main_file->file_name)), - 1); - - // Later files stay unopened on early Close, and an open failure stops at that file. - const std::string main_path = - PathUtil::JoinPath(counted_split->BucketPath(), main_file->file_name); - const std::string failed_path = PathUtil::JoinPath( - counted_split->BucketPath(), std::next(vector_files.begin())->second->file_name); - for (bool fail_next_file : {false, true}) { - auto lazy_fs = std::make_shared(fail_next_file ? failed_path : ""); - ReadContextBuilder lazy_builder(table_path); - lazy_builder.WithFileSystem(lazy_fs).EnablePrefetch(false).SetReadFieldNames( - {"id", "embedding", "tag"}); - ASSERT_OK_AND_ASSIGN(std::unique_ptr lazy_context, lazy_builder.Finish()); - ASSERT_OK_AND_ASSIGN(std::unique_ptr lazy_read, - TableRead::Create(std::move(lazy_context))); - ASSERT_OK_AND_ASSIGN(std::unique_ptr lazy_reader, - lazy_read->CreateReader(splits)); - lazy_read.reset(); - ASSERT_EQ(lazy_fs->OpenCount(main_path), 0); - for (const auto& [first, file] : vector_files) { - ASSERT_EQ(lazy_fs->OpenCount( - PathUtil::JoinPath(counted_split->BucketPath(), file->file_name)), - 0); - } - ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch first_batch, lazy_reader->NextBatch()); - ScopeGuard release_first_batch( - [&first_batch]() { ReaderUtils::ReleaseReadBatch(std::move(first_batch)); }); - ASSERT_FALSE(BatchReader::IsEofBatch(first_batch)); - if (fail_next_file) { - auto result = ReadResultCollector::CollectResult(std::move(lazy_reader)); - ASSERT_FALSE(result.ok()); - ASSERT_TRUE(result.status().IsIOError()); - ASSERT_NE(result.status().ToString().find(failed_path), std::string::npos); - } else { - lazy_reader->Close(); - } - ASSERT_EQ(lazy_fs->OpenCount(main_path), 1); - for (const auto& [first, file] : vector_files) { - std::string path = PathUtil::JoinPath(counted_split->BucketPath(), file->file_name); - int64_t expected_opens = first == 0 || (fail_next_file && path == failed_path) ? 1 : 0; - ASSERT_EQ(lazy_fs->OpenCount(path), expected_opens); - } - } - int64_t next_row_id = 0; std::vector selected_ranges = {Range(0, 0)}; for (const auto& [first_row_id, file] : vector_files) { @@ -2176,11 +2076,8 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { {std::vector{"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"}, std::vector{"id", "_ROW_ID", "_SEQUENCE_NUMBER"}, std::vector{"_ROW_ID", "_SEQUENCE_NUMBER"}}) { - auto sequence_fs = std::make_shared(); ReadContextBuilder sequence_read_builder(table_path); - sequence_read_builder.WithFileSystem(sequence_fs) - .EnablePrefetch(false) - .SetReadFieldNames(projection); + sequence_read_builder.SetReadFieldNames(projection); ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_context, sequence_read_builder.Finish()); ASSERT_OK_AND_ASSIGN(std::unique_ptr sequence_read, @@ -2200,9 +2097,6 @@ TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) { arrow::StructArray::Make(projected_arrays, projected_names).ValueOrDie(); ASSERT_TRUE(sequence_result->Equals(arrow::ChunkedArray(projected_expected))) << sequence_result->ToString(); - if (sequential) { - ASSERT_EQ(sequence_fs->OpenCount(main_path), 1); - } } } }