This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 020211a4b4 [GLUTEN-12743][VL] Preserve transaction observers in native
Delta commands (#13134)
020211a4b4 is described below
commit 020211a4b4c132f8133d0b3786b2d1928d28f0bf
Author: Felipe Pessoto <[email protected]>
AuthorDate: Thu Oct 15 08:31:57 2026 -0700
[GLUTEN-12743][VL] Preserve transaction observers in native Delta commands
(#13134)
* [GLUTEN-12743][VL] Preserve transaction observers in native Delta commands
Delegate native transaction creation and lifecycle callbacks to the
existing Delta observer. Preserve next-observer advancement and nested command
scopes, and add lifecycle and native DELETE regressions.
Co-authored-by: Copilot <[email protected]>
* [GLUTEN-12743][VL] Keep observer integration coverage in gluten-delta
Export the native DELETE regression as an abstract gluten-delta suite and
run it through the existing Velox Delta test-jar dependency. Remove the added
gluten-ut dependency and use a concrete Velox suite instead.
Co-authored-by: Copilot <[email protected]>
* [GLUTEN-12743][VL] Share observer tests and refresh Delta baseline
Remove the 64 observer-related cases confirmed passing in Delta CI run
36155269247, retaining the remaining footer-metadata failure. Share the
unchanged observer regression suites from src-delta33 through existing-style
src-delta40 symlinks, with both Spark 3.5 and Spark 4.0 coverage.
Co-authored-by: Copilot <[email protected]>
* [GLUTEN-12743][VL] Address observer regression review comments
Keep native DELETE coverage in the Velox suite with typed references and
assert a data-file rewrite. Tag both suites for extended CI, use the supported
transaction API, and document the Delta 4.3 observer getter upgrade requirement.
Co-authored-by: Copilot <[email protected]>
---------
Co-authored-by: Copilot <[email protected]>
---
.../util/delta-spark-ut/known-failures.txt | 64 ------
.../datasources/v2/DeltaWriteOperators.scala | 83 +++++--
.../sql/delta/DeltaTransactionObserverSuite.scala | 249 +++++++++++++++++++++
.../delta/VeloxDeltaTransactionObserverSuite.scala | 88 ++++++++
.../sql/delta/DeltaTransactionObserverSuite.scala | 1 +
.../delta/VeloxDeltaTransactionObserverSuite.scala | 1 +
6 files changed, 406 insertions(+), 80 deletions(-)
diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt
b/.github/workflows/util/delta-spark-ut/known-failures.txt
index 51bbf2b9ed..a8b68d06be 100644
--- a/.github/workflows/util/delta-spark-ut/known-failures.txt
+++ b/.github/workflows/util/delta-spark-ut/known-failures.txt
@@ -65,7 +65,6 @@ org.apache.spark.sql.delta.ConvertToDeltaSQLSuite#external
tables use correct pa
org.apache.spark.sql.delta.ConvertToDeltaScalaSuite#external tables use
correct path scheme
org.apache.spark.sql.delta.DeleteMetricsSuite#delete-metrics: delete one row
per file - Partitioned = false, cdfEnabled = false
org.apache.spark.sql.delta.DeleteMetricsSuite#delete-metrics: delete one row
per file - Partitioned = false, cdfEnabled = true
-org.apache.spark.sql.delta.DeltaAllFilesInCrcSuite#test all-files-in-crc
verification failure also triggers and logs incremental-commit verification
result
org.apache.spark.sql.delta.DeltaCDCIdColumnMappingSuite#filters with special
characters in name should be pushed down - column mapping id mode
org.apache.spark.sql.delta.DeltaCDCNameColumnMappingSuite#filters with special
characters in name should be pushed down - column mapping name mode
org.apache.spark.sql.delta.DeltaCDCSQLIdColumnMappingSuite#filters with
special characters in name should be pushed down - column mapping id mode
@@ -266,15 +265,7 @@
org.apache.spark.sql.delta.DescribeDeltaHistoryWithCatalogOwnedBatch100Suite#mer
org.apache.spark.sql.delta.DescribeDeltaHistoryWithCatalogOwnedBatch100Suite#merge-metrics:
upsert and delete with conditions - Partitioned = true, CDF = false
org.apache.spark.sql.delta.DescribeDeltaHistoryWithCatalogOwnedBatch100Suite#merge-metrics:
upsert and delete with conditions - Partitioned = true, CDF = true
org.apache.spark.sql.delta.DescribeDeltaHistoryWithCatalogOwnedBatch100Suite#operation
metrics - merge
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Disable Deletion
Vectors feature - withUnset: false
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Disable Deletion
Vectors feature - withUnset: true
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Disable row
tracking feature - withUnset: false
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Disable row
tracking feature - withUnset: true
org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Enable column
mapping feature - txnInterleaved: true
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Enable deletion
vectors feature
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Enable row
tracking feature concurrent txn: delete
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Removing column
mapping mode produces conflict - startMode: id
-org.apache.spark.sql.delta.FeatureEnablementConcurrencySuite#Removing column
mapping mode produces conflict - startMode: name
org.apache.spark.sql.delta.FileSizeHistogramSuite#check CommitStats with
deletes
org.apache.spark.sql.delta.FileSizeHistogramSuite#histogram is re-calculated
when files are removed
org.apache.spark.sql.delta.GeneratedColumnSuite#update_generated_column_with_incorrect_value
@@ -390,49 +381,6 @@
org.apache.spark.sql.delta.SnapshotManagementWithCoordinatedCommitsBatch1Suite#s
org.apache.spark.sql.delta.SnapshotManagementWithCoordinatedCommitsBatch2Suite#recover
from a corrupt checkpoint: previous checkpoint doesn't exist
org.apache.spark.sql.delta.SnapshotManagementWithCoordinatedCommitsBatch2Suite#should
not recover when both the current and previous checkpoints are broken
org.apache.spark.sql.delta.UpdateMetricsSuite#update-metrics: update one row
per file - Partitioned = false, cdfEnabled = false
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 1
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 2
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 3
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 4
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 5
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 6
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#DELETE
- Scenario 7
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 1
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 2
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 3
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 4
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 5
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 6
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#OPTIMIZE
- Scenario 7
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 1
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 2
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 3
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 4
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 5
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 6
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsDVSuite#UPDATE
- Scenario 7
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 1
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 2
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 3
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 4
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 5
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 6
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#DELETE
- Scenario 7
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 1
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 2
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 3
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 4
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 5
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 6
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#OPTIMIZE
- Scenario 7
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 1
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 2
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 3
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 4
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 5
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 6
-org.apache.spark.sql.delta.commands.backfill.RowTrackingBackfillConflictsSuite#UPDATE
- Scenario 7
-org.apache.spark.sql.delta.concurrency.TransactionExecutionObserverSuite#Phase
Locking - delete command
org.apache.spark.sql.delta.coordinatedcommits.CoordinatedCommitsSuite#Incomplete
backfills are handled properly by next commit after CC to FS conversion
org.apache.spark.sql.delta.deletionvectors.DeletionVectorsSuite#DELETE with
DVs with column mapping mode=id
org.apache.spark.sql.delta.deletionvectors.DeletionVectorsSuite#huge table:
delete a small number of rows from tables of 2B rows with DVs
@@ -547,28 +495,16 @@
org.apache.spark.sql.delta.generatedsuites.UpdateBaseTempViewSQLPathBasedSuite#t
org.apache.spark.sql.delta.generatedsuites.UpdateBaseTempViewSQLPathBasedSuite#test
update on temp view - nontrivial projection - SQL TempView
org.apache.spark.sql.delta.optimize.OptimizeCompactionSQLSuite#optimize -
multiple jobs start executing at once
org.apache.spark.sql.delta.optimize.OptimizeCompactionScalaSuite#optimize -
multiple jobs start executing at once
-org.apache.spark.sql.delta.optimize.OptimizeConflictSuite#conflict handling
between Optimize and Business Txn
org.apache.spark.sql.delta.optimize.OptimizeMetricsSuite#optimize ZOrderBy
operation metrics in Delta table history
org.apache.spark.sql.delta.optimize.OptimizeMetricsSuite#optimize metrics on
idempotent operations
org.apache.spark.sql.delta.perf.OptimizeGeneratedColumnSuite#end-to-end test
of behaviors of write/read null on partition column
org.apache.spark.sql.delta.perf.OptimizeGeneratedColumnSuite#five digits year
in a date_format yyyy-MM partition column
org.apache.spark.sql.delta.perf.OptimizeGeneratedColumnSuite#five digits year
in a date_format yyyy-MM-dd-HH partition column
-org.apache.spark.sql.delta.rowid.ConflictCheckerRowIdSuite#Re-added files keep
their row IDs after conflict with txn not updating high watermark
-org.apache.spark.sql.delta.rowid.ConflictCheckerRowIdSuite#concurrent
transactions do not assign overlapping row IDs
-org.apache.spark.sql.delta.rowid.ConflictCheckerRowIdSuite#re-added files keep
their row ids
org.apache.spark.sql.delta.rowid.RowIdSuite#Filter by base Row IDs
org.apache.spark.sql.delta.rowid.RowIdSuite#Filter by base Row IDs in subquery
org.apache.spark.sql.delta.rowid.RowIdSuite#No dictionary filtering on
_metadata.row_id
org.apache.spark.sql.delta.rowid.RowIdSuite#missing base row ids and default
row commit versions
org.apache.spark.sql.delta.rowid.RowIdSuite#row ids can be read back
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencySuite#Interleaved
delete right after protocol downgrade should abort due to protocol change
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencySuite#Interleaved
update right after protocol downgrade should abort due to protocol change
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencySuite#Single
Unbackfill batch interleaves delete
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencySuite#Single
Unbackfill batch interleaves update
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencyWithoutDVsSuite#Interleaved
delete right after protocol downgrade should abort due to protocol change
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencyWithoutDVsSuite#Interleaved
update right after protocol downgrade should abort due to protocol change
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencyWithoutDVsSuite#Single
Unbackfill batch interleaves delete
-org.apache.spark.sql.delta.rowid.RowTrackingRemovalConcurrencyWithoutDVsSuite#Single
Unbackfill batch interleaves update
org.apache.spark.sql.delta.rowtracking.RowTrackingReadWriteSuite#write and
read table with all-null materialized columns
org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#Data
skipping handles aliasing for _metadata fields
org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
diff --git
a/backends-velox/src-delta33/main/scala/org/apache/spark/sql/execution/datasources/v2/DeltaWriteOperators.scala
b/backends-velox/src-delta33/main/scala/org/apache/spark/sql/execution/datasources/v2/DeltaWriteOperators.scala
index ebd9e7a3e4..e5b40e7182 100644
---
a/backends-velox/src-delta33/main/scala/org/apache/spark/sql/execution/datasources/v2/DeltaWriteOperators.scala
+++
b/backends-velox/src-delta33/main/scala/org/apache/spark/sql/execution/datasources/v2/DeltaWriteOperators.scala
@@ -28,8 +28,7 @@ case class GlutenDeltaLeafV2CommandExec(delegate:
LeafV2CommandExec) extends Lea
override def metrics: Map[String, SQLMetric] = delegate.metrics
override protected def run(): Seq[InternalRow] = {
- TransactionExecutionObserver.withObserver(
- DeltaV2WriteOperators.UseColumnarDeltaTransactionLog) {
+ DeltaV2WriteOperators.withColumnarTransaction {
delegate.executeCollect()
}
}
@@ -50,8 +49,7 @@ case class GlutenDeltaLeafRunnableCommand(delegate:
LeafRunnableCommand)
}
override def run(sparkSession: SparkSession): Seq[Row] = {
- TransactionExecutionObserver.withObserver(
- DeltaV2WriteOperators.UseColumnarDeltaTransactionLog) {
+ DeltaV2WriteOperators.withColumnarTransaction {
delegate.run(sparkSession)
}
}
@@ -67,8 +65,7 @@ case class GlutenDeltaRunnableCommand(delegate:
RunnableCommand) extends LeafRun
}
override def run(sparkSession: SparkSession): Seq[Row] = {
- TransactionExecutionObserver.withObserver(
- DeltaV2WriteOperators.UseColumnarDeltaTransactionLog) {
+ DeltaV2WriteOperators.withColumnarTransaction {
delegate.run(sparkSession)
}
}
@@ -77,26 +74,80 @@ case class GlutenDeltaRunnableCommand(delegate:
RunnableCommand) extends LeafRun
}
object DeltaV2WriteOperators {
- object UseColumnarDeltaTransactionLog extends TransactionExecutionObserver {
+ private[sql] def withColumnarTransaction[T](f: => T): T = {
+ TransactionExecutionObserver.getObserver match {
+ case _: UseColumnarDeltaTransactionLog => f
+ case observer =>
+ TransactionExecutionObserver.setObserver(wrap(observer))
+ try {
+ f
+ } finally {
+ // A completed transaction may have advanced to the next observer.
+
TransactionExecutionObserver.setObserver(unwrap(TransactionExecutionObserver.getObserver))
+ }
+ }
+ }
+
+ private def wrap(observer: TransactionExecutionObserver):
TransactionExecutionObserver =
+ observer match {
+ case _: UseColumnarDeltaTransactionLog => observer
+ case _ => new UseColumnarDeltaTransactionLog(observer)
+ }
+
+ private def unwrap(observer: TransactionExecutionObserver):
TransactionExecutionObserver =
+ observer match {
+ case columnar: UseColumnarDeltaTransactionLog => columnar.underlying
+ case _ => observer
+ }
+
+ private class UseColumnarDeltaTransactionLog(val underlying:
TransactionExecutionObserver)
+ extends TransactionExecutionObserver {
override def startingTransaction(f: => OptimisticTransaction):
OptimisticTransaction = {
- val delegate = f
- new GlutenOptimisticTransaction(delegate)
+ underlying.startingTransaction {
+ new GlutenOptimisticTransaction(f)
+ }
}
- override def preparingCommit[T](f: => T): T = f
+ override def preparingCommit[T](f: => T): T = underlying.preparingCommit(f)
- override def beginDoCommit(): Unit = ()
+ override def beginDoCommit(): Unit = underlying.beginDoCommit()
- override def beginBackfill(): Unit = ()
+ override def beginBackfill(): Unit = underlying.beginBackfill()
- override def beginPostCommit(): Unit = ()
+ override def beginPostCommit(): Unit = underlying.beginPostCommit()
- override def transactionCommitted(): Unit = ()
+ override def transactionCommitted(): Unit = withObserverAdvance {
+ underlying.transactionCommitted()
+ }
- override def transactionAborted(): Unit = ()
+ override def transactionAborted(): Unit = withObserverAdvance {
+ underlying.transactionAborted()
+ }
override def createChild(): TransactionExecutionObserver = {
- TransactionExecutionObserver.getObserver
+ wrap(underlying.createChild())
+ }
+
+ override def setNextObserver(nextTxnObserver:
TransactionExecutionObserver): Unit = {
+ // Delta 4.3 adds getNextObserver; also delegate that getter when
upgrading the API.
+ underlying.setNextObserver(unwrap(nextTxnObserver))
+ }
+
+ override def advanceToNextThreadObserver(): Unit = withObserverAdvance {
+ underlying.advanceToNextThreadObserver()
+ }
+
+ private def withObserverAdvance(f: => Unit): Unit = {
+ val inColumnarScope =
+
TransactionExecutionObserver.getObserver.isInstanceOf[UseColumnarDeltaTransactionLog]
+ try {
+ f
+ } finally {
+ if (inColumnarScope) {
+ // Delta's observer can replace itself from inside its commit/abort
callbacks.
+
TransactionExecutionObserver.setObserver(wrap(TransactionExecutionObserver.getObserver))
+ }
+ }
}
}
}
diff --git
a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
new file mode 100644
index 0000000000..3290f2c858
--- /dev/null
+++
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
@@ -0,0 +1,249 @@
+/*
+ * 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.spark.sql.delta
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.sql.delta.catalog.DeltaCatalog
+import org.apache.spark.sql.delta.fuzzer.{OptimisticTransactionPhases,
PhaseLockingTransactionExecutionObserver}
+import
org.apache.spark.sql.execution.datasources.v2.DeltaV2WriteOperators.withColumnarTransaction
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.tags.ExtendedSQLTest
+
+import io.delta.sql.DeltaSparkSessionExtension
+
+import scala.collection.mutable.ListBuffer
+
+@ExtendedSQLTest
+class DeltaTransactionObserverSuite extends SparkFunSuite with
SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.sql.extensions", classOf[DeltaSparkSessionExtension].getName)
+ .set("spark.sql.catalog.spark_catalog", classOf[DeltaCatalog].getName)
+ .set("spark.ui.enabled", "false")
+ }
+
+ private class RecordingObserver extends TransactionExecutionObserver {
+ val events: ListBuffer[String] = ListBuffer.empty
+ var child: Option[TransactionExecutionObserver] = None
+
+ override def startingTransaction(f: => OptimisticTransaction):
OptimisticTransaction = {
+ events += "starting"
+ val transaction = f
+ events += "started"
+ transaction
+ }
+
+ override def preparingCommit[T](f: => T): T = {
+ events += "preparing"
+ val result = f
+ events += "prepared"
+ result
+ }
+
+ override def beginDoCommit(): Unit = { events += "commit" }
+ override def beginBackfill(): Unit = { events += "backfill" }
+ override def beginPostCommit(): Unit = { events += "postCommit" }
+ override def transactionCommitted(): Unit = { events += "committed" }
+ override def transactionAborted(): Unit = { events += "aborted" }
+
+ override def createChild(): TransactionExecutionObserver = {
+ events += "child"
+ child.getOrElse(TransactionExecutionObserver.getObserver)
+ }
+ }
+
+ private def phaseObserver(name: String):
PhaseLockingTransactionExecutionObserver = {
+ val observer =
+ new
PhaseLockingTransactionExecutionObserver(OptimisticTransactionPhases.forName(name))
+ observer.phaseLocks.foreach(_.entryBarrier.unblock())
+ observer
+ }
+
+ private def finishCommit(observer: TransactionExecutionObserver): Unit = {
+ observer.preparingCommit(())
+ observer.beginDoCommit()
+ observer.beginBackfill()
+ observer.beginPostCommit()
+ observer.transactionCommitted()
+ }
+
+ test("native transactions preserve the existing observer and all lifecycle
callbacks") {
+ withTempDir {
+ dir =>
+ val original = new RecordingObserver
+ val deltaLog = DeltaLog.forTable(spark, dir.getCanonicalPath)
+ TransactionExecutionObserver.withObserver(original) {
+ withColumnarTransaction {
+ val transaction = deltaLog.startTransaction(catalogTableOpt = None)
+ assert(transaction.isInstanceOf[GlutenOptimisticTransaction])
+ val observer = TransactionExecutionObserver.getObserver
+ assert(transaction.executionObserver eq observer)
+ assert(original.events.toList == List("starting", "started"))
+ assert(observer.preparingCommit(42) == 42)
+ observer.beginDoCommit()
+ observer.beginBackfill()
+ observer.beginPostCommit()
+ observer.transactionCommitted()
+ observer.transactionAborted()
+ }
+ assert(TransactionExecutionObserver.getObserver eq original)
+ }
+ assert(
+ original.events.toList == List(
+ "starting",
+ "started",
+ "preparing",
+ "prepared",
+ "commit",
+ "backfill",
+ "postCommit",
+ "committed",
+ "aborted"))
+ }
+ }
+
+ test("creation, preparation, and command exceptions retain their identity
and observer scope") {
+ val original = new RecordingObserver
+ val failure = new IllegalStateException("observer test")
+ TransactionExecutionObserver.withObserver(original) {
+ assert(intercept[IllegalStateException] {
+ withColumnarTransaction {
+ TransactionExecutionObserver.getObserver.startingTransaction(throw
failure)
+ }
+ } eq failure)
+ assert(TransactionExecutionObserver.getObserver eq original)
+
+ assert(intercept[IllegalStateException] {
+ withColumnarTransaction {
+ TransactionExecutionObserver.getObserver.preparingCommit(throw
failure)
+ }
+ } eq failure)
+ assert(TransactionExecutionObserver.getObserver eq original)
+
+ assert(intercept[IllegalStateException] {
+ withColumnarTransaction(throw failure)
+ } eq failure)
+ assert(TransactionExecutionObserver.getObserver eq original)
+ }
+ }
+
+ for (aborted <- Seq(false, true)) {
+ test(s"Delta's phase observer advances after completion:
aborted=$aborted") {
+ withTempDir {
+ dir =>
+ val original = phaseObserver("first")
+ val next = phaseObserver("next")
+ original.setNextObserver(next, autoAdvance = true)
+ original.phases.postCommitPhase.exitBarrier.unblock()
+ val deltaLog = DeltaLog.forTable(spark, dir.getCanonicalPath)
+
+ TransactionExecutionObserver.withObserver(original) {
+ withColumnarTransaction {
+ val first = deltaLog.startTransaction(catalogTableOpt = None)
+ assert(first.isInstanceOf[GlutenOptimisticTransaction])
+ val firstObserver = first.executionObserver
+ if (aborted) {
+ firstObserver.transactionAborted()
+ } else {
+ finishCommit(firstObserver)
+ }
+
+ assert(TransactionExecutionObserver.getObserver ne firstObserver)
+ val second = deltaLog.startTransaction(catalogTableOpt = None)
+ assert(second.isInstanceOf[GlutenOptimisticTransaction])
+ finishCommit(second.executionObserver)
+ assert(next.allPhasesHavePassed)
+ }
+ assert(TransactionExecutionObserver.getObserver eq next)
+ }
+ }
+ }
+ }
+
+ test("nested native commands reuse the observer and preserve advancement on
scope exit") {
+ val original = new RecordingObserver
+ val next = new RecordingObserver
+ TransactionExecutionObserver.withObserver(original) {
+ withColumnarTransaction {
+ val observer = TransactionExecutionObserver.getObserver
+ withColumnarTransaction {
+ assert(TransactionExecutionObserver.getObserver eq observer)
+ observer.setNextObserver(next)
+ TransactionExecutionObserver.advanceToNextObserver()
+ TransactionExecutionObserver.getObserver.beginDoCommit()
+ }
+ assert(TransactionExecutionObserver.getObserver ne observer)
+ assert(TransactionExecutionObserver.getObserver ne next)
+ TransactionExecutionObserver.getObserver.beginBackfill()
+ }
+ assert(TransactionExecutionObserver.getObserver eq next)
+ }
+ assert(next.events.toList == List("commit", "backfill"))
+ }
+
+ test("native child observers preserve Delta's current-thread and
custom-child behavior") {
+ val original = new RecordingObserver
+ val child = new RecordingObserver
+ TransactionExecutionObserver.withObserver(original) {
+ withColumnarTransaction {
+ val observer = TransactionExecutionObserver.getObserver
+ assert(observer.createChild() eq observer)
+ original.child = Some(child)
+ val childObserver = observer.createChild()
+ assert(childObserver ne child)
+ childObserver.beginDoCommit()
+ assert(child.events.toList == List("commit"))
+ assert(TransactionExecutionObserver.getObserver eq observer)
+ }
+ assert(TransactionExecutionObserver.getObserver eq original)
+ }
+ }
+
+ test("a callback exception does not discard an observer advancement") {
+ val next = new RecordingObserver
+ val failure = new IllegalStateException("commit callback")
+ val original = new RecordingObserver {
+ override def transactionCommitted(): Unit = {
+ advanceToNextThreadObserver()
+ throw failure
+ }
+ }
+ original.setNextObserver(next)
+ TransactionExecutionObserver.withObserver(original) {
+ assert(intercept[IllegalStateException] {
+ withColumnarTransaction {
+ TransactionExecutionObserver.getObserver.transactionCommitted()
+ }
+ } eq failure)
+ assert(TransactionExecutionObserver.getObserver eq next)
+ }
+ }
+
+ test("callbacks outside a native command do not install a native observer") {
+ val original = new RecordingObserver
+ TransactionExecutionObserver.withObserver(original) {
+ val observer = withColumnarTransaction {
+ TransactionExecutionObserver.getObserver
+ }
+ observer.transactionCommitted()
+ observer.transactionAborted()
+ assert(TransactionExecutionObserver.getObserver eq original)
+ }
+ assert(original.events.toList == List("committed", "aborted"))
+ }
+}
diff --git
a/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
new file mode 100644
index 0000000000..d0817853c0
--- /dev/null
+++
b/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
@@ -0,0 +1,88 @@
+/*
+ * 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.spark.sql.delta
+
+import org.apache.gluten.config.{GlutenConfig, VeloxDeltaConfig}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{GlutenQueryTest, Row}
+import org.apache.spark.sql.delta.catalog.DeltaCatalog
+import org.apache.spark.sql.delta.fuzzer.{OptimisticTransactionPhases,
PhaseLockingTransactionExecutionObserver}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.tags.ExtendedSQLTest
+
+import io.delta.sql.DeltaSparkSessionExtension
+
+@ExtendedSQLTest
+class VeloxDeltaTransactionObserverSuite extends GlutenQueryTest with
SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.sql.extensions", classOf[DeltaSparkSessionExtension].getName)
+ .set("spark.sql.catalog.spark_catalog", classOf[DeltaCatalog].getName)
+ .set("spark.plugins", "org.apache.gluten.GlutenPlugin")
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "1024MB")
+ .set("spark.default.parallelism", "1")
+ .set("spark.sql.shuffle.partitions", "1")
+ .set("spark.databricks.delta.snapshotPartitions", "1")
+ .set("spark.sql.ansi.enabled", "false")
+ .set(GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key, "false")
+ .set(VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key, "true")
+ }
+
+ test("native Delta DELETE rewrites rows while preserving the phase-locking
observer") {
+ withTempDir {
+ dir =>
+ withSQLConf(VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key -> "false") {
+ spark
+ .range(4)
+ .coalesce(1)
+ .write
+ .format("delta")
+ .option(DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key, "false")
+ .save(dir.getCanonicalPath)
+ }
+ val deltaLog = DeltaLog.forTable(spark, dir.getCanonicalPath)
+ val originalFiles = deltaLog.update().allFiles.collect()
+ assert(originalFiles.length == 1)
+ val observer = new PhaseLockingTransactionExecutionObserver(
+ OptimisticTransactionPhases.forName("native-delete")) {
+ override def startingTransaction(f: => OptimisticTransaction):
OptimisticTransaction = {
+ val transaction = super.startingTransaction(f)
+ assert(transaction.isInstanceOf[GlutenOptimisticTransaction])
+ transaction
+ }
+ }
+ observer.phaseLocks.foreach(_.entryBarrier.unblock())
+
+ TransactionExecutionObserver.withObserver(observer) {
+ sql(s"DELETE FROM delta.`${dir.getCanonicalPath}` WHERE id <
2").collect()
+ assert(TransactionExecutionObserver.getObserver eq observer)
+ }
+ assert(observer.allPhasesHavePassed)
+ val remainingFiles = deltaLog.update().allFiles.collect()
+ assert(remainingFiles.nonEmpty)
+ assert(remainingFiles.forall(_.path != originalFiles.head.path))
+ assert(remainingFiles.forall(_.deletionVector == null))
+ checkAnswer(
+ spark.read.format("delta").load(dir.getCanonicalPath),
+ Seq(Row(2L), Row(3L)))
+ }
+ }
+}
diff --git
a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
new file mode 120000
index 0000000000..b7864d0a58
--- /dev/null
+++
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
@@ -0,0 +1 @@
+../../../../../../../../src-delta33/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala
\ No newline at end of file
diff --git
a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
new file mode 120000
index 0000000000..b7da75bc7f
--- /dev/null
+++
b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
@@ -0,0 +1 @@
+../../../../../../../../src-delta33/test/scala/org/apache/spark/sql/delta/VeloxDeltaTransactionObserverSuite.scala
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
