malinjawi commented on code in PR #13134: URL: https://github.com/apache/gluten/pull/13134#discussion_r4257240340
########## 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: nit: `startTransaction()` is deprecated and adds a build warning here. The other Velox Delta tests use `startTransaction(catalogTableOpt = None)`. Could you change this? ########## 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: nit: `DELETE` with no `WHERE` is metadata-only in Delta. It just removes the files from the log and never calls `writeFiles`, so the native writer doesn't actually run in this test. Can we add a predicate like `WHERE id < 2` and check that 2 rows are left? Then the native rewrite runs with the observer in place. ########## 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: @felipepessoto agreed, nothing is needed for the Delta versions we use today. One heads-up: Delta 4.3.0 adds `getNextObserver` to `ChainableExecutionObserver` (delta-io/delta#6585). This class would inherit it and quietly return `None`, since `setNextObserver` goes to `underlying`. Nothing calls it on transaction observers yet, and it can't be overridden in this shared file. Could you leave a short comment here so whoever bumps Delta remembers to forward it? ########## 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: nit: This test only works on Velox, but gluten-delta is shared with ClickHouse and Bolt. That's why the config key and class name end up as strings here. Could we move the test body into `VeloxDeltaTransactionObserverSuite` and drop the abstract class? Then it can use the real references instead of strings - `VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key` for the config key - `transaction.isInstanceOf[GlutenOptimisticTransaction]` for the class check The gluten-delta file and its `src-delta40` symlink can go. The `backends-velox/src-delta40` symlink still needs to stay for Delta 4.x. ########## 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: The CI only ran these suites on Spark 3.5. The 4.0 and 4.1 jobs compile them but filter them out, so they never actually run there. Tagging both concrete suites with `@ExtendedSQLTest`, like `DeleteSQLSuite`, would get them running on 4.0 too. 4.1 would still be covered only by the Delta UT run, since the 4.1 slow job doesn't build with `-Pdelta`. -- 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]
