fix: [branch-1.2] copy rows retained by identity broadcasts (#7260)

.github/workflows/pr_build_macos.yml

@@ -68,7 +68,7 @@ jobs:
   build-native:
     needs: lint
     name: Build Native Library (macOS)
-    runs-on: macos-14
+    runs-on: macos-15
     steps:
       - uses: actions/checkout@v7
 
@@ -143,7 +143,7 @@ jobs:
     if: ${{ !inputs.cache-refresh-only }}
     strategy:
       matrix:
-        os: [macos-14]
+        os: [macos-15]
 
         # macOS runs cover platform-specific concerns (native library loading, FFI,
         # threading, shuffle on Apple Silicon) that are largely independent of the

spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala

@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
-import org.apache.spark.sql.catalyst.plans.physical.Partitioning
+import org.apache.spark.sql.catalyst.plans.physical.{IdentityBroadcastMode, Partitioning}
 import org.apache.spark.sql.comet.util.{Utils => CometUtils}
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.execution.{CodegenSupport, ColumnarToRowTransition, SparkPlan, SQLExecution}
@@ -137,7 +137,13 @@ case class CometColumnarToRowExec(child: SparkPlan)
           numOutputRows)
 
         val mode = cometBroadcastExchange.get.mode
-        val relation = mode.transform(rows, Some(numOutputRows.value))
+        // Identity broadcasts retain rows, but both JVM conversion paths reuse the returned row.
+        // Hashed relations copy row bytes into their own storage while consuming the iterator.
+        val broadcastRows = mode match {
+          case IdentityBroadcastMode => rows.map(_.copy())
+          case _ => rows
+        }
+        val relation = mode.transform(broadcastRows, Some(numOutputRows.value))
         val broadcasted = sparkContext.broadcastInternal(relation, serializedOnly = true)
         val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
         SQLMetrics.postDriverMetricUpdates(sparkContext, executionId, metrics.values.toSeq)

spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala

@@ -2344,7 +2344,9 @@ class CometParquetWriterSuite extends CometParquetWriterTestBase {
             captureWritePlan(sql("INSERT INTO v2_file_target SELECT * FROM v2_file_source"))
           assertHasCometNativeWriteExec(plan)
           assert(RecordingFileSystem.created.size == 2, RecordingFileSystem.created)
-          assert(RecordingFileSystem.closed.asScala == RecordingFileSystem.created.asScala)
+          // The two tasks run concurrently, so their files close in any order.
+          assert(
+            RecordingFileSystem.closed.asScala.toSet == RecordingFileSystem.created.asScala.toSet)
           checkAnswer(spark.table("v2_file_target"), df)
         }
       }

spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala

@@ -28,12 +28,12 @@ import org.scalatest.PrivateMethodTester._
 
 import org.apache.logging.log4j.Level
 import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.FunctionIdentifier
+import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow}
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, ExpressionInfo, In, InSet, KnownFloatingPointNormalized, Literal, Not}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate, Final, Min, Partial, PartialMerge}
 import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero
 import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
-import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, RoundRobinPartitioning}
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, IdentityBroadcastMode, RoundRobinPartitioning}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.comet._
 import org.apache.spark.sql.comet.CometExec.nativeLeaves
@@ -97,6 +97,31 @@ class CometExecRuleSuite extends CometTestBase {
     CometExecRule(spark).apply(stripAQEPlan(plan))
   }
 
+  test("columnar-to-row identity broadcasts retain every build row") {
+    for (direct <- Seq(false, true)) {
+      withSQLConf(
+        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+        SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
+        CometConf.COMET_EXEC_RANGE_ENABLED.key -> "true",
+        CometConf.COMET_DIRECT_COLUMNAR_TO_ROW_ENABLED.key -> direct.toString,
+        CometConf.COMET_DIRECT_COLUMNAR_TO_ROW_MIN_BATCH_SIZE.key -> "1") {
+        val build = applyCometExecRule(
+          createSparkPlan(spark, "SELECT CAST(id AS INT) AS y FROM range(2024, 2028, 1, 2)"))
+        assert(build.isInstanceOf[CometNativeExec], build.treeString)
+        val mode = IdentityBroadcastMode
+        val original = BroadcastExchangeExec(mode, build)
+        val exchange = CometBroadcastExchangeExec(original, build.output, mode, build)
+        val rows = CometColumnarToRowExec(exchange)
+          .executeBroadcast[Array[InternalRow]]()
+          .value
+        // Two rows per partition expose row reuse in both JVM conversion paths.
+        withClue(s"direct=$direct: ") {
+          assert(rows.map(_.getInt(0)).sorted.toSeq == Seq(2024, 2025, 2026, 2027))
+        }
+      }
+    }
+  }
+
   /** Create a test data frame that is used in all tests */
   private def createTestDataFrame = {
     val testSchema = new StructType(