This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new cea0caaf [core] Read sparse bitmap matches from data evolution ROW
sidecars (#1173)
cea0caaf is described below
commit cea0caafd638f11203357a92c69d3629e880a848
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 20:17:25 2026 +0800
[core] Read sparse bitmap matches from data evolution ROW sidecars (#1173)
---
crates/paimon/src/table/data_evolution_reader.rs | 152 +++++-
crates/paimon/src/table/data_file_reader.rs | 28 +-
crates/paimon/src/table/row_sidecar.rs | 5 +-
.../tests/data_evolution_bitmap_merge_test.rs | 577 +++++++++++++++++++++
.../tests/data_evolution_row_sidecar_test.rs | 145 +++++-
.../tests/data_evolution_search_sidecar_test.rs | 440 ++++++++++++++++
.../tests/data_evolution_variant_sidecar_test.rs | 457 ++++++++++++++++
7 files changed, 1758 insertions(+), 46 deletions(-)
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 77ac3f90..da0f950c 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -477,7 +477,10 @@ impl DataEvolutionReader {
let push_down_raw_predicates = !self.predicates.is_empty()
&& self.row_id_index.is_none()
&& filter_before_blob_resolution;
- let read_raw_file_index = push_down_raw_predicates
+ // A bitmap selection preserves physical positions, including when
+ // `_ROW_ID` needs attaching before the residual predicate runs.
+ let read_file_index = !self.predicates.is_empty()
+ && filter_before_blob_resolution
&&
CoreOptions::new(&self.table_options).file_index_read_enabled();
// A managed BLOB file fetches payloads as its batch is decoded.
// Keep batches small and restrict predicate-free file selections
@@ -508,7 +511,9 @@ impl DataEvolutionReader {
if remaining == Some(0) {
break;
}
- let row_ranges = split.row_ranges().map(|r| r.to_vec());
+ let row_ranges = split
+ .row_ranges()
+ .map(|ranges|
crate::table::merge_row_ranges(ranges.to_vec()));
// A chunk may span several disjoint row-id groups while one
// group still needs column-wise merging. Process each aligned
// group independently; treating the whole chunk as one merge
@@ -541,10 +546,8 @@ impl DataEvolutionReader {
let has_row_id = file_meta.first_row_id.is_some();
let requested_row_ranges = if has_row_id {
row_ranges.clone() } else { None };
let mut effective_row_ranges =
requested_row_ranges.clone();
- if read_raw_file_index {
- // Independent files can apply a physical row
selection
- // before decoding. Column-merge groups
cannot: an older
- // file may hold values still needed by a
newer file.
+ let mut has_bitmap_selection = false;
+ if read_file_index {
match evaluate_file_index(
&self.file_io,
split.bucket_path(),
@@ -560,6 +563,7 @@ impl DataEvolutionReader {
&selection,
file_meta.row_count,
)? {
+ has_bitmap_selection =
selection.len() < file_meta.row_count as u64;
let base =
file_meta.first_row_id.unwrap_or(0);
let index_ranges = local_ranges
.into_iter()
@@ -578,6 +582,18 @@ impl DataEvolutionReader {
FileIndexResult::Remain => {}
}
}
+ // Java applies DV to a selective bitmap before
+ // choosing ROW. DV alone and LIMIT cannot trigger
it.
+ let sidecar_row_ranges = if has_bitmap_selection {
+ selected_absolute_row_ranges_for_file(
+ file_meta.first_row_id.unwrap_or(0),
+ file_meta.row_count,
+ effective_row_ranges.as_deref(),
+ deletion_vector.as_deref(),
+ )?
+ } else {
+ effective_row_ranges.clone()
+ };
if self.predicates.is_empty() {
if let Some(left) = remaining {
let selected =
selected_absolute_row_ranges_for_file(
@@ -628,7 +644,7 @@ impl DataEvolutionReader {
data_fields,
deletion_vector,
effective_row_ranges,
- requested_row_ranges,
+ sidecar_row_ranges,
)?;
while remaining != Some(0) {
let Some(batch) = stream.next().await else {
break };
@@ -659,6 +675,35 @@ impl DataEvolutionReader {
}
} else {
let prepared_group = PreparedMergeGroup::new(&files)?;
+ let mut indexed_row_ranges = row_ranges.clone();
+ let mut has_bitmap_selection = false;
+ if read_file_index {
+ match self.evaluate_group_file_indexes(&split,
&prepared_group).await? {
+ FileIndexResult::Skip => continue,
+ FileIndexResult::Selection(selection) => {
+ if let Some(local_ranges) =
file_index_selection_to_local_ranges(
+ &selection,
+ prepared_group.logical_row_count,
+ )? {
+ has_bitmap_selection = selection.len()
< prepared_group.logical_row_count as u64;
+ let index_ranges =
local_ranges.into_iter().map(|range| {
+ RowRange::new(
+ prepared_group.first_row_id +
range.from(),
+ prepared_group.first_row_id +
range.to(),
+ )
+ }).collect::<Vec<_>>();
+ indexed_row_ranges = Some(match
indexed_row_ranges {
+ Some(ref ranges) =>
intersect_local_ranges(ranges, &index_ranges),
+ None => index_ranges,
+ });
+ }
+ }
+ FileIndexResult::Remain => {}
+ }
+ }
+ if
indexed_row_ranges.as_ref().is_some_and(Vec::is_empty) {
+ continue;
+ }
let anchor_deletion_vector =
read_anchor_deletion_vector(
&self.file_io,
&split,
@@ -668,11 +713,19 @@ impl DataEvolutionReader {
let mut selected_ranges =
selected_absolute_row_ranges_for_file(
prepared_group.first_row_id,
prepared_group.logical_row_count,
- row_ranges.as_deref(),
+ indexed_row_ranges.as_deref(),
anchor_deletion_vector
.as_ref()
.map(|ctx| ctx.deletion_vector.as_ref()),
)?;
+ // Every provider shares the DV-aware bitmap. Without
+ // a selective bitmap, only caller ranges choose ROW.
+ // LIMIT never changes sidecar eligibility.
+ let sidecar_row_ranges = if has_bitmap_selection {
+ selected_ranges.clone()
+ } else {
+ indexed_row_ranges.clone()
+ };
let effective_row_ranges = if
self.predicates.is_empty() {
if let Some(left) = remaining {
selected_ranges = Some(prefix_row_ranges(
@@ -683,10 +736,10 @@ impl DataEvolutionReader {
));
selected_ranges.clone()
} else {
- row_ranges.clone()
+ indexed_row_ranges.clone()
}
} else {
- row_ranges.clone()
+ indexed_row_ranges
};
let expected_output_rows = match
selected_ranges.as_ref() {
Some(ranges) => ranges.iter().map(|r| r.count() as
usize).sum(),
@@ -711,7 +764,7 @@ impl DataEvolutionReader {
&split,
&prepared_group,
effective_row_ranges,
- row_ranges.clone(),
+ sidecar_row_ranges,
expected_output_rows,
anchor_deletion_vector,
)?;
@@ -944,6 +997,83 @@ impl DataEvolutionReader {
.await
}
+ /// Evaluate only the fields each normal file wins in the merged row.
+ ///
+ /// The group is newest first. A newer writer claims its fields even when
+ /// it has no index, so an obsolete value can never reject a live row.
+ async fn evaluate_group_file_indexes(
+ &self,
+ split: &DataSplit,
+ group: &PreparedMergeGroup,
+ ) -> crate::Result<FileIndexResult> {
+ // Dedicated sources may cover only part of the normal files' range.
+ // Like Java, they prevent a shared positional bitmap selection.
+ let can_select = group.files.iter().all(|file| {
+ !is_blob_or_video_file_name(&file.file_name)
+ && !is_vector_store_file_name(&file.file_name)
+ });
+ let mut claimed = HashSet::new();
+ let mut result = FileIndexResult::Remain;
+ for file in &group.files {
+ if is_blob_or_video_file_name(&file.file_name)
+ || is_vector_store_file_name(&file.file_name)
+ {
+ continue;
+ }
+ let (data_fields, _) = raw_file_physical_fields(
+ &self.schema_manager,
+ self.table_schema_id,
+ &self.table_fields,
+ file,
+ )
+ .await?;
+ let data_fields =
data_fields.as_deref().unwrap_or(&self.table_fields);
+ let winning_ids = data_fields
+ .iter()
+ .filter_map(|field|
claimed.insert(field.id()).then_some(field.id()))
+ .collect::<HashSet<_>>();
+ let mapping = self
+ .table_fields
+ .iter()
+ .enumerate()
+ .map(|(index, field)| {
+ // Partition values come from the split. A partially
written
+ // ROW can have several providers for its parent nullness.
+ (winning_ids.contains(&field.id())
+ && !matches!(field.data_type(), DataType::Row(_))
+ && !self.partition_fields.iter().any(|p| p.id() ==
field.id()))
+ .then_some(index)
+ })
+ .collect::<Vec<_>>();
+ let predicates = self
+ .predicates
+ .iter()
+ .filter_map(|predicate|
predicate.project_field_index_inclusive(&mapping))
+ .collect::<Vec<_>>();
+ let file_result = evaluate_file_index(
+ &self.file_io,
+ split.bucket_path(),
+ file,
+ &self.table_fields,
+ data_fields,
+ &predicates,
+ )
+ .await?;
+ if !file_result.remain() {
+ return Ok(FileIndexResult::Skip);
+ }
+ if can_select {
+ // Each file contributes a necessary condition, so intersect
+ // candidates rather than independently filtering its stream.
+ result = result.and(file_result);
+ if !result.remain() {
+ return Ok(FileIndexResult::Skip);
+ }
+ }
+ }
+ Ok(result)
+ }
+
/// Merge multiple logical sources column-wise for data evolution.
///
/// Normal partial-column files remain one source per file. Rolling `.blob`
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 841e347e..4ea4c9bd 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -361,7 +361,7 @@ impl DataFileReader {
)
});
split_file_offset += file_meta.row_count;
- let sidecar_ranges = split_ranges.clone();
+ let mut has_bitmap_selection = false;
let selected_ranges = match file_index_result {
FileIndexResult::Remain => split_ranges,
FileIndexResult::Skip => Some(Vec::new()),
@@ -370,12 +370,15 @@ impl DataFileReader {
&selection,
file_meta.row_count,
)? {
- Some(index_ranges) => Some(match split_ranges {
- Some(split_ranges) => {
- intersect_sorted_ranges(&index_ranges,
&split_ranges)
- }
- None => index_ranges,
- }),
+ Some(index_ranges) => {
+ has_bitmap_selection = selection.len() <
file_meta.row_count as u64;
+ Some(match split_ranges {
+ Some(split_ranges) => {
+
intersect_sorted_ranges(&index_ranges, &split_ranges)
+ }
+ None => index_ranges,
+ })
+ }
None => split_ranges,
}
}
@@ -385,6 +388,13 @@ impl DataFileReader {
dv.as_deref(),
selected_ranges.as_deref(),
);
+ // Like Java, only a selective bitmap allows DV to narrow
+ // sidecar eligibility. DV alone cannot trigger a ROW read.
+ let sidecar_ranges = if has_bitmap_selection {
+ row_selection.clone()
+ } else {
+ selected_ranges
+ };
let mut stream =
reader.read_single_file_stream_with_selection(
&split,
@@ -480,8 +490,8 @@ impl DataFileReader {
)
}
- /// Physical selection may be narrowed by file indexes or a LIMIT, while
- /// Java selects the ROW sidecar from the caller's original row ranges.
+ /// The caller supplies sidecar candidates separately from physical ranges:
+ /// a selective bitmap includes DV filtering, while LIMIT never affects it.
pub(super) fn read_single_file_stream_with_requested_ranges(
&self,
split: &DataSplit,
diff --git a/crates/paimon/src/table/row_sidecar.rs
b/crates/paimon/src/table/row_sidecar.rs
index 9e7dd87d..624ca6ef 100644
--- a/crates/paimon/src/table/row_sidecar.rs
+++ b/crates/paimon/src/table/row_sidecar.rs
@@ -213,8 +213,9 @@ fn selected_sidecar<'a>(
{
return Ok(None);
}
- // The caller clips, sorts and merges the requested ranges into local
- // positions before intersecting them with indexes and deletion vectors.
+ // The caller intersects requested ranges with bitmap indexes in local
+ // positions, including DV only when a selective bitmap exists. LIMIT
+ // never affects this decision.
let selected: i64 = ranges.iter().map(RowRange::count).sum();
Ok((selected > 0
&& selected < file.row_count
diff --git a/crates/paimon/tests/data_evolution_bitmap_merge_test.rs
b/crates/paimon/tests/data_evolution_bitmap_merge_test.rs
new file mode 100644
index 00000000..1118cddd
--- /dev/null
+++ b/crates/paimon/tests/data_evolution_bitmap_merge_test.rs
@@ -0,0 +1,577 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{Array, ArrayRef, BinaryArray, Int32Array, Int64Array,
RecordBatch, StructArray};
+use arrow_buffer::NullBuffer;
+use arrow_schema::DataType as ArrowDataType;
+use arrow_select::concat::concat_batches;
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::arrow::variant_arrow_type;
+use paimon::spec::{
+ DataType, Datum, IntType, Predicate, PredicateBuilder, Schema,
SchemaChange, TableSchema,
+ VariantType,
+};
+use paimon::table::{CommitMessage, RowRange, Table};
+use paimon::variant::GenericVariant;
+use std::sync::Arc;
+
+async fn table(name: &str) -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("a", DataType::Int(IntType::new()))
+ .column("b", DataType::Int(IntType::new()))
+ .column("payload", DataType::Variant(VariantType::new()))
+ .option("file.format", "parquet")
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .option("global-index.enabled", "false")
+ .option("file-index.bitmap.columns", "id,a,b")
+ .option("file-index.in-manifest-threshold", "1mb")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(
+ &format!("memory:/bitmap_merge_{name}"),
+ TableSchema::new(0, &schema),
+ );
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ table
+}
+
+fn values(start: i32, count: i32, updated: bool) ->
Vec<Option<GenericVariant>> {
+ (start..start + count)
+ .map(|id| {
+ let json = match id.rem_euclid(5) {
+ 1 => return None,
+ 2 => "null".to_string(),
+ 3 => format!(r#""{}""#, if updated { "updated" } else {
"scalar" }),
+ 4 => format!("[{},null,true]", id + if updated { 1000 } else {
0 }),
+ _ => format!(
+ r#"{{"x":{},"nested":{{"items":[true,null,"中"]}}}}"#,
+ id + if updated { 1000 } else { 0 }
+ ),
+ };
+ Some(GenericVariant::parse_json(&json).unwrap())
+ })
+ .collect()
+}
+
+fn variants(values: &[Option<GenericVariant>]) -> StructArray {
+ let ArrowDataType::Struct(fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ StructArray::new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(
+ values
+ .iter()
+ .map(|v| v.as_ref().map_or(&[][..],
GenericVariant::value)),
+ )),
+ Arc::new(BinaryArray::from_iter_values(
+ values
+ .iter()
+ .map(|v| v.as_ref().map_or(&[][..],
GenericVariant::metadata)),
+ )),
+ ],
+ Some(NullBuffer::from(
+ values.iter().map(Option::is_some).collect::<Vec<_>>(),
+ )),
+ )
+}
+
+async fn append(table: &Table, start: i32, count: i32) -> Vec<CommitMessage> {
+ let ids = (start..start + count).collect::<Vec<_>>();
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(ids.clone())) as ArrayRef),
+ ("a", Arc::new(Int32Array::from(ids.clone())) as ArrayRef),
+ ("b", Arc::new(Int32Array::from(ids)) as ArrayRef),
+ (
+ "payload",
+ Arc::new(variants(&values(start, count, false))) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ messages
+}
+
+async fn patch(table: &Table, column: &str, values: ArrayRef) ->
Vec<CommitMessage> {
+ let batch = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from_iter_values(20..120)) as ArrayRef,
+ ),
+ (column, values),
+ ])
+ .unwrap();
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap();
+ let files = messages
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .collect::<Vec<_>>();
+ assert_eq!(files.len(), 1);
+ assert_eq!(files[0].row_count, 100);
+ assert_eq!(files[0].first_row_id, Some(20));
+ assert_eq!(
+ files[0].write_cols.as_deref(),
+ Some(&[column.to_string()][..])
+ );
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages.clone())
+ .await
+ .unwrap();
+ messages
+}
+
+async fn remove_primaries(table: &Table, messages: &[CommitMessage]) {
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ assert!(file.file_name.ends_with(".parquet"));
+ assert_eq!(file.extra_files, vec![format!("{}.row", file.file_name)]);
+ assert!(table
+ .file_io()
+ .exists(&file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0]
+ ))
+ .await
+ .unwrap());
+ table
+ .file_io()
+ .delete_file(&file.data_file_path(&format!("{}/bucket-0",
table.location())))
+ .await
+ .unwrap();
+ }
+}
+
+async fn query(table: &Table, predicate: Predicate, projection: &[&str]) ->
RecordBatch {
+ let mut builder = table.new_read_builder();
+ builder.with_filter(predicate);
+ builder.with_projection(projection).unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert!(!plan.splits().is_empty());
+ for split in plan.splits() {
+ assert!(
+ split.row_ranges().is_none(),
+ "no global index or caller row selection"
+ );
+ assert!(split
+ .data_files()
+ .iter()
+ .all(|f| f.first_row_id == Some(20)));
+ }
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert!(!batches.is_empty());
+ concat_batches(&batches[0].schema(), &batches).unwrap()
+}
+
+fn assert_payload(batch: &RecordBatch, ids: &[i32], updated: bool) {
+ let payload = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let row_ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(batch.num_rows(), ids.len());
+ for (row, id) in ids.iter().enumerate() {
+ assert_eq!(row_ids.value(row), i64::from(*id) + 20);
+ let expected = values(*id, 1, updated).pop().unwrap();
+ assert_eq!(payload.is_null(row), expected.is_none());
+ if let Some(expected) = expected {
+ for (column, expected) in [expected.value(),
expected.metadata()].iter().enumerate() {
+ let actual = payload
+ .column(column)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ assert!(actual.is_valid(row));
+ assert_eq!(actual.value(row), *expected);
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn bitmap_reads_variant_column_merge_sidecars_without_requested_ranges()
{
+ let table = table("variant").await;
+ append(&table, -20, 20).await;
+ let mut messages = append(&table, 0, 100).await;
+ assert!(messages
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .all(|f| f.embedded_index.is_some()));
+ messages.extend(patch(&table, "payload", Arc::new(variants(&values(0, 100,
true)))).await);
+ remove_primaries(&table, &messages).await;
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .is_in("id", (5..10).map(Datum::Int).collect())
+ .unwrap();
+ // The indexed predicate field is absent from the output. The same sparse
+ // physical selection must still reach both the ID and Variant providers.
+ let batch = query(&table, predicate, &["payload", "_ROW_ID"]).await;
+ assert_payload(&batch, &[5, 6, 7, 8, 9], true);
+}
+
+#[tokio::test]
+async fn bitmap_uses_latest_column_provider_instead_of_overwritten_values() {
+ let table = table("latest").await;
+ append(&table, -20, 20).await;
+ let mut messages = append(&table, 0, 100).await;
+ let updated = (0..100)
+ .map(|id| match id {
+ 5 => 999,
+ 42 => 5,
+ _ => id,
+ })
+ .collect::<Vec<_>>();
+ let patches = patch(&table, "b",
Arc::new(Int32Array::from(updated))).await;
+ assert!(patches
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .all(|f| f.embedded_index.is_some()));
+ messages.extend(patches);
+ let schema = table
+ .schema()
+ .apply_changes(vec![SchemaChange::rename_column(
+ "b".into(),
+ "renamed_b".into(),
+ )])
+ .unwrap();
+ persist_table_schema(table.file_io(), table.location(), &schema).await;
+ let table = table.copy_with_resolved_schema(schema, "main").unwrap();
+ remove_primaries(&table, &messages).await;
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ // The old index says b=5 is row 5, whereas the winning column provider
+ // moved it to row 42. The old file has no b=999 at all, so consulting its
+ // index as an independent conjunct would incorrectly prune the second hit.
+ for (value, expected_id) in [(5, 42), (999, 5)] {
+ let batch = query(
+ &table,
+ predicates.equal("renamed_b", Datum::Int(value)).unwrap(),
+ &["payload", "_ROW_ID"],
+ )
+ .await;
+ assert_payload(&batch, &[expected_id], false);
+ }
+}
+
+#[tokio::test]
+async fn bitmap_boolean_predicates_select_the_same_rows_for_every_column() {
+ let table = table("boolean").await;
+ append(&table, -20, 20).await;
+ let mut messages = append(&table, 0, 100).await;
+ messages.extend(
+ patch(
+ &table,
+ "b",
+ Arc::new(Int32Array::from_iter_values(1000..1100)),
+ )
+ .await,
+ );
+ messages.extend(patch(&table, "payload", Arc::new(variants(&values(0, 100,
true)))).await);
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ // A cross-provider OR has no safe per-file necessary condition. Reading
+ // the primaries must retain the union rather than intersect its branches.
+ let pure_or = Predicate::or(vec![
+ predicates.equal("a", Datum::Int(5)).unwrap(),
+ predicates.equal("b", Datum::Int(1007)).unwrap(),
+ ]);
+ let fallback = query(&table, pure_or, &["payload", "_ROW_ID"]).await;
+ assert_payload(&fallback, &[5, 7], true);
+ let cases = [
+ (
+ Predicate::and(vec![
+ predicates
+ .is_in("a", vec![Datum::Int(5), Datum::Int(6)])
+ .unwrap(),
+ predicates
+ .is_in("b", vec![Datum::Int(1006), Datum::Int(1007)])
+ .unwrap(),
+ ]),
+ vec![6],
+ ),
+ (
+ Predicate::and(vec![
+ Predicate::or(vec![
+ predicates.equal("a", Datum::Int(5)).unwrap(),
+ predicates.equal("b", Datum::Int(1007)).unwrap(),
+ ]),
+ predicates
+ .is_in("id", vec![Datum::Int(5), Datum::Int(6),
Datum::Int(7)])
+ .unwrap(),
+ ]),
+ vec![5, 7],
+ ),
+ (
+ Predicate::and(vec![
+ Predicate::negate(Predicate::and(vec![
+ predicates.equal("a", Datum::Int(5)).unwrap(),
+ predicates.equal("b", Datum::Int(1007)).unwrap(),
+ ])),
+ predicates
+ .is_in("id", (5..10).map(Datum::Int).collect())
+ .unwrap(),
+ ]),
+ // No row satisfies both inner predicates. Negating only one
+ // provider's half would incorrectly discard row 5 or row 7.
+ vec![5, 6, 7, 8, 9],
+ ),
+ ];
+ let without_index =
table.copy_with_options(std::collections::HashMap::from([(
+ "file-index.read.enabled".into(),
+ "false".into(),
+ )]));
+ for (predicate, expected_ids) in &cases {
+ let batch = query(&without_index, predicate.clone(), &["payload",
"_ROW_ID"]).await;
+ assert_payload(&batch, expected_ids, true);
+ }
+ remove_primaries(&table, &messages).await;
+ for (predicate, expected_ids) in cases {
+ // Only the third provider is projected; predicates require both the
+ // original provider (a/id) and a later scalar-only provider (b).
+ let batch = query(&table, predicate, &["payload", "_ROW_ID"]).await;
+ assert_payload(&batch, &expected_ids, true);
+ }
+}
+
+#[tokio::test]
+async fn missing_latest_bitmap_keeps_primary_and_does_not_reuse_stale_index() {
+ use std::collections::HashMap;
+
+ let table = table("missing_latest_index").await;
+ append(&table, -20, 20).await;
+ let mut messages = append(&table, 0, 100).await;
+ let unindexed_writer = table.copy_with_options(HashMap::from([(
+ "file-index.bitmap.columns".into(),
+ "id,a".into(),
+ )]));
+ let updated = (0..100)
+ .map(|id| match id {
+ 5 => 999,
+ 42 => 5,
+ _ => id,
+ })
+ .collect::<Vec<_>>();
+ let patches = patch(&unindexed_writer, "b",
Arc::new(Int32Array::from(updated))).await;
+ assert!(patches
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .all(|f| f.embedded_index.is_none()));
+ messages.extend(patches);
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ assert_eq!(file.extra_files.len(), 1);
+ table
+ .file_io()
+ .delete_file(&file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0],
+ ))
+ .await
+ .unwrap();
+ }
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ // The latest b has no bitmap. Fall back to reading primary providers;
+ // neither an old index result nor an empty result is a valid replacement.
+ for (value, expected_id) in [(5, 42), (999, 5)] {
+ let batch = query(
+ &table,
+ predicates.equal("b", Datum::Int(value)).unwrap(),
+ &["payload", "_ROW_ID"],
+ )
+ .await;
+ assert_payload(&batch, &[expected_id], false);
+ }
+ let batch = query(
+ &table,
+ Predicate::or(vec![
+ predicates.equal("a", Datum::Int(5)).unwrap(),
+ predicates.equal("b", Datum::Int(5)).unwrap(),
+ ]),
+ &["payload", "_ROW_ID"],
+ )
+ .await;
+ assert_payload(&batch, &[5, 42], false);
+}
+
+#[tokio::test]
+async fn
bitmap_requested_ranges_deletions_and_limit_keep_merged_providers_aligned() {
+ use std::collections::HashMap;
+
+ for (name, max_rows, sidecar) in [
+ ("aligned", "4", true),
+ ("after_dv", "3", true),
+ ("before_limit", "2", false),
+ ] {
+ let table = table(name).await.copy_with_options(HashMap::from([(
+ "data-evolution.row-sidecar.max-selected-rows".into(),
+ max_rows.into(),
+ )]));
+ append(&table, -20, 20).await;
+ let mut messages = append(&table, 0, 100).await;
+ messages.extend(patch(&table, "payload", Arc::new(variants(&values(0,
100, true)))).await);
+ let deletions = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![27])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletions)
+ .await
+ .unwrap();
+ if sidecar {
+ remove_primaries(&table, &messages).await;
+ } else {
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ table
+ .file_io()
+ .delete_file(&file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0],
+ ))
+ .await
+ .unwrap();
+ }
+ }
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["payload", "_ROW_ID"]).unwrap();
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .is_in("id", (5..10).map(Datum::Int).collect())
+ .unwrap(),
+ );
+ builder.with_row_ranges(vec![RowRange::new(26, 29)]);
+ builder.with_limit(2);
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert!(!plan.splits().is_empty());
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert!(!batches.is_empty());
+ // Four index hits remain after requested-range intersection. Deleting
+ // global row 27 must remove local selected position 1 from every
+ // provider, then LIMIT returns rows 26/28, including a SQL-null
Variant.
+ // DV reduces the bitmap's four candidates to three before choosing
+ // ROW. LIMIT must not reduce that threshold further to two.
+ let batch = concat_batches(&batches[0].schema(), &batches).unwrap();
+ assert_payload(&batch, &[6, 8], true);
+ }
+}
+
+#[tokio::test]
+async fn deletions_narrow_sidecar_candidates_only_for_a_selective_bitmap() {
+ use std::collections::HashMap;
+
+ for merged in [false, true] {
+ for selective in [false, true] {
+ let table = table(&format!("bitmap_dv_{merged}_{selective}"))
+ .await
+ .copy_with_options(HashMap::from([(
+ "data-evolution.row-sidecar.max-selected-rows".into(),
+ "3".into(),
+ )]));
+ append(&table, -20, 20).await;
+ let mut messages = append(&table, 0, 100).await;
+ if merged {
+ messages.extend(
+ patch(&table, "payload", Arc::new(variants(&values(0, 100,
true)))).await,
+ );
+ }
+ let deleted = if selective {
+ vec![25, 27]
+ } else {
+ (20..119).collect()
+ };
+ let deletions = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(deleted)
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletions)
+ .await
+ .unwrap();
+ if selective {
+ remove_primaries(&table, &messages).await;
+ } else {
+ // Java normalizes a full-coverage bitmap to REMAIN. Even
+ // though DV leaves one live row, it cannot choose ROW alone.
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ table
+ .file_io()
+ .delete_file(&file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0],
+ ))
+ .await
+ .unwrap();
+ }
+ }
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ let predicate = if selective {
+ predicates
+ .is_in("id", (5..10).map(Datum::Int).collect())
+ .unwrap()
+ } else {
+ predicates.greater_or_equal("id", Datum::Int(0)).unwrap()
+ };
+ let batch = query(&table, predicate, &["payload",
"_ROW_ID"]).await;
+ let expected = if selective { vec![6, 8, 9] } else { vec![99] };
+ assert_payload(&batch, &expected, merged);
+ }
+ }
+}
diff --git a/crates/paimon/tests/data_evolution_row_sidecar_test.rs
b/crates/paimon/tests/data_evolution_row_sidecar_test.rs
index b2d924ee..e25d7fd4 100644
--- a/crates/paimon/tests/data_evolution_row_sidecar_test.rs
+++ b/crates/paimon/tests/data_evolution_row_sidecar_test.rs
@@ -22,7 +22,7 @@ use common::incremental_helpers::{memory_table,
persist_table_schema, setup_dirs
use futures::TryStreamExt;
use paimon::io::FileIO;
use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
-use paimon::table::{CommitMessage, RowRange, Table};
+use paimon::table::{CommitMessage, DataSplitBuilder, RowRange, Table};
use std::collections::HashMap;
use std::sync::Arc;
@@ -371,8 +371,113 @@ async fn
explicit_abort_deletes_sidecar_while_writer_close_keeps_prepared_files(
}
#[tokio::test]
-async fn file_index_and_limit_do_not_turn_full_reads_into_sidecar_reads() {
+async fn bitmap_index_selection_reads_sidecar_without_primary_file() {
use paimon::spec::{Datum, PredicateBuilder};
+ let table = table(&[
+ ("file-index.bitmap.columns", "id"),
+ ("file-index.in-manifest-threshold", "1mb"),
+ ])
+ .await;
+ append(&table, -20, 20).await;
+ let messages = append(&table, 0, 100).await;
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.embedded_index.is_some()));
+ remove_primary(table.file_io(), &table, &messages).await;
+
+ for project_row_id in [false, true] {
+ for (requested_ranges, expected) in [
+ (None, vec![5]),
+ (Some(vec![RowRange::new(20, 119)]), vec![5]),
+ (Some(vec![RowRange::new(25, 27)]), vec![5]),
+ (Some(vec![RowRange::new(26, 30)]), vec![]),
+ ] {
+ let mut builder = table.new_read_builder();
+ if project_row_id {
+ builder
+ .with_projection(&["id", "value", "_ROW_ID"])
+ .unwrap();
+ }
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(5))
+ .unwrap(),
+ );
+ if let Some(ranges) = requested_ranges {
+ builder.with_row_ranges(ranges);
+ }
+ let plan = builder.new_scan().plan().await.unwrap();
+ let actual: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let ids = actual
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(ids, expected);
+ if project_row_id && !actual.is_empty() {
+ assert_eq!(rows(&actual), vec![(5, "v5".into(), 25)]);
+ }
+ }
+ }
+
+ // Direct split callers need not pass through ReadBuilder's range
+ // normalization. A later range must not hide an earlier bitmap hit.
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id", "value", "_ROW_ID"])
+ .unwrap();
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(5))
+ .unwrap(),
+ );
+ let plan = builder.new_scan().plan().await.unwrap();
+ let direct_splits = plan
+ .splits()
+ .iter()
+ .map(|split| {
+ DataSplitBuilder::new()
+ .with_snapshot(split.snapshot_id())
+ .with_partition(split.partition().clone())
+ .with_bucket(split.bucket())
+ .with_bucket_path(split.bucket_path().to_string())
+ .with_total_buckets(split.total_buckets())
+ .with_data_files(split.data_files().to_vec())
+ .with_raw_convertible(split.raw_convertible())
+ .with_row_ranges(vec![RowRange::new(80, 90), RowRange::new(25,
27)])
+ .build()
+ .unwrap()
+ })
+ .collect::<Vec<_>>();
+ let actual: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(&direct_splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(rows(&actual), vec![(5, "v5".into(), 25)]);
+}
+
+#[tokio::test]
+async fn limit_and_deletion_vectors_do_not_trigger_sidecar_reads() {
let table = table(&[("file-index.bitmap.columns", "id")]).await;
let messages = append(&table, 0, 100).await;
for file in messages.iter().flat_map(|m| &m.new_files) {
@@ -389,28 +494,6 @@ async fn
file_index_and_limit_do_not_turn_full_reads_into_sidecar_reads() {
.await
.unwrap();
}
- // A one-row bitmap selection must still decode the existing primary file.
- for requested_ranges in [None, Some(vec![RowRange::new(0, 99)])] {
- let mut builder = table.new_read_builder();
- builder.with_filter(
- PredicateBuilder::new(table.schema().fields())
- .equal("id", Datum::Int(5))
- .unwrap(),
- );
- if let Some(ranges) = requested_ranges {
- builder.with_row_ranges(ranges);
- }
- let plan = builder.new_scan().plan().await.unwrap();
- let actual: Vec<RecordBatch> = builder
- .new_read()
- .unwrap()
- .to_arrow(plan.splits())
- .unwrap()
- .try_collect()
- .await
- .unwrap();
- assert_eq!(actual.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
- }
let mut builder = table.new_read_builder();
builder.with_limit(1);
let plan = builder.new_scan().plan().await.unwrap();
@@ -475,6 +558,20 @@ async fn
file_index_and_limit_do_not_turn_full_reads_into_sidecar_reads() {
.value(0),
"changed"
);
+
+ // Keeping only one live row must not make an otherwise full read use a
+ // sidecar either. All the sidecars above are absent, but primaries remain.
+ let deletions = update.delete_by_row_id((1..100).collect()).await.unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletions)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&read(&table, None).await.unwrap()),
+ vec![(0, "changed".into(), 0)]
+ );
}
#[tokio::test]
diff --git a/crates/paimon/tests/data_evolution_search_sidecar_test.rs
b/crates/paimon/tests/data_evolution_search_sidecar_test.rs
new file mode 100644
index 00000000..58fde33e
--- /dev/null
+++ b/crates/paimon/tests/data_evolution_search_sidecar_test.rs
@@ -0,0 +1,440 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::types::Float32Type;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, Float32Array, Int32Array, Int64Array,
ListArray, RecordBatch,
+ StringArray, StructArray,
+};
+use arrow_buffer::NullBuffer;
+use arrow_schema::DataType as ArrowDataType;
+use arrow_select::concat::concat_batches;
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::arrow::variant_arrow_type;
+use paimon::spec::{
+ ArrayType, DataType, Datum, FloatType, IntType, PredicateBuilder, Schema,
TableSchema,
+ VarCharType, VariantType,
+};
+use paimon::table::{CommitMessage, GlobalIndexResult, Table};
+use paimon::variant::GenericVariant;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+fn value(id: i32, updated: bool) -> Option<GenericVariant> {
+ let json = match id.rem_euclid(5) {
+ 1 => return None,
+ 2 => "null".into(),
+ 3 => r#""scalar""#.into(),
+ 4 => "[1,null,true]".into(),
+ _ => format!(
+ r#"{{"x":{},"nested":{{"items":[true,null,"中"]}}}}"#,
+ id + if updated { 1000 } else { 0 }
+ ),
+ };
+ Some(GenericVariant::parse_json(&json).unwrap())
+}
+
+fn variants(start: i32, count: i32, updated: bool) -> StructArray {
+ let values = (start..start + count)
+ .map(|id| value(id, updated))
+ .collect::<Vec<_>>();
+ let ArrowDataType::Struct(fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ StructArray::new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(
+ values
+ .iter()
+ .map(|v| v.as_ref().map_or(&[][..],
GenericVariant::value)),
+ )),
+ Arc::new(BinaryArray::from_iter_values(
+ values
+ .iter()
+ .map(|v| v.as_ref().map_or(&[][..],
GenericVariant::metadata)),
+ )),
+ ],
+ Some(NullBuffer::from(
+ values.iter().map(Option::is_some).collect::<Vec<_>>(),
+ )),
+ )
+}
+
+fn input(start: i32, count: i32) -> RecordBatch {
+ let embeddings =
+ ListArray::from_iter_primitive::<Float32Type, _, _>((start..start +
count).map(|id| {
+ let x = match id {
+ 5 => 0.0,
+ 7 => 1.0,
+ 6 => 2.0,
+ 8 => 3.0,
+ _ => 1000.0 + id.abs() as f32,
+ };
+ Some(vec![Some(x), Some(0.0)])
+ }));
+ let text = StringArray::from_iter_values((start..start + count).map(|id|
match id {
+ 5 => "needle spare spare spare spare spare spare spare",
+ 6 => "needle",
+ 7 => "needle needle needle needle",
+ _ => "unrelated",
+ }));
+ RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(start..start + count)) as
ArrayRef,
+ ),
+ ("embedding", Arc::new(embeddings) as ArrayRef),
+ ("text", Arc::new(text) as ArrayRef),
+ (
+ "payload",
+ Arc::new(variants(start, count, false)) as ArrayRef,
+ ),
+ ])
+ .unwrap()
+}
+
+async fn append(table: &Table, batch: &RecordBatch) -> Vec<CommitMessage> {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ messages
+}
+
+async fn fixture(name: &str, full_text: bool) -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .column("text", DataType::VarChar(VarCharType::string_type()))
+ .column("payload", DataType::Variant(VariantType::new()))
+ .option("file.format", "parquet")
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ // Full-text index builds currently require DV to be disabled. The
+ // vector fixture separately verifies deletion-aware Top-K and reads.
+ .option(
+ "deletion-vectors.enabled",
+ if full_text { "false" } else { "true" },
+ )
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .option("global-index.enabled", "true")
+ .option("global-index.row-count-per-shard", "1000")
+ .option("file-index.bitmap.columns", "id")
+ .option("file-index.in-manifest-threshold", "1mb")
+ .option("vector-index.search-mode", "fast")
+ .option("full-text-index.search-mode", "fast")
+ .option("ivf-flat.dimension", "2")
+ .option("ivf-flat.nlist", "1")
+ .option("ivf-flat.distance.metric", "l2")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(
+ &format!("memory:/search_sidecar_{name}"),
+ TableSchema::new(0, &schema),
+ );
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let mut messages = append(&table, &input(-20, 20)).await;
+ messages.extend(append(&table, &input(0, 100)).await);
+ assert!(
+ table
+ .new_global_index_build_builder()
+ .with_index_column("embedding")
+ .with_index_type("ivf-flat")
+ .execute()
+ .await
+ .unwrap()
+ > 0
+ );
+ if full_text {
+ assert!(
+ table
+ .new_global_index_build_builder()
+ .with_index_column("text")
+ .with_index_type("full-text")
+ .execute()
+ .await
+ .unwrap()
+ > 0
+ );
+ }
+ let patch = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from_iter_values(20..120)) as ArrayRef,
+ ),
+ ("payload", Arc::new(variants(0, 100, true)) as ArrayRef),
+ ])
+ .unwrap();
+ let updates = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![patch])
+ .await
+ .unwrap();
+ assert_eq!(updates.iter().map(|m| m.new_files.len()).sum::<usize>(), 1);
+ let file = &updates[0].new_files[0];
+ assert_eq!(file.row_count, 100);
+ assert_eq!(file.first_row_id, Some(20));
+ assert_eq!(
+ file.write_cols.as_deref(),
+ Some(&["payload".to_string()][..])
+ );
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(updates.clone())
+ .await
+ .unwrap();
+ messages.extend(updates);
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ assert!(file.file_name.ends_with(".parquet"));
+ assert_eq!(file.extra_files, vec![format!("{}.row", file.file_name)]);
+ assert!(io
+ .exists(&file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0]
+ ))
+ .await
+ .unwrap());
+ io.delete_file(&file.data_file_path(&format!("{}/bucket-0",
table.location())))
+ .await
+ .unwrap();
+ }
+ table
+}
+
+async fn assert_materialized(result: &GlobalIndexResult) {
+ let expected = result.row_ids().unwrap();
+ assert_eq!(expected.row_ids.len(), 3);
+ let batches = result
+ .new_read_builder()
+ .with_projection(&["payload", "id"])
+ .read()
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert!(!batches.is_empty());
+ let batch = concat_batches(&batches[0].schema(), &batches).unwrap();
+ assert_eq!(
+ batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().as_str())
+ .collect::<Vec<_>>(),
+ vec!["payload", "id", "__paimon_search_score"]
+ );
+ assert_eq!(batch.num_rows(), expected.row_ids.len());
+ let ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let scores = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap();
+ let payload = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ for (row, (&global_id, &score)) in
expected.row_ids.iter().zip(&expected.scores).enumerate() {
+ let id = i32::try_from(global_id - 20).unwrap();
+ assert_eq!(ids.value(row), id, "materialized order follows search
rank");
+ assert!(
+ (scores.value(row) - score).abs() < 1e-6,
+ "score belongs to the same global row ID"
+ );
+ let expected = value(id, true);
+ assert_eq!(payload.is_null(row), expected.is_none());
+ if let Some(expected) = expected {
+ for (index, expected) in [expected.value(),
expected.metadata()].iter().enumerate() {
+ let child = payload
+ .column(index)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ assert!(child.is_valid(row));
+ assert_eq!(
+ child.value(row),
+ *expected,
+ "latest Variant buffer at row ID {global_id}"
+ );
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn
real_vector_hits_read_updated_variant_sidecars_in_score_order_with_deletions() {
+ let table = fixture("vector", false).await;
+ let mut builder = table.new_vector_search_builder();
+ builder
+ .with_vector_column("embedding")
+ .with_query_vector(vec![0.0, 0.0])
+ .with_limit(3)
+ .with_options(vector_options());
+ let result = builder.execute().await.unwrap();
+ assert_eq!(result.row_ids().unwrap().row_ids, vec![25, 27, 26]);
+ for (actual, expected) in
result.row_ids().unwrap().scores.iter().zip([1.0, 0.5, 0.2]) {
+ assert!((*actual - expected).abs() < 1e-6);
+ }
+ // Physical rows are 5/6/7; ranking is 5/7/6, crossing JSON and SQL null.
+ // Both the original ID provider and the latest Variant provider must read
+ // their ROW sidecars because every primary Parquet has been deleted.
+ assert_materialized(&result).await;
+
+ // Scalar search filters materialize matching row IDs before searching.
+ // That internal `_ROW_ID` projection must retain bitmap pruning and ROW
+ // access when its internal projection contains only the row-ID column.
+ let mut filtered = table.new_vector_search_builder();
+ filtered
+ .with_vector_column("embedding")
+ .with_query_vector(vec![0.0, 0.0])
+ .with_limit(3)
+ .with_options(vector_options())
+ .with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .is_in("id", (6..10).map(Datum::Int).collect())
+ .unwrap(),
+ );
+ let filtered = filtered.execute().await.unwrap();
+ assert_eq!(filtered.row_ids().unwrap().row_ids, vec![27, 26, 28]);
+ assert_materialized(&filtered).await;
+
+ let deletions = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![25])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletions)
+ .await
+ .unwrap();
+ // Also exercise the public Scan -> Plan -> Read API, without rerouting
+ // materialization through a hand-constructed row-range request.
+ let result = builder
+ .new_read()
+ .unwrap()
+ .read(builder.new_scan().unwrap().plan().await.unwrap())
+ .await
+ .unwrap();
+ assert_eq!(result.row_ids().unwrap().row_ids, vec![27, 26, 28]);
+ for (actual, expected) in
result.row_ids().unwrap().scores.iter().zip([0.5, 0.2, 0.1]) {
+ assert!((*actual - expected).abs() < 1e-6);
+ }
+ assert_materialized(&result).await;
+}
+
+fn vector_options() -> HashMap<String, String> {
+ HashMap::from([
+ ("index_type".into(), "ivf-flat".into()),
+ ("ivf.nprobe".into(), "1".into()),
+ ])
+}
+
+#[cfg(feature = "fulltext")]
+#[tokio::test]
+async fn
full_text_and_hybrid_hits_read_latest_variant_sidecars_with_ranked_scores() {
+ let table = fixture("full_text_hybrid", true).await;
+ let text = table
+ .new_full_text_search_builder()
+ .with_text_column("text")
+ .with_query_text("needle")
+ .with_limit(3)
+ .execute_scored()
+ .await
+ .unwrap();
+ let text_hits = text.row_ids().unwrap();
+ let mut ids = text_hits.row_ids.clone();
+ ids.sort_unstable();
+ assert_eq!(ids, vec![25, 26, 27]);
+ assert_ne!(
+ text_hits.row_ids,
+ vec![25, 26, 27],
+ "different document lengths must produce a nonphysical relevance order"
+ );
+ assert!(text_hits.scores.iter().all(|s| s.is_finite() && *s > 0.0));
+ assert!(text_hits.scores.windows(2).all(|s| s[0] >= s[1]));
+ assert_materialized(&text).await;
+
+ let vector = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![0.0, 0.0])
+ .with_limit(3)
+ .with_options(vector_options())
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(vector.row_ids().unwrap().row_ids, vec![25, 27, 26]);
+ let mut expected_scores = HashMap::<u64, f32>::new();
+ for hits in [vector.row_ids().unwrap(), text_hits] {
+ for (rank, row_id) in hits.row_ids.iter().enumerate() {
+ *expected_scores.entry(*row_id).or_default() += 1.0 / (60.0 + rank
as f32 + 1.0);
+ }
+ }
+ let mut expected = expected_scores.into_iter().collect::<Vec<_>>();
+ expected.sort_by(|left, right| {
+ right
+ .1
+ .total_cmp(&left.1)
+ .then_with(|| left.0.cmp(&right.0))
+ });
+ let mut hybrid = table.new_hybrid_search_builder();
+ hybrid
+ .add_vector_route("embedding", vec![0.0, 0.0], 3, 1.0,
vector_options())
+ .unwrap()
+ .add_full_text_route(
+ "text",
+ r#"{"match":{"query":"needle"}}"#,
+ 3,
+ 1.0,
+ HashMap::new(),
+ )
+ .unwrap()
+ .with_limit(3);
+ let fused = hybrid.execute_scored().await.unwrap();
+ let hits = fused.row_ids().unwrap();
+ assert_eq!(
+ hits.row_ids,
+ expected.iter().map(|(id, _)| *id).collect::<Vec<_>>()
+ );
+ for (score, (_, expected)) in hits.scores.iter().zip(expected) {
+ assert!((*score - expected).abs() < 1e-6);
+ }
+ assert_materialized(&fused).await;
+}
diff --git a/crates/paimon/tests/data_evolution_variant_sidecar_test.rs
b/crates/paimon/tests/data_evolution_variant_sidecar_test.rs
new file mode 100644
index 00000000..93983f02
--- /dev/null
+++ b/crates/paimon/tests/data_evolution_variant_sidecar_test.rs
@@ -0,0 +1,457 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{Array, ArrayRef, BinaryArray, Int32Array, Int64Array,
RecordBatch, StructArray};
+use arrow_buffer::NullBuffer;
+use arrow_schema::{DataType as ArrowDataType, Field};
+use arrow_select::concat::concat_batches;
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::arrow::variant_arrow_type;
+use paimon::spec::{
+ variant_extraction_row, BigIntType, DataField, DataType, Datum, IntType,
Predicate,
+ PredicateBuilder, RowType, Schema, SchemaChange, TableSchema, VariantType,
ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME,
+};
+use paimon::table::{CommitMessage, Table};
+use paimon::variant::GenericVariant;
+use std::sync::Arc;
+
+fn variants(values: &[Option<GenericVariant>]) -> StructArray {
+ let ArrowDataType::Struct(fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ StructArray::new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(values.iter().map(|value| {
+ value.as_ref().map_or(&[][..], GenericVariant::value)
+ }))),
+ Arc::new(BinaryArray::from_iter_values(values.iter().map(|value| {
+ value.as_ref().map_or(&[][..], GenericVariant::metadata)
+ }))),
+ ],
+ Some(NullBuffer::from(
+ values.iter().map(Option::is_some).collect::<Vec<_>>(),
+ )),
+ )
+}
+
+fn values(start: i32, count: i32, updated: bool) ->
Vec<Option<GenericVariant>> {
+ (start..start + count)
+ .map(|id| {
+ let json = if updated && id == 5 {
+ Some(r#"{"x":999,"other":[true,null]}"#.to_string())
+ } else if id == 50 {
+ // An unselected value must not fail the strict extraction.
+ Some(r#"{"x":"invalid integer"}"#.to_string())
+ } else {
+ match id.rem_euclid(5) {
+ 1 => None,
+ 2 => Some("null".to_string()),
+ 4 => Some(r#""scalar""#.to_string()),
+ _ => Some(format!(r#"{{"x":{id},"other":[true,null]}}"#)),
+ }
+ };
+ json.map(|json| GenericVariant::parse_json(&json).unwrap())
+ })
+ .collect()
+}
+
+fn input(start: i32, count: i32) -> RecordBatch {
+ let nested_values = (start..start + count)
+ .map(|id| {
+ if id.rem_euclid(5) == 1 {
+ None
+ } else if id.rem_euclid(5) == 2 {
+ Some(GenericVariant::parse_json("null").unwrap())
+ } else {
+ Some(GenericVariant::parse_json(&format!(r#"{{"x":{}}}"#, id +
100)).unwrap())
+ }
+ })
+ .collect::<Vec<_>>();
+ let nested = StructArray::new(
+ vec![Arc::new(Field::new("detail", variant_arrow_type(),
true))].into(),
+ vec![Arc::new(variants(&nested_values))],
+ Some(NullBuffer::from(
+ (start..start + count)
+ .map(|id| id.rem_euclid(5) != 3)
+ .collect::<Vec<_>>(),
+ )),
+ );
+ RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(start..start + count)) as
ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(variants(&values(start, count, false))) as ArrayRef,
+ ),
+ ("nested", Arc::new(nested) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+async fn append(table: &Table, batch: &RecordBatch) -> Vec<CommitMessage> {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ messages
+}
+
+fn extraction_type() -> DataType {
+ DataType::Row(
+ variant_extraction_row(
+ true,
+ ["$.x", "$.missing"].map(|path| {
+ (
+ DataType::Int(IntType::new()),
+ path.into(),
+ true,
+ "UTC".into(),
+ )
+ }),
+ )
+ .unwrap(),
+ )
+}
+
+async fn query(table: &Table, extraction: bool) -> RecordBatch {
+ let fields = table.schema().fields();
+ let mut builder = table.new_read_builder();
+ if extraction {
+ let DataType::Row(nested) = fields[2].data_type() else {
+ unreachable!()
+ };
+ let detail = &nested.fields()[0];
+ builder.with_read_type(vec![
+ DataField::new(
+ fields[2].id(),
+ "nested".into(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ detail.id(),
+ "detail".into(),
+ extraction_type(),
+ )])),
+ ),
+ DataField::new(
+ ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME.into(),
+ DataType::BigInt(BigIntType::new()),
+ ),
+ DataField::new(fields[1].id(), fields[1].name().into(),
extraction_type()),
+ fields[0].clone(),
+ ]);
+ } else {
+ builder
+ .with_projection(&["nested", "_ROW_ID", fields[1].name(), "id"])
+ .unwrap();
+ }
+ let predicates = PredicateBuilder::new(fields);
+ builder.with_filter(Predicate::and(vec![
+ predicates.greater_or_equal("id", Datum::Int(5)).unwrap(),
+ predicates.less_or_equal("id", Datum::Int(9)).unwrap(),
+ ]));
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert!(!plan.splits().is_empty());
+ for split in plan.splits() {
+ let ranges = split
+ .row_ranges()
+ .expect("the real global index must select row IDs");
+ assert!(ranges
+ .iter()
+ .all(|range| range.from() >= 25 && range.to() <= 29));
+ assert!(split
+ .data_files()
+ .iter()
+ .all(|file| file.file_name.ends_with(".parquet")));
+ assert!(split
+ .data_files()
+ .iter()
+ .all(|file| file.first_row_id == Some(20)));
+ }
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ concat_batches(&batches[0].schema(), &batches).unwrap()
+}
+
+fn assert_result(batch: &RecordBatch, updated: bool, extraction: bool,
deleted: bool) {
+ let expected_ids = if deleted {
+ vec![5, 6, 7, 9]
+ } else {
+ vec![5, 6, 7, 8, 9]
+ };
+ let ids = batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(ids.values().as_ref(), expected_ids.as_slice());
+ let row_ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(
+ row_ids.values().as_ref(),
+ expected_ids
+ .iter()
+ .map(|id| i64::from(*id) + 20)
+ .collect::<Vec<_>>()
+ );
+ let payload = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert!(payload.is_null(1), "SQL NULL keeps a null Variant parent");
+ assert!(payload.is_valid(2), "JSON null remains a non-null Variant");
+ if extraction {
+ let x = payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let expected = expected_ids
+ .iter()
+ .map(|id| match *id {
+ 5 => Some(if updated { 999 } else { 5 }),
+ 8 => Some(8),
+ _ => None,
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(x, &Int32Array::from(expected));
+ assert_eq!(payload.column(1).null_count(), expected_ids.len());
+ let nested = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let detail = nested
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let x = detail
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for (row, id) in expected_ids.iter().enumerate() {
+ if *id == 5 || *id == 9 {
+ assert!(detail.is_valid(row));
+ assert_eq!(x.value(row), id + 100);
+ }
+ if *id == 6 {
+ assert!(nested.is_valid(row));
+ assert!(detail.is_null(row), "nested SQL NULL keeps a null
parent");
+ }
+ if *id == 7 {
+ assert!(nested.is_valid(row));
+ assert!(detail.is_valid(row), "nested JSON null keeps its
parent");
+ assert!(x.is_null(row));
+ }
+ if *id == 8 {
+ assert!(nested.is_null(row));
+ } else {
+ assert!(detail.column(1).is_null(row), "missing paths remain
NULL");
+ }
+ }
+ } else {
+ let expected = values(0, 100, updated);
+ let actual_values = payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let metadata = payload
+ .column(1)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ for (row, id) in expected_ids.iter().enumerate() {
+ if let Some(expected) = &expected[*id as usize] {
+ assert_eq!(actual_values.value(row), expected.value());
+ assert_eq!(metadata.value(row), expected.metadata());
+ }
+ }
+ let nested = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let detail = nested
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let detail_values = detail
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let detail_metadata = detail
+ .column(1)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ for (row, id) in expected_ids.iter().enumerate() {
+ assert_eq!(nested.is_null(row), id.rem_euclid(5) == 3);
+ if nested.is_null(row) {
+ continue;
+ }
+ assert_eq!(detail.is_null(row), *id == 6);
+ if detail.is_valid(row) {
+ let expected = GenericVariant::parse_json(&if *id == 7 {
+ "null".to_string()
+ } else {
+ format!(r#"{{"x":{}}}"#, id + 100)
+ })
+ .unwrap();
+ assert_eq!(detail_values.value(row), expected.value());
+ assert_eq!(detail_metadata.value(row), expected.metadata());
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn
global_indexes_read_variant_sidecars_after_column_updates_and_rename() {
+ for (index_type, shredded) in [("bitmap", false), ("btree", true)] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Variant(VariantType::new()))
+ .column(
+ "nested",
+ DataType::Row(RowType::new(vec![DataField::new(
+ 0,
+ "detail".into(),
+ DataType::Variant(VariantType::new()),
+ )])),
+ )
+ .option("file.format", "parquet")
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .option("global-index.enabled", "true");
+ if shredded {
+ schema = schema.option("parquet.variant.shreddingSchema",
+
r#"{"type":"ROW","fields":[{"name":"payload","type":{"type":"ROW","fields":[{"name":"x","type":"INT"}]}}]}"#);
+ }
+ let (io, table) = memory_table(
+ &format!("memory:/variant_sidecar_{index_type}"),
+ TableSchema::new(0, &schema.build().unwrap()),
+ );
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ append(&table, &input(-20, 20)).await;
+ let initial = append(&table, &input(0, 100)).await;
+ assert!(
+ table
+ .new_global_index_build_builder()
+ .with_index_column("id")
+ .with_index_type(index_type)
+ .execute()
+ .await
+ .unwrap()
+ > 0
+ );
+ assert_result(&query(&table, false).await, false, false, false);
+ let file = &initial[0].new_files[0];
+ assert_eq!(file.extra_files, vec![format!("{}.row", file.file_name)]);
+ let path = file.data_file_path(&format!("{}/bucket-0",
table.location()));
+ let primary = io.new_input(&path).unwrap().read().await.unwrap();
+ io.delete_file(&path).await.unwrap();
+ for extraction in [false, true] {
+ assert_result(&query(&table, extraction).await, false, extraction,
false);
+ }
+ // Restore only for producing the column patch; reads below remove all
providers again.
+ io.new_output(&path).unwrap().write(primary).await.unwrap();
+ let patch = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from_iter_values(20..120)) as ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(variants(&values(0, 100, true))) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let update = table.new_write_builder().new_update().unwrap();
+ let updates = update
+ .update_by_arrow_with_row_id(vec![patch])
+ .await
+ .unwrap();
+ assert!(updates.iter().flat_map(|m| &m.new_files).all(|f| f
+ .write_cols
+ .as_ref()
+ .is_some_and(|columns| columns == &["payload"])));
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(updates.clone())
+ .await
+ .unwrap();
+ let schema = table
+ .schema()
+ .apply_changes(vec![SchemaChange::rename_column(
+ "payload".into(),
+ "renamed_payload".into(),
+ )])
+ .unwrap();
+ persist_table_schema(&io, table.location(), &schema).await;
+ let table = table.copy_with_resolved_schema(schema, "main").unwrap();
+ for file in initial.iter().chain(&updates).flat_map(|m| &m.new_files) {
+ assert_eq!(file.row_count, 100);
+ io.delete_file(&file.data_file_path(&format!("{}/bucket-0",
table.location())))
+ .await
+ .unwrap();
+ }
+ for extraction in [false, true] {
+ assert_result(&query(&table, extraction).await, true, extraction,
false);
+ }
+ let deletions = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![28])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletions)
+ .await
+ .unwrap();
+ for extraction in [false, true] {
+ assert_result(&query(&table, extraction).await, true, extraction,
true);
+ }
+ }
+}
