diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index 8324e3a03e5..db74861e3fe 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -5341,14 +5341,16 @@ async fn test_reset_state_discards_dynamic_field_pruning() { ))) .unwrap(); - // Dynamic pruning skips the first row group but does not filter individual - // rows in the second. Drop before consuming the third row group. + // Dynamic pruning skips the first row group, and the dynamic prefilter prunes the + // row the producer rejects in the second. Only the row above the producer's + // threshold is left; the live filter above the scan still owns the final result. + // Drop before consuming the third row group. let mut stream = scanner .scan_partition(&Default::default(), &ExecutionPlanMetricsSet::default(), 0) .unwrap(); let first_batch = stream.try_next().await.unwrap().unwrap(); assert_eq!( - vec![2.0, 3.0], + vec![3.0], scan_field_values(std::slice::from_ref(&first_batch)) ); drop(stream); diff --git a/src/mito2/src/sst/parquet/prefilter.rs b/src/mito2/src/sst/parquet/prefilter.rs index d5401f4a6de..fbb1e0b5101 100644 --- a/src/mito2/src/sst/parquet/prefilter.rs +++ b/src/mito2/src/sst/parquet/prefilter.rs @@ -24,6 +24,9 @@ use std::sync::Arc; use api::v1::SemanticType; use common_recordbatch::filter::SimpleFilterEvaluator; +use common_telemetry::debug; +use datafusion::physical_expr::utils::collect_columns; +use datafusion::physical_plan::PhysicalExpr; use datafusion_common::ScalarValue; use datafusion_expr::Expr; use datatypes::arrow::array::{Array, BinaryArray, BooleanArray, BooleanBufferBuilder}; @@ -53,8 +56,8 @@ use crate::sst::parquet::file_range::PreFilterMode; use crate::sst::parquet::flat_format::FlatReadFormat; use crate::sst::parquet::format::{PrimaryKeyArray, StatValues}; use crate::sst::parquet::reader::{ - MaybeFilter, PhysicalFilterContext, RowGroupBuildContext, RowGroupReaderBuilder, - SimpleFilterContext, + DynamicFilterSnapshot, MaybeFilter, PhysicalFilterContext, RowGroupBuildContext, + RowGroupReaderBuilder, SimpleFilterContext, }; pub(crate) fn matching_row_ranges_by_primary_key( @@ -401,6 +404,12 @@ pub(crate) fn build_bulk_filter_plan( /// the prefilter pass. A caller can postpone simple timestamp filters to the normal /// precise-filter path when the scan time range covers the SST. When predicate /// prefiltering is disabled, all simple filters remain on the normal path instead. +/// +/// Producer-owned dynamic filters, such as the membership filter a single-key `HashJoin` +/// publishes, are admitted the same best-effort way, from their published physical +/// expression instead of a logical [`Expr`]: see [`PhysicalFilterContext::new_dynamic`]. +/// Their masks are never cached, because the producer may republish the expression +/// between row groups. #[allow(clippy::too_many_arguments)] pub(crate) fn build_reader_filter_plan( predicate: Option<&Predicate>, @@ -517,6 +526,20 @@ pub(crate) fn build_reader_filter_plan( }; } + // Producer-owned dynamic filters are physical expressions the producer republishes + // while the scan runs. Evaluate their current snapshot as a prefilter pass so the row + // selection shrinks before the full projection is read. A dynamic filter never + // replaces the live filter above the scan, and its mask is never cached. + let dyn_filters = predicate.dyn_filters(); + for handle in dyn_filters.iter() { + if let Some(filter) = + PhysicalFilterContext::new_dynamic(metadata, expected_metadata, read_format, handle) + && can_direct_prefilter(filter.semantic_type()) + { + prefilter_physical_filters.push(filter); + } + } + let pk_filter_expr_strs = (!pk_filter_contexts.is_empty()).then(|| { let mut expr_strs = pk_filter_contexts .iter() @@ -567,6 +590,12 @@ pub(crate) struct PrefilterContext { /// Physical filters that can be evaluated directly from the prefilter batch. /// Physical expressions are only applied in the prefilter phase. physical_filters: Vec, + /// Producer-owned dynamic filter snapshots, aligned with `physical_filters`. + /// + /// Captured when this row group's prefilter starts and evaluated for the whole pass. + /// `None` for static filters and for dynamic filters this pass cannot use, which + /// keeps every row instead. + dynamic_snapshots: Vec>, /// Region schema version used in per-filter cache keys. schema_version: u64, /// Sorted expression strings for the encoded-PK filter group. @@ -577,6 +606,41 @@ pub(crate) struct PrefilterContext { proven_simple_filters: Vec, } +impl PrefilterContext { + /// Returns true when the dynamic filter at `physical_filters[idx]` contributes no mask + /// for this row group: no snapshot was captured, or the snapshot references no column. + /// + /// A producer that has not published yet still holds its `true` placeholder, and + /// reading the column for this row group would only cost I/O. A constant that rejects + /// every row is left to the row-group statistics pruning, which skips the whole row + /// group before any read. + fn dynamic_filter_is_noop(&self, idx: usize) -> bool { + let Some(filter) = self.physical_filters.get(idx) else { + return false; + }; + if !filter.is_dynamic() { + return false; + } + match self.dynamic_snapshots.get(idx) { + Some(Some(snapshot)) => collect_columns(&snapshot.expr).is_empty(), + _ => true, + } + } + + /// Returns true when the producer has not republished the dynamic filter at + /// `physical_filters[idx]` since this pass captured its snapshot. + fn dynamic_snapshot_is_current(&self, idx: usize) -> bool { + let (Some(filter), Some(Some(snapshot))) = ( + self.physical_filters.get(idx), + self.dynamic_snapshots.get(idx), + ) else { + return false; + }; + + filter.dynamic_snapshot_is_current(snapshot) + } +} + /// Pre-built state for constructing [PrefilterContext] per row group. /// /// Fields invariant across row groups (projection mask, codec, metadata, filters) @@ -675,6 +739,10 @@ impl PrefilterContextBuilder { } /// Builds a [PrefilterContext] for a specific row group. + /// + /// This is where a row group's prefilter starts, so producer-owned dynamic filters + /// are snapshotted here: one stable expression per filter, evaluated for the whole + /// pass and rechecked against the producer before its mask is committed. pub(crate) fn build(&self, row_group_idx: usize) -> PrefilterContext { let pk_filter = self .build_primary_key_filter() @@ -683,6 +751,11 @@ impl PrefilterContextBuilder { pk_filter, filters: self.filters.clone(), physical_filters: self.physical_filters.clone(), + dynamic_snapshots: self + .physical_filters + .iter() + .map(|filter| filter.capture_dynamic_snapshot()) + .collect(), schema_version: self.schema_version, pk_filter_expr_strs: self.pk_filter_expr_strs.clone(), arrow_schema: self.arrow_schema.clone(), @@ -935,8 +1008,10 @@ async fn execute_prefilter_with_result_cache( .copied() .map(|idx| PrefilterEntry::without_cache(PrefilterEntryKind::Physical(idx))), ); - let (uncached_mask, read_rows) = + let masks = build_prefilter_masks(prefilter_ctx, reader_builder, build_ctx, &uncached_entries).await?; + let read_rows = masks.rows; + let uncached_mask = masks.commit(prefilter_ctx); let final_mask = match (hit_mask, uncached_mask) { (Some(hit_mask), Some(uncached_mask)) => hit_mask.bitand(&uncached_mask), @@ -960,16 +1035,90 @@ fn non_cacheable_physical_filters(prefilter_ctx: &PrefilterContext) -> Vec, + /// Masks of producer-owned dynamic filters, keyed by their `physical_filters` index. + /// Only commit them while the producer stays at the generation they were snapshotted + /// at. + dynamic_masks: Vec<(usize, BooleanBuffer)>, + /// Number of rows this pass read. + rows: usize, +} + +impl PrefilterMasks { + /// ANDs the masks that may be committed for this row group. + /// + /// A producer may publish a new expression while this pass runs. The committed mask + /// must never be tighter than the live filter, and nothing here orders one generation + /// against another, so a dynamic mask whose generation moved on is discarded and the + /// incoming selection of the row group is kept as it was. Only the live filter above + /// the scan decides which rows a query returns. + /// + /// # Producer contract + /// + /// Committing a dynamic mask takes two properties from the producer: + /// + /// * **Stable snapshot.** The mask was computed from the single expression captured + /// at `generation` and is only committed while the handle is still at that + /// generation (see [`PhysicalFilterContext::capture_dynamic_snapshot`]). + /// * **A publication is a necessary condition for the same execution.** A committed + /// mask removes every row that generation rejects, so a publication is only allowed + /// to reject rows the enclosing join or limit rejects too. The supported producers + /// publish completed state: a `HashJoin` publishes its immutable, completed + /// build-side map, and `TopK` publishes a monotone (only ever tightening) threshold. + /// Arbitrary independently mutable nested expressions are not promised, and nothing + /// here compares two generations: a newer publication is simply re-evaluated by the + /// next pass, while the live filter above the scan stays authoritative. + /// + /// # Why a stale mask is dropped instead of retained + /// + /// For a monotone tightening producer such as `TopK`, the newer expression accepts a + /// subset of the rows the snapshotted one accepts, so keeping the older mask would + /// stay correct and prune more. Nothing observable here tells that producer apart from + /// one whose next publication widens, reverses, or rewrites its expression, and + /// keeping a mask across such a publication would return rows the live filter rejects. + /// A mask whose generation moved on is therefore treated as untrusted rather than + /// assumed safe: the mask is dropped, the incoming selection of the row group is kept, + /// and the next prefilter pass evaluates the new expression from scratch. Retaining a + /// stale mask needs an explicit monotone/tightening contract published by the producer + /// next to its generation; no producer exposes one today, so it can be added together + /// with the producer that relies on it. + fn commit(self, prefilter_ctx: &PrefilterContext) -> Option { + let mut mask = self.static_mask; + for (idx, dynamic_mask) in self.dynamic_masks { + // The producer moved on. Keeping this mask would assume the new expression only + // tightens the snapshotted one, which nothing here proves. + if !prefilter_ctx.dynamic_snapshot_is_current(idx) { + continue; + } + mask = Some(match mask { + Some(mask) => mask.bitand(&dynamic_mask), + None => dynamic_mask, + }); + } + + mask + } +} + async fn build_prefilter_masks( prefilter_ctx: &mut PrefilterContext, reader_builder: &RowGroupReaderBuilder, build_ctx: &RowGroupBuildContext<'_>, entries: &[PrefilterEntry], -) -> Result<(Option, usize)> { +) -> Result { let prefilter_column_names = prefilter_column_names_for_entries(prefilter_ctx, entries); let parquet_schema = reader_builder .parquet_metadata() @@ -994,7 +1143,14 @@ async fn build_prefilter_masks( .iter() .map(|entry| entry.key.is_some().then(|| BooleanBufferBuilder::new(0))) .collect::>(); - let mut combined_builder = (!entries.is_empty()).then(|| BooleanBufferBuilder::new(0)); + // Dynamic filter masks are kept per filter, so the commit step can still drop one + // whose generation moved on without discarding the others. + let mut dynamic_builders = prefilter_ctx + .dynamic_snapshots + .iter() + .map(|snapshot| snapshot.is_some().then(|| BooleanBufferBuilder::new(0))) + .collect::>(); + let mut combined_builder = BooleanBufferBuilder::new(0); let mut rows_before_filter = 0usize; while let Some(batch_result) = stream.next().await { @@ -1013,14 +1169,21 @@ async fn build_prefilter_masks( entry.kind, reader_builder.file_path(), )?; - batch_mask = batch_mask.bitand(&mask); + let dynamic_builder = match entry.kind { + PrefilterEntryKind::Physical(physical_idx) => dynamic_builders + .get_mut(physical_idx) + .and_then(Option::as_mut), + PrefilterEntryKind::Simple(_) | PrefilterEntryKind::PkGroup => None, + }; + match dynamic_builder { + Some(builder) => builder.append_buffer(&mask), + None => batch_mask = batch_mask.bitand(&mask), + } if let Some(Some(builder)) = cache_builders.get_mut(idx) { builder.append_buffer(&mask); } } - if let Some(builder) = &mut combined_builder { - builder.append_buffer(&batch_mask); - } + combined_builder.append_buffer(&batch_mask); } for (entry, builder) in entries.iter().zip(cache_builders) { @@ -1031,10 +1194,18 @@ async fn build_prefilter_masks( } } - Ok(( - combined_builder.map(|mut builder| builder.finish()), - rows_before_filter, - )) + Ok(PrefilterMasks { + static_mask: Some(combined_builder.finish()) + .filter(|mask: &BooleanBuffer| !mask.is_empty()), + dynamic_masks: dynamic_builders + .into_iter() + .enumerate() + .filter_map(|(idx, builder)| builder.map(|mut builder| (idx, builder.finish()))) + // A dynamic filter without a usable snapshot reads no row for this pass. + .filter(|(_, mask)| !mask.is_empty()) + .collect(), + rows: rows_before_filter, + }) } fn prefilter_column_names_for_entries( @@ -1073,8 +1244,9 @@ async fn execute_prefilter_by_reading_columns( if entries.is_empty() { return Ok(identity_prefilter_result(reader_builder, build_ctx)); } - let (mask, rows_before_filter) = - build_prefilter_masks(prefilter_ctx, reader_builder, build_ctx, &entries).await?; + let masks = build_prefilter_masks(prefilter_ctx, reader_builder, build_ctx, &entries).await?; + let rows_before_filter = masks.rows; + let mask = masks.commit(prefilter_ctx); let final_mask = mask.unwrap_or_else(|| BooleanBuffer::new_set(rows_before_filter)); let rows_selected = final_mask.count_set_bits(); @@ -1111,6 +1283,7 @@ fn all_prefilter_entries(prefilter_ctx: &PrefilterContext) -> Vec { - eval_physical_filter_mask(batch, &prefilter_ctx.physical_filters[idx], file_path) + let filter_ctx = &prefilter_ctx.physical_filters[idx]; + let Some(snapshot) = prefilter_ctx + .dynamic_snapshots + .get(idx) + .and_then(Option::as_ref) + else { + // A static filter; a dynamic filter without a usable snapshot keeps all + // rows instead of being evaluated at its live, ever-changing generation. + let Some(filter) = filter_ctx.filter() else { + return Ok(BooleanBuffer::new_set(batch.num_rows())); + }; + return evaluate_physical_mask(batch, filter_ctx, filter, file_path); + }; + // A producer-owned expression is only checked against the prefilter schema + // when it is snapshotted. Keep every row if evaluating it fails, exactly like + // any other unsupported prefilter candidate. + match evaluate_physical_mask(batch, filter_ctx, &snapshot.expr, file_path) { + Ok(mask) => Ok(mask), + Err(error) => { + // The producer owns the expression, so a snapshot that cannot be + // evaluated is a miss for this pass, not a scan error. + debug!( + "Ignored dynamic prefilter snapshot of {} that failed to evaluate: {}, file: {}", + filter_ctx.column_name(), + error, + file_path + ); + Ok(BooleanBuffer::new_set(batch.num_rows())) + } + } } PrefilterEntryKind::PkGroup => { let pk_filter = prefilter_ctx.pk_filter.as_mut().context(UnexpectedSnafu { @@ -1322,13 +1524,18 @@ fn eval_simple_filter_mask( filter.evaluate_array(&column).context(RecordBatchSnafu) } -fn eval_physical_filter_mask( +/// Evaluates a physical prefilter expression against the single filter column of `batch`. +/// +/// `NULL` results count as filtered out. This is safe because the published predicate is +/// already a necessary condition for the enclosing operation; legacy consumer-side NULL +/// guards may admit additional rows that the operation does not need. No `IS NULL` guard +/// is added here because the producer owns the null semantics of its expression. +pub(crate) fn evaluate_physical_mask( batch: &RecordBatch, filter_ctx: &PhysicalFilterContext, + filter: &Arc, file_path: &str, ) -> Result { - let filter = filter_ctx.filter(); - let (idx, _) = batch .schema() .column_with_name(filter_ctx.column_name()) @@ -1371,12 +1578,15 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; use bytes::Bytes; - use common_recordbatch::filter::SimpleFilterEvaluator; + use common_recordbatch::filter::{SimpleFilterEvaluator, batch_filter}; + use datafusion::physical_plan::expressions::{ + Column, DynamicFilterPhysicalExpr, lit as physical_lit, + }; use datafusion_common::ScalarValue; use datafusion_expr::{col, lit}; use datatypes::arrow::array::{ - ArrayRef, DictionaryArray, Int32Array, TimestampMillisecondArray, UInt8Array, UInt32Array, - UInt64Array, + ArrayRef, DictionaryArray, Float64Array, Int32Array, TimestampMillisecondArray, UInt8Array, + UInt32Array, UInt64Array, }; use datatypes::arrow::datatypes::{DataType, Field, Schema, UInt32Type}; use datatypes::arrow::record_batch::RecordBatch; @@ -2011,6 +2221,7 @@ mod tests { pk_filter: None, filters, physical_filters: Vec::new(), + dynamic_snapshots: Vec::new(), schema_version: metadata.schema_version, pk_filter_expr_strs: None, arrow_schema: metadata.schema.arrow_schema().clone(), @@ -2027,6 +2238,102 @@ mod tests { )); } + /// A generation change discards the dynamic mask even when the new expression is a + /// monotone tightening of the snapshotted one. + /// + /// A `TopK` only ever tightens its threshold, so the mask of the generation a pass + /// snapshotted would still be implied by the newer expression and keeping it would + /// prune more. Nothing here can tell such a producer apart from one whose next + /// publication widens, so a mask from a superseded generation is treated as untrusted + /// and dropped: the row group keeps its incoming selection and the next pass evaluates + /// the new expression from scratch. + #[test] + fn test_commit_discards_dynamic_mask_after_monotone_tightening_publication() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = FlatReadFormat::new( + metadata.clone(), + ReadColumns::new(metadata.column_metadatas.iter().map(|c| c.column_id)), + None, + "memtable", + false, + ) + .unwrap(); + let producer_schema = Arc::new(Schema::new(vec![Field::new( + "field_0", + DataType::UInt64, + true, + )])); + let publish = |members: Vec| { + let expr = col("field_0").in_list(members.into_iter().map(lit).collect(), false); + Predicate::to_physical_expr(&expr, &producer_schema).unwrap() + }; + let key = Arc::new(Column::new("field_0", 0)) as Arc; + // The producer starts from `{1, 2}` and only ever tightens the membership below. + let handle = Arc::new(DynamicFilterPhysicalExpr::new( + vec![key], + physical_lit(true), + )); + handle.update(publish(vec![1, 2])).unwrap(); + + // A prefilter pass of three rows that snapshots the current generation. + let new_context = || { + let filter = + PhysicalFilterContext::new_dynamic(&metadata, None, &read_format, &handle).unwrap(); + let snapshot = filter.capture_dynamic_snapshot(); + PrefilterContext { + pk_filter: None, + filters: Vec::new(), + physical_filters: vec![filter], + dynamic_snapshots: vec![snapshot], + schema_version: metadata.schema_version, + pk_filter_expr_strs: None, + arrow_schema: read_format.arrow_schema().clone(), + proven_simple_filters: Vec::new(), + } + }; + let mask_bits = |mask: BooleanBuffer| mask.iter().collect::>(); + let keep_all = BooleanBuffer::new_set(3); + // `{1, 2}` keeps rows 1 and 2 of the three rows the pass read. + let membership_12 = BooleanBuffer::from(vec![false, true, true]); + + let context = new_context(); + let committed = PrefilterMasks { + static_mask: Some(keep_all.clone()), + dynamic_masks: vec![(0, membership_12.clone())], + rows: 3, + } + .commit(&context); + assert_eq!( + vec![false, true, true], + mask_bits(committed.unwrap()), + "a mask of the snapshotted generation is committed" + ); + + // Tightening `{1, 2}` to `{2}` supersedes the generation the pass snapshotted. + handle.update(publish(vec![2])).unwrap(); + let committed = PrefilterMasks { + static_mask: Some(keep_all.clone()), + dynamic_masks: vec![(0, membership_12.clone())], + rows: 3, + } + .commit(&context); + assert_eq!( + vec![true, true, true], + mask_bits(committed.unwrap()), + "the superseded mask must be dropped, not retained as a tightening" + ); + + // The next pass snapshots the tightened generation and prunes with it. + let context = new_context(); + let committed = PrefilterMasks { + static_mask: Some(keep_all), + dynamic_masks: vec![(0, BooleanBuffer::from(vec![false, false, true]))], + rows: 3, + } + .commit(&context); + assert_eq!(vec![false, false, true], mask_bits(committed.unwrap())); + } + #[test] fn test_identity_row_selection_preserves_input() { let sparse = RowSelection::from(vec![ @@ -2242,6 +2549,116 @@ mod tests { ); } + /// Creates the dynamic filter a single-key `HashJoin` publishes for `members`. + fn new_dynamic_field_filter( + read_format: &FlatReadFormat, + members: Vec, + ) -> Arc { + let arrow_schema = read_format.arrow_schema(); + let key = Arc::new(Column::new("field_0", 0)) as Arc; + let handle = Arc::new(DynamicFilterPhysicalExpr::new( + vec![key], + Predicate::to_physical_expr(&lit(true), arrow_schema).unwrap(), + )); + let expr = col("field_0").in_list(members.into_iter().map(lit).collect(), false); + handle + .update(Predicate::to_physical_expr(&expr, arrow_schema).unwrap()) + .unwrap(); + handle + } + + fn dynamic_filter_count(builder: &PrefilterContextBuilder) -> usize { + builder + .physical_filters + .iter() + .filter(|filter| filter.is_dynamic()) + .count() + } + + /// A dynamic filter on a field column is admitted only where field pruning is safe. + /// `SkipFields` preserves the dedup semantics of a multi-source `LastRow` / + /// `LastNonNull` merge, so the dynamic filter must not prune fields there either. + #[test] + fn test_build_reader_filter_plan_gates_dynamic_filters_by_prefilter_mode() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = FlatReadFormat::new( + metadata.clone(), + ReadColumns::new(metadata.column_metadatas.iter().map(|c| c.column_id)), + None, + "test", + true, + ) + .unwrap(); + let codec = build_primary_key_codec(metadata.as_ref()); + let parquet_metadata = stats_metadata(&[vec![1]], None); + let predicate = || { + Predicate::with_dyn_filters( + vec![], + vec![new_dynamic_field_filter(&read_format, vec![1])], + ) + }; + + let all_plan = build_reader_filter_plan( + Some(&predicate()), + None, + PreFilterMode::All, + true, + false, + &read_format, + &codec, + &parquet_metadata, + ); + assert_eq!( + 1, + dynamic_filter_count(all_plan.prefilter_builder.as_ref().unwrap()) + ); + + // The dynamic filter is the only prefilter candidate, and it is gated out. + let skip_fields_plan = build_reader_filter_plan( + Some(&predicate()), + None, + PreFilterMode::SkipFields, + true, + false, + &read_format, + &codec, + &parquet_metadata, + ); + assert!(skip_fields_plan.prefilter_builder.is_none()); + + // With a tag filter the prefilter still runs, without the dynamic field filter. + let skip_fields_plan = build_reader_filter_plan( + Some(&Predicate::with_dyn_filters( + vec![col("tag_0").eq(lit("a"))], + vec![new_dynamic_field_filter(&read_format, vec![1])], + )), + None, + PreFilterMode::SkipFields, + true, + false, + &read_format, + &codec, + &parquet_metadata, + ); + assert_eq!( + 0, + dynamic_filter_count(skip_fields_plan.prefilter_builder.as_ref().unwrap()) + ); + + // Disabled predicate prefiltering keeps dynamic filters on the normal path. + let disabled_plan = build_reader_filter_plan( + Some(&predicate()), + None, + PreFilterMode::All, + false, + false, + &read_format, + &codec, + &parquet_metadata, + ); + assert!(disabled_plan.prefilter_builder.is_none()); + } + #[test] fn test_pk_filter_expr_strings_are_stable_under_expr_order() { let metadata: RegionMetadataRef = Arc::new(sst_region_metadata_with_encoding( @@ -2348,10 +2765,177 @@ mod tests { let pk = new_primary_key(&["a", "x"]); let batch = new_raw_batch(&[pk.as_slice(), pk.as_slice(), pk.as_slice()], &[9, 10, 11]); - let mask = eval_physical_filter_mask(&batch, &physical_filters[0], "test").unwrap(); + let filter = physical_filters[0].filter().unwrap(); + let mask = evaluate_physical_mask(&batch, &physical_filters[0], filter, "test").unwrap(); assert_eq!(mask.count_set_bits(), 1); } + /// The producer's input schema, where `field_0` is its fourth column rather than the + /// only column of a prefilter batch. + fn producer_schema(field_type: DataType) -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("tag_0", DataType::Utf8, true), + Field::new("tag_1", DataType::Utf8, true), + Field::new("ts", DataType::UInt64, true), + Field::new("field_0", field_type, true), + ])) + } + + fn full_read_format(metadata: &RegionMetadataRef) -> FlatReadFormat { + FlatReadFormat::new( + metadata.clone(), + ReadColumns::new(metadata.column_metadatas.iter().map(|c| c.column_id)), + None, + "test", + false, + ) + .unwrap() + } + + /// Evaluates a producer-owned dynamic filter over `batch` both ways: as a prefilter + /// mask, and as the live `batch_filter` that runs above the scan. + /// + /// `batch` carries the filter column plus a `row` column holding row indexes, so the + /// rows the upper filter keeps come back as values. Both evaluations see the same + /// rows, so the returned row lists are directly comparable. + fn prefilter_and_upper_rows( + metadata: &RegionMetadataRef, + read_format: &FlatReadFormat, + producer_schema: &SchemaRef, + published: &Expr, + batch: &RecordBatch, + ) -> (Vec, Vec) { + let key = Arc::new(Column::new("field_0", 3)) as Arc; + let handle = Arc::new(DynamicFilterPhysicalExpr::new( + vec![key], + Predicate::to_physical_expr(&lit(true), producer_schema).unwrap(), + )); + handle + .update(Predicate::to_physical_expr(published, producer_schema).unwrap()) + .unwrap(); + let filter_ctx = + PhysicalFilterContext::new_dynamic(metadata, None, read_format, &handle).unwrap(); + let snapshot = filter_ctx.capture_dynamic_snapshot().unwrap(); + + let mask = evaluate_physical_mask(batch, &filter_ctx, &snapshot.expr, "test").unwrap(); + let prefilter_rows = mask + .iter() + .enumerate() + .filter_map(|(row, keep)| keep.then_some(row)) + .collect::>(); + + let filtered = batch_filter(batch, &snapshot.expr).unwrap(); + let upper_rows = filtered + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .iter() + .map(|row| *row as usize) + .collect::>(); + + (prefilter_rows, upper_rows) + } + + /// A batch of `field_0` values plus a `row` column holding their indexes. + fn indexed_batch(field: ArrayRef, data_type: DataType) -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("field_0", data_type, true), + Field::new("row", DataType::UInt64, false), + ])); + let rows = Arc::new(UInt64Array::from_iter_values(0..field.len() as u64)) as ArrayRef; + RecordBatch::try_new(schema, vec![field, rows]).unwrap() + } + + /// A bare `col IN (...)` leaves a NULL comparison UNKNOWN, and the prefilter must + /// reject that row exactly like the live filter above the scan: it adds no NULL guard + /// of its own, because the producer owns the NULL semantics of its expression. + #[test] + fn test_dynamic_filter_mask_matches_upper_filter_for_nullable_bare_in() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = full_read_format(&metadata); + let published = col("field_0").in_list(vec![lit(1_u64), lit(3_u64)], false); + // Row 1 is NULL, and `NULL IN (1, 3)` is neither a member nor a non-member. + let batch = indexed_batch( + Arc::new(UInt64Array::from(vec![Some(1), None, Some(2), Some(3)])) as ArrayRef, + DataType::UInt64, + ); + + let (prefilter_rows, upper_rows) = prefilter_and_upper_rows( + &metadata, + &read_format, + &producer_schema(DataType::UInt64), + &published, + &batch, + ); + + assert_eq!(vec![0, 3], prefilter_rows); + assert_eq!( + prefilter_rows, upper_rows, + "the prefilter must keep exactly the rows the upper filter keeps" + ); + } + + /// Float membership compares with a total order, so `NaN` is a member of a list that + /// holds `NaN` while `-0.0` is not a member of a list that holds `0.0`. The prefilter + /// must reproduce those semantics exactly, which it can only do by evaluating the + /// producer's whole expression instead of substituting an IEEE `==` or a re-derived + /// comparison. + #[test] + fn test_dynamic_filter_mask_matches_upper_filter_for_float_nan_membership() { + let metadata = metadata_with_field_type(ConcreteDataType::float64_datatype()); + let read_format = full_read_format(&metadata); + let published = col("field_0").in_list(vec![lit(0.0_f64), lit(f64::NAN)], false); + // Rows: `0.0` and `NaN` are members, `-0.0` is not (total order, not IEEE `==`), + // and the NULL row is UNKNOWN. + let batch = indexed_batch( + Arc::new(Float64Array::from(vec![ + Some(0.0), + Some(-0.0), + Some(f64::NAN), + None, + ])) as ArrayRef, + DataType::Float64, + ); + let unlisted = indexed_batch( + Arc::new(Float64Array::from(vec![ + Some(1.5), + Some(f64::NAN), + Some(0.0), + None, + ])) as ArrayRef, + DataType::Float64, + ); + + let (prefilter_rows, upper_rows) = prefilter_and_upper_rows( + &metadata, + &read_format, + &producer_schema(DataType::Float64), + &published, + &batch, + ); + + assert_eq!( + prefilter_rows, upper_rows, + "the prefilter must keep exactly the rows the upper filter keeps" + ); + assert_eq!(vec![0, 2], prefilter_rows); + + let (unlisted_prefilter_rows, unlisted_upper_rows) = prefilter_and_upper_rows( + &metadata, + &read_format, + &producer_schema(DataType::Float64), + &published, + &unlisted, + ); + assert_eq!( + unlisted_prefilter_rows, unlisted_upper_rows, + "the prefilter must keep exactly the rows the upper filter keeps" + ); + assert_eq!(vec![1, 2], unlisted_prefilter_rows); + } + #[test] fn test_eval_pk_group_mask_finds_pk_column_by_name() { let metadata = Arc::new(sst_region_metadata()); diff --git a/src/mito2/src/sst/parquet/reader.rs b/src/mito2/src/sst/parquet/reader.rs index e92ac60f93b..f395e48bf58 100644 --- a/src/mito2/src/sst/parquet/reader.rs +++ b/src/mito2/src/sst/parquet/reader.rs @@ -25,7 +25,9 @@ use arrow_schema::extension::{ use common_recordbatch::QueryMemoryTracker; use common_recordbatch::filter::{SimpleFilterEvaluator, TimestampUnitCast}; use common_telemetry::{debug, error, tracing, warn}; +use datafusion::physical_expr::utils::{collect_columns, reassign_expr_columns}; use datafusion::physical_plan::PhysicalExpr; +use datafusion::physical_plan::expressions::DynamicFilterPhysicalExpr; use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; use datafusion_expr::utils::expr_to_columns; use datafusion_expr::{Expr, Volatility}; @@ -2159,7 +2161,12 @@ impl SimpleFilterContext { #[derive(Clone)] pub(crate) struct PhysicalFilterContext { /// Filter to evaluate. - filter: Arc, + /// + /// `None` for a producer-owned dynamic filter, whose expression changes while the + /// scan runs and is therefore captured per prefilter pass. + filter: Option>, + /// Producer-owned dynamic filter to snapshot for each prefilter pass. + dynamic: Option>, /// Debug string of the original logical expression. expr_str: String, /// Id of the column to evaluate. @@ -2170,10 +2177,24 @@ pub(crate) struct PhysicalFilterContext { semantic_type: SemanticType, /// Schema containing only the referenced column. schema: SchemaRef, - /// Whether the original logical expression is immutable across queries. + /// Whether the filter is immutable across queries. immutable: bool, } +/// A stable snapshot of the expression a producer published for a dynamic filter. +/// +/// The expression is rebound to the single-column prefilter schema and is only +/// trustworthy while the producer stays at `generation`. See +/// [`PhysicalFilterContext::capture_dynamic_snapshot`] for the producer contract this +/// snapshot depends on. +#[derive(Clone)] +pub(crate) struct DynamicFilterSnapshot { + /// Captured expression, rebound to the prefilter schema. + pub(crate) expr: Arc, + /// Producer generation the expression was captured at. + generation: u64, +} + impl PhysicalFilterContext { /// Creates a context for the `expr`. /// @@ -2190,27 +2211,7 @@ impl PhysicalFilterContext { } let expr_str = format!("{expr:?}"); let column_name = Self::single_column_name(expr)?; - let column_metadata = match expected_meta { - Some(meta) => { - let column = meta.column_by_name(&column_name)?; - let sst_column = sst_meta.column_by_id(column.column_id)?; - // Physical expr requires the column name to match the SST column name. - if sst_column.column_schema.name != column_name { - return None; - } - // Schema evolution (an altered field column, or a widened time - // index unit) can make the column's file type differ from the - // expected type; evaluating the original expr against the - // file's column would raise a cross-type comparison error - // (e.g. Timestamp(ms) >= Timestamp(µs)). Drop the prefilter; - // the query layer re-applies the predicate above the scan. - if sst_column.column_schema.data_type != column.column_schema.data_type { - return None; - } - column - } - None => sst_meta.column_by_name(&column_name)?, - }; + let column_metadata = Self::validated_column(sst_meta, expected_meta, &column_name)?; // The column must be present in the projected arrow schema for the // prefilter to be able to read it. @@ -2225,7 +2226,8 @@ impl PhysicalFilterContext { let immutable = expr_is_immutable(expr); Some(Self { - filter: physical_expr, + filter: Some(physical_expr), + dynamic: None, expr_str, column_id: column_metadata.column_id, column_name, @@ -2235,6 +2237,167 @@ impl PhysicalFilterContext { }) } + /// Creates a context for a producer-owned dynamic filter. + /// + /// Unlike [`Self::new_opt`], the expression is built by the producer (for example a + /// single-key `HashJoin`) against its own input schema, so this constructor admits only + /// filters whose declared dependencies are the filter column itself, and rebinds that + /// column to the single-column prefilter schema. It does not restrict the shape of the + /// published expression: the whole expression is evaluated as-is. + /// + /// Returns `None` when the dependencies are not exactly the filter column, or when + /// the SST and the expected metadata disagree on the column's identity or type. The + /// prefilter then keeps every row. + pub(crate) fn new_dynamic( + sst_meta: &RegionMetadataRef, + expected_meta: Option<&RegionMetadata>, + read_format: &FlatReadFormat, + handle: &Arc, + ) -> Option { + // The columns a dynamic filter may reference are fixed when it is created, so + // they decide admission before any expression is published. + let column_name = Self::single_dynamic_column(handle)?; + let column_metadata = Self::validated_column(sst_meta, expected_meta, &column_name)?; + let (_, field) = read_format.arrow_schema().column_with_name(&column_name)?; + // The prefilter batch must carry the type the producer built its expression + // against. A changed type (an altered column, or a tag column that the flat + // format dictionary-encodes) would fail the comparison or silently compare + // across types. + if *field.data_type() != column_metadata.column_schema.data_type.as_arrow_type() { + return None; + } + let schema = Arc::new(ArrowSchema::new(vec![field.clone()])); + + Some(Self { + filter: None, + dynamic: Some(Arc::clone(handle)), + // A stable debug string: the published expression changes between generations. + expr_str: format!( + "dynamic filter {:?} on {}", + handle.expression_id(), + column_name + ), + column_id: column_metadata.column_id, + column_name, + semantic_type: column_metadata.semantic_type, + schema, + // The producer republishes the expression while the scan runs. + immutable: false, + }) + } + + /// Captures the current expression of a producer-owned dynamic filter. + /// + /// Returns `None` for a static filter, and whenever no trustworthy snapshot exists: + /// the producer updated the filter while the snapshot was taken (the generation read + /// before and after [`DynamicFilterPhysicalExpr::current`] differs), or the published + /// expression cannot be rebound to the single-column prefilter schema, which includes + /// every reference outside this filter's column. Callers must keep every row then. + /// + /// # Producer contract + /// + /// The snapshot is evaluated as-is for a whole row-group prefilter pass, so the + /// published expression has to be usable on its own: + /// + /// * **A publication is a stable snapshot.** It is a value the producer hands over, + /// not a view of state the producer keeps mutating: one expression is captured per + /// pass and evaluated against every batch of that pass. + /// * **A publication is a necessary condition for the same execution.** The prefilter + /// only ever drops rows its snapshot rejects, so a publication that accepts every + /// row the enclosing join or limit keeps can never change the query result, however + /// loose it is. A publication must therefore never reject a row the execution's own + /// semantics keeps. + /// * **Only this handle's generation is observed.** A + /// [`DynamicFilterPhysicalExpr`] nested inside the published expression that + /// republishes on its own is invisible to + /// [`Self::dynamic_snapshot_is_current`]; such an expression is not promised to be + /// snapshotted at all. + /// + /// The supported producers publish completed, self-contained state: a `HashJoin` + /// publishes its immutable, completed build-side map (the build side does not change + /// once the filter is published), and `TopK` publishes its monotone (only ever + /// tightening) threshold as a fresh comparison expression. Arbitrary independently + /// mutable nested expressions are not promised: the snapshot is then either rejected + /// here and the pass keeps every row, or the producer's own generation keeps it + /// honest. The live filter above the scan stays authoritative either way. + pub(crate) fn capture_dynamic_snapshot(&self) -> Option { + let handle = self.dynamic.as_ref()?; + let generation = handle.snapshot_generation(); + let expr = handle.current().ok()?; + if handle.snapshot_generation() != generation { + return None; + } + let expr = reassign_expr_columns(expr, &self.schema).ok()?; + + Some(DynamicFilterSnapshot { expr, generation }) + } + + /// Returns true when the producer has not published a newer expression since + /// `snapshot` was captured. + pub(crate) fn dynamic_snapshot_is_current(&self, snapshot: &DynamicFilterSnapshot) -> bool { + self.dynamic + .as_ref() + .is_some_and(|handle| handle.snapshot_generation() == snapshot.generation) + } + + /// Returns true for a producer-owned dynamic filter. + pub(crate) fn is_dynamic(&self) -> bool { + self.dynamic.is_some() + } + + /// Resolves the column a filter reads after checking that the SST and the expected + /// metadata agree on its identity (id and name) and type. + fn validated_column<'a>( + sst_meta: &'a RegionMetadataRef, + expected_meta: Option<&'a RegionMetadata>, + column_name: &str, + ) -> Option<&'a ColumnMetadata> { + match expected_meta { + Some(meta) => { + // Gets the column metadata from the expected metadata. + let column = meta.column_by_name(column_name)?; + // Checks if the column is present in the SST metadata. We still uses the + // column from the expected metadata. + let sst_column = sst_meta.column_by_id(column.column_id)?; + debug_assert_eq!(column.semantic_type, sst_column.semantic_type); + // Physical exprs reference the column by the expected name; a renamed + // column keeps its old name in the SST. + if sst_column.column_schema.name != column_name { + return None; + } + // Schema evolution (an altered field column, or a widened time index + // unit) can make columns with the same id have different types across + // SSTs; evaluating the expression against the file's column would raise + // a cross-type comparison error. Drop the prefilter; the query layer + // re-applies the predicate above the scan. + if sst_column.column_schema.data_type != column.column_schema.data_type { + return None; + } + Some(column) + } + None => sst_meta.column_by_name(column_name), + } + } + + /// Returns the single column all declared dependencies of a dynamic filter read. + /// + /// A multi-key producer publishes one filter over several key columns; such a filter + /// cannot be evaluated on a single-column prefilter batch, so it is rejected. + fn single_dynamic_column(handle: &DynamicFilterPhysicalExpr) -> Option { + let mut column_name = None; + for child in handle.children() { + for column in collect_columns(child) { + match &column_name { + Some(name) if *name != column.name() => return None, + Some(_) => {} + None => column_name = Some(column.name().to_string()), + } + } + } + + column_name + } + /// Returns true if the expression is a variant we want to evaluate as a /// physical prefilter. Binary exprs are intentionally excluded because /// [`SimpleFilterEvaluator`] already handles them. @@ -2266,8 +2429,12 @@ impl PhysicalFilterContext { } /// Returns the filter to evaluate. - pub(crate) fn filter(&self) -> &Arc { - &self.filter + /// + /// `None` for a producer-owned dynamic filter: its expression changes while the scan + /// runs, so callers capture a snapshot with [`Self::capture_dynamic_snapshot`] + /// instead of evaluating this field. + pub(crate) fn filter(&self) -> Option<&Arc> { + self.filter.as_ref() } /// Returns the original logical expression string. @@ -2498,12 +2665,20 @@ mod tests { use common_function::scalars::json::json_get::JsonGetWithType; use common_function::scalars::udf::create_udf; use common_recordbatch::ext::RecordBatchExt; + use common_recordbatch::filter::batch_filter; use datafusion::arrow::datatypes::DataType; + use datafusion::physical_plan::expressions::{ + BinaryExpr, CaseExpr, Column, lit as physical_lit, + }; + use datafusion::physical_plan::joins::join_hash_map::{JoinHashMapType, JoinHashMapU32}; + use datafusion::physical_plan::joins::{HashExpr, HashTableLookupExpr, Map, SeededRandomState}; + use datafusion::physical_plan::repartition::REPARTITION_RANDOM_STATE; use datafusion_common::ScalarValue; + use datafusion_common::hash_utils::create_hashes; use datafusion_expr::expr::ScalarFunction; use datafusion_expr::{ - ColumnarValue, Expr, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, - col, lit, + ColumnarValue, Expr, Operator, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, + Volatility, col, lit, }; use datatypes::arrow::array::{ ArrayRef, BinaryDictionaryBuilder, Int64Array, StringArray, StructArray, @@ -2530,10 +2705,14 @@ mod tests { use crate::cache::CacheManager; use crate::cache::test_util::{GatedStorageReader, parquet_file_data_with_region_metadata}; use crate::sst::parquet::metadata::MetadataLoader; - use crate::sst::parquet::prefilter::{build_reader_filter_plan, execute_prefilter}; + use crate::sst::parquet::prefilter::{ + build_reader_filter_plan, evaluate_physical_mask, execute_prefilter, + }; use crate::sst::parquet::read_columns::{ParquetReadColumn, ParquetReadColumns}; use crate::sst::parquet::row_group::ParquetFetchMetrics; - use crate::test_util::sst_util::{sst_file_handle, sst_region_metadata}; + use crate::test_util::sst_util::{ + new_record_batch_with_custom_sequence, sst_file_handle, sst_region_metadata, + }; #[tokio::test] async fn test_shared_file_metadata_resolution_counts_without_duplicate_io() { @@ -2802,6 +2981,30 @@ mod tests { object_store: ObjectStore, predicate: Predicate, cache_strategy: CacheStrategy, + ) -> (RowGroupReaderBuilder, Arc) { + prefilter_test_builder_with( + object_store, + 3, + 2, + PageIndexPolicy::Skip, + None, + predicate, + cache_strategy, + ) + .await + } + + /// Builds a reader over one SST with `num_row_groups` row groups of + /// `rows_per_group` rows. `writer_props` lets a test write several pages per row + /// group. + async fn prefilter_test_builder_with( + object_store: ObjectStore, + rows_per_group: i64, + num_row_groups: i64, + page_index_policy: PageIndexPolicy, + writer_props: Option, + predicate: Predicate, + cache_strategy: CacheStrategy, ) -> (RowGroupReaderBuilder, Arc) { let metadata = Arc::new( crate::test_util::sst_util::sst_region_metadata_with_encoding( @@ -2843,27 +3046,26 @@ mod tests { ) .unwrap() }; - let first_batch = batch(0, 3); - let second_batch = batch(3, 6); + let first_batch = batch(0, rows_per_group); let mut bytes = Vec::new(); - let mut writer = ArrowWriter::try_new(&mut bytes, first_batch.schema(), None).unwrap(); - writer.write(&first_batch).unwrap(); - writer.flush().unwrap(); - writer.write(&second_batch).unwrap(); + let mut writer = + ArrowWriter::try_new(&mut bytes, first_batch.schema(), writer_props).unwrap(); + for row_group in 0..num_row_groups { + let start = row_group * rows_per_group; + writer.write(&batch(start, start + rows_per_group)).unwrap(); + writer.flush().unwrap(); + } writer.close().unwrap(); - let file_handle = sst_file_handle(0, 6); + let file_handle = sst_file_handle(0, num_row_groups * rows_per_group); let file_path = file_handle.file_path("prefilter_test", PathType::Bare); let file_size = bytes.len() as u64; object_store.write(&file_path, bytes).await.unwrap(); let mut cache_metrics = MetadataCacheMetrics::default(); - let parquet_meta = Arc::new( - MetadataLoader::new(object_store.clone(), &file_path, file_size) - .load(&mut cache_metrics) - .await - .unwrap(), - ); + let mut loader = MetadataLoader::new(object_store.clone(), &file_path, file_size); + loader.with_page_index_policy(page_index_policy); + let parquet_meta = Arc::new(loader.load(&mut cache_metrics).await.unwrap()); let read_format = FlatReadFormat::new( metadata.clone(), ReadColumns::new( @@ -2926,6 +3128,9 @@ mod tests { ) } + /// A row group whose filters are already proven by its statistics keeps the incoming + /// selection, and the prefilter pass requests no pages or ranges for it: every fetch + /// counter stays at zero. #[tokio::test(flavor = "current_thread")] async fn test_execute_prefilter_proven_filters_preserve_selection_without_fetching() { let object_store = ObjectStore::new(Memory::default()).unwrap(); @@ -2965,6 +3170,8 @@ mod tests { assert_eq!(result.refined_selection, expected); assert_eq!(result.filtered_rows, 0); let metrics = fetch_metrics.data.lock().unwrap(); + // `pages_to_fetch_*` count the page ranges requested per source, so these + // are store, mem-cache and write-cache range counts rather than pages. assert_eq!(metrics.pages_to_fetch_store, 0); assert_eq!(metrics.pages_to_fetch_mem, 0); assert_eq!(metrics.pages_to_fetch_write_cache, 0); @@ -3013,6 +3220,8 @@ mod tests { RowSelection::from(vec![RowSelector::select(1), RowSelector::skip(2),]) ); if pass == 1 { + // The cached pass adds no store range count: the second pass reuses the + // cached masks instead of reading the row group again. assert_eq!(fetch_metrics.data.lock().unwrap().pages_to_fetch_store, 0); } } @@ -3047,6 +3256,495 @@ mod tests { ); } + /// Builds a dynamic filter over `field_0` that a `HashJoin` starts as `true` and + /// republishes with the membership of its build side. + fn new_test_dynamic_filter() -> Arc { + let key = Arc::new(Column::new("field_0", 0)) as Arc; + Arc::new(DynamicFilterPhysicalExpr::new( + vec![key], + physical_lit(true), + )) + } + + fn publish_membership( + handle: &Arc, + producer_schema: &SchemaRef, + members: Vec, + ) { + let expr = col("field_0").in_list(members.into_iter().map(lit).collect(), false); + let published = Predicate::to_physical_expr(&expr, producer_schema).unwrap(); + handle.update(published).unwrap(); + } + + async fn execute_test_prefilter( + reader_builder: &RowGroupReaderBuilder, + row_group_idx: usize, + ) -> crate::sst::parquet::prefilter::PrefilterResult { + execute_test_prefilter_with_metrics(reader_builder, row_group_idx) + .await + .0 + } + + /// Runs one prefilter pass over a whole row group and keeps the fetch metrics of that + /// pass, so a test can tell a pass that read the row group from one served by the + /// prefilter-result cache. + async fn execute_test_prefilter_with_metrics( + reader_builder: &RowGroupReaderBuilder, + row_group_idx: usize, + ) -> ( + crate::sst::parquet::prefilter::PrefilterResult, + ParquetFetchMetrics, + ) { + let fetch_metrics = ParquetFetchMetrics::default(); + let prefilter_builder = reader_builder.prefilter_builder.as_ref().unwrap(); + let mut prefilter_ctx = prefilter_builder.build(row_group_idx); + let result = execute_prefilter( + &mut prefilter_ctx, + reader_builder, + &RowGroupBuildContext { + row_group_idx, + row_selection: None, + fetch_metrics: Some(&fetch_metrics), + }, + ) + .await + .unwrap(); + (result, fetch_metrics) + } + + /// A dynamic filter starts as `true` and is published after the plan and the reader + /// are constructed. Each row group snapshots the current generation when its prefilter + /// starts, so a later publication still prunes. + #[tokio::test(flavor = "current_thread")] + async fn test_execute_prefilter_dynamic_filter_published_after_reader_construction() { + let producer_schema = dynamic_producer_schema(); + let handle = new_test_dynamic_filter(); + let (reader_builder, _) = prefilter_test_builder_with( + ObjectStore::new(Memory::default()).unwrap(), + 3, + 3, + PageIndexPolicy::Skip, + None, + Predicate::with_dyn_filters(vec![], vec![handle.clone()]), + CacheStrategy::Disabled, + ) + .await; + + // Nothing is published yet: the placeholder keeps every row. + let result = execute_test_prefilter(&reader_builder, 0).await; + assert_eq!(3, result.refined_selection.row_count()); + assert_eq!(0, result.filtered_rows); + + // The build side finishes and the producer publishes its membership. + publish_membership(&handle, &producer_schema, vec![1]); + let result = execute_test_prefilter(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![ + RowSelector::skip(1), + RowSelector::select(1), + RowSelector::skip(1), + ]), + result.refined_selection + ); + assert_eq!(2, result.filtered_rows); + + // The mask is never cached across generations: the next publication is evaluated + // instead of replaying the previous result. + publish_membership(&handle, &producer_schema, vec![2]); + let result = execute_test_prefilter(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![RowSelector::skip(2), RowSelector::select(1)]), + result.refined_selection + ); + assert_eq!(2, result.filtered_rows); + } + + /// A producer may republish while a row group's prefilter runs. Nothing proves the + /// new expression keeps the rows the stale snapshot kept, so the mask is discarded and + /// the incoming selection is kept: the upper filter stays authoritative. + #[tokio::test(flavor = "current_thread")] + async fn test_execute_prefilter_discards_dynamic_mask_after_republish() { + let producer_schema = dynamic_producer_schema(); + let handle = new_test_dynamic_filter(); + publish_membership(&handle, &producer_schema, vec![1]); + let (reader_builder, _) = prefilter_test_builder_with( + ObjectStore::new(Memory::default()).unwrap(), + 3, + 3, + PageIndexPolicy::Skip, + None, + Predicate::with_dyn_filters(vec![], vec![handle.clone()]), + CacheStrategy::Disabled, + ) + .await; + + // The row group's prefilter starts and snapshots the current generation. + let prefilter_builder = reader_builder.prefilter_builder.as_ref().unwrap(); + let mut prefilter_ctx = prefilter_builder.build(0); + // The producer publishes a new expression while the pass is in flight. + publish_membership(&handle, &producer_schema, vec![2]); + let incoming = RowSelection::from(vec![RowSelector::select(2), RowSelector::skip(1)]); + let result = execute_prefilter( + &mut prefilter_ctx, + &reader_builder, + &RowGroupBuildContext { + row_group_idx: 0, + row_selection: Some(incoming.clone()), + fetch_metrics: None, + }, + ) + .await + .unwrap(); + + assert_eq!(incoming, result.refined_selection); + assert_eq!(0, result.filtered_rows); + } + + /// With the prefilter-result cache enabled, a static predicate keeps its cached mask + /// while a producer republishes a dynamic filter. + /// + /// The static `ts` predicate is cacheable, so the row selection it produces is written + /// to the cache and reused. The dynamic membership is not: each pass evaluates the + /// generation it snapshotted, a newer publication replaces the previous membership + /// instead of replaying a cached mask, and disabling the filter leaves no stale mask + /// behind. Only the static mask is reused, and the rows the pass keeps stay the rows + /// the static predicate and the current membership keep. + #[tokio::test(flavor = "current_thread")] + async fn test_prefilter_result_cache_keeps_static_mask_but_not_dynamic_mask() { + let producer_schema = dynamic_producer_schema(); + let handle = new_test_dynamic_filter(); + let cache = CacheStrategy::EnableAll(Arc::new( + CacheManager::builder() + .prefilter_result_cache_size(1024) + .build(), + )); + // `ts >= 1` keeps the last two of the three rows, and the row group's time index + // statistics cannot prove it, so it stays a cacheable prefilter entry. The dynamic + // filter reads another column, so the cached static mask is the only way a later + // pass can skip reading `ts`. + let static_predicate = + col("ts").gt_eq(lit(ScalarValue::TimestampMillisecond(Some(1), None))); + let predicate = Predicate::with_dyn_filters(vec![static_predicate], vec![handle.clone()]); + let (reader_builder, _) = prefilter_test_builder_with( + ObjectStore::new(Memory::default()).unwrap(), + 3, + 1, + PageIndexPolicy::Skip, + None, + predicate.clone(), + cache, + ) + .await; + + // Pass 1: membership A keeps row 1. This pass writes the static mask to the cache. + publish_membership(&handle, &producer_schema, vec![1]); + let (result, metrics) = execute_test_prefilter_with_metrics(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![ + RowSelector::skip(1), + RowSelector::select(1), + RowSelector::skip(1), + ]), + result.refined_selection, + "the pass keeps the row membership A keeps" + ); + assert_eq!(2, result.filtered_rows); + assert!( + metrics.data.lock().unwrap().pages_to_fetch_store > 0, + "the first pass has to read the row group" + ); + + // Pass 2: publishing membership B keeps row 2. A pass that replayed a cached mask + // would still keep row 1. + publish_membership(&handle, &producer_schema, vec![2]); + let (result, _) = execute_test_prefilter_with_metrics(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![RowSelector::skip(2), RowSelector::select(1)]), + result.refined_selection, + "the new membership must apply instead of the cached static-only selection" + ); + assert_eq!(2, result.filtered_rows); + + // Pass 3: a wider membership keeps every row, so only the cached static mask + // prunes. A stale mask of A or B would still drop row 1 or row 2. + publish_membership(&handle, &producer_schema, vec![0, 1, 2]); + let (result, metrics) = execute_test_prefilter_with_metrics(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![RowSelector::skip(1), RowSelector::select(2)]), + result.refined_selection, + "no mask of a superseded membership may survive" + ); + assert_eq!(1, result.filtered_rows); + assert!( + metrics.data.lock().unwrap().pages_to_fetch_store > 0, + "the dynamic filter still reads its column" + ); + + // Pass 4: the producer republishes its `true` placeholder, which disables the + // dynamic filter for this row group. The pass then prunes nothing dynamically and + // reads no page at all: the static mask comes from the cache. + handle.update(physical_lit(true)).unwrap(); + let (result, metrics) = execute_test_prefilter_with_metrics(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![RowSelector::skip(1), RowSelector::select(2)]), + result.refined_selection, + "a disabled dynamic filter keeps the rows the static predicate keeps" + ); + assert_eq!(1, result.filtered_rows); + assert_eq!( + 0, + metrics.data.lock().unwrap().pages_to_fetch_store, + "the static mask must come from the cache" + ); + + // The rows those selections promise are the rows the projected reader returns. + let (batches, _) = + read_test_row_group(&reader_builder, 0, Some(result.refined_selection)).await; + assert_eq!(vec![1, 2], test_field_values(&batches)); + + // Control: the same predicate (the shared handle still holds the `true` placeholder) + // without a cache reads the row group and keeps the same rows, so the pass above + // was a cache hit rather than a skipped read. + let (control, _) = prefilter_test_builder_with( + ObjectStore::new(Memory::default()).unwrap(), + 3, + 1, + PageIndexPolicy::Skip, + None, + predicate, + CacheStrategy::Disabled, + ) + .await; + let (result, metrics) = execute_test_prefilter_with_metrics(&control, 0).await; + assert_eq!( + RowSelection::from(vec![RowSelector::skip(1), RowSelector::select(2)]), + result.refined_selection + ); + assert!( + metrics.data.lock().unwrap().pages_to_fetch_store > 0, + "the uncached control reads the row group" + ); + } + + /// The prefilter of the expression a real `HashJoin` publishes keeps exactly the rows + /// the live `batch_filter` above the same scan keeps. + /// + /// The published shape is what a partitioned join builds for a large build side: a + /// hash-table lookup over the probe key per build partition, routed by + /// `hash_repartition(key) % 2` inside a `CASE` — not a logical `IN` list. The + /// prefilter evaluates it on the reduced projection; the filter above the scan + /// evaluates the same expression on the projected read. Both must select the same + /// rows of the same row group. + #[tokio::test(flavor = "current_thread")] + async fn test_execute_prefilter_hash_lookup_matches_upper_filter() { + // The build side holds 1, 3 and 5; rows 0, 2 and 4 are not members. + let build_keys = [1_u64, 3, 5]; + let handle = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::new(Column::new("field_0", 3)) as Arc], + physical_lit(true), + )); + handle + .update(partitioned_hash_join_lookup( + &dynamic_producer_schema(), + &build_keys, + )) + .unwrap(); + let published = handle.current().unwrap(); + assert!( + published.is::(), + "a partitioned join routes its lookups: {published}" + ); + assert!( + published.to_string().contains("hash_lookup") + && published.to_string().contains("hash_repartition"), + "the published shape must be the hash lookup a producer builds: {published}" + ); + + let (reader_builder, _) = prefilter_test_builder_with( + ObjectStore::new(Memory::default()).unwrap(), + 6, + 1, + PageIndexPolicy::Skip, + None, + Predicate::with_dyn_filters(vec![], vec![handle.clone()]), + CacheStrategy::Disabled, + ) + .await; + + let result = execute_test_prefilter(&reader_builder, 0).await; + assert_eq!( + 3, result.filtered_rows, + "the lookup must prune the rows outside the build side" + ); + + let (all_batches, _) = read_test_row_group(&reader_builder, 0, None).await; + let (selected_batches, _) = + read_test_row_group(&reader_builder, 0, Some(result.refined_selection)).await; + + // The upper filter evaluates the same published expression, but on the wide + // projected batch it guards instead of the single prefilter column. + let schema = all_batches[0].schema(); + let upper_expr = reassign_expr_columns(Arc::clone(&published), schema.as_ref()).unwrap(); + let upper_batches = all_batches + .iter() + .map(|batch| batch_filter(batch, &upper_expr).unwrap()) + .collect::>(); + let upper_values = test_field_values(&upper_batches); + assert_eq!( + build_keys.to_vec(), + upper_values, + "the upper filter keeps exactly the build-side members" + ); + assert_eq!( + upper_values, + test_field_values(&selected_batches), + "the prefilter must keep exactly the rows the upper filter keeps" + ); + } + + /// A multi-page row group returns the same rows for the rows a dynamic prefilter + /// selects, and the projected reader fetches fewer bytes for them. + /// + /// The metrics below belong to the projected read only: `page_size_to_fetch_store` is + /// the number of bytes of the projected columns that read covers, and + /// `pages_to_fetch_store` the number of store ranges it requests. The saving is a + /// projection-byte saving, not a net scan saving: this row group's filter column was + /// already read by the prefilter pass, and store range count is not part of the claim — + /// a partial selection can split a column into more ranges than a full read. The row + /// counts asserted here are the rows the projected reader returns, not the rows of the + /// filter column the prefilter decoded. + /// + /// The row group is large enough that the pages a partial selection skips span more + /// than the store's fetch merge gap; below that the fetched byte ranges merge back + /// into the whole row group whatever the selection is. + #[tokio::test(flavor = "current_thread")] + async fn test_execute_prefilter_dynamic_selection_saves_projected_store_bytes() { + const ROWS: i64 = 100_000; + const PAGE_ROWS: usize = 20_000; + + let props = WriterProperties::builder() + .set_data_page_row_count_limit(PAGE_ROWS) + .set_write_batch_size(8192) + .build(); + let producer_schema = dynamic_producer_schema(); + let handle = new_test_dynamic_filter(); + // Only the first row of the row group survives the producer's membership. + publish_membership(&handle, &producer_schema, vec![0]); + let (reader_builder, _) = prefilter_test_builder_with( + ObjectStore::new(Memory::default()).unwrap(), + ROWS, + 1, + // Translating a row selection into byte ranges needs the offset index. + PageIndexPolicy::Optional, + Some(props), + Predicate::with_dyn_filters(vec![], vec![handle.clone()]), + CacheStrategy::Disabled, + ) + .await; + assert_eq!( + ROWS as usize, + reader_builder.parquet_meta.row_group(0).num_rows() as usize + ); + assert!( + reader_builder.parquet_meta.offset_index().is_some(), + "the row selection can only skip pages with the offset index" + ); + assert!( + reader_builder + .parquet_meta + .row_group(0) + .columns() + .iter() + .any(|column| column.num_values() as usize > PAGE_ROWS), + "the row group must span several pages" + ); + + let result = execute_test_prefilter(&reader_builder, 0).await; + assert_eq!( + RowSelection::from(vec![ + RowSelector::select(1), + RowSelector::skip(ROWS as usize - 1), + ]), + result.refined_selection + ); + + let (all_batches, all_metrics) = read_test_row_group(&reader_builder, 0, None).await; + let (selected_batches, selected_metrics) = + read_test_row_group(&reader_builder, 0, Some(result.refined_selection)).await; + + // The selected read returns exactly the rows of the full read that the producer + // keeps. + let all_values = test_field_values(&all_batches); + assert_eq!( + (0..ROWS).map(|value| value as u64).collect::>(), + all_values + ); + assert_eq!( + vec![0], + test_field_values(&selected_batches), + "the row selection must not change the rows the projected reader returns" + ); + + let all = all_metrics.data.lock().unwrap().clone(); + let selected = selected_metrics.data.lock().unwrap().clone(); + assert!( + all.page_size_to_fetch_store > 0, + "the control read must fetch projected bytes from the store" + ); + // Store range count (`pages_to_fetch_store`) is deliberately not asserted: a + // partial selection can request more ranges than a full read, because one range + // per column can become one range per page. + // + // Projection-byte savings: the projected read covers fewer bytes, because the + // selection drops the pages of the rows the producer rejects. + assert!( + selected.page_size_to_fetch_store < all.page_size_to_fetch_store, + "selected {} bytes vs all {} bytes", + selected.page_size_to_fetch_store, + all.page_size_to_fetch_store + ); + } + + /// Reads one row group of the projected columns through the projected reader and + /// returns its batches with the fetch metrics of exactly that read. + async fn read_test_row_group( + reader_builder: &RowGroupReaderBuilder, + row_group_idx: usize, + row_selection: Option, + ) -> (Vec, ParquetFetchMetrics) { + let fetch_metrics = ParquetFetchMetrics::default(); + let mut stream = reader_builder + .build_with_projection( + row_group_idx, + row_selection, + reader_builder.projection.mask.clone(), + Some(&fetch_metrics), + ) + .await + .unwrap(); + let mut batches = Vec::new(); + while let Some(batch) = stream.next().await { + batches.push(batch.unwrap()); + } + + (batches, fetch_metrics) + } + + fn test_field_values(batches: &[RecordBatch]) -> Vec { + batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect() + } + #[test] fn test_skip_prefilter_for_json_get() -> WhateverResult<()> { fn json_get_expr(base: Expr, path: &str) -> Expr { @@ -3623,6 +4321,306 @@ mod tests { assert!(PhysicalFilterContext::new_opt(&metadata, None, &read_format, &binary).is_none()); } + /// The producer's input schema, where `field_0` is its fourth column. + fn dynamic_producer_schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("tag_0", DataType::Utf8, true), + Field::new("tag_1", DataType::Utf8, true), + Field::new("ts", DataType::UInt64, true), + Field::new("field_0", DataType::UInt64, true), + ])) + } + + fn full_read_format(metadata: &RegionMetadataRef) -> FlatReadFormat { + FlatReadFormat::new( + metadata.clone(), + ReadColumns::new(metadata.column_metadatas.iter().map(|c| c.column_id)), + None, + "test", + true, + ) + .unwrap() + } + + /// Creates the dynamic filter a single-key `HashJoin` produces: the probe key is + /// declared as the filter's only child at the producer's column index, the + /// placeholder is `true`, and `published` is what the build side sends once known. + fn new_hash_join_dynamic_filter( + producer_schema: &SchemaRef, + published: &Expr, + ) -> Arc { + let key = Arc::new(Column::new("field_0", 3)) as Arc; + let handle = Arc::new(DynamicFilterPhysicalExpr::new( + vec![key], + physical_lit(true), + )); + let published = Predicate::to_physical_expr(published, producer_schema).unwrap(); + handle.update(published).unwrap(); + handle + } + + /// Returns a hash-table lookup over the probe key that contains `keys`, i.e. one + /// build partition of a `HashJoin` that picked the hash-map pushdown strategy. + fn hash_lookup_expr(keys: &[u64], seed: &SeededRandomState) -> Arc { + // The lookup compares hashes computed with the seed the producer built its map + // with, so the map is filled with the same `create_hashes` output. + let keys: ArrayRef = Arc::new(UInt64Array::from(keys.to_vec())); + let mut hashes = vec![0_u64; keys.len()]; + create_hashes([keys], seed.random_state(), &mut hashes).unwrap(); + let mut map = JoinHashMapU32::with_capacity(hashes.len().max(1)); + map.update_from_iter(Box::new(hashes.iter().enumerate()), 0); + + Arc::new(HashTableLookupExpr::new( + vec![Arc::new(Column::new("field_0", 3)) as Arc], + seed.clone(), + Arc::new(Map::HashMap(Box::new(map))), + "hash_lookup".to_string(), + None, + )) + } + + /// The dynamic filter a partitioned `HashJoin` publishes once every build partition + /// reported: one hash-table lookup per build partition, selected by + /// `hash_repartition(key) % partitions` in a `CASE`. + /// + /// `build_keys` is the whole build side. Each key is routed like the join routes a + /// probe row, so the branch that owns a key is the branch whose map contains it. + fn partitioned_hash_join_lookup( + producer_schema: &SchemaRef, + build_keys: &[u64], + ) -> Arc { + const PARTITIONS: usize = 2; + let seed = SeededRandomState::with_seed(0); + let probe_key = || Arc::new(Column::new("field_0", 3)) as Arc; + let route = Arc::new(BinaryExpr::new( + Arc::new(HashExpr::new( + vec![probe_key()], + REPARTITION_RANDOM_STATE, + "hash_repartition".to_string(), + )), + Operator::Modulo, + physical_lit(ScalarValue::UInt64(Some(PARTITIONS as u64))), + )) as Arc; + + // Route the build keys exactly like probe rows, with the producer schema so the + // routing expression reads `field_0` at the producer's column index. + let build_batch = RecordBatch::try_new( + Arc::clone(producer_schema), + vec![ + Arc::new(StringArray::from(vec![""; build_keys.len()])) as ArrayRef, + Arc::new(StringArray::from(vec![""; build_keys.len()])) as ArrayRef, + Arc::new(UInt64Array::from(vec![0_u64; build_keys.len()])) as ArrayRef, + Arc::new(UInt64Array::from(build_keys.to_vec())) as ArrayRef, + ], + ) + .unwrap(); + let routes = route + .evaluate(&build_batch) + .unwrap() + .into_array(build_keys.len()) + .unwrap(); + let routes = routes.as_any().downcast_ref::().unwrap(); + let mut partition_keys = vec![Vec::new(); PARTITIONS]; + for (key, partition) in build_keys.iter().zip(routes.values()) { + partition_keys[*partition as usize].push(*key); + } + + // The last partition is the `ELSE` branch, as the producer emits it. + let when_then_expr = partition_keys[..PARTITIONS - 1] + .iter() + .enumerate() + .map(|(partition, keys)| { + ( + physical_lit(ScalarValue::UInt64(Some(partition as u64))), + hash_lookup_expr(keys, &seed), + ) + }) + .collect::>(); + let else_expr = hash_lookup_expr(&partition_keys[PARTITIONS - 1], &seed); + + Arc::new(CaseExpr::try_new(Some(route), when_then_expr, Some(else_expr)).unwrap()) + } + + #[test] + fn test_dynamic_filter_context_rebinds_published_column_to_prefilter_schema() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = full_read_format(&metadata); + let handle = new_hash_join_dynamic_filter( + &dynamic_producer_schema(), + &col("field_0").in_list(vec![lit(1_u64), lit(2_u64)], false), + ); + + let ctx = + PhysicalFilterContext::new_dynamic(&metadata, None, &read_format, &handle).unwrap(); + assert!(ctx.is_dynamic()); + assert!( + !ctx.is_immutable(), + "a producer-owned mask must never enter the result cache" + ); + assert!(ctx.filter().is_none()); + assert_eq!("field_0", ctx.column_name()); + + let snapshot = ctx.capture_dynamic_snapshot().unwrap(); + assert!(ctx.dynamic_snapshot_is_current(&snapshot)); + // The producer published `field_0` as its fourth column; the prefilter batch + // carries that column alone, so the snapshot must use index 0. + let columns = collect_columns(&snapshot.expr); + assert_eq!(1, columns.len()); + assert_eq!(0, columns.iter().next().unwrap().index()); + + // `new_record_batch_with_custom_sequence` sets field_0 to 0, 1, 2, 3. + let batch = new_record_batch_with_custom_sequence(&["a", "x"], 0, 4, 1); + let mask = evaluate_physical_mask(&batch, &ctx, &snapshot.expr, "test").unwrap(); + assert_eq!( + vec![false, true, true, false], + mask.iter().collect::>() + ); + } + + /// The published expression is evaluated whole: a producer that routes NULL keys + /// itself keeps them, so the prefilter must not narrow the expression to its `IN` + /// leaf. + #[test] + fn test_dynamic_filter_prefilter_keeps_producer_null_semantics() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = full_read_format(&metadata); + let producer_schema = dynamic_producer_schema(); + let key = || col("field_0"); + // Duplicate members are the producer's list verbatim. + let members = || vec![lit(1_u64), lit(1_u64), lit(3_u64)]; + + // Both shapes a single-key `HashJoin` publishes for a nullable probe key: the + // membership filter ORed with the producer-owned null guard, and the same choice + // routed through `CASE`. + let guarded = key().is_null().or(key().in_list(members(), false)); + let case_routed = Expr::Case(datafusion_expr::expr::Case::new( + None, + vec![(Box::new(key().is_null()), Box::new(lit(true)))], + Some(Box::new(key().in_list(members(), false))), + )); + + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "field_0", + DataType::UInt64, + true, + )])), + vec![Arc::new(UInt64Array::from(vec![Some(1), None, Some(2), Some(3)])) as ArrayRef], + ) + .unwrap(); + + for published in [guarded, case_routed] { + let handle = new_hash_join_dynamic_filter(&producer_schema, &published); + let ctx = + PhysicalFilterContext::new_dynamic(&metadata, None, &read_format, &handle).unwrap(); + let snapshot = ctx.capture_dynamic_snapshot().unwrap(); + let mask = evaluate_physical_mask(&batch, &ctx, &snapshot.expr, "test").unwrap(); + // The NULL key and both members survive, only `2` is dropped. Evaluating the + // bare `IN` leaf would have dropped the NULL row. + assert_eq!( + vec![true, true, false, true], + mask.iter().collect::>(), + "{published}" + ); + } + } + + #[test] + fn test_dynamic_filter_context_keeps_all_without_a_single_filter_column() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = full_read_format(&metadata); + let producer_schema = dynamic_producer_schema(); + + // A two-key producer declares both keys, so the filter cannot be evaluated on a + // single-column prefilter batch. + let multi_key = Arc::new(DynamicFilterPhysicalExpr::new( + vec![ + Arc::new(Column::new("field_0", 3)) as Arc, + Arc::new(Column::new("ts", 2)) as Arc, + ], + physical_lit(true), + )); + assert!( + PhysicalFilterContext::new_dynamic(&metadata, None, &read_format, &multi_key).is_none() + ); + + let published = col("field_0").in_list(vec![lit(1_u64)], false); + let handle = new_hash_join_dynamic_filter(&producer_schema, &published); + + // The expected metadata points the filter column at an id the SST does not have. + let mut missing = (*metadata).clone(); + for column in missing.column_metadatas.iter_mut() { + if column.column_schema.name == "field_0" { + column.column_id = 20; + } + } + let missing = Arc::new(missing); + assert!( + PhysicalFilterContext::new_dynamic( + &metadata, + Some(missing.as_ref()), + &read_format, + &handle + ) + .is_none() + ); + + // A producer that publishes a wider expression after admission keeps every row: + // the snapshot is rejected instead of evaluated against the wrong columns. + let ctx = + PhysicalFilterContext::new_dynamic(&metadata, None, &read_format, &handle).unwrap(); + let widened = + Predicate::to_physical_expr(&published.and(col("ts").gt(lit(0_u64))), &producer_schema) + .unwrap(); + handle.update(widened).unwrap(); + assert!(ctx.capture_dynamic_snapshot().is_none()); + } + + #[test] + fn test_dynamic_filter_context_skips_renamed_and_type_changed_columns() { + let metadata: RegionMetadataRef = Arc::new(sst_region_metadata()); + let read_format = full_read_format(&metadata); + let producer_schema = dynamic_producer_schema(); + let handle = new_hash_join_dynamic_filter( + &producer_schema, + &col("field_0").in_list(vec![lit(1_u64)], false), + ); + + // The SST keeps the pre-rename name for the filter column's id. + let mut sst_metadata = (*metadata).clone(); + for column in sst_metadata.column_metadatas.iter_mut() { + if column.column_schema.name == "field_0" { + column.column_schema.name = "field_old".to_string(); + } + } + assert!( + PhysicalFilterContext::validated_column( + &Arc::new(sst_metadata), + Some(metadata.as_ref()), + "field_0" + ) + .is_none() + ); + + // An altered column type between the file and the expected metadata. + let mut type_changed = (*metadata).clone(); + for column in type_changed.column_metadatas.iter_mut() { + if column.column_schema.name == "field_0" { + column.column_schema.data_type = ConcreteDataType::int64_datatype(); + } + } + let type_changed = Arc::new(type_changed); + assert!( + PhysicalFilterContext::new_dynamic( + &metadata, + Some(type_changed.as_ref()), + &read_format, + &handle + ) + .is_none() + ); + } + fn write_test_parquet_with_pk_column(values: &[&[u8]]) -> bytes::Bytes { use datatypes::arrow::array::{ BinaryArray, TimestampMillisecondArray, UInt8Array, UInt64Array,