You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "viirya (via GitHub)" <gi...@apache.org> on 2026/10/09 16:56:11 UTC
[PR] fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer [datafusion-comet]
viirya opened a new pull request, #6821:
URL: https://github.com/apache/datafusion-comet/pull/6821
## Which issue does this PR close?
Closes #6820.
## Rationale for this change
On Spark 3.x `CometNativeWriteExec` stands in for the whole `InsertIntoHadoopFsRelationCommand`, but only runs the commit lifecycle. After a native write, cached DataFrames and cached tables over the output path return stale rows, the table relation's file listing is stale (the Spark 3.x side of #3521), and table stats are neither updated nor cleared. It also skips the command's duplicate-column check, so a write Spark rejects succeeds natively. Spark 4.0+ is not affected because Spark's command still runs there.
## What changes are included in this PR?
`CometNativeWriteExec` now does what the command it replaces does, using the command it already keeps as `originalPlan`:
- before writing, it rejects duplicate output column names with `SchemaUtils.checkColumnNameDuplication`
- after the job commits, it refreshes the command's `fileIndex`, calls `cacheManager.recacheByPath` for the output path, and calls `CommandUtils.updateTableStats` when the command has a catalog table
These steps run after the commit block, so a failure there does not abort a job that has already committed. As in Spark, they are skipped when `SaveMode.Ignore` finds an existing target. Partition metadata updates are not needed because partitioned writes are never converted.
## How are these changes tested?
New tests in `CometParquetWriterSuite`, which run on every Spark profile:
- a cached path is recached after a native overwrite
- a cached table is recached after a native `INSERT`
- table stats after native `INSERT INTO` / `INSERT OVERWRITE`, with `spark.sql.statistics.size.autoUpdate.enabled` on (stats equal the data size) and off (stale stats cleared)
- a write with case-insensitively duplicate column names fails with `COLUMN_ALREADY_EXISTS` and writes nothing
The existing "INSERT INTO ... SELECT is visible to subsequent reads" test no longer assumes Spark 4.0+.
On Spark 3.5 all of these fail without the fix and pass with it. `CometParquetWriterSuite`, `CometNativeWriteSuite` and `RevertNativeForTransitionHeavyStagesSuite` pass on Spark 3.4, 3.5 and 4.1. On 4.x the new tests pass through Spark's own command.
This pull request and its description were written by Isaac.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #6821:
URL: https://github.com/apache/datafusion-comet/pull/6821#discussion_r4233248376
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -1340,6 +1340,122 @@ 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 e = intercept[AnalysisException] {
Review Comment:
Good catch, thanks. Added `captureFailedWritePlan` to `CometParquetWriterTestBase`, which keeps the plan from `onFailure` as well, and the test now asserts the failed write carried a native write before checking the error. With the support level forced to `Unsupported` it now fails with "found 0" native write operators, and it still fails without the new check in `CometNativeWriteExec`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6821:
URL: https://github.com/apache/datafusion-comet/pull/6821#discussion_r4233109670
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -1340,6 +1340,122 @@ 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 e = intercept[AnalysisException] {
Review Comment:
This is the only new test that doesn't check the write actually went native. It does today. I reverted the `CometNativeWriteExec` change locally on 3.5 and this test fails with "no exception was thrown". But if this shape ever starts falling back, Spark's own command throws the same `COLUMN_ALREADY_EXISTS` and the test stays green without exercising the new check. That's the trap the comment in the `INSERT INTO ... SELECT` test at line 1333 warns about.
Could `captureWritePlan` also keep the plan from `onFailure`, so this test can call `assertHasCometNativeWriteExec` on the failed write? `CometIcebergWriteDetectionSuite.captureWritePlan` already does this with its `allowWriteFailure` flag.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #6821:
URL: https://github.com/apache/datafusion-comet/pull/6821#issuecomment-6100966176
Thanks @andygrove @sunchao !
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #6821:
URL: https://github.com/apache/datafusion-comet/pull/6821
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6821:
URL: https://github.com/apache/datafusion-comet/pull/6821#issuecomment-6086466382
This is really nice! Thanks @viirya. AI review coming up, just requesting a test change
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org