fix: predicate cache stats calculation

datafusion/core/tests/parquet/filter_pushdown.rs

@@ -644,6 +644,22 @@ async fn predicate_cache_pushdown_default() -> datafusion_common::Result<()> {
     .await
 }
 
+#[tokio::test]
+async fn predicate_cache_stats_issue_19561() -> datafusion_common::Result<()> {
+    let mut config = SessionConfig::new();
+    config.options_mut().execution.parquet.pushdown_filters = true;
+    // force to get multiple batches to trigger repeated metric compound bug
+    config.options_mut().execution.batch_size = 1;
+    let ctx = SessionContext::new_with_config(config);
+    // The cache is on by default, and used when filter pushdown is enabled
+    PredicateCacheTest {
+        expected_inner_records: 8,
+        expected_records: 4,
+    }
+    .run(&ctx)
+    .await
+}
+
 #[tokio::test]
 async fn predicate_cache_pushdown_default_selections_only()
 -> datafusion_common::Result<()> {

datafusion/datasource-parquet/src/metrics.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use datafusion_physical_plan::metrics::{
-    Count, ExecutionPlanMetricsSet, MetricBuilder, MetricType, PruningMetrics,
+    Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, MetricType, PruningMetrics,
     RatioMergeStrategy, RatioMetrics, Time,
 };
 
@@ -77,11 +77,16 @@ pub struct ParquetFileMetrics {
     /// Parquet.
     ///
     /// This is the expensive path (IO + Decompression + Decoding).
-    pub predicate_cache_inner_records: Count,
+    ///
+    /// We use a Gauge here as arrow-rs reports absolute numbers rather
+    /// than incremental readings, we want a `set` operation here rather
+    /// than `add`. Earlier it was `Count`, which led to this issue:
+    /// github.com/apache/datafusion/issues/19334
+    pub predicate_cache_inner_records: Gauge,
     /// Predicate Cache: number of records read from the cache. This is the
     /// number of rows that were stored in the cache after evaluating predicates
     /// reused for the output.
-    pub predicate_cache_records: Count,
+    pub predicate_cache_records: Gauge,
 }
 
 impl ParquetFileMetrics {
@@ -162,11 +167,11 @@ impl ParquetFileMetrics {
 
         let predicate_cache_inner_records = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
-            .counter("predicate_cache_inner_records", partition);
+            .gauge("predicate_cache_inner_records", partition);
 
         let predicate_cache_records = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
-            .counter("predicate_cache_records", partition);
+            .gauge("predicate_cache_records", partition);
 
         Self {
             files_ranges_pruned_statistics,

datafusion/datasource-parquet/src/opener.rs

@@ -47,7 +47,7 @@ use datafusion_physical_expr_common::physical_expr::{
     PhysicalExpr, is_dynamic_physical_expr,
 };
 use datafusion_physical_plan::metrics::{
-    Count, ExecutionPlanMetricsSet, MetricBuilder, PruningMetrics,
+    Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, PruningMetrics,
 };
 use datafusion_pruning::{FilePruner, PruningPredicate, build_pruning_predicate};
 
@@ -674,15 +674,15 @@ impl FileOpener for ParquetOpener {
 /// arrow-rs parquet reader) to the parquet file metrics for DataFusion
 fn copy_arrow_reader_metrics(
     arrow_reader_metrics: &ArrowReaderMetrics,
-    predicate_cache_inner_records: &Count,
-    predicate_cache_records: &Count,
+    predicate_cache_inner_records: &Gauge,
+    predicate_cache_records: &Gauge,
 ) {
     if let Some(v) = arrow_reader_metrics.records_read_from_inner() {
-        predicate_cache_inner_records.add(v);
+        predicate_cache_inner_records.set(v);
     }
 
     if let Some(v) = arrow_reader_metrics.records_read_from_cache() {
-        predicate_cache_records.add(v);
+        predicate_cache_records.set(v);
     }
 }