philo-he commented on code in PR #12726:
URL: https://github.com/apache/gluten/pull/12726#discussion_r4240463137
##########
backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala:
##########
@@ -56,12 +56,18 @@ 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 newField = names.find(n => resolver(n, field.name)) match {
Review Comment:
Perhaps, naming it to `resolvedField` for being more descriptive?
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/PushDownInputFileExpression.scala:
##########
@@ -68,52 +101,181 @@ object PushDownInputFileExpression {
plan
}
- private def rewriteExpr(expr: Expression, replacedExprs: mutable.Map[String,
Alias]): Expression =
+ /**
+ * Create a tagged Alias for an input-file expression.
+ *
+ * The alias carries INPUT_FILE_COL_METADATA (so makeColumnTypeNode emits
METADATA_COL) and, when
+ * the schema name must be mangled to avoid a case-insensitive collision
with a user column, also
+ * stores the canonical prettyName under GLUTEN_INPUT_FILE_CANON_KEY so that
+ * BasicScanExecTransformer can recover it for the Velox split infoColumns
map.
+ *
+ * @param child
+ * The input-file function expression.
+ * @param schemaName
+ * The name to give the alias in the scan output schema. Equals prettyName
when there is no
+ * collision; is a mangled private name when a caseSensitive=false
collision exists.
+ * @param canonicalName
+ * The Spark prettyName (e.g. "input_file_name"). Stored in metadata only
when it differs from
+ * schemaName.
+ */
+ private def makeInputFileAlias(
+ child: Expression,
+ schemaName: String,
+ canonicalName: String): Alias = {
+ val meta =
+ if (schemaName == canonicalName) {
+ INPUT_FILE_COL_METADATA
+ } else {
+ new MetadataBuilder()
+ .withMetadata(INPUT_FILE_COL_METADATA)
+ .putString(GLUTEN_INPUT_FILE_CANON_KEY, canonicalName)
+ .build()
+ }
+ Alias(child, schemaName)(explicitMetadata = Some(meta))
+ }
+
+ /**
+ * Rewrite input-file function expressions in `expr`, replacing each with an
AttributeReference to
+ * a fresh alias stored in `replacedExprs`.
+ *
+ * @param mangledPrettyNames
+ * Set of canonical prettyNames whose injected aliases must use a private
mangled schema name
+ * (because the user table has a column that normalises to the same
lowercase name under
+ * caseSensitive=false). An empty set means no mangling is needed.
+ */
+ private def rewriteExpr(
+ expr: Expression,
+ replacedExprs: mutable.Map[String, Alias],
+ mangledPrettyNames: Set[String] = Set.empty): Expression =
expr match {
case _: InputFileName =>
+ val pretty = expr.prettyName
replacedExprs
- .getOrElseUpdate(expr.prettyName, Alias(InputFileName(),
expr.prettyName)())
+ .getOrElseUpdate(
+ pretty,
+ makeInputFileAlias(
+ InputFileName(),
+ if (mangledPrettyNames.contains(pretty))
mangledSchemaName(pretty) else pretty,
+ pretty))
.toAttribute
case _: InputFileBlockStart =>
+ val pretty = expr.prettyName
replacedExprs
- .getOrElseUpdate(expr.prettyName, Alias(InputFileBlockStart(),
expr.prettyName)())
+ .getOrElseUpdate(
+ pretty,
+ makeInputFileAlias(
+ InputFileBlockStart(),
+ if (mangledPrettyNames.contains(pretty))
mangledSchemaName(pretty) else pretty,
+ pretty))
.toAttribute
case _: InputFileBlockLength =>
+ val pretty = expr.prettyName
replacedExprs
- .getOrElseUpdate(expr.prettyName, Alias(InputFileBlockLength(),
expr.prettyName)())
+ .getOrElseUpdate(
+ pretty,
+ makeInputFileAlias(
+ InputFileBlockLength(),
+ if (mangledPrettyNames.contains(pretty))
mangledSchemaName(pretty) else pretty,
+ pretty))
.toAttribute
case other =>
- other.withNewChildren(other.children.map(child => rewriteExpr(child,
replacedExprs)))
+ other.withNewChildren(
+ other.children.map(child => rewriteExpr(child, replacedExprs,
mangledPrettyNames)))
}
+ /**
+ * Returns the private schema name used for an injected input-file alias
when a
+ * caseSensitive=false name collision is detected. The mangled name starts
with the
+ * GLUTEN_INPUT_FILE_COL_ATTR_KEY prefix, guaranteeing it can never collide
with a real user
+ * column (real columns cannot be named with double-underscore
Gluten-internal identifiers).
+ */
+ private[columnar] def mangledSchemaName(prettyName: String): String =
+ s"$GLUTEN_INPUT_FILE_COL_ATTR_KEY${prettyName}__"
+
object PreOffload extends Rule[SparkPlan] {
override def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
case ProjectExec(projectList, child)
if projectList.exists(containsInputFileRelatedExpr) &&
hasInputFileRelatedSource(child) =>
+ val mangledNames = collidingInputFilePrettyNames(child)
val replacedExprs = mutable.Map[String, Alias]()
val newProjectList = projectList.map {
- expr => rewriteExpr(expr,
replacedExprs).asInstanceOf[NamedExpression]
+ expr => rewriteExpr(expr, replacedExprs,
mangledNames).asInstanceOf[NamedExpression]
}
val newChild = addMetadataCol(child, replacedExprs)
ProjectExec(newProjectList, newChild)
case f @ FilterExec(condition, child)
if containsInputFileRelatedExpr(condition) &&
hasInputFileRelatedSource(child) =>
+ val mangledNames = collidingInputFilePrettyNames(child)
val replacedExprs = mutable.Map[String, Alias]()
- val newCondition = rewriteExpr(condition, replacedExprs)
+ val newCondition = rewriteExpr(condition, replacedExprs, mangledNames)
val newChild = addMetadataCol(child, replacedExprs)
ProjectExec(f.output, FilterExec(newCondition, newChild))
}
+ /**
+ * Returns the set of input-file function prettyNames (e.g.
"input_file_name") whose injected
+ * aliases must be given mangled schema names because a user column in the
scan output
+ * normalises (via ConverterUtils.normalizeColName) to the same lowercase
string.
+ *
+ * Under caseSensitive=true, normalizeColName preserves case, so
"Input_File_Name" and
+ * "input_file_name" remain distinct -- no mangling needed, empty set
returned. However, a user
+ * column whose name is an exact lowercase match to the function
prettyName (e.g.
+ * "input_file_name") will still be detected as a collision even under
caseSensitive=true,
+ * because both names normalise to the same string. In that case the
injected alias is mangled
+ * AND the scan is fallback-tagged (in addMetadataCol) so the JVM scan
sets the InputFileName
+ * thread-local.
+ *
+ * Under caseSensitive=false, normalizeColName lowercases all names. A
user column
+ * "Input_File_Name" lowercases to "input_file_name", colliding with the
injected alias of the
+ * same name in the Velox NamedStruct schema. In this case we return
{"input_file_name"} so that
+ * the alias is given the private name
mangledSchemaName("input_file_name") instead, avoiding
+ * the Velox "Cannot map from same table column to different outputs"
error.
Review Comment:
Could you open a GitHub issue to document the detailed background and
rationale? We could then shorten this code comment to the key points and
reference the issue for further details.
Ditto for other long comments if applicable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
