iemejia commented on code in PR #12612:
URL: https://github.com/apache/gluten/pull/12612#discussion_r3822549703
##########
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:
Done in 4e6b4ad — dropped the stale "resolving/caching table path" wording
at L33-35 and L88. The `normalize` path no longer resolves or caches the root
now that it's passed in explicitly.
##########
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:
Good catch — you're right. With `partitionColumnCount = 1` the old walk-up
hits `_delta_log` on the first probe, so this test would be green on `main`
too. Reworded in 4e6b4ad to describe what it actually covers (end-to-end DV
read over a partitioned table) and to point at
`DeltaDeletionVectorScanInfoSuite` ("normalize materializes DV read options
using the supplied table path") as the test that actually discriminates the
root, since it points the `PartitionedFile` at an unrelated directory.
##########
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:
Agreed that this arm is under-pinned: the only root-discriminating unit test
calls `normalize()` directly (bypassing `DeltaScanTransformer`), and the
integration test uses a table where both candidate roots coincide, so today
only the delta40-only clone shards would catch a wrong root selected here. The
shallow-clone-then-`DELETE` read you suggested on the known-failures thread is
the right way to pin this on 3.5 too — happy to add it in a follow-up commit to
this PR. Let me know if you'd prefer it in-PR before merge.
##########
.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:
Thanks for the detailed breakdown — it matches exactly what I see. I've
reframed the PR description around this being a correctness fix: a
shallow-clone `DELETE` writes a clone-root-relative `"u"` DV while the data
files point absolute into the upstream table, so the old walk-up stopped at the
upstream `_delta_log` and resolved the wrong root; the 2 tests that passed only
did so because `makePathsAbsolute` had already rewritten their DVs to absolute
`"p"` paths. Using `TahoeFileIndex.path` fixes all four, which is why they can
drop from the baseline.
On the not-blocking suggestion: I agree a shallow-clone-then-`DELETE` read
in `DeltaSuite` would extend this to 3.5 (the caller is shared, so the risk is
low, but it would pin the `DeltaScanTransformer` Tahoe arm directly). I'm happy
to add it — replied on the `DeltaScanTransformer` thread too.
--
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]
