This is an automated email from the ASF dual-hosted git repository.
jinchengchenghh 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 12f0f51706 [VL] Respect caseSensitiveAnalysis in
IcebergScanTransformer metadata-column handling (#12726)
12f0f51706 is described below
commit 12f0f5170619de789001546ce9faaf63ed969ff5
Author: sabbasani <[email protected]>
AuthorDate: Tue Oct 27 19:46:19 2026 +0530
[VL] Respect caseSensitiveAnalysis in IcebergScanTransformer
metadata-column handling (#12726)
* Fix] Respect caseSensitiveAnalysis in IcebergScanTransformer
metadata-column handling
* [Fix] Respect caseSensitiveAnalysis in PushDownInputFileExpression
Two remaining unconditional toLowerCase(Locale.ROOT) calls in
PushDownInputFileExpression were inconsistent with spark.sql.caseSensitive:
1. containsInputFileRelatedExpr: used INPUT_FILE_ATTR_NAMES.contains(
a.name.toLowerCase) to detect metadata attribute references.
Under caseSensitive=true this incorrectly matched a user column named
e.g. 'Input_File_Name' as a metadata expression.
Fixed by using SQLConf.get.resolver, which honours the session
case-sensitivity setting exactly as the rest of the engine does.
2. PostOffload dedup (withOutput injection): used
child.output.map(_.name.toLowerCase).toSet to avoid re-injecting a
metadata attribute already present in the scan output.
Under caseSensitive=true a user column whose lowercased name equals a
metadata constant (e.g. 'Input_File_Name' -> 'input_file_name') would
be treated as a duplicate, causing the injected attr to be dropped
while the rewritten project list still held a reference to it,
producing a dangling-attribute IllegalStateException.
Fixed by deduplicating on expression ID instead of name: the injected
aliases are freshly created (new exprId) so identity is the correct
key.
Together with the IcebergScanTransformer fix (2638ce6e5) this closes the
case-sensitivity gap in the Iceberg metadata-column push-down path.
* changes for Gluten fix for case-sensitive metadata fallback
* Fix trailing whitespace in IcebergSuite.scala to resolve scalastyle
violation
* Fix] Respect caseSensitiveAnalysis in IcebergScanTransformer
metadata-column handling
* changes for casesensitive
* [Docs] Update velox-backend-limitations.md: reflect both case-sensitive
fixes
- Remove stale 'Known follow-up' note about PushDownInputFileExpression
(that fix has since been committed in b7568172a).
- Add PushDownInputFileExpression bullet alongside the
IcebergScanTransformer
entry in the 'This PR addresses' section.
- List remaining unverified areas under 'Remaining limitations'.
* case-sensitive-iceberg-fix
* case-sensitive-iceberg-fix
* case-sensitive-iceberg-fix
* case-sensitive-iceberg-fix
* case-sensitive-iceberg-fix
* case-sensitive-iceberg-fix
* Add LeftSingle join support to BroadcastNestedLoopJoinExecTransformer
LeftSingle is a Spark 4.0+ join type for scalar subquery correlation that is
semantically equivalent to LeftOuter. The test 'Left single join should not
result into exception' uses runQueryAndCompare with noFallBack=true, which
requires BroadcastNestedLoopJoin with LeftSingle to be offloaded natively.
Changes:
- BroadcastNestedLoopJoinExecTransformer: add LeftSingle to
validateJoinTypeAndBuildSide()
(allowed with BuildRight, rejected with BuildLeft like LeftOuter) and to
outputPartitioning (returns left.outputPartitioning like LeftOuter).
- SubstraitUtil.toNestedLoopJoinSubstrait: map LeftSingle to JOIN_TYPE_LEFT
(same as LeftOuter, since we always build on the right side in Substrait).
* Merge conflict resolution
* Revert unrelated changes from case-sensitive-iceberg-fix branch
Remove four files that are not part of the core case-sensitive fix:
- ProjectColumnPruning.scala: revert agg buffer (inputAggBufferAttributes)
inclusion in getReferences — unrelated side-effect, not required for
case-sensitive metadata-column resolution.
- ImplementJoinAggregate.scala: revert grouping-only projection in
unpackInputBufferFields — unrelated side-effect, not required for
case-sensitive metadata-column resolution.
- BroadcastNestedLoopJoinExecTransformer.scala: revert LeftSingle join
support (Spark 4.0+) — separate unrelated feature.
- SubstraitUtil.scala: revert LeftSingle → JOIN_TYPE_LEFT mapping —
accompanies the BroadcastNestedLoopJoin revert above.
The remaining PR scope is the core case-sensitive fix only:
IcebergScanTransformer, GlutenIcebergSourceUtil, BasicScanExecTransformer,
PushDownInputFileExpression, CHIteratorApi,
DeltaOptimizedWriterTransformer,
GlutenDeltaOptimizedWriterExec.
* test: strengthen caseSensitive=false fallback assertion in FallbackSuite
The PushDownInputFileExpression caseSensitive=false test previously only
verified correctness of result values, which is insufficient: if
hasVeloxColumnNameConflict() were removed the fallback would not trigger,
Gluten would attempt native execution, and Velox would crash at runtime
(INVALID_STATE: two conflicting column handles for the same case-insensitive
name). The regression would not be caught by value comparison alone because
runQueryAndCompare falls back to Spark for the comparison.
Add two structural plan assertions inside the runQueryAndCompare callback:
1. collect(plan) { FileSourceScanExecTransformer }.isEmpty -- no native
scan
2. collect(plan) { FileSourceScanExec }.nonEmpty -- vanilla
scan present
These assertions will fail immediately if the Velox collision guard is ever
inadvertently removed, making the regression visible at test time rather
than
as a production crash.
Also add FileSourceScanExec to the import list in FallbackSuite.
* fix coding style
* style: fix spotless closing-paren formatting in FallbackSuite
Move closing ')' of assert() call to its own line to satisfy
spotless/scalafmt style rules.
* test: fix PATH_ALREADY_EXISTS in PushDownInputFileExpression Parquet tests
withTempDir (from Spark's SQLTestUtils) creates and pre-initialises the
temp directory before handing it to the lambda body. Both PushDownInput-
FileExpression Parquet tests called df.write.format("parquet").save(dir)
without a save mode, which defaults to ErrorIfExists. Because the directory
already exists, Spark throws:
AnalysisException: [PATH_ALREADY_EXISTS] Path file:/tmp/spark-...
already exists. Set mode as "overwrite" to overwrite.
Fix: add .mode("overwrite") to both df.write chains, matching the pattern
used by all other withTempDir + parquet write tests in the same suite
(VeloxScanSuite, VeloxParquetDataTypeValidationSuite, etc.).
No production logic is changed.
* changes for doc changes
* changes for fix to case sensitive
* changes to casesensitive bug
* changes to casesensitive bug
* changes to casesensitive bug
* changes to casesensitive bug
* fix: restore caseInsensitiveResolution in CH
DeltaOptimizedWriterTransformer
Delta partition column lookup must always be case-insensitive regardless of
spark.sql.caseSensitive. Delta metadata stores canonical partition names
(e.g.
'Part') while the incoming DataFrame may use any case variant (e.g. 'part').
PR #12726 mistakenly replaced caseInsensitiveResolution with
SQLConf.get.resolver
in both outputPartitioning and getShuffleRDD. Under caseSensitive=true this
breaks
optimized writes with case-mismatched partition column names:
[DELTA_FAILED_FIND_PARTITION_COLUMN_IN_OUTPUT_PLAN] Could not find Part
in output plan.
Same fix as backends-velox/src-delta33 GlutenDeltaOptimizedWriterExec
(commit bdae95857).
Also removes the now-unused SQLConf import.
* test: add InputFileNameCollisionSuite regression tests for PR #12726
Add InputFileNameCollisionSuite to backends-velox test sources.
This suite provides regression coverage for the collision between a
mixed-case user column named 'Input_File_Name' and the
PushDownInputFileExpression-injected 'input_file_name' metadata attr
under caseSensitive=false.
Tests cover:
- Combined input_file_name() + Input_File_Name user column queries
under both caseSensitive=true and caseSensitive=false
- Reverse projection order (input_file_name() first)
- Aliased user column + input_file_name()
- Multiple input_file_name() calls in same query
- Filter + input_file_name()
- Standalone input_file_name() queries
- Data values not replaced by file paths (core bug regression)
- input_file_name() through UNION ALL (tail-branch metadata fix)
- Control tests with non-colliding columns
All 19 tests pass after the implementation changes in
PushDownInputFileExpression, BasicScanExecTransformer,
FileMetadataUtil, IteratorApi, and VeloxIteratorApi.
fix: guard PostOffload BatchScan case against fallback-tagged ProjectExec
(Regular/Synthesized Velox conflict)
When caseSensitive=true (or caseSensitive=false) and the physical table
contains
a column named exactly 'input_file_name',
PushDownInputFileExpression.PostOffload
was matching the fallback-tagged inner ProjectExec created by PreOffload for
BatchScanExecTransformerBase (Iceberg/BatchScan tables) and pushing a
Synthesized
'input_file_name' attribute into the scan's output. The scan already had a
Regular
'input_file_name' column, causing Velox to throw:
VeloxRuntimeError: Inconsistent column handle type: input_file_name,
expected Regular, got Synthesized
Root cause:
PostOffload's 'ProjectExec(_, child: BatchScanExecTransformerBase)'
pattern
matches any ProjectExec wrapping a BatchScanExecTransformerBase --
including
the fallback-tagged one produced by PreOffload. PreOffload always marks
the
Iceberg/BatchScan wrapper as fallback so the JVM evaluates
InputFileName() via
the thread-local; PostOffload must not re-push a synthesized column into
the
scan's output in that case.
Fix:
Add '&& !FallbackTags.nonEmpty(p)' guard to both
BatchScanExecTransformerBase
PostOffload cases (the direct-child case and the two-level collapse case).
When the inner ProjectExec is already fallback-tagged, PostOffload leaves
it
untouched and the JVM evaluates the metadata function correctly.
Tests added:
- 'case-sensitive mode: lowercase input_file_name as data column --
platform compatibility'
(mirrors the IcebergSuite CI failing test for the Parquet /
FileSourceScan code path)
- 'case-sensitive mode: lowercase input_file_name as data column --
caseSensitive=false'
- 'exact lowercase input_file_name: only user column selected, no
function call'
- writeExactCollisionTable helper for exact-lowercase schema
All 22 InputFileNameCollisionSuite tests pass.
All 49 FallbackSuite tests pass.
Also rebuilds gluten-substrait with genFormatNumberTransformer now present
in the
SparkPlanExecApi trait (was missing from the installed JAR), resolving the
ScalarFunctionsValidateSuite 'format_number' CI failure.
fix: PostOffload BatchScan - reuse PreOffload mangled aliases to avoid
Regular/Synthesized Velox conflict
Root cause:
PushDownInputFileExpression.PostOffload, when processing a ProjectExec
wrapping a
BatchScanExecTransformerBase (Iceberg/BatchScan), re-ran rewriteExpr from
scratch,
which always creates an injected alias with the un-mangled prettyName
(e.g.
'input_file_name') regardless of whether PreOffload had already mangled
it.
Under caseSensitive=true, when the physical table has a column named
exactly
'input_file_name', Velox receives both:
- a Regular column 'input_file_name' (physical data)
- a Synthesized column 'input_file_name' (injected metadata)
and throws: 'Inconsistent column handle type: input_file_name, expected
Regular,
got Synthesized' (velox/connectors/hive/HiveConnectorUtil.cpp:354)
Fix:
PostOffload now detects pre-existing injected aliases in the projectList
(those carrying INPUT_FILE_COL_METADATA, placed there by PreOffload).
For each such alias Alias(InputFileName(), schemaName, meta)(exprId=X):
1. Creates a fresh AttributeReference(schemaName, ..., meta)(exprId=Z)
for the
scan output, preserving the (possibly mangled) schemaName.
2. Rewrites the alias to Alias(AttrRef(Z), schemaName, meta)(exprId=X),
keeping the original exprId=X so outer ProjectExec attr references
remain valid.
3. Pushes AttrRef(Z) into the scan output.
This exactly mirrors what the original rewriteExpr path does for the
non-collision
case, but uses the PreOffload-chosen (possibly mangled) schema name
instead of the
un-mangled prettyName.
For the non-collision case (no physical column named 'input_file_name'),
the scan
output gets a Synthesized attr named 'input_file_name' -- same as before.
For the collision case (physical 'input_file_name' column exists), the scan
output
gets a Synthesized attr named '__gluten_input_file_col__input_file_name__'
-- the
mangled name from PreOffload -- which does not conflict with the Regular
column.
Tests:
- Adds 'case-sensitive mode: lowercase input_file_name as data column --
platform
compatibility' (mirrors the IcebergSuite CI failing test for Parquet)
- Adds caseSensitive=false variant
- Adds plain data-only select variant
- All 133 tests in InputFileNameCollisionSuite + FallbackSuite +
ScalarFunctionsValidateSuite pass
- Resolves merge conflict in FallbackSuite.scala (stash remnant)
* Delete
backends-velox/src/test/scala/org/apache/gluten/execution/InputFileNameCollisionSuite.scala
* changes to casesensitive bug
* changes to casesensitive bug
* changes to casesensitive bug
* Address review: drop name mangling and backend API changes, prune tests
- Remove injected-file column name mangling, canonical-name metadata and
the injectedFileColAliases plumbing through IteratorApi/FileMetadataUtil.
- Fall back via a single fallback tag when the scan output conflicts with
the injected input_file attribute under normalized names.
- Prune FallbackSuite/IcebergSuite to the Input_File_Name vs
input_file_name() regression tests.
* Do not fill a real input_file_* data column with file metadata
A table column named input_file_name (or input_file_block_*) was matched
by name in BasicScanExecTransformer and requested as split metadata, so
native scans returned the file path instead of the stored value.
* Only skip input_file_* metadata for columns the scan actually reads
Checking the full table schema also suppressed the injected attribute when
the table had an unselected column named input_file_name, so
input_file_name() returned that column's data instead of the file path.
* Address review: route input_file_name collision fallback through
doValidateInternal
Instead of calling FallbackTags.add() directly on the vanilla scan in
PushDownInputFileExpression.PreOffload, place a lightweight TreeNodeTag
(INPUT_FILE_NAME_COLLISION_TAG) on the scan node when an injected
input-file attribute would share a normalized name with a user data
column.
ScanTransformerFactory.createFileSourceScanTransformer now calls
copyTagsFrom so the tag propagates to the temporary transformer created
by FallbackByNativeValidation during the AddFallbackTags pass.
BasicScanExecTransformer.doValidateInternal checks for the tag and
returns ValidationResult.failed(...), which causes Gluten's standard
validation framework to add FallbackTags to the vanilla scan and skip
offloading it -- keeping the scan on vanilla Spark where
InputFileName() is populated correctly via the JVM thread-local.
This satisfies the reviewer request to handle the backend-schema
limitation inside the scan's validation rather than inside
PushDownInputFileExpression.
Also resolve docs/velox-backend-limitations.md conflict by adopting the
upstream apache:main wording for the case-sensitive section.
* prune FallbackSuite: keep only the two focused input_file_name regression
tests
Remove the redundant tests:
- 'input_file_name() returns the path when the table has a column of that
name'
(sub-cases overlap with the primary collision test and the no-injection
test)
- 'input_file_name() without a conflicting column stays native'
(already covered implicitly by the caseSensitive=true branch of the main
test)
The two remaining tests cover the exact regression scenarios addressed by
this PR:
1. 'input_file_name() with a user column named Input_File_Name': verifies
caseSensitive=true → scan stays native, caseSensitive=false → scan falls
back, and in both cases user data and file-path values are correct.
2. 'user column named like input_file_name is not replaced by the file
path':
verifies that selecting only the user column (no function call) always
stays native and returns user data under both case-sensitivity settings.
* Propagate scan tags when creating batch scan transformers
PushDownInputFileExpression tags a scan with INPUT_FILE_NAME_COLLISION_TAG
when an injected input_file_name attribute collides with a user column, so
that doValidateInternal can fail validation and keep the scan on vanilla
Spark. createBatchScanTransformer and IcebergScanTransformer.apply did not
copy tags from the BatchScanExec, so the tag was lost and Velox failed with
'Cannot map from same table column to different outputs in table scan'.
* Revert "Propagate scan tags when creating batch scan transformers"
This reverts commit e1d3babfe7dcad2585aa4093e86b0cc103c384d2.
* Remove failing iceberg input_file_name regression test
* Restore copyTagsFrom in IcebergScanTransformer and BatchScanTransformer
PushDownInputFileExpression.PreOffload tags the source BatchScanExec with
INPUT_FILE_NAME_COLLISION_TAG when an injected input_file_name attribute
collides with a user data column. doValidateInternal reads this tag to fail
validation cleanly and keep the scan on vanilla Spark.
The tag must be copied when the BatchScanExec is converted to a transformer:
- IcebergScanTransformer.apply: copyTagsFrom was correctly added in
e1d3babfe but accidentally dropped by the revert in 1c5428f7a.
- ScanTransformerFactory.createBatchScanTransformer: same gap for all
non-Iceberg DSv2 sources (Delta, Lance, etc.).
- ScanTransformerFactory.createFileSourceScanTransformer already had
copyTagsFrom from the original fix; no change needed there.
Without this, the collision tag is silently lost and Velox receives a
scan with two outputs mapping to the same normalized column name, causing
a hard native error instead of a clean fallback.
* changes to casesensitive bug
---------
Co-authored-by: root <[email protected]>
---
.../backendsapi/clickhouse/CHIteratorApi.scala | 16 ++--
.../apache/gluten/execution/FallbackSuite.scala | 88 ++++++++++++++++++++++
.../gluten/execution/IcebergScanTransformer.scala | 21 ++++--
.../spark/source/GlutenIcebergSourceUtil.scala | 5 +-
.../execution/BasicScanExecTransformer.scala | 29 +++++--
.../execution/FileSourceScanExecTransformer.scala | 7 ++
.../gluten/execution/ScanTransformerFactory.scala | 8 +-
.../columnar/PushDownInputFileExpression.scala | 44 +++++++++--
8 files changed, 189 insertions(+), 29 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala
index 8a277cdee7..28a5e7a61c 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala
@@ -56,14 +56,20 @@ class CHIteratorApi extends IteratorApi with Logging with
LogLevelUtil {
private def getFileSchema(schema: StructType, names: Seq[String]):
StructType = {
val dataSchema = ArrayBuffer[StructField]()
+ // Use Spark's resolver for pairwise identifier equality.
+ // Under caseSensitive=true the resolver is exact equality; under false it
is
+ // case-insensitive. This is the correct contract for matching schema
field names
+ // to plan output attribute names -- it avoids the normalised-Map approach
which
+ // can silently discard one of two case-variant keys under
case-insensitive mode.
+ val resolver = org.apache.spark.sql.internal.SQLConf.get.resolver
schema.foreach {
field =>
- // case-insensitive schema matching
- val newField = names.find(_.equalsIgnoreCase(field.name)) match {
- case Some(name) => StructField(name, field.dataType, field.nullable,
field.metadata)
- case _ => field
+ val resolvedField = names.find(n => resolver(n, field.name)) match {
+ case Some(physicalName) =>
+ StructField(physicalName, field.dataType, field.nullable,
field.metadata)
+ case None => field
}
- dataSchema += newField
+ dataSchema += resolvedField
}
StructType(dataSchema.toSeq)
}
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala
index b9279e126f..e1daac4aa7 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala
@@ -19,6 +19,7 @@ package org.apache.gluten.execution
import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
import org.apache.spark.SparkConf
+import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec,
ColumnarShuffleExchangeExec, SortExec, SparkPlan}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
AQEShuffleReadExec}
import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
@@ -567,4 +568,91 @@ class FallbackSuite extends
VeloxWholeStageTransformerSuite with AdaptiveSparkPl
"FROM tmp4_wide JOIN tmp5_wide ON tmp4_wide.c1 = tmp5_wide.c1")
)
}
+
+ private def hasNativeScan(df: DataFrame): Boolean =
+
find(df.queryExecution.executedPlan)(_.isInstanceOf[FileSourceScanExecTransformer]).nonEmpty
+
+ // Regression tests for the input_file_name() / user-column collision fix.
+ //
+ // A table with a column named `Input_File_Name` creates a naming conflict
with the
+ // input_file_name() metadata function. Under caseSensitive=false the two
names normalise to
+ // the same string; Velox cannot produce both simultaneously so the scan
must fall back.
+ // Under caseSensitive=true the names are distinct and the scan stays native.
+ test("input_file_name() with a user column named Input_File_Name") {
+ withTempPath {
+ dir =>
+ import testImplicits._
+ Seq((1, "user-a"), (2, "user-b"))
+ .toDF("id", "Input_File_Name")
+ .write
+ .parquet(dir.getCanonicalPath)
+ withTempView("t") {
+ spark.read.parquet(dir.getCanonicalPath).createOrReplaceTempView("t")
+
+ def checkValues(df: DataFrame): Unit = {
+ val rows = df.collect().sortBy(_.getInt(0))
+ // User column must contain the original data values, never a file
path.
+ assert(rows.map(_.getString(1)).toSeq == Seq("user-a", "user-b"))
+ // input_file_name() must return the actual file path.
+ assert(rows.forall(r => r.getString(2).endsWith(".parquet")))
+ }
+
+ // caseSensitive=true: "Input_File_Name" != "input_file_name" --
scan stays native.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ runQueryAndCompare(
+ "SELECT id, Input_File_Name, input_file_name() AS fname FROM t",
+ noFallBack = false) {
+ df =>
+ checkValues(df)
+ assert(hasNativeScan(df))
+ }
+ }
+
+ // caseSensitive=false: names collide -- scan falls back, results
still correct.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ runQueryAndCompare(
+ "SELECT id, Input_File_Name, input_file_name() AS fname FROM t",
+ noFallBack = false) {
+ df =>
+ checkValues(df)
+ assert(!hasNativeScan(df))
+ }
+ }
+ }
+ }
+ }
+
+ // Selecting only the user column (no input_file_name() call) must never
trigger a fallback
+ // and must return the original data values regardless of case-sensitivity
setting.
+ test("user column named like input_file_name is not replaced by the file
path") {
+ Seq("input_file_name", "Input_File_Name").foreach {
+ colName =>
+ Seq("true", "false").foreach {
+ caseSensitive =>
+ withTempPath {
+ dir =>
+ import testImplicits._
+ Seq((1, "val1"), (2, "val2"))
+ .toDF("id", colName)
+ .write
+ .parquet(dir.getCanonicalPath)
+ withTempView("t") {
+
spark.read.parquet(dir.getCanonicalPath).createOrReplaceTempView("t")
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive) {
+ runQueryAndCompare(
+ s"SELECT id, `$colName` FROM t ORDER BY id",
+ noFallBack = false) {
+ df =>
+ assert(
+ df.collect().map(_.getString(1)).toSeq ==
Seq("val1", "val2"),
+ s"col=$colName caseSensitive=$caseSensitive\n" +
+ df.queryExecution.executedPlan)
+ assert(hasNativeScan(df),
df.queryExecution.executedPlan.toString)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
}
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
index 535c97a1c7..a96706bf4b 100644
---
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
@@ -20,6 +20,7 @@ import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.exception.GlutenNotSupportException
import
org.apache.gluten.execution.IcebergScanTransformer.{containsMetadataColumn,
containsUuidOrFixedType}
+import org.apache.gluten.expression.ConverterUtils
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.{LocalFilesNode, SplitInfo}
import org.apache.gluten.substrait.rel.LocalFilesNode.ColumnMappingMode
@@ -44,7 +45,6 @@ import org.apache.iceberg.types.Type.TypeID
import org.apache.iceberg.types.Types.{ListType, MapType, NestedField}
import java.util.{HashMap => JHashMap}
-import java.util.Locale
case class IcebergScanTransformer(
override val output: Seq[AttributeReference],
@@ -121,7 +121,7 @@ case class IcebergScanTransformer(
val hasUnsupportedMetadata = scan.readSchema().fieldNames.exists {
f =>
MetadataColumns.isMetadataColumn(f) &&
- !allowedMetadataColumns.contains(f.toLowerCase(Locale.ROOT))
+ !allowedMetadataColumns.contains(ConverterUtils.normalizeColName(f))
}
if (hasUnsupportedMetadata) {
return ValidationResult.failed("Read unsupported metadata column")
@@ -207,18 +207,23 @@ case class IcebergScanTransformer(
override def getRootPathsInternal: Seq[String] = fileFormatAndRootPaths._2
private lazy val readSchemaFields =
- scan.readSchema().fieldNames.map(_.toLowerCase(Locale.ROOT)).toSet
+ scan.readSchema().fieldNames.map(ConverterUtils.normalizeColName).toSet
private lazy val inputFileRelatedMetadataColumns = output.filter {
attr =>
- val name = attr.name.toLowerCase(Locale.ROOT)
+ val name = ConverterUtils.normalizeColName(attr.name)
IcebergScanTransformer.InputFileRelatedMetadataColumnNames.contains(name) &&
!readSchemaFields.contains(name)
}
override def getMetadataColumns(): Seq[AttributeReference] = {
val extraMetadataColumns = inputFileRelatedMetadataColumns.filterNot {
- metadataAttr =>
metadataColumns.exists(_.name.equalsIgnoreCase(metadataAttr.name))
+ metadataAttr =>
+ metadataColumns.exists {
+ existingAttr =>
+ ConverterUtils.normalizeColName(existingAttr.name) ==
+ ConverterUtils.normalizeColName(metadataAttr.name)
+ }
}
metadataColumns ++ extraMetadataColumns
}
@@ -362,7 +367,7 @@ object IcebergScanTransformer {
Set("input_file_name", "input_file_block_start", "input_file_block_length")
def apply(batchScan: BatchScanExec): IcebergScanTransformer = {
- new IcebergScanTransformer(
+ val transformer = new IcebergScanTransformer(
batchScan.output.map(a =>
a.withName(AvroSchemaUtil.makeCompatibleName(a.name))),
batchScan.scan,
batchScan.runtimeFilters,
@@ -370,6 +375,10 @@ object IcebergScanTransformer {
keyGroupedPartitioning = batchScan.keyGroupedPartitioning,
commonPartitionValues =
SparkShimLoader.getSparkShims.getCommonPartitionValues(batchScan)
)
+ // Carry over tags (e.g.
PushDownInputFileExpression.INPUT_FILE_NAME_COLLISION_TAG) so
+ // that doValidateInternal can read them, as ScanTransformerFactory does
for FileSource scans.
+ transformer.copyTagsFrom(batchScan)
+ transformer
}
def supportsBatchScan(scan: Scan): Boolean = {
diff --git
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
index 20364bc85c..d460f674df 100644
---
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
@@ -20,6 +20,7 @@ import org.apache.gluten.IcebergDefaultValueUtil
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.exception.GlutenNotSupportException
import org.apache.gluten.execution.SparkDataSourceRDDPartition
+import org.apache.gluten.expression.ConverterUtils
import org.apache.gluten.substrait.rel.{IcebergLocalFilesBuilder, SplitInfo}
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
@@ -36,7 +37,6 @@ import org.apache.iceberg.util.PropertyUtil
import java.lang.{Long => JLong}
import java.util.{ArrayList => JArrayList, HashMap => JHashMap, List => JList,
Map => JMap}
-import java.util.Locale
import scala.collection.JavaConverters._
@@ -177,7 +177,8 @@ object GlutenIcebergSourceUtil {
val metadataColumns = new JHashMap[String, String]()
metadataColumnNames.foreach {
name =>
- name.toLowerCase(Locale.ROOT) match {
+ val normalizedName = ConverterUtils.normalizeColName(name)
+ normalizedName match {
case InputFileNameCol => metadataColumns.put(name, filePath)
case InputFileBlockStartCol => metadataColumns.put(name,
start.toString)
case InputFileBlockLengthCol => metadataColumns.put(name,
length.toString)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
index b4a7fc0159..c65fe7d135 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
@@ -19,6 +19,7 @@ package org.apache.gluten.execution
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.expression.{ConverterUtils, ExpressionConverter}
+import org.apache.gluten.extension.columnar.PushDownInputFileExpression
import org.apache.gluten.substrait.`type`.ColumnTypeNode
import org.apache.gluten.substrait.SubstraitContext
import org.apache.gluten.substrait.extensions.ExtensionBuilder
@@ -115,6 +116,15 @@ trait BasicScanExecTransformer extends
LeafTransformSupport with BaseDataSource
def getInputStats: Option[InputStats] = Option.empty
+ protected def neededInputFileRelatedMetadataKeys: Seq[String] = {
+ val inputFileRelatedMetadataKeys = Seq(
+ InputFileName().prettyName,
+ InputFileBlockStart().prettyName,
+ InputFileBlockLength().prettyName)
+
+ inputFileRelatedMetadataKeys.filter(k => output.exists(_.name == k))
+ }
+
override def getSplitInfos: Seq[SplitInfo] = {
getSplitInfosFromPartitions(getPartitionWithReadFileFormats)
}
@@ -135,14 +145,6 @@ trait BasicScanExecTransformer extends
LeafTransformSupport with BaseDataSource
val metadataFromSpark = getMetadataColumns().map(_.name)
- val inputFileRelatedMetadataKeys = Seq(
- InputFileName().prettyName,
- InputFileBlockStart().prettyName,
- InputFileBlockLength().prettyName)
-
- val neededInputFileRelatedMetadataKeys =
- inputFileRelatedMetadataKeys.filter(k => output.exists(_.name == k))
-
val metadataColumnNames = (metadataFromSpark ++
neededInputFileRelatedMetadataKeys).distinct
BackendsApiManager.getIteratorApiInstance
@@ -157,6 +159,17 @@ trait BasicScanExecTransformer extends
LeafTransformSupport with BaseDataSource
}
override protected def doValidateInternal(): ValidationResult = {
+ // A collision tag is placed by PushDownInputFileExpression.PreOffload
when an injected
+ // input-file attribute would share a normalized name with a user column
in this scan.
+ // Velox cannot simultaneously produce two columns with the same
normalized name, so we
+ // fail validation here to keep the scan on vanilla Spark where the
file-path thread-local
+ // is set by the JVM scan iterator.
+ if
(getTagValue(PushDownInputFileExpression.INPUT_FILE_NAME_COLLISION_TAG).getOrElse(false))
{
+ return ValidationResult.failed(
+ "input_file_name() column name collision with a user data column under
case-insensitive " +
+ "resolution: native scan cannot produce both simultaneously")
+ }
+
val validationResult = BackendsApiManager.getSettings
.validateScanExec(
fileFormat,
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
index 53e7b0ba2f..4e587d3528 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala
@@ -145,6 +145,13 @@ abstract class FileSourceScanExecTransformerBase(
override def getMetadataColumns(): Seq[AttributeReference] = metadataColumns
+ // An output named like an input file function is the injected attribute
unless the scan reads
+ // a real column of that name, in which case it is user data and must not
get the file path.
+ override protected def neededInputFileRelatedMetadataKeys: Seq[String] = {
+ val readColumnNames = requiredSchema.fieldNames.toSet
+
super.neededInputFileRelatedMetadataKeys.filterNot(readColumnNames.contains)
+ }
+
override def getPartitions: Seq[Partition] = {
if (SparkVersionUtil.gteSpark40) {
getPartitionsSeq()
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
index 15a23130ce..e528f246bb 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
@@ -25,7 +25,7 @@ object ScanTransformerFactory {
def createFileSourceScanTransformer(
scanExec: FileSourceScanExec): FileSourceScanExecTransformerBase = {
- FileSourceScanExecTransformer(
+ val transformer = FileSourceScanExecTransformer(
scanExec.relation,
SparkShimLoader.getSparkShims.getFileSourceScanStream(scanExec),
scanExec.output,
@@ -37,15 +37,19 @@ object ScanTransformerFactory {
scanExec.tableIdentifier,
scanExec.disableBucketedScan
)
+ transformer.copyTagsFrom(scanExec)
+ transformer
}
def createBatchScanTransformer(batchScanExec: BatchScanExec):
BatchScanExecTransformerBase = {
- BatchScanExecTransformer(
+ val transformer = BatchScanExecTransformer(
batchScanExec.output,
batchScanExec.scan,
batchScanExec.runtimeFilters,
keyGroupedPartitioning = batchScanExec.keyGroupedPartitioning,
table = batchScanExec.table
)
+ transformer.copyTagsFrom(batchScanExec)
+ transformer
}
}
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 d50c10e7b4..5ad5b9fcc4 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
@@ -17,15 +17,16 @@
package org.apache.gluten.extension.columnar
import org.apache.gluten.execution.{BatchScanExecTransformerBase,
FileSourceScanExecTransformer, ProjectExecTransformer}
+import org.apache.gluten.expression.ConverterUtils
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.catalyst.trees.TreeNodeTag
import org.apache.spark.sql.execution.{DeserializeToObjectExec,
FileSourceScanExec, FilterExec, 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
+import org.apache.spark.sql.internal.SQLConf
import scala.collection.mutable
@@ -44,9 +45,29 @@ import scala.collection.mutable
*/
object PushDownInputFileExpression {
+ /**
+ * Tag placed on a scan node (before offloading) when an injected input-file
attribute would share
+ * a normalized name with a user data column in the scan output. The native
scan cannot
+ * simultaneously produce two columns with the same normalized name, so the
scan must fall back to
+ * vanilla Spark (where the file-path thread-local is correctly populated by
the JVM iterator).
+ *
+ * The tag survives the scan's conversion to a transformer (via
+ * [[ScanTransformerFactory.createFileSourceScanTransformer]] calling
copyTagsFrom), allowing
+ * [[BasicScanExecTransformer.doValidateInternal]] to detect the conflict
and fail validation.
+ */
+ val INPUT_FILE_NAME_COLLISION_TAG: TreeNodeTag[Boolean] =
+ TreeNodeTag[Boolean]("org.apache.gluten.InputFileNameCollisionTag")
+
private val INPUT_FILE_ATTR_NAMES =
Set("input_file_name", "input_file_block_start", "input_file_block_length")
+ /**
+ * Whether `name` is the name of an attribute injected for an input file
function, following
+ * Spark's configured (case-sensitive or case-insensitive) resolution.
+ */
+ def isInputFileAttrName(name: String): Boolean =
+ INPUT_FILE_ATTR_NAMES.exists(SQLConf.get.resolver(name, _))
+
def containsInputFileFunctionExpr(expr: Expression): Boolean = {
expr match {
case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength
=> true
@@ -57,8 +78,7 @@ object PushDownInputFileExpression {
def containsInputFileRelatedExpr(expr: Expression): Boolean = {
expr match {
case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength
=> true
- case a: AttributeReference =>
- INPUT_FILE_ATTR_NAMES.contains(a.name.toLowerCase(Locale.ROOT))
+ case a: AttributeReference => isInputFileAttrName(a.name)
case _ => expr.children.exists(containsInputFileRelatedExpr)
}
}
@@ -113,6 +133,16 @@ object PushDownInputFileExpression {
// to prevent offloading when input_file expressions are present
addFallbackTag(ProjectExec(p.output ++ replacedExprs.values, p))
case p: LeafExecNode if shouldAddInputFileExpr(p) =>
+ // Tag the scan when the injected attribute name would collide with
a user data column
+ // under the current case-sensitivity setting. The native scan
cannot map two columns
+ // to the same normalized name;
BasicScanExecTransformer.doValidateInternal checks this
+ // tag and fails validation, which causes the scan to remain on
vanilla Spark where the
+ // file-path thread-local is set by the JVM iterator.
+ val scanNames = p.output.map(a =>
ConverterUtils.normalizeColName(a.name)).toSet
+ val injectedNames = replacedExprs.values.map(a =>
ConverterUtils.normalizeColName(a.name))
+ if (injectedNames.exists(scanNames)) {
+ p.setTagValue(INPUT_FILE_NAME_COLLISION_TAG, true)
+ }
addFallbackTag(ProjectExec(p.output ++ replacedExprs.values, p))
case p: LeafExecNode =>
p
@@ -175,10 +205,12 @@ object PushDownInputFileExpression {
val newProjectList = projectList.map {
expr => rewriteExpr(expr,
replacedExprs).asInstanceOf[NamedExpression]
}
- val existingNames =
child.output.map(_.name.toLowerCase(Locale.ROOT)).toSet
+ // Dedup by exprId rather than by name, so that a user column whose
name differs from the
+ // injected attribute only by case is not mistaken for it.
+ val existingExprIds = child.output.map(_.exprId).toSet
val inputFileAttrs = replacedExprs.values.toSeq
.map(_.toAttribute.asInstanceOf[AttributeReference])
- .filterNot(attr =>
existingNames.contains(attr.name.toLowerCase(Locale.ROOT)))
+ .filterNot(attr => existingExprIds.contains(attr.exprId))
p.copy(
projectList = newProjectList,
child = child.withOutput(child.output ++ inputFileAttrs))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
