You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "zhztheplayer (via GitHub)" <gi...@apache.org> on 2026/01/21 17:26:26 UTC
[PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
zhztheplayer opened a new pull request, #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461
The PR aims to offload the Spark shuffle generated during executing Delta write command with write optimization enabled, to fix https://github.com/apache/incubator-gluten/issues/11421.
The PR is WIP.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3784891349
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#discussion_r2714235664
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,352 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
+ }
+ }
+
+ override protected def doValidateInternal(): ValidationResult = {
+ columnarShufflePlan.doValidate()
+ }
+
+ override def doExecute(): RDD[InternalRow] = throw new UnsupportedOperationException
+
+ override def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ // Single partitioned tasks can simply be written
+ if (childNumPartitions <= 1) return child.executeColumnar()
+
+ val shuffledRDD = getShuffleRDD
+
+ val partitions = computeBins()
+
+ recordDeltaEvent(
+ deltaLog,
+ "delta.optimizeWrite.planned",
+ data = Map(
+ "originalPartitions" -> childNumPartitions,
+ "outputPartitions" -> partitions.length,
+ "shufflePartitions" -> numPartitions,
+ "numShuffleBlocks" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS),
+ "binSize" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE),
+ "maxShufflePartitions" ->
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS)
+ )
+ )
+
+ new GlutenDeltaOptimizedWriterRDD(
+ sparkContext,
+ shuffledRDD.dependency,
+ readMetrics,
+ new OptimizedWriterBlocks(partitions))
+ }
+
+ private def getConf[T](entry: ConfigEntry[T]): T = {
+ conf.getConf(entry)
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): GlutenDeltaOptimizedWriterExec =
+ copy(child = newChild)
+
+ override def batchType(): Convention.BatchType = VeloxBatchType
+
+ override def rowType0(): Convention.RowType = Convention.RowType.None
+}
+
+/**
+ * A specialized implementation similar to `ShuffledRowRDD`, where a partition reads a prepared set
+ * of shuffle blocks.
+ */
+private class GlutenDeltaOptimizedWriterRDD(
+ @transient sparkContext: SparkContext,
+ var dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ metrics: Map[String, SQLMetric],
+ @transient blocks: OptimizedWriterBlocks)
+ extends RDD[ColumnarBatch](sparkContext, Seq(dep))
+ with DeltaLogging {
+
+ override def getPartitions: Array[Partition] = Array.tabulate(blocks.bins.length) {
+ i => ShuffleBlockRDDPartition(i, blocks.bins(i))
+ }
+
+ override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = {
+ val tempMetrics = context.taskMetrics().createTempShuffleReadMetrics()
+ val sqlMetricsReporter = new SQLShuffleReadMetricsReporter(tempMetrics, metrics)
+
+ val blocks = if (context.stageAttemptNumber() > 0) {
+ // We lost shuffle blocks, so we need to now get new manager addresses
+ val executorTracker = SparkEnv.get.mapOutputTracker
+ val oldBlockLocations = split.asInstanceOf[ShuffleBlockRDDPartition].blocks
+
+ // assumes we bin-pack by reducerId
+ val reducerId = oldBlockLocations.head._2.head._1.asInstanceOf[ShuffleBlockId].reduceId
+ // Get block addresses
+ val newLocations = executorTracker
+ .getMapSizesByExecutorId(dep.shuffleId, reducerId)
+ .flatMap {
+ case (bmId, newBlocks) =>
+ newBlocks.map(blockInfo => (blockInfo._3, (bmId, blockInfo)))
+ }
+ .toMap
+
+ val blockLocations = new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ oldBlockLocations.foreach {
+ case (_, oldBlocks) =>
+ oldBlocks.foreach {
+ oldBlock =>
+ val (bmId, blockInfo) = newLocations(oldBlock._3)
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append(blockInfo)
+ }
+ }
+
+ blockLocations.iterator
+ } else {
+ split.asInstanceOf[ShuffleBlockRDDPartition].blocks.iterator
+ }
+
+ val reader = new GlutenOptimizedWriterShuffleReader(dep, context, blocks, sqlMetricsReporter)
+ reader.read().map(_._2)
+ }
+
+ override def clearDependencies(): Unit = {
+ super.clearDependencies()
+ dep = null
+ }
+}
+
+/** A simplified implementation of the `BlockStoreShuffleReader` for reading shuffle blocks. */
+private class GlutenOptimizedWriterShuffleReader(
+ dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ context: TaskContext,
+ blocks: Iterator[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])],
+ readMetrics: ShuffleReadMetricsReporter)
+ extends ShuffleReader[Int, ColumnarBatch] {
+
+ /** Read the combined key-values for this reduce task */
+ override def read(): Iterator[Product2[Int, ColumnarBatch]] = {
+ val wrappedStreams = new ShuffleBlockFetcherIterator(
+ context,
+ SparkEnv.get.blockManager.blockStoreClient,
+ SparkEnv.get.blockManager,
+ SparkEnv.get.mapOutputTracker,
+ blocks,
+ SparkEnv.get.serializerManager.wrapStream,
Review Comment:
Should use ` ColumnarShuffleManager.bypassDecompressionSerializerManger` for ColumnarShuffleDependency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3783944392
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3779928874
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#discussion_r2716552670
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,352 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
+ }
+ }
+
+ override protected def doValidateInternal(): ValidationResult = {
+ columnarShufflePlan.doValidate()
+ }
+
+ override def doExecute(): RDD[InternalRow] = throw new UnsupportedOperationException
+
+ override def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ // Single partitioned tasks can simply be written
+ if (childNumPartitions <= 1) return child.executeColumnar()
+
+ val shuffledRDD = getShuffleRDD
+
+ val partitions = computeBins()
+
+ recordDeltaEvent(
+ deltaLog,
+ "delta.optimizeWrite.planned",
+ data = Map(
+ "originalPartitions" -> childNumPartitions,
+ "outputPartitions" -> partitions.length,
+ "shufflePartitions" -> numPartitions,
+ "numShuffleBlocks" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS),
+ "binSize" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE),
+ "maxShufflePartitions" ->
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS)
+ )
+ )
+
+ new GlutenDeltaOptimizedWriterRDD(
+ sparkContext,
+ shuffledRDD.dependency,
+ readMetrics,
+ new OptimizedWriterBlocks(partitions))
+ }
+
+ private def getConf[T](entry: ConfigEntry[T]): T = {
+ conf.getConf(entry)
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): GlutenDeltaOptimizedWriterExec =
+ copy(child = newChild)
+
+ override def batchType(): Convention.BatchType = VeloxBatchType
+
+ override def rowType0(): Convention.RowType = Convention.RowType.None
+}
+
+/**
+ * A specialized implementation similar to `ShuffledRowRDD`, where a partition reads a prepared set
+ * of shuffle blocks.
+ */
+private class GlutenDeltaOptimizedWriterRDD(
+ @transient sparkContext: SparkContext,
+ var dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ metrics: Map[String, SQLMetric],
+ @transient blocks: OptimizedWriterBlocks)
+ extends RDD[ColumnarBatch](sparkContext, Seq(dep))
+ with DeltaLogging {
+
+ override def getPartitions: Array[Partition] = Array.tabulate(blocks.bins.length) {
+ i => ShuffleBlockRDDPartition(i, blocks.bins(i))
+ }
+
+ override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = {
+ val tempMetrics = context.taskMetrics().createTempShuffleReadMetrics()
+ val sqlMetricsReporter = new SQLShuffleReadMetricsReporter(tempMetrics, metrics)
+
+ val blocks = if (context.stageAttemptNumber() > 0) {
+ // We lost shuffle blocks, so we need to now get new manager addresses
+ val executorTracker = SparkEnv.get.mapOutputTracker
+ val oldBlockLocations = split.asInstanceOf[ShuffleBlockRDDPartition].blocks
+
+ // assumes we bin-pack by reducerId
+ val reducerId = oldBlockLocations.head._2.head._1.asInstanceOf[ShuffleBlockId].reduceId
+ // Get block addresses
+ val newLocations = executorTracker
+ .getMapSizesByExecutorId(dep.shuffleId, reducerId)
+ .flatMap {
+ case (bmId, newBlocks) =>
+ newBlocks.map(blockInfo => (blockInfo._3, (bmId, blockInfo)))
+ }
+ .toMap
+
+ val blockLocations = new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ oldBlockLocations.foreach {
+ case (_, oldBlocks) =>
+ oldBlocks.foreach {
+ oldBlock =>
+ val (bmId, blockInfo) = newLocations(oldBlock._3)
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append(blockInfo)
+ }
+ }
+
+ blockLocations.iterator
+ } else {
+ split.asInstanceOf[ShuffleBlockRDDPartition].blocks.iterator
+ }
+
+ val reader = new GlutenOptimizedWriterShuffleReader(dep, context, blocks, sqlMetricsReporter)
+ reader.read().map(_._2)
+ }
+
+ override def clearDependencies(): Unit = {
+ super.clearDependencies()
+ dep = null
+ }
+}
+
+/** A simplified implementation of the `BlockStoreShuffleReader` for reading shuffle blocks. */
+private class GlutenOptimizedWriterShuffleReader(
Review Comment:
> Seems like this simplified shuffle reader implementation is quite similar to ColumnarShuffleReader.
I was following Delta's practice to add a new inner reader class here. I thought it was the safest move in the first iteration.
> Can we just use ColumnarShuffleReader here?
Yes, we can give it a try later on.
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,352 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
+ }
+ }
+
+ override protected def doValidateInternal(): ValidationResult = {
+ columnarShufflePlan.doValidate()
+ }
+
+ override def doExecute(): RDD[InternalRow] = throw new UnsupportedOperationException
+
+ override def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ // Single partitioned tasks can simply be written
+ if (childNumPartitions <= 1) return child.executeColumnar()
+
+ val shuffledRDD = getShuffleRDD
+
+ val partitions = computeBins()
+
+ recordDeltaEvent(
+ deltaLog,
+ "delta.optimizeWrite.planned",
+ data = Map(
+ "originalPartitions" -> childNumPartitions,
+ "outputPartitions" -> partitions.length,
+ "shufflePartitions" -> numPartitions,
+ "numShuffleBlocks" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS),
+ "binSize" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE),
+ "maxShufflePartitions" ->
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS)
+ )
+ )
+
+ new GlutenDeltaOptimizedWriterRDD(
+ sparkContext,
+ shuffledRDD.dependency,
+ readMetrics,
+ new OptimizedWriterBlocks(partitions))
+ }
+
+ private def getConf[T](entry: ConfigEntry[T]): T = {
+ conf.getConf(entry)
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): GlutenDeltaOptimizedWriterExec =
+ copy(child = newChild)
+
+ override def batchType(): Convention.BatchType = VeloxBatchType
+
+ override def rowType0(): Convention.RowType = Convention.RowType.None
+}
+
+/**
+ * A specialized implementation similar to `ShuffledRowRDD`, where a partition reads a prepared set
+ * of shuffle blocks.
+ */
+private class GlutenDeltaOptimizedWriterRDD(
+ @transient sparkContext: SparkContext,
+ var dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ metrics: Map[String, SQLMetric],
+ @transient blocks: OptimizedWriterBlocks)
+ extends RDD[ColumnarBatch](sparkContext, Seq(dep))
+ with DeltaLogging {
+
+ override def getPartitions: Array[Partition] = Array.tabulate(blocks.bins.length) {
+ i => ShuffleBlockRDDPartition(i, blocks.bins(i))
+ }
+
+ override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = {
+ val tempMetrics = context.taskMetrics().createTempShuffleReadMetrics()
+ val sqlMetricsReporter = new SQLShuffleReadMetricsReporter(tempMetrics, metrics)
+
+ val blocks = if (context.stageAttemptNumber() > 0) {
+ // We lost shuffle blocks, so we need to now get new manager addresses
+ val executorTracker = SparkEnv.get.mapOutputTracker
+ val oldBlockLocations = split.asInstanceOf[ShuffleBlockRDDPartition].blocks
+
+ // assumes we bin-pack by reducerId
+ val reducerId = oldBlockLocations.head._2.head._1.asInstanceOf[ShuffleBlockId].reduceId
+ // Get block addresses
+ val newLocations = executorTracker
+ .getMapSizesByExecutorId(dep.shuffleId, reducerId)
+ .flatMap {
+ case (bmId, newBlocks) =>
+ newBlocks.map(blockInfo => (blockInfo._3, (bmId, blockInfo)))
+ }
+ .toMap
+
+ val blockLocations = new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ oldBlockLocations.foreach {
+ case (_, oldBlocks) =>
+ oldBlocks.foreach {
+ oldBlock =>
+ val (bmId, blockInfo) = newLocations(oldBlock._3)
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append(blockInfo)
+ }
+ }
+
+ blockLocations.iterator
+ } else {
+ split.asInstanceOf[ShuffleBlockRDDPartition].blocks.iterator
+ }
+
+ val reader = new GlutenOptimizedWriterShuffleReader(dep, context, blocks, sqlMetricsReporter)
+ reader.read().map(_._2)
+ }
+
+ override def clearDependencies(): Unit = {
+ super.clearDependencies()
+ dep = null
+ }
+}
+
+/** A simplified implementation of the `BlockStoreShuffleReader` for reading shuffle blocks. */
+private class GlutenOptimizedWriterShuffleReader(
+ dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ context: TaskContext,
+ blocks: Iterator[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])],
+ readMetrics: ShuffleReadMetricsReporter)
+ extends ShuffleReader[Int, ColumnarBatch] {
+
+ /** Read the combined key-values for this reduce task */
+ override def read(): Iterator[Product2[Int, ColumnarBatch]] = {
+ val wrappedStreams = new ShuffleBlockFetcherIterator(
+ context,
+ SparkEnv.get.blockManager.blockStoreClient,
+ SparkEnv.get.blockManager,
+ SparkEnv.get.mapOutputTracker,
+ blocks,
+ SparkEnv.get.serializerManager.wrapStream,
Review Comment:
Thanks for the suggestion. The issue has been fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3784001932
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3782081168
Cluster format still have fallback shuffle
SQL:
```
spark.sql(f''' CREATE TABLE {databasename}.{tbl}
USING DELTA
TBLPROPERTIES (
delta.autoOptimize.autoCompact = true,
delta.autoOptimize.optimizeWrite = true,
delta.dataSkippingNumIndexedCols = 128
)
CLUSTER BY ({partcol})
LOCATION 's3a://presto-workload/{databasename}/{tbl}'
AS SELECT *
FROM tpcds_sf2500_parquet_zstd_135_nonpart.{tbl}''')
```
<img width="441" height="437" alt="image" src="https://github.com/user-attachments/assets/1a863790-b689-4090-93ca-a765c165d277" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3783457775
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on code in PR #11461:
URL: https://github.com/apache/gluten/pull/11461#discussion_r3127409410
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,365 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.shuffle.sort.ColumnarShuffleManager
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private lazy val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(
+ ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
Review Comment:
https://github.com/apache/gluten/pull/11978
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on PR #11461:
URL: https://github.com/apache/gluten/pull/11461#issuecomment-4502541357
@zhztheplayer I cant find the GlutenDeltaOptimizedWriterExec class in the JAR downloaded from https://www.apache.org/dyn/closer.lua/gluten/1.6.0/apache-gluten-1.6.0-bin-spark-4.0.tar.gz. Do you know what I'm doing wrong?
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3784065658
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3784497759
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3783937725
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#discussion_r2716565590
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala:
##########
@@ -370,7 +370,15 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
val newShuffle = shuffle.outputPartitioning match {
case HashPartitioning(exprs, _) =>
- val hashExpr = new Murmur3Hash(exprs)
+ val hashExpr = if (exprs.isEmpty) {
+ // In Spark, a hash expression with empty input is not resolvable and an
+ // `WRONG_NUM_ARGS.WITHOUT_SUGGESTION` error will be reported when validating the project
+ // transformer. So we directly return the seed here, which is the intended hashed value
+ // for empty input given Spark's murmur3 hash logic.
+ Literal(new Murmur3Hash(Nil).seed, IntegerType)
+ } else {
+ new Murmur3Hash(exprs)
+ }
Review Comment:
cc @marin-ma Let me know if any thoughts about this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#discussion_r2714126908
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,352 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
+ }
+ }
+
+ override protected def doValidateInternal(): ValidationResult = {
+ columnarShufflePlan.doValidate()
+ }
+
+ override def doExecute(): RDD[InternalRow] = throw new UnsupportedOperationException
+
+ override def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ // Single partitioned tasks can simply be written
+ if (childNumPartitions <= 1) return child.executeColumnar()
+
+ val shuffledRDD = getShuffleRDD
+
+ val partitions = computeBins()
+
+ recordDeltaEvent(
+ deltaLog,
+ "delta.optimizeWrite.planned",
+ data = Map(
+ "originalPartitions" -> childNumPartitions,
+ "outputPartitions" -> partitions.length,
+ "shufflePartitions" -> numPartitions,
+ "numShuffleBlocks" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS),
+ "binSize" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE),
+ "maxShufflePartitions" ->
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS)
+ )
+ )
+
+ new GlutenDeltaOptimizedWriterRDD(
+ sparkContext,
+ shuffledRDD.dependency,
+ readMetrics,
+ new OptimizedWriterBlocks(partitions))
+ }
+
+ private def getConf[T](entry: ConfigEntry[T]): T = {
+ conf.getConf(entry)
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): GlutenDeltaOptimizedWriterExec =
+ copy(child = newChild)
+
+ override def batchType(): Convention.BatchType = VeloxBatchType
+
+ override def rowType0(): Convention.RowType = Convention.RowType.None
+}
+
+/**
+ * A specialized implementation similar to `ShuffledRowRDD`, where a partition reads a prepared set
+ * of shuffle blocks.
+ */
+private class GlutenDeltaOptimizedWriterRDD(
+ @transient sparkContext: SparkContext,
+ var dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ metrics: Map[String, SQLMetric],
+ @transient blocks: OptimizedWriterBlocks)
+ extends RDD[ColumnarBatch](sparkContext, Seq(dep))
+ with DeltaLogging {
+
+ override def getPartitions: Array[Partition] = Array.tabulate(blocks.bins.length) {
+ i => ShuffleBlockRDDPartition(i, blocks.bins(i))
+ }
+
+ override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = {
+ val tempMetrics = context.taskMetrics().createTempShuffleReadMetrics()
+ val sqlMetricsReporter = new SQLShuffleReadMetricsReporter(tempMetrics, metrics)
+
+ val blocks = if (context.stageAttemptNumber() > 0) {
+ // We lost shuffle blocks, so we need to now get new manager addresses
+ val executorTracker = SparkEnv.get.mapOutputTracker
+ val oldBlockLocations = split.asInstanceOf[ShuffleBlockRDDPartition].blocks
+
+ // assumes we bin-pack by reducerId
+ val reducerId = oldBlockLocations.head._2.head._1.asInstanceOf[ShuffleBlockId].reduceId
+ // Get block addresses
+ val newLocations = executorTracker
+ .getMapSizesByExecutorId(dep.shuffleId, reducerId)
+ .flatMap {
+ case (bmId, newBlocks) =>
+ newBlocks.map(blockInfo => (blockInfo._3, (bmId, blockInfo)))
+ }
+ .toMap
+
+ val blockLocations = new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ oldBlockLocations.foreach {
+ case (_, oldBlocks) =>
+ oldBlocks.foreach {
+ oldBlock =>
+ val (bmId, blockInfo) = newLocations(oldBlock._3)
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append(blockInfo)
+ }
+ }
+
+ blockLocations.iterator
+ } else {
+ split.asInstanceOf[ShuffleBlockRDDPartition].blocks.iterator
+ }
+
+ val reader = new GlutenOptimizedWriterShuffleReader(dep, context, blocks, sqlMetricsReporter)
+ reader.read().map(_._2)
+ }
+
+ override def clearDependencies(): Unit = {
+ super.clearDependencies()
+ dep = null
+ }
+}
+
+/** A simplified implementation of the `BlockStoreShuffleReader` for reading shuffle blocks. */
+private class GlutenOptimizedWriterShuffleReader(
Review Comment:
Seems like this simplified shuffle reader implementation is quite similar to `ColumnarShuffleReader`. Can we just use `ColumnarShuffleReader` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #11461:
URL: https://github.com/apache/gluten/pull/11461#issuecomment-4519117136
@felipepessoto please refer to the build script used: https://github.com/apache/gluten/blob/v1.6.0/dev/release/build-release.sh, the build command may be different from GHA CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #11461:
URL: https://github.com/apache/gluten/pull/11461#issuecomment-4519128062
@felipepessoto the build process is also documented at https://github.com/apache/gluten/blob/main/docs/developers/HowToRelease.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3789319089
> There is still a C2R/R2C after exchange
Yes, it was expected and I've marked the reason [here](https://github.com/apache/incubator-gluten/pull/11461/files#diff-44537c8d54133707205e1318508c50245a2956642d5b96d14d03b5ee5f09b10fR111-R113). Will follow up later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#discussion_r2713631258
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,352 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
Review Comment:
This class is modified from the Delta 3.3 class: https://github.com/delta-io/delta/blob/v3.3.0/spark/src/main/scala/org/apache/spark/sql/delta/perf/DeltaOptimizedWriterExec.scala.
We mainly need to offload the row-based shuffle as an `columnarShufflePlan` which is placed as a member field of this new class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #11461:
URL: https://github.com/apache/gluten/pull/11461#discussion_r3054970050
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,365 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.shuffle.sort.ColumnarShuffleManager
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private lazy val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(
+ ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
Review Comment:
Thanks, and yes, this looks a bit off, could be unexpectedly brought to the PR from some test code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3783962442
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#discussion_r2716622622
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala:
##########
@@ -370,7 +370,15 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
val newShuffle = shuffle.outputPartitioning match {
case HashPartitioning(exprs, _) =>
- val hashExpr = new Murmur3Hash(exprs)
+ val hashExpr = if (exprs.isEmpty) {
+ // In Spark, a hash expression with empty input is not resolvable and an
+ // `WRONG_NUM_ARGS.WITHOUT_SUGGESTION` error will be reported when validating the project
+ // transformer. So we directly return the seed here, which is the intended hashed value
+ // for empty input given Spark's murmur3 hash logic.
+ Literal(new Murmur3Hash(Nil).seed, IntegerType)
+ } else {
+ new Murmur3Hash(exprs)
+ }
Review Comment:
Looks good. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3787042509
@zhli1142015 Have you solved this in your case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3779909874
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3779933656
PR status:
When running the test case: https://github.com/apache/incubator-gluten/blob/4cd6440d23665d8807fbf1fc0d470a148accf020/backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/perf/OptimizedWritesSuite.scala#L124, an error "Stream is corrupted" from the shuffle reader is raised:
```
E20260121 18:29:26.731045 420393 Exceptions.h:53] Line: /opt/code/incubator-gluten/ep/build-velox/build/velox_ep/velox/exec/Driver.cpp:582, Function:operator(), Expression: Operator::getOutput failed for [operator: ValueStream, plan node ID: 0]: Error during calling Java code from native code: org.apache.gluten.exception.GlutenException: org.apache.gluten.exception.GlutenException: Error during calling Java code from native code: org.apache.spark.shuffle.FetchFailedException: Block shuffle_0_0_1805 is corrupted but the cause is unknown
at org.apache.spark.errors.SparkCoreErrors$.fetchFailedError(SparkCoreErrors.scala:437)
at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:1239)
at org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1398)
at org.apache.spark.storage.BufferReleasingInputStream.read(ShuffleBlockFetcherIterator.scala:1374)
at org.apache.gluten.vectorized.OnHeapJniByteInputStream.read(OnHeapJniByteInputStream.java:39)
at org.apache.gluten.vectorized.ColumnarBatchOutIterator.nativeNext(Native Method)
at org.apache.gluten.vectorized.ColumnarBatchOutIterator.next0(ColumnarBatchOutIterator.java:63)
at org.apache.gluten.vectorized.ColumnarBatchOutIterator.next0(ColumnarBatchOutIterator.java:28)
at org.apache.gluten.iterator.ClosableIterator.next(ClosableIterator.java:48)
at org.apache.gluten.vectorized.ColumnarBatchSerializerInstanceImpl$TaskDeserializationStream.liftedTree1$1(ColumnarBatchSerializer.scala:188)
at org.apache.gluten.vectorized.ColumnarBatchSerializerInstanceImpl$TaskDeserializationStream.readValue(ColumnarBatchSerializer.scala:187)
at org.apache.spark.serializer.DeserializationStream$$anon$2.getNext(Serializer.scala:188)
at org.apache.spark.serializer.DeserializationStream$$anon$2.getNext(Serializer.scala:185)
at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at scala.collection.convert.Wrappers$IteratorWrapper.hasNext(Wrappers.scala:32)
at org.apache.gluten.vectorized.ColumnarBatchInIterator.hasNext(ColumnarBatchInIterator.java:36)
at org.apache.gluten.vectorized.ColumnarBatchOutIterator.nativeHasNext(Native Method)
at org.apache.gluten.vectorized.ColumnarBatchOutIterator.hasNext0(ColumnarBatchOutIterator.java:58)
at org.apache.gluten.iterator.ClosableIterator.hasNext(ClosableIterator.java:36)
at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:45)
at org.apache.gluten.iterator.IteratorsV1$InvocationFlowProtection.hasNext(IteratorsV1.scala:154)
at org.apache.gluten.iterator.IteratorsV1$IteratorCompleter.hasNext(IteratorsV1.scala:66)
at org.apache.gluten.iterator.IteratorsV1$PayloadCloser.hasNext(IteratorsV1.scala:38)
at org.apache.gluten.iterator.IteratorsV1$LifeTimeAccumulator.hasNext(IteratorsV1.scala:95)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
at org.apache.spark.sql.delta.files.GlutenDeltaFileFormatWriter$.executeTask(GlutenDeltaFileFormatWriter.scala:452)
at org.apache.spark.sql.delta.files.GlutenDeltaFileFormatWriter$.$anonfun$executeWrite$4(GlutenDeltaFileFormatWriter.scala:311)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
at org.apache.spark.scheduler.Task.run(Task.scala:141)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.io.IOException: Stream is corrupted
at net.jpountz.lz4.LZ4BlockInputStream.refill(LZ4BlockInputStream.java:202)
at net.jpountz.lz4.LZ4BlockInputStream.read(LZ4BlockInputStream.java:159)
at net.jpountz.lz4.LZ4BlockInputStream.read(LZ4BlockInputStream.java:172)
at org.apache.spark.storage.BufferReleasingInputStream.$anonfun$read$2(ShuffleBlockFetcherIterator.scala:1374)
at scala.runtime.java8.JFunction0$mcI$sp.apply(JFunction0$mcI$sp.java:23)
at org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1389)
... 37 more
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3783954997
> Cluster format still have fallback shuffle
>
> SQL:
>
> ```
> spark.sql(f''' CREATE TABLE {databasename}.{tbl}
> USING DELTA
> TBLPROPERTIES (
> delta.autoOptimize.autoCompact = true,
> delta.autoOptimize.optimizeWrite = true,
> delta.dataSkippingNumIndexedCols = 128
> )
> CLUSTER BY ({partcol})
> LOCATION 's3a://presto-workload/{databasename}/{tbl}'
> AS SELECT *
> FROM tpcds_sf2500_parquet_zstd_135_nonpart.{tbl}''')
> ```
>
> <img alt="image" width="441" height="437" src="https://private-user-images.githubusercontent.com/47296334/538894684-1a863790-b689-4090-93ca-a765c165d277.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NjkwODIyOTYsIm5iZiI6MTc2OTA4MTk5NiwicGF0aCI6Ii80NzI5NjMzNC81Mzg4OTQ2ODQtMWE4NjM3OTAtYjY4OS00MDkwLTkzY2EtYTc2NWMxNjVkMjc3LnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjAxMjIlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYwMTIyVDExMzk1NlomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTExMTJhODk5OTE3NzJiMGVmNzE3OTVmMGRhMDljMTJjYTdiNTZjZDJlNmU2YWI1MmMyZjAzNDk0ZjYxY2M1ZWImWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0In0._t4E5rJTom8SYWHVdR8hqEyTOVBSR51b3ZgXT0MZ0A8">
If I repeated it correctly, this should be fixed in https://github.com/apache/incubator-gluten/pull/11461/commits/7f1fe98e38bdc876eb008433562a094ed012f0a6.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3780828281
Is it that the shuffle writer uses native shuffle write, but the reader is trying to decompress on the java side?
```
Caused by: java.io.IOException: Stream is corrupted
at net.jpountz.lz4.LZ4BlockInputStream.refill(LZ4BlockInputStream.java:202)
at net.jpountz.lz4.LZ4BlockInputStream.read(LZ4BlockInputStream.java:159)
at net.jpountz.lz4.LZ4BlockInputStream.read(LZ4BlockInputStream.java:172)
at org.apache.spark.storage.BufferReleasingInputStream.$anonfun$read$2(ShuffleBlockFetcherIterator.scala:1374)
at scala.runtime.java8.JFunction0$mcI$sp.apply(JFunction0$mcI$sp.java:23)
at org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1389)
... 37 more
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW commented on PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461#issuecomment-3787297117
> If I repeated it correctly, this should be fixed in [7f1fe98](https://github.com/apache/incubator-gluten/commit/7f1fe98e38bdc876eb008433562a094ed012f0a6).
confirm solved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer merged PR #11461:
URL: https://github.com/apache/incubator-gluten/pull/11461
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #11461:
URL: https://github.com/apache/gluten/pull/11461#issuecomment-4507410630
@felipepessoto the 1.6 release excludes all lake format supports for 4.0: https://github.com/apache/gluten/releases/tag/v1.6.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on PR #11461:
URL: https://github.com/apache/gluten/pull/11461#issuecomment-4510804203
@zhztheplayer the https://github.com/apache/gluten/blob/main/.github/workflows/velox_backend_x86.yml has the `-Pdelta` flag. Is the `Lake = OFF` set by another flag?
Do you have any PR/issue/doc where I can find more info about it?
At least for previous versions I wasn't aware of this `Lake` flag, is it new?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-10215][VL] Delta Write: Offload DeltaOptimizedWriterExec [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on code in PR #11461:
URL: https://github.com/apache/gluten/pull/11461#discussion_r3054692961
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/perf/GlutenDeltaOptimizedWriterExec.scala:
##########
@@ -0,0 +1,365 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.delta.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{ValidatablePlan, ValidationResult}
+import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
+
+// scalastyle:off import.ordering.noEmptyLine
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.shuffle._
+import org.apache.spark.shuffle.sort.ColumnarShuffleManager
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, ColumnarShuffleExchangeExec, GenerateTransformStageId}
+import org.apache.spark.sql.execution.{ShuffledColumnarBatchRDD, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage._
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/** Gluten's vectorized version of [[DeltaOptimizedWriterExec]]. */
+case class GlutenDeltaOptimizedWriterExec(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends ValidatablePlan
+ with UnaryExecNode
+ with DeltaLogging {
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+ private lazy val readMetrics =
+ SQLShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size")
+ ) ++ readMetrics ++ writeMetrics
+
+ private lazy val childNumPartitions = child.executeColumnar().getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ val targetShuffleBlocks = getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private lazy val columnarShufflePlan = {
+ val resolver = org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffle =
+ ShuffleExchangeExec(saltedPartitioning, child)
+ val columnarShuffle =
+ BackendsApiManager.getSparkPlanExecApiInstance.genColumnarShuffleExchange(shuffle)
+ val columnarShuffleWithWst =
+ GenerateTransformStageId()(
+ ColumnarCollapseTransformStages(new GlutenConfig(conf))(columnarShuffle))
+ columnarShuffleWithWst.asInstanceOf[ColumnarShuffleExchangeExec]
+ }
+
+ /** Creates a ShuffledRowRDD for facilitating the shuffle in the map side. */
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val columnarShuffleRdd =
+ columnarShufflePlan.executeColumnar().asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = columnarShuffleRdd
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+ ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE), ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long, Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId, Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ /** Performs the shuffle before the write, so that we can bin-pack output data. */
+ private def getShuffleStats(): Array[(BlockManagerId, collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+ // Gets the shuffle output stats
+ def getStats() =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0, numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats()
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats()
+ } catch {
+ case e: FetchFailedException =>
+ logWarning(log"Failed to fetch shuffle blocks for the optimized writer. Retrying", e)
+ awaitShuffleMapStage()
+ getStats()
+ throw e
Review Comment:
@zhztheplayer is this `throw e` expected? It seems we have a retry, but after it, we rethrow the exception
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org