diff --git a/include/paimon/utils/roaring_bitmap32.h b/include/paimon/utils/roaring_bitmap32.h index 29d458097..85f6a87e7 100644 --- a/include/paimon/utils/roaring_bitmap32.h +++ b/include/paimon/utils/roaring_bitmap32.h @@ -65,6 +65,8 @@ class PAIMON_EXPORT RoaringBitmap32 { bool operator!=(const Iterator& other) const; private: + Iterator() = default; + // Null represents the end sentinel without an underlying iterator allocation. void* iterator_ = nullptr; }; @@ -136,6 +138,7 @@ class PAIMON_EXPORT RoaringBitmap32 { std::string ToString() const; Iterator Begin() const; + /// Returns an end sentinel without allocating an underlying iterator. Iterator End() const; /// @return the iterator moved to the value which is equal or larger than key Iterator EqualOrLarger(int32_t key) const; diff --git a/include/paimon/utils/roaring_bitmap64.h b/include/paimon/utils/roaring_bitmap64.h index 289622b0a..931f1b528 100644 --- a/include/paimon/utils/roaring_bitmap64.h +++ b/include/paimon/utils/roaring_bitmap64.h @@ -65,9 +65,12 @@ class PAIMON_EXPORT RoaringBitmap64 { bool operator==(const Iterator& other) const; bool operator!=(const Iterator& other) const; /// Move the iterator to the value which is equal or larger than input value + /// An End() sentinel remains at the end. void EqualOrLarger(int64_t value); private: + Iterator() = default; + // Null represents the end sentinel without an underlying iterator allocation. void* iterator_ = nullptr; }; @@ -149,6 +152,7 @@ class PAIMON_EXPORT RoaringBitmap64 { std::string ToString() const; Iterator Begin() const; + /// Returns an end sentinel without allocating an underlying iterator. Iterator End() const; /// @return the iterator moved to the value which is equal or larger than key Iterator EqualOrLarger(int64_t key) const; diff --git a/src/paimon/common/reader/blob_fallback_batch_reader.cpp b/src/paimon/common/reader/blob_fallback_batch_reader.cpp index 830b0df4b..a5e238557 100644 --- a/src/paimon/common/reader/blob_fallback_batch_reader.cpp +++ b/src/paimon/common/reader/blob_fallback_batch_reader.cpp @@ -42,10 +42,8 @@ Result> BlobFallbackBatchReader::Create std::vector>&& sequence_groups, const std::shared_ptr& read_schema, int32_t read_batch_size, const std::shared_ptr& arrow_pool) { - if (sequence_groups.size() < 2) { - return Status::Invalid( - "Blob fallback needs at least two sequence groups; a single group should be read " - "sequentially."); + if (sequence_groups.empty()) { + return Status::Invalid("Blob fallback needs at least one sequence group."); } if (read_schema == nullptr) { return Status::Invalid("Blob fallback read schema cannot be nullptr."); @@ -339,6 +337,9 @@ Result BlobFallbackBatchReader::NextBatch() { return Status::Invalid("blob fallback batch reader is closed"); } std::vector> group_chunks(groups_.size()); + // TODO(xinyu.lxy): Older layers are materialized even when newer values cover them, adding + // unnecessary reads and potentially exposing stale payload errors. Add Java-like read/skip + // support to materialize older values only when fallback is needed. // the first (newest) group defines the window; the others must step in lockstep PAIMON_ASSIGN_OR_RAISE(int64_t row_count, FillWindow(0, read_batch_size_, &group_chunks[0])); for (size_t g = 1; g < groups_.size(); g++) { diff --git a/src/paimon/common/reader/blob_fallback_batch_reader.h b/src/paimon/common/reader/blob_fallback_batch_reader.h index e15667b4a..36bceb990 100644 --- a/src/paimon/common/reader/blob_fallback_batch_reader.h +++ b/src/paimon/common/reader/blob_fallback_batch_reader.h @@ -73,9 +73,9 @@ class BlobFallbackBatchReader : public BatchReader { }; /// `sequence_groups` must be ordered by descending max sequence number and contain at least - /// two groups (a single group needs no fallback). `read_schema` is the schema every file - /// reader emits; it must contain exactly one blob field and may additionally contain the - /// row-tracking fields _ROW_ID and _SEQUENCE_NUMBER (completed per file by + /// one group (a single group can have gaps to fill with NULLs). `read_schema` is the schema + /// every file reader emits; it must contain exactly one blob field and may additionally contain + /// the row-tracking fields _ROW_ID and _SEQUENCE_NUMBER (completed per file by /// CompleteRowTrackingFieldsBatchReader), which stay correct for rows that are a /// placeholder in every layer. static Result> Create( diff --git a/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp b/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp index 7eac84b4a..9f67737de 100644 --- a/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp +++ b/src/paimon/common/reader/blob_fallback_batch_reader_test.cpp @@ -145,6 +145,11 @@ TEST_F(BlobFallbackBatchReaderTest, TestBasicFallback) { {"b0", "u1", "b2"}); } +TEST_F(BlobFallbackBatchReaderTest, TestSingleLayerFillsUnwrittenRows) { + CheckFallback({{SegmentSpec::Gap(2), SegmentSpec::File({"b2", "PH"}), SegmentSpec::Gap(2)}}, + {std::nullopt, std::nullopt, "b2", std::nullopt, std::nullopt, std::nullopt}); +} + TEST_F(BlobFallbackBatchReaderTest, TestGapPadding) { // the newer layer only covers rows 2-3; the gaps stand for placeholders CheckFallback({{SegmentSpec::Gap(2), SegmentSpec::File({"u2", "PH"})}, @@ -325,12 +330,10 @@ TEST_F(BlobFallbackBatchReaderTest, TestMisalignedGroupsFail) { } TEST_F(BlobFallbackBatchReaderTest, TestCreateValidation) { - // a single group needs no fallback - std::vector> single_group; - single_group.push_back(MakeGroup({SegmentSpec::File({"b0"})}, 1024)); - ASSERT_NOK_WITH_MSG(BlobFallbackBatchReader::Create(std::move(single_group), read_schema_, 1024, + std::vector> empty_groups; + ASSERT_NOK_WITH_MSG(BlobFallbackBatchReader::Create(std::move(empty_groups), read_schema_, 1024, GetArrowPool(pool_)), - "at least two sequence groups"); + "at least one sequence group"); // the read schema must contain a blob field std::vector> groups; diff --git a/src/paimon/common/utils/roaring_bitmap32.cpp b/src/paimon/common/utils/roaring_bitmap32.cpp index 8a90952ab..7644c550a 100644 --- a/src/paimon/common/utils/roaring_bitmap32.cpp +++ b/src/paimon/common/utils/roaring_bitmap32.cpp @@ -56,6 +56,11 @@ RoaringBitmap32::Iterator& RoaringBitmap32::Iterator::operator=( if (&other == this) { return *this; } + if (!other.iterator_) { + delete static_cast(iterator_); + iterator_ = nullptr; + return *this; + } if (!iterator_) { iterator_ = new roaring::Roaring::const_iterator(GetIterator(other.iterator_)); } else { @@ -92,7 +97,11 @@ bool RoaringBitmap32::Iterator::operator==(const Iterator& other) const { if (&other == this) { return true; } - assert(iterator_ && other.iterator_); + bool is_end = !iterator_ || !GetIterator(iterator_).i.has_value; + bool other_is_end = !other.iterator_ || !GetIterator(other.iterator_).i.has_value; + if (is_end || other_is_end) { + return is_end == other_is_end; + } return GetIterator(iterator_) == GetIterator(other.iterator_); } bool RoaringBitmap32::Iterator::operator!=(const Iterator& other) const { @@ -301,9 +310,7 @@ RoaringBitmap32::Iterator RoaringBitmap32::Begin() const { } RoaringBitmap32::Iterator RoaringBitmap32::End() const { - RoaringBitmap32::Iterator iter(*this); - GetIterator(iter.iterator_) = GetRoaringBitmap(roaring_bitmap_).end(); - return iter; + return Iterator(); } RoaringBitmap32::Iterator RoaringBitmap32::EqualOrLarger(int32_t key) const { diff --git a/src/paimon/common/utils/roaring_bitmap32_test.cpp b/src/paimon/common/utils/roaring_bitmap32_test.cpp index 239381f22..b4334a9c7 100644 --- a/src/paimon/common/utils/roaring_bitmap32_test.cpp +++ b/src/paimon/common/utils/roaring_bitmap32_test.cpp @@ -256,6 +256,21 @@ TEST(RoaringBitmap32Test, TestIterator) { ASSERT_EQ(iter, roaring.End()); } +TEST(RoaringBitmap32Test, TestEndIterator) { + RoaringBitmap32 empty; + ASSERT_EQ(empty.Begin(), empty.End()); + ASSERT_EQ(empty.EqualOrLarger(0), empty.End()); + + auto bitmap = RoaringBitmap32::From({0, 10, 20}); + auto iter = bitmap.Begin(); + std::vector values; + for (; iter != bitmap.End(); ++iter) { + values.push_back(*iter); + } + ASSERT_EQ(values, (std::vector{0, 10, 20})); + ASSERT_EQ(iter, bitmap.End()); +} + TEST(RoaringBitmap32Test, TestNextAndPreviousValue) { RoaringBitmap32 roaring = RoaringBitmap32::From({10, 20, 30}); @@ -307,6 +322,18 @@ TEST(RoaringBitmap32Test, TestIteratorAssignAndMove) { iter3 = std::move(iter2); ASSERT_EQ(10, *iter3); ASSERT_FALSE(iter2.iterator_); // NOLINT(bugprone-use-after-move) + + iter3 = roaring.End(); + auto copied_end = iter3; + auto moved_end = std::move(copied_end); + ASSERT_EQ(moved_end, roaring.End()); + iter3 = roaring.Begin(); + iter3 = moved_end; + ASSERT_EQ(iter3, roaring.End()); + iter3 = roaring.Begin(); + ASSERT_EQ(*iter3, 10); + iter3 = std::move(moved_end); + ASSERT_EQ(iter3, roaring.End()); } TEST(RoaringBitmap32Test, TestDeserializeFromInputStream) { diff --git a/src/paimon/common/utils/roaring_bitmap64.cpp b/src/paimon/common/utils/roaring_bitmap64.cpp index 510ef4e8d..0bd4f8c1e 100644 --- a/src/paimon/common/utils/roaring_bitmap64.cpp +++ b/src/paimon/common/utils/roaring_bitmap64.cpp @@ -30,8 +30,19 @@ namespace paimon { namespace { -roaring::Roaring64Map::const_iterator& GetIterator(void* iter) { - return *(static_cast(iter)); +// Expose exhaustion without constructing another native iterator or modifying CRoaring. +class Bitmap64Iterator : public roaring::Roaring64Map::const_iterator { + public: + explicit Bitmap64Iterator(const roaring::Roaring64Map& bitmap) + : roaring::Roaring64Map::const_iterator(bitmap) {} + + bool IsEnd() const { + return map_iter == map_end; + } +}; + +Bitmap64Iterator& GetIterator(void* iter) { + return *(static_cast(iter)); } roaring::Roaring64Map& GetRoaringBitmap(void* bitmap) { return *(static_cast(bitmap)); @@ -39,13 +50,12 @@ roaring::Roaring64Map& GetRoaringBitmap(void* bitmap) { } // namespace RoaringBitmap64::Iterator::Iterator(const RoaringBitmap64& bitmap) { - iterator_ = - new roaring::Roaring64Map::const_iterator(GetRoaringBitmap(bitmap.roaring_bitmap_).begin()); + iterator_ = new Bitmap64Iterator(GetRoaringBitmap(bitmap.roaring_bitmap_)); } RoaringBitmap64::Iterator::~Iterator() { if (iterator_) { - delete static_cast(iterator_); + delete static_cast(iterator_); } } @@ -57,8 +67,13 @@ RoaringBitmap64::Iterator& RoaringBitmap64::Iterator::operator=( if (&other == this) { return *this; } + if (!other.iterator_) { + delete static_cast(iterator_); + iterator_ = nullptr; + return *this; + } if (!iterator_) { - iterator_ = new roaring::Roaring64Map::const_iterator(GetIterator(other.iterator_)); + iterator_ = new Bitmap64Iterator(GetIterator(other.iterator_)); } else { GetIterator(iterator_) = GetIterator(other.iterator_); } @@ -75,7 +90,7 @@ RoaringBitmap64::Iterator& RoaringBitmap64::Iterator::operator=( return *this; } if (iterator_) { - delete static_cast(iterator_); + delete static_cast(iterator_); } iterator_ = other.iterator_; other.iterator_ = nullptr; @@ -93,7 +108,11 @@ bool RoaringBitmap64::Iterator::operator==(const Iterator& other) const { if (&other == this) { return true; } - assert(iterator_ && other.iterator_); + bool is_end = !iterator_ || GetIterator(iterator_).IsEnd(); + bool other_is_end = !other.iterator_ || GetIterator(other.iterator_).IsEnd(); + if (is_end || other_is_end) { + return is_end == other_is_end; + } return GetIterator(iterator_) == GetIterator(other.iterator_); } bool RoaringBitmap64::Iterator::operator!=(const Iterator& other) const { @@ -101,6 +120,9 @@ bool RoaringBitmap64::Iterator::operator!=(const Iterator& other) const { } void RoaringBitmap64::Iterator::EqualOrLarger(int64_t value) { + if (!iterator_) { + return; + } [[maybe_unused]] bool _ = GetIterator(iterator_).move(value); } @@ -372,17 +394,12 @@ RoaringBitmap64::Iterator RoaringBitmap64::Begin() const { } RoaringBitmap64::Iterator RoaringBitmap64::End() const { - RoaringBitmap64::Iterator iter(*this); - GetIterator(iter.iterator_) = GetRoaringBitmap(roaring_bitmap_).end(); - return iter; + return Iterator(); } RoaringBitmap64::Iterator RoaringBitmap64::EqualOrLarger(int64_t key) const { RoaringBitmap64::Iterator iter(*this); - bool not_end = GetIterator(iter.iterator_).move(key); - if (!not_end) { - return End(); - } + iter.EqualOrLarger(key); return iter; } diff --git a/src/paimon/common/utils/roaring_bitmap64_test.cpp b/src/paimon/common/utils/roaring_bitmap64_test.cpp index 6a51d29ac..e77aba039 100644 --- a/src/paimon/common/utils/roaring_bitmap64_test.cpp +++ b/src/paimon/common/utils/roaring_bitmap64_test.cpp @@ -275,6 +275,22 @@ TEST(RoaringBitmap64Test, TestIterator) { ASSERT_EQ(iter, roaring.End()); } +TEST(RoaringBitmap64Test, TestEndIterator) { + RoaringBitmap64 empty; + ASSERT_EQ(empty.Begin(), empty.End()); + ASSERT_EQ(empty.EqualOrLarger(0), empty.End()); + + const std::vector expected = {0, 1LL << 32, RoaringBitmap64::MAX_VALUE}; + auto bitmap = RoaringBitmap64::From(expected); + auto iter = bitmap.Begin(); + std::vector values; + for (; iter != bitmap.End(); ++iter) { + values.push_back(*iter); + } + ASSERT_EQ(values, expected); + ASSERT_EQ(iter, bitmap.End()); +} + TEST(RoaringBitmap64Test, TestIteratorAssignAndMove) { RoaringBitmap64 roaring = RoaringBitmap64::From({10000000000010l, 100000000000100l, 100000000000200l}); @@ -308,6 +324,18 @@ TEST(RoaringBitmap64Test, TestIteratorAssignAndMove) { iter3 = std::move(iter2); ASSERT_EQ(10000000000010l, *iter3); ASSERT_FALSE(iter2.iterator_); // NOLINT(bugprone-use-after-move) + + iter3 = roaring.End(); + auto copied_end = iter3; + auto moved_end = std::move(copied_end); + ASSERT_EQ(moved_end, roaring.End()); + iter3 = roaring.Begin(); + iter3 = moved_end; + ASSERT_EQ(iter3, roaring.End()); + iter3 = roaring.Begin(); + ASSERT_EQ(*iter3, 10000000000010l); + iter3 = std::move(moved_end); + ASSERT_EQ(iter3, roaring.End()); } TEST(RoaringBitmap64Test, TestDeserializeFromInputStream) { @@ -417,6 +445,14 @@ TEST(RoaringBitmap64Test, TestIteratorEqualOrLarger) { ASSERT_EQ(*iter, 100l); iter.EqualOrLarger(200l); ASSERT_EQ(iter, roaring.End()); + ASSERT_EQ(roaring.End(), iter); + iter.EqualOrLarger(3l); + ASSERT_EQ(*iter, 3l); + + iter = roaring.EqualOrLarger(200l); + ASSERT_EQ(iter, roaring.End()); + iter.EqualOrLarger(5l); + ASSERT_EQ(*iter, 5l); } // Helper function to convert a RoaringBitmap64 to a list of contiguous ranges. diff --git a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h index c428a0795..704247e95 100644 --- a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h +++ b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h @@ -57,8 +57,7 @@ class ApplyDeletionVectorBatchReader : public FileBatchReader { return batch_with_bitmap; } auto& [batch, bitmap] = batch_with_bitmap; - PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 valid_bitmap, Filter(batch.first->length)); - bitmap &= valid_bitmap; + PAIMON_ASSIGN_OR_RAISE(bitmap, Filter(bitmap)); if (bitmap.IsEmpty()) { ReaderUtils::ReleaseReadBatch(std::move(batch)); continue; @@ -101,9 +100,12 @@ class ApplyDeletionVectorBatchReader : public FileBatchReader { } private: - Result Filter(int32_t batch_size) const { + Result Filter(const RoaringBitmap32& selection) const { RoaringBitmap32 is_valid; - for (int32_t i = 0; i < batch_size; ++i) { + // Unselected rows may lie outside the group's anchor range. Probe only rows that the + // format/index selection actually returns, just as ApplyDeletionVectorReader does. + for (auto iter = selection.Begin(); iter != selection.End(); ++iter) { + int32_t i = *iter; PAIMON_ASSIGN_OR_RAISE(uint64_t file_row_id, reader_->GetPreviousBatchFileRowId(i)); PAIMON_ASSIGN_OR_RAISE(bool is_deleted, deletion_vector_->IsDeleted(file_row_id)); if (!is_deleted) { diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 5fea0f219..87dc99ccd 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -52,6 +52,7 @@ #include "paimon/common/utils/vector_store_utils.h" #include "paimon/core/core_options.h" #include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h" +#include "paimon/core/deletionvectors/bitmap_deletion_vector.h" #include "paimon/core/global_index/indexed_split_impl.h" #include "paimon/core/io/file_index_evaluator.h" #include "paimon/core/io/file_index_options.h" @@ -137,63 +138,53 @@ class PositionShiftedDeletionVector : public DeletionVector { }; } // namespace -int64_t DataEvolutionSplitRead::BlobBunch::RowCount() const { - if (files_.empty()) { - return 0; +Result DataEvolutionSplitRead::BlobBunch::RowCount() const { + const auto merged = Range::SortAndMergeOverlap(ranges_, /*adjacent=*/true); + if (expected_row_range_) { + for (const auto& range : merged) { + if (range.from < expected_row_range_->from || range.to > expected_row_range_->to) { + return Status::Invalid("Blob file range should be within normal file range."); + } + } + return expected_row_range_->Count(); } - if (!has_row_ids_selection_) { - // Add enforces the union range to be contiguous - return union_end_row_id_ - union_first_row_id_; + if (!has_row_ids_selection_ && merged.size() != 1) { + return Status::Invalid("Blob file bunch should always contain a contiguous row range."); } - // with a row-ids selection the scan may have pruned files, leaving holes in the union int64_t row_count = 0; - for (const auto& range : Range::SortAndMergeOverlap(ranges_, /*adjacent=*/true)) { + for (const auto& range : merged) { row_count += range.Count(); } return row_count; } +Result DataEvolutionSplitRead::BlobBunch::LogicalRange() const { + if (expected_row_range_) { + return *expected_row_range_; + } + const auto merged = Range::SortAndMergeOverlap(ranges_, /*adjacent=*/true); + if (merged.empty()) { + return Status::Invalid("Blob file bunch should not be empty."); + } + return Range(merged.front().from, merged.back().to); +} + Status DataEvolutionSplitRead::BlobBunch::Add(const std::shared_ptr& file) { if (!BlobUtils::IsBlobFile(file->file_name)) { return Status::Invalid("Only blob file can be added to a blob bunch."); } - PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, file->NonNullFirstRowId()); if (!files_.empty()) { // Blob files for the same field may span schema ids. if (file->write_cols != files_[0]->write_cols) { return Status::Invalid("All files in a blob bunch should have the same write columns."); } } - // files sharing a max sequence number form one layer, whose row id ranges must be disjoint; - // overlaps across layers are the expected shape of partial updates and are kept for the - // row-level placeholder fallback - auto [layer_iter, layer_inserted] = - sequence_group_end_.try_emplace(file->max_sequence_number, 0); - if (!layer_inserted && first_row_id < layer_iter->second) { - return Status::Invalid(fmt::format( - "Blob files with the same max sequence number should not have overlapping row id " - "ranges: file {} (max sequence number {}) starts at row id {} before the previous " - "file's end {}", - file->file_name, file->max_sequence_number, first_row_id, layer_iter->second)); - } - if (!has_row_ids_selection_ && !files_.empty() && first_row_id > union_end_row_id_) { - // a hole no layer covers cannot be aligned with the data files - return Status::Invalid( - fmt::format("Blob file first row id should be continuous, expect {} but got {}", - union_end_row_id_, first_row_id)); - } - int64_t end_row_id = first_row_id + file->row_count; - layer_iter->second = end_row_id; - union_first_row_id_ = std::min(union_first_row_id_, first_row_id); - union_end_row_id_ = std::max(union_end_row_id_, end_row_id); - ranges_.emplace_back(first_row_id, end_row_id - 1); + PAIMON_ASSIGN_OR_RAISE(Range range, file->NonNullRowIdRange()); files_.push_back(file); - if (!has_row_ids_selection_ && expected_row_count_ >= 0 && RowCount() > expected_row_count_) { - return Status::Invalid( - fmt::format("Blob files row count exceed the expect {}", expected_row_count_)); - } + ranges_.push_back(range); return Status::OK(); } + DataEvolutionSplitRead::DataEvolutionSplitRead( const std::shared_ptr& path_factory, const std::shared_ptr& context, @@ -346,7 +337,7 @@ Result> DataEvolutionSplitRead::CreateBlobViewReade PAIMON_ASSIGN_OR_RAISE(std::optional group_dv, ReadGroupDeletionVector(group, split_dv_factory)); PAIMON_ASSIGN_OR_RAISE(DeletionVector::Factory group_dv_factory, - CreateGroupDvFactory(data_files, group_dv)); + CreateGroupDvFactory(data_files, group_dv, row_ranges)); PAIMON_ASSIGN_OR_RAISE( std::vector> raw_file_readers, CreateRawFileReaders(split_impl->Partition(), data_files, blob_view_schema, @@ -431,26 +422,35 @@ Result> DataEvolutionSplitRead::InnerCreateReader( std::vector> sub_readers; for (const std::vector>& need_merge_files : split_by_row_id) { - if (need_merge_files.size() > 1) { - PAIMON_ASSIGN_OR_RAISE( - bool skip_group, - SkipByFileIndex(push_down_predicate, need_merge_files, data_file_path_factory)); - if (skip_group) { - continue; - } + PAIMON_ASSIGN_OR_RAISE(std::optional group_selection, + SelectByFileIndex(push_down_predicate, need_merge_files, read_schema, + row_ranges, data_file_path_factory)); + if (group_selection && group_selection->IsEmpty()) { + continue; } PAIMON_ASSIGN_OR_RAISE(std::optional group_dv, ReadGroupDeletionVector(need_merge_files, split_dv_factory)); PAIMON_ASSIGN_OR_RAISE(DeletionVector::Factory group_dv_factory, - CreateGroupDvFactory(need_merge_files, group_dv)); + CreateGroupDvFactory(need_merge_files, group_dv, row_ranges)); + if (group_dv && group_selection) { + PAIMON_ASSIGN_OR_RAISE( + *group_selection, + ExcludeDeletedPositions(std::move(*group_selection), group_dv->deletion_vector)); + if (group_selection->IsEmpty()) { + continue; + } + } + DataEvolutionSplitRead group_read(path_factory_, context_, pool_, executor_); + group_read.file_indexes_evaluated_ = true; + group_read.group_selection_ = std::move(group_selection); if (need_merge_files.size() == 1) { // No need to merge fields, just create a single file reader PAIMON_ASSIGN_OR_RAISE( std::vector> raw_file_readers, - CreateRawFileReaders(split_impl->Partition(), need_merge_files, read_schema, - push_down_predicate, group_dv_factory, row_ranges, - data_file_path_factory, - /*extra_format_options=*/{})); + group_read.CreateRawFileReaders(split_impl->Partition(), need_merge_files, + read_schema, push_down_predicate, group_dv_factory, + row_ranges, data_file_path_factory, + /*extra_format_options=*/{})); if (raw_file_readers.empty()) { continue; } @@ -461,8 +461,9 @@ Result> DataEvolutionSplitRead::InnerCreateReader( } else { PAIMON_ASSIGN_OR_RAISE( std::unique_ptr evolution_reader, - CreateUnionReader(split_impl->Partition(), need_merge_files, row_ranges, - data_file_path_factory, group_dv_factory, group_dv, read_schema)); + group_read.CreateUnionReader(split_impl->Partition(), need_merge_files, row_ranges, + data_file_path_factory, group_dv_factory, group_dv, + read_schema)); sub_readers.push_back(std::move(evolution_reader)); } } @@ -487,22 +488,55 @@ Result> DataEvolutionSplitRead::CreatePushDownPredica return PredicateUtils::CreatePickedFieldFilter(predicate, picked_field_name_to_idx); } -Result DataEvolutionSplitRead::SkipByFileIndex( +Result> DataEvolutionSplitRead::SelectByFileIndex( const std::shared_ptr& predicate, const std::vector>& files, + const std::shared_ptr& read_schema, + const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory) const { + if (row_ranges) { + std::vector file_ranges; + for (const auto& file : files) { + PAIMON_ASSIGN_OR_RAISE(Range range, file->NonNullRowIdRange()); + file_ranges.push_back(range); + } + if (Range::And(Range::SortAndMergeOverlap(*row_ranges, /*adjacent=*/true), + Range::SortAndMergeOverlap(file_ranges, /*adjacent=*/true)) + .empty()) { + return std::optional(RoaringBitmap32()); + } + } if (!options_.FileIndexReadEnabled() || !predicate) { - return false; + return std::optional(); + } + + // Like Java's canPushDownGroupSelection, only share exact bitmap selections across + // normal files with the same row-id range. Other groups can still be rejected by indexes. + bool can_push_down_selection = true; + if (files.size() > 1) { + PAIMON_ASSIGN_OR_RAISE(Range group_range, files[0]->NonNullRowIdRange()); + for (const auto& file : files) { + if (!DataEvolutionUtils::IsNormalFile(file->file_name)) { + can_push_down_selection = false; + break; + } + PAIMON_ASSIGN_OR_RAISE(Range file_range, file->NonNullRowIdRange()); + if (!(file_range == group_range)) { + can_push_down_selection = false; + break; + } + } } PAIMON_ASSIGN_OR_RAISE( std::unique_ptr field_mapping_builder, - FieldMappingBuilder::Create(raw_read_schema_, context_->GetPartitionKeys(), predicate)); + FieldMappingBuilder::Create(read_schema, context_->GetPartitionKeys(), predicate)); + std::optional selection; std::set claimed_field_ids; for (const auto& file : files) { // Blob and vector-store files may cover only part of the row range, so their indexes // cannot prove that the complete merged group misses the predicate. - if (!DataEvolutionUtils::IsNormalFile(file->file_name)) { + if (files.size() > 1 && !DataEvolutionUtils::IsNormalFile(file->file_name)) { continue; } @@ -555,10 +589,28 @@ Result DataEvolutionSplitRead::SkipByFileIndex( file, options_.GetFileSystem(), pool_)); PAIMON_ASSIGN_OR_RAISE(bool is_remain, index_result->IsRemain()); if (!is_remain) { - return true; + return std::optional(RoaringBitmap32()); + } + if (!can_push_down_selection) { + continue; + } + if (auto* bitmap_index = dynamic_cast(index_result.get())) { + PAIMON_ASSIGN_OR_RAISE(const RoaringBitmap32* bitmap, bitmap_index->GetBitmap()); + if (selection) { + *selection &= *bitmap; + } else { + selection = *bitmap; + } + } + } + if (selection && row_ranges) { + PAIMON_ASSIGN_OR_RAISE(std::optional range_selection, + files[0]->ToFileSelection(row_ranges)); + if (range_selection) { + *selection &= *range_selection; } } - return false; + return selection; } Result> DataEvolutionSplitRead::ApplyIndexAndDvReaderIfNeeded( @@ -570,7 +622,7 @@ Result> DataEvolutionSplitRead::ApplyIndexAndDv const std::map& file_index_options, std::vector* index_scores) const { std::shared_ptr file_index_result; - if (options_.FileIndexReadEnabled()) { + if (options_.FileIndexReadEnabled() && !file_indexes_evaluated_) { PAIMON_ASSIGN_OR_RAISE(file_index_result, FileIndexEvaluator::Evaluate(data_schema, file_index_options, predicate, data_file_path_factory, file, @@ -580,7 +632,7 @@ Result> DataEvolutionSplitRead::ApplyIndexAndDv return std::unique_ptr(); } } - const RoaringBitmap32* index_selection = nullptr; + const RoaringBitmap32* index_selection = group_selection_ ? &*group_selection_ : nullptr; if (auto* bitmap_index = dynamic_cast(file_index_result.get())) { PAIMON_ASSIGN_OR_RAISE(index_selection, bitmap_index->GetBitmap()); } @@ -633,14 +685,14 @@ DataEvolutionSplitRead::SplitFieldBunches( bool has_row_ranges_selection) { std::vector> fields_files; std::map> blob_bunch_map; - int64_t row_count = -1; + std::optional expected_row_range; for (const auto& file : need_merge_files) { if (BlobUtils::IsBlobFile(file->file_name)) { PAIMON_ASSIGN_OR_RAISE(int32_t field_id, blob_field_to_field_id(file)); auto iter = blob_bunch_map.find(field_id); if (iter == blob_bunch_map.end()) { auto blob_bunch = std::make_shared( - row_count, has_row_ranges_selection); + expected_row_range, has_row_ranges_selection); PAIMON_RETURN_NOT_OK(blob_bunch->Add(file)); blob_bunch_map.insert(std::make_pair(field_id, blob_bunch)); } else { @@ -650,7 +702,9 @@ DataEvolutionSplitRead::SplitFieldBunches( } else { // Normal file, just add it to the current merge split fields_files.push_back(std::make_shared(file)); - row_count = file->row_count; + if (DataEvolutionUtils::IsNormalFile(file->file_name)) { + PAIMON_ASSIGN_OR_RAISE(expected_row_range, file->NonNullRowIdRange()); + } } } for (const auto& [field_id, blob_bunch] : blob_bunch_map) { @@ -889,6 +943,8 @@ Result> DataEvolutionSplitRead::CreateUnionReader( TryPlanSequentialVectorRead(need_merge_files, file_to_field_ids)); auto split_read = std::make_shared(path_factory_, context_, pool_, executor_); + split_read->file_indexes_evaluated_ = file_indexes_evaluated_; + split_read->group_selection_ = group_selection_; if (!sequential_bunches.empty()) { std::vector sources; sources.push_back(LazyConcatBatchReader::Source{ @@ -970,16 +1026,23 @@ Result> DataEvolutionSplitRead::CreateFieldUnionRea 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()); + auto bunch_first_row_id = [](const std::shared_ptr& bunch) -> Result { + if (auto blob_bunch = std::dynamic_pointer_cast(bunch)) { + PAIMON_ASSIGN_OR_RAISE(Range range, blob_bunch->LogicalRange()); + return range.from; + } + return bunch->Files()[0]->NonNullFirstRowId(); + }; + PAIMON_ASSIGN_OR_RAISE(int64_t row_count, fields_files[0]->RowCount()); + PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, bunch_first_row_id(fields_files[0])); if (!row_ranges) { for (const auto& bunch : fields_files) { - if (bunch->RowCount() != row_count) { + PAIMON_ASSIGN_OR_RAISE(int64_t bunch_row_count, bunch->RowCount()); + if (bunch_row_count != row_count) { return Status::Invalid( "All files in a field merge split should have the same row count."); } - PAIMON_ASSIGN_OR_RAISE(int64_t current_bunch_first_row_id, - bunch->Files()[0]->NonNullFirstRowId()); + PAIMON_ASSIGN_OR_RAISE(int64_t current_bunch_first_row_id, bunch_first_row_id(bunch)); if (current_bunch_first_row_id != first_row_id) { return Status::Invalid( "All files in a field merge split should have the same first row id and could " @@ -1048,6 +1111,8 @@ Result> DataEvolutionSplitRead::CreateFieldUnionRea std::vector sources; auto split_read = std::make_shared(path_factory_, context_, pool_, executor_); + split_read->file_indexes_evaluated_ = file_indexes_evaluated_; + split_read->group_selection_ = group_selection_; for (const auto& file : bunch->Files()) { std::optional> selected = row_ranges; if (row_ranges) { @@ -1077,14 +1142,13 @@ Result> DataEvolutionSplitRead::CreateFieldUnionRea } 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 + } else if (blob_bunch) { + PAIMON_ASSIGN_OR_RAISE(Range logical_range, blob_bunch->LogicalRange()); PAIMON_ASSIGN_OR_RAISE( file_batch_readers[file_idx], - CreateBlobFallbackReader(partition, blob_bunch->Files(), file_read_schema, - row_ranges, data_file_path_factory, group_dv_factory, - group_dv)); + CreateBlobFallbackReader(partition, blob_bunch->Files(), logical_range, + file_read_schema, row_ranges, data_file_path_factory, + group_dv_factory, group_dv)); } else { PAIMON_ASSIGN_OR_RAISE( std::vector> file_readers, @@ -1097,7 +1161,6 @@ Result> DataEvolutionSplitRead::CreateFieldUnionRea } else { auto raw_readers = ObjectUtils::MoveVector>( std::move(file_readers)); - // Concat multiple blob files that map to the same data file. file_batch_readers[file_idx] = std::make_unique(std::move(raw_readers), arrow_pool_); } @@ -1148,20 +1211,28 @@ std::vector SelectedRangesInRange(int64_t from, int64_t to, Result> DataEvolutionSplitRead::CreateBlobFallbackReader( const BinaryRow& partition, const std::vector>& files, - const std::shared_ptr& file_read_schema, + const Range& logical_range, const std::shared_ptr& file_read_schema, 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 { - int64_t union_first_row_id = std::numeric_limits::max(); - int64_t union_last_row_id = std::numeric_limits::min(); + if (files.empty()) { + return Status::Invalid("Blob bunch should not be empty."); + } + int64_t union_first_row_id = logical_range.from; + int64_t union_last_row_id = logical_range.to; using FileWithFirstRowId = std::pair>; std::map, std::greater<>> sequence_groups; for (const auto& file : files) { - PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, file->NonNullFirstRowId()); - union_first_row_id = std::min(union_first_row_id, first_row_id); - union_last_row_id = std::max(union_last_row_id, first_row_id + file->row_count - 1); - sequence_groups[file->max_sequence_number].emplace_back(first_row_id, file); + PAIMON_ASSIGN_OR_RAISE(Range file_range, file->NonNullRowIdRange()); + if (!row_ranges && + (file_range.from < logical_range.from || file_range.to > logical_range.to)) { + return Status::Invalid("Blob file range should be within logical range."); + } + if (row_ranges && !Range::HasIntersection(file_range, logical_range)) { + return Status::Invalid("Blob file range should intersect logical range."); + } + sequence_groups[file->max_sequence_number].emplace_back(file_range.from, file); } // the blob format reader must emit placeholder sentinels instead of failing on them const std::map blob_format_options = { @@ -1173,21 +1244,24 @@ Result> DataEvolutionSplitRead::CreateBlobFallbackR [](const FileWithFirstRowId& f1, const FileWithFirstRowId& f2) { return f1.first < f2.first; }); - std::vector segments; - // pad row ids this layer does not cover with placeholder gaps, so that every layer spans - // the same union range and the groups can be stepped in lockstep - int64_t next_row_id = union_first_row_id; + int64_t previous_file_end = std::numeric_limits::min(); for (const auto& [first_row_id, file] : group_files) { - if (first_row_id < next_row_id) { + if (first_row_id < previous_file_end) { return Status::Invalid(fmt::format( "Blob files with the same max sequence number should not have overlapping " "row id ranges: file {} (max sequence number {}) starts at row id {} before " "the previous file's end {}", - file->file_name, max_sequence_number, first_row_id, next_row_id)); + file->file_name, max_sequence_number, first_row_id, previous_file_end)); } - if (first_row_id > next_row_id) { - std::vector gap_selected_ranges = - SelectedRangesInRange(next_row_id, first_row_id - 1, row_ranges); + previous_file_end = first_row_id + file->row_count; + } + std::vector segments; + // Pad rows this layer does not cover, so that every layer spans the logical range. + int64_t next_row_id = union_first_row_id; + for (const auto& [first_row_id, file] : group_files) { + if (first_row_id > next_row_id && next_row_id <= union_last_row_id) { + std::vector gap_selected_ranges = SelectedRangesInRange( + next_row_id, std::min(first_row_id - 1, union_last_row_id), row_ranges); if (group_dv && !gap_selected_ranges.empty()) { // gap segments have no file reader to wrap, so the deleted row ids are // dropped from their ranges to stay aligned with the other layers @@ -1200,12 +1274,15 @@ Result> DataEvolutionSplitRead::CreateBlobFallbackR BlobFallbackBatchReader::Segment{nullptr, std::move(gap_selected_ranges)}); } } - if (!SelectedRangesInRange(first_row_id, first_row_id + file->row_count - 1, row_ranges) + std::optional> file_ranges = + SelectedRangesInRange(union_first_row_id, union_last_row_id, row_ranges); + if (!SelectedRangesInRange(first_row_id, first_row_id + file->row_count - 1, + file_ranges) .empty()) { PAIMON_ASSIGN_OR_RAISE( std::vector> file_readers, CreateRawFileReaders(partition, {file}, file_read_schema, - /*predicate=*/nullptr, group_dv_factory, row_ranges, + /*predicate=*/nullptr, group_dv_factory, file_ranges, data_file_path_factory, blob_format_options)); if (file_readers.size() != 1) { return Status::Invalid("Unexpected: blob fallback file reader was skipped."); @@ -1213,7 +1290,7 @@ Result> DataEvolutionSplitRead::CreateBlobFallbackR segments.push_back( BlobFallbackBatchReader::Segment{std::move(file_readers[0]), {}}); } - next_row_id = first_row_id + file->row_count; + next_row_id = std::max(next_row_id, first_row_id + file->row_count); } if (next_row_id <= union_last_row_id) { std::vector gap_selected_ranges = @@ -1227,6 +1304,10 @@ Result> DataEvolutionSplitRead::CreateBlobFallbackR BlobFallbackBatchReader::Segment{nullptr, std::move(gap_selected_ranges)}); } } + if (segments.empty()) { + return std::make_unique(std::vector>(), + arrow_pool_); + } groups.push_back(std::move(segments)); } PAIMON_ASSIGN_OR_RAISE( @@ -1334,14 +1415,6 @@ DataEvolutionSplitRead::ReadGroupDeletionVector( if (!split_dv_factory) { return std::optional(); } - // A split may pack a group holding nothing but blob files, written when only the blob column - // was, next to one that does carry a deletion vector. No anchor means no writer could have - // keyed a vector by it, so it reads undeleted instead of failing the whole split. - if (std::none_of(group.begin(), group.end(), [](const std::shared_ptr& file) { - return DataEvolutionUtils::IsNormalFile(file->file_name); - })) { - return std::optional(); - } PAIMON_ASSIGN_OR_RAISE(std::shared_ptr anchor, DataEvolutionUtils::RetrieveAnchorFile(group)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr deletion_vector, @@ -1356,7 +1429,8 @@ DataEvolutionSplitRead::ReadGroupDeletionVector( Result DataEvolutionSplitRead::CreateGroupDvFactory( const std::vector>& group, - const std::optional& group_dv) { + const std::optional& group_dv, + const std::optional>& row_ranges) { if (!group_dv) { return DeletionVector::Factory(); } @@ -1365,11 +1439,18 @@ Result DataEvolutionSplitRead::CreateGroupDvFactory( for (const auto& file : group) { PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, file->NonNullFirstRowId()); int64_t last_row_id = first_row_id + file->row_count - 1; - if (first_row_id < group_dv->anchor_range.from || last_row_id > group_dv->anchor_range.to) { - return Status::Invalid(fmt::format( - "Deletion vector range [{}, {}] should contain row id range [{}, {}] of file {}.", - group_dv->anchor_range.from, group_dv->anchor_range.to, first_row_id, last_row_id, - file->file_name)); + const std::vector selected_ranges = + row_ranges ? Range::And({Range(first_row_id, last_row_id)}, + Range::SortAndMergeOverlap(*row_ranges, /*adjacent=*/true)) + : std::vector{Range(first_row_id, last_row_id)}; + for (const auto& range : selected_ranges) { + if (range.from < group_dv->anchor_range.from || range.to > group_dv->anchor_range.to) { + return Status::Invalid(fmt::format( + "Deletion vector range [{}, {}] should contain selected row id range [{}, {}] " + "of file {}.", + group_dv->anchor_range.from, group_dv->anchor_range.to, range.from, range.to, + file->file_name)); + } } int64_t offset = first_row_id - group_dv->anchor_range.from; bool covers_whole_anchor_range = offset == 0 && last_row_id == group_dv->anchor_range.to; @@ -1388,8 +1469,81 @@ Result DataEvolutionSplitRead::CreateGroupDvFactory( }); } +Result DataEvolutionSplitRead::ExcludeDeletedPositions( + RoaringBitmap32 selection, const std::shared_ptr& deletion_vector) { + if (selection.IsEmpty() || deletion_vector->IsEmpty()) { + return selection; + } + if (auto* bitmap_dv = dynamic_cast(deletion_vector.get())) { + selection -= *bitmap_dv->GetBitmap(); + return selection; + } + PAIMON_ASSIGN_OR_RAISE(int64_t cardinality, deletion_vector->GetCardinality()); + if (cardinality < selection.Cardinality()) { + RoaringBitmap32 deleted; + PAIMON_RETURN_NOT_OK(deletion_vector->ForEachDeletedPosition([&](int64_t position) { + if (position >= 0 && position <= std::numeric_limits::max()) { + deleted.Add(static_cast(position)); + } + })); + selection -= deleted; + return selection; + } + RoaringBitmap32 live; + const auto end = selection.End(); + for (auto iter = selection.Begin(); iter != end; ++iter) { + PAIMON_ASSIGN_OR_RAISE(bool is_deleted, + deletion_vector->IsDeleted(static_cast(*iter))); + if (!is_deleted) { + live.Add(*iter); + } + } + return live; +} + Result> DataEvolutionSplitRead::ExcludeDeletedRowIds( const std::vector& ranges, const GroupDeletionVector& group_dv) { + if (ranges.empty()) { + return std::vector(); + } + PAIMON_ASSIGN_OR_RAISE(int64_t cardinality, group_dv.deletion_vector->GetCardinality()); + int64_t selected_count = 0; + for (const auto& range : ranges) { + if (range.Count() > cardinality - selected_count) { + // Sparse deletions in a large selection: visit deleted positions instead of every + // selected row. Coalesce runs before taking the complement to keep memory bounded + // for contiguous deletions. + std::vector deleted; + PAIMON_RETURN_NOT_OK( + group_dv.deletion_vector->ForEachDeletedPosition([&](int64_t position) { + if (position < 0 || + position > group_dv.anchor_range.to - group_dv.anchor_range.from) { + return; + } + int64_t row_id = group_dv.anchor_range.from + position; + // The caller supplies sorted, disjoint selected gap ranges. Do not expand + // deletions elsewhere in the anchor into temporary ranges. + auto selected = std::lower_bound( + ranges.begin(), ranges.end(), row_id, + [](const Range& range, int64_t value) { return range.to < value; }); + if (selected == ranges.end() || selected->from > row_id) { + return; + } + if (!deleted.empty() && deleted.back().to < row_id && + row_id - deleted.back().to == 1) { + deleted.back().to = row_id; + } else { + deleted.emplace_back(row_id, row_id); + } + })); + if (deleted.empty()) { + return ranges; + } + return Range::And(ranges, + Range(ranges.front().from, ranges.back().to).Exclude(deleted)); + } + selected_count += range.Count(); + } std::vector remaining; for (const auto& range : ranges) { std::optional run_start; diff --git a/src/paimon/core/operation/data_evolution_split_read.h b/src/paimon/core/operation/data_evolution_split_read.h index e0406495c..b7610dc1c 100644 --- a/src/paimon/core/operation/data_evolution_split_read.h +++ b/src/paimon/core/operation/data_evolution_split_read.h @@ -20,7 +20,6 @@ #include #include -#include #include #include #include @@ -37,6 +36,7 @@ #include "paimon/reader/batch_reader.h" #include "paimon/result.h" #include "paimon/status.h" +#include "paimon/utils/roaring_bitmap32.h" namespace arrow { class Schema; @@ -62,7 +62,7 @@ struct DeletionFile; /// splits)->(BlobViewResolvingBatchReader)->(CompleteIndexScoreBatchReader)-> /// CompleteRowKindBatchReader->(PredicateBatchReader) /// ->ConcatBatchReader across files->DataEvolutionFileReader -/// ->(ConcatBatchReader across blob files | BlobFallbackBatchReader across blob sequence layers) +/// ->(ConcatBatchReader across vector files | BlobFallbackBatchReader across blob sequence layers) /// ->FieldMappingReader->(ApplyDeletionVectorBatchReader)->(ApplyBitmapIndexBatchReader) /// ->(CompleteRowTrackingFieldsBatchReader)->(ShreddingFileReader) /// ->(VectorFileBatchReader)->(DelegatingPrefetchReader)->(PrefetchFileBatchReader) @@ -70,9 +70,9 @@ struct DeletionFile; /// /// /// A union `SplitRead` to read multiple inner files to merge columns. A single-file row range -/// group gets both file-index and format-level predicate pushdown. A merged group only uses file -/// indexes to skip the whole group: filtering its child readers independently would break their -/// positional alignment. +/// group gets both file-index and format-level predicate pushdown. A merged group of normal +/// files sharing one row-id range shares the intersection of its file-index selections across +/// all child readers to preserve alignment. /// /// Deletion vectors are supported: a row range group's vector is maintained against the /// group's anchor file (DataEvolutionUtils::RetrieveAnchorFile), so its positions are @@ -105,7 +105,7 @@ class DataEvolutionSplitRead : public AbstractSplitRead { class FieldBunch { public: virtual ~FieldBunch() = default; - virtual int64_t RowCount() const = 0; + virtual Result RowCount() const = 0; virtual const std::vector>& Files() const = 0; }; @@ -113,7 +113,7 @@ class DataEvolutionSplitRead : public AbstractSplitRead { public: explicit DataBunch(const std::shared_ptr& data_file) : data_files_({data_file}) {} - int64_t RowCount() const override { + Result RowCount() const override { return data_files_[0]->row_count; } const std::vector>& Files() const override { @@ -129,7 +129,7 @@ class DataEvolutionSplitRead : public AbstractSplitRead { public: VectorBunch(std::vector> files, int64_t row_count) : files_(std::move(files)), row_count_(row_count) {} - int64_t RowCount() const override { + Result RowCount() const override { return row_count_; } const std::vector>& Files() const override { @@ -145,38 +145,22 @@ class DataEvolutionSplitRead : public AbstractSplitRead { /// (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 /// placeholder entries and reading falls back row by row to older layers. - /// Files must be added ordered by first row id ascending (then max sequence number - /// descending), as produced by MergeRangesAndSort. class BlobBunch : public FieldBunch { public: - explicit BlobBunch(int64_t expected_row_count, bool has_row_ids_selection) - : expected_row_count_(expected_row_count), + BlobBunch(const std::optional& expected_row_range, bool has_row_ids_selection) + : expected_row_range_(expected_row_range), has_row_ids_selection_(has_row_ids_selection) {} - /// Number of distinct row ids covered by the added files. Without a row-ids selection - /// the covered range is contiguous (enforced by Add) and matches the data files. - int64_t RowCount() const override; + /// The normal file owns the logical rows, including rows with no physical blob value. + Result RowCount() const override; + Result LogicalRange() const; const std::vector>& Files() const override { return files_; } Status Add(const std::shared_ptr& file); - /// True when every file shares one max sequence number, so the files are read - /// sequentially without the fallback merge. A lone layer is expected to hold no - /// placeholder entries; if one does (a user value equal to the placeholder sentinel - /// written by a first write), the strict blob reader rejects the read, since no older - /// layer exists to resolve it. - bool SequentialReadOptimize() const { - return sequence_group_end_.size() <= 1; - } private: - int64_t expected_row_count_ = -1; + std::optional expected_row_range_; bool has_row_ids_selection_ = false; - int64_t union_first_row_id_ = std::numeric_limits::max(); - /// Exclusive end of the union row id range covered so far. - int64_t union_end_row_id_ = std::numeric_limits::min(); - /// Per max sequence number: exclusive end of the last added range, to reject - /// overlapping files within one layer. - std::map sequence_group_end_; std::vector ranges_; std::vector> files_; }; @@ -201,12 +185,15 @@ class DataEvolutionSplitRead : public AbstractSplitRead { const std::shared_ptr& predicate, const std::shared_ptr& read_schema); - /// Returns true when file indexes prove that no row in a merged row range group can match. - /// Only normal files are considered, and an older copy of a field is excluded after a newer - /// file has claimed the same field id. - Result SkipByFileIndex( + /// Evaluates the winning files' indexes before opening the group's DV or data files. Shared + /// bitmap selection is restricted to normal files with the same row-id range; a single-file + /// group may always use its bitmap. The bitmap uses file-local positions and is intersected + /// with `row_ranges`. Nullopt means no bitmap restriction; an empty bitmap rejects the group. + Result> SelectByFileIndex( const std::shared_ptr& predicate, const std::vector>& files, + const std::shared_ptr& read_schema, + const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory) const; /// Builds the deletion vector factory over the split's deletion files, keyed by data file @@ -225,11 +212,16 @@ class DataEvolutionSplitRead : public AbstractSplitRead { /// the anchor range and limited to its row count. Null factory when the group has no vector. static Result CreateGroupDvFactory( const std::vector>& group, - const std::optional& group_dv); + const std::optional& group_dv, + const std::optional>& row_ranges); + + /// Removes deleted positions without expanding the selection into row-id ranges. Both the + /// selection and the deletion vector must use the same file-local positions. + static Result ExcludeDeletedPositions( + RoaringBitmap32 selection, const std::shared_ptr& deletion_vector); /// Removes the row ids deleted by `group_dv` from the sorted disjoint `ranges` (absolute - /// row ids). Probes the vector once per row id, since it exposes no range API; the cost is - /// paid once per gap segment while building the reader, not per batch. + /// row ids). Iterates over deletions when they are sparser than the selected rows. static Result> ExcludeDeletedRowIds(const std::vector& ranges, const GroupDeletionVector& group_dv); @@ -318,11 +310,19 @@ class DataEvolutionSplitRead : public AbstractSplitRead { /// emitting the same surviving rows. Result> CreateBlobFallbackReader( const BinaryRow& partition, const std::vector>& files, - const std::shared_ptr& file_read_schema, + const Range& logical_range, const std::shared_ptr& file_read_schema, 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; + + /// Set only on a per-group reader after SelectByFileIndex has evaluated its indexes. + /// Lazy child readers inherit it so no sidecar is reopened during reader construction. + bool file_indexes_evaluated_ = false; + + /// Shared file-local selection on a per-group reader. Present only for a single-file group + /// or normal files with identical row-id ranges; every column reader consumes the same bitmap. + std::optional group_selection_; }; } // namespace paimon 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 4bd0e0c29..a68dfdf8b 100644 --- a/src/paimon/core/operation/data_evolution_split_read_test.cpp +++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp @@ -30,11 +30,14 @@ #include "arrow/ipc/json_simple.h" #include "gtest/gtest.h" #include "paimon/common/data/binary_row.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/file_index/bitmap/bitmap_file_index.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/apply_deletion_vector_batch_reader.h" #include "paimon/core/deletionvectors/bitmap64_deletion_vector.h" #include "paimon/core/deletionvectors/bitmap_deletion_vector.h" #include "paimon/core/io/data_file_meta.h" @@ -46,6 +49,7 @@ #include "paimon/core/table/source/data_split_impl.h" #include "paimon/defs.h" #include "paimon/executor.h" +#include "paimon/file_index/file_index_writer.h" #include "paimon/format/file_format_factory.h" #include "paimon/format/format_writer.h" #include "paimon/format/writer_builder.h" @@ -55,6 +59,8 @@ #include "paimon/predicate/predicate_builder.h" #include "paimon/read_context.h" #include "paimon/status.h" +#include "paimon/testing/mock/mock_file_batch_reader.h" +#include "paimon/testing/mock/mock_search_file_index.h" #include "paimon/testing/utils/read_result_collector.h" #include "paimon/testing/utils/testharness.h" @@ -206,12 +212,13 @@ TEST_F(DataEvolutionSplitReadTest, TestAddSingleBlobEntry) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); ASSERT_EQ(blob_bunch->Files().size(), 1); ASSERT_EQ(blob_bunch->Files()[0], blob_entry); - ASSERT_EQ(blob_bunch->RowCount(), 100); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 100); } TEST_F(DataEvolutionSplitReadTest, TestAddBlobEntryAndTail) { @@ -224,21 +231,22 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobEntryAndTail) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); ASSERT_OK(blob_bunch->Add(blob_tail)); ASSERT_EQ(blob_bunch->Files().size(), 2); ASSERT_EQ(blob_bunch->Files()[0], blob_entry); ASSERT_EQ(blob_bunch->Files()[1], blob_tail); - ASSERT_EQ(blob_bunch->RowCount(), 300); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 300); } TEST_F(DataEvolutionSplitReadTest, TestAddNonBlobFileInvalid) { auto blob_entry = CreateDataFileMeta("normal1.parquet", /*first_row_id=*/0, FileSource::Append()); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_NOK_WITH_MSG(blob_bunch->Add(blob_entry), "Only blob file can be added to a blob bunch."); } @@ -257,7 +265,7 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobWithSameFirstRowId) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); // Files with the same first row id and lower sequence numbers are older layers of a // partial update; they are kept for the row-level placeholder fallback, whether they @@ -269,8 +277,8 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobWithSameFirstRowId) { ASSERT_EQ(blob_bunch->Files()[0], blob_entry); ASSERT_EQ(blob_bunch->Files()[1], blob_full_tail); ASSERT_EQ(blob_bunch->Files()[2], blob_short_tail); - ASSERT_EQ(blob_bunch->RowCount(), 100); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 100); } TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithOverlappingRowId) { @@ -283,14 +291,14 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithOverlappingRowId) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); // Overlapping layers with different sequence numbers are kept for the fallback. ASSERT_OK(blob_bunch->Add(blob_tail)); ASSERT_EQ(blob_bunch->Files().size(), 2); - ASSERT_EQ(blob_bunch->RowCount(), 200); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 200); } TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithOverlappingRowIdAndHigherSequenceNumber) { @@ -303,13 +311,13 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithOverlappingRowIdAndHigherS /*max_sequence_number=*/2, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); ASSERT_OK(blob_bunch->Add(blob_tail)); ASSERT_EQ(blob_bunch->Files().size(), 2); - ASSERT_EQ(blob_bunch->RowCount(), 200); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 200); } TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithOverlappingRowIdInSameLayer) { @@ -322,12 +330,12 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithOverlappingRowIdInSameLaye /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); - // Files sharing a max sequence number form one layer and must not overlap. - ASSERT_NOK_WITH_MSG(blob_bunch->Add(blob_tail), - "Blob files with the same max sequence number should not have overlapping " - "row id ranges"); + // The bunch collects ranges; the fallback reader validates overlaps within each layer. + ASSERT_OK(blob_bunch->Add(blob_tail)); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 200); } TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithNonContinuousRowId) { @@ -340,11 +348,11 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithNonContinuousRowId) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); - // Adding file with non-continuous row id should return bad status - ASSERT_NOK_WITH_MSG(blob_bunch->Add(blob_tail), - "Blob file first row id should be continuous, expect 100 but got 200"); + ASSERT_OK(blob_bunch->Add(blob_tail)); + ASSERT_NOK_WITH_MSG(blob_bunch->RowCount(), + "Blob file bunch should always contain a contiguous row range."); } TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithDifferentWriteCols) { @@ -357,13 +365,79 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithDifferentWriteCols) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"diff_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry)); // Adding file with different write columns should return bad status ASSERT_NOK_WITH_MSG(blob_bunch->Add(blob_tail), "All files in a blob bunch should have the same write columns."); } +TEST_F(DataEvolutionSplitReadTest, TestBlobBunchChecksContinuityAfterCollectingAllFiles) { + DataEvolutionSplitRead::BlobBunch bunch(std::nullopt, /*has_row_ids_selection=*/false); + for (int64_t first_row_id : {0, 20, 10}) { + ASSERT_OK(bunch.Add(CreateBlobFile(std::to_string(first_row_id), first_row_id, 10, 1, + std::optional>({"blob_col"})))); + } + ASSERT_OK_AND_ASSIGN(int64_t row_count, bunch.RowCount()); + ASSERT_EQ(row_count, 30); + ASSERT_OK_AND_ASSIGN(Range range, bunch.LogicalRange()); + ASSERT_EQ(range, Range(0, 29)); +} + +TEST_F(DataEvolutionSplitReadTest, TestBlobBunchRowCountValidatesExpectedRange) { + for (bool has_row_ids_selection : {false, true}) { + DataEvolutionSplitRead::BlobBunch bunch(Range(100, 109), has_row_ids_selection); + ASSERT_OK(bunch.Add(CreateBlobFile("blob", 99, 4, 1, + std::optional>({"blob_col"})))); + ASSERT_NOK_WITH_MSG(bunch.RowCount(), + "Blob file range should be within normal file range."); + ASSERT_OK_AND_ASSIGN(Range range, bunch.LogicalRange()); + ASSERT_EQ(range, Range(100, 109)); + } +} + +TEST_F(DataEvolutionSplitReadTest, TestEmptyBlobBunchLogicalRange) { + DataEvolutionSplitRead::BlobBunch bunch(std::nullopt, /*has_row_ids_selection=*/true); + ASSERT_NOK_WITH_MSG(bunch.LogicalRange(), "Blob file bunch should not be empty."); +} + +TEST_F(DataEvolutionSplitReadTest, TestBlobFallbackValidatesRangesBeforeOpeningFiles) { + auto dir = UniqueTestDirectory::Create("local"); + auto schema = DataField::ConvertDataFieldsToArrowSchema( + {DataField(0, arrow::field("id", arrow::int32()))}); + const std::map options = {{Options::FILE_FORMAT, "parquet"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, + TableSchema::Create(0, schema, {}, {}, options)); + ReadContextBuilder builder(dir->Str()); + 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)); + DataEvolutionSplitRead split_read(/*path_factory=*/nullptr, context, GetDefaultPool(), + /*executor=*/nullptr); + auto blob = + CreateBlobFile("blob", 100, 10, 1, std::optional>({"blob_col"})); + auto overlap = + CreateBlobFile("overlap", 105, 5, 1, std::optional>({"blob_col"})); + auto create_reader = [&](const std::vector>& files, + const Range& range, + const std::optional>& row_ranges) { + return split_read.CreateBlobFallbackReader( + BinaryRow::EmptyRow(), files, range, schema, row_ranges, + /*data_file_path_factory=*/nullptr, /*group_dv_factory=*/nullptr, + /*group_dv=*/std::nullopt); + }; + ASSERT_NOK_WITH_MSG(create_reader({}, Range(100, 109), std::nullopt), + "Blob bunch should not be empty."); + ASSERT_NOK_WITH_MSG(create_reader({blob}, Range(101, 109), std::nullopt), + "Blob file range should be within logical range."); + ASSERT_NOK_WITH_MSG(create_reader({blob}, Range(200, 209), std::vector{Range(200, 200)}), + "Blob file range should intersect logical range."); + ASSERT_NOK_WITH_MSG(create_reader({overlap, blob}, Range(100, 109), std::nullopt), + "should not have overlapping row id ranges"); +} + TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap) { auto blob_entry = CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/10, @@ -380,7 +454,7 @@ TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap) { /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/true); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/true); ASSERT_OK(blob_bunch->Add(blob_entry)); // blob_sub1 was pruned by the row-ids selection in the scan process; blob_sub2 is a newer // layer and both files are kept for the row-level placeholder fallback. @@ -388,8 +462,8 @@ TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap) { ASSERT_EQ(blob_bunch->Files().size(), 2); ASSERT_EQ(blob_bunch->Files()[0], blob_entry); ASSERT_EQ(blob_bunch->Files()[1], blob_sub2); - ASSERT_EQ(blob_bunch->RowCount(), 10); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 10); } TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap2) { @@ -408,7 +482,7 @@ TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap2) { /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/true); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/true); ASSERT_OK(blob_bunch->Add(blob_entry)); // blob_sub1 was pruned by the row-ids selection in the scan process; blob_sub2 is an older // layer and both files are kept for the row-level placeholder fallback. @@ -416,8 +490,8 @@ TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap2) { ASSERT_EQ(blob_bunch->Files().size(), 2); ASSERT_EQ(blob_bunch->Files()[0], blob_entry); ASSERT_EQ(blob_bunch->Files()[1], blob_sub2); - ASSERT_EQ(blob_bunch->RowCount(), 10); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 10); } TEST_F(DataEvolutionSplitReadTest, TestRowIdSelection) { @@ -436,14 +510,15 @@ TEST_F(DataEvolutionSplitReadTest, TestRowIdSelection) { /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/true); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/true); ASSERT_OK(blob_bunch->Add(blob_entry0)); // blob_entry1 will not be added, for it has been skipped by row_ids in scan process. ASSERT_OK(blob_bunch->Add(blob_entry2)); ASSERT_EQ(blob_bunch->Files().size(), 2); ASSERT_EQ(blob_bunch->Files()[0], blob_entry0); ASSERT_EQ(blob_bunch->Files()[1], blob_entry2); - ASSERT_EQ(blob_bunch->RowCount(), 20); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 20); } TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario) { @@ -464,7 +539,7 @@ TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario) { /*max_sequence_number=*/1, /*write_cols=*/std::optional>({"blob_col"})); auto blob_bunch = std::make_shared( - INT64_MAX, /*has_row_ids_selection=*/false); + /*expected_row_range=*/std::nullopt, /*has_row_ids_selection=*/false); ASSERT_OK(blob_bunch->Add(blob_entry1)); ASSERT_OK(blob_bunch->Add(blob_entry2)); ASSERT_OK(blob_bunch->Add(blob_entry3)); @@ -475,7 +550,8 @@ TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario) { ASSERT_EQ(blob_bunch->Files()[1], blob_entry2); ASSERT_EQ(blob_bunch->Files()[2], blob_entry3); ASSERT_EQ(blob_bunch->Files()[3], blob_entry4); - ASSERT_EQ(blob_bunch->RowCount(), 1000); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 1000); } TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario2) { @@ -571,8 +647,8 @@ TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario2) { ASSERT_EQ(blob_bunch->Files()[6], blob_entry7); ASSERT_EQ(blob_bunch->Files()[7], blob_entry4); ASSERT_EQ(blob_bunch->Files()[8], blob_entry8); - ASSERT_EQ(blob_bunch->RowCount(), 1000); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 1000); } TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario3) { @@ -699,8 +775,8 @@ TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario3) { ASSERT_EQ(blob_bunch->Files()[6], blob_entry7); ASSERT_EQ(blob_bunch->Files()[7], blob_entry4); ASSERT_EQ(blob_bunch->Files()[8], blob_entry8); - ASSERT_EQ(blob_bunch->RowCount(), 1000); - ASSERT_FALSE(blob_bunch->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count, blob_bunch->RowCount()); + ASSERT_EQ(row_count, 1000); auto blob_bunch2 = std::dynamic_pointer_cast(bunch[2]); ASSERT_EQ(blob_bunch2->Files().size(), 9); @@ -713,8 +789,8 @@ TEST_F(DataEvolutionSplitReadTest, TestComplexBlobBunchScenario3) { ASSERT_EQ(blob_bunch2->Files()[6], blob_entry17); ASSERT_EQ(blob_bunch2->Files()[7], blob_entry14); ASSERT_EQ(blob_bunch2->Files()[8], blob_entry18); - ASSERT_EQ(blob_bunch2->RowCount(), 1000); - ASSERT_FALSE(blob_bunch2->SequentialReadOptimize()); + ASSERT_OK_AND_ASSIGN(int64_t row_count2, blob_bunch2->RowCount()); + ASSERT_EQ(row_count2, 1000); } TEST_F(DataEvolutionSplitReadTest, TestDifferentRowIdRange) { @@ -804,8 +880,10 @@ TEST_F(DataEvolutionSplitReadTest, TestPlanSequentialVectorRead) { 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_OK_AND_ASSIGN(int64_t row_count, bunches[0]->RowCount()); + ASSERT_EQ(row_count, 10); + ASSERT_OK_AND_ASSIGN(int64_t row_count2, bunches[1]->RowCount()); + ASSERT_EQ(row_count2, 10); ASSERT_EQ(vector1->first_row_id, 103); ASSERT_EQ(vector1->row_count, 4); } @@ -1167,6 +1245,116 @@ TEST_F(DataEvolutionSplitReadTest, TestSplitWithMultipleBlobFilesPerGroup) { ASSERT_EQ(expected_metas, split_metas); } +TEST_F(DataEvolutionSplitReadTest, TestFileIndexSelectionOnlyPushesDownNormalGroups) { + auto dir = UniqueTestDirectory::Create("local"); + auto pool = GetDefaultPool(); + auto schema = DataField::ConvertDataFieldsToArrowSchema( + {DataField(0, arrow::field("f0", arrow::int32())), + DataField(1, BlobUtils::ToArrowField("b0")), + DataField(2, arrow::field("embedding", arrow::list(arrow::float32())))}); + const std::map options = {{Options::FILE_FORMAT, "parquet"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::FILE_INDEX_READ_ENABLED, "true"}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, + TableSchema::Create(0, schema, {}, {}, options)); + ReadContextBuilder builder(dir->Str()); + 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)); + DataEvolutionSplitRead split_read(/*path_factory=*/nullptr, context, pool, + /*executor=*/nullptr); + + // Use a real embedded bitmap index without creating any data files. + auto index_schema = arrow::schema({schema->field(0)}); + auto data = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(index_schema->fields()), + R"([[0],[1],[2],[3]])") + .ValueOrDie(); + ArrowSchema c_schema; + ASSERT_TRUE(arrow::ExportSchema(*index_schema, &c_schema).ok()); + BitmapFileIndex index({}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr writer, + index.CreateWriter(&c_schema, pool)); + ArrowArray c_array; + ASSERT_TRUE(arrow::ExportArray(*data, &c_array).ok()); + ASSERT_OK(writer->AddBatch(&c_array)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr payload, writer->SerializedBytes()); + FileIndexFormat::ColumnIndexes indexes; + indexes["f0"]["bitmap"] = payload; + auto normal = CreateNormalFile("normal.parquet", 100, 4, 20); + normal->write_cols = std::vector{"f0"}; + ASSERT_OK_AND_ASSIGN(normal->embedded_index, MockSearchFileIndex::Serialize(indexes, pool)); + + auto blob = CreateBlobFile("partial", 101, 2, 30, std::vector{"b0"}); + auto vector = CreateNormalFile("partial.vector.parquet", 101, 2, 30); + vector->write_cols = std::vector{"embedding"}; + auto same_range = CreateNormalFile("same.parquet", 100, 4, 10); + same_range->write_cols = std::vector{"b0"}; + auto different_range = CreateNormalFile("different.parquet", 101, 2, 10); + different_range->write_cols = std::vector{"b0"}; + auto match = PredicateBuilder::Equal(0, "f0", FieldType::INT, Literal(1)); + auto miss = PredicateBuilder::Equal(0, "f0", FieldType::INT, Literal(9)); + auto last_match = PredicateBuilder::Equal(0, "f0", FieldType::INT, Literal(3)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr sparse_match, + PredicateBuilder::Or({match, last_match})); + + for (const auto& other : {blob, vector, different_range}) { + ASSERT_OK_AND_ASSIGN( + std::optional selection, + split_read.SelectByFileIndex(match, {normal, other}, schema, std::nullopt, + /*data_file_path_factory=*/nullptr)); + ASSERT_FALSE(selection.has_value()); + + std::vector row_ranges = {Range(100, 102)}; + ASSERT_OK_AND_ASSIGN( + selection, split_read.SelectByFileIndex(match, {normal, other}, schema, row_ranges, + /*data_file_path_factory=*/nullptr)); + ASSERT_FALSE(selection.has_value()); + + // The normal file can still reject the entire group when its index has no matches. + ASSERT_OK_AND_ASSIGN( + selection, split_read.SelectByFileIndex(miss, {normal, other}, schema, std::nullopt, + /*data_file_path_factory=*/nullptr)); + ASSERT_TRUE(selection.has_value()); + ASSERT_TRUE(selection->IsEmpty()); + } + for (const auto& files : {std::vector>{normal}, + std::vector>{normal, same_range}}) { + ASSERT_OK_AND_ASSIGN(std::optional selection, + split_read.SelectByFileIndex(match, files, schema, std::nullopt, + /*data_file_path_factory=*/nullptr)); + ASSERT_EQ(selection, std::optional(RoaringBitmap32::From({1}))); + + // Row ranges use absolute IDs; the returned bitmap keeps file-local positions. + const std::vector, std::vector>> cases = { + {{Range(101, 101)}, {1}}, + {{Range(103, 103)}, {3}}, + {{Range(100, 100)}, {}}, + {{Range(102, 102)}, {}}, + {{Range(104, 105)}, {}}, + {{}, {}}, + {{Range(103, 103), Range(100, 101), Range(101, 102)}, {1, 3}}, + {{Range(103, 110), Range(99, 101)}, {1, 3}}}; + for (const auto& [row_ranges, expected] : cases) { + ASSERT_OK_AND_ASSIGN( + selection, split_read.SelectByFileIndex(sparse_match, files, schema, row_ranges, + /*data_file_path_factory=*/nullptr)); + ASSERT_EQ(selection, std::optional(RoaringBitmap32::From(expected))); + } + ASSERT_OK_AND_ASSIGN(selection, + split_read.SelectByFileIndex(/*predicate=*/nullptr, files, schema, + std::vector{Range(100, 102)}, + /*data_file_path_factory=*/nullptr)); + ASSERT_FALSE(selection.has_value()); + ASSERT_OK_AND_ASSIGN(selection, + split_read.SelectByFileIndex(/*predicate=*/nullptr, files, schema, + std::vector{Range(200, 201)}, + /*data_file_path_factory=*/nullptr)); + ASSERT_TRUE(selection.has_value()); + ASSERT_TRUE(selection->IsEmpty()); + } +} + namespace { std::shared_ptr MakeBitmapDv(const std::vector& deleted_positions) { RoaringBitmap32 bitmap; @@ -1188,6 +1376,62 @@ DeletionVector::Factory MakeSingleFileDvFactory(const std::string& expected_file } } // namespace +TEST_F(DataEvolutionSplitReadTest, TestExcludeDeletedPositionsKeepsBitmapSelection) { + const std::vector, std::vector>> cases = { + {{0, 1, 2, 3, 4, 5, 6, 7}, {0, 2, 4, 5, 6, 7}}, {{1, 3}, {}}, {{0, 2}, {0, 2}}, {{}, {}}}; + for (bool bitmap64 : {false, true}) { + std::shared_ptr dv = + bitmap64 ? std::shared_ptr(std::make_shared()) + : MakeBitmapDv({}); + ASSERT_OK(dv->Delete(1)); + ASSERT_OK(dv->Delete(3)); + if (bitmap64) { + ASSERT_OK(dv->Delete(1LL << 32)); + } + for (const auto& [selected, expected] : cases) { + ASSERT_OK_AND_ASSIGN(RoaringBitmap32 live, + DataEvolutionSplitRead::ExcludeDeletedPositions( + RoaringBitmap32::From(selected), dv)); + ASSERT_EQ(live, RoaringBitmap32::From(expected)); + } + } +} + +TEST_F(DataEvolutionSplitReadTest, TestGroupBitmapCombinesWithRowRangesAndDvInFileReader) { + auto dir = UniqueTestDirectory::Create("local"); + auto schema = DataField::ConvertDataFieldsToArrowSchema( + {DataField(0, arrow::field("f0", arrow::int32()))}); + const std::map options = {{Options::FILE_FORMAT, "parquet"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, + TableSchema::Create(0, schema, {}, {}, options)); + ReadContextBuilder builder(dir->Str()); + 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)); + DataEvolutionSplitRead split_read(/*path_factory=*/nullptr, context, GetDefaultPool(), + /*executor=*/nullptr); + split_read.file_indexes_evaluated_ = true; + split_read.group_selection_ = RoaringBitmap32::From({0, 1, 3}); + auto file = CreateNormalFile("normal.parquet", 100, 4, 20); + auto type = arrow::struct_(schema->fields()); + auto data = + arrow::ipc::internal::json::ArrayFromJSON(type, R"([[0],[1],[2],[3]])").ValueOrDie(); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + split_read.ApplyIndexAndDvReaderIfNeeded( + std::make_unique(data, type, 2), file, schema, schema, + /*predicate=*/nullptr, MakeSingleFileDvFactory(file->file_name, MakeBitmapDv({1})), + std::vector{Range(101, 103)}, /*data_file_path_factory=*/nullptr, + /*file_index_options=*/{}, /*index_scores=*/nullptr)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr result, + ReadResultCollector::CollectResult(std::move(reader))); + auto expected = arrow::ipc::internal::json::ArrayFromJSON(type, R"([[3]])").ValueOrDie(); + ASSERT_TRUE(result); + ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected))) << result->ToString(); +} + TEST_F(DataEvolutionSplitReadTest, TestReadGroupDeletionVector) { // the anchor is the oldest normal file; blob files never anchor the deletion vector std::vector> group = { @@ -1221,17 +1465,17 @@ TEST_F(DataEvolutionSplitReadTest, TestReadGroupDeletionVector) { group, DeletionVector::Factory())); ASSERT_FALSE(group_dv.has_value()); - // a group of nothing but dedicated storage files has no anchor, so no deletion vector can be - // keyed by it. The split's factory is non-null because another group of the same split - // carries one, which must not fail this group's read. + // Match Java: a split carrying deletion files requires a normal anchor in every read group. std::vector> anchorless_group = { CreateDataFileMeta("blob1.blob", /*first_row_id=*/100, /*row_count=*/100, /*max_seq=*/5), CreateDataFileMeta("blob2.blob", /*first_row_id=*/100, /*row_count=*/100, /*max_seq=*/6), CreateDataFileMeta("vector-store.vector.parquet", /*first_row_id=*/100, /*row_count=*/100, /*max_seq=*/7)}; - ASSERT_OK_AND_ASSIGN(group_dv, - DataEvolutionSplitRead::ReadGroupDeletionVector( - anchorless_group, MakeSingleFileDvFactory("anchor.parquet", dv))); + ASSERT_NOK_WITH_MSG(DataEvolutionSplitRead::ReadGroupDeletionVector( + anchorless_group, MakeSingleFileDvFactory("anchor.parquet", dv)), + "normal anchor file"); + ASSERT_OK_AND_ASSIGN(group_dv, DataEvolutionSplitRead::ReadGroupDeletionVector( + anchorless_group, DeletionVector::Factory())); ASSERT_FALSE(group_dv.has_value()); } @@ -1245,8 +1489,9 @@ TEST_F(DataEvolutionSplitReadTest, TestCreateGroupDvFactoryShiftsPositions) { // deletion vector positions are anchor-relative: positions {10, 45} are row ids {110, 145} DataEvolutionSplitRead::GroupDeletionVector group_dv{MakeBitmapDv({10, 45}), Range(100, 199)}; - ASSERT_OK_AND_ASSIGN(DeletionVector::Factory factory, - DataEvolutionSplitRead::CreateGroupDvFactory(group, group_dv)); + ASSERT_OK_AND_ASSIGN( + DeletionVector::Factory factory, + DataEvolutionSplitRead::CreateGroupDvFactory(group, group_dv, /*row_ranges=*/std::nullopt)); ASSERT_TRUE(factory); // the anchor file reads the deletion vector unshifted @@ -1285,7 +1530,8 @@ TEST_F(DataEvolutionSplitReadTest, TestCreateGroupDvFactoryShiftsPositions) { ASSERT_FALSE(unknown_dv); ASSERT_OK_AND_ASSIGN(DeletionVector::Factory null_factory, - DataEvolutionSplitRead::CreateGroupDvFactory(group, std::nullopt)); + DataEvolutionSplitRead::CreateGroupDvFactory(group, std::nullopt, + /*row_ranges=*/std::nullopt)); ASSERT_FALSE(null_factory); } @@ -1297,8 +1543,72 @@ TEST_F(DataEvolutionSplitReadTest, TestCreateGroupDvFactoryRejectsFileOutsideAnc std::vector> group = {anchor, blob}; DataEvolutionSplitRead::GroupDeletionVector group_dv{MakeBitmapDv({10}), Range(100, 199)}; - ASSERT_NOK_WITH_MSG(DataEvolutionSplitRead::CreateGroupDvFactory(group, group_dv), - "should contain row id range"); + ASSERT_NOK_WITH_MSG( + DataEvolutionSplitRead::CreateGroupDvFactory(group, group_dv, /*row_ranges=*/std::nullopt), + "should contain selected row id range"); +} + +TEST_F(DataEvolutionSplitReadTest, TestGroupDvChecksOnlySelectedRanges) { + auto anchor = CreateNormalFile("anchor.parquet", 100, 100, 10); + auto prefix = CreateDataFileMeta("prefix.blob", 90, 30, 20); + auto suffix = CreateDataFileMeta("suffix.blob", 180, 30, 20); + std::vector> group = {anchor, prefix, suffix}; + for (bool bitmap64 : {false, true}) { + std::shared_ptr dv = + bitmap64 ? std::shared_ptr(std::make_shared()) + : MakeBitmapDv({}); + ASSERT_OK(dv->Delete(5)); + ASSERT_OK(dv->Delete(85)); + DataEvolutionSplitRead::GroupDeletionVector group_dv{dv, Range(100, 199)}; + ASSERT_OK_AND_ASSIGN( + DeletionVector::Factory factory, + DataEvolutionSplitRead::CreateGroupDvFactory( + group, group_dv, std::vector{Range(180, 199), Range(100, 119)})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr prefix_dv, factory("prefix.blob")); + ASSERT_OK_AND_ASSIGN(std::shared_ptr suffix_dv, factory("suffix.blob")); + ASSERT_TRUE(prefix_dv->IsDeleted(15).value()); + ASSERT_TRUE(suffix_dv->IsDeleted(5).value()); + ASSERT_NOK_WITH_MSG(DataEvolutionSplitRead::CreateGroupDvFactory( + group, group_dv, std::vector{Range(99, 105)}), + "should contain selected row id range"); + ASSERT_NOK_WITH_MSG(DataEvolutionSplitRead::CreateGroupDvFactory( + group, group_dv, std::vector{Range(195, 200)}), + "should contain selected row id range"); + ASSERT_OK(DataEvolutionSplitRead::CreateGroupDvFactory( + group, group_dv, std::vector{Range(300, 310)})); + + // A format can return an entire batch with only some rows selected. The DV wrapper + // must never probe unselected prefix/suffix positions outside the anchor range. + auto type = arrow::struct_({arrow::field("f0", arrow::int32())}); + auto data = + arrow::ipc::internal::json::ArrayFromJSON(type, R"([[0],[1],[2],[3]])").ValueOrDie(); + auto small_file = CreateDataFileMeta("small.blob", 99, 4, 20); + ASSERT_OK_AND_ASSIGN(factory, + DataEvolutionSplitRead::CreateGroupDvFactory( + {small_file}, group_dv, std::vector{Range(100, 101)})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr shifted_dv, factory("small.blob")); + RoaringBitmap32 selection = RoaringBitmap32::From({1, 2}); + ApplyDeletionVectorBatchReader reader( + std::make_unique(data, type, selection, 4), shifted_dv); + ASSERT_OK_AND_ASSIGN(std::shared_ptr result, + ReadResultCollector::CollectResult(&reader)); + ASSERT_TRUE(result); + ASSERT_EQ(result->length(), 2); + } +} + +TEST_F(DataEvolutionSplitReadTest, TestPartialBlobBunchUsesNormalRange) { + auto blob = CreateBlobFile("blob", 103, 2, 20, std::optional>({"b0"})); + DataEvolutionSplitRead::BlobBunch bunch(Range(100, 109), /*has_row_ids_selection=*/false); + ASSERT_OK(bunch.Add(blob)); + ASSERT_OK_AND_ASSIGN(int64_t row_count, bunch.RowCount()); + ASSERT_EQ(row_count, 10); + ASSERT_OK_AND_ASSIGN(Range logical_range, bunch.LogicalRange()); + ASSERT_EQ(logical_range, Range(100, 109)); + auto tail = CreateBlobFile("tail", 108, 2, 20, std::optional>({"b0"})); + ASSERT_OK(bunch.Add(tail)); + ASSERT_OK_AND_ASSIGN(int64_t row_count2, bunch.RowCount()); + ASSERT_EQ(row_count2, 10); } TEST_F(DataEvolutionSplitReadTest, TestExcludeDeletedRowIds) { @@ -1322,6 +1632,47 @@ TEST_F(DataEvolutionSplitReadTest, TestExcludeDeletedRowIds) { ASSERT_TRUE(remaining.empty()); } +TEST_F(DataEvolutionSplitReadTest, TestExcludeDeletedRowIdsRestrictsDeletionsToSelectedGaps) { + for (bool bitmap64 : {false, true}) { + std::shared_ptr dv = + bitmap64 ? std::shared_ptr(std::make_shared()) + : MakeBitmapDv({}); + for (int64_t position = 0; position < 200; position += 2) { + ASSERT_OK(dv->Delete(position)); + } + DataEvolutionSplitRead::GroupDeletionVector group_dv{dv, Range(100, 599)}; + + // Selected gaps exceed the DV cardinality, but all deletions precede them. + ASSERT_OK_AND_ASSIGN( + std::vector remaining, + DataEvolutionSplitRead::ExcludeDeletedRowIds({Range(300, 499)}, group_dv)); + ASSERT_EQ(remaining, (std::vector{Range(300, 499)})); + + // Deletions between selected ranges and after them must also be ignored. + ASSERT_OK(dv->Delete(450)); + ASSERT_OK_AND_ASSIGN(remaining, DataEvolutionSplitRead::ExcludeDeletedRowIds( + {Range(100, 110), Range(300, 499)}, group_dv)); + ASSERT_EQ(remaining, + (std::vector{Range(101, 101), Range(103, 103), Range(105, 105), + Range(107, 107), Range(109, 109), Range(300, 499)})); + + // Deleting row id 450 (anchor position 350) splits the second selected gap. + ASSERT_OK(dv->Delete(350)); + ASSERT_OK_AND_ASSIGN(remaining, DataEvolutionSplitRead::ExcludeDeletedRowIds( + {Range(100, 110), Range(400, 499)}, group_dv)); + ASSERT_EQ(remaining, (std::vector{Range(101, 101), Range(103, 103), Range(105, 105), + Range(107, 107), Range(109, 109), Range(400, 449), + Range(451, 499)})); + + // Keep both endpoints of a selected range subject to deletion filtering. + ASSERT_OK(dv->Delete(200)); + ASSERT_OK(dv->Delete(399)); + ASSERT_OK_AND_ASSIGN( + remaining, DataEvolutionSplitRead::ExcludeDeletedRowIds({Range(300, 499)}, group_dv)); + ASSERT_EQ(remaining, (std::vector{Range(301, 449), Range(451, 498)})); + } +} + TEST_F(DataEvolutionSplitReadTest, Bitmap64GroupWindowWithHighOffset) { const int64_t offset = 1LL << 32; auto anchor = CreateNormalFile("anchor.parquet", 100, offset + 30, 10); @@ -1331,8 +1682,8 @@ TEST_F(DataEvolutionSplitReadTest, Bitmap64GroupWindowWithHighOffset) { ASSERT_OK(dv->Delete(p)); } DataEvolutionSplitRead::GroupDeletionVector group_dv{dv, Range(100, 100 + offset + 29)}; - ASSERT_OK_AND_ASSIGN(auto factory, - DataEvolutionSplitRead::CreateGroupDvFactory({anchor, blob}, group_dv)); + ASSERT_OK_AND_ASSIGN(auto factory, DataEvolutionSplitRead::CreateGroupDvFactory( + {anchor, blob}, group_dv, /*row_ranges=*/std::nullopt)); ASSERT_OK_AND_ASSIGN(auto view, factory("blob0.blob")); ASSERT_TRUE(view); ASSERT_TRUE(view->IsDeleted(0).value()); @@ -1342,6 +1693,11 @@ TEST_F(DataEvolutionSplitReadTest, Bitmap64GroupWindowWithHighOffset) { std::vector positions; ASSERT_OK(view->ForEachDeletedPosition([&](int64_t p) { positions.push_back(p); })); ASSERT_EQ(positions, (std::vector{0, 5})); + ASSERT_OK_AND_ASSIGN(std::vector live, DataEvolutionSplitRead::ExcludeDeletedRowIds( + {group_dv.anchor_range}, group_dv)); + ASSERT_EQ(live, (std::vector{ + Range(100, 100 + offset - 2), Range(101 + offset, 104 + offset), + Range(106 + offset, 119 + offset), Range(121 + offset, 129 + offset)})); } } // namespace paimon::test diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 4cbfda4df..c721658f5 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -2004,6 +2004,132 @@ TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateWithDeletionVectors) /*predicate=*/nullptr, /*row_ranges=*/{Range(0, 1)})); } +TEST_P(BlobTableInteTest, TestDataEvolutionPartialBlobBackfillWithDeletionVectors) { + arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), + BlobUtils::ToArrowField("b0", /*nullable=*/true)}; + CreateTable(fields, {}, + {{Options::FILE_FORMAT, GetParam()}, + {Options::FILE_SYSTEM, "local"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::DELETION_VECTORS_ENABLED, "true"}, + {Options::DELETION_VECTOR_BITMAP64, "true"}, + {Options::READ_BATCH_SIZE, "2"}}); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + auto normal_type = arrow::struct_({fields[0]}); + auto normal = arrow::ipc::internal::json::ArrayFromJSON( + normal_type, R"([[0],[1],[2],[3],[4],[5],[6],[7],[8],[9]])") + .ValueOrDie(); + ASSERT_OK_AND_ASSIGN(std::vector> base, + WriteArray(table_path, {}, {"f0"}, {normal})); + SetFirstRowId(100, base); + ASSERT_OK(Commit(table_path, base)); + auto blobs = MakeBlobUpdateArray(fields[1], {"b3", "b4", "b5"}); + ASSERT_OK_AND_ASSIGN(std::vector> backfill, + WriteArray(table_path, {}, {"b0"}, {blobs})); + SetFirstRowId(103, backfill); + ASSERT_OK(Commit(table_path, backfill)); + + auto full = checked_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), + R"([[0,null],[1,null],[2,null],[3,"b3"],[4,"b4"], + [5,"b5"],[6,null],[7,null],[8,null],[9,null]])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"f0", "b0"}, full)); + auto blob_only = checked_pointer_cast( + arrow::StructArray::Make({full->field(1)}, {fields[1]}).ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"b0"}, blob_only)); + + ASSERT_OK_AND_ASSIGN(std::string anchor, + DeletionVectorTestHelper::RetrieveAnchorFileName(base)); + // Delete rows in the physical blob file, the leading gap, and the trailing gap. + ASSERT_OK(CommitDeletionVectors(table_path, base[0], {{anchor, {1, 3, 7, 9}}})); + arrow::FieldVector read_fields = {fields[0], fields[1], SpecialFields::RowId().ArrowField()}; + auto expected = checked_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON( + arrow::struct_(read_fields), + R"([[0,null,100],[2,null,102],[4,"b4",104],[5,"b5",105],[6,null,106],[8,null,108]])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"f0", "b0", "_ROW_ID"}, expected)); + blob_only = checked_pointer_cast( + arrow::StructArray::Make({expected->field(1)}, {fields[1]}).ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"b0"}, blob_only)); + auto selected = checked_pointer_cast(blob_only->Slice(1, 4)); + ASSERT_OK(ScanAndRead(table_path, {"b0"}, selected, nullptr, {Range(102, 106)})); + auto gap_only = checked_pointer_cast(blob_only->Slice(4, 2)); + ASSERT_OK(ScanAndRead(table_path, {"b0"}, gap_only, nullptr, {Range(106, 108)})); + ASSERT_OK(ScanAndRead(table_path, {"b0"}, nullptr, nullptr, {Range(101, 101)})); +} + +TEST_P(BlobTableInteTest, TestDataEvolutionDvValidatesSelectedBlobRange) { + arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), + BlobUtils::ToArrowField("b0", /*nullable=*/true)}; + CreateTable(fields, {}, + {{Options::FILE_FORMAT, GetParam()}, + {Options::FILE_SYSTEM, "local"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::DELETION_VECTORS_ENABLED, "true"}}); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + auto normal = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0]}), + R"([[0],[1],[2],[3]])") + .ValueOrDie(); + ASSERT_OK_AND_ASSIGN(std::vector> base, + WriteArray(table_path, {}, {"f0"}, {normal})); + SetFirstRowId(100, base); + ASSERT_OK(Commit(table_path, base)); + auto blobs = MakeBlobUpdateArray(fields[1], {"b0", "b1", "b2", "b3", "b4", "b5", "b6", "b7"}); + ASSERT_OK_AND_ASSIGN(std::vector> backfill, + WriteArray(table_path, {}, {"b0"}, {blobs})); + SetFirstRowId(98, backfill); + ASSERT_OK(Commit(table_path, backfill)); + ASSERT_OK_AND_ASSIGN(std::string anchor, + DeletionVectorTestHelper::RetrieveAnchorFileName(base)); + ASSERT_OK(CommitDeletionVectors(table_path, base[0], {{anchor, {1}}})); + auto expected = checked_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), + R"([[0,"b2"],[2,"b4"],[3,"b5"]])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"f0", "b0"}, expected, nullptr, {Range(100, 103)})); + ASSERT_NOK_WITH_MSG(ScanAndRead(table_path, {"f0", "b0"}, nullptr), + "should contain selected row id range"); + ASSERT_NOK_WITH_MSG(ScanAndRead(table_path, {"f0", "b0"}, expected, nullptr, {Range(99, 100)}), + "should contain selected row id range"); +} + +TEST_P(BlobTableInteTest, TestDataEvolutionRejectsAnchorlessGroupWithDeletionVectors) { + arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), + BlobUtils::ToArrowField("b0", /*nullable=*/true)}; + CreateTable(fields, {}, + {{Options::FILE_FORMAT, GetParam()}, + {Options::FILE_SYSTEM, "local"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::DELETION_VECTORS_ENABLED, "true"}}); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + auto normal = + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0]}), R"([[0],[1],[2]])") + .ValueOrDie(); + ASSERT_OK_AND_ASSIGN(std::vector> base, + WriteArray(table_path, {}, {"f0"}, {normal})); + SetFirstRowId(100, base); + ASSERT_OK(Commit(table_path, base)); + auto blobs = MakeBlobUpdateArray(fields[1], {"b0", "b1"}); + ASSERT_OK_AND_ASSIGN(std::vector> dedicated, + WriteArray(table_path, {}, {"b0"}, {blobs})); + SetFirstRowId(200, dedicated); + ASSERT_OK(Commit(table_path, dedicated)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr rows, + ScanAndReadResult(table_path, {"b0", "_ROW_ID"})); + ASSERT_TRUE(rows); + ASSERT_EQ(rows->length(), 5); + ASSERT_OK_AND_ASSIGN(std::string anchor, + DeletionVectorTestHelper::RetrieveAnchorFileName(base)); + ASSERT_OK(CommitDeletionVectors(table_path, base[0], {{anchor, {1}}})); + // The same split now carries a DV. Match Java by requiring a normal anchor in each group. + ASSERT_NOK_WITH_MSG(ScanAndReadResult(table_path, {"b0", "_ROW_ID"}), "normal anchor file"); +} + TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateCompactedLayers) { arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")}; @@ -2350,11 +2476,10 @@ TEST_P(BlobTableInteTest, TestBlobSentinelValueInBaseLayerDegradesToNull) { ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); } -TEST_P(BlobTableInteTest, TestUserSuppliedInternalPlaceholderOptionsIgnored) { +TEST_P(BlobTableInteTest, TestSingleBlobLayerPlaceholderBecomesNull) { // blob.internal.emit-placeholder-sentinel is reserved for the data-evolution fallback read - // path. Set in the user table options it must be ignored: were it honored, a read of a - // placeholder entry that no older layer can resolve would return the sentinel bytes as the - // blob value instead of failing. + // path. A single layer also uses fallback: an unresolved placeholder becomes null, and the + // internal sentinel must never be exposed as the blob value. arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0", /*nullable=*/true)}; @@ -2376,9 +2501,11 @@ TEST_P(BlobTableInteTest, TestUserSuppliedInternalPlaceholderOptionsIgnored) { WriteArray(table_path, {}, schema->field_names(), {src_array})); ASSERT_OK(Commit(table_path, commit_msgs)); - // the user-supplied key was stripped, so the single-layer read rejects the placeholder - ASSERT_NOK_WITH_MSG(ScanAndReadResult(table_path, schema->field_names()), - "contains a placeholder entry"); + auto expected_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), + R"([[1, "a", null], [2, "b", "blob_b"]])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); } TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyFirstCommitFailsWithoutFirstRowId) { diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index 395de88d0..475ae9a63 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -107,10 +107,14 @@ class DataEvolutionTableTest : public ::testing::Test, Result>> WriteArrays( const std::string& table_path, const std::map& partition, const std::vector& write_cols, - const std::vector>& write_arrays) const { + const std::vector>& write_arrays, + const std::map& write_options = {}) const { // write WriteContextBuilder write_builder(table_path, "commit_user_1"); write_builder.WithWriteSchema(write_cols); + for (const auto& [key, value] : write_options) { + write_builder.AddOption(key, value); + } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr write_context, write_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(auto file_store_write, FileStoreWrite::Create(std::move(write_context))); @@ -132,8 +136,9 @@ class DataEvolutionTableTest : public ::testing::Test, Result>> WriteArray( const std::string& table_path, const std::map& partition, const std::vector& write_cols, - const std::shared_ptr& write_array) const { - return WriteArrays(table_path, partition, write_cols, {write_array}); + const std::shared_ptr& write_array, + const std::map& write_options = {}) const { + return WriteArrays(table_path, partition, write_cols, {write_array}, write_options); } Result>> WriteArray( @@ -213,7 +218,8 @@ class DataEvolutionTableTest : public ::testing::Test, /// free to serve f2 from the full-row write instead. Result>> WriteAndCommitGroup( const std::string& table_path, int64_t first_row_id, const std::vector& f0_values, - const std::map& partition = {}) const { + const std::map& partition = {}, + const std::map& partial_write_options = {}) const { auto partition_f1 = partition.find("f1"); std::string base_json = "["; std::string f2_json = "["; @@ -245,8 +251,9 @@ class DataEvolutionTableTest : public ::testing::Test, auto f2_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f2_fields), f2_json) .ValueOrDie()); - PAIMON_ASSIGN_OR_RAISE(std::vector> f2_msgs, - WriteArray(table_path, partition, {"f2"}, f2_array)); + PAIMON_ASSIGN_OR_RAISE( + std::vector> f2_msgs, + WriteArray(table_path, partition, {"f2"}, f2_array, partial_write_options)); SetFirstRowId(first_row_id, f2_msgs); PAIMON_RETURN_NOT_OK(Commit(table_path, f2_msgs)); @@ -2859,15 +2866,12 @@ TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) { expected_array)); } { - // The old file's f2 index does not contain 102, but the newer file owns f2. + // The newer file owns f2; its bitmap selects the same row in every merged reader. auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2", FieldType::INT, Literal(102)); auto expected_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([ - ["Lily", 2, 102, 2.1], - ["Alice", 4, 104, 3.1], - ["Bob", 6, 106, 4.1], - ["David", 8, 108, 5.1] + ["Lily", 2, 102, 2.1] ])") .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(), @@ -2926,15 +2930,13 @@ TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) { { // test row id with predicate std::vector row_ranges = {Range(0l, 2l)}; - // A merged group keeps all selected row ids to preserve column alignment. + // Intersect the group's shared bitmap with row ranges while preserving column alignment. auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2", FieldType::INT, Literal(106)); CheckScanResult(table_path, /*predicate=*/predicate, /*row_ranges=*/row_ranges, /*expected_first_row_ids=*/{0, 0}, /*expected_row_counts=*/{4, 4}); auto expected_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([ - ["Lily", 2, 102, 2.1], - ["Alice", 4, 104, 3.1], ["Bob", 6, 106, 4.1] ])") .ValueOrDie()); @@ -3657,6 +3659,143 @@ TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectors) { /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 2)})); } +TEST_P(DataEvolutionDvTableTest, TestIndexSkipsBeforeReadingDeletionVector) { + CreateDataEvolutionTable( + /*deletion_vectors_enabled=*/true, + {{"file-index.bitmap.columns", "f2"}, {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1B"}}); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + auto data = checked_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), + R"([[0,"a0","x0"],[1,"a1","x1"], + [2,"a2","x2"],[3,"a3","x3"]])") + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(std::vector> messages, + WriteArray(table_path, {}, {"f0", "f1", "f2"}, data)); + SetFirstRowId(100, messages); + ASSERT_OK(Commit(table_path, messages)); + ASSERT_OK_AND_ASSIGN(std::vector> merged_messages, + WriteAndCommitGroup(table_path, 200, {0, 1, 2, 3})); + ASSERT_OK_AND_ASSIGN(std::vector anchors, PlannedAnchorFileNames(table_path)); + ASSERT_EQ(anchors.size(), 2); + ASSERT_OK(CommitDeletionVectors(table_path, messages[0], {{anchors[0], {2}}})); + ASSERT_OK(CommitDeletionVectors(table_path, merged_messages[0], {{anchors[1], {2}}})); + + { + // Indexes keep positions 1 and 2 in both groups; their DVs remove position 2. + auto predicate = PredicateBuilder::In( + 2, "f2", FieldType::STRING, + {Literal(FieldType::STRING, "x1", 2), Literal(FieldType::STRING, "x2", 2), + Literal(FieldType::STRING, "y1", 2), Literal(FieldType::STRING, "y2", 2)}); + auto expected = checked_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), + R"([[1,"a1","x1"],[1,"a1","y1"]])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2"}, expected, predicate, {}, + /*check_scan_plan_when_empty_result=*/true, + /*apply_predicate_to_scan=*/false)); + } + + ASSERT_OK_AND_ASSIGN(std::vector> splits, PlanSplits(table_path)); + std::set dv_paths; + for (const auto& split : splits) { + auto data_split = checked_pointer_cast(split); + for (const auto& deletion_file : data_split->DeletionFiles()) { + if (deletion_file) { + dv_paths.insert(deletion_file->path); + } + } + } + ASSERT_FALSE(dv_paths.empty()); + for (const auto& path : dv_paths) { + ASSERT_OK(dir_->GetFileSystem()->Delete(path, /*recursive=*/false)); + } + auto missing = PredicateBuilder::Equal(2, "f2", FieldType::STRING, + Literal(FieldType::STRING, "missing", 7)); + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2"}, nullptr, missing, {}, + /*check_scan_plan_when_empty_result=*/false, + /*apply_predicate_to_scan=*/false)); + auto single_match = + PredicateBuilder::Equal(2, "f2", FieldType::STRING, Literal(FieldType::STRING, "x1", 2)); + auto merged_match = + PredicateBuilder::Equal(2, "f2", FieldType::STRING, Literal(FieldType::STRING, "y1", 2)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr match, + PredicateBuilder::Or({single_match, merged_match})); + // Both single-file and merged groups are rejected by rowRanges intersected with the bitmap. + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2"}, nullptr, match, + {Range(100, 100), Range(200, 200)}, + /*check_scan_plan_when_empty_result=*/false, + /*apply_predicate_to_scan=*/false)); + // A selected matching row still needs its DV; do not hide an actual missing sidecar. + ASSERT_NOK(ScanAndRead(table_path, {"f0", "f1", "f2"}, nullptr, match, {Range(101, 101)}, + /*check_scan_plan_when_empty_result=*/false, + /*apply_predicate_to_scan=*/false)); +} + +TEST_P(DataEvolutionDvTableTest, TestSharedBitmapSelectionComposesWithDeletionVectors) { + CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true, + {{"file-index.bitmap.columns", "f0,f2"}, + {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1B"}, + {Options::READ_BATCH_SIZE, "2"}}); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + ASSERT_OK_AND_ASSIGN(std::vector> first, + WriteAndCommitGroup(table_path, 100, {0, 1, 2, 3}, {}, + {{"file-index.bitmap.columns", "f2"}})); + ASSERT_OK_AND_ASSIGN(std::vector> second, + WriteAndCommitGroup(table_path, 200, {0, 1, 2, 3}, {}, + {{"file-index.bitmap.columns", "f2"}})); + ASSERT_OK_AND_ASSIGN(std::vector anchors, PlannedAnchorFileNames(table_path)); + ASSERT_EQ(anchors.size(), 2); + ASSERT_OK(CommitDeletionVectors(table_path, first[0], {{anchors[0], {2}}})); + ASSERT_OK(CommitDeletionVectors(table_path, second[0], {{anchors[1], {1}}})); + auto one = + PredicateBuilder::Equal(2, "f2", FieldType::STRING, Literal(FieldType::STRING, "y1", 2)); + auto two = + PredicateBuilder::Equal(2, "f2", FieldType::STRING, Literal(FieldType::STRING, "y2", 2)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr predicate, PredicateBuilder::Or({one, two})); + arrow::FieldVector fields = fields_; + fields.push_back(SpecialFields::RowId().ArrowField()); + auto expected = checked_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), + R"([[1,"a1","y1",101],[2,"a2","y2",202]])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID"}, expected, predicate, {}, + /*check_scan_plan_when_empty_result=*/true, + /*apply_predicate_to_scan=*/false)); + + // The newer f2 index must win over its stale copy in the anchor file. + auto newer = + PredicateBuilder::Equal(2, "f2", FieldType::STRING, Literal(FieldType::STRING, "y1", 2)); + auto expected_one = checked_pointer_cast(expected->Slice(0, 1)); + auto small = PredicateBuilder::Equal(0, "f0", FieldType::INT, Literal(1)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr combined, + PredicateBuilder::And({small, predicate})); + // f0 and f2 are won by different files. Intersect both indexes before applying the DV. + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID"}, expected_one, combined, {}, + /*check_scan_plan_when_empty_result=*/true, + /*apply_predicate_to_scan=*/false)); + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID"}, expected_one, newer, {}, + /*check_scan_plan_when_empty_result=*/true, + /*apply_predicate_to_scan=*/false)); + + ASSERT_OK_AND_ASSIGN(std::vector> splits, PlanSplits(table_path)); + for (const auto& split : splits) { + auto data_split = checked_pointer_cast(split); + for (const auto& file : data_split->DataFiles()) { + if (file->first_row_id == 200) { + ASSERT_OK(dir_->GetFileSystem()->Delete( + PathUtil::JoinPath(data_split->BucketPath(), file->file_name), false)); + } + } + } + // The second group's only bitmap hit is deleted: skip it before opening any data file. + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID"}, expected_one, one, {}, + /*check_scan_plan_when_empty_result=*/true, + /*apply_predicate_to_scan=*/false)); + ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2"}, nullptr, one, {Range(200, 203)}, + /*check_scan_plan_when_empty_result=*/false, + /*apply_predicate_to_scan=*/false)); +} + TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectorsAcrossReadBatches) { // the 12 rows below span several read batches, so the deletion vector empties a whole // batch of every file of the group: each file reader then skips that batch entirely and