diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala index d45b7a3f20..3204ed8a36 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala @@ -30,7 +30,7 @@ import org.apache.gluten.substrait.SubstraitContext import org.apache.gluten.substrait.expression.{ExpressionBuilder, ExpressionNode, WindowFunctionNode} import org.apache.gluten.vectorized.{ColumnarBatchSerializer, ColumnarBatchSerializeResult} -import org.apache.spark.{ShuffleDependency, SparkEnv, SparkException} +import org.apache.spark.{broadcast, ShuffleDependency, SparkEnv, SparkException} import org.apache.spark.api.python.{ColumnarArrowEvalPythonExec, PullOutArrowEvalPythonPreProjectHelper} import org.apache.spark.internal.Logging import org.apache.spark.memory.SparkMemoryUtil @@ -879,57 +879,15 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging { val useOffheapBroadcastBuildRelation = VeloxConfig.get.enableBroadcastBuildRelationInOffheap - val serialized: Seq[ColumnarBatchSerializeResult] = newChild - .executeColumnar() - .mapPartitions(itr => Iterator(BroadcastUtils.serializeStream(itr))) - .filter(_.numRows != 0) - .collect - val buildSideRowCount = serialized.map(_.numRows).sum - val rawSize = serialized.map(_.sizeInBytes()).sum - if (rawSize >= GlutenConfig.get.maxBroadcastTableSize) { - throw new SparkException( - "Cannot broadcast the table that is larger than " + - s"${SparkMemoryUtil.bytesToString(GlutenConfig.get.maxBroadcastTableSize)}: " + - s"${SparkMemoryUtil.bytesToString(rawSize)}") - } - numOutputRows += buildSideRowCount - dataSize += rawSize - - val rawThreads = - math - .ceil(dataSize.value.toDouble / VeloxConfig.get.veloxBroadcastHashTableBuildTargetBytes) - .toInt - val buildThreadsValue = if (rawThreads < 1) 1 else rawThreads - buildThreads += buildThreadsValue - - // Create the base ColumnarBuildSideRelation first - val columnarRelation = if (useOffheapBroadcastBuildRelation) { - TaskResources.runUnsafe { - UnsafeColumnarBuildSideRelation( - newOutput, - serialized.flatMap(_.offHeapData().asScala), - mode, - newBuildKeys, - offload, - buildThreadsValue) - } - } else { - ColumnarBuildSideRelation( - newOutput, - serialized.flatMap(_.onHeapData().asScala).toArray, - mode, - newBuildKeys, - offload, - buildThreadsValue) - } - // Check if we should build hash table on driver (Spark-native approach) // Only do this for HashedRelationBroadcastMode and when offload is enabled val shouldBuildOnDriver = VeloxConfig.get.enableDriverSideBroadcastHashTableBuild && mode.isInstanceOf[HashedRelationBroadcastMode] && offload - if (shouldBuildOnDriver) { + // The join context is resolved before the build side is collected, because it is also part of + // the key of the driver-side broadcast relation cache: a cache hit skips the collect. + val buildContextOpt: Option[BroadcastHashJoinContext] = if (shouldBuildOnDriver) { // Try to get broadcast join context from logical plan tag // In multi-join scenarios, there may be multiple contexts. Find the one that matches // the current broadcast child's output. @@ -951,8 +909,13 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging { } } - joinContextOpt match { - case Some(joinContext) => + if (joinContextOpt.isEmpty) { + // No join context available - fall back to executor-side build + logInfo(s"No broadcast join context found in logical plan, using executor-side build") + } + + joinContextOpt.map { + joinContext => // We have join context information - build hash table on driver logInfo( s"Building hash table on driver in BroadcastExchangeExec " + @@ -1009,7 +972,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging { } else { joinContext.originalLeftKeys } - val buildContext = BroadcastHashJoinContext( + BroadcastHashJoinContext( buildSideJoinKeys = if (newBuildKeys.nonEmpty) newBuildKeys else joinKeys, substraitJoinType = substraitJoinType, buildRight = joinContext.buildRight, @@ -1026,65 +989,167 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging { serializeHashTableTimeMetric = Option(serializeHashTableTimeMetric), serializedHashTableSizeMetric = Option(serializedHashTableSizeMetric) ) + } + } else { + None + } - try { - // Build and serialize hash table on driver - val (serializedHashTable, safeMode) = columnarRelation match { - case rel: ColumnarBuildSideRelation => - ( - VeloxBroadcastBuildSideCache - .buildAndSerializeOnDriverInBroadcastExchange( - rel, - buildContext, - buildSideRowCount), - rel.safeBroadcastMode - ) - case rel: UnsafeColumnarBuildSideRelation => - ( - VeloxBroadcastBuildSideCache - .buildAndSerializeOnDriverInBroadcastExchange( - rel, - buildContext, - buildSideRowCount), - rel.getSafeBroadcastMode - ) - } + // A hash table built by a previous query can be reused as long as the build side plan and the + // join properties it was built with are the same. This is what makes repeated workloads, e.g. + // the concurrent streams of a TPC-DS throughput run, skip the whole build side pipeline. + val relationCacheKeyOpt = buildContextOpt + .filter(_ => VeloxDriverBroadcastRelationCache.enabled) + .flatMap { + buildContext => + VeloxDriverBroadcastRelationCache.keyOf( + newChild, + newOutput, + newBuildKeys, + buildContext, + useOffheapBroadcastBuildRelation) + } - logInfo( - s"Successfully built hash table on driver: " + - s"size=${serializedHashTable.sizeInBytes} bytes, " + - s"rows=${serializedHashTable.numRows}, " + - s"joinType=${joinContext.joinType}, " + - s"broadcastId=$broadcastId") - - // Return SerializedHashTableBroadcastRelation - SerializedHashTableBroadcastRelation( - serializedHashTable, - safeMode, - newOutput, - 0L, // buildTimeMs - tracked inside SerializedBroadcastHashTable - 0L // serializeTimeMs - tracked inside SerializedBroadcastHashTable - ) - } catch { - case e: Exception => - logWarning( - s"Failed to build hash table on driver for broadcastId=$broadcastId, " + - s"falling back to executor-side build: ${e.getMessage}", - e) - columnarRelation - } + val cachedRelation = relationCacheKeyOpt.flatMap(VeloxDriverBroadcastRelationCache.getIfPresent) + if (cachedRelation.isDefined) { + val cached = cachedRelation.get + logInfo( + s"Reusing the broadcast relation built by a previous query: " + + s"size=${cached.serializedSize} bytes, rows=${cached.numRows}") + // The build side job is skipped, so report the metrics of the job that built the relation. + numOutputRows += cached.numRows + dataSize += cached.rawSize + Option(buildThreads).foreach(_ += cached.buildThreads) + Option(serializedHashTableSizeMetric).foreach(_.set(cached.serializedSize)) + return cached.relation + } - case None => - // No join context available - fall back to executor-side build - logInfo(s"No broadcast join context found in logical plan, using executor-side build") - columnarRelation + val serialized: Seq[ColumnarBatchSerializeResult] = newChild + .executeColumnar() + .mapPartitions(itr => Iterator(BroadcastUtils.serializeStream(itr))) + .filter(_.numRows != 0) + .collect + val buildSideRowCount = serialized.map(_.numRows).sum + val rawSize = serialized.map(_.sizeInBytes()).sum + if (rawSize >= GlutenConfig.get.maxBroadcastTableSize) { + throw new SparkException( + "Cannot broadcast the table that is larger than " + + s"${SparkMemoryUtil.bytesToString(GlutenConfig.get.maxBroadcastTableSize)}: " + + s"${SparkMemoryUtil.bytesToString(rawSize)}") + } + numOutputRows += buildSideRowCount + dataSize += rawSize + + val rawThreads = + math + .ceil(dataSize.value.toDouble / VeloxConfig.get.veloxBroadcastHashTableBuildTargetBytes) + .toInt + val buildThreadsValue = if (rawThreads < 1) 1 else rawThreads + buildThreads += buildThreadsValue + + // Create the base ColumnarBuildSideRelation first + val columnarRelation = if (useOffheapBroadcastBuildRelation) { + TaskResources.runUnsafe { + UnsafeColumnarBuildSideRelation( + newOutput, + serialized.flatMap(_.offHeapData().asScala), + mode, + newBuildKeys, + offload, + buildThreadsValue) } } else { - // Return ColumnarBuildSideRelation for executor-side build (legacy approach) - columnarRelation + ColumnarBuildSideRelation( + newOutput, + serialized.flatMap(_.onHeapData().asScala).toArray, + mode, + newBuildKeys, + offload, + buildThreadsValue) + } + + buildContextOpt match { + case Some(buildContext) => + try { + // Build and serialize hash table on driver. When the relation is kept by the driver-side + // relation cache, that cache owns the serialized hash table, so it is not tracked by the + // per-exchange cache as well. + val trackSerializedHashTable = relationCacheKeyOpt.isEmpty + val (serializedHashTable, safeMode) = columnarRelation match { + case rel: ColumnarBuildSideRelation => + ( + VeloxBroadcastBuildSideCache + .buildAndSerializeOnDriverInBroadcastExchange( + rel, + buildContext, + buildSideRowCount, + trackSerializedHashTable), + rel.safeBroadcastMode + ) + case rel: UnsafeColumnarBuildSideRelation => + ( + VeloxBroadcastBuildSideCache + .buildAndSerializeOnDriverInBroadcastExchange( + rel, + buildContext, + buildSideRowCount, + trackSerializedHashTable), + rel.getSafeBroadcastMode + ) + } + + logInfo( + s"Successfully built hash table on driver: " + + s"size=${serializedHashTable.sizeInBytes} bytes, " + + s"rows=${serializedHashTable.numRows}, " + + s"joinType=${buildContext.substraitJoinType}, " + + s"broadcastId=${buildContext.buildHashTableId}") + + // Return SerializedHashTableBroadcastRelation + val relation = SerializedHashTableBroadcastRelation( + serializedHashTable, + safeMode, + newOutput, + 0L, // buildTimeMs - tracked inside SerializedBroadcastHashTable + 0L // serializeTimeMs - tracked inside SerializedBroadcastHashTable + ) + + relationCacheKeyOpt.foreach { + key => + VeloxDriverBroadcastRelationCache.put( + key, + relation, + buildSideRowCount, + rawSize, + buildThreadsValue, + serializedHashTable.sizeInBytes) + } + + relation + } catch { + case e: Exception => + logWarning( + s"Failed to build hash table on driver for " + + s"broadcastId=${buildContext.buildHashTableId}, " + + s"falling back to executor-side build: ${e.getMessage}", + e + ) + columnarRelation + } + + case None => + // Return ColumnarBuildSideRelation for executor-side build (legacy approach) + columnarRelation } } + override def cachedBroadcast(relation: BuildSideRelation): Option[broadcast.Broadcast[Any]] = + VeloxDriverBroadcastRelationCache.cachedBroadcast(relation) + + override def cacheBroadcast( + relation: BuildSideRelation, + broadcasted: broadcast.Broadcast[Any]): Unit = + VeloxDriverBroadcastRelationCache.registerBroadcast(relation, broadcasted) + override def doCanonicalizeForBroadcastMode(mode: BroadcastMode): BroadcastMode = { mode match { case hash: HashedRelationBroadcastMode diff --git a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala index 802a47cadd..e807e43d5d 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala @@ -120,6 +120,12 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) { def enableDriverSideBroadcastHashTableBuild: Boolean = getConf(VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_BUILD) + def enableDriverSideBroadcastHashTableCache: Boolean = + getConf(VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_CACHE_ENABLED) + + def driverSideBroadcastHashTableCacheMaxSize: Long = + getConf(VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_CACHE_MAX_SIZE) + def enableGpuAsyncShuffleReader: Boolean = getConf(ENABLE_GPU_ASYNC_SHUFFLE_READER) def gpuAsyncReaderMaxPrefetchBytes: Long = getConf(GPU_ASYNC_SHUFFLE_READER_MAX_PREFETCH_BYTES) @@ -773,6 +779,30 @@ object VeloxConfig extends ConfigRegistry { .booleanConf .createWithDefault(false) + val VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_CACHE_ENABLED = + buildConf("spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableCache.enabled") + .doc( + "Reuse the hash tables built by driver-side broadcast hash table build across queries of " + + "the same application. Cache entries are keyed by the canonicalized build side plan " + + "plus the join properties the hash table was built with, so identical broadcast build " + + "sides coming from different queries, e.g. the concurrent streams of a TPC-DS " + + "throughput run, share one hash table instead of collecting, building and " + + "broadcasting it again. Only takes effect when " + + "spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild is enabled. " + + "Reusing assumes the data behind a build side does not change while the application " + + "is running, and the cached hash tables hold driver off-heap memory until they are " + + "evicted, so this is disabled by default.") + .booleanConf + .createWithDefault(false) + + val VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_CACHE_MAX_SIZE = + buildConf("spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableCache.maxSize") + .doc( + "The maximum total size of the serialized hash tables kept by the driver-side broadcast " + + "hash table cache. Least-recently-used entries are evicted once the limit is exceeded.") + .bytesConf(ByteUnit.BYTE) + .createWithDefaultString("1GB") + val QUERY_TRACE_ENABLED = buildConf("spark.gluten.sql.columnar.backend.velox.queryTraceEnabled") .doc("Enable query tracing flag.") .booleanConf diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala index 02b18b5d2b..abdd5b2e2d 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala @@ -111,11 +111,17 @@ object VeloxBroadcastBuildSideCache * BroadcastExchangeExec and doesn't need a broadcast variable. * * This is the Spark-native approach where hash table is built in BroadcastExchangeExec. + * + * @param trackInDriverCache + * whether the serialized hash table is kept in `driverSerializedCache`, which also owns its + * off-heap memory. Set to false when the caller keeps the built relation alive itself, e.g. + * `VeloxDriverBroadcastRelationCache`, so that the hash table has a single owner. */ def buildAndSerializeOnDriverInBroadcastExchange( relation: BuildSideRelation, broadcastContext: BroadcastHashJoinContext, - numRows: Long): SerializedBroadcastHashTable = { + numRows: Long, + trackInDriverCache: Boolean = true): SerializedBroadcastHashTable = { val broadcastId = broadcastContext.buildHashTableId @@ -180,7 +186,9 @@ object VeloxBroadcastBuildSideCache broadcastContext.serializeHashTableTimeMetric.foreach(_ += serializeTimeMs) broadcastContext.serializedHashTableSizeMetric.foreach(_ += result.sizeInBytes) - driverSerializedCache.put(broadcastId, result) + if (trackInDriverCache) { + driverSerializedCache.put(broadcastId, result) + } result } finally { resetRelation(droppedDuplicates) @@ -233,6 +241,7 @@ object VeloxBroadcastBuildSideCache def cleanAll(): Unit = { buildSideRelationCache.invalidateAll() driverSerializedCache.invalidateAll() + VeloxDriverBroadcastRelationCache.cleanAll() } override def onRemoval( diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxDriverBroadcastRelationCache.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxDriverBroadcastRelationCache.scala new file mode 100644 index 0000000000..9fcb281554 --- /dev/null +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxDriverBroadcastRelationCache.scala @@ -0,0 +1,292 @@ +/* + * 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.gluten.execution + +import org.apache.gluten.backendsapi.velox.VeloxBackendSettings +import org.apache.gluten.config.VeloxConfig +import org.apache.gluten.expression.ConverterUtils + +import org.apache.spark.SparkEnv +import org.apache.spark.broadcast.Broadcast +import org.apache.spark.internal.Logging +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, PlanExpression} +import org.apache.spark.sql.catalyst.plans.QueryPlan +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.joins.BuildSideRelation +import org.apache.spark.sql.types.{StructField, StructType} + +import com.github.benmanes.caffeine.cache.{Cache, Caffeine, RemovalCause, RemovalListener, Weigher} +import com.github.benmanes.caffeine.cache.stats.CacheStats + +import java.util.IdentityHashMap +import java.util.concurrent.TimeUnit + +/** + * Identifies a broadcast hash table that was built on the driver. + * + * Every field is independent of the query instance the build side came from: the plan is + * canonicalized and the join keys are normalized against the build side output, so two structurally + * identical broadcast build sides taken from two different queries map to the same key. Fields that + * are not part of this key must not influence the content of the built hash table. + */ +case class DriverBroadcastRelationKey( + canonicalizedBuildPlan: SparkPlan, + buildSchema: StructType, + normalizedJoinKeys: Seq[Expression], + normalizedRelationKeys: Seq[Expression], + substraitJoinType: Int, + buildRight: Boolean, + hasMixedFiltCondition: Boolean, + isExistenceJoin: Boolean, + filterBuildColumnOrdinals: Seq[Int], + filterPropagatesNulls: Boolean, + isNullAwareAntiJoin: Boolean, + bloomFilterPushdownSize: Long, + droppedDuplicates: Boolean, + useOffheapRelation: Boolean) + +/** + * A broadcast relation that is kept alive on the driver so that following queries can reuse it, + * together with the statistics of the job that produced it. The statistics are needed because a + * query that hits the cache does not run the build side job and still has to report the same + * `numOutputRows` / `dataSize` metrics as a query that builds the relation. + */ +class CachedBroadcastRelation( + val relation: BuildSideRelation, + val numRows: Long, + val rawSize: Long, + val buildThreads: Int, + val serializedSize: Long) { + + @volatile private var broadcasted: Broadcast[Any] = _ + + /** The broadcast that was created for [[relation]], if it has already been broadcast once. */ + def broadcast: Option[Broadcast[Any]] = Option(broadcasted) + + def setBroadcast(value: Broadcast[Any]): Unit = broadcasted = value +} + +/** + * Driver-side cache of the broadcast relations produced by driver-side broadcast hash table build. + * + * With `spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild` enabled the hash + * table is built and serialized once per `BroadcastExchangeExec` instance, which means every query + * rebuilds the hash tables of its build sides even when a previous query has just built the very + * same one. That is the common case for workloads that run the same queries over and over, e.g. the + * concurrent streams of a TPC-DS throughput run. + * + * This cache keys the built relation by the canonicalized build side plan (see + * [[DriverBroadcastRelationKey]]) instead of by the exchange instance, so a later query can skip: + * - collecting the build side to the driver, + * - building and serializing the hash table, + * - broadcasting the serialized hash table, when the very same relation object is broadcast again + * the broadcast created the first time is reused as well. + * + * The off-heap memory of a cached serialized hash table is released by `UnsafeByteArray#finalize` + * once neither this cache nor any broadcast refers to it, so eviction only drops the reference and + * never frees memory that a running query may still be using. + */ +object VeloxDriverBroadcastRelationCache extends Logging { + + // Caffeine weights are ints, so the cache is bounded in KiB rather than in bytes. + private val WEIGHT_UNIT = 1024L + + private lazy val cache: Cache[DriverBroadcastRelationKey, CachedBroadcastRelation] = { + val maxSize = VeloxConfig.get.driverSideBroadcastHashTableCacheMaxSize + val expiredTime = SparkEnv.get.conf.getLong( + VeloxBackendSettings.GLUTEN_VELOX_BROADCAST_CACHE_EXPIRED_TIME, + VeloxBackendSettings.GLUTEN_VELOX_BROADCAST_CACHE_EXPIRED_TIME_DEFAULT + ) + logInfo( + s"Creating driver-side broadcast relation cache, maxSize=$maxSize bytes, " + + s"expiredTime=$expiredTime seconds") + Caffeine.newBuilder + .expireAfterAccess(expiredTime, TimeUnit.SECONDS) + .maximumWeight(math.max(1L, maxSize / WEIGHT_UNIT)) + .weigher(new Weigher[DriverBroadcastRelationKey, CachedBroadcastRelation] { + override def weigh( + key: DriverBroadcastRelationKey, + value: CachedBroadcastRelation): Int = { + // A cached relation keeps both the serialized hash table and the collected build side + // batches it was built from alive, the latter is still needed by DPP. + val bytes = value.serializedSize + value.rawSize + math.max(1L, bytes / WEIGHT_UNIT).min(Int.MaxValue.toLong).toInt + } + }) + .removalListener(new RemovalListener[DriverBroadcastRelationKey, CachedBroadcastRelation] { + override def onRemoval( + key: DriverBroadcastRelationKey, + value: CachedBroadcastRelation, + cause: RemovalCause): Unit = { + if (value != null) { + broadcastIndex.synchronized(broadcastIndex.remove(value.relation)) + logInfo( + s"Evicted a cached broadcast relation of ${value.serializedSize} bytes, " + + s"cause: $cause") + } + } + }) + .recordStats() + .build[DriverBroadcastRelationKey, CachedBroadcastRelation]() + } + + // Maps a cached relation to its entry by identity, used to reuse the broadcast that was created + // for the relation. Kept in sync with `cache` by the removal listener above. + private val broadcastIndex = + new IdentityHashMap[BuildSideRelation, CachedBroadcastRelation]() + + def enabled: Boolean = { + val conf = VeloxConfig.get + conf.enableDriverSideBroadcastHashTableBuild && conf.enableDriverSideBroadcastHashTableCache + } + + /** + * Build the cache key of a broadcast build side, or return None if the build side must not be + * shared between queries. + * + * @param buildPlan + * the plan whose output is broadcast, after the pre-projection rewriting + * @param buildOutput + * the output of `buildPlan` + * @param relationKeys + * the build keys carried by the build side relation, empty if the relation takes its keys from + * the join context + * @param context + * the join context the hash table is built with + * @param useOffheapRelation + * whether the build side batches are kept off-heap + */ + def keyOf( + buildPlan: SparkPlan, + buildOutput: Seq[Attribute], + relationKeys: Seq[Expression], + context: BroadcastHashJoinContext, + useOffheapRelation: Boolean): Option[DriverBroadcastRelationKey] = { + if (!isCacheable(buildPlan)) { + logInfo( + "The broadcast build side is not shareable between queries, " + + "skipping the driver-side broadcast relation cache") + return None + } + + // `filterBuildColumns` carries expression ids, which differ between two instances of the same + // query. Turn the names back into positions in the build side output, and give up caching if + // any of them cannot be resolved. + val ordinalByName = buildOutput.zipWithIndex.map { + case (attr, ordinal) => ConverterUtils.genColumnNameWithExprId(attr) -> ordinal + }.toMap + val filterOrdinals = context.filterBuildColumns.map(ordinalByName.get) + if (filterOrdinals.exists(_.isEmpty)) { + logInfo( + "Failed to resolve the filter build columns of the broadcast build side, " + + "skipping the driver-side broadcast relation cache") + return None + } + + Some( + DriverBroadcastRelationKey( + canonicalizedBuildPlan = buildPlan.canonicalized, + buildSchema = StructType( + buildOutput.map(attr => StructField(attr.name, attr.dataType, attr.nullable))), + normalizedJoinKeys = + context.buildSideJoinKeys.map(QueryPlan.normalizeExpressions(_, buildOutput)), + normalizedRelationKeys = relationKeys.map(QueryPlan.normalizeExpressions(_, buildOutput)), + substraitJoinType = context.substraitJoinType.ordinal(), + buildRight = context.buildRight, + hasMixedFiltCondition = context.hasMixedFiltCondition, + isExistenceJoin = context.isExistenceJoin, + filterBuildColumnOrdinals = filterOrdinals.flatten.toSeq.sorted, + filterPropagatesNulls = context.filterPropagatesNulls, + isNullAwareAntiJoin = context.isNullAwareAntiJoin, + bloomFilterPushdownSize = context.bloomFilterPushdownSize, + droppedDuplicates = context.droppedDuplicates, + useOffheapRelation = useOffheapRelation + )) + } + + def getIfPresent(key: DriverBroadcastRelationKey): Option[CachedBroadcastRelation] = { + if (!enabled) { + None + } else { + Option(cache.getIfPresent(key)) + } + } + + def put( + key: DriverBroadcastRelationKey, + relation: BuildSideRelation, + numRows: Long, + rawSize: Long, + buildThreads: Int, + serializedSize: Long): Unit = { + val cached = + new CachedBroadcastRelation(relation, numRows, rawSize, buildThreads, serializedSize) + // Insert into `cache` first so that any removal listener triggered by a replaced entry fires + // before the new entry is visible in `broadcastIndex`. Reversing the order would leave a + // window where the removal listener could remove the new entry from `broadcastIndex`. + cache.put(key, cached) + broadcastIndex.synchronized(broadcastIndex.put(relation, cached)) + } + + /** The broadcast that was created for a cached relation, if any. */ + def cachedBroadcast(relation: BuildSideRelation): Option[Broadcast[Any]] = + entryOf(relation).flatMap(_.broadcast) + + /** Remember the broadcast created for a relation, so that a later query can reuse it. */ + def registerBroadcast(relation: BuildSideRelation, broadcasted: Broadcast[Any]): Unit = + entryOf(relation).foreach(_.setBroadcast(broadcasted)) + + private def entryOf(relation: BuildSideRelation): Option[CachedBroadcastRelation] = { + if (relation == null) { + None + } else { + broadcastIndex.synchronized(Option(broadcastIndex.get(relation))) + } + } + + /** + * A build side can only be shared between queries if it produces the same data every time it is + * evaluated. Non-deterministic expressions and subqueries, including the runtime filters of + * dynamic partition pruning, rule the build side out. + */ + private def isCacheable(plan: SparkPlan): Boolean = { + plan + .find { + p => + p.expressions.exists { + e => !e.deterministic || e.find(_.isInstanceOf[PlanExpression[_]]).isDefined + } + } + .isEmpty + } + + def size(): Long = { + // Run the pending maintenance work first, so that the returned size is exact. + cache.cleanUp() + cache.estimatedSize() + } + + def stats(): CacheStats = cache.stats() + + def cleanAll(): Unit = { + // Clear `broadcastIndex` first so that the removal listener fired by `cleanUp` does not + // observe stale entries after the index has been cleared. + broadcastIndex.synchronized(broadcastIndex.clear()) + cache.invalidateAll() + cache.cleanUp() + } +} diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala index 6c7da6f0f1..908fff2774 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala @@ -489,6 +489,99 @@ class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite { } } + test("driver-side broadcast hash table cache is reused by an identical later query") { + withSQLConf( + ("spark.sql.autoBroadcastJoinThreshold", "10MB"), + ("spark.sql.adaptive.enabled", "false"), + (VeloxConfig.VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_BUILD.key, "true"), + (VeloxConfig.VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_CACHE_ENABLED.key, "true"), + (VeloxConfig.VELOX_BROADCAST_BUILD_RELATION_USE_OFFHEAP.key, "true") + ) { + withTable("cached_bhj_fact", "cached_bhj_dim") { + spark.range(0, 200).selectExpr("id as k", "id % 7 as v").write + .saveAsTable("cached_bhj_fact") + spark.range(0, 50).selectExpr("id as k", "concat('dim_', cast(id as string)) as name").write + .saveAsTable("cached_bhj_dim") + + val query = + """ + |SELECT /*+ BROADCAST(d) */ + | f.k, f.v, d.name + |FROM cached_bhj_fact f + |JOIN cached_bhj_dim d + |ON f.k = d.k + |ORDER BY f.k + |""".stripMargin + + val first = spark.sql(query) + val expected = first.collect().toSeq + val firstRelations = collectBroadcastRelations(first) + assert( + firstRelations.size == 1 && + firstRelations.head.isInstanceOf[SerializedHashTableBroadcastRelation], + s"Expected a single serialized broadcast relation, got $firstRelations" + ) + assert( + VeloxDriverBroadcastRelationCache.size() == 1, + s"Expected the built relation to be cached, got " + + s"${VeloxDriverBroadcastRelationCache.size()} entries" + ) + // The relation cache owns the serialized hash table, the per-exchange cache must not + // track it as well. + assert(VeloxBroadcastBuildSideCache.driverSerializedCacheSize() == 0) + + val hitsBefore = VeloxDriverBroadcastRelationCache.stats().hitCount() + + val second = spark.sql(query) + checkAnswer(second, expected) + assert( + VeloxDriverBroadcastRelationCache.stats().hitCount() > hitsBefore, + "The second query should have hit the driver-side broadcast relation cache") + assert( + VeloxDriverBroadcastRelationCache.size() == 1, + "The second query must not add another entry for the same build side") + + // A query that reuses a relation does not run the build side job, it still has to report + // the statistics of the exchange. + val exchanges = collectWithSubqueries(second.queryExecution.executedPlan) { + case exchange: ColumnarBroadcastExchangeExec => exchange + } + assert(exchanges.size == 1) + assert(exchanges.head.metrics("numOutputRows").value == 50) + assert(exchanges.head.metrics("dataSize").value > 0) + + VeloxBroadcastBuildSideCache.cleanAll() + assert(VeloxDriverBroadcastRelationCache.size() == 0) + } + } + } + + test("driver-side broadcast hash table cache is not used for different build keys") { + withSQLConf( + ("spark.sql.autoBroadcastJoinThreshold", "10MB"), + ("spark.sql.adaptive.enabled", "false"), + (VeloxConfig.VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_BUILD.key, "true"), + (VeloxConfig.VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_CACHE_ENABLED.key, "true") + ) { + withTable("key_bhj_fact", "key_bhj_dim") { + spark.range(0, 100).selectExpr("id as k1", "id % 5 as k2").write + .saveAsTable("key_bhj_fact") + spark.range(0, 20).selectExpr("id as k1", "id % 5 as k2", "id as v").write + .saveAsTable("key_bhj_dim") + + val onK1 = spark.sql("""SELECT /*+ BROADCAST(d) */ count(*) FROM key_bhj_fact f + | JOIN key_bhj_dim d ON f.k1 = d.k1""".stripMargin) + val onK2 = spark.sql("""SELECT /*+ BROADCAST(d) */ count(*) FROM key_bhj_fact f + | JOIN key_bhj_dim d ON f.k2 = d.k2""".stripMargin) + checkAnswer(onK1, Seq(Row(20))) + checkAnswer(onK2, Seq(Row(400))) + assert( + VeloxDriverBroadcastRelationCache.size() == 2, + "Build sides joined on different keys must not share a hash table") + } + } + } + test("Broadcast join with multiple cast expressions in join keys") { withSQLConf( ("spark.sql.autoBroadcastJoinThreshold", "10MB"), diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 712260f984..fdb74012e5 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -9,94 +9,96 @@ nav_order: 16 ## Gluten Velox backend configurations -| Key | Modifiability | Default | Description | -|----------------------------------------------------------------------------------|---------------|-------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| spark.gluten.sql.columnar.backend.velox.IOThreads | ⚓ Static | <undefined> | The Size of the IO thread pool in the Connector. This thread pool is used for split preloading and DirectBufferedInput. By default, the value is the same as the maximum task slots per Spark executor. | -| spark.gluten.sql.columnar.backend.velox.SplitPreloadPerDriver | 🔄 Dynamic | 2 | The split preload per task | -| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinPct | 🔄 Dynamic | 90 | If partial aggregation aggregationPct greater than this value, partial aggregation may be early abandoned. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | -| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinRows | 🔄 Dynamic | 100000 | If partial aggregation input rows number greater than this value, partial aggregation may be early abandoned. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | -| spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping | ⚓ Static | 30000ms | Timeout in milliseconds when waiting for runtime-scoped async work to finish during teardown. | -| spark.gluten.sql.columnar.backend.velox.cacheEnabled | ⚓ Static | false | Enable Velox cache, default off. It's recommended to enablesoft-affinity as well when enable velox cache. | -| spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct | ⚓ Static | 0 | Set prefetch cache min pct for velox file scan | -| spark.gluten.sql.columnar.backend.velox.checkUsageLeak | ⚓ Static | true | Enable check memory usage leak. | -| spark.gluten.sql.columnar.backend.velox.cudf.allowCpuFallback | ⚓ Static | true | Allow cuDF to fall back to CPU execution for unsupported operators. | -| spark.gluten.sql.columnar.backend.velox.cudf.batchSize | 🔄 Dynamic | 2147483647 | Cudf input batch size after shuffle reader | -| spark.gluten.sql.columnar.backend.velox.cudf.concurrentGpuTasks | ⚓ Static | 1 | The number of concurrent GPU tasks to run. | -| spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan | ⚓ Static | false | Enable cudf table scan | -| spark.gluten.sql.columnar.backend.velox.cudf.enableValidation | 🔄 Dynamic | true | Heuristics you can apply to validate a cuDF/GPU plan and only offload when the entire stage can be fully and profitably executed on GPU | -| spark.gluten.sql.columnar.backend.velox.cudf.memoryPercent | ⚓ Static | 50 | The initial percent of GPU memory to allocate for memory resource for one thread. | -| spark.gluten.sql.columnar.backend.velox.cudf.memoryResource | ⚓ Static | async | GPU RMM memory resource. | -| spark.gluten.sql.columnar.backend.velox.cudf.shuffleMaxPrefetchBytes | 🔄 Dynamic | 1028MB | Maximum bytes to prefetch in CPU memory during GPU shuffle read while waiting for GPU available. | -| spark.gluten.sql.columnar.backend.velox.directorySizeGuess | ⚓ Static | 32KB | Deprecated, rename to spark.gluten.sql.columnar.backend.velox.footerEstimatedSize | -| spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild | 🔄 Dynamic | false | Enable driver-side broadcast hash table build. When enabled, the hash table is built and serialized on the driver, then broadcast to executors. When disabled, each executor builds its own hash table from the broadcast data. | -| spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation | 🔄 Dynamic | false | Enable validation fallback for TimestampNTZ type. When true, any plan containing TimestampNTZ will fall back to Spark execution. When false, allows native execution for TimestampNTZ scan. | -| spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | true | Enables caching of open file handles to avoid repeated open/close overhead. Benefits both local filesystems (fewer open/close syscalls and file descriptor churn) and remote filesystems/object stores (reused connection state). Should be disabled if files are mutable, i.e. file content may change while the file path stays the same. | -| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 10m | Expiration time for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). Accepts a Spark duration string (e.g., "10m", "600s") or a plain number interpreted as milliseconds. A value of 0 disables TTL-based eviction. | -| spark.gluten.sql.columnar.backend.velox.filePreloadThreshold | ⚓ Static | 1MB | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold | -| spark.gluten.sql.columnar.backend.velox.floatingPointMode | 🔄 Dynamic | loose | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled. | -| spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation | 🔄 Dynamic | true | Enable flushable aggregation. If true, Gluten will try converting regular aggregation into Velox's flushable aggregation when applicable. A flushable aggregation could emit intermediate result at anytime when memory is full / data reduction ratio is low. | -| spark.gluten.sql.columnar.backend.velox.footerEstimatedSize | ⚓ Static | 32KB | Set the footer estimated size for velox file scan, refer to Velox's footer-estimated-size | -| spark.gluten.sql.columnar.backend.velox.gpuAsyncShuffleReader.enabled | 🔄 Dynamic | false | Experimental: Enable GPU async shuffle reader. When true, the gpu shuffle reader will use a thread pool to read and deserialize the input streams. When false, the shuffle reader will execute in the current thread. | -| spark.gluten.sql.columnar.backend.velox.gpuAsyncShuffleReader.maxPrefetchBytes | 🔄 Dynamic | 1GB | The maximum number of bytes to prefetch in CPU memory during GPU async shuffle read. | -| spark.gluten.sql.columnar.backend.velox.gpuAsyncShuffleReader.threadPoolSize | ⚓ Static | 1 | The number of threads used by GPU async shuffle reader for decompressing and deserializing input streams. | -| spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilter.bypassMinPct | 🔄 Dynamic | 85 | Bypass the build-side Bloom filter when its acceptance percentage reaches this value. | -| spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilter.bypassMinRows | 🔄 Dynamic | 0 | Number of probe rows used to decide whether to bypass the build-side Bloom filter for left outer, existence, and left anti joins. | -| spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilterPushdown.maxSize | 🔄 Dynamic | 0b | The maximum byte size of Bloom filter that can be generated from hash probe. When set to 0, no Bloom filter will be generated. To achieve optimal performance, this should not be too larger than the CPU cache size on the host. | -| spark.gluten.sql.columnar.backend.velox.hashProbe.dynamicFilterPushdown.enabled | 🔄 Dynamic | true | Whether hash probe can generate any dynamic filter (including Bloom filter) and push down to upstream operators. | -| spark.gluten.sql.columnar.backend.velox.hashShuffle.reader.streamMerge.enabled | 🔄 Dynamic | false | Enables a reader-side raw payload merge fast path for plain hash shuffle payloads within each shuffle input stream. This path merges payload buffers before Velox vectors are materialized, so it has lower per-batch overhead than generic VeloxResizeBatchesExec resizing, but it only covers plain payloads. Complex types and dictionary-encoded payloads are not merged by this path. VeloxResizeBatchesExec can still be enabled separately as a generic complement for types and encodings not covered by this fast path. If false, each hash shuffle payload is returned as its own columnar batch. | -| spark.gluten.sql.columnar.backend.velox.loadQuantum | ⚓ Static | 256MB | Set the load quantum for velox file scan, recommend to use the default value (256MB) for performance consideration. If Velox cache is enabled, it can be 8MB at most. | -| spark.gluten.sql.columnar.backend.velox.maxCoalescedBytes | ⚓ Static | 64MB | Set the max coalesced bytes for velox file scan | -| spark.gluten.sql.columnar.backend.velox.maxCoalescedDistance | ⚓ Static | 512KB | Set the max coalesced distance bytes for velox file scan | -| spark.gluten.sql.columnar.backend.velox.maxCompiledRegexes | 🔄 Dynamic | 100 | Controls maximum number of compiled regular expression patterns per function instance per thread of execution. | -| spark.gluten.sql.columnar.backend.velox.maxExtendedPartialAggregationMemory | 🔄 Dynamic | <undefined> | Set the max extended memory of partial aggregation in bytes. When this option is set to a value greater than 0, it will override spark.gluten.sql.columnar.backend.velox.maxExtendedPartialAggregationMemoryRatio. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | -| spark.gluten.sql.columnar.backend.velox.maxExtendedPartialAggregationMemoryRatio | 🔄 Dynamic | 0.15 | Set the max extended memory of partial aggregation as maxExtendedPartialAggregationMemoryRatio of offheap size. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | -| spark.gluten.sql.columnar.backend.velox.maxPartialAggregationMemory | 🔄 Dynamic | <undefined> | Set the max memory of partial aggregation in bytes. When this option is set to a value greater than 0, it will override spark.gluten.sql.columnar.backend.velox.maxPartialAggregationMemoryRatio. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | -| spark.gluten.sql.columnar.backend.velox.maxPartialAggregationMemoryRatio | 🔄 Dynamic | 0.1 | Set the max memory of partial aggregation as maxPartialAggregationMemoryRatio of offheap size. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | -| spark.gluten.sql.columnar.backend.velox.maxPartitionsPerWritersSession | 🔄 Dynamic | 10000 | Maximum number of partitions per a single table writer instance. | -| spark.gluten.sql.columnar.backend.velox.maxSpillBytes | 🔄 Dynamic | 100G | The maximum file size of a query | -| spark.gluten.sql.columnar.backend.velox.maxSpillFileSize | 🔄 Dynamic | 1GB | The maximum size of a single spill file created | -| spark.gluten.sql.columnar.backend.velox.maxSpillLevel | 🔄 Dynamic | 4 | The max allowed spilling level with zero being the initial spilling level | -| spark.gluten.sql.columnar.backend.velox.maxSpillRunRows | 🔄 Dynamic | 3M | The maximum row size of a single spill run | -| spark.gluten.sql.columnar.backend.velox.memCacheSize | ⚓ Static | 1GB | The memory cache size | -| spark.gluten.sql.columnar.backend.velox.memInitCapacity | 🔄 Dynamic | 8MB | The initial memory capacity to reserve for a newly created Velox query memory pool. | -| spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks | 🔄 Dynamic | true | Whether to allow memory capacity transfer between memory pools from different tasks. | -| spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. | -| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent network connections/sockets rather than per-file OS file descriptors, but they can still count toward OS resource limits (ulimit -n). | -| spark.gluten.sql.columnar.backend.velox.orc.scan.enabled | 🔄 Dynamic | true | Enable velox orc scan. If disabled, vanilla spark orc scan will be used. | -| spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes | 🔄 Dynamic | 2MB | The maximum size in bytes for a Parquet dictionary page | -| spark.gluten.sql.columnar.backend.velox.parquet.pageSizeBytes | 🔄 Dynamic | 1MB | The page size in bytes is for compression. | -| spark.gluten.sql.columnar.backend.velox.parquetMaxTargetFileSize | 🔄 Dynamic | 0b | The target file size for each output file when writing data. 0 means no limit on target file size, and the actual file size will be determined by other factors such as max partition number and shuffle batch size. | -| spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for Parquet files. | -| spark.gluten.sql.columnar.backend.velox.prefetchRowGroups | ⚓ Static | 1 | Set the prefetch row groups for velox file scan | -| spark.gluten.sql.columnar.backend.velox.queryTraceEnabled | 🔄 Dynamic | false | Enable query tracing flag. | -| spark.gluten.sql.columnar.backend.velox.reclaimMaxWaitMs | 🔄 Dynamic | 3600000ms | The max time in ms to wait for memory reclaim. | -| spark.gluten.sql.columnar.backend.velox.resizeBatches.copyRanges.enabled | 🔄 Dynamic | true | Enables a VeloxResizeBatchesExec fast path that combines eligible batches using Velox vector copyRanges instead of generic RowVector append. When possible, it collects the small input batches for one VeloxResizeBatchesExec output, allocates the output RowVector once, and bulk-copies child vector ranges. This is most useful for shuffle-read outputs where plain hash shuffle payloads are materialized as dense flat vectors. Complex vectors can also use copyRanges, but ARRAY and MAP still rebuild nested offsets and sizes while bulk-copying child ranges. Unsupported encodings such as dictionary and constant vectors fall back to the generic copy path. This option is enabled by default and complements the reader-side raw payload merge fast path: that path avoids materializing small plain payload batches, while this option optimizes VeloxResizeBatchesExec when that operator is enabled. | -| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput | 🔄 Dynamic | true | If true, combine small columnar batches together before sending to shuffle. The default minimum output batch size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize | -| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput.minSize | 🔄 Dynamic | <undefined> | The minimum batch size for shuffle. If size of an input batch is smaller than the value, it will be combined with other batches before sending to shuffle. Only functions when spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput is set to true. Default value: 0.25 * | -| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInputOutput.minSize | 🔄 Dynamic | <undefined> | The minimum batch size for shuffle input and output. If size of an input batch is smaller than the value, it will be combined with other batches before sending to shuffle. The same applies for batches output by shuffle read. Only functions when spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput or spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput is set to true. Default value: 0.25 * | -| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput | 🔄 Dynamic | false | If true, combine small columnar batches together right after shuffle read. The default minimum output batch size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize | -| spark.gluten.sql.columnar.backend.velox.scan.bloomFilterPushdown.enabled | ⚓ Static | false | Whether to push Bloom filters into Velox scans. | -| spark.gluten.sql.columnar.backend.velox.showTaskMetricsWhenFinished | 🔄 Dynamic | false | Show velox full task metrics when finished. | -| spark.gluten.sql.columnar.backend.velox.spillFileSystem | 🔄 Dynamic | local | The filesystem used to store spill data. local: The local file system. heap-over-local: Write file to JVM heap if having extra heap space. Otherwise write to local file system. | -| spark.gluten.sql.columnar.backend.velox.spillStrategy | 🔄 Dynamic | auto | none: Disable spill on Velox backend; auto: Let Spark memory manager manage Velox's spilling | -| spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads | ⚓ Static | 4 | The number of IO threads for SSD cache read/write operations | -| spark.gluten.sql.columnar.backend.velox.ssdCachePath | ⚓ Static | /tmp | The folder to store the cache files, better on SSD | -| spark.gluten.sql.columnar.backend.velox.ssdCacheShards | ⚓ Static | 1 | The cache shards | -| spark.gluten.sql.columnar.backend.velox.ssdCacheSize | ⚓ Static | 1GB | The SSD cache size, will do memory caching only if this value = 0 | -| spark.gluten.sql.columnar.backend.velox.ssdCheckpointIntervalBytes | ⚓ Static | 0 | Checkpoint after every 'checkpointIntervalBytes' for SSD cache. 0 means no checkpointing. | -| spark.gluten.sql.columnar.backend.velox.ssdChecksumEnabled | ⚓ Static | false | If true, checksum write to SSD is enabled. | -| spark.gluten.sql.columnar.backend.velox.ssdChecksumReadVerificationEnabled | ⚓ Static | false | If true, checksum read verification from SSD is enabled. | -| spark.gluten.sql.columnar.backend.velox.ssdDisableFileCow | ⚓ Static | false | True if copy on write should be disabled. | -| spark.gluten.sql.columnar.backend.velox.ssdODirect | ⚓ Static | false | The O_DIRECT flag for cache writing | -| spark.gluten.sql.columnar.backend.velox.valueStream.dynamicFilter.enabled | 🔄 Dynamic | false | Whether to apply dynamic filters pushed down from hash probe in the ValueStream (shuffle reader) operator to filter rows before they reach the hash join. | -| spark.gluten.sql.enable.enhancedFeatures | 🔄 Dynamic | true | Enable some features including iceberg native write and other features. | -| spark.gluten.sql.rewrite.castArrayToString | 🔄 Dynamic | true | When true, rewrite `cast(array as String)` to `concat('[', array_join(array, ', ', null), ']')` to allow offloading to Velox. | -| spark.gluten.velox.broadcast.build.targetBytesPerThread | ⚓ Static | 32MB | It is used to calculate the number of hash table build threads. Based on our testing across various thresholds (1MB to 128MB), we recommend a value of 32MB or 64MB, as these consistently provided the most significant performance gains. | -| spark.gluten.velox.broadcastBuild.mergeBatches | 🔄 Dynamic | false | If enabled, all columnar batches in a broadcast build relation will be serialized into a single buffer to reduce the number of addInput calls in HashBuild operator. This can significantly improve BHJ performance when the broadcast table has many small batches, but may increase driver-side peak memory and is not suitable for very large broadcasts. | -| spark.gluten.velox.castFromVarcharAddTrimNode | 🔄 Dynamic | false | If true, will add a trim node which has the same semantic as vanilla Spark to CAST-from-varchar.Otherwise, do nothing. | -| spark.gluten.velox.s3MaxConcurrentUploadNum | ⚓ Static | 4 | The maximum number of in-flight S3 part uploads per file. | -| spark.gluten.velox.s3UploadPartAsync | ⚓ Static | false | If true, S3 multipart upload parts are uploaded asynchronously. | -| spark.gluten.velox.s3UploadThreads | ⚓ Static | 16 | The number of shared S3 part upload threads. | +| Key | Modifiability | Default | Description | +|-----------------------------------------------------------------------------------|---------------|-------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| spark.gluten.sql.columnar.backend.velox.IOThreads | ⚓ Static | <undefined> | The Size of the IO thread pool in the Connector. This thread pool is used for split preloading and DirectBufferedInput. By default, the value is the same as the maximum task slots per Spark executor. | +| spark.gluten.sql.columnar.backend.velox.SplitPreloadPerDriver | 🔄 Dynamic | 2 | The split preload per task | +| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinPct | 🔄 Dynamic | 90 | If partial aggregation aggregationPct greater than this value, partial aggregation may be early abandoned. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | +| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinRows | 🔄 Dynamic | 100000 | If partial aggregation input rows number greater than this value, partial aggregation may be early abandoned. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | +| spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping | ⚓ Static | 30000ms | Timeout in milliseconds when waiting for runtime-scoped async work to finish during teardown. | +| spark.gluten.sql.columnar.backend.velox.cacheEnabled | ⚓ Static | false | Enable Velox cache, default off. It's recommended to enablesoft-affinity as well when enable velox cache. | +| spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct | ⚓ Static | 0 | Set prefetch cache min pct for velox file scan | +| spark.gluten.sql.columnar.backend.velox.checkUsageLeak | ⚓ Static | true | Enable check memory usage leak. | +| spark.gluten.sql.columnar.backend.velox.cudf.allowCpuFallback | ⚓ Static | true | Allow cuDF to fall back to CPU execution for unsupported operators. | +| spark.gluten.sql.columnar.backend.velox.cudf.batchSize | 🔄 Dynamic | 2147483647 | Cudf input batch size after shuffle reader | +| spark.gluten.sql.columnar.backend.velox.cudf.concurrentGpuTasks | ⚓ Static | 1 | The number of concurrent GPU tasks to run. | +| spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan | ⚓ Static | false | Enable cudf table scan | +| spark.gluten.sql.columnar.backend.velox.cudf.enableValidation | 🔄 Dynamic | true | Heuristics you can apply to validate a cuDF/GPU plan and only offload when the entire stage can be fully and profitably executed on GPU | +| spark.gluten.sql.columnar.backend.velox.cudf.memoryPercent | ⚓ Static | 50 | The initial percent of GPU memory to allocate for memory resource for one thread. | +| spark.gluten.sql.columnar.backend.velox.cudf.memoryResource | ⚓ Static | async | GPU RMM memory resource. | +| spark.gluten.sql.columnar.backend.velox.cudf.shuffleMaxPrefetchBytes | 🔄 Dynamic | 1028MB | Maximum bytes to prefetch in CPU memory during GPU shuffle read while waiting for GPU available. | +| spark.gluten.sql.columnar.backend.velox.directorySizeGuess | ⚓ Static | 32KB | Deprecated, rename to spark.gluten.sql.columnar.backend.velox.footerEstimatedSize | +| spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild | 🔄 Dynamic | false | Enable driver-side broadcast hash table build. When enabled, the hash table is built and serialized on the driver, then broadcast to executors. When disabled, each executor builds its own hash table from the broadcast data. | +| spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableCache.enabled | 🔄 Dynamic | false | Reuse the hash tables built by driver-side broadcast hash table build across queries of the same application. Cache entries are keyed by the canonicalized build side plan plus the join properties the hash table was built with, so identical broadcast build sides coming from different queries, e.g. the concurrent streams of a TPC-DS throughput run, share one hash table instead of collecting, building and broadcasting it again. Only takes effect when spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild is enabled. Reusing assumes the data behind a build side does not change while the application is running, and the cached hash tables hold driver off-heap memory until they are evicted, so this is disabled by default. | +| spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableCache.maxSize | 🔄 Dynamic | 1GB | The maximum total size of the serialized hash tables kept by the driver-side broadcast hash table cache. Least-recently-used entries are evicted once the limit is exceeded. | +| spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation | 🔄 Dynamic | false | Enable validation fallback for TimestampNTZ type. When true, any plan containing TimestampNTZ will fall back to Spark execution. When false, allows native execution for TimestampNTZ scan. | +| spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled | ⚓ Static | true | Enables caching of open file handles to avoid repeated open/close overhead. Benefits both local filesystems (fewer open/close syscalls and file descriptor churn) and remote filesystems/object stores (reused connection state). Should be disabled if files are mutable, i.e. file content may change while the file path stays the same. | +| spark.gluten.sql.columnar.backend.velox.fileHandleExpirationDurationMs | ⚓ Static | 10m | Expiration time for cached file handles. Handles not accessed within this duration are evicted from the cache. This prevents stale handles from accumulating (e.g., expired HDFS leases, closed remote connections). Accepts a Spark duration string (e.g., "10m", "600s") or a plain number interpreted as milliseconds. A value of 0 disables TTL-based eviction. | +| spark.gluten.sql.columnar.backend.velox.filePreloadThreshold | ⚓ Static | 1MB | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold | +| spark.gluten.sql.columnar.backend.velox.floatingPointMode | 🔄 Dynamic | loose | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled. | +| spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation | 🔄 Dynamic | true | Enable flushable aggregation. If true, Gluten will try converting regular aggregation into Velox's flushable aggregation when applicable. A flushable aggregation could emit intermediate result at anytime when memory is full / data reduction ratio is low. | +| spark.gluten.sql.columnar.backend.velox.footerEstimatedSize | ⚓ Static | 32KB | Set the footer estimated size for velox file scan, refer to Velox's footer-estimated-size | +| spark.gluten.sql.columnar.backend.velox.gpuAsyncShuffleReader.enabled | 🔄 Dynamic | false | Experimental: Enable GPU async shuffle reader. When true, the gpu shuffle reader will use a thread pool to read and deserialize the input streams. When false, the shuffle reader will execute in the current thread. | +| spark.gluten.sql.columnar.backend.velox.gpuAsyncShuffleReader.maxPrefetchBytes | 🔄 Dynamic | 1GB | The maximum number of bytes to prefetch in CPU memory during GPU async shuffle read. | +| spark.gluten.sql.columnar.backend.velox.gpuAsyncShuffleReader.threadPoolSize | ⚓ Static | 1 | The number of threads used by GPU async shuffle reader for decompressing and deserializing input streams. | +| spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilter.bypassMinPct | 🔄 Dynamic | 85 | Bypass the build-side Bloom filter when its acceptance percentage reaches this value. | +| spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilter.bypassMinRows | 🔄 Dynamic | 0 | Number of probe rows used to decide whether to bypass the build-side Bloom filter for left outer, existence, and left anti joins. | +| spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilterPushdown.maxSize | 🔄 Dynamic | 0b | The maximum byte size of Bloom filter that can be generated from hash probe. When set to 0, no Bloom filter will be generated. To achieve optimal performance, this should not be too larger than the CPU cache size on the host. | +| spark.gluten.sql.columnar.backend.velox.hashProbe.dynamicFilterPushdown.enabled | 🔄 Dynamic | true | Whether hash probe can generate any dynamic filter (including Bloom filter) and push down to upstream operators. | +| spark.gluten.sql.columnar.backend.velox.hashShuffle.reader.streamMerge.enabled | 🔄 Dynamic | false | Enables a reader-side raw payload merge fast path for plain hash shuffle payloads within each shuffle input stream. This path merges payload buffers before Velox vectors are materialized, so it has lower per-batch overhead than generic VeloxResizeBatchesExec resizing, but it only covers plain payloads. Complex types and dictionary-encoded payloads are not merged by this path. VeloxResizeBatchesExec can still be enabled separately as a generic complement for types and encodings not covered by this fast path. If false, each hash shuffle payload is returned as its own columnar batch. | +| spark.gluten.sql.columnar.backend.velox.loadQuantum | ⚓ Static | 256MB | Set the load quantum for velox file scan, recommend to use the default value (256MB) for performance consideration. If Velox cache is enabled, it can be 8MB at most. | +| spark.gluten.sql.columnar.backend.velox.maxCoalescedBytes | ⚓ Static | 64MB | Set the max coalesced bytes for velox file scan | +| spark.gluten.sql.columnar.backend.velox.maxCoalescedDistance | ⚓ Static | 512KB | Set the max coalesced distance bytes for velox file scan | +| spark.gluten.sql.columnar.backend.velox.maxCompiledRegexes | 🔄 Dynamic | 100 | Controls maximum number of compiled regular expression patterns per function instance per thread of execution. | +| spark.gluten.sql.columnar.backend.velox.maxExtendedPartialAggregationMemory | 🔄 Dynamic | <undefined> | Set the max extended memory of partial aggregation in bytes. When this option is set to a value greater than 0, it will override spark.gluten.sql.columnar.backend.velox.maxExtendedPartialAggregationMemoryRatio. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | +| spark.gluten.sql.columnar.backend.velox.maxExtendedPartialAggregationMemoryRatio | 🔄 Dynamic | 0.15 | Set the max extended memory of partial aggregation as maxExtendedPartialAggregationMemoryRatio of offheap size. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | +| spark.gluten.sql.columnar.backend.velox.maxPartialAggregationMemory | 🔄 Dynamic | <undefined> | Set the max memory of partial aggregation in bytes. When this option is set to a value greater than 0, it will override spark.gluten.sql.columnar.backend.velox.maxPartialAggregationMemoryRatio. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | +| spark.gluten.sql.columnar.backend.velox.maxPartialAggregationMemoryRatio | 🔄 Dynamic | 0.1 | Set the max memory of partial aggregation as maxPartialAggregationMemoryRatio of offheap size. Note: this option only works when flushable partial aggregation is enabled. Ignored when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false. | +| spark.gluten.sql.columnar.backend.velox.maxPartitionsPerWritersSession | 🔄 Dynamic | 10000 | Maximum number of partitions per a single table writer instance. | +| spark.gluten.sql.columnar.backend.velox.maxSpillBytes | 🔄 Dynamic | 100G | The maximum file size of a query | +| spark.gluten.sql.columnar.backend.velox.maxSpillFileSize | 🔄 Dynamic | 1GB | The maximum size of a single spill file created | +| spark.gluten.sql.columnar.backend.velox.maxSpillLevel | 🔄 Dynamic | 4 | The max allowed spilling level with zero being the initial spilling level | +| spark.gluten.sql.columnar.backend.velox.maxSpillRunRows | 🔄 Dynamic | 3M | The maximum row size of a single spill run | +| spark.gluten.sql.columnar.backend.velox.memCacheSize | ⚓ Static | 1GB | The memory cache size | +| spark.gluten.sql.columnar.backend.velox.memInitCapacity | 🔄 Dynamic | 8MB | The initial memory capacity to reserve for a newly created Velox query memory pool. | +| spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks | 🔄 Dynamic | true | Whether to allow memory capacity transfer between memory pools from different tasks. | +| spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. | +| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent network connections/sockets rather than per-file OS file descriptors, but they can still count toward OS resource limits (ulimit -n). | +| spark.gluten.sql.columnar.backend.velox.orc.scan.enabled | 🔄 Dynamic | true | Enable velox orc scan. If disabled, vanilla spark orc scan will be used. | +| spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes | 🔄 Dynamic | 2MB | The maximum size in bytes for a Parquet dictionary page | +| spark.gluten.sql.columnar.backend.velox.parquet.pageSizeBytes | 🔄 Dynamic | 1MB | The page size in bytes is for compression. | +| spark.gluten.sql.columnar.backend.velox.parquetMaxTargetFileSize | 🔄 Dynamic | 0b | The target file size for each output file when writing data. 0 means no limit on target file size, and the actual file size will be determined by other factors such as max partition number and shuffle batch size. | +| spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for Parquet files. | +| spark.gluten.sql.columnar.backend.velox.prefetchRowGroups | ⚓ Static | 1 | Set the prefetch row groups for velox file scan | +| spark.gluten.sql.columnar.backend.velox.queryTraceEnabled | 🔄 Dynamic | false | Enable query tracing flag. | +| spark.gluten.sql.columnar.backend.velox.reclaimMaxWaitMs | 🔄 Dynamic | 3600000ms | The max time in ms to wait for memory reclaim. | +| spark.gluten.sql.columnar.backend.velox.resizeBatches.copyRanges.enabled | 🔄 Dynamic | true | Enables a VeloxResizeBatchesExec fast path that combines eligible batches using Velox vector copyRanges instead of generic RowVector append. When possible, it collects the small input batches for one VeloxResizeBatchesExec output, allocates the output RowVector once, and bulk-copies child vector ranges. This is most useful for shuffle-read outputs where plain hash shuffle payloads are materialized as dense flat vectors. Complex vectors can also use copyRanges, but ARRAY and MAP still rebuild nested offsets and sizes while bulk-copying child ranges. Unsupported encodings such as dictionary and constant vectors fall back to the generic copy path. This option is enabled by default and complements the reader-side raw payload merge fast path: that path avoids materializing small plain payload batches, while this option optimizes VeloxResizeBatchesExec when that operator is enabled. | +| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput | 🔄 Dynamic | true | If true, combine small columnar batches together before sending to shuffle. The default minimum output batch size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize | +| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput.minSize | 🔄 Dynamic | <undefined> | The minimum batch size for shuffle. If size of an input batch is smaller than the value, it will be combined with other batches before sending to shuffle. Only functions when spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput is set to true. Default value: 0.25 * | +| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInputOutput.minSize | 🔄 Dynamic | <undefined> | The minimum batch size for shuffle input and output. If size of an input batch is smaller than the value, it will be combined with other batches before sending to shuffle. The same applies for batches output by shuffle read. Only functions when spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput or spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput is set to true. Default value: 0.25 * | +| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput | 🔄 Dynamic | false | If true, combine small columnar batches together right after shuffle read. The default minimum output batch size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize | +| spark.gluten.sql.columnar.backend.velox.scan.bloomFilterPushdown.enabled | ⚓ Static | false | Whether to push Bloom filters into Velox scans. | +| spark.gluten.sql.columnar.backend.velox.showTaskMetricsWhenFinished | 🔄 Dynamic | false | Show velox full task metrics when finished. | +| spark.gluten.sql.columnar.backend.velox.spillFileSystem | 🔄 Dynamic | local | The filesystem used to store spill data. local: The local file system. heap-over-local: Write file to JVM heap if having extra heap space. Otherwise write to local file system. | +| spark.gluten.sql.columnar.backend.velox.spillStrategy | 🔄 Dynamic | auto | none: Disable spill on Velox backend; auto: Let Spark memory manager manage Velox's spilling | +| spark.gluten.sql.columnar.backend.velox.ssdCacheIOThreads | ⚓ Static | 4 | The number of IO threads for SSD cache read/write operations | +| spark.gluten.sql.columnar.backend.velox.ssdCachePath | ⚓ Static | /tmp | The folder to store the cache files, better on SSD | +| spark.gluten.sql.columnar.backend.velox.ssdCacheShards | ⚓ Static | 1 | The cache shards | +| spark.gluten.sql.columnar.backend.velox.ssdCacheSize | ⚓ Static | 1GB | The SSD cache size, will do memory caching only if this value = 0 | +| spark.gluten.sql.columnar.backend.velox.ssdCheckpointIntervalBytes | ⚓ Static | 0 | Checkpoint after every 'checkpointIntervalBytes' for SSD cache. 0 means no checkpointing. | +| spark.gluten.sql.columnar.backend.velox.ssdChecksumEnabled | ⚓ Static | false | If true, checksum write to SSD is enabled. | +| spark.gluten.sql.columnar.backend.velox.ssdChecksumReadVerificationEnabled | ⚓ Static | false | If true, checksum read verification from SSD is enabled. | +| spark.gluten.sql.columnar.backend.velox.ssdDisableFileCow | ⚓ Static | false | True if copy on write should be disabled. | +| spark.gluten.sql.columnar.backend.velox.ssdODirect | ⚓ Static | false | The O_DIRECT flag for cache writing | +| spark.gluten.sql.columnar.backend.velox.valueStream.dynamicFilter.enabled | 🔄 Dynamic | false | Whether to apply dynamic filters pushed down from hash probe in the ValueStream (shuffle reader) operator to filter rows before they reach the hash join. | +| spark.gluten.sql.enable.enhancedFeatures | 🔄 Dynamic | true | Enable some features including iceberg native write and other features. | +| spark.gluten.sql.rewrite.castArrayToString | 🔄 Dynamic | true | When true, rewrite `cast(array as String)` to `concat('[', array_join(array, ', ', null), ']')` to allow offloading to Velox. | +| spark.gluten.velox.broadcast.build.targetBytesPerThread | ⚓ Static | 32MB | It is used to calculate the number of hash table build threads. Based on our testing across various thresholds (1MB to 128MB), we recommend a value of 32MB or 64MB, as these consistently provided the most significant performance gains. | +| spark.gluten.velox.broadcastBuild.mergeBatches | 🔄 Dynamic | false | If enabled, all columnar batches in a broadcast build relation will be serialized into a single buffer to reduce the number of addInput calls in HashBuild operator. This can significantly improve BHJ performance when the broadcast table has many small batches, but may increase driver-side peak memory and is not suitable for very large broadcasts. | +| spark.gluten.velox.castFromVarcharAddTrimNode | 🔄 Dynamic | false | If true, will add a trim node which has the same semantic as vanilla Spark to CAST-from-varchar.Otherwise, do nothing. | +| spark.gluten.velox.s3MaxConcurrentUploadNum | ⚓ Static | 4 | The maximum number of in-flight S3 part uploads per file. | +| spark.gluten.velox.s3UploadPartAsync | ⚓ Static | false | If true, S3 multipart upload parts are uploaded asynchronously. | +| spark.gluten.velox.s3UploadThreads | ⚓ Static | 16 | The number of shared S3 part upload threads. | ## Gluten Velox backend *experimental* configurations diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala index 4ca08a5ad6..7f502996ac 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala @@ -24,7 +24,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.SubstraitContext import org.apache.gluten.substrait.expression.WindowFunctionNode -import org.apache.spark.ShuffleDependency +import org.apache.spark.{broadcast, ShuffleDependency} import org.apache.spark.rdd.RDD import org.apache.spark.serializer.Serializer import org.apache.spark.shuffle.{GenShuffleReaderParameters, GenShuffleWriterParameters, GlutenShuffleReaderWrapper, GlutenShuffleWriterWrapper} @@ -454,6 +454,16 @@ trait SparkPlanExecApi { mode.canonicalized } + /** + * The broadcast that was already created for the given relation, if the backend keeps broadcast + * relations alive on the driver to share them between queries. None means the relation has to be + * broadcast. + */ + def cachedBroadcast(relation: BuildSideRelation): Option[broadcast.Broadcast[Any]] = None + + /** Remember the broadcast created for a relation, so that a later query can reuse it. */ + def cacheBroadcast(relation: BuildSideRelation, broadcasted: broadcast.Broadcast[Any]): Unit = {} + /** Create ColumnarWriteFilesExec */ def createColumnarWriteFilesExec( child: WriteFilesExecTransformer, diff --git a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala index f8d5d1f858..cffeb18b9e 100644 --- a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala +++ b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala @@ -85,10 +85,21 @@ case class ColumnarBroadcastExchangeExec(mode: BroadcastMode, child: SparkPlan) val broadcasted = GlutenTimeMetric.millis(longMetric("broadcastTime")) { _ => - // Broadcast the relation - SparkShimLoader.getSparkShims.broadcastInternal( - sparkContext, - relation.asInstanceOf[Any]) + val execApi = BackendsApiManager.getSparkPlanExecApiInstance + execApi.cachedBroadcast(relation) match { + case Some(cached) => + // The relation comes from a driver-side cache and has been broadcast already by a + // previous query, so the very same broadcast can be handed out again. + logInfo("Reusing the broadcast of a cached broadcast relation") + cached + case None => + // Broadcast the relation + val broadcastResult = SparkShimLoader.getSparkShims.broadcastInternal( + sparkContext, + relation.asInstanceOf[Any]) + execApi.cacheBroadcast(relation, broadcastResult) + broadcastResult + } } // Update driver metrics