[VL] Gluten fails to satisfy the ordering required by dynamic partition column

[littlelittlewhite09]

Backend

VL (Velox)

Bug description

Example

set spark.sql.maxConcurrentOutputFileWriters = 0;
set spark.sql.sources.partitionOverwriteMode = DYNAMIC;

-- src table
DROP TABLE IF EXISTS sql_migration.tmp_gluten_repro_src;
CREATE TABLE sql_migration.tmp_gluten_repro_src (k string, v string, day string) STORED AS ORC;

INSERT INTO sql_migration.tmp_gluten_repro_src VALUES
  ('k1','v01','2026-06-01'), ('k2','v02','2026-06-02'),
  ('k3','v03','2026-06-01'), ('k4','v04','2026-06-02'),
  ('k5','v05','2026-06-01'), ('k6','v06','2026-06-02'),
  ('k7','v07','2026-06-01'), ('k8','v08','2026-06-02'),
  ('k9','v09','2026-06-01'), ('k10','v10','2026-06-02');

-- target table
DROP TABLE IF EXISTS sql_migration.tmp_gluten_repro_tgt;
CREATE TABLE sql_migration.tmp_gluten_repro_tgt (k string, m string)
  PARTITIONED BY (day string) STORED AS ORC;

-- insert into partitions dynamically
INSERT OVERWRITE TABLE sql_migration.tmp_gluten_repro_tgt PARTITION(day)
SELECT k, max(v) AS m, day
FROM sql_migration.tmp_gluten_repro_src
GROUP BY day, k;

Exception:
Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: /xx/xx/sql_migration/tmp_gluten_repro_tgt/.spark-staging-46b3ca99-cab6-4c7e-9e12-dd6a8ad81bc9/_temporary/0/_temporary/attempt_202607081201073995302620617304707_0039_m_000000_2127/day=2026-06-02/part-00000-46b3ca99-cab6-4c7e-9e12-dd6a8ad81bc9.c000.snappy.orc for client 10.69.52.150 already exists
	at org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.startFile(FSDirWriteFileOp.java:386)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFileInt(FSNamesystem.java:2464)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFile(FSNamesystem.java:2361)
	at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.create(NameNodeRpcServer.java:774)
	at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.create(ClientNamenodeProtocolServerSideTranslatorPB.java:462)
	at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
	at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:522)
	at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1025)
	at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:931)
	at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:870)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1730)
	at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2763)

	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
	at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499)
	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:480)
	at org.apache.hadoop.ipc.RemoteException.instantiateException(RemoteException.java:121)
	at org.apache.hadoop.ipc.RemoteException.unwrapRemoteException(RemoteException.java:88)
	at org.apache.hadoop.hdfs.DFSOutputStream.newStreamForCreate(DFSOutputStream.java:284)
	at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1219)
	at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1198)
	at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1136)
	at org.apache.hadoop.hdfs.DistributedFileSystem$8.doCall(DistributedFileSystem.java:531)
	at org.apache.hadoop.hdfs.DistributedFileSystem$8.doCall(DistributedFileSystem.java:528)
	at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
	at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:542)
	at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:469)
	at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1123)
	at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1103)
	at org.apache.orc.impl.PhysicalFsWriter.<init>(PhysicalFsWriter.java:94)
	at org.apache.orc.impl.WriterImpl.<init>(WriterImpl.java:230)
	at org.apache.orc.OrcFile.createWriter(OrcFile.java:1040)
	at org.apache.spark.sql.execution.datasources.orc.OrcOutputWriter.<init>(OrcOutputWriter.scala:50)
	at org.apache.spark.sql.execution.datasources.orc.OrcFileFormat$$anon$1.newInstance(OrcFileFormat.scala:92)
	at org.apache.spark.sql.execution.datasources.BaseDynamicPartitionDataWriter.renewCurrentWriter(FileFormatDataWriter.scala:298)
	at org.apache.spark.sql.execution.datasources.DynamicPartitionDataSingleWriter.write(FileFormatDataWriter.scala:365)
	at org.apache.spark.sql.execution.datasources.FileFormatDataWriter.writeWithMetrics(FileFormatDataWriter.scala:85)
	at org.apache.spark.sql.execution.datasources.FileFormatDataWriter.writeWithIterator(FileFormatDataWriter.scala:92)
	at org.apache.spark.sql.execution.datasources.FileFormatWriter$.$anonfun$executeTask$2(FileFormatWriter.scala:481)
	at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1398)
	at org.apache.spark.sql.execution.datasources.FileFormatWriter$.executeTask(FileFormatWriter.scala:488)
	... 17 more

Advanced Description

Gluten plan:
Execute InsertIntoHadoopFsRelationCommand (16)
   +- WriteFiles (15)
      +- VeloxColumnarToRow (14)
         +- ^ ProjectExecTransformer (12)
            +- ^ RegularHashAggregateExecTransformer (11)
               +- ^ InputIteratorTransformer (10)
                  +- AQEShuffleRead (8)
                     +- ShuffleQueryStage (7), Statistics(sizeInBytes=271.0 B, rowCount=10)
                        +- ColumnarExchange (6)
                           +- VeloxResizeBatches (5)
                              +- ^ ProjectExecTransformer (3)
                                 +- ^ FlushableHashAggregateExecTransformer (2)
                                    +- ^ FileSourceScanExecTransformer orc spark_catalog.sql_migration.tmp_gluten_repro_src (1)

initial plan:
   Execute InsertIntoHadoopFsRelationCommand (24)
   +- WriteFiles (23)
      +- SortAggregate (22)
         +- Sort (21)
            +- Exchange (20)
               +- SortAggregate (19)
                  +- Sort (18)
                     +- Scan orc spark_catalog.sql_migration.tmp_gluten_repro_src (17)

Gluten transforms SortAggregate into HashAggregate，it destroys the ordering requirement for dynamic partitions writing. When dynamicPartitionDataSingleWriter is adopted, Task may process data that is not sorted by the partition column, which leading the FileAlreadyExistsException. Actually, when spark.sql.maxConcurrentOutputFileWriters is greater than the num of dynamic partitions, dynamicPartitionDataConcurrentWriter can be adopted, and the problem can be avoided. But, I do not think it is a graceful way.

Gluten version

1.6.0

Spark version

3.5.8

Spark configurations

No response

System information

No response

Relevant logs

[littlelittlewhite09]

This issue may be the same as issue #9789

[littlelittlewhite09]

@philo-he Can you help me confirm this issue？Thank you so much.

[philo-he]

@littlelittlewhite09, Thanks for raising this. Gluten already has an EnsureLocalSortRequirements rule that restores the local sort needed to satisfy operators' requiredChildOrdering. However, in your case it looks like the file write doesn't expose its ordering requirement through requiredChildOrdering.

One more thing: I notice the native write isn't used here. Was that a real fallback, or did you disable native write? Have you tried with native write enabled? That would help narrow down the fix.

gluten/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala

Line 33 in 5d9bced

object EnsureLocalSortRequirements extends Rule[SparkPlan] {

[littlelittlewhite09]

@philo-he Thank you for your replying. I pick the Gluten v1.6.0 tag to run my case. And I found the code below may cause the issue:

gluten/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala

Lines 217 to 218 in 8971898

	case plan: SortAggregateExec =>
	HashAggregateExecBaseTransformer.from(plan)

HashAggregateExecBaseTransformer may fail to satisfy the ordering.
Native write is not used here, because the table is orc format, which is not support to be native.
Meanwhile, I notice that #9473 has modified this code, and it may work for me. I will try it later.

[littlelittlewhite09]

@philo-he Thank you for your replying. I pick the Gluten v1.6.0 tag to run my case. And I found the code below may cause the issue:

gluten/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala

Lines 217 to 218 in 8971898

	case plan: SortAggregateExec =>
	HashAggregateExecBaseTransformer.from(plan)

HashAggregateExecBaseTransformer may fail to satisfy the ordering.
Native write is not used here, because the table is orc format, which is not support to be native.
Meanwhile, I notice that #9473 has modified this code, and it may work for me. I will try it later.

This patch does not work for me.

[philo-he]

@littlelittlewhite09, Could you check whether this issue still reproduces on Gluten's main branch? If so, I can try to fix it.