felipepessoto commented on code in PR #13134: URL: https://github.com/apache/gluten/pull/13134#discussion_r4277647822
########## backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala: ########## @@ -0,0 +1,247 @@ +/* + * 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 io.delta.sql.DeltaSparkSessionExtension + +import scala.collection.mutable.ListBuffer + +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() Review Comment: Addressed in `fb6868738f`: all three calls now use `startTransaction(catalogTableOpt = None)`. These are path-based test tables, so this preserves the existing behavior without using the deprecated overload. ########## backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaTransactionObserverSuite.scala: ########## @@ -0,0 +1,247 @@ +/* + * 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 io.delta.sql.DeltaSparkSessionExtension + +import scala.collection.mutable.ListBuffer + +class DeltaTransactionObserverSuite extends SparkFunSuite with SharedSparkSession { Review Comment: Addressed in `fb6868738f`. Added `@ExtendedSQLTest` to both concrete observer suites, following `DeleteSQLSuite`. Verified all nine tests execute with `-DtagsToInclude=org.apache.spark.tags.ExtendedSQLTest` on both Spark 3.5 / Delta 3.3 and Spark 4.0 / Delta 4.0. The existing Spark 4.0 extended CI job now selects them; the Spark 4.1 slow-job profile limitation is unchanged. ########## gluten-delta/src-delta33/test/scala/org/apache/spark/sql/delta/GlutenDeltaTransactionObserverSuite.scala: ########## @@ -0,0 +1,77 @@ +/* + * 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 + +import org.apache.spark.SparkConf +import org.apache.spark.sql.GlutenQueryTest +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 io.delta.sql.DeltaSparkSessionExtension + +abstract class GlutenDeltaTransactionObserverSuite + extends GlutenQueryTest + with SharedSparkSession { + + private val nativeWriteKey = "spark.gluten.sql.columnar.backend.velox.delta.enableNativeWrite" Review Comment: Addressed in `fb6868738f`. Moved the test body directly into `VeloxDeltaTransactionObserverSuite` and removed the abstract gluten-delta suite and its symlink. It now uses `VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key` and `isInstanceOf[GlutenOptimisticTransaction]`. Both backend src-delta40 symlinks remain; there are no POM changes. ########## backends-velox/src-delta33/main/scala/org/apache/spark/sql/execution/datasources/v2/DeltaWriteOperators.scala: ########## @@ -77,26 +74,79 @@ 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 = { + underlying.setNextObserver(unwrap(nextTxnObserver)) + } Review Comment: @malinjawi Added the short reminder in `fb6868738f`, beside `setNextObserver`, to delegate `getNextObserver` when upgrading to Delta 4.3’s API. No getter override is added for the older supported versions. ########## gluten-delta/src-delta33/test/scala/org/apache/spark/sql/delta/GlutenDeltaTransactionObserverSuite.scala: ########## @@ -0,0 +1,77 @@ +/* + * 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 + +import org.apache.spark.SparkConf +import org.apache.spark.sql.GlutenQueryTest +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 io.delta.sql.DeltaSparkSessionExtension + +abstract class GlutenDeltaTransactionObserverSuite + extends GlutenQueryTest + with SharedSparkSession { + + private val nativeWriteKey = "spark.gluten.sql.columnar.backend.velox.delta.enableNativeWrite" + + 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(nativeWriteKey, "true") + } + + test("native Delta DELETE preserves the phase-locking transaction observer") { + withTempDir { + dir => + withSQLConf(nativeWriteKey -> "false") { + spark.range(4).write.format("delta").save(dir.getCanonicalPath) + } + val observer = new PhaseLockingTransactionExecutionObserver( + OptimisticTransactionPhases.forName("native-delete")) { + override def startingTransaction(f: => OptimisticTransaction): OptimisticTransaction = { + val transaction = super.startingTransaction(f) + assert( + transaction.getClass.getName == + "org.apache.spark.sql.delta.GlutenOptimisticTransaction") + transaction + } + } + observer.phaseLocks.foreach(_.entryBarrier.unblock()) + + TransactionExecutionObserver.withObserver(observer) { + sql(s"DELETE FROM delta.`${dir.getCanonicalPath}`").collect() + assert(TransactionExecutionObserver.getObserver eq observer) + } Review Comment: Addressed in `fb6868738f`. The regression now uses `DELETE ... WHERE id < 2` and checks the exact remaining rows, `2` and `3`. The fixture writes all four rows into one file with deletion vectors explicitly disabled; the test also verifies that the original file was replaced and that the resulting files have no DVs. The observer still asserts that the transaction is a `GlutenOptimisticTransaction`. -- 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]
