You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by ho...@apache.org on 2026/06/09 09:25:12 UTC
(gluten) branch main updated: [GLUTEN-10511][VL][Delta] Fix wrong result with partition filters under column mapping (#12240)
This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new bdaad2a047 [GLUTEN-10511][VL][Delta] Fix wrong result with partition filters under column mapping (#12240)
bdaad2a047 is described below
commit bdaad2a047c91d98e98f610e64cb75ce80f0f6df
Author: EJ Song <51...@users.noreply.github.com>
AuthorDate: Tue Jun 9 02:25:05 2026 -0700
[GLUTEN-10511][VL][Delta] Fix wrong result with partition filters under column mapping (#12240)
Co-authored-by: Claude Opus 4.7 (1M context) <no...@anthropic.com>
---
.../gluten/execution/DeltaScanTransformer.scala | 31 ++-
.../gluten/extension/DeltaPostTransformRules.scala | 90 ++++++---
.../org/apache/gluten/execution/DeltaSuite.scala | 213 +++++++++++++++++++++
3 files changed, 305 insertions(+), 29 deletions(-)
diff --git a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
index 1be03dd404..154b735554 100644
--- a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
+++ b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
@@ -20,9 +20,10 @@ import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.delta.{DeltaParquetFileFormat, NoMapping}
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.datasources.HadoopFsRelation
import org.apache.spark.sql.types.StructType
@@ -55,6 +56,34 @@ case class DeltaScanTransformer(
override lazy val fileFormat: ReadFileFormat = ReadFileFormat.ParquetReadFormat
+ // For Delta column-mapping tables, `dataFilters` on the scan node are LOGICAL-named so Delta's
+ // file index (`PreparedDeltaFileIndex.matchingFiles`, `Snapshot.filesForScan`) can do partition
+ // pruning and stats-based file skipping -- both resolve filter attrs against logical schemas.
+ //
+ // The native (Velox) side, however, must see PHYSICAL names: `output` and `dataSchema` are
+ // physical (so the parquet reader finds the right column), and `BasicScanExecTransformer`
+ // matches `scanFilters` against `pushDownFilters` (built from a `Filter` that references the
+ // physical-named scan output) by `AttributeReference.equals`, which compares names. Without
+ // this override, the logical-named `scanFilters` and physical-named `pushDownFilters` would
+ // never match, causing duplicate filter evaluation in the substrait plan.
+ //
+ // Translate by exprId match against `output` rather than by re-running Delta's column-mapping
+ // helpers; exprIds are stable across the post-transform rewrite and don't require a second
+ // metadata lookup.
+ //
+ // See `DeltaPostTransformRules.transformColumnMappingPlan` for the full picture of which
+ // fields stay logical vs. become physical, and the longer-term cleanup direction (do all
+ // physical translation at substrait emission time so this override and the alias-back
+ // ProjectExec both go away).
+ override def scanFilters: Seq[Expression] = relation.fileFormat match {
+ case d: DeltaParquetFileFormat if d.columnMappingMode != NoMapping =>
+ val physicalByExprId = output.collect { case ar: AttributeReference => ar.exprId -> ar }.toMap
+ dataFilters.map(_.transformDown {
+ case ar: AttributeReference => physicalByExprId.getOrElse(ar.exprId, ar)
+ })
+ case _ => dataFilters
+ }
+
override protected def doValidateInternal(): ValidationResult = {
if (
requiredSchema.fields.exists(
diff --git a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala
index e16a6d12fd..421db09732 100644
--- a/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala
+++ b/gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala
@@ -164,20 +164,55 @@ object DeltaPostTransformRules {
}
/**
- * This method is only used for Delta ColumnMapping FileFormat(e.g. nameMapping and idMapping)
- * transform the metadata of Delta into Parquet's, each plan should only be transformed once.
+ * Used for Delta ColumnMapping FileFormat (nameMapping and idMapping). Each plan is transformed
+ * at most once; the first run is tagged so re-runs are no-ops.
+ *
+ * Background: with column mapping, Delta files are written with PHYSICAL column names while
+ * Delta's metadata (partition schema, column stats) keeps LOGICAL names. Vanilla Spark + Delta
+ * resolves this asymmetry inside `DeltaParquetFileFormat.buildReaderWithPartitionValues`:
+ * everything on the scan node stays logical, and physical translation happens just-in-time when
+ * handing data and filters to the parquet reader. Gluten bypasses that hook (it goes to native
+ * via Substrait), so the translation has to live somewhere on our side.
+ *
+ * What this rule produces -- the parts that diverge from vanilla Spark are commented at each
+ * site. The split-by-consumer is asymmetric on purpose:
+ *
+ * - `output`, `dataSchema`, and the data fields of `requiredSchema` ==> PHYSICAL. These flow
+ * into the substrait `NamedStruct` that Velox uses to look up columns in the parquet file.
+ * The parquet column name is the physical name, so Velox needs the physical name on the
+ * schema side. A `ProjectExecTransformer` is added below to alias these back to logical names
+ * for downstream Spark operators.
+ * - `partitionSchema`, `partitionFilters`, `dataFilters`, partition fields of `requiredSchema`
+ * ==> LOGICAL. These are consumed by Delta's `PreparedDeltaFileIndex.matchingFiles` and
+ * `Snapshot.filesForScan`, which resolve filters and partition values against
+ * `metadata.partitionSchema` and the column-stats schema -- both LOGICAL. Rewriting any of
+ * these to physical names was the cause of issue #10511 (partition pruning silently no-op'd)
+ * and would also disable file-level stats skipping.
+ * - `DeltaScanTransformer.scanFilters` (override) ==> PHYSICAL, translated from `dataFilters`
+ * by exprId match against `output`. Substrait binds filters by exprId rather than name, so it
+ * would be tempting to pass logical-named filters straight through; but
+ * `BasicScanExecTransformer.filterExprs()` does a name-and-exprId equality check
+ * (`scanFilters.partition(pushDownFilters.contains(_))`) against the physical-named
+ * `pushDownFilters` from the upstream `Filter`. The override ensures both sides match.
+ *
+ * Future cleanup (out of scope for this fix): the cleaner shape is to mirror vanilla Spark
+ * exactly -- keep EVERYTHING on the scan node logical, and do physical translation only at
+ * substrait emission time (e.g. inside the `NamedStruct`/`ReadRel` build in
+ * `BasicScanExecTransformer.doTransform`). That removes the alias-back project below and the
+ * `scanFilters` override, but it requires plumbing Delta-specific physical-name lookup into the
+ * substrait emitter and is a multi-module refactor.
*/
private def transformColumnMappingPlan(plan: SparkPlan): SparkPlan = plan match {
case plan: DeltaScanTransformer =>
val fmt = plan.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat]
- // transform HadoopFsRelation
val relation = plan.relation
+ val partitionColNames = relation.partitionSchema.fields.iterator.map(_.name).toSet
+ def isPartitionCol(name: String): Boolean = partitionColNames.contains(name)
+
+ // transform HadoopFsRelation: only `dataSchema` needs physical names (those are the
+ // columns actually stored in parquet). `partitionSchema` stays logical.
val newFsRelation = relation.copy(
- partitionSchema = DeltaColumnMapping.createPhysicalSchema(
- relation.partitionSchema,
- fmt.referenceSchema,
- fmt.columnMappingMode),
dataSchema = DeltaColumnMapping.createPhysicalSchema(
relation.dataSchema,
fmt.referenceSchema,
@@ -193,6 +228,8 @@ object DeltaPostTransformRules {
attr
} else if (isInputFileRelatedAttribute(attr)) {
attr
+ } else if (isPartitionCol(attr.name)) {
+ attr
} else {
DeltaColumnMapping
.createPhysicalAttributes(Seq(attr), fmt.referenceSchema, fmt.columnMappingMode)
@@ -204,31 +241,28 @@ object DeltaPostTransformRules {
newAttr
}
val newOutput = plan.output.map(o => mapAttribute(o))
- // transform dataFilters
- val newDataFilters = plan.dataFilters.map {
- e =>
- e.transformDown {
- case attr: AttributeReference =>
- mapAttribute(attr)
- }
- }
- // transform partitionFilters
- val newPartitionFilters = plan.partitionFilters.map {
- e =>
- e.transformDown {
- case attr: AttributeReference =>
- mapAttribute(attr)
- }
- }
- // replace tableName in schema with physicalName
+ // dataFilters / partitionFilters: kept LOGICAL on the scan node so Delta's file index
+ // (partition pruning + stats-based file skipping) resolves columns correctly. The native
+ // (Velox) side gets physical-translated copies via `DeltaScanTransformer.scanFilters`.
+ val newDataFilters = plan.dataFilters
+ val newPartitionFilters = plan.partitionFilters
+
+ // requiredSchema: rewrite data fields to physical, keep partition fields logical.
+ val physicalRequiredSchema = DeltaColumnMapping.createPhysicalSchema(
+ plan.requiredSchema,
+ fmt.referenceSchema,
+ fmt.columnMappingMode)
+ val newRequiredSchema = StructType(
+ physicalRequiredSchema.fields.zip(plan.requiredSchema.fields).map {
+ case (_, logical) if isPartitionCol(logical.name) => logical
+ case (physical, _) => physical
+ })
+
val scanExecTransformer = new DeltaScanTransformer(
newFsRelation,
plan.stream,
newOutput,
- DeltaColumnMapping.createPhysicalSchema(
- plan.requiredSchema,
- fmt.referenceSchema,
- fmt.columnMappingMode),
+ newRequiredSchema,
newPartitionFilters,
plan.optionalBucketSet,
plan.optionalNumCoalescedBuckets,
diff --git a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
index 031bf46034..1f95ec7dac 100644
--- a/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
+++ b/gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala
@@ -82,6 +82,219 @@ abstract class DeltaSuite extends WholeStageTransformerSuite {
}
}
+ // Counts files Delta will read for `df`. Driven by `PreparedDeltaFileIndex.inputFiles`, which
+ // is the post-pruning, post-stats-skipping file set computed by `PrepareDeltaScan`. Useful for
+ // asserting that Delta's file index actually pruned, regardless of what Gluten does later.
+ private def deltaInputFileCount(df: org.apache.spark.sql.DataFrame): Int =
+ df.inputFiles.length
+
+ // Counts the partition directories selected by the executed scan after Gluten's rewrite.
+ // Backed by `selectedPartitions` -> `relation.location.listFiles(partitionFilters, dataFilters)`,
+ // which is the exact call site of issue #10511: pre-fix, physical-named partition filters
+ // could not match Delta's logical partition schema, so this returned all directories. We use
+ // `getPartitionArray` rather than `getPartitions` because the latter reflects post-coalesce
+ // splits (Velox may merge small files into one split, hiding the per-partition count).
+ private def selectedPartitionCount(df: org.apache.spark.sql.DataFrame): Int = {
+ val scan = df.queryExecution.executedPlan.collect {
+ case f: DeltaScanTransformer => f
+ }.head
+ scan.getPartitionArray.length
+ }
+
+ // Regression for issue #10511: with column mapping, a partition column filter must prune
+ // partitions correctly. Pre-fix, Gluten rewrote partition filters to physical names, which
+ // broke `PreparedDeltaFileIndex.matchingFiles` and silently returned all files.
+ Seq("name", "id").foreach {
+ mode =>
+ testWithMinSparkVersion(
+ s"column mapping mode = $mode with partition filter (single partition col)",
+ "3.2") {
+ withTable("delta_cm_part") {
+ spark.sql(s"""
+ |create table delta_cm_part (id int, name string) using delta
+ |partitioned by (id)
+ |tblproperties ("delta.columnMapping.mode" = "$mode")
+ |""".stripMargin)
+ // Use multiple inserts so each value lands in its own partition directory & file.
+ spark.sql("insert into delta_cm_part values (1, \"v1\")")
+ spark.sql("insert into delta_cm_part values (2, \"v2\")")
+ spark.sql("insert into delta_cm_part values (3, \"v3\")")
+
+ // Equality on partition column. 1 of 3 partitions matches.
+ val df1 = runQueryAndCompare("select name from delta_cm_part where id = 2") { _ => }
+ checkLengthAndPlan(df1, 1)
+ checkAnswer(df1, Row("v2") :: Nil)
+ assert(deltaInputFileCount(df1) == 1, "Delta should prune to 1 file")
+ assert(selectedPartitionCount(df1) == 1, "native scan should see 1 split")
+
+ // Range on partition column (the exact case from the bug report).
+ val df2 = runQueryAndCompare("select name from delta_cm_part where id > 2") { _ => }
+ checkLengthAndPlan(df2, 1)
+ checkAnswer(df2, Row("v3") :: Nil)
+ assert(deltaInputFileCount(df2) == 1)
+ assert(selectedPartitionCount(df2) == 1)
+
+ // IN list on partition column. 2 of 3 partitions match.
+ val df3 =
+ runQueryAndCompare("select name from delta_cm_part where id in (1, 3)") { _ => }
+ checkLengthAndPlan(df3, 2)
+ checkAnswer(df3, Row("v1") :: Row("v3") :: Nil)
+ assert(deltaInputFileCount(df3) == 2)
+ assert(selectedPartitionCount(df3) == 2)
+
+ // No filter -- baseline: all 3 partitions read.
+ val dfAll = runQueryAndCompare("select name from delta_cm_part") { _ => }
+ assert(deltaInputFileCount(dfAll) == 3)
+ assert(selectedPartitionCount(dfAll) == 3)
+ }
+ }
+
+ testWithMinSparkVersion(
+ s"column mapping mode = $mode with partition filter (multi partition col)",
+ "3.2") {
+ withTable("delta_cm_part_multi") {
+ spark.sql(s"""
+ |create table delta_cm_part_multi
+ | (id int, region string, name string)
+ |using delta partitioned by (region, id)
+ |tblproperties ("delta.columnMapping.mode" = "$mode")
+ |""".stripMargin)
+ spark.sql("insert into delta_cm_part_multi values (1, \"us\", \"v1\")")
+ spark.sql("insert into delta_cm_part_multi values (2, \"us\", \"v2\")")
+ spark.sql("insert into delta_cm_part_multi values (1, \"eu\", \"v3\")")
+ spark.sql("insert into delta_cm_part_multi values (2, \"eu\", \"v4\")")
+
+ val df = runQueryAndCompare(
+ "select name from delta_cm_part_multi where region = 'us' and id > 1") { _ => }
+ checkLengthAndPlan(df, 1)
+ checkAnswer(df, Row("v2") :: Nil)
+ assert(deltaInputFileCount(df) == 1, "Delta should prune to 1 file with both filters")
+ assert(selectedPartitionCount(df) == 1)
+
+ // Filter on only one of two partition columns.
+ val df2 = runQueryAndCompare(
+ "select name from delta_cm_part_multi where region = 'eu'") { _ => }
+ checkLengthAndPlan(df2, 2)
+ checkAnswer(df2, Row("v3") :: Row("v4") :: Nil)
+ assert(deltaInputFileCount(df2) == 2)
+ assert(selectedPartitionCount(df2) == 2)
+ }
+ }
+
+ testWithMinSparkVersion(
+ s"column mapping mode = $mode with partition + data filter",
+ "3.2") {
+ withTable("delta_cm_part_data") {
+ spark.sql(s"""
+ |create table delta_cm_part_data (id int, name string, age int)
+ |using delta partitioned by (id)
+ |tblproperties ("delta.columnMapping.mode" = "$mode")
+ |""".stripMargin)
+ spark.sql("insert into delta_cm_part_data values (1, \"a\", 10), (1, \"b\", 20)")
+ spark.sql("insert into delta_cm_part_data values (2, \"c\", 30), (2, \"d\", 40)")
+ spark.sql("insert into delta_cm_part_data values (3, \"e\", 50), (3, \"f\", 60)")
+
+ // Combined: partition pruning to id > 1 keeps 2 files; data stats-skipping on age >= 50
+ // further drops the id=2 file (max age 40 < 50). Should leave 1 file.
+ val df1 = runQueryAndCompare(
+ "select name from delta_cm_part_data where id > 1 and age >= 50") { _ => }
+ checkLengthAndPlan(df1, 2)
+ checkAnswer(df1, Row("e") :: Row("f") :: Nil)
+ assert(
+ deltaInputFileCount(df1) == 1,
+ "partition + stats-skipping should leave 1 file out of 3")
+
+ // Data filter alone -- file-level stats skipping should resolve column names.
+ // Only the id=2 file (age 30..40) matches age = 30.
+ val df2 = runQueryAndCompare(
+ "select name from delta_cm_part_data where age = 30") { _ => }
+ checkLengthAndPlan(df2, 1)
+ checkAnswer(df2, Row("c") :: Nil)
+ assert(
+ deltaInputFileCount(df2) == 1,
+ "stats-based file skipping should leave 1 file out of 3")
+ }
+ }
+
+ testWithMinSparkVersion(
+ s"column mapping mode = $mode with IS [NOT] NULL on partition col",
+ "3.2") {
+ withTable("delta_cm_part_null") {
+ spark.sql(s"""
+ |create table delta_cm_part_null (id int, name string)
+ |using delta partitioned by (id)
+ |tblproperties ("delta.columnMapping.mode" = "$mode")
+ |""".stripMargin)
+ spark.sql("insert into delta_cm_part_null values (1, \"v1\")")
+ spark.sql("insert into delta_cm_part_null values (2, \"v2\")")
+ spark.sql("insert into delta_cm_part_null values (cast(null as int), \"vn\")")
+
+ val df1 = runQueryAndCompare(
+ "select name from delta_cm_part_null where id is null") { _ => }
+ checkAnswer(df1, Row("vn") :: Nil)
+ assert(deltaInputFileCount(df1) == 1)
+ assert(selectedPartitionCount(df1) == 1)
+
+ val df2 = runQueryAndCompare(
+ "select name from delta_cm_part_null where id is not null") { _ => }
+ checkAnswer(df2, Row("v1") :: Row("v2") :: Nil)
+ assert(deltaInputFileCount(df2) == 2)
+ assert(selectedPartitionCount(df2) == 2)
+ }
+ }
+
+ testWithMinSparkVersion(
+ s"column mapping mode = $mode partition filter survives column rename",
+ "3.2") {
+ withTable("delta_cm_part_rename") {
+ spark.sql(s"""
+ |create table delta_cm_part_rename (id int, name string)
+ |using delta partitioned by (id)
+ |tblproperties ("delta.columnMapping.mode" = "$mode")
+ |""".stripMargin)
+ spark.sql("insert into delta_cm_part_rename values (1, \"v1\")")
+ spark.sql("insert into delta_cm_part_rename values (2, \"v2\")")
+ spark.sql("insert into delta_cm_part_rename values (3, \"v3\")")
+ // Rename the partition column. The physical name in storage stays the same; only the
+ // logical name changes, so the logical-name-based partition filter must still resolve.
+ spark.sql("alter table delta_cm_part_rename rename column id to pid")
+
+ val df = runQueryAndCompare(
+ "select name from delta_cm_part_rename where pid >= 2") { _ => }
+ checkLengthAndPlan(df, 2)
+ checkAnswer(df, Row("v2") :: Row("v3") :: Nil)
+ assert(deltaInputFileCount(df) == 2)
+ assert(selectedPartitionCount(df) == 2)
+ }
+ }
+
+ testWithMinSparkVersion(
+ s"column mapping mode = $mode data column rename + filter (file skipping)",
+ "3.2") {
+ withTable("delta_cm_data_rename") {
+ spark.sql(s"""
+ |create table delta_cm_data_rename (id int, age int, name string)
+ |using delta
+ |tblproperties ("delta.columnMapping.mode" = "$mode")
+ |""".stripMargin)
+ spark.sql("insert into delta_cm_data_rename values (1, 10, \"a\")")
+ spark.sql("insert into delta_cm_data_rename values (2, 20, \"b\")")
+ spark.sql("insert into delta_cm_data_rename values (3, 30, \"c\")")
+ // Rename a data column. Filter pushdown must still match physical column in parquet,
+ // and Delta's stats-based skipping must still resolve the logical name `years`.
+ spark.sql("alter table delta_cm_data_rename rename column age to years")
+
+ val df = runQueryAndCompare(
+ "select name from delta_cm_data_rename where years = 20") { _ => }
+ checkLengthAndPlan(df, 1)
+ checkAnswer(df, Row("b") :: Nil)
+ assert(
+ deltaInputFileCount(df) == 1,
+ "stats skipping on renamed data column should leave 1 file")
+ }
+ }
+ }
+
test("delta: time travel") {
withTable("delta_tm") {
spark.sql(s"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org