From be4ab7770d2068560dc215eaacfe3bf2e4d016c0 Mon Sep 17 00:00:00 2001 From: parmesant Date: Thu, 22 Oct 2026 12:42:06 +0530 Subject: [PATCH 1/2] fix: manifest pruning (#1828) `extract_primary_filter()` incorrectly prunes manifests based on time filters even if they are not on `P_TIMESTAMP` (cherry picked from commit af11d36eeb98c12e383a4ee148d0b1ab899813c3) --- src/query/stream_schema_provider.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/query/stream_schema_provider.rs b/src/query/stream_schema_provider.rs index 7dce6f732..abd0cac89 100644 --- a/src/query/stream_schema_provider.rs +++ b/src/query/stream_schema_provider.rs @@ -959,6 +959,12 @@ impl PartialTimeFilter { let Expr::BinaryExpr(binexpr) = expr else { return None; }; + + let is_p_timestamp = expr + .column_refs() + .iter() + .map(|c| &c.name) + .contains(&DEFAULT_TIMESTAMP_KEY.to_string()); let (op, time) = extract_timestamp_bound(binexpr, time_partition)?; let value = match op { Operator::Gt => PartialTimeFilter::Low(Bound::Excluded(time)), @@ -969,7 +975,7 @@ impl PartialTimeFilter { Operator::IsNotDistinctFrom => PartialTimeFilter::Eq(time), _ => return None, }; - Some(value) + if is_p_timestamp { Some(value) } else { None } } pub fn binary_expr(&self, left: Expr) -> Expr { From 1aa3c1cbe70af5e6aaa8006bb57009cb77ee17ff Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Fri, 23 Oct 2026 19:23:15 -0500 Subject: [PATCH 2/2] fix(query): prune manifests by the stream's time partition column Follow-up to the upstream port of #1828. Upstream only accepts bounds on p_timestamp, which is wrong for streams with a custom time_partition: - Their manifests, staging prefixes and listing prefixes are bucketed by the time_partition column (catalog::calculate_time_bound, Event::process), so bounds on it are the only ones that can prune. Upstream dropped these, so the time range the query handler injects on the partition column no longer pruned anything. - p_timestamp there is ingestion time, unrelated to the event-time buckets, so pruning by it can drop back-filled rows. Accept a bound only when its left side is the column manifests are keyed by: time_partition if set, otherwise p_timestamp. Co-Authored-By: Claude Opus 5.5 --- src/query/stream_schema_provider.rs | 152 ++++++++++++++++++++++++++-- 1 file changed, 143 insertions(+), 9 deletions(-) diff --git a/src/query/stream_schema_provider.rs b/src/query/stream_schema_provider.rs index abd0cac89..55d04f1c2 100644 --- a/src/query/stream_schema_provider.rs +++ b/src/query/stream_schema_provider.rs @@ -960,11 +960,18 @@ impl PartialTimeFilter { return None; }; - let is_p_timestamp = expr - .column_refs() - .iter() - .map(|c| &c.name) - .contains(&DEFAULT_TIMESTAMP_KEY.to_string()); + // Manifests (and staging/listing prefixes) are bucketed by the stream's + // time partition column, or `p_timestamp` when none is configured. A + // bound on any other column, including `p_timestamp` on a stream with a + // custom time partition, says nothing about which manifests hold + // matching rows and must not be used to prune them. + let partition_column = time_partition.as_deref().unwrap_or(DEFAULT_TIMESTAMP_KEY); + let Expr::Column(column) = binexpr.left.as_ref() else { + return None; + }; + if column.name != partition_column { + return None; + } let (op, time) = extract_timestamp_bound(binexpr, time_partition)?; let value = match op { Operator::Gt => PartialTimeFilter::Low(Bound::Excluded(time)), @@ -975,7 +982,7 @@ impl PartialTimeFilter { Operator::IsNotDistinctFrom => PartialTimeFilter::Eq(time), _ => return None, }; - if is_p_timestamp { Some(value) } else { None } + Some(value) } pub fn binary_expr(&self, left: Expr) -> Expr { @@ -1325,16 +1332,19 @@ mod tests { }; use crate::catalog::{ + Snapshot as _, column::{Column, Int64Type, TypedStatistics}, manifest::{File, SortOrder}, - snapshot::ManifestItem, + snapshot::{ManifestItem, Snapshot}, }; + use crate::event::DEFAULT_TIMESTAMP_KEY; use super::{ HotTierStreamKey, PartialTimeFilter, StandardTableProvider, balanced_file_groups, build_parquet_scan_components, build_parquet_scan_components_with_full_filters, - exact_source_filters, extract_timestamp_bound, guarded_hot_tier_table_source, - hot_tier_stream_key, is_overlapping_query, standard_table_provider, + exact_source_filters, extract_primary_filter, extract_timestamp_bound, + guarded_hot_tier_table_source, hot_tier_stream_key, is_overlapping_query, + standard_table_provider, }; #[test] @@ -1933,6 +1943,130 @@ mod tests { assert_eq!(result, expected); } + + fn ts_millis(time: DateTime) -> Expr { + Expr::Literal( + ScalarValue::TimestampMillisecond(Some(time.timestamp_millis()), None), + None, + ) + } + + fn column(name: &str) -> Expr { + Expr::Column(name.into()) + } + + fn pruned_manifest_paths(filters: &[Expr], time_partition: &Option) -> Vec { + let time_filters = extract_primary_filter(filters, time_partition); + manifest_snapshot() + .manifests(&time_filters) + .into_iter() + .map(|item| item.manifest_path) + .collect() + } + + fn manifest_snapshot() -> Snapshot { + Snapshot { + manifest_list: manifest_items(), + ..Snapshot::default() + } + } + + #[test] + fn non_timestamp_column_filter_does_not_prune_manifests() { + // `created_at` is unrelated to how manifests are bucketed: rows with a + // late `created_at` can live in any day's manifest. + let filters = [ + column("created_at").gt_eq(ts_millis(datetime_min(2023, 12, 17))), + column("created_at").lt(ts_millis(datetime_max(2023, 12, 17))), + ]; + + assert!(extract_primary_filter(&filters, &None).is_empty()); + assert_eq!(pruned_manifest_paths(&filters, &None), ["1", "2", "3"]); + } + + #[test] + fn non_timestamp_string_literal_filter_does_not_prune_manifests() { + let filters = [column("created_at").gt(Expr::Literal( + ScalarValue::Utf8(Some("2023-12-17T00:00:00".to_owned())), + None, + ))]; + + assert!(extract_primary_filter(&filters, &None).is_empty()); + assert_eq!(pruned_manifest_paths(&filters, &None), ["1", "2", "3"]); + } + + #[test] + fn p_timestamp_filter_prunes_manifests() { + let filters = [ + column(DEFAULT_TIMESTAMP_KEY).gt_eq(ts_millis(datetime_min(2023, 12, 16))), + column(DEFAULT_TIMESTAMP_KEY).lt(ts_millis(datetime_max(2023, 12, 16))), + ]; + + assert_eq!(extract_primary_filter(&filters, &None).len(), 2); + assert_eq!(pruned_manifest_paths(&filters, &None), ["2"]); + } + + #[test] + fn qualified_p_timestamp_filter_prunes_manifests() { + // The query handler injects table-qualified time filters. + let filters = [Expr::Column(datafusion::common::Column::new( + Some("app_logs"), + DEFAULT_TIMESTAMP_KEY, + )) + .gt_eq(ts_millis(datetime_min(2023, 12, 17)))]; + + assert_eq!(pruned_manifest_paths(&filters, &None), ["3"]); + } + + #[test] + fn literal_on_left_or_non_comparison_does_not_prune_manifests() { + let filters = [ + ts_millis(datetime_min(2023, 12, 17)).lt(column(DEFAULT_TIMESTAMP_KEY)), + column(DEFAULT_TIMESTAMP_KEY).not_eq(ts_millis(datetime_min(2023, 12, 17))), + ]; + + assert!(extract_primary_filter(&filters, &None).is_empty()); + } + + #[test] + fn custom_time_partition_filter_prunes_manifests() { + // Manifests of time-partitioned streams are bucketed by the partition + // column, so bounds on it (timestamp or parseable string) prune. + let time_partition = Some("event_time".to_owned()); + let filters = [ + column("event_time").gt_eq(ts_millis(datetime_min(2023, 12, 16))), + column("event_time").lt(Expr::Literal( + ScalarValue::Utf8(Some("2023-12-16T12:00:00".to_owned())), + None, + )), + ]; + + assert_eq!(extract_primary_filter(&filters, &time_partition).len(), 2); + assert_eq!(pruned_manifest_paths(&filters, &time_partition), ["2"]); + } + + #[test] + fn p_timestamp_filter_does_not_prune_custom_time_partition_manifests() { + // On a time-partitioned stream `p_timestamp` is ingestion time, while + // manifests are bucketed by event time: back-filled events ingested on + // the 17th sit in the 15th's manifest and must not be pruned away. + let time_partition = Some("event_time".to_owned()); + let filters = [column(DEFAULT_TIMESTAMP_KEY).gt_eq(ts_millis(datetime_min(2023, 12, 17)))]; + + assert!(extract_primary_filter(&filters, &time_partition).is_empty()); + assert_eq!( + pruned_manifest_paths(&filters, &time_partition), + ["1", "2", "3"] + ); + } + + #[test] + fn unrelated_column_does_not_prune_custom_time_partition_manifests() { + let time_partition = Some("event_time".to_owned()); + let filters = [column("created_at").gt_eq(ts_millis(datetime_min(2023, 12, 17)))]; + + assert!(extract_primary_filter(&filters, &time_partition).is_empty()); + } } #[cfg(test)]