This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new a965a7d4 fix(read): correct data evolution selection and blob backfill
reads (#472)
a965a7d4 is described below
commit a965a7d4c70d63d84bad1a8dfd619fce68282c6c
Author: lxy <[email protected]>
AuthorDate: Wed Oct 28 19:25:52 2026 +0800
fix(read): correct data evolution selection and blob backfill reads (#472)
---
include/paimon/utils/roaring_bitmap32.h | 3 +
include/paimon/utils/roaring_bitmap64.h | 4 +
.../common/reader/blob_fallback_batch_reader.cpp | 9 +-
.../common/reader/blob_fallback_batch_reader.h | 6 +-
.../reader/blob_fallback_batch_reader_test.cpp | 13 +-
src/paimon/common/utils/roaring_bitmap32.cpp | 15 +-
src/paimon/common/utils/roaring_bitmap32_test.cpp | 27 ++
src/paimon/common/utils/roaring_bitmap64.cpp | 47 ++-
src/paimon/common/utils/roaring_bitmap64_test.cpp | 36 ++
.../apply_deletion_vector_batch_reader.h | 10 +-
.../core/operation/data_evolution_split_read.cpp | 368 +++++++++++-----
.../core/operation/data_evolution_split_read.h | 76 ++--
.../operation/data_evolution_split_read_test.cpp | 466 ++++++++++++++++++---
test/inte/blob_table_inte_test.cpp | 141 ++++++-
test/inte/data_evolution_table_test.cpp | 167 +++++++-
15 files changed, 1132 insertions(+), 256 deletions(-)
diff --git a/include/paimon/utils/roaring_bitmap32.h
b/include/paimon/utils/roaring_bitmap32.h
index 29d45809..85f6a87e 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 289622b0..931f1b52 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 830b0df4..a5e23855 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<std::unique_ptr<BlobFallbackBatchReader>>
BlobFallbackBatchReader::Create
std::vector<std::vector<Segment>>&& sequence_groups,
const std::shared_ptr<arrow::Schema>& read_schema, int32_t read_batch_size,
const std::shared_ptr<arrow::MemoryPool>& 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<BatchReader::ReadBatch>
BlobFallbackBatchReader::NextBatch() {
return Status::Invalid("blob fallback batch reader is closed");
}
std::vector<std::vector<Chunk>> 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 e15667b4..36bceb99 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<std::unique_ptr<BlobFallbackBatchReader>> 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 7eac84b4..9f67737d 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<std::vector<BlobFallbackBatchReader::Segment>> 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<std::vector<BlobFallbackBatchReader::Segment>> 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<std::vector<BlobFallbackBatchReader::Segment>> groups;
diff --git a/src/paimon/common/utils/roaring_bitmap32.cpp
b/src/paimon/common/utils/roaring_bitmap32.cpp
index 8a90952a..7644c550 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<roaring::Roaring::const_iterator*>(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 239381f2..b4334a9c 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<int32_t> values;
+ for (; iter != bitmap.End(); ++iter) {
+ values.push_back(*iter);
+ }
+ ASSERT_EQ(values, (std::vector<int32_t>{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 510ef4e8..0bd4f8c1 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<roaring::Roaring64Map::const_iterator*>(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<Bitmap64Iterator*>(iter));
}
roaring::Roaring64Map& GetRoaringBitmap(void* bitmap) {
return *(static_cast<roaring::Roaring64Map*>(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<roaring::Roaring64Map::const_iterator*>(iterator_);
+ delete static_cast<Bitmap64Iterator*>(iterator_);
}
}
@@ -57,8 +67,13 @@ RoaringBitmap64::Iterator&
RoaringBitmap64::Iterator::operator=(
if (&other == this) {
return *this;
}
+ if (!other.iterator_) {
+ delete static_cast<Bitmap64Iterator*>(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<roaring::Roaring64Map::const_iterator*>(iterator_);
+ delete static_cast<Bitmap64Iterator*>(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 6a51d29a..e77aba03 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<int64_t> expected = {0, 1LL << 32,
RoaringBitmap64::MAX_VALUE};
+ auto bitmap = RoaringBitmap64::From(expected);
+ auto iter = bitmap.Begin();
+ std::vector<int64_t> 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 c428a079..704247e9 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<RoaringBitmap32> Filter(int32_t batch_size) const {
+ Result<RoaringBitmap32> 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 5fea0f21..87dc99cc 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<int64_t> 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<Range> 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<DataFileMeta>& 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<FileStorePathFactory>& path_factory,
const std::shared_ptr<InternalReadContext>& context,
@@ -346,7 +337,7 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateBlobViewReade
PAIMON_ASSIGN_OR_RAISE(std::optional<GroupDeletionVector> 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<std::unique_ptr<FileBatchReader>> raw_file_readers,
CreateRawFileReaders(split_impl->Partition(), data_files,
blob_view_schema,
@@ -431,26 +422,35 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::InnerCreateReader(
std::vector<std::unique_ptr<BatchReader>> sub_readers;
for (const std::vector<std::shared_ptr<DataFileMeta>>& 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<RoaringBitmap32> 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<GroupDeletionVector> 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<std::unique_ptr<FileBatchReader>> 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<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::InnerCreateReader(
} else {
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<BatchReader> 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<std::shared_ptr<Predicate>>
DataEvolutionSplitRead::CreatePushDownPredica
return PredicateUtils::CreatePickedFieldFilter(predicate,
picked_field_name_to_idx);
}
-Result<bool> DataEvolutionSplitRead::SkipByFileIndex(
+Result<std::optional<RoaringBitmap32>>
DataEvolutionSplitRead::SelectByFileIndex(
const std::shared_ptr<Predicate>& predicate,
const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ if (row_ranges) {
+ std::vector<Range> 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>(RoaringBitmap32());
+ }
+ }
if (!options_.FileIndexReadEnabled() || !predicate) {
- return false;
+ return std::optional<RoaringBitmap32>();
+ }
+
+ // 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<FieldMappingBuilder> field_mapping_builder,
- FieldMappingBuilder::Create(raw_read_schema_,
context_->GetPartitionKeys(), predicate));
+ FieldMappingBuilder::Create(read_schema, context_->GetPartitionKeys(),
predicate));
+ std::optional<RoaringBitmap32> selection;
std::set<int32_t> 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<bool> 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>(RoaringBitmap32());
+ }
+ if (!can_push_down_selection) {
+ continue;
+ }
+ if (auto* bitmap_index =
dynamic_cast<BitmapIndexResult*>(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<RoaringBitmap32> range_selection,
+ files[0]->ToFileSelection(row_ranges));
+ if (range_selection) {
+ *selection &= *range_selection;
}
}
- return false;
+ return selection;
}
Result<std::unique_ptr<FileBatchReader>>
DataEvolutionSplitRead::ApplyIndexAndDvReaderIfNeeded(
@@ -570,7 +622,7 @@ Result<std::unique_ptr<FileBatchReader>>
DataEvolutionSplitRead::ApplyIndexAndDv
const std::map<std::string, std::string>& file_index_options,
std::vector<float>* index_scores) const {
std::shared_ptr<FileIndexResult> 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<std::unique_ptr<FileBatchReader>>
DataEvolutionSplitRead::ApplyIndexAndDv
return std::unique_ptr<FileBatchReader>();
}
}
- const RoaringBitmap32* index_selection = nullptr;
+ const RoaringBitmap32* index_selection = group_selection_ ?
&*group_selection_ : nullptr;
if (auto* bitmap_index =
dynamic_cast<BitmapIndexResult*>(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<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>
fields_files;
std::map<int32_t, std::shared_ptr<DataEvolutionSplitRead::BlobBunch>>
blob_bunch_map;
- int64_t row_count = -1;
+ std::optional<Range> 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<DataEvolutionSplitRead::BlobBunch>(
- 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<DataEvolutionSplitRead::DataBunch>(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<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateUnionReader(
TryPlanSequentialVectorRead(need_merge_files,
file_to_field_ids));
auto split_read =
std::make_shared<DataEvolutionSplitRead>(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<LazyConcatBatchReader::Source> sources;
sources.push_back(LazyConcatBatchReader::Source{
@@ -970,16 +1026,23 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
const std::optional<GroupDeletionVector>& group_dv,
const std::shared_ptr<arrow::Schema>& read_schema) const {
assert(!fields_files.empty());
- int64_t row_count = fields_files[0]->RowCount();
- PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id,
fields_files[0]->Files()[0]->NonNullFirstRowId());
+ auto bunch_first_row_id = [](const std::shared_ptr<FieldBunch>& bunch) ->
Result<int64_t> {
+ if (auto blob_bunch = std::dynamic_pointer_cast<BlobBunch>(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<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
std::vector<LazyConcatBatchReader::Source> sources;
auto split_read =
std::make_shared<DataEvolutionSplitRead>(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<std::vector<Range>> selected = row_ranges;
if (row_ranges) {
@@ -1077,14 +1142,13 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
}
file_batch_readers[file_idx] =
std::make_unique<LazyConcatBatchReader>(std::move(sources), arrow_pool_);
- } else if (blob_bunch && !blob_bunch->SequentialReadOptimize()) {
- // blob files span multiple max sequence number layers:
placeholder entries of
- // newer layers must fall back row by row to older layers
+ } 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<std::unique_ptr<FileBatchReader>> file_readers,
@@ -1097,7 +1161,6 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateFieldUnionRea
} else {
auto raw_readers =
ObjectUtils::MoveVector<std::unique_ptr<BatchReader>>(
std::move(file_readers));
- // Concat multiple blob files that map to the same data
file.
file_batch_readers[file_idx] =
std::make_unique<ConcatBatchReader>(std::move(raw_readers), arrow_pool_);
}
@@ -1148,20 +1211,28 @@ std::vector<Range> SelectedRangesInRange(int64_t from,
int64_t to,
Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateBlobFallbackReader(
const BinaryRow& partition, const
std::vector<std::shared_ptr<DataFileMeta>>& files,
- const std::shared_ptr<arrow::Schema>& file_read_schema,
+ const Range& logical_range, const std::shared_ptr<arrow::Schema>&
file_read_schema,
const std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
const DeletionVector::Factory& group_dv_factory,
const std::optional<GroupDeletionVector>& group_dv) const {
- int64_t union_first_row_id = std::numeric_limits<int64_t>::max();
- int64_t union_last_row_id = std::numeric_limits<int64_t>::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<int64_t,
std::shared_ptr<DataFileMeta>>;
std::map<int64_t, std::vector<FileWithFirstRowId>, 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<std::string, std::string> blob_format_options = {
@@ -1173,21 +1244,24 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateBlobFallbackR
[](const FileWithFirstRowId& f1, const
FileWithFirstRowId& f2) {
return f1.first < f2.first;
});
- std::vector<BlobFallbackBatchReader::Segment> 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<int64_t>::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<Range> gap_selected_ranges =
- SelectedRangesInRange(next_row_id, first_row_id - 1,
row_ranges);
+ previous_file_end = first_row_id + file->row_count;
+ }
+ std::vector<BlobFallbackBatchReader::Segment> 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<Range> 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<std::unique_ptr<BatchReader>>
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<std::vector<Range>> 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<std::unique_ptr<FileBatchReader>> 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<std::unique_ptr<BatchReader>>
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<Range> gap_selected_ranges =
@@ -1227,6 +1304,10 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateBlobFallbackR
BlobFallbackBatchReader::Segment{nullptr,
std::move(gap_selected_ranges)});
}
}
+ if (segments.empty()) {
+ return
std::make_unique<ConcatBatchReader>(std::vector<std::unique_ptr<BatchReader>>(),
+ 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<GroupDeletionVector>();
}
- // 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<DataFileMeta>& file) {
- return DataEvolutionUtils::IsNormalFile(file->file_name);
- })) {
- return std::optional<GroupDeletionVector>();
- }
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> anchor,
DataEvolutionUtils::RetrieveAnchorFile(group));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DeletionVector> deletion_vector,
@@ -1356,7 +1429,8 @@ DataEvolutionSplitRead::ReadGroupDeletionVector(
Result<DeletionVector::Factory> DataEvolutionSplitRead::CreateGroupDvFactory(
const std::vector<std::shared_ptr<DataFileMeta>>& group,
- const std::optional<GroupDeletionVector>& group_dv) {
+ const std::optional<GroupDeletionVector>& group_dv,
+ const std::optional<std::vector<Range>>& row_ranges) {
if (!group_dv) {
return DeletionVector::Factory();
}
@@ -1365,11 +1439,18 @@ Result<DeletionVector::Factory>
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<Range> selected_ranges =
+ row_ranges ? Range::And({Range(first_row_id, last_row_id)},
+ Range::SortAndMergeOverlap(*row_ranges,
/*adjacent=*/true))
+ : std::vector<Range>{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<DeletionVector::Factory>
DataEvolutionSplitRead::CreateGroupDvFactory(
});
}
+Result<RoaringBitmap32> DataEvolutionSplitRead::ExcludeDeletedPositions(
+ RoaringBitmap32 selection, const std::shared_ptr<DeletionVector>&
deletion_vector) {
+ if (selection.IsEmpty() || deletion_vector->IsEmpty()) {
+ return selection;
+ }
+ if (auto* bitmap_dv =
dynamic_cast<BitmapDeletionVector*>(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<uint32_t>::max()) {
+ deleted.Add(static_cast<int32_t>(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<uint32_t>(*iter)));
+ if (!is_deleted) {
+ live.Add(*iter);
+ }
+ }
+ return live;
+}
+
Result<std::vector<Range>> DataEvolutionSplitRead::ExcludeDeletedRowIds(
const std::vector<Range>& ranges, const GroupDeletionVector& group_dv) {
+ if (ranges.empty()) {
+ return std::vector<Range>();
+ }
+ 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<Range> 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<Range> remaining;
for (const auto& range : ranges) {
std::optional<int64_t> 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 e0406495..b7610dc1 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 <cstdint>
#include <functional>
-#include <limits>
#include <map>
#include <memory>
#include <string>
@@ -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<int64_t> RowCount() const = 0;
virtual const std::vector<std::shared_ptr<DataFileMeta>>& Files()
const = 0;
};
@@ -113,7 +113,7 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
public:
explicit DataBunch(const std::shared_ptr<DataFileMeta>& data_file)
: data_files_({data_file}) {}
- int64_t RowCount() const override {
+ Result<int64_t> RowCount() const override {
return data_files_[0]->row_count;
}
const std::vector<std::shared_ptr<DataFileMeta>>& Files() const
override {
@@ -129,7 +129,7 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
public:
VectorBunch(std::vector<std::shared_ptr<DataFileMeta>> files, int64_t
row_count)
: files_(std::move(files)), row_count_(row_count) {}
- int64_t RowCount() const override {
+ Result<int64_t> RowCount() const override {
return row_count_;
}
const std::vector<std::shared_ptr<DataFileMeta>>& 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<Range>& 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<int64_t> RowCount() const override;
+ Result<Range> LogicalRange() const;
const std::vector<std::shared_ptr<DataFileMeta>>& Files() const
override {
return files_;
}
Status Add(const std::shared_ptr<DataFileMeta>& 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<Range> expected_row_range_;
bool has_row_ids_selection_ = false;
- int64_t union_first_row_id_ = std::numeric_limits<int64_t>::max();
- /// Exclusive end of the union row id range covered so far.
- int64_t union_end_row_id_ = std::numeric_limits<int64_t>::min();
- /// Per max sequence number: exclusive end of the last added range, to
reject
- /// overlapping files within one layer.
- std::map<int64_t, int64_t> sequence_group_end_;
std::vector<Range> ranges_;
std::vector<std::shared_ptr<DataFileMeta>> files_;
};
@@ -201,12 +185,15 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
const std::shared_ptr<Predicate>& predicate,
const std::shared_ptr<arrow::Schema>& 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<bool> 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<std::optional<RoaringBitmap32>> SelectByFileIndex(
const std::shared_ptr<Predicate>& predicate,
const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& 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<DeletionVector::Factory> CreateGroupDvFactory(
const std::vector<std::shared_ptr<DataFileMeta>>& group,
- const std::optional<GroupDeletionVector>& group_dv);
+ const std::optional<GroupDeletionVector>& group_dv,
+ const std::optional<std::vector<Range>>& 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<RoaringBitmap32> ExcludeDeletedPositions(
+ RoaringBitmap32 selection, const std::shared_ptr<DeletionVector>&
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<std::vector<Range>> ExcludeDeletedRowIds(const
std::vector<Range>& ranges,
const
GroupDeletionVector& group_dv);
@@ -318,11 +310,19 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
/// emitting the same surviving rows.
Result<std::unique_ptr<BatchReader>> CreateBlobFallbackReader(
const BinaryRow& partition, const
std::vector<std::shared_ptr<DataFileMeta>>& files,
- const std::shared_ptr<arrow::Schema>& file_read_schema,
+ const Range& logical_range, const std::shared_ptr<arrow::Schema>&
file_read_schema,
const std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
const DeletionVector::Factory& group_dv_factory,
const std::optional<GroupDeletionVector>& group_dv) const;
+
+ /// 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<RoaringBitmap32> 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 4bd0e0c2..a68dfdf8 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"diff_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"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<std::vector<std::string>>({"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<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"}};
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+ TableSchema::Create(0, schema, {}, {}, options));
+ ReadContextBuilder builder(dir->Str());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+ DataEvolutionSplitRead split_read(/*path_factory=*/nullptr, context,
GetDefaultPool(),
+ /*executor=*/nullptr);
+ auto blob =
+ CreateBlobFile("blob", 100, 10, 1,
std::optional<std::vector<std::string>>({"blob_col"}));
+ auto overlap =
+ CreateBlobFile("overlap", 105, 5, 1,
std::optional<std::vector<std::string>>({"blob_col"}));
+ auto create_reader = [&](const std::vector<std::shared_ptr<DataFileMeta>>&
files,
+ const Range& range,
+ const std::optional<std::vector<Range>>&
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>{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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<std::vector<std::string>>({"blob_col"}));
auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
- 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<DataEvolutionSplitRead::BlobBunch>(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<std::shared_ptr<DataFileMeta>>{main});
ASSERT_EQ(bunches[1]->Files(),
(std::vector<std::shared_ptr<DataFileMeta>>{vector0, vector1,
vector2}));
- ASSERT_EQ(bunches[0]->RowCount(), 10);
- ASSERT_EQ(bunches[1]->RowCount(), 10);
+ ASSERT_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<std::string, std::string> 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<TableSchema> table_schema,
+ TableSchema::Create(0, schema, {}, {}, options));
+ ReadContextBuilder builder(dir->Str());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+ 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<FileIndexWriter> 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<Bytes> payload,
writer->SerializedBytes());
+ FileIndexFormat::ColumnIndexes indexes;
+ indexes["f0"]["bitmap"] = payload;
+ auto normal = CreateNormalFile("normal.parquet", 100, 4, 20);
+ normal->write_cols = std::vector<std::string>{"f0"};
+ ASSERT_OK_AND_ASSIGN(normal->embedded_index,
MockSearchFileIndex::Serialize(indexes, pool));
+
+ auto blob = CreateBlobFile("partial", 101, 2, 30,
std::vector<std::string>{"b0"});
+ auto vector = CreateNormalFile("partial.vector.parquet", 101, 2, 30);
+ vector->write_cols = std::vector<std::string>{"embedding"};
+ auto same_range = CreateNormalFile("same.parquet", 100, 4, 10);
+ same_range->write_cols = std::vector<std::string>{"b0"};
+ auto different_range = CreateNormalFile("different.parquet", 101, 2, 10);
+ different_range->write_cols = std::vector<std::string>{"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<Predicate> sparse_match,
+ PredicateBuilder::Or({match, last_match}));
+
+ for (const auto& other : {blob, vector, different_range}) {
+ ASSERT_OK_AND_ASSIGN(
+ std::optional<RoaringBitmap32> selection,
+ split_read.SelectByFileIndex(match, {normal, other}, schema,
std::nullopt,
+ /*data_file_path_factory=*/nullptr));
+ ASSERT_FALSE(selection.has_value());
+
+ std::vector<Range> 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<std::shared_ptr<DataFileMeta>>{normal},
+
std::vector<std::shared_ptr<DataFileMeta>>{normal, same_range}}) {
+ ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap32> selection,
+ split_read.SelectByFileIndex(match, files,
schema, std::nullopt,
+
/*data_file_path_factory=*/nullptr));
+ ASSERT_EQ(selection,
std::optional<RoaringBitmap32>(RoaringBitmap32::From({1})));
+
+ // Row ranges use absolute IDs; the returned bitmap keeps file-local
positions.
+ const std::vector<std::pair<std::vector<Range>, std::vector<int32_t>>>
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>(RoaringBitmap32::From(expected)));
+ }
+ ASSERT_OK_AND_ASSIGN(selection,
+
split_read.SelectByFileIndex(/*predicate=*/nullptr, files, schema,
+
std::vector<Range>{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>{Range(200, 201)},
+
/*data_file_path_factory=*/nullptr));
+ ASSERT_TRUE(selection.has_value());
+ ASSERT_TRUE(selection->IsEmpty());
+ }
+}
+
namespace {
std::shared_ptr<DeletionVector> MakeBitmapDv(const std::vector<int64_t>&
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::pair<std::vector<int32_t>, std::vector<int32_t>>>
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<DeletionVector> dv =
+ bitmap64 ?
std::shared_ptr<DeletionVector>(std::make_shared<Bitmap64DeletionVector>())
+ : 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<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"}};
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+ TableSchema::Create(0, schema, {}, {}, options));
+ ReadContextBuilder builder(dir->Str());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+ InternalReadContext::Create(read_context,
table_schema, options));
+ 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<FileBatchReader> reader,
+ split_read.ApplyIndexAndDvReaderIfNeeded(
+ std::make_unique<MockFileBatchReader>(data, type, 2), file,
schema, schema,
+ /*predicate=*/nullptr, MakeSingleFileDvFactory(file->file_name,
MakeBitmapDv({1})),
+ std::vector<Range>{Range(101, 103)},
/*data_file_path_factory=*/nullptr,
+ /*file_index_options=*/{}, /*index_scores=*/nullptr));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> 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<std::shared_ptr<DataFileMeta>> 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<std::shared_ptr<DataFileMeta>> 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<std::shared_ptr<DataFileMeta>> 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<std::shared_ptr<DataFileMeta>> group = {anchor, prefix,
suffix};
+ for (bool bitmap64 : {false, true}) {
+ std::shared_ptr<DeletionVector> dv =
+ bitmap64 ?
std::shared_ptr<DeletionVector>(std::make_shared<Bitmap64DeletionVector>())
+ : 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>{Range(180, 199),
Range(100, 119)}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DeletionVector> prefix_dv,
factory("prefix.blob"));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DeletionVector> 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>{Range(99,
105)}),
+ "should contain selected row id range");
+ ASSERT_NOK_WITH_MSG(DataEvolutionSplitRead::CreateGroupDvFactory(
+ group, group_dv, std::vector<Range>{Range(195,
200)}),
+ "should contain selected row id range");
+ ASSERT_OK(DataEvolutionSplitRead::CreateGroupDvFactory(
+ group, group_dv, std::vector<Range>{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>{Range(100, 101)}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DeletionVector> shifted_dv,
factory("small.blob"));
+ RoaringBitmap32 selection = RoaringBitmap32::From({1, 2});
+ ApplyDeletionVectorBatchReader reader(
+ std::make_unique<MockFileBatchReader>(data, type, selection, 4),
shifted_dv);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> 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<std::vector<std::string>>({"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<std::vector<std::string>>({"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<DeletionVector> dv =
+ bitmap64 ?
std::shared_ptr<DeletionVector>(std::make_shared<Bitmap64DeletionVector>())
+ : 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<Range> remaining,
+ DataEvolutionSplitRead::ExcludeDeletedRowIds({Range(300, 499)},
group_dv));
+ ASSERT_EQ(remaining, (std::vector<Range>{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>{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>{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>{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<int64_t> positions;
ASSERT_OK(view->ForEachDeletedPosition([&](int64_t p) {
positions.push_back(p); }));
ASSERT_EQ(positions, (std::vector<int64_t>{0, 5}));
+ ASSERT_OK_AND_ASSIGN(std::vector<Range> live,
DataEvolutionSplitRead::ExcludeDeletedRowIds(
+ {group_dv.anchor_range},
group_dv));
+ ASSERT_EQ(live, (std::vector<Range>{
+ 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 4cbfda4d..c721658f 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<std::shared_ptr<CommitMessage>> 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<std::shared_ptr<CommitMessage>> backfill,
+ WriteArray(table_path, {}, {"b0"}, {blobs}));
+ SetFirstRowId(103, backfill);
+ ASSERT_OK(Commit(table_path, backfill));
+
+ auto full = checked_pointer_cast<arrow::StructArray>(
+ 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>(
+ 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::StructArray>(
+ 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>(
+ arrow::StructArray::Make({expected->field(1)},
{fields[1]}).ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"b0"}, blob_only));
+ auto selected =
checked_pointer_cast<arrow::StructArray>(blob_only->Slice(1, 4));
+ ASSERT_OK(ScanAndRead(table_path, {"b0"}, selected, nullptr, {Range(102,
106)}));
+ auto gap_only =
checked_pointer_cast<arrow::StructArray>(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<std::shared_ptr<CommitMessage>> 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<std::shared_ptr<CommitMessage>> 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::StructArray>(
+ 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<std::shared_ptr<CommitMessage>> 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<std::shared_ptr<CommitMessage>> dedicated,
+ WriteArray(table_path, {}, {"b0"}, {blobs}));
+ SetFirstRowId(200, dedicated);
+ ASSERT_OK(Commit(table_path, dedicated));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> 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::StructArray>(
+ 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 395de88d..475ae9a6 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<std::vector<std::shared_ptr<CommitMessage>>> WriteArrays(
const std::string& table_path, const std::map<std::string,
std::string>& partition,
const std::vector<std::string>& write_cols,
- const std::vector<std::shared_ptr<arrow::Array>>& write_arrays) const {
+ const std::vector<std::shared_ptr<arrow::Array>>& write_arrays,
+ const std::map<std::string, std::string>& 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<WriteContext> 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<std::vector<std::shared_ptr<CommitMessage>>> WriteArray(
const std::string& table_path, const std::map<std::string,
std::string>& partition,
const std::vector<std::string>& write_cols,
- const std::shared_ptr<arrow::Array>& write_array) const {
- return WriteArrays(table_path, partition, write_cols, {write_array});
+ const std::shared_ptr<arrow::Array>& write_array,
+ const std::map<std::string, std::string>& write_options = {}) const {
+ return WriteArrays(table_path, partition, write_cols, {write_array},
write_options);
}
Result<std::vector<std::shared_ptr<CommitMessage>>> WriteArray(
@@ -213,7 +218,8 @@ class DataEvolutionTableTest : public ::testing::Test,
/// free to serve f2 from the full-row write instead.
Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommitGroup(
const std::string& table_path, int64_t first_row_id, const
std::vector<int32_t>& f0_values,
- const std::map<std::string, std::string>& partition = {}) const {
+ const std::map<std::string, std::string>& partition = {},
+ const std::map<std::string, std::string>& 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::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f2_fields), f2_json)
.ValueOrDie());
- PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>>
f2_msgs,
- WriteArray(table_path, partition, {"f2"},
f2_array));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::shared_ptr<CommitMessage>> 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::StructArray>(
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<Range> 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::StructArray>(
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::StructArray>(
+ 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<std::shared_ptr<CommitMessage>> messages,
+ WriteArray(table_path, {}, {"f0", "f1", "f2"}, data));
+ SetFirstRowId(100, messages);
+ ASSERT_OK(Commit(table_path, messages));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>>
merged_messages,
+ WriteAndCommitGroup(table_path, 200, {0, 1, 2, 3}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> 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::StructArray>(
+ 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<std::shared_ptr<Split>> splits,
PlanSplits(table_path));
+ std::set<std::string> dv_paths;
+ for (const auto& split : splits) {
+ auto data_split = checked_pointer_cast<DataSplitImpl>(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<Predicate> 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<std::shared_ptr<CommitMessage>> first,
+ WriteAndCommitGroup(table_path, 100, {0, 1, 2, 3}, {},
+ {{"file-index.bitmap.columns",
"f2"}}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> second,
+ WriteAndCommitGroup(table_path, 200, {0, 1, 2, 3}, {},
+ {{"file-index.bitmap.columns",
"f2"}}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> 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> predicate,
PredicateBuilder::Or({one, two}));
+ arrow::FieldVector fields = fields_;
+ fields.push_back(SpecialFields::RowId().ArrowField());
+ auto expected = checked_pointer_cast<arrow::StructArray>(
+ 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<arrow::StructArray>(expected->Slice(0, 1));
+ auto small = PredicateBuilder::Equal(0, "f0", FieldType::INT, Literal(1));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> 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<std::shared_ptr<Split>> splits,
PlanSplits(table_path));
+ for (const auto& split : splits) {
+ auto data_split = checked_pointer_cast<DataSplitImpl>(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
