fix: predicate cache stats calculation (#19561)

## Which issue does this PR close?

- Closes #19334

## Rationale for this change

Explained in issue

## What changes are included in this PR?

Changed counter to gauge to set value provided by arrow-rs correctly

## Are these changes tested?

Yes, have added a new test which fails with old code as it reports
compounded metric

## Are there any user-facing changes?

There are user-facing changes, but it's more of a fix to a metric
reported to the user. No change to API itself.

Author: feniljain

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,
 };
 
@@ -79,11 +79,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 {
@@ -169,11 +174,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};
 
@@ -682,15 +682,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);
     }
 }