(datafusion-comet) branch main updated: perf: unwrap identity casts in schema adapter to enable Parquet stats pruning (#4730)

[mb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git


The following commit(s) were added to refs/heads/main by this push:
     new 2473aaa9be perf: unwrap identity casts in schema adapter to enable Parquet stats pruning (#4730)
2473aaa9be is described below

commit 2473aaa9bed30ad1d5266f02ff31dd3a62a4b380
Author: Matt Butrovich <mb...@users.noreply.github.com>
AuthorDate: Thu Jun 25 16:03:05 2026 -0400

    perf: unwrap identity casts in schema adapter to enable Parquet stats pruning (#4730)
---
 native/core/src/parquet/schema_adapter.rs          | 15 +++++++
 .../apache/comet/exec/CometNativeReaderSuite.scala | 47 +++++++++++++++++++++-
 2 files changed, 61 insertions(+), 1 deletion(-)

diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs
index 38bcff91a1..c6586b4681 100644
--- a/native/core/src/parquet/schema_adapter.rs
+++ b/native/core/src/parquet/schema_adapter.rs
@@ -645,6 +645,21 @@ impl SparkPhysicalExprAdapter {
             };
             let physical_type = input_field.data_type();
 
+            // Identity cast: DataFusion's default adapter inserts a CastExpr
+            // whenever the logical and physical Arrow Fields differ in any
+            // attribute (data type, nullability, or metadata), so with identical
+            // data types but mismatched nullability or metadata, we receive a
+            // no-op cast. Unwrapping is safe because Spark `Cast` with equal
+            // source and target types is value-level identity (it does not
+            // null-strip or enforce non-null), and field nullability/metadata is
+            // informational rather than computational. Leaving the wrapper in
+            // place blocks DataFusion's pruning-predicate analyzer from
+            // recognizing the column reference, defeating row-group / page-index
+            // stats pruning.
+            if physical_type == target_type {
+                return Ok(Transformed::yes(child));
+            }
+
             // Reject reading a string/binary Parquet column as anything else. Spark's
             // `ParquetVectorUpdaterFactory.getUpdater` BINARY case allows StringType /
             // BinaryType, or DecimalType only when the column carries a
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala
index fc2adcbe6d..5b7b9c8e6a 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala
@@ -24,12 +24,13 @@ import org.scalatest.Tag
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
-import org.apache.parquet.hadoop.ParquetWriter
+import org.apache.parquet.hadoop.{ParquetFileReader, ParquetWriter}
 import org.apache.parquet.hadoop.api.WriteSupport
 import org.apache.parquet.hadoop.api.WriteSupport.WriteContext
 import org.apache.parquet.io.api.RecordConsumer
 import org.apache.parquet.schema.MessageTypeParser
 import org.apache.spark.sql.{CometTestBase, Row}
+import org.apache.spark.sql.comet.CometNativeScanExec
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.functions.{array, col}
 import org.apache.spark.sql.internal.SQLConf
@@ -827,6 +828,50 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
       }
     }
   }
+
+  test("row-group statistics pruning fires for native Parquet scan dataFilters") {
+    // Regression test for the identity-cast pruning gap. DataFusion's default
+    // PhysicalExprAdapter inserts a CastExpr around Column refs whenever the logical
+    // and physical Arrow Fields differ at all, including metadata or nullability with
+    // identical data types. Comet then translates that CastExpr into a Spark Cast,
+    // which DataFusion's pruning-predicate analyzer cannot peel back to a column,
+    // so build_pruning_predicates returns None and no row groups are pruned. The
+    // fix skips the wrap when source and target types are equal.
+    withTempPath { dir =>
+      withSQLConf(SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1") {
+        spark
+          .range(0, 1000)
+          .toDF("c1")
+          .repartition(1)
+          .write
+          .option("parquet.block.size", "1024")
+          .format("parquet")
+          .save(dir.toString)
+
+        val parquetFile = dir
+          .listFiles()
+          .find(_.getName.endsWith(".parquet"))
+          .getOrElse(fail("No parquet file was written"))
+        val reader = ParquetFileReader.open(
+          org.apache.parquet.hadoop.util.HadoopInputFile
+            .fromPath(new Path(parquetFile.getAbsolutePath), spark.sessionState.newHadoopConf()))
+        val numRowGroups =
+          try reader.getRowGroups.size()
+          finally reader.close()
+        assert(numRowGroups > 1, s"Test setup needs >1 row groups, got $numRowGroups")
+
+        val df = spark.read.parquet(dir.toString).where("c1 > 500")
+        val (_, cometPlan) = checkSparkAnswerAndOperator(df)
+        val nativeScans = cometPlan.collect { case n: CometNativeScanExec => n }
+        assert(nativeScans.nonEmpty, "Expected a CometNativeScanExec")
+        val scanRows = nativeScans.head.metrics("numOutputRows").value
+        assert(
+          scanRows < 1000,
+          s"Row-group statistics pruning did not fire (scan read $scanRows of 1000 rows " +
+            s"across $numRowGroups row groups)")
+      }
+    }
+  }
 }
 
 private class DirectWriteSupport(schema: org.apache.parquet.schema.MessageType)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org