[VL] Follow up optimizing BHJ by building hash table in driver side like spark#11993
Merged
Conversation
JkSelf
force-pushed
the
driver-build-hashtable
branch
8 times, most recently
from
415a9df to
38347b3
Compare
JkSelf
force-pushed
the
driver-build-hashtable
branch
3 times, most recently
from
37c4e55 to
85251b6
Compare
|
Run Gluten Clickhouse CI on x86 |
2 similar comments
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
JkSelf
force-pushed
the
driver-build-hashtable
branch
from
3059c90 to
1ddf514
Compare
JkSelf
force-pushed
the
driver-build-hashtable
branch
10 times, most recently
from
401e02c to
5c7d7fa
Compare
JkSelf
force-pushed
the
driver-build-hashtable
branch
from
1b7c47c to
d8bb296
Compare
|
Run Gluten Clickhouse CI on x86 |
JkSelf
force-pushed
the
driver-build-hashtable
branch
from
d8bb296 to
8748a71
Compare
|
Run Gluten Clickhouse CI on x86 |
Comment on lines
+935
to
+940
| contexts.find { | ||
| ctx => | ||
| val buildOutputMatches = childOutputSet.subsetOf(ctx.buildOutputSet) && | ||
| ctx.buildOutputSet.subsetOf(childOutputSet) | ||
| buildOutputMatches | ||
| } |
Comment on lines
+74
to
+83
| new RemovalListener[String, SerializedBroadcastHashTable] { | ||
| override def onRemoval( | ||
| key: String, | ||
| value: SerializedBroadcastHashTable, | ||
| cause: RemovalCause): Unit = { | ||
| if (value != null && value.serializedData != null) { | ||
| value.serializedData.release() | ||
| } | ||
| } | ||
| } |
Comment on lines
+124
to
+129
| val serializedSize = HashJoinBuilder.serializedHashTableSizeDirect(hashTableHandle) | ||
| val byteBuffer = JniUnsafeByteBuffer.allocate(serializedSize) | ||
| HashJoinBuilder.serializeHashTableDirect( | ||
| hashTableHandle, | ||
| byteBuffer.address(), | ||
| byteBuffer.size()) |
Comment on lines
+49
to
+53
| out.writeLong(bloomFilterBlocksByteSize) | ||
| out.writeLong(hashProbeDynamicFiltersProduced) | ||
| serializedData.writeExternal(out) | ||
| out.writeObject(buildSideRelation) | ||
| } |
Comment on lines
19
to
22
| CURRENT_DIR=$(cd "$(dirname "$BASH_SOURCE")"; pwd) | ||
| VELOX_REPO=https://github.com/IBM/velox.git | ||
| VELOX_BRANCH=dft-2026_06_30 | ||
| VELOX_REPO=https://github.com/JkSelf/velox.git | ||
| VELOX_BRANCH=dft-2026_06_30_hashtable_ser | ||
| VELOX_ENHANCED_BRANCH=ibm-2026_06_30 |
|
Run Gluten Clickhouse CI on x86 |
JkSelf
force-pushed
the
driver-build-hashtable
branch
from
ead9402 to
1763724
Compare
|
Run Gluten Clickhouse CI on x86 |
Comment on lines
+185
to
189
| if (!dropDuplicates_) { | ||
| uniqueTable_->setAllowDuplicates(true); | ||
| } | ||
| lookup_.reset(); | ||
| } |
Comment on lines
+49
to
+53
| out.writeLong(bloomFilterBlocksByteSize) | ||
| out.writeLong(hashProbeDynamicFiltersProduced) | ||
| serializedData.writeExternal(out) | ||
| out.writeObject(buildSideRelation) | ||
| } |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
Comment on lines
+1069
to
+1076
| // Return SerializedHashTableBroadcastRelation | ||
| SerializedHashTableBroadcastRelation( | ||
| serializedHashTable, | ||
| safeMode, | ||
| newOutput, | ||
| 0L, // buildTimeMs - tracked inside SerializedBroadcastHashTable | ||
| 0L // serializeTimeMs - tracked inside SerializedBroadcastHashTable | ||
| ) |
JkSelf
force-pushed
the
driver-build-hashtable
branch
from
c260a52 to
8c4d8a0
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
Comment on lines
+926
to
+942
| val joinContextOpt: Option[BroadcastJoinContextInfo] = | ||
| findLogicalLink(child).flatMap { | ||
| logicalPlan => | ||
| logicalPlan.getTagValue( | ||
| BroadcastJoinContextTag.BROADCAST_JOIN_CONTEXT | ||
| ).flatMap { | ||
| contexts => | ||
| val childOutputSet = AttributeSet(newOutput) | ||
| // Find the context whose build output matches the child's output | ||
| contexts.find { | ||
| ctx => | ||
| val buildOutputMatches = childOutputSet.subsetOf(ctx.buildOutputSet) && | ||
| ctx.buildOutputSet.subsetOf(childOutputSet) | ||
| buildOutputMatches | ||
| } | ||
| } | ||
| } |
Comment on lines
+994
to
+1018
| // Use the join keys from the matched context | ||
| // Since we already matched the context by comparing outputs, | ||
| // we know this is the correct one | ||
| val joinKeys = if (joinContext.buildRight) { | ||
| joinContext.originalRightKeys | ||
| } else { | ||
| joinContext.originalLeftKeys | ||
| } | ||
| val buildContext = BroadcastHashJoinContext( | ||
| buildSideJoinKeys = if (newBuildKeys.nonEmpty) newBuildKeys else joinKeys, | ||
| substraitJoinType = substraitJoinType, | ||
| buildRight = joinContext.buildRight, | ||
| hasMixedFiltCondition = hasMixedFiltCondition, | ||
| isExistenceJoin = joinContext.joinType | ||
| .isInstanceOf[ExistenceJoin], | ||
| buildSideStructure = newOutput, | ||
| filterBuildColumns = filterBuildColumns, | ||
| filterPropagatesNulls = filterPropagatesNulls, | ||
| buildHashTableId = broadcastId, | ||
| isNullAwareAntiJoin = joinContext.isNullAwareAntiJoin, | ||
| bloomFilterPushdownSize = bloomFilterPushdownSize, | ||
| buildHashTableTimeMetric = Option(buildHashTableTimeMetric), | ||
| serializeHashTableTimeMetric = Option(serializeHashTableTimeMetric), | ||
| serializedHashTableSizeMetric = Option(serializedHashTableSizeMetric) | ||
| ) |
Comment on lines
725
to
+729
| "buildHashTableTime" -> SQLMetrics.createTimingMetric( | ||
| sparkContext, | ||
| "time to build hash table") | ||
| "time to build hash table"), | ||
| "deserializeHashTableTime" -> SQLMetrics.createTimingMetric( | ||
| sparkContext, |
Comment on lines
+136
to
+139
| val bloomFilterBlocksByteSize = HashJoinBuilder | ||
| .getHashTableBloomFilterBlocksByteSize(hashTableHandle) | ||
| val hashProbeDynamicFiltersProduced = if (bloomFilterBlocksByteSize > 0) 1L else 0L | ||
|
|
Comment on lines
+259
to
+265
| val startTime = System.nanoTime() | ||
| val jniWrapper = ColumnarBatchSerializerJniWrapper.create(runtime) | ||
| val serializeHandle: Long = { | ||
| val allocator = ArrowBufferAllocators.globalInstance() | ||
| val cSchema = ArrowSchema.allocateNew(allocator) | ||
| val arrowSchema = SparkArrowUtil.toArrowSchema( | ||
| SparkShimLoader.getSparkShims.structFromAttributes(output), |
Comment on lines
+231
to
+239
| val startTime = System.nanoTime() | ||
| val jniWrapper = ColumnarBatchSerializerJniWrapper.create(runtime) | ||
| val serializeHandle: Long = { | ||
| val allocator = ArrowBufferAllocators.globalInstance() | ||
| val cSchema = ArrowSchema.allocateNew(allocator) | ||
| val arrowSchema = SparkArrowUtil.toArrowSchema( | ||
| SparkShimLoader.getSparkShims.structFromAttributes(output), | ||
| SQLConf.get.sessionLocalTimeZone) | ||
| ArrowAbiUtil.exportSchema(allocator, arrowSchema, cSchema) |
Comment on lines
+975
to
+980
| case Some(cond) => | ||
| val buildAttrs = joinContext.buildOutputSet | ||
| val cols: Array[String] = cond.references.toSeq.collect { | ||
| case a: Attribute if buildAttrs.contains(a) => | ||
| ConverterUtils.genColumnNameWithExprId(a) | ||
| }.toArray |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes are proposed in this pull request?
Further optimize bhj performance by building hash table in driver side.
How was this patch tested?
Existing tests
Was this patch authored or co-authored using generative AI tooling?
Generated-by: bob