This is an automated email from the ASF dual-hosted git repository.
taiyang-li 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 99f8383746 [GLUTEN] Fix wrong input_file_name() for BHJ build-side
LocalRelation (#12272)
99f8383746 is described below
commit 99f838374607bf443662e52a024dd2da4dde3e56
Author: 李扬 <[email protected]>
AuthorDate: Fri Jun 12 16:45:29 2026 +0800
[GLUTEN] Fix wrong input_file_name() for BHJ build-side LocalRelation
(#12272)
---
.../functions/ScalarFunctionsValidateSuite.scala | 44 ++++++++++++++++++++++
.../columnar/PushDownInputFileExpression.scala | 27 +++++++++++--
2 files changed, 68 insertions(+), 3 deletions(-)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
index cd49c23029..74b2933a67 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
@@ -1077,6 +1077,50 @@ abstract class ScalarFunctionsValidateSuite extends
FunctionsValidateSuite {
}
}
+ test("input_file_name() with BHJ build-side LocalRelation must return real
path") {
+ withTempPath {
+ path =>
+ Seq(("event_a", 1001L, "param1"))
+ .toDF("event", "device_id", "params")
+ .write
+ .parquet(path.getCanonicalPath)
+
spark.read.parquet(path.getCanonicalPath).createOrReplaceTempView("event_log")
+
+ withSQLConf(
+ "spark.sql.autoBroadcastJoinThreshold" -> "10MB",
+ "spark.sql.adaptive.enabled" -> "true"
+ ) {
+ val sql =
+ """
+ |SELECT a.event,
+ | a.params,
+ | a.device_id,
+ | input_file_name() AS fname
+ |FROM event_log a
+ |JOIN
+ | (
+ | SELECT 'event_a' AS envent,
+ | 1001 AS device_id
+ | ) b
+ |ON a.event = b.envent
+ |AND a.device_id = b.device_id
+ |""".stripMargin
+
+ compareResultsAgainstVanillaSpark(sql, true, { _ => })
+
+ val df = spark.sql(sql)
+ val rows = df.collect()
+ assert(rows.nonEmpty, "Join should match at least one row")
+ rows.foreach {
+ r =>
+ val fname = r.getAs[String]("fname")
+ assert(fname != null && fname.nonEmpty)
+ assert(fname.contains(path.getName))
+ }
+ }
+ }
+ }
+
testWithMinSparkVersion("array insert", "3.4") {
withTempPath {
path =>
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala
index 7c9d1bea2d..8e3a2d53ad 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala
@@ -21,7 +21,8 @@ import
org.apache.gluten.execution.{BatchScanExecTransformerBase, FileSourceScan
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, Expression, InputFileBlockLength, InputFileBlockStart,
InputFileName, NamedExpression}
import org.apache.spark.sql.catalyst.optimizer.CollapseProjectShim
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.{DeserializeToObjectExec, LeafExecNode,
ProjectExec, SerializeFromObjectExec, SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.{DeserializeToObjectExec,
FileSourceScanExec, LeafExecNode, ProjectExec, SerializeFromObjectExec,
SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.hive.HiveTableScanExecTransformer
import java.util.Locale
@@ -87,7 +88,8 @@ object PushDownInputFileExpression {
object PreOffload extends Rule[SparkPlan] {
override def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
- case ProjectExec(projectList, child) if
projectList.exists(containsInputFileRelatedExpr) =>
+ case ProjectExec(projectList, child)
+ if projectList.exists(containsInputFileRelatedExpr) &&
hasInputFileRelatedSource(child) =>
val replacedExprs = mutable.Map[String, Alias]()
val newProjectList = projectList.map {
expr => rewriteExpr(expr,
replacedExprs).asInstanceOf[NamedExpression]
@@ -104,8 +106,10 @@ object PushDownInputFileExpression {
// For BatchScanExecTransformerBase (includes Iceberg scans), add
fallback tag
// to prevent offloading when input_file expressions are present
addFallbackTag(ProjectExec(p.output ++ replacedExprs.values, p))
- case p: LeafExecNode =>
+ case p: LeafExecNode if shouldAddInputFileExpr(p) =>
addFallbackTag(ProjectExec(p.output ++ replacedExprs.values, p))
+ case p: LeafExecNode =>
+ p
// Output of SerializeFromObjectExec's child and output of
DeserializeToObjectExec must be
// a single-field row.
case p @ (_: SerializeFromObjectExec | _: DeserializeToObjectExec) =>
@@ -127,6 +131,23 @@ object PushDownInputFileExpression {
u.copy(children = newFirstChild +: newOtherChildren)
case p => p.withNewChildren(p.children.map(child =>
addMetadataCol(child, replacedExprs)))
}
+
+ private def hasInputFileRelatedSource(plan: SparkPlan): Boolean = {
+ plan match {
+ case _: BatchScanExecTransformerBase => true
+ case p: LeafExecNode => shouldAddInputFileExpr(p)
+ case _ => plan.children.exists(hasInputFileRelatedSource)
+ }
+ }
+
+ private def shouldAddInputFileExpr(plan: SparkPlan): Boolean = {
+ plan match {
+ case _: FileSourceScanExec => true
+ case _: BatchScanExec => true
+ case p if HiveTableScanExecTransformer.isHiveTableScan(p) => true
+ case _ => false
+ }
+ }
}
object PostOffload extends Rule[SparkPlan] {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
