diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index fcbe077cc6f..0ca54391702 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -28,11 +28,14 @@ use common_telemetry::tracing::Instrument; use common_telemetry::{debug, error, tracing, warn}; use common_time::range::TimestampRange; use datafusion::physical_plan::expressions::DynamicFilterPhysicalExpr; -use datafusion_common::Column; +use datafusion_common::pruning::PruningStatistics; +use datafusion_common::{Column, ScalarValue}; use datafusion_expr::Expr; use datafusion_expr::utils::expr_to_columns; +use datatypes::arrow::array::{ArrayRef, BooleanArray, UInt64Array}; use datatypes::extension::json::is_structured_json_field; use datatypes::types::json_type::JsonNativeType; +use datatypes::value::timestamp_to_scalar_value; use futures::StreamExt; use itertools::Itertools; use partition::expr::PartitionExpr; @@ -1071,6 +1074,27 @@ impl ScanInput { } } + /// Tries to build file-level pruning statistics using only the [FileHandle]'s manifest-level + /// time range, without reading any parquet metadata. + /// + /// Returns `None` if timestamp unit conversion overflows (conservative: keep the file). + fn try_file_level_pruning_stats(&self, file: &FileHandle) -> Option { + let (ts_min, ts_max) = file.time_range(); + let time_index = self.mapper.metadata().time_index_column(); + let time_index_unit = time_index.column_schema.data_type.as_timestamp()?.unit(); + + // Convert file timestamps to the time index column's unit. Use `convert_to_ceil` for + // the upper bound to avoid accidentally shrinking the manifest range. + let min_ts = ts_min.convert_to(time_index_unit)?; + let max_ts = ts_max.convert_to_ceil(time_index_unit)?; + + Some(FileLevelPruningStats { + min_scalar: timestamp_to_scalar_value(time_index_unit, Some(min_ts.value())), + max_scalar: timestamp_to_scalar_value(time_index_unit, Some(max_ts.value())), + time_index_col_name: time_index.column_schema.name.clone(), + }) + } + /// Prunes a file to scan and returns the builder to build readers. #[tracing::instrument( skip_all, @@ -1086,6 +1110,24 @@ impl ScanInput { reader_metrics: &mut ReaderMetrics, ) -> Result { let predicate = self.predicate_for_file(file); + + // Early file-level pruning using manifest time range before any parquet metadata access. + // This avoids I/O for files that definitely can't match the current predicate snapshot, + // especially after TopK dynamic filters have established a timestamp threshold. + if let Some(ref pred) = predicate + && !pred.is_empty() + && let Some(file_level_stats) = self.try_file_level_pruning_stats(file) + { + let pruning_results = pred.prune_with_stats( + &file_level_stats, + self.mapper.metadata().schema.arrow_schema(), + ); + if pruning_results.first() == Some(&false) { + reader_metrics.filter_metrics.files_time_range_pruned += 1; + return Ok(FileRangeBuilder::default()); + } + } + let may_build_selective_row_selection = predicate.is_some(); let decode_pk_values = !self.compaction && self @@ -1305,6 +1347,57 @@ impl ScanInput { } } +/// Lightweight [PruningStatistics] that only uses the file-level time range from manifest +/// metadata, avoiding any parquet metadata reads. Used for early file-level pruning before +/// accessing row-group-level statistics. +pub(crate) struct FileLevelPruningStats { + /// Scalar value for the file's minimum timestamp in the time index column's unit. + pub(crate) min_scalar: ScalarValue, + /// Scalar value for the file's maximum timestamp in the time index column's unit. + pub(crate) max_scalar: ScalarValue, + /// Name of the time index column. + pub(crate) time_index_col_name: String, +} + +impl PruningStatistics for FileLevelPruningStats { + fn min_values(&self, column: &Column) -> Option { + if column.name == self.time_index_col_name { + ScalarValue::iter_to_array(std::iter::once(self.min_scalar.clone())).ok() + } else { + None + } + } + + fn max_values(&self, column: &Column) -> Option { + if column.name == self.time_index_col_name { + ScalarValue::iter_to_array(std::iter::once(self.max_scalar.clone())).ok() + } else { + None + } + } + + fn num_containers(&self) -> usize { + 1 + } + + fn null_counts(&self, column: &Column) -> Option { + if column.name == self.time_index_col_name { + // The time index column is NOT NULL. + Some(Arc::new(UInt64Array::from(vec![0u64]))) + } else { + None + } + } + + fn row_counts(&self, _column: &Column) -> Option { + None + } + + fn contained(&self, _column: &Column, _values: &HashSet) -> Option { + None + } +} + #[cfg(test)] impl ScanInput { /// Returns SST file ids to scan. @@ -1861,9 +1954,13 @@ impl PredicateGroup { mod tests { use std::sync::Arc; + use common_time::timestamp::{TimeUnit, Timestamp}; use datafusion::physical_plan::expressions::lit as physical_lit; use datafusion_common::ScalarValue; use datafusion_expr::{col, lit}; + use datatypes::arrow::datatypes::{ + DataType as ArrowDataType, Field, Schema as ArrowSchema, TimeUnit as ArrowTimeUnit, + }; use datatypes::prelude::ConcreteDataType; use datatypes::schema::ColumnSchema; use datatypes::types::json_type::JsonObjectType; @@ -1877,6 +1974,7 @@ mod tests { use crate::error::InvalidMetadataSnafu; use crate::read::range_cache::ScanRequestFingerprintBuilder; use crate::read::read_columns::ReadColumn; + use crate::sst::file::FileMeta; use crate::test_util::memtable_util::metadata_with_primary_key; use crate::test_util::scheduler_util::SchedulerEnv; @@ -1904,6 +2002,60 @@ mod tests { lit(ScalarValue::TimestampMillisecond(Some(val), None)) } + fn metadata_with_time_index_unit(unit: TimeUnit) -> RegionMetadataRef { + let mut builder = RegionMetadataBuilder::new(RegionId::new(123, 456)); + builder + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "k0".to_string(), + ConcreteDataType::string_datatype(), + false, + ), + semantic_type: SemanticType::Tag, + column_id: 0, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "k1".to_string(), + ConcreteDataType::uint32_datatype(), + false, + ), + semantic_type: SemanticType::Tag, + column_id: 1, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts".to_string(), + ConcreteDataType::timestamp_datatype(unit), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 2, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "v0".to_string(), + ConcreteDataType::int64_datatype(), + true, + ), + semantic_type: SemanticType::Field, + column_id: 3, + }) + .primary_key(vec![0, 1]); + + Arc::new(builder.build().unwrap()) + } + + fn file_handle_with_time_range(start: Timestamp, end: Timestamp) -> FileHandle { + FileHandle::new( + FileMeta { + time_range: (start, end), + ..Default::default() + }, + Arc::new(crate::sst::file_purger::NoopFilePurger), + ) + } + #[test] fn test_fill_json_nested_paths_from_hint() -> Result<()> { fn json_projection_test_metadata() -> Result { @@ -2129,6 +2281,144 @@ mod tests { assert_eq!(1, predicate_without_region.dyn_filters().len()); } + #[test] + fn test_file_level_pruning_stats_prunes_old_file() { + let ts_col_name = "ts"; + let predicate = Predicate::new(vec![col(ts_col_name).gt(ts_lit(1000))]); + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + ts_col_name, + ArrowDataType::Timestamp(ArrowTimeUnit::Millisecond, None), + false, + )])); + + // File with time range [0ms, 500ms] is completely before `ts > 1000ms`. + let stats = FileLevelPruningStats { + min_scalar: ScalarValue::TimestampMillisecond(Some(0), None), + max_scalar: ScalarValue::TimestampMillisecond(Some(500), None), + time_index_col_name: ts_col_name.to_string(), + }; + assert_eq!( + vec![false], + predicate.prune_with_stats(&stats, &arrow_schema) + ); + + // File with time range [0ms, 2000ms] overlaps `ts > 1000ms`, so keep it. + let stats = FileLevelPruningStats { + min_scalar: ScalarValue::TimestampMillisecond(Some(0), None), + max_scalar: ScalarValue::TimestampMillisecond(Some(2000), None), + time_index_col_name: ts_col_name.to_string(), + }; + assert_eq!( + vec![true], + predicate.prune_with_stats(&stats, &arrow_schema) + ); + } + + #[test] + fn test_file_level_pruning_stats_no_predicate_keeps_all() { + let predicate = Predicate::new(vec![]); + assert!(predicate.is_empty()); + + let stats = FileLevelPruningStats { + min_scalar: ScalarValue::TimestampMillisecond(Some(0), None), + max_scalar: ScalarValue::TimestampMillisecond(Some(500), None), + time_index_col_name: "ts".to_string(), + }; + let arrow_schema = Arc::new(ArrowSchema::new(Vec::::new())); + assert_eq!( + vec![true], + predicate.prune_with_stats(&stats, &arrow_schema) + ); + } + + #[tokio::test] + async fn test_file_level_pruning_stats_ceil_max_unit_conversion() { + let metadata = metadata_with_time_index_unit(TimeUnit::Millisecond); + let input = new_scan_input(metadata, vec![]).await; + let file = file_handle_with_time_range( + Timestamp::new(1_000_001, TimeUnit::Nanosecond), + Timestamp::new(1_000_001, TimeUnit::Nanosecond), + ); + + let stats = input.try_file_level_pruning_stats(&file).unwrap(); + assert_eq!( + ScalarValue::TimestampMillisecond(Some(1), None), + stats.min_scalar + ); + assert_eq!( + ScalarValue::TimestampMillisecond(Some(2), None), + stats.max_scalar + ); + + // The actual max timestamp is slightly greater than 1ms. It must be kept for `ts > 1ms`. + let predicate = Predicate::new(vec![col("ts").gt(ts_lit(1))]); + assert_eq!( + vec![true], + predicate.prune_with_stats(&stats, input.mapper.metadata().schema.arrow_schema()) + ); + } + + #[tokio::test] + async fn test_file_level_pruning_stats_overflow_keeps_file() { + let metadata = metadata_with_time_index_unit(TimeUnit::Nanosecond); + let input = new_scan_input(metadata, vec![]).await; + let file = file_handle_with_time_range( + Timestamp::new(0, TimeUnit::Second), + Timestamp::new(i64::MAX, TimeUnit::Second), + ); + + assert!(input.try_file_level_pruning_stats(&file).is_none()); + } + + #[test] + fn test_file_level_pruning_stats_keeps_inclusive_boundary() { + let ts_col_name = "ts"; + let predicate = Predicate::new(vec![col(ts_col_name).gt_eq(ts_lit(1000))]); + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + ts_col_name, + ArrowDataType::Timestamp(ArrowTimeUnit::Millisecond, None), + false, + )])); + let stats = FileLevelPruningStats { + min_scalar: ScalarValue::TimestampMillisecond(Some(0), None), + max_scalar: ScalarValue::TimestampMillisecond(Some(1000), None), + time_index_col_name: ts_col_name.to_string(), + }; + + assert_eq!( + vec![true], + predicate.prune_with_stats(&stats, &arrow_schema) + ); + } + + #[tokio::test] + async fn test_file_level_pruning_with_dyn_filter_only_predicate() { + let metadata = Arc::new(metadata_with_primary_key(vec![0, 1], false)); + let mapper = FlatProjectionMapper::new(&metadata, [0, 2, 3]).unwrap(); + let predicate_group = PredicateGroup::new(metadata.as_ref(), &[]).unwrap(); + predicate_group.add_dyn_filters(vec![Arc::new(DynamicFilterPhysicalExpr::new( + vec![], + physical_lit(false), + ))]); + let input = ScanInput::new(SchedulerEnv::new().await.access_layer.clone(), mapper) + .with_predicate(predicate_group); + let file = file_handle_with_time_range( + Timestamp::new_millisecond(0), + Timestamp::new_millisecond(1000), + ); + let mut reader_metrics = ReaderMetrics::default(); + + let builder = input + .prune_file(&file, PreFilterMode::SkipFields, &mut reader_metrics) + .await + .unwrap(); + + assert_eq!(1, reader_metrics.filter_metrics.files_time_range_pruned); + let mut ranges = SmallVec::new(); + builder.build_ranges(-1, &mut ranges); + assert!(ranges.is_empty()); + } + #[tokio::test] async fn test_range_pre_filter_mode() { let metadata = Arc::new(metadata_with_primary_key(vec![0, 1], false)); diff --git a/src/mito2/src/read/scan_util.rs b/src/mito2/src/read/scan_util.rs index 3e1ab18d382..428cf22740d 100644 --- a/src/mito2/src/read/scan_util.rs +++ b/src/mito2/src/read/scan_util.rs @@ -195,6 +195,8 @@ pub(crate) struct ScanMetricsSet { pruner_cache_miss: usize, /// Duration spent waiting for pruner to build file ranges. pruner_prune_cost: Duration, + /// Number of files filtered by manifest time-range pruning. + files_time_range_pruned: usize, /// Number of record batches read from SST. num_sst_record_batches: usize, /// Number of batches decoded from SST. @@ -326,6 +328,7 @@ impl fmt::Debug for ScanMetricsSet { pruner_cache_hit, pruner_cache_miss, pruner_prune_cost, + files_time_range_pruned, num_sst_record_batches, num_sst_batches, num_sst_rows, @@ -390,6 +393,9 @@ impl fmt::Debug for ScanMetricsSet { } // Write non-zero filter counters + if *files_time_range_pruned > 0 { + write!(f, ", \"files_time_range_pruned\":{files_time_range_pruned}")?; + } if *rg_fulltext_filtered > 0 { write!(f, ", \"rg_fulltext_filtered\":{rg_fulltext_filtered}")?; } @@ -689,6 +695,7 @@ impl ScanMetricsSet { pruner_cache_hit, pruner_cache_miss, pruner_prune_cost, + files_time_range_pruned, inverted_index_apply_metrics, bloom_filter_apply_metrics, fulltext_index_apply_metrics, @@ -706,6 +713,8 @@ impl ScanMetricsSet { self.build_parts_cost += *build_cost; self.sst_scan_cost += *scan_cost; + self.files_time_range_pruned += *files_time_range_pruned; + self.rg_total += *rg_total; self.rg_fulltext_filtered += *rg_fulltext_filtered; self.rg_inverted_filtered += *rg_inverted_filtered; diff --git a/src/mito2/src/sst/parquet/reader.rs b/src/mito2/src/sst/parquet/reader.rs index e936bf8d8b6..6459a9877e8 100644 --- a/src/mito2/src/sst/parquet/reader.rs +++ b/src/mito2/src/sst/parquet/reader.rs @@ -1428,6 +1428,8 @@ pub(crate) struct ReaderFilterMetrics { pub(crate) pruner_cache_miss: usize, /// Duration spent waiting for pruner to build file ranges. pub(crate) pruner_prune_cost: Duration, + /// Number of files filtered by manifest time-range pruning. + pub(crate) files_time_range_pruned: usize, } impl ReaderFilterMetrics { @@ -1460,6 +1462,7 @@ impl ReaderFilterMetrics { self.pruner_cache_hit += other.pruner_cache_hit; self.pruner_cache_miss += other.pruner_cache_miss; self.pruner_prune_cost += other.pruner_prune_cost; + self.files_time_range_pruned += other.files_time_range_pruned; // Merge optional applier metrics if let Some(other_metrics) = &other.inverted_index_apply_metrics {