You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/10 12:46:36 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6821-a3b297c4d1b0aa26776213342716aef9cb4ec34d created (now dc880d6d26)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6821-a3b297c4d1b0aa26776213342716aef9cb4ec34d
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at dc880d6d26 fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer (#6821)
This branch includes the following new commits:
new dc880d6d26 fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer (#6821)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer (#6821)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6821-a3b297c4d1b0aa26776213342716aef9cb4ec34d
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit dc880d6d265f8391b6202efa43574cfeea71e75a
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Sat Oct 10 12:46:12 2026 +0000
fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer (#6821)
* fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer
On Spark 3.x CometNativeWriteExec replaces the whole InsertIntoHadoopFsRelationCommand, but
only ran the job/task commit lifecycle. After the job commits it now refreshes the table's
file index, recaches cached plans that read the output path, and updates or clears the
table stats, as the command does. It also rejects duplicate output column names up front,
which the command checks before writing.
Co-authored-by: Isaac <no...@databricks.com>
* test: assert the duplicate-column write is native
Spark's own writer rejects the same columns, so a fallback would keep the test green
without exercising the native writer's check. Capture the plan of the failed write as
well, and assert it carried a native write.
Co-authored-by: Isaac <no...@databricks.com>
---------
Co-authored-by: Isaac <no...@databricks.com>
---
.../spark/sql/comet/CometNativeWriteExec.scala | 36 +++++-
.../comet/parquet/CometParquetWriterSuite.scala | 125 ++++++++++++++++++++-
.../comet/parquet/CometParquetWriterTestBase.scala | 43 +++++--
3 files changed, 189 insertions(+), 15 deletions(-)
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeWriteExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeWriteExec.scala
index 200c783132..ec903fc893 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeWriteExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeWriteExec.scala
@@ -37,9 +37,10 @@ import org.apache.spark.sql.comet.execution.arrow.CometArrowStream
import org.apache.spark.sql.comet.util.{Utils => CometUtils}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
-import org.apache.spark.sql.execution.command.DataWritingCommandExec
-import org.apache.spark.sql.execution.datasources.{OutputWriterFactory, WriteFilesExec}
+import org.apache.spark.sql.execution.command.{CommandUtils, DataWritingCommandExec}
+import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCommand, OutputWriterFactory, WriteFilesExec}
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.util.SchemaUtils
import org.apache.spark.sql.vectorized.ColumnarBatch
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -106,7 +107,21 @@ case class CometNativeWriteExec(
sparkContext.emptyRDD[ColumnarBatch]
}
+ /** The command this exec replaces, which holds the table this write updates, if any. */
+ private def command: InsertIntoHadoopFsRelationCommand = originalPlan.cmd match {
+ case cmd: InsertIntoHadoopFsRelationCommand => cmd
+ case other =>
+ throw new IllegalStateException(
+ s"${getClass.getSimpleName} replaced an unexpected write command: ${other.getClass}")
+ }
+
private def executeWriteAndCommit(): Unit = {
+ val cmd = command
+ // Like InsertIntoHadoopFsRelationCommand.run, reject duplicate columns before anything else.
+ SchemaUtils.checkColumnNameDuplication(
+ cmd.outputColumnNames,
+ session.sessionState.conf.caseSensitiveAnalysis)
+
if (!prepareOutputPathForMode()) {
logInfo(s"Skipping insertion into $outputPath - already exists (SaveMode.$mode)")
return
@@ -124,6 +139,23 @@ case class CometNativeWriteExec(
s"${metrics("files_written").value} files, " +
s"${metrics("bytes_written").value} bytes, ${metrics("rows_written").value} rows")
})(catchBlock = committer.abortJob(job))
+ // Outside the block above: the job has committed, so a failure here must not abort it.
+ refreshAfterWrite(cmd)
+ }
+
+ /**
+ * Repeats what InsertIntoHadoopFsRelationCommand.run does once the write has committed, since
+ * this exec replaces that command. Partition metadata updates are left out because partitioned
+ * writes never reach this exec.
+ */
+ private def refreshAfterWrite(cmd: InsertIntoHadoopFsRelationCommand): Unit = {
+ val fs = cmd.outputPath.getFileSystem(serializableHadoopConf.value)
+ // Refresh the files the table's relation has listed.
+ cmd.fileIndex.foreach(_.refresh())
+ // Recache any cached plan that reads the output path.
+ session.sharedState.cacheManager.recacheByPath(session, cmd.outputPath, fs)
+ // Update or clear the table stats, depending on spark.sql.statistics.size.autoUpdate.
+ cmd.catalogTable.foreach(CommandUtils.updateTableStats(session, _))
}
private def runNativeWriteJob(
diff --git a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala
index 466f6cad9e..2feb4e12e2 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala
@@ -34,7 +34,7 @@ import org.apache.parquet.hadoop.util.HadoopInputFile
import org.apache.parquet.schema.{MessageType, Type}
import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode}
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.comet.{CometBatchScanExec, CometNativeColumnarToRowExec, CometNativeScanExec, CometNativeWriteExec, CometScanExec, CometSparkToColumnarExec, CometWriteFilesExec}
import org.apache.spark.sql.execution.{ColumnarToRowTransition, FileSourceScanExec, SparkPlan, SQLExecution}
import org.apache.spark.sql.execution.command.DataWritingCommandExec
@@ -1318,9 +1318,9 @@ class CometParquetWriterSuite extends CometParquetWriterTestBase {
}
test("INSERT INTO ... SELECT is visible to subsequent reads") {
- assume(isSpark40Plus, "Requires the WriteFilesExec seam")
// https://github.com/apache/datafusion-comet/issues/3521 - reads returned no rows because the
- // bespoke write path never refreshed the catalog cache. Spark's command does that itself.
+ // bespoke write path never refreshed the catalog cache. On Spark 4.0+ Spark's command does
+ // that itself; on 3.x CometNativeWriteExec repeats it after the job commits.
withTable("comet_write_target", "comet_write_source") {
withNativeWriter {
sql("CREATE TABLE comet_write_source(id bigint, name string) USING parquet")
@@ -1340,6 +1340,125 @@ class CometParquetWriterSuite extends CometParquetWriterTestBase {
}
}
+ test("a native overwrite of a cached path recaches the data") {
+ withTempPath { dir =>
+ val target = new File(dir, "target").getAbsolutePath
+ val source = new File(dir, "source").getAbsolutePath
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark.range(0, 5).write.parquet(target)
+ spark.range(100, 103).write.parquet(source)
+ }
+ val cached = spark.read.parquet(target).cache()
+ try {
+ assert(cached.count() == 5)
+ withNativeWriter {
+ assertHasCometNativeWriteExec(
+ captureWritePlan(
+ spark.read.parquet(source).write.mode(SaveMode.Overwrite).parquet(target)))
+ }
+ // Spark recaches every cached plan that reads the output path once the write commits.
+ checkAnswer(cached, (100L until 103L).map(Row(_)))
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+
+ test("a native write rejects case-insensitively duplicate column names") {
+ withTempPath { dir =>
+ val source = new File(dir, "source").getAbsolutePath
+ val target = new File(dir, "target")
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark.range(3).selectExpr("id AS a", "id AS b").write.parquet(source)
+ }
+ withNativeWriter {
+ val (plan, e) = captureFailedWritePlan(
+ spark.read
+ .parquet(source)
+ .selectExpr("a", "b AS A")
+ .write
+ .mode(SaveMode.Overwrite)
+ .parquet(target.getAbsolutePath))
+ // Spark's own writer rejects these columns too, so a fallback would also pass the checks
+ // below without exercising the native writer's check.
+ assertHasCometNativeWriteExec(plan)
+ assert(e.isInstanceOf[AnalysisException], e)
+ assert(e.getMessage.contains("COLUMN_ALREADY_EXISTS"), e.getMessage)
+ }
+ assert(!target.exists(), "nothing may be written when the columns are rejected")
+ }
+ }
+
+ test("a native INSERT into a cached table recaches the table") {
+ withTable("comet_write_target", "comet_write_source") {
+ sql("CREATE TABLE comet_write_source(id bigint) USING parquet")
+ sql("CREATE TABLE comet_write_target(id bigint) USING parquet")
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ sql("INSERT INTO comet_write_source VALUES (1), (2)")
+ sql("INSERT INTO comet_write_target VALUES (0)")
+ }
+ sql("CACHE TABLE comet_write_target")
+ try {
+ checkAnswer(spark.table("comet_write_target"), Row(0L))
+ withNativeWriter {
+ assertHasCometNativeWriteExec(
+ captureWritePlan(
+ sql("INSERT INTO comet_write_target SELECT id FROM comet_write_source")))
+ }
+ checkAnswer(spark.table("comet_write_target"), Seq(Row(0L), Row(1L), Row(2L)))
+ } finally {
+ sql("UNCACHE TABLE IF EXISTS comet_write_target")
+ }
+ }
+ }
+
+ for {
+ autoUpdate <- Seq(true, false)
+ overwrite <- Seq(false, true)
+ } {
+ val insert = if (overwrite) "INSERT OVERWRITE" else "INSERT INTO"
+ test(s"native $insert maintains table stats with size autoUpdate=$autoUpdate") {
+ withTable("comet_write_target", "comet_write_source") {
+ withSQLConf(SQLConf.AUTO_SIZE_UPDATE_ENABLED.key -> autoUpdate.toString) {
+ sql("CREATE TABLE comet_write_source(id bigint) USING parquet")
+ sql("CREATE TABLE comet_write_target(id bigint) USING parquet")
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ sql("INSERT INTO comet_write_source SELECT id FROM range(100)")
+ sql("INSERT INTO comet_write_target VALUES (0)")
+ }
+ sql("ANALYZE TABLE comet_write_target COMPUTE STATISTICS NOSCAN")
+ val target = TableIdentifier("comet_write_target")
+ val catalog = spark.sessionState.catalog
+ val statsBefore = catalog.getTableMetadata(target).stats
+ assert(statsBefore.isDefined)
+
+ withNativeWriter {
+ assertHasCometNativeWriteExec(
+ captureWritePlan(
+ sql(s"$insert comet_write_target SELECT id FROM comet_write_source")))
+ }
+
+ // Mirrors CommandUtils.updateTableStats, which Spark runs after the write commits.
+ val statsAfter = catalog.getTableMetadata(target).stats
+ if (autoUpdate) {
+ val location = new Path(catalog.getTableMetadata(target).location)
+ val fs = location.getFileSystem(spark.sessionState.newHadoopConf())
+ val dataSize = fs
+ .listStatus(location)
+ .filterNot(f =>
+ f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))
+ .map(_.getLen)
+ .sum
+ assert(statsAfter.map(_.sizeInBytes) == Some(BigInt(dataSize)))
+ assert(statsAfter != statsBefore)
+ } else {
+ assert(statsAfter.isEmpty, s"stale stats were left in place: $statsAfter")
+ }
+ }
+ }
+ }
+ }
+
test("dynamic partition overwrite falls back to Spark") {
// A dynamic overwrite is a partitioned write, which both native writers decline - but the
// consequence of getting it wrong is silent data loss across untouched partitions, so assert
diff --git a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala
index 20caba36b1..605dcbbb3e 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala
@@ -21,6 +21,8 @@ package org.apache.comet.parquet
import java.util.concurrent.atomic.AtomicReference
+import scala.util.control.NonFatal
+
import org.apache.spark.CometListenerBusUtils
import org.apache.spark.sql.CometTestBase
import org.apache.spark.sql.comet.{CometNativeWriteExec, CometWriteFilesExec}
@@ -64,19 +66,37 @@ abstract class CometParquetWriterTestBase extends CometTestBase {
/** As above, for a write that names its own target (an `INSERT INTO`, for example). */
protected def captureWritePlan(writeOp: => Unit): SparkPlan = {
+ val (plan, failure) = captureWriteOutcome(writeOp)
+ failure.foreach(throw _)
+ plan.getOrElse(fail("Listener was not called - no execution plan captured"))
+ }
+
+ /**
+ * As above, for a write that is expected to fail. Returns the plan of the failed write, so a
+ * test can check that the failure came from the native writer rather than from Spark's.
+ */
+ protected def captureFailedWritePlan(writeOp: => Unit): (SparkPlan, Throwable) = {
+ val (plan, failure) = captureWriteOutcome(writeOp)
+ val error = failure.getOrElse(fail("Expected the write to fail, but it succeeded"))
+ (plan.getOrElse(fail("Listener was not called - no execution plan captured")), error)
+ }
+
+ private def captureWriteOutcome(writeOp: => Unit): (Option[SparkPlan], Option[Throwable]) = {
val capturedPlan = new AtomicReference[QueryExecution]()
+ def isWrite(funcName: String): Boolean = funcName == "save" || funcName.contains("command")
val listener = new org.apache.spark.sql.util.QueryExecutionListener {
override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
- if (funcName == "save" || funcName.contains("command")) {
+ if (isWrite(funcName)) {
capturedPlan.set(qe)
}
}
- override def onFailure(
- funcName: String,
- qe: QueryExecution,
- exception: Exception): Unit = {}
+ override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = {
+ if (isWrite(funcName)) {
+ capturedPlan.set(qe)
+ }
+ }
}
// Listener events are delivered asynchronously, so drain the bus before registering: an
@@ -85,12 +105,15 @@ abstract class CometParquetWriterTestBase extends CometTestBase {
spark.listenerManager.register(listener)
try {
- writeOp
+ val failure =
+ try {
+ writeOp
+ None
+ } catch {
+ case NonFatal(e) => Some(e)
+ }
CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
-
- val plan = capturedPlan.get()
- assert(plan != null, "Listener was not called - no execution plan captured")
- stripAQEPlan(plan.executedPlan)
+ (Option(capturedPlan.get()).map(qe => stripAQEPlan(qe.executedPlan)), failure)
} finally {
spark.listenerManager.unregister(listener)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org