Copilot commented on code in PR #12884:
URL: https://github.com/apache/gluten/pull/12884#discussion_r3893867507
##########
gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala:
##########
@@ -142,11 +144,43 @@ 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 _ =>
+ }
+ // PreparedDeltaFileIndex contains the exact AddFiles selected for this
scan. Use these as the
+ // source of truth because PartitionedFile metadata can retain an older DV
descriptor after
+ // repeated DML updates the same data file.
relation.location match {
+ case prepared: PreparedDeltaFileIndex =>
+ val tableRootPath = prepared.path
+ val addFiles = prepared.preparedScan.files
+ splitInfos.zip(partitions).map {
+ case (localFiles: LocalFilesNode, (filePartition: FilePartition, _))
=>
+ DeltaDeletionVectorScanInfo
+ .normalizeFromAddFiles(filePartition.files.toSeq, tableRootPath,
addFiles)
+ .map {
+ case (otherMetadataColumns, deltaReadOptions) =>
+ DeltaLocalFilesBuilder.makeDeltaLocalFiles(
+ localFiles,
+ otherMetadataColumns.asJava,
+ deltaReadOptions.asJava): SplitInfo
+ }
+ .getOrElse(localFiles)
+ case (splitInfo, _) => splitInfo
+ }
Review Comment:
`normalizeFromAddFiles` is invoked once per `FilePartition` while being
passed the full `addFiles` list each time. Inside `normalizeFromAddFiles`, each
`PartitionedFile` does a linear `.find` over `addFiles`, which can devolve into
O(totalFiles^2) work for large scans and many partitions. Consider pre-indexing
`addFiles` once (e.g., a Map from a canonicalized path key to `AddFile`)
outside the `splitInfos.zip(partitions)` loop and changing
`normalizeFromAddFiles` to use that indexed structure (or introducing a new
overload that takes the precomputed index).
##########
gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala:
##########
@@ -132,6 +171,65 @@ object DeltaDeletionVectorScanInfo {
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
+ private def extract(
+ file: PartitionedFile,
+ hadoopConf: Configuration,
+ tablePath: Path,
+ addFile: AddFile): PartitionFileScanInfo = {
+ val metadata = otherMetadataColumns(file)
+ val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded,
RowIndexFilterTypeKey)
+ val dvInfo = Option(addFile.deletionVector) match {
+ case Some(descriptor) =>
+ DeletionVectorInfo(
+ true,
+ IF_CONTAINED,
+ descriptor.cardinality,
+ serializePayload(hadoopConf, tablePath, descriptor))
+ case None =>
+ DeletionVectorInfo(false, KEEP_ALL, 0L, Array.emptyByteArray)
+ }
+ PartitionFileScanInfo(normalizedMetadata, dvInfo)
+ }
+
+ private def findAddFile(
+ file: PartitionedFile,
+ tablePath: Path,
+ addFiles: Seq[AddFile]): AddFile = {
+ val partitionedFilePath = new Path(file.filePath.toString)
+ addFiles
+ .find {
+ addFile =>
+ val addFilePath =
DeltaFileOperations.absolutePath(tablePath.toString, addFile.path)
+ samePath(partitionedFilePath, addFilePath)
+ }
+ .getOrElse {
+ throw new IllegalStateException(
+ s"Unable to find Delta AddFile metadata for split ${file.filePath}")
+ }
+ }
Review Comment:
`findAddFile` throws if path matching fails, which can fail an entire query
at runtime even if the scan could potentially proceed using split-provided
metadata. A more resilient approach would be to (a) avoid per-file linear
search by building an indexed lookup (see perf note), and (b) consider a safe
fallback when the AddFile cannot be matched (e.g., return `None` from
`normalizeFromAddFiles` for that partition and let the existing
`normalize(...)` path handle it, or preserve original metadata rather than
throwing).
##########
cpp/velox/compute/VeloxPlanConverter.cc:
##########
@@ -53,6 +53,8 @@ VeloxPlanConverter::VeloxPlanConverter(
}
namespace {
Review Comment:
The metadata key string is duplicated across JVM
(`LocalFilesNode.COLUMN_MAPPING_MODE_METADATA_KEY`) and C++
(`kColumnMappingModeMetadataKey`). To reduce drift risk, consider centralizing
this in a single shared definition (e.g., a generated constant from a Substrait
extension/proto, or at least a shared C++ header + JVM-side reference in one
place with a strong comment/test ensuring literals remain identical).
##########
gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java:
##########
@@ -130,6 +152,14 @@ public void setFileSchema(StructType schema) {
this.fileSchema = schema;
}
+ public void clearFileSchema() {
+ this.fileSchema = null;
+ }
Review Comment:
`clearFileSchema()` introduces a `null` state for `fileSchema`. Since
DeltaScanTransformer now calls `clearFileSchema()` before split serialization,
`toProtobuf()` / any code that builds schema-related protobuf fields must
tolerate `fileSchema == null` to avoid runtime NPEs. If current serialization
assumes a non-null schema, prefer representing “no schema” explicitly (e.g.,
conditionally omitting the schema field during protobuf conversion, or using an
empty `StructType`) and document the expected invariants for callers.
--
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]
