This is an automated email from the ASF dual-hosted git repository.
jackylee-ch 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 b02dfc5253 [GLUTEN-13197][CORE] Bind GetStructField chains by exprId
(#13198)
b02dfc5253 is described below
commit b02dfc5253fabaf6eef48bea5ed7cfede31da666
Author: YangJie <[email protected]>
AuthorDate: Sun Oct 11 21:41:39 2026 -0400
[GLUTEN-13197][CORE] Bind GetStructField chains by exprId (#13198)
---
.../gluten/expression/StructFieldBindSuite.scala | 168 +++++++++++++++++++++
.../gluten/expression/StructFieldBindSuite.scala | 168 +++++++++++++++++++++
.../gluten/expression/ExpressionConverter.scala | 53 +++----
3 files changed, 363 insertions(+), 26 deletions(-)
diff --git
a/backends-bolt/src/test/scala/org/apache/gluten/expression/StructFieldBindSuite.scala
b/backends-bolt/src/test/scala/org/apache/gluten/expression/StructFieldBindSuite.scala
new file mode 100644
index 0000000000..7aa90dea4d
--- /dev/null
+++
b/backends-bolt/src/test/scala/org/apache/gluten/expression/StructFieldBindSuite.scala
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.expression
+
+import org.apache.gluten.execution.{BoltWholeStageTransformerSuite,
ProjectExecTransformer}
+
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression}
+import org.apache.spark.sql.catalyst.expressions.GetStructField
+import org.apache.spark.sql.execution.ProjectExec
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType,
StructField, StructType}
+
+class StructFieldBindSuite extends BoltWholeStageTransformerSuite {
+ override protected val resourcePath: String = ""
+ override protected val fileFormat: String = "parquet"
+
+ private def bind(expr: Expression, input: Seq[AttributeReference]) =
+ ExpressionConverter
+ .replaceWithExpressionTransformer(expr, input)
+ .asInstanceOf[BoltGetStructFieldTransformer]
+
+ private def readsStructField(expressions: Seq[Expression]): Boolean =
+ expressions.exists(_.find(_.isInstanceOf[GetStructField]).isDefined)
+
+ // The struct fields must be read by a native projection, or the result
comparison would not
+ // exercise the binding.
+ private def assertStructReadOffloaded(df: DataFrame): Unit = {
+ val plan = df.queryExecution.executedPlan
+ assert(
+ collect(plan) { case p: ProjectExecTransformer if
readsStructField(p.projectList) => p }
+ .nonEmpty,
+ plan)
+ }
+
+ test("struct field binding resolves the root attribute by exprId, not by
name") {
+ // Two same-named struct attributes with different layouts, as after a
join of two tables
+ // that both have a struct column `s`.
+ val target =
+ AttributeReference(
+ "s",
+ StructType(Seq(StructField("id", IntegerType), StructField("v",
StringType))))()
+ val decoy =
+ AttributeReference(
+ "s",
+ StructType(Seq(StructField("v", IntegerType), StructField("z",
StringType))))()
+
+ val transformer = bind(GetStructField(target, ordinal = 1, Some("v")),
Seq(target, decoy))
+
+ assert(transformer.ordinal == 1)
+ }
+
+ test("struct field binding keeps the ordinal of a duplicate field name") {
+ val s = AttributeReference(
+ "s",
+ StructType(Seq(StructField("a", IntegerType), StructField("a",
LongType))))()
+
+ assert(bind(GetStructField(s, ordinal = 0), Seq(s)).ordinal == 0)
+ assert(bind(GetStructField(s, ordinal = 1), Seq(s)).ordinal == 1)
+ }
+
+ test("struct field binding looks fields up by name in a pruned input
struct") {
+ // The input carries the same attribute with a pruned struct type, so the
expression's
+ // ordinals do not apply and each level is found by name: s.b.c is field 0
of field 1.
+ val inner = StructType(Seq(StructField("a", IntegerType), StructField("c",
StringType)))
+ val s = AttributeReference(
+ "s",
+ StructType(Seq(StructField("id", IntegerType), StructField("b",
inner))))()
+ val pruned = AttributeReference(
+ "s",
+ StructType(
+ Seq(
+ StructField("id", IntegerType),
+ StructField("b", StructType(Seq(StructField("c",
StringType)))))))(exprId = s.exprId)
+ val level1 = GetStructField(s, ordinal = 1, Some("b"))
+ val level2 = GetStructField(level1, ordinal = 1, Some("c"))
+
+ val transformer = bind(level2, Seq(pruned))
+
+ assert(transformer.ordinal == 0)
+
assert(transformer.child.asInstanceOf[BoltGetStructFieldTransformer].ordinal ==
1)
+ }
+
+ test("struct field binding falls back on an ambiguous name in a pruned input
struct") {
+ val s = AttributeReference(
+ "s",
+ StructType(
+ Seq(
+ StructField("a", IntegerType),
+ StructField("a", LongType),
+ StructField("z", StringType))))()
+ val pruned = AttributeReference(
+ "s",
+ StructType(Seq(StructField("a", IntegerType), StructField("a",
LongType))))(
+ exprId = s.exprId)
+
+ val e = intercept[UnsupportedOperationException] {
+ bind(GetStructField(s, ordinal = 1), Seq(pruned))
+ }
+ assert(e.getMessage.contains("ambiguous field a"), e.getMessage)
+ }
+
+ test("same-named struct columns with different layouts in a join") {
+ // Both join sides carry a struct column `s`, with the fields in a
different order. The
+ // query reads every field, so nested column pruning keeps the whole
structs above the join
+ // and the projection there sees both `s` attributes.
+ withTempPath {
+ dir =>
+ val left = new java.io.File(dir, "left").getCanonicalPath
+ val right = new java.io.File(dir, "right").getCanonicalPath
+ spark
+ .sql("select id, named_struct('x', id, 'y', id * 100) as s from
range(5)")
+ .write
+ .parquet(left)
+ spark
+ .sql("select id, named_struct('y', id * 100, 'x', id) as s from
range(5)")
+ .write
+ .parquet(right)
+ withTempView("l", "r") {
+ spark.read.parquet(left).createOrReplaceTempView("l")
+ spark.read.parquet(right).createOrReplaceTempView("r")
+ runQueryAndCompare("select l.s.x, l.s.y, r.s.x, r.s.y from l join r
on l.id = r.id") {
+ assertStructReadOffloaded
+ }
+ }
+ }
+ }
+
+ test("struct fields with duplicate names fall back when the input struct
type differs") {
+ // The empty branch makes the first field of the union's struct nullable.
Once the empty branch
+ // is removed, the projection reads the other branch's struct under the
same exprId but with a
+ // different type, so fields are looked up by name, and two fields named a
must fall back.
+ runQueryAndCompare(
+ "select s.* from (" +
+ "select /*+ REPARTITION(2) */ named_struct('a', id, 'a', id * 10) as s
from range(5) " +
+ "union all " +
+ "select named_struct('a', cast(null as bigint), 'a', id) as s from
range(5) where false)",
+ noFallBack = false
+ ) {
+ df =>
+ val plan = df.queryExecution.executedPlan
+ assert(
+ collect(plan) { case p: ProjectExec if
readsStructField(p.projectList) => p }.nonEmpty,
+ plan)
+ }
+ }
+
+ test("struct fields with duplicate names read through the generate rewrite")
{
+ // The rewrite reads the inline output through a struct whose fields are
named after the
+ // aliases, so both columns read a field named x and only the ordinal
tells them apart.
+ runQueryAndCompare(
+ "select v.* from (select id, array(named_struct('p', id, 'q', id * 10))
as arr " +
+ "from range(5)) lateral view inline(arr) v as x,
x")(assertStructReadOffloaded)
+ }
+}
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/expression/StructFieldBindSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/expression/StructFieldBindSuite.scala
new file mode 100644
index 0000000000..2603e9e89b
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/expression/StructFieldBindSuite.scala
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.expression
+
+import org.apache.gluten.execution.{ProjectExecTransformer,
VeloxWholeStageTransformerSuite}
+
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression}
+import org.apache.spark.sql.catalyst.expressions.GetStructField
+import org.apache.spark.sql.execution.ProjectExec
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType,
StructField, StructType}
+
+class StructFieldBindSuite extends VeloxWholeStageTransformerSuite {
+ override protected val resourcePath: String = ""
+ override protected val fileFormat: String = "parquet"
+
+ private def bind(expr: Expression, input: Seq[AttributeReference]) =
+ ExpressionConverter
+ .replaceWithExpressionTransformer(expr, input)
+ .asInstanceOf[VeloxGetStructFieldTransformer]
+
+ private def readsStructField(expressions: Seq[Expression]): Boolean =
+ expressions.exists(_.find(_.isInstanceOf[GetStructField]).isDefined)
+
+ // The struct fields must be read by a native projection, or the result
comparison would not
+ // exercise the binding.
+ private def assertStructReadOffloaded(df: DataFrame): Unit = {
+ val plan = df.queryExecution.executedPlan
+ assert(
+ collect(plan) { case p: ProjectExecTransformer if
readsStructField(p.projectList) => p }
+ .nonEmpty,
+ plan)
+ }
+
+ test("struct field binding resolves the root attribute by exprId, not by
name") {
+ // Two same-named struct attributes with different layouts, as after a
join of two tables
+ // that both have a struct column `s`.
+ val target =
+ AttributeReference(
+ "s",
+ StructType(Seq(StructField("id", IntegerType), StructField("v",
StringType))))()
+ val decoy =
+ AttributeReference(
+ "s",
+ StructType(Seq(StructField("v", IntegerType), StructField("z",
StringType))))()
+
+ val transformer = bind(GetStructField(target, ordinal = 1, Some("v")),
Seq(target, decoy))
+
+ assert(transformer.ordinal == 1)
+ }
+
+ test("struct field binding keeps the ordinal of a duplicate field name") {
+ val s = AttributeReference(
+ "s",
+ StructType(Seq(StructField("a", IntegerType), StructField("a",
LongType))))()
+
+ assert(bind(GetStructField(s, ordinal = 0), Seq(s)).ordinal == 0)
+ assert(bind(GetStructField(s, ordinal = 1), Seq(s)).ordinal == 1)
+ }
+
+ test("struct field binding looks fields up by name in a pruned input
struct") {
+ // The input carries the same attribute with a pruned struct type, so the
expression's
+ // ordinals do not apply and each level is found by name: s.b.c is field 0
of field 1.
+ val inner = StructType(Seq(StructField("a", IntegerType), StructField("c",
StringType)))
+ val s = AttributeReference(
+ "s",
+ StructType(Seq(StructField("id", IntegerType), StructField("b",
inner))))()
+ val pruned = AttributeReference(
+ "s",
+ StructType(
+ Seq(
+ StructField("id", IntegerType),
+ StructField("b", StructType(Seq(StructField("c",
StringType)))))))(exprId = s.exprId)
+ val level1 = GetStructField(s, ordinal = 1, Some("b"))
+ val level2 = GetStructField(level1, ordinal = 1, Some("c"))
+
+ val transformer = bind(level2, Seq(pruned))
+
+ assert(transformer.ordinal == 0)
+
assert(transformer.child.asInstanceOf[VeloxGetStructFieldTransformer].ordinal
== 1)
+ }
+
+ test("struct field binding falls back on an ambiguous name in a pruned input
struct") {
+ val s = AttributeReference(
+ "s",
+ StructType(
+ Seq(
+ StructField("a", IntegerType),
+ StructField("a", LongType),
+ StructField("z", StringType))))()
+ val pruned = AttributeReference(
+ "s",
+ StructType(Seq(StructField("a", IntegerType), StructField("a",
LongType))))(
+ exprId = s.exprId)
+
+ val e = intercept[UnsupportedOperationException] {
+ bind(GetStructField(s, ordinal = 1), Seq(pruned))
+ }
+ assert(e.getMessage.contains("ambiguous field a"), e.getMessage)
+ }
+
+ test("same-named struct columns with different layouts in a join") {
+ // Both join sides carry a struct column `s`, with the fields in a
different order. The
+ // query reads every field, so nested column pruning keeps the whole
structs above the join
+ // and the projection there sees both `s` attributes.
+ withTempPath {
+ dir =>
+ val left = new java.io.File(dir, "left").getCanonicalPath
+ val right = new java.io.File(dir, "right").getCanonicalPath
+ spark
+ .sql("select id, named_struct('x', id, 'y', id * 100) as s from
range(5)")
+ .write
+ .parquet(left)
+ spark
+ .sql("select id, named_struct('y', id * 100, 'x', id) as s from
range(5)")
+ .write
+ .parquet(right)
+ withTempView("l", "r") {
+ spark.read.parquet(left).createOrReplaceTempView("l")
+ spark.read.parquet(right).createOrReplaceTempView("r")
+ runQueryAndCompare("select l.s.x, l.s.y, r.s.x, r.s.y from l join r
on l.id = r.id") {
+ assertStructReadOffloaded
+ }
+ }
+ }
+ }
+
+ test("struct fields with duplicate names fall back when the input struct
type differs") {
+ // The empty branch makes the first field of the union's struct nullable.
Once the empty branch
+ // is removed, the projection reads the other branch's struct under the
same exprId but with a
+ // different type, so fields are looked up by name, and two fields named a
must fall back.
+ runQueryAndCompare(
+ "select s.* from (" +
+ "select /*+ REPARTITION(2) */ named_struct('a', id, 'a', id * 10) as s
from range(5) " +
+ "union all " +
+ "select named_struct('a', cast(null as bigint), 'a', id) as s from
range(5) where false)",
+ noFallBack = false
+ ) {
+ df =>
+ val plan = df.queryExecution.executedPlan
+ assert(
+ collect(plan) { case p: ProjectExec if
readsStructField(p.projectList) => p }.nonEmpty,
+ plan)
+ }
+ }
+
+ test("struct fields with duplicate names read through the generate rewrite")
{
+ // The rewrite reads the inline output through a struct whose fields are
named after the
+ // aliases, so both columns read a field named x and only the ordinal
tells them apart.
+ runQueryAndCompare(
+ "select v.* from (select id, array(named_struct('p', id, 'q', id * 10))
as arr " +
+ "from range(5)) lateral view inline(arr) v as x,
x")(assertStructReadOffloaded)
+ }
+}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
index 9bfd90cdd7..0464734c32 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
@@ -980,8 +980,6 @@ object ExpressionConverter extends SQLConfHelper with
Logging {
private def bindGetStructField(
structField: GetStructField,
input: AttributeSeq): BoundReference = {
- // get the new ordinal base input
- var newOrdinal: Int = -1
val names = new ArrayBuffer[String]
var root: Expression = structField
while (root.isInstanceOf[GetStructField]) {
@@ -994,31 +992,34 @@ object ExpressionConverter extends SQLConfHelper with
Logging {
if (!root.isInstanceOf[AttributeReference]) {
return BoundReference(structField.ordinal, structField.dataType,
structField.nullable)
}
- names += root.asInstanceOf[AttributeReference].name
- input.attrs.foreach(
- attribute => {
- var level = names.size - 1
- if (names(level) == attribute.name) {
- var candidateFields: Array[StructField] = null
- var dtType = attribute.dataType
- while (dtType.isInstanceOf[StructType] && level >= 1) {
- candidateFields = dtType.asInstanceOf[StructType].fields
- level -= 1
- val curName = names(level)
- for (i <- 0 until candidateFields.length) {
- if (candidateFields(i).name == curName) {
- dtType = candidateFields(i).dataType
- newOrdinal = i
- }
- }
- }
- }
- })
- if (newOrdinal == -1) {
+ def cannotBind(reason: String): Nothing =
throw new IllegalStateException(
- s"Couldn't find $structField in ${input.attrs.mkString("[", ",",
"]")}")
- } else {
- BoundReference(newOrdinal, structField.dataType, structField.nullable)
+ s"Couldn't bind $structField ($reason) in ${input.attrs.mkString("[",
",", "]")}")
+ // Bind the root by exprId like BindReferences: the input can hold several
same-named struct
+ // attributes with different layouts, e.g. a struct column of the same
name on both join sides.
+ val rootAttr = root.asInstanceOf[AttributeReference]
+ val rootIndex = input.indexOf(rootAttr.exprId)
+ if (rootIndex < 0) {
+ cannotBind(s"no input attribute with the exprId of ${rootAttr.name}")
+ }
+ val attribute = input.attrs(rootIndex)
+ // Same layout as the one the expression was resolved against, so its own
ordinal is right.
+ // It is also the only right answer when the struct has duplicate field
names.
+ if (attribute.dataType == rootAttr.dataType) {
+ return BoundReference(structField.ordinal, structField.dataType,
structField.nullable)
+ }
+ // The input struct type differs (e.g. nested nullability merged by a
union): look each level
+ // up by name from the root down, and fall back instead of guessing on a
duplicate name.
+ val (_, ordinal) = names.reverseIterator.foldLeft[(DataType,
Int)]((attribute.dataType, -1)) {
+ case ((struct: StructType, _), name) =>
+ struct.fields.indexWhere(_.name == name) match {
+ case -1 => cannotBind(s"no field $name")
+ case idx if struct.fields.lastIndexWhere(_.name == name) != idx =>
+ cannotBind(s"ambiguous field $name")
+ case idx => (struct.fields(idx).dataType, idx)
+ }
+ case ((other, _), name) => cannotBind(s"field $name looked up in
non-struct $other")
}
+ BoundReference(ordinal, structField.dataType, structField.nullable)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
