malinjawi commented on code in PR #12612:
URL: https://github.com/apache/gluten/pull/12612#discussion_r3820907003
##########
.github/workflows/util/delta-spark-ut/known-failures.txt:
##########
@@ -59,8 +59,6 @@ org.apache.spark.sql.delta.CloneTableSQLSuite#shallow clone
across file systems
org.apache.spark.sql.delta.CloneTableSQLWithCatalogOwnedBatch100Suite#shallow
clone across file systems
org.apache.spark.sql.delta.CloneTableSQLWithCatalogOwnedBatch1Suite#shallow
clone across file systems
org.apache.spark.sql.delta.CloneTableSQLWithCatalogOwnedBatch2Suite#shallow
clone across file systems
-org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#Cloning table
with persistent DVs and absolute parquet paths
-org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#Shallow clone
round-trip with DVs
Review Comment:
The 4 DV-clone tests in `CloneTableScalaDeletionVectorSuite` split exactly
as your fix predicts. The 2 that failed write a `"u"` DV after a shallow clone
DV relative to the clone root, data files absolute into the upstream table, so
the old walk-up stops at the upstream `_delta_log`. The 2 that passed had
`makePathsAbsolute` rewrite their DVs to absolute `"p"` first. Worth putting in
the description as it's a real bug fix, not cleanup.
Not blocking: these only run on delta40, since `delta_spark_ut.yml` defaults
to `spark-4.1`. The caller is shared so I don't think there's real risk, but a
shallow-clone-then-`DELETE` read in `DeltaSuite` would cover 3.5 too if you
think it's worth it.
##########
gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala:
##########
@@ -98,20 +99,29 @@ case class DeltaScanTransformer(
override def getSplitInfosFromPartitions(
partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = {
val splitInfos = super.getSplitInfosFromPartitions(partitions)
- val partitionColumnCount = getPartitionSchema.fields.length
- splitInfos.zip(partitions).map {
- case (localFiles: LocalFilesNode, (filePartition: FilePartition, _)) =>
- DeltaDeletionVectorScanInfo
- .normalize(partitionColumnCount, filePartition.files.toSeq)
- .map {
- case (otherMetadataColumns, deltaReadOptions) =>
- DeltaLocalFilesBuilder.makeDeltaLocalFiles(
- localFiles,
- otherMetadataColumns.asJava,
- deltaReadOptions.asJava): SplitInfo
- }
- .getOrElse(localFiles)
- case (splitInfo, _) => splitInfo
+ // 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.
+ relation.location match {
+ case tahoe: TahoeFileIndex =>
+ val tableRootPath = tahoe.path
Review Comment:
Nothing pins this arm. `DeltaDeletionVectorScanInfoSuite:140` is the only
root-discriminating test and it calls `normalize` directly, bypassing
`DeltaScanTransformer`. The integration test uses a table where both candidate
roots coincide. A wrong root here is caught only by the delta40-only clone
shards.
##########
gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala:
##########
@@ -427,6 +430,41 @@ abstract class DeltaSuite extends
WholeStageTransformerSuite {
}
}
+ testWithMinSparkVersion("deletion vector on partitioned table", "3.4") {
+ withTempPath {
+ p =>
+ import testImplicits._
+ val path = p.getCanonicalPath
+ // Partitioned so data files live under partition subdirs
(region=.../...). The DV path is
+ // resolved from the table root (TahoeFileIndex.path) regardless of
partition nesting; this
+ // guards the removal of the old partition-count-based table-path
walk-up.
Review Comment:
I think this comment claims more than the test delivers.
`getPartitionSchema` is `relation.partitionSchema`, so `partitionBy("region")`
gives `partitionColumnCount = 1` the old loop goes `<root>/region=a` →
`<root>`, finds `_delta_log`, and returns on the first probe. This test would
be green on `main` too.
No issue with keeping it, but could you reword? The test that actually
discriminates is `DeltaDeletionVectorScanInfoSuite:140`, since it points the
`PartitionedFile` at an unrelated directory worth naming that one instead.
##########
backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala:
##########
@@ -97,10 +97,7 @@ object DeltaPlanningBenchmark extends SqlBasedBenchmark {
withDeltaTableWithDVs(numFiles, rowsPerFile) {
(path, partitionedFiles) =>
benchmark.addCase(s"normalize() - $numFiles DV files", benchmarkIters)
{
- _ =>
- DeltaDeletionVectorScanInfo.normalize(
- partitionColumnCount = 0,
- partitionFiles = partitionedFiles)
+ _ => DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new
Path(path))
Review Comment:
nit: scaladoc at L33-35 and L88 still talks about resolving/caching the
table path. Would you mind updating?
--
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]
