diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 317653f1c50..cb996b4df05 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.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 diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala index d3eacddf5e0..3a60be3f788 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala +++ b/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) diff --git a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala index e3a22569a24..af57ff0efac 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala +++ b/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) } } diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala index 62d968c95d3..96a49adbe88 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/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(