Copilot commented on code in PR #12884:
URL: https://github.com/apache/gluten/pull/12884#discussion_r3893974591
##########
gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala:
##########
@@ -132,6 +172,75 @@ object DeltaDeletionVectorScanInfo {
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
+ private def extract(
+ file: PartitionedFile,
+ hadoopConf: Configuration,
+ serializableHadoopConf: SerializableConfiguration,
+ 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,
+ deletionVectorPayload(
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ descriptor,
+ None))
+ case None =>
+ DeletionVectorInfo(
+ false,
+ KEEP_ALL,
+ 0L,
+ new InMemoryDeletionVectorPayload(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:
In `normalizeFromAddFiles`, `findAddFile` performs a linear search over
`addFiles` for every `PartitionedFile`, making the overall complexity O(N*M).
This can become a hotspot for large scans. Consider pre-indexing `addFiles`
once (e.g., build a `Map` keyed by normalized path variants) and doing O(1)
lookups per split; apply the same improvement to the delta33 variant as well.
##########
gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala:
##########
@@ -132,6 +172,75 @@ object DeltaDeletionVectorScanInfo {
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
+ private def extract(
+ file: PartitionedFile,
+ hadoopConf: Configuration,
+ serializableHadoopConf: SerializableConfiguration,
+ 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,
+ deletionVectorPayload(
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ descriptor,
+ None))
+ case None =>
+ DeletionVectorInfo(
+ false,
+ KEEP_ALL,
+ 0L,
+ new InMemoryDeletionVectorPayload(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}")
+ }
+ }
+
+ 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.
Review Comment:
`samePath` compares only `(authority, path)` and ignores the URI scheme. If
two different schemes (e.g., `file:` vs `s3a:`) ever appear with the same
authority/path, this could match the wrong `AddFile` and attach incorrect DV
metadata. Consider including the scheme in the comparison key (e.g., `(scheme,
authority, path)`) or otherwise enforcing scheme equality before treating paths
as identical.
##########
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:
##########
@@ -593,32 +595,22 @@ object GlutenConfig extends ConfigRegistry {
val confPrefixSession = prefixSessionOf(backendName)
val confPrefix = prefixOf(backendName)
+ // Column mapping mode is passed to Velox through scan splits, not through
+ // native session configs.
+ val veloxSplitColumnMappingConfigs = Set(
+ VELOX_PARQUET_USE_COLUMN_NAMES,
+ SPARK_ORC_FORCE_POSITIONAL_EVOLUTION)
Review Comment:
`SPARK_ORC_FORCE_POSITIONAL_EVOLUTION`
(`spark.hadoop.orc.force.positional.evolution`) does not match either
`prefixOf(velox)` or `prefixSessionOf(velox)`, so it won't be included in the
`conf.filter` set in the first place and therefore doesn't need to be excluded
here. Keeping it in `veloxSplitColumnMappingConfigs` is misleading; consider
removing it from this set (or adding a clarifying comment about why it is
listed despite never being in the native conf map).
--
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]
