From 7ecfbea71105e0816851f5d7ed4b04772620502f Mon Sep 17 00:00:00 2001 From: zhangyizhong Date: Tue, 8 Sep 2026 23:13:03 +0800 Subject: [PATCH] use NativePartition wrapper and @transient rddPartitions to reduce serialization overhead --- .../auron/plan/NativeShuffleExchangeExec.scala | 2 +- .../org/apache/spark/sql/auron/NativeRDD.scala | 18 ++++++++++++++++-- .../execution/auron/plan/NativeAggBase.scala | 10 ++++++++-- .../plan/NativeBroadcastExchangeBase.scala | 6 ++++-- .../auron/plan/NativeBroadcastJoinBase.scala | 12 +++++++++--- .../auron/plan/NativeCollectLimitBase.scala | 14 +++++++++++--- .../auron/plan/NativeExpandBase.scala | 9 +++++++-- .../auron/plan/NativeFileSourceScanBase.scala | 1 + .../auron/plan/NativeFilterBase.scala | 9 +++++++-- .../auron/plan/NativeGenerateBase.scala | 9 +++++++-- .../auron/plan/NativeGlobalLimitBase.scala | 9 +++++++-- .../auron/plan/NativeLocalLimitBase.scala | 9 +++++++-- .../auron/plan/NativeOrcSinkBase.scala | 9 +++++++-- .../NativeParquetInsertIntoHiveTableBase.scala | 2 +- .../auron/plan/NativeParquetSinkBase.scala | 9 +++++++-- .../auron/plan/NativeProjectBase.scala | 9 +++++++-- .../auron/plan/NativeRenameColumnsBase.scala | 10 ++++++++-- .../auron/plan/NativeShuffleExchangeBase.scala | 15 ++++++++++----- .../plan/NativeShuffledHashJoinBase.scala | 14 +++++++++++--- .../execution/auron/plan/NativeSortBase.scala | 9 +++++++-- .../auron/plan/NativeSortMergeJoinBase.scala | 14 +++++++++++--- .../auron/plan/NativeTakeOrderedBase.scala | 12 ++++++++---- .../auron/plan/NativeWindowBase.scala | 9 +++++++-- .../auron/shuffle/AuronShuffleDependency.scala | 5 +++-- .../auron/plan/NativeHiveTableScanBase.scala | 2 +- 25 files changed, 173 insertions(+), 54 deletions(-) diff --git a/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeExec.scala b/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeExec.scala index 090a22d79..5d07d8799 100644 --- a/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeExec.scala +++ b/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeExec.scala @@ -148,7 +148,7 @@ case class NativeShuffleExchangeExec( "Input iterator must be empty (SPARK-44605: adapt to Spark 4+ ShuffleWriteProcessor API changes)") val rdd = dep.asInstanceOf[AuronShuffleDependency[_, _, _]].inputRdd - val partition = rdd.partitions(mapIndex) + val partition = dep.asInstanceOf[AuronShuffleDependency[_, _, _]].rddPartitions(mapIndex) internalWrite(rdd, dep, mapId, context, partition) } diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeRDD.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeRDD.scala index f2629241a..7184dbf83 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeRDD.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeRDD.scala @@ -36,7 +36,7 @@ import org.apache.auron.util.SparkVersionUtil class NativeRDD( @transient private val rddSparkContext: SparkContext, val metrics: SparkMetricNode, - private val rddPartitions: Array[Partition], + @transient private val rddPartitions: Array[Partition], private val rddPartitioner: Option[Partitioner], private val rddDependencies: Seq[Dependency[_]], private val rddShuffleReadFull: Boolean, @@ -46,6 +46,8 @@ class NativeRDD( with Logging with Serializable { + private val numRddPartitions: Int = rddPartitions.length + // use serializable wrapper to avoid serializing nativePlan val nativePlanWrapper = new NativePlanWrapper(nativePlan) @@ -60,7 +62,10 @@ class NativeRDD( def isShuffleReadFull: Boolean = Shims.get.getRDDShuffleReadFull(this) Shims.get.setRDDShuffleReadFull(this, rddShuffleReadFull) - override protected def getPartitions: Array[Partition] = rddPartitions + // Spark 4+ needs the number of partitions + override protected def getPartitions: Array[Partition] = + if (rddPartitions != null) rddPartitions + else Array.tabulate(numRddPartitions)(i => new Partition { override def index: Int = i }) override protected def getDependencies: Seq[Dependency[_]] = rddDependencies override val partitioner: Option[Partitioner] = rddPartitioner @@ -101,6 +106,15 @@ class EmptyNativeRDD(@transient private val rddSparkContext: SparkContext) } +case class NativePartition[P](override val index: Int, payload: P) extends Partition {} + +object NativePartition { + def unwrap(partition: Partition): Partition = partition match { + case np: NativePartition[_] => np.payload.asInstanceOf[Partition] + case other => other + } +} + class NativePlanWrapper(var p: (Partition, TaskContext) => PhysicalPlanNode) extends Serializable { def plan(split: Partition, context: TaskContext): PhysicalPlanNode = { diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeAggBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeAggBase.scala index 755fb6466..42574c10f 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeAggBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeAggBase.scala @@ -21,9 +21,11 @@ import scala.collection.immutable.SortedMap import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.internal.Logging import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.auron.Shims @@ -174,18 +176,22 @@ abstract class NativeAggBase( val nativeAggrModes = this.nativeAggrModes val nativeAggrs = this.nativeAggrs val nativeGroupingExprs = this.nativeGroupingExprs + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { + val inputPartition = NativePartition.unwrap(partition) lazy val inputPlan = - inputRDD.nativePlan(inputRDD.partitions(partition.index), taskContext) + inputRDD.nativePlan(inputPartition, taskContext) pb.PhysicalPlanNode .newBuilder() .setAgg( diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastExchangeBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastExchangeBase.scala index b775b6c06..25e458e36 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastExchangeBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastExchangeBase.scala @@ -66,7 +66,9 @@ import org.apache.auron.{protobuf => pb, sparkver} import org.apache.auron.jni.JniBridge import org.apache.auron.metric.SparkMetricNode -abstract class NativeBroadcastExchangeBase(mode: BroadcastMode, override val child: SparkPlan) +abstract class NativeBroadcastExchangeBase( + @transient mode: BroadcastMode, + override val child: SparkPlan) extends BroadcastExchangeLike with NativeSupports { @@ -245,7 +247,7 @@ abstract class NativeBroadcastExchangeBase(mode: BroadcastMode, override val chi metrics("dataSize") += byteArray.length }) - val input = inputRDD.nativePlan(inputRDD.partitions(split.index), context) + val input = inputRDD.nativePlan(split, context) val nativeIpcWriterExec = pb.PhysicalPlanNode .newBuilder() .setIpcWriter( diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastJoinBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastJoinBase.scala index 82a50ca16..19c335809 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastJoinBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeBroadcastJoinBase.scala @@ -23,6 +23,7 @@ import org.apache.spark.OneToOneDependency import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.auron.Shims @@ -153,10 +154,14 @@ abstract class NativeBroadcastJoinBase( Seq(FullOuter, LeftOuter, LeftSemi, LeftAnti).contains(joinType) }) + val nativePartitions = probedRDD.partitions.map { p => + NativePartition[Partition](p.index, p) + } + new NativeRDD( sparkContext, nativeMetrics, - probedRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = probedRDD.partitioner, rddDependencies = new OneToOneDependency(probedRDD) :: Nil, probedShuffleReadFull, @@ -164,14 +169,15 @@ abstract class NativeBroadcastJoinBase( val partition0 = new Partition() { override def index: Int = 0 } + val probedPartition = NativePartition.unwrap(partition) val (leftChild, rightChild) = broadcastSide match { case JoinBuildLeft => ( leftRDD.nativePlan(partition0, context), - rightRDD.nativePlan(rightRDD.partitions(partition.index), context)) + rightRDD.nativePlan(probedPartition, context)) case JoinBuildRight => ( - leftRDD.nativePlan(leftRDD.partitions(partition.index), context), + leftRDD.nativePlan(probedPartition, context), rightRDD.nativePlan(partition0, context)) } val cachedBuildHashMapId = s"bhm_stage${context.stageId}_rdd${builtRDD.id}" diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeCollectLimitBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeCollectLimitBase.scala index 4a226fc65..ab9b2ebd5 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeCollectLimitBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeCollectLimitBase.scala @@ -19,7 +19,12 @@ package org.apache.spark.sql.execution.auron.plan import scala.collection.mutable import org.apache.spark.OneToOneDependency -import org.apache.spark.sql.auron.{NativeHelper, NativeRDD, NativeSupports, Shims} +import org.apache.spark.Partition +import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition +import org.apache.spark.sql.auron.NativeRDD +import org.apache.spark.sql.auron.NativeSupports +import org.apache.spark.sql.auron.Shims import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.plans.physical.{SinglePartition, UnknownPartitioning} @@ -56,16 +61,19 @@ abstract class NativeCollectLimitBase(limit: Int, offset: Int, override val chil // merge all LocalLimit child partitions into a single partition val shuffled = Shims.get.createNativeShuffleExchangeExec(SinglePartition, partial) val singlePartitionRDD = NativeHelper.executeNative(shuffled) + val nativePartitions = singlePartitionRDD.partitions.map { p => + NativePartition[Partition](p.index, p) + } new NativeRDD( sparkContext, SparkMetricNode(metrics, singlePartitionRDD.metrics :: Nil), - singlePartitionRDD.partitions, + rddPartitions = nativePartitions.toArray, singlePartitionRDD.partitioner, new OneToOneDependency(singlePartitionRDD) :: Nil, rddShuffleReadFull = false, (partition, taskContext) => { - val inputPartition = singlePartitionRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeLimitExec = LimitExecNode .newBuilder() .setInput(singlePartitionRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeExpandBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeExpandBase.scala index 8eaa86f60..96c0e8a75 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeExpandBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeExpandBase.scala @@ -20,8 +20,10 @@ import scala.collection.immutable.SortedMap import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, SortOrder} @@ -75,16 +77,19 @@ abstract class NativeExpandBase( val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) val nativeSchema = this.nativeSchema val nativeProjections = this.nativeProjections + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeExpandExec = ExpandExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFileSourceScanBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFileSourceScanBase.scala index 84e886060..eb9919060 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFileSourceScanBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFileSourceScanBase.scala @@ -55,6 +55,7 @@ abstract class NativeFileSourceScanBase(basedFileScan: FileSourceScanExec) override val output: Seq[Attribute] = basedFileScan.output override val outputPartitioning: Partitioning = basedFileScan.outputPartitioning + @transient protected val inputFileScanRDD: FileScanRDD = { MethodUtils.invokeMethod(basedFileScan, true, "prepare") MethodUtils.invokeMethod(basedFileScan, true, "waitForSubqueries") diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFilterBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFilterBase.scala index cf06c2d86..14ab87a78 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFilterBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeFilterBase.scala @@ -21,8 +21,10 @@ import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.And @@ -91,15 +93,18 @@ abstract class NativeFilterBase(condition: Expression, override val child: Spark val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) val nativeFilterExprs = this.nativeFilterExprs + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeFilterExec = FilterExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGenerateBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGenerateBase.scala index 202645b6a..3d04d6f00 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGenerateBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGenerateBase.scala @@ -22,8 +22,10 @@ import scala.jdk.CollectionConverters._ import com.google.protobuf.ByteString import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Attribute @@ -134,16 +136,19 @@ abstract class NativeGenerateBase( val nativeGenerator = this.nativeGenerator val nativeGeneratorOutput = this.nativeGeneratorOutput val nativeRequiredChildOutput = this.nativeRequiredChildOutput + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeGenerateExec = pb.GenerateExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGlobalLimitBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGlobalLimitBase.scala index cf1dc3413..34c954dcd 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGlobalLimitBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeGlobalLimitBase.scala @@ -19,7 +19,9 @@ package org.apache.spark.sql.execution.auron.plan import scala.collection.immutable.SortedMap import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Attribute @@ -52,16 +54,19 @@ abstract class NativeGlobalLimitBase(limit: Int, offset: Int, override val child override def doExecuteNative(): NativeRDD = { val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - inputRDD.partitions, + nativePartitions.toArray, inputRDD.partitioner, new OneToOneDependency(inputRDD) :: Nil, rddShuffleReadFull = false, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeLimitExec = LimitExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeLocalLimitBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeLocalLimitBase.scala index a54e1f3ff..0df717a10 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeLocalLimitBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeLocalLimitBase.scala @@ -19,7 +19,9 @@ package org.apache.spark.sql.execution.auron.plan import scala.collection.immutable.SortedMap import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Attribute @@ -49,16 +51,19 @@ abstract class NativeLocalLimitBase(limit: Int, override val child: SparkPlan) override def doExecuteNative(): NativeRDD = { val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, rddShuffleReadFull = false, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeLimitExec = LimitExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeOrcSinkBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeOrcSinkBase.scala index 2523b28e7..a966f1e9d 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeOrcSinkBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeOrcSinkBase.scala @@ -29,9 +29,11 @@ import org.apache.hadoop.hive.ql.plan.TableDesc import org.apache.hadoop.mapred.JobConf import org.apache.hadoop.mapreduce.Job import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.SparkSession import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.catalog.CatalogTable @@ -85,10 +87,13 @@ abstract class NativeOrcSinkBase( val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) val nativeDependencies = new OneToOneDependency(inputRDD) :: Nil + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkSession.sparkContext, nativeMetrics, - inputRDD.partitions, + nativePartitions.toArray, inputRDD.partitioner, nativeDependencies, inputRDD.isShuffleReadFull, @@ -116,7 +121,7 @@ abstract class NativeOrcSinkBase( .setValue(entry.getValue) .build()) - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val orcSink = OrcSinkExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, context)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetInsertIntoHiveTableBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetInsertIntoHiveTableBase.scala index 32dfa4c0d..e434b5c46 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetInsertIntoHiveTableBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetInsertIntoHiveTableBase.scala @@ -56,7 +56,7 @@ import org.apache.spark.sql.execution.metric.SQLMetrics import org.apache.spark.sql.hive.execution.InsertIntoHiveTable abstract class NativeParquetInsertIntoHiveTableBase( - cmd: InsertIntoHiveTable, + @transient cmd: InsertIntoHiveTable, override val child: SparkPlan) extends UnaryExecNode with NativeSupports { diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetSinkBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetSinkBase.scala index 689b46e60..9b74325df 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetSinkBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetSinkBase.scala @@ -37,8 +37,10 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils import org.apache.hadoop.mapred.JobConf import org.apache.hadoop.mapreduce.Job import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.SparkSession import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.catalog.CatalogTable @@ -91,10 +93,13 @@ abstract class NativeParquetSinkBase( val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) val nativeDependencies = new OneToOneDependency(inputRDD) :: Nil + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkSession.sparkContext, nativeMetrics, - inputRDD.partitions, + rddPartitions = nativePartitions.toArray, inputRDD.partitioner, nativeDependencies, inputRDD.isShuffleReadFull, @@ -146,7 +151,7 @@ abstract class NativeParquetSinkBase( .setValue(entry.getValue) .build()) - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val parquetSink = ParquetSinkExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, context)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeProjectBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeProjectBase.scala index 04ae0b55b..ea3dd3a90 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeProjectBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeProjectBase.scala @@ -21,8 +21,10 @@ import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.analysis.ResolvedStar @@ -71,16 +73,19 @@ abstract class NativeProjectBase(projectList: Seq[NamedExpression], override val val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) val nativeProject = this.nativeProject + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeProjectExec = nativeProject.toBuilder .setInput(inputRDD.nativePlan(inputPartition, taskContext)) .build() diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeRenameColumnsBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeRenameColumnsBase.scala index 3d2b21fc9..aef7256c1 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeRenameColumnsBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeRenameColumnsBase.scala @@ -19,7 +19,9 @@ package org.apache.spark.sql.execution.auron.plan import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Attribute @@ -54,16 +56,20 @@ abstract class NativeRenameColumnsBase( override def doExecuteNative(): NativeRDD = { val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPlan = inputRDD.nativePlan(inputRDD.partitions(partition.index), taskContext) + val inputPartition = NativePartition.unwrap(partition) + val inputPlan = inputRDD.nativePlan(inputPartition, taskContext) buildRenameColumnsExec(inputPlan, renamedColumnNames) }, friendlyName = "NativeRDD.RenameColumns") diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeBase.scala index e8ce4dfdf..c46aab6d3 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffleExchangeBase.scala @@ -25,12 +25,13 @@ import scala.jdk.CollectionConverters._ import scala.reflect.ClassTag import scala.util.hashing.byteswap32 -import org.apache.spark.{OneToOneDependency, Partitioner, RangePartitioner, ShuffleDependency, SparkEnv, TaskContext} +import org.apache.spark.{OneToOneDependency, Partition, Partitioner, RangePartitioner, ShuffleDependency, SparkEnv, TaskContext} import org.apache.spark.rdd.{PartitionPruningRDD, RDD} import org.apache.spark.serializer.Serializer import org.apache.spark.shuffle.ShuffleWriteProcessor import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.auron.Shims @@ -125,6 +126,7 @@ abstract class NativeShuffleExchangeBase( override def doExecuteNative(): NativeRDD = { val shuffleHandle = shuffleDependency.shuffleHandle val rdd = doExecuteNonNative() + val nativeSchema = this.nativeSchema val nativeMetrics = SparkMetricNode( Map(), @@ -148,7 +150,6 @@ abstract class NativeShuffleExchangeBase( (partition, taskContext) => { val shuffleReadMetrics = taskContext.taskMetrics().createTempShuffleReadMetrics() val metricReporter = new SQLShuffleReadMetricsReporter(shuffleReadMetrics, metrics) - val nativeSchema = this.nativeSchema // store fetch iterator in jni resource before native compute val jniResourceId = s"NativeShuffleReadExec:${UUID.randomUUID().toString}" @@ -244,18 +245,21 @@ abstract class NativeShuffleExchangeBase( case _ => null } + val nativeInputPartitions = nativeInputRDD.partitions.map { p => + NativePartition[Partition](p.index, p) + } val nativeHashExprs = this.nativeHashExprs val nativeSortExecNode = this.nativeSortExecNode val nativeShuffleRDD = new NativeRDD( nativeInputRDD.sparkContext, nativeMetrics, - nativeInputRDD.partitions, + rddPartitions = nativeInputPartitions.toArray, nativeInputRDD.partitioner, new OneToOneDependency(nativeInputRDD) :: Nil, nativeInputRDD.isShuffleReadFull, (partition, taskContext) => { - val nativeInputPartition = nativeInputRDD.partitions(partition.index) + val nativeInputPartition = NativePartition.unwrap(partition) val repartitionBuilder = PhysicalRepartition.newBuilder() val nativeOutputPartitioning = outputPartitioning match { case SinglePartition => @@ -305,7 +309,8 @@ abstract class NativeShuffleExchangeBase( override def getPartition(key: Any): Int = key.asInstanceOf[Int] }, - schema = Util.getSchema(outputAttributes, useExprId = false)) + schema = Util.getSchema(outputAttributes, useExprId = false), + rddPartitions = nativeInputPartitions.toArray) metrics("numPartitions").set(numPartitionsRest) val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) SQLMetrics.postDriverMetricUpdates(sparkContext, executionId, metrics("numPartitions") :: Nil) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala index 908b3ce0b..deb826f73 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala @@ -20,8 +20,10 @@ import scala.collection.immutable.SortedMap import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.auron.join.JoinBuildSides.{JoinBuildLeft, JoinBuildRight, JoinBuildSide} @@ -109,6 +111,7 @@ abstract class NativeShuffledHashJoinBase( val nativeJoinType = this.nativeJoinType val nativeJoinFilter = this.nativeJoinFilter val nativeBuildSide = this.nativeBuildSide + val nativeSchema = this.nativeSchema val (partitions, partitioner) = if (joinType != RightOuter) { (leftRDD.partitions, leftRDD.partitioner) @@ -116,19 +119,24 @@ abstract class NativeShuffledHashJoinBase( (rightRDD.partitions, rightRDD.partitioner) } val dependencies = Seq(new OneToOneDependency(leftRDD), new OneToOneDependency(rightRDD)) + val nativePartitions = partitions.map { p => + NativePartition[(Partition, Partition)]( + p.index, + (leftRDD.partitions(p.index), rightRDD.partitions(p.index))) + } new NativeRDD( sparkContext, nativeMetrics, - partitions, + rddPartitions = nativePartitions.toArray, partitioner, dependencies, leftRDD.isShuffleReadFull && rightRDD.isShuffleReadFull, (partition, taskContext) => { - val leftPartition = leftRDD.partitions(partition.index) + val (leftPartition, rightPartition) = + partition.asInstanceOf[NativePartition[(Partition, Partition)]].payload val leftChild = leftRDD.nativePlan(leftPartition, taskContext) - val rightPartition = rightRDD.partitions(partition.index) val rightChild = rightRDD.nativePlan(rightPartition, taskContext) val hashJoinExec = pb.HashJoinExecNode diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortBase.scala index bc6672313..bc030e98d 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortBase.scala @@ -20,8 +20,10 @@ import scala.collection.immutable.SortedMap import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Ascending @@ -98,16 +100,19 @@ abstract class NativeSortBase( val inputRDD = NativeHelper.executeNative(child) val nativeMetrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil) val nativeSortExprs = this.nativeSortExprs + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeSortExec = SortExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortMergeJoinBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortMergeJoinBase.scala index 566f63d11..9d71b1dca 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortMergeJoinBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeSortMergeJoinBase.scala @@ -20,8 +20,10 @@ import scala.collection.immutable.SortedMap import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Ascending @@ -113,6 +115,7 @@ abstract class NativeSortMergeJoinBase( val nativeSortOptions = this.nativeSortOptions val nativeJoinOn = this.nativeJoinOn val nativeJoinType = this.nativeJoinType + val nativeSchema = this.nativeSchema val nativeJoinFilter = this.nativeJoinFilter val (partitions, partitioner) = if (joinType != RightOuter) { @@ -121,6 +124,11 @@ abstract class NativeSortMergeJoinBase( (rightRDD.partitions, rightRDD.partitioner) } val dependencies = Seq(new OneToOneDependency(leftRDD), new OneToOneDependency(rightRDD)) + val nativePartitions = partitions.map { p => + NativePartition[(Partition, Partition)]( + p.index, + (leftRDD.partitions(p.index), rightRDD.partitions(p.index))) + } val isShuffleReadFull = joinType match { case _: InnerLike => logInfo("SortMergeJoin Inner mark shuffleReadFull = false") @@ -137,15 +145,15 @@ abstract class NativeSortMergeJoinBase( new NativeRDD( sparkContext, nativeMetrics, - partitions, + rddPartitions = nativePartitions.toArray, partitioner, dependencies, isShuffleReadFull, (partition, taskContext) => { - val leftPartition = leftRDD.partitions(partition.index) + val (leftPartition, rightPartition) = + partition.asInstanceOf[NativePartition[(Partition, Partition)]].payload val leftChild = leftRDD.nativePlan(leftPartition, taskContext) - val rightPartition = rightRDD.partitions(partition.index) val rightChild = rightRDD.nativePlan(rightPartition, taskContext) val sortMergeJoinExec = SortMergeJoinExecNode diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeTakeOrderedBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeTakeOrderedBase.scala index 16b338977..df66b4b25 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeTakeOrderedBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeTakeOrderedBase.scala @@ -21,8 +21,10 @@ import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.auron.Shims @@ -137,8 +139,7 @@ abstract class NativeTakeOrderedBase( shuffledRDD.partitioner, new OneToOneDependency(shuffledRDD) :: Nil, rddShuffleReadFull = false, - (_, taskContext) => { - val inputPartition = shuffledRDD.partitions(0) + (inputPartition, taskContext) => { val nativeTakeOrderedExec = SortExecNode .newBuilder() .setInput(shuffledRDD.nativePlan(inputPartition, taskContext)) @@ -181,16 +182,19 @@ abstract class NativePartialTakeOrderedBase( override def doExecuteNative(): NativeRDD = { val inputRDD = NativeHelper.executeNative(child) val nativeSortExprs = this.nativeSortExprs + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, metrics = SparkMetricNode(metrics, inputRDD.metrics :: Nil), - inputRDD.partitions, + nativePartitions.toArray, inputRDD.partitioner, new OneToOneDependency(inputRDD) :: Nil, rddShuffleReadFull = false, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeTakeOrderedExec = SortExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeWindowBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeWindowBase.scala index f0336be8b..60b7557c4 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeWindowBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeWindowBase.scala @@ -20,8 +20,10 @@ import scala.collection.immutable.SortedMap import scala.jdk.CollectionConverters._ import org.apache.spark.OneToOneDependency +import org.apache.spark.Partition import org.apache.spark.sql.auron.NativeConverters import org.apache.spark.sql.auron.NativeHelper +import org.apache.spark.sql.auron.NativePartition import org.apache.spark.sql.auron.NativeRDD import org.apache.spark.sql.auron.NativeSupports import org.apache.spark.sql.catalyst.expressions.Ascending @@ -271,16 +273,19 @@ abstract class NativeWindowBase( val nativeWindowExprs = this.nativeWindowExprs val nativeOrderSpecExprs = this.nativeOrderSpecExprs val nativePartitionSpecExprs = this.nativePartitionSpecExprs + val nativePartitions = inputRDD.partitions.map { inputPartition => + NativePartition[Partition](inputPartition.index, inputPartition) + } new NativeRDD( sparkContext, nativeMetrics, - rddPartitions = inputRDD.partitions, + rddPartitions = nativePartitions.toArray, rddPartitioner = inputRDD.partitioner, rddDependencies = new OneToOneDependency(inputRDD) :: Nil, inputRDD.isShuffleReadFull, (partition, taskContext) => { - val inputPartition = inputRDD.partitions(partition.index) + val inputPartition = NativePartition.unwrap(partition) val nativeWindowExec = pb.WindowExecNode .newBuilder() .setInput(inputRDD.nativePlan(inputPartition, taskContext)) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/shuffle/AuronShuffleDependency.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/shuffle/AuronShuffleDependency.scala index 4c5b47c1d..0a61f3525 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/shuffle/AuronShuffleDependency.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/shuffle/AuronShuffleDependency.scala @@ -18,7 +18,7 @@ package org.apache.spark.sql.execution.auron.shuffle import scala.reflect.ClassTag -import org.apache.spark.{Aggregator, Partitioner, ShuffleDependency, SparkEnv} +import org.apache.spark.{Aggregator, Partition, Partitioner, ShuffleDependency, SparkEnv} import org.apache.spark.internal.Logging import org.apache.spark.rdd.RDD import org.apache.spark.serializer.Serializer @@ -35,7 +35,8 @@ class AuronShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag]( override val aggregator: Option[Aggregator[K, V, C]] = None, override val mapSideCombine: Boolean = false, override val shuffleWriterProcessor: ShuffleWriteProcessor = new ShuffleWriteProcessor, - val schema: StructType) + val schema: StructType, + val rddPartitions: Array[Partition] = null) extends ShuffleDependency[K, V, C]( _rdd, partitioner, diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/hive/execution/auron/plan/NativeHiveTableScanBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/hive/execution/auron/plan/NativeHiveTableScanBase.scala index 6dfc8be79..e1d56d9ff 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/hive/execution/auron/plan/NativeHiveTableScanBase.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/hive/execution/auron/plan/NativeHiveTableScanBase.scala @@ -45,7 +45,7 @@ import org.apache.auron.{protobuf => pb} import org.apache.auron.jni.JniBridge import org.apache.auron.sparkver -abstract class NativeHiveTableScanBase(basedHiveScan: HiveTableScanExec) +abstract class NativeHiveTableScanBase(@transient basedHiveScan: HiveTableScanExec) extends LeafExecNode with NativeSupports {

AltStyle によって変換されたページ (->オリジナル) /