[25.10] Add missing handling for the 0 input partition count in computing partition count in GpuOptimizeWriteExchangeExec

delta-lake/common/src/main/delta-33x/scala/org/apache/spark/sql/delta/rapids/GpuOptimizeWriteExchangeExec.scala

@@ -108,27 +108,31 @@ case class GpuOptimizeWriteExchangeExec(
 
   @transient lazy val inputRDD: RDD[ColumnarBatch] = child.executeColumnar()
 
+  @transient private lazy val childNumPartitions = inputRDD.getNumPartitions
+
   @transient lazy val mapOutputStatisticsFuture: Future[MapOutputStatistics] = {
-    if (inputRDD.getNumPartitions == 0) {
+    if (childNumPartitions == 0) {
       Future.successful(null)
     } else {
       sparkContext.submitMapStage(shuffleDependency)
     }
   }
 
-  private lazy val childNumPartitions = inputRDD.getNumPartitions
-
-  private lazy val actualNumPartitions: Int = {
-    val targetShuffleBlocks = conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
-    math.min(
-      math.max(targetShuffleBlocks / childNumPartitions, 1),
-      conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+  @transient private lazy val actualNumPartitions: Int = {
+    if (childNumPartitions == 0) {
+      0
+    } else {
+      val targetShuffleBlocks = conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+      math.min(
+        math.max(targetShuffleBlocks / childNumPartitions, 1),
+        conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+    }
   }
 
   // The actual partitioning to use for the shuffle exchange. The input partition count can be
   // adjusted based on the number of partitions in the input RDD and the target number of shuffle
   // blocks.
-  private lazy val actualPartitioning: GpuPartitioning = partitioning match {
+  @transient private lazy val actualPartitioning: GpuPartitioning = partitioning match {
     // Currently only hash and round-robin partitioning are supported.
     // See DeltaShufflePartitionsUtil.partitioningForRebalance() for more details.
     case p: GpuHashPartitioning => p.copy(numPartitions = actualNumPartitions)

integration_tests/src/main/python/delta_lake_write_test.py

@@ -1012,6 +1012,27 @@ def setup_tables(spark):
             conf=conf)
     do_test_optimize_write(spark_tmp_path, aqe_enabled, do_write, num_chunks)
 
+@allow_non_gpu(*delta_meta_allow)
+@delta_lake
+@ignore_order
+@pytest.mark.skipif(not is_databricks_runtime() and is_before_spark_353(), reason="Delta Lake optimized writes are not supported before Spark 3.5.3 on Apache Spark")
+def test_delta_write_optimized_empty_output(spark_tmp_path):
+    num_chunks = 20
+    data_path = spark_tmp_path + "/DELTA_DATA"
+    gen = IntegerGen(nullable=False)
+    confs=copy_and_update(delta_writes_enabled_conf, {
+        "spark.databricks.delta.optimizeWrite.enabled" : "true"
+    })
+
+    assert_gpu_and_cpu_writes_are_equal_collect(
+        # filter everything out in the generated data
+        lambda spark, path: unary_op_df(spark, gen) \
+            .filter("a is null") \
+            .repartition(num_chunks).write.format("delta").save(path),
+        lambda spark, path: spark.read.format("delta").load(path),
+        data_path,
+        conf=confs)
+
 @allow_non_gpu(*delta_meta_allow)
 @delta_lake
 @ignore_order(local=True)