You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ul...@apache.org on 2026/07/31 05:22:35 UTC
(spark) branch master updated: [SPARK-58444][SQL] Do not enable concurrent output writer when ordering already matched
This is an automated email from the ASF dual-hosted git repository.
ulysses-you pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8c38c50661cd [SPARK-58444][SQL] Do not enable concurrent output writer when ordering already matched
8c38c50661cd is described below
commit 8c38c50661cd2e786c5b71f5370e0017ad7771a6
Author: Xiduo You <ul...@gmail.com>
AuthorDate: Fri Jul 31 13:22:22 2026 +0800
[SPARK-58444][SQL] Do not enable concurrent output writer when ordering already matched
### What changes were proposed in this pull request?
When planned write is enabled, `FileFormatWriter` builds a `concurrentOutputWriterSpecFunc` that always wraps the plan with a sort and creates a concurrent output writer spec. This PR skips building the concurrent output writer spec when the output ordering already matches the required ordering, by making the func return `None` in that case.
### Why are the changes needed?
The concurrent output writer keeps one open writer per dynamic partition and falls back to the sort-based sequential writer once the number of open writers exceeds `spark.sql.maxConcurrentOutputFileWriters`. When the input is already sorted by the required ordering, enabling the concurrent writer is pure overhead: it may open many writers and then fall back to the sort-based writer anyway. Skipping it avoids this wasteful fall-back.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New UT in `V1WriteCommandSuite` that asserts, via the fall-back log, that the concurrent writer is disabled when the ordering already matches, with a control case (unsorted input) proving the assertion is discriminating.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57650 from ulysses-you/concurrent-writer.
Authored-by: Xiduo You <ul...@gmail.com>
Signed-off-by: Xiduo You <ul...@apache.org>
---
.../execution/datasources/FileFormatWriter.scala | 10 ++--
.../datasources/V1WriteCommandSuite.scala | 57 ++++++++++++++++++++++
2 files changed, 64 insertions(+), 3 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
index de642a3e850a..50d5dd28ab74 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
@@ -189,9 +189,13 @@ object FileFormatWriter extends Logging {
if (writeFilesOpt.isDefined) {
// build `WriteFilesSpec` for `WriteFiles`
- val concurrentOutputWriterSpecFunc = (plan: SparkPlan) => {
- val sortPlan = createSortPlan(plan, requiredOrdering, outputSpec)
- createConcurrentOutputWriterSpec(sparkSession, sortPlan, sortColumns)
+ val concurrentOutputWriterSpecFunc = if (orderingMatched) {
+ (_: SparkPlan) => None
+ } else {
+ (plan: SparkPlan) => {
+ val sortPlan = createSortPlan(plan, requiredOrdering, outputSpec)
+ createConcurrentOutputWriterSpec(sparkSession, sortPlan, sortColumns)
+ }
}
val writeSpec = WriteFilesSpec(
description = description,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
index bac194255581..9f7e23a5c6de 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
@@ -438,4 +438,61 @@ class V1WriteCommandSuite extends SharedSparkSession with V1WriteCommandSuiteBas
}
}
}
+
+ test("SPARK-58444: planned write should not enable concurrent writer when ordering " +
+ "already matched") {
+ // The concurrent output writer keeps one open writer per dynamic partition and falls back
+ // to the sort-based sequential writer once the number of open writers reaches
+ // `maxConcurrentOutputFileWriters`. When the input is already sorted by the required
+ // ordering, FileFormatWriter should NOT enable the concurrent writer at all, so that this
+ // wasteful fall-back never happens. This test pins that behavior via the fall-back log.
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.PLANNED_WRITE_ENABLED.key -> "true",
+ SQLConf.MAX_CONCURRENT_OUTPUT_FILE_WRITERS.key -> "2") {
+ withTable("t") {
+ sql("CREATE TABLE t(i INT, k STRING) USING PARQUET PARTITIONED BY (j INT)")
+
+ // t0 has 5 distinct values of `j` (i % 5), which is greater than the
+ // maxConcurrentOutputFileWriters threshold (2), so a single-task concurrent write over
+ // all partitions would trigger the fall-back. Use an int partition column to avoid the
+ // empty2null projection that a string partition column would add.
+ val fallbackMsg = "Fall back from concurrent writers"
+ val loggerName = classOf[DynamicPartitionDataConcurrentWriter].getName
+ val expected = spark.table("t0").select($"i", $"k", $"j")
+
+ // Case 1: input already sorted by the dynamic partition column, collapsed into a single
+ // task. The output ordering matches the required ordering, so the concurrent writer must
+ // be disabled -> no fall-back log.
+ val matchedAppender = new LogAppender("ordering matched, no concurrent writer")
+ withLogAppender(matchedAppender, Seq(loggerName)) {
+ expected.repartition(1).sortWithinPartitions("j")
+ .write.mode("overwrite").insertInto("t")
+ }
+ assert(FileFormatWriter.outputOrderingMatched,
+ "Expected the output ordering to match the required ordering.")
+ assert(!matchedAppender.loggingEvents.exists(
+ _.getMessage.getFormattedMessage.contains(fallbackMsg)),
+ "Concurrent writer should be disabled when ordering already matches, " +
+ "so no fall-back to the sort-based writer should happen.")
+ checkAnswer(spark.table("t"), expected)
+
+ // Case 2 (control): input NOT sorted. Ordering does not match, so the concurrent writer
+ // stays enabled and falls back once open writers exceed the threshold. This proves the
+ // log assertion above is actually discriminating and not vacuously true.
+ val unmatchedAppender = new LogAppender("ordering not matched, concurrent writer")
+ withLogAppender(unmatchedAppender, Seq(loggerName)) {
+ expected.repartition(1)
+ .write.mode("overwrite").insertInto("t")
+ }
+ assert(!FileFormatWriter.outputOrderingMatched,
+ "Expected the output ordering NOT to match the required ordering.")
+ assert(unmatchedAppender.loggingEvents.exists(
+ _.getMessage.getFormattedMessage.contains(fallbackMsg)),
+ "Concurrent writer should fall back to the sort-based writer when ordering " +
+ "does not match.")
+ checkAnswer(spark.table("t"), expected)
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org