wip

Author: adriangb

datafusion/common/src/file_options/parquet_writer.rs

@@ -210,7 +210,7 @@ impl ParquetOptions {
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
             filter_pushdown_min_bytes_per_sec: _, // not used for writer props
-            filter_correlation_threshold: _,   // not used for writer props
+            filter_correlation_threshold: _,      // not used for writer props
             filter_statistics_collection_min_rows: _, // not used for writer props
             filter_statistics_collection_fraction: _, // not used for writer props
         } = self;
@@ -465,6 +465,11 @@ mod tests {
             coerce_int96: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
             filter_pushdown_min_bytes_per_sec: defaults.filter_pushdown_min_bytes_per_sec,
+            filter_correlation_threshold: defaults.filter_correlation_threshold,
+            filter_statistics_collection_fraction: defaults
+                .filter_statistics_collection_fraction,
+            filter_statistics_collection_min_rows: defaults
+                .filter_statistics_collection_min_rows,
         }
     }
 
@@ -581,6 +586,12 @@ mod tests {
                 coerce_int96: None,
                 filter_pushdown_min_bytes_per_sec: global_options_defaults
                     .filter_pushdown_min_bytes_per_sec,
+                filter_correlation_threshold: global_options_defaults
+                    .filter_correlation_threshold,
+                filter_statistics_collection_fraction: global_options_defaults
+                    .filter_statistics_collection_fraction,
+                filter_statistics_collection_min_rows: global_options_defaults
+                    .filter_statistics_collection_min_rows,
             },
             column_specific_options,
             key_value_metadata,

datafusion/datasource-parquet/src/file_format.rs

@@ -524,7 +524,8 @@ impl FileFormat for ParquetFormat {
             .cloned()
             .ok_or_else(|| internal_datafusion_err!("Expected ParquetSource"))?;
         source = source.with_table_parquet_options(self.options.clone());
-        source = source.with_filter_pushdown_min_bytes_per_sec(filter_pushdown_min_bytes_per_sec);
+        source = source
+            .with_filter_pushdown_min_bytes_per_sec(filter_pushdown_min_bytes_per_sec);
 
         // Use the CachedParquetFileReaderFactory
         let metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache();

datafusion/datasource-parquet/src/opener.rs

@@ -588,11 +588,24 @@ impl FileOpener for ParquetOpener {
             // metrics from the arrow reader itself
             let arrow_reader_metrics = ArrowReaderMetrics::enabled();
 
-            // Compute bytes_per_row from file metadata for the bytes/sec metric.
+            // Compute bytes_per_row from file metadata and feed it into the
+            // selectivity tracker's running average (used by partition_filters_grouped).
             let bytes_per_row = compute_bytes_per_row(
                 file_metadata.as_ref(),
                 &projection.column_indices(),
             );
+            if let Some(bpr) = bytes_per_row {
+                let file_rows: u64 = file_metadata
+                    .row_groups()
+                    .iter()
+                    .map(|rg| rg.num_rows() as u64)
+                    .sum();
+                if file_rows > 0 {
+                    selectivity_tracker
+                        .write()
+                        .update_bytes_per_row(bpr, file_rows);
+                }
+            }
 
             // Acquire tracker lock once for both partitioning and row filter building.
             let (post_scan_filters, projection, mask) = {
@@ -611,7 +624,7 @@ impl FileOpener for ParquetOpener {
                             .into_iter()
                             .map(Arc::clone)
                             .collect();
-                    tracker.partition_filters_grouped(conjuncts, bytes_per_row)
+                    tracker.partition_filters_grouped(conjuncts)
                 } else {
                     PartitionedFiltersGrouped {
                         row_filter_groups: vec![],
@@ -623,18 +636,14 @@ impl FileOpener for ParquetOpener {
                 // but don't add filter expressions as projection columns.
                 let mut all_indices: Vec<usize> = projection.column_indices();
                 for filter in &post_scan {
-                    for col in datafusion_physical_expr::utils::collect_columns(filter)
-                    {
+                    for col in datafusion_physical_expr::utils::collect_columns(filter) {
                         let idx = col.index();
                         if !all_indices.contains(&idx) {
                             all_indices.push(idx);
                         }
                     }
                 }
-                let mask = ProjectionMask::roots(
-                    builder.parquet_schema(),
-                    all_indices,
-                );
+                let mask = ProjectionMask::roots(builder.parquet_schema(), all_indices);
 
                 // Build row filter from groups of correlated filters
                 if !row_filter_groups.is_empty() {

datafusion/datasource-parquet/src/row_filter.rs

@@ -895,9 +895,11 @@ pub fn build_row_filter_from_groups(
                 // Compound can't push down -- these filters will need to be
                 // handled as post-scan. For now, try each individually.
                 for expr in group {
-                    if let Ok(Some(candidate)) =
-                        FilterCandidateBuilder::new(Arc::clone(&expr), Arc::clone(file_schema))
-                            .build(metadata)
+                    if let Ok(Some(candidate)) = FilterCandidateBuilder::new(
+                        Arc::clone(&expr),
+                        Arc::clone(file_schema),
+                    )
+                    .build(metadata)
                     {
                         candidates_with_exprs.push((vec![expr], candidate));
                     }