Copilot commented on code in PR #12884:
URL: https://github.com/apache/gluten/pull/12884#discussion_r3895344030
##########
cpp/velox/compute/VeloxPlanConverter.cc:
##########
@@ -53,6 +53,10 @@ VeloxPlanConverter::VeloxPlanConverter(
}
namespace {
+// Keep this key in sync with the JVM-side constant
+// LocalFilesNode.COLUMN_MAPPING_MODE_METADATA_KEY.
+constexpr std::string_view kColumnMappingModeMetadataKey =
"__gluten.column_mapping_mode";
Review Comment:
`kColumnMappingModeMetadataKey` is a `std::string_view` but
`otherMetadataColumn.key()` is a `std::string` (protobuf). If this project is
compiled as C++17 (common for Velox), `std::string == std::string_view` is not
guaranteed to compile. Use a `constexpr const char*` for the key, or compare
via `std::string_view(otherMetadataColumn.key()) ==
kColumnMappingModeMetadataKey` to keep this C++17-safe.
##########
gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala:
##########
@@ -142,11 +144,63 @@ case class DeltaScanTransformer(
override def getSplitInfosFromPartitions(
partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = {
val splitInfos = super.getSplitInfosFromPartitions(partitions)
- // Deletion vectors only exist on Delta tables read through a
TahoeFileIndex (which also covers
- // PreparedDeltaFileIndex). Its `path` is the authoritative table root and
is used to resolve
- // per-file DV locations. Any other location cannot carry Delta DV
metadata, so the generic
- // split representation is returned unchanged.
+ // Keep Delta's split decoration narrow. The generic Parquet path has
already attached the
+ // session-derived split mapping mode and only attaches file schema when
position mapping
+ // needs it. Delta name column mapping is the one case that must force
name mapping regardless
+ // of the generic Parquet setting because Gluten rewrites the scan schema
to physical names.
+ splitInfos.foreach {
+ case localFiles: LocalFilesNode =>
+ deltaColumnMappingMode.foreach {
+ mode =>
+ localFiles.clearFileSchema()
+ localFiles.setColumnMappingMode(mode)
+ }
+ case _ =>
+ }
Review Comment:
`deltaColumnMappingMode` is computed for every element in `splitInfos`.
Since it depends only on `relation.fileFormat`, compute it once (e.g., `val
modeOpt = deltaColumnMappingMode`) and apply it inside the loop. This avoids
repeated pattern matching per split and makes the intent clearer.
##########
gluten-delta/src/main/scala/org/apache/gluten/delta/DeltaAddFileLookup.scala:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.delta
+
+import org.apache.spark.sql.delta.actions.AddFile
+import org.apache.spark.sql.delta.util.DeltaFileOperations
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+
+import org.apache.hadoop.fs.Path
+
+import scala.collection.mutable
+
+/** Driver-side AddFile index shared by all FilePartitions in one prepared
Delta scan. */
+final private[gluten] class DeltaAddFileLookup private (
+ addFiles: IndexedSeq[AddFile],
+ absolutePaths: IndexedSeq[Path],
+ candidateIndexes: Map[(Option[String], String), Vector[Int]],
+ val hasDeletionVector: Boolean) {
+
+ def find(file: PartitionedFile): AddFile = {
+ val partitionedFilePath = new Path(file.filePath.toString)
+ val candidates = mutable.BitSet.empty
+ DeltaAddFileLookup.pathVariants(partitionedFilePath).foreach {
+ key => candidateIndexes.get(key).foreach(indexes => candidates ++=
indexes)
+ }
+
+ // BitSet iteration preserves the original AddFile order used by the
previous Seq.find lookup.
+ candidates.iterator
+ .find(index => DeltaAddFileLookup.samePath(partitionedFilePath,
absolutePaths(index)))
+ .map(addFiles.apply)
+ .getOrElse {
+ throw new IllegalStateException(
+ s"Unable to find Delta AddFile metadata for split ${file.filePath}")
+ }
+ }
+}
+
+private[gluten] object DeltaAddFileLookup {
+ val empty: DeltaAddFileLookup =
+ new DeltaAddFileLookup(Vector.empty, Vector.empty, Map.empty,
hasDeletionVector = false)
+
+ def apply(
+ tablePath: Path,
+ addFiles: Seq[AddFile],
+ hasDeletionVector: Boolean): DeltaAddFileLookup = {
+ val indexedAddFiles = addFiles.toIndexedSeq
+ val absolutePaths = indexedAddFiles.map {
+ addFile => DeltaFileOperations.absolutePath(tablePath.toString,
addFile.path)
+ }
+ val candidateIndexes = mutable.HashMap.empty[
+ (Option[String], String),
+ mutable.ArrayBuffer[Int]]
+
+ absolutePaths.zipWithIndex.foreach {
+ case (absolutePath, index) =>
+ pathVariants(absolutePath).foreach {
+ key => candidateIndexes.getOrElseUpdate(key,
mutable.ArrayBuffer.empty) += index
+ }
+ }
+
+ new DeltaAddFileLookup(
+ indexedAddFiles,
+ absolutePaths,
+ candidateIndexes.iterator.map { case (key, indexes) => key ->
indexes.toVector }.toMap,
+ hasDeletionVector)
+ }
+
+ private def samePath(left: Path, right: Path): Boolean = {
+ pathVariants(left).intersect(pathVariants(right)).nonEmpty
+ }
+
+ private def pathVariants(path: Path): Set[(Option[String], String)] = {
+ val uri = path.toUri.normalize()
+ val authority = Option(uri.getAuthority)
+ Seq(uri.getRawPath, uri.getPath)
+ .filter(_ != null)
+ .flatMap(percentVariants)
+ .map(pathValue => authority -> pathValue)
+ .toSet
+ }
+
+ // SparkPath and DeltaFileOperations can expose literal '%' characters at
different URI escaping
+ // levels. Compare a bounded set of full-path variants while retaining the
URI authority.
+ private def percentVariants(path: String): Set[String] = {
+ (0 until 4).foldLeft(Set(path)) {
+ (variants, _) => variants ++ variants.map(_.replace("%25", "%"))
+ }
Review Comment:
The `(0 until 4)` iteration count is a magic number and makes it unclear why
“4” is sufficient. Consider making this either (a) a named constant with a
brief rationale, or (b) a loop that iterates until the set stabilizes
(optionally with a max-iteration cap) to keep behavior robust while still
bounded.
##########
cpp/velox/compute/VeloxPlanConverter.cc:
##########
@@ -164,6 +177,15 @@ std::shared_ptr<SplitInfo> parseScanSplitInfo(
metadataColumnMap[metadataColumn.key()] = metadataColumn.value();
}
for (const auto& otherMetadataColumn :
file.other_const_metadata_columns()) {
+ if (otherMetadataColumn.key() == kColumnMappingModeMetadataKey) {
Review Comment:
`kColumnMappingModeMetadataKey` is a `std::string_view` but
`otherMetadataColumn.key()` is a `std::string` (protobuf). If this project is
compiled as C++17 (common for Velox), `std::string == std::string_view` is not
guaranteed to compile. Use a `constexpr const char*` for the key, or compare
via `std::string_view(otherMetadataColumn.key()) ==
kColumnMappingModeMetadataKey` to keep this C++17-safe.
##########
gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala:
##########
@@ -178,10 +185,22 @@ case class PaimonScanTransformer(
.asJava,
new JHashMap[String, String]()
)
+ localFiles.setFileSchema(getDataSchema)
+
paimonColumnMappingMode(fileFormat).foreach(localFiles.setColumnMappingMode)
+ localFiles
case _ => throw new GlutenNotSupportException()
}
}
+ private def paimonColumnMappingMode(fileFormat: ReadFileFormat):
Option[ColumnMappingMode] = {
+ fileFormat match {
+ case ReadFileFormat.ParquetReadFormat | ReadFileFormat.OrcReadFormat =>
+ Some(ColumnMappingMode.NAME)
+ case _ =>
+ None
+ }
+ }
Review Comment:
This mapping-mode selection logic is duplicated (very similarly) in
`IcebergScanTransformer`. If both are intended to enforce the same scan-level
semantics for table formats, consider extracting a small shared helper (e.g.,
in a common utility/object under `org.apache.gluten.execution` or
`substrait.rel`) to reduce drift risk when supported formats/modes change.
--
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]
