This is an automated email from the ASF dual-hosted git repository.
philo-he pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 1d09d3293d [MINOR] Decouple BHJ/BNLJ type flag from broadcast
build-side cache key (#12309)
1d09d3293d is described below
commit 1d09d3293d14f6defa911b7f3ed1a9ec0bbd6199
Author: Philo He <[email protected]>
AuthorDate: Tue Jun 23 09:52:17 2026 -0700
[MINOR] Decouple BHJ/BNLJ type flag from broadcast build-side cache key
(#12309)
---
.../gluten/vectorized/StorageJoinBuilder.java | 30 ++++++++++++----------
.../execution/CHBroadcastBuildSideCache.scala | 21 +++++++++------
.../CHBroadcastNestedLoopJoinExecTransformer.scala | 11 ++++----
.../execution/CHHashJoinExecTransformer.scala | 10 +++++---
.../apache/spark/rpc/GlutenDriverEndpoint.scala | 8 ++++--
.../org/apache/spark/rpc/GlutenRpcMessages.scala | 4 ++-
.../joins/ClickHouseBuildSideRelation.scala | 27 ++++++++++---------
.../benchmarks/CHHashBuildBenchmark.scala | 14 +++++++---
.../benchmarks/CHStorageJoinBenchmark.scala | 4 +--
.../gluten/execution/HashJoinExecTransformer.scala | 4 +--
.../execution/VeloxBroadcastBuildSideCache.scala | 7 +++--
.../apache/spark/rpc/GlutenDriverEndpoint.scala | 8 ++++--
.../org/apache/spark/rpc/GlutenRpcMessages.scala | 4 ++-
...astJoinBuilder.cpp => BroadcastJoinBuilder.cpp} | 25 +++++++-----------
...oadCastJoinBuilder.h => BroadcastJoinBuilder.h} | 8 +++---
.../Parser/AdvancedParametersParseUtil.cpp | 1 +
.../Parser/RelParsers/CrossRelParser.cpp | 6 ++---
.../Parser/RelParsers/JoinRelParser.cpp | 4 +--
cpp-ch/local-engine/local_engine_jni.cpp | 12 +++++----
cpp/velox/benchmarks/data/plan/q17_joins.json | 2 +-
cpp/velox/jni/VeloxJniWrapper.cc | 4 +--
.../BroadcastNestedLoopJoinExecTransformer.scala | 4 +--
.../gluten/execution/JoinExecTransformer.scala | 12 +++++----
23 files changed, 130 insertions(+), 100 deletions(-)
diff --git
a/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
b/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
index 11127f37ba..f3b39865c6 100644
---
a/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
+++
b/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.vectorized;
-import org.apache.gluten.execution.BroadCastHashJoinContext;
+import org.apache.gluten.execution.BroadcastJoinContext;
import org.apache.gluten.execution.JoinTypeTransform;
import org.apache.gluten.expression.ConverterUtils$;
import org.apache.gluten.utils.SubstraitUtil;
@@ -36,11 +36,12 @@ public class StorageJoinBuilder {
public static native long nativeCloneBuildHashTable(long hashTableData);
private static native long nativeBuild(
- String buildHashTableId,
+ String buildTableId,
byte[] in,
long rowCount,
String joinKeys,
int joinType,
+ boolean isBhj,
boolean hasMixedFiltCondition,
boolean isExistenceJoin,
byte[] namedStruct,
@@ -53,7 +54,7 @@ public class StorageJoinBuilder {
public static long build(
byte[] batches,
long rowCount,
- BroadCastHashJoinContext broadCastContext,
+ BroadcastJoinContext broadcastContext,
List<Expression> newBuildKeys,
List<Attribute> newOutput,
boolean hasNullKeyValues) {
@@ -61,8 +62,8 @@ public class StorageJoinBuilder {
List<Expression> keys;
List<Attribute> output;
if (newBuildKeys.isEmpty()) {
- keys =
JavaConverters.<Expression>seqAsJavaList(broadCastContext.buildSideJoinKeys());
- output =
JavaConverters.<Attribute>seqAsJavaList(broadCastContext.buildSideStructure());
+ keys =
JavaConverters.<Expression>seqAsJavaList(broadcastContext.buildSideJoinKeys());
+ output =
JavaConverters.<Attribute>seqAsJavaList(broadcastContext.buildSideStructure());
} else {
keys = newBuildKeys;
output = newOutput;
@@ -77,24 +78,25 @@ public class StorageJoinBuilder {
.collect(Collectors.joining(","));
int joinType;
- if
(broadCastContext.buildHashTableId().startsWith("BuiltBNLJBroadcastTable-")) {
- joinType =
SubstraitUtil.toCrossRelSubstrait(broadCastContext.joinType()).ordinal();
- } else {
- boolean buildRight = broadCastContext.buildRight();
+ if (broadcastContext.isBhj()) {
+ boolean buildRight = broadcastContext.buildRight();
joinType =
- JoinTypeTransform.toSubstraitJoinType(broadCastContext.joinType(),
buildRight).ordinal();
+ JoinTypeTransform.toSubstraitJoinType(broadcastContext.joinType(),
buildRight).ordinal();
+ } else {
+ joinType =
SubstraitUtil.toCrossRelSubstrait(broadcastContext.joinType()).ordinal();
}
return nativeBuild(
- broadCastContext.buildHashTableId(),
+ broadcastContext.buildTableId(),
batches,
rowCount,
joinKey,
joinType,
- broadCastContext.hasMixedFiltCondition(),
- broadCastContext.isExistenceJoin(),
+ broadcastContext.isBhj(),
+ broadcastContext.hasMixedFiltCondition(),
+ broadcastContext.isExistenceJoin(),
SubstraitUtil.toNameStruct(output).toByteArray(),
- broadCastContext.isNullAwareAntiJoin(),
+ broadcastContext.isNullAwareAntiJoin(),
hasNullKeyValues);
}
}
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastBuildSideCache.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastBuildSideCache.scala
index 91763ca84f..c8dac8f019 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastBuildSideCache.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastBuildSideCache.scala
@@ -36,7 +36,9 @@ case class BroadcastHashTable(pointer: Long, relation:
ClickHouseBuildSideRelati
* The complicated part is due to reuse exchange, where multiple BHJ IDs
correspond to a
* `ClickHouseBuildSideRelation`.
*/
-object CHBroadcastBuildSideCache extends Logging with RemovalListener[String,
BroadcastHashTable] {
+object CHBroadcastBuildSideCache
+ extends Logging
+ with RemovalListener[String, BroadcastHashTable] {
private lazy val expiredTime = SparkEnv.get.conf.getLong(
CHBackendSettings.GLUTEN_CLICKHOUSE_BROADCAST_CACHE_EXPIRED_TIME,
@@ -53,23 +55,23 @@ object CHBroadcastBuildSideCache extends Logging with
RemovalListener[String, Br
def getOrBuildBroadcastHashTable(
broadcast: Broadcast[BuildSideRelation],
- broadCastContext: BroadCastHashJoinContext): BroadcastHashTable = {
+ broadcastContext: BroadcastJoinContext): BroadcastHashTable = {
buildSideRelationCache
.get(
- broadCastContext.buildHashTableId,
- (broadcast_id: String) => {
+ broadcastContext.buildTableId,
+ (broadcastId: String) => {
val (pointer, relation) =
broadcast.value
.asInstanceOf[ClickHouseBuildSideRelation]
- .buildHashTable(broadCastContext)
- logDebug(s"Create bhj $broadcast_id = 0x${pointer.toHexString}")
+ .buildHashTable(broadcastContext)
+ logDebug(s"Create bhj $broadcastId = 0x${pointer.toHexString}")
BroadcastHashTable(pointer, relation)
}
)
}
- /** This is callback from c++ backend. */
+ /** This is called from c++ side. */
def get(broadcastHashtableId: String): Long =
Option(buildSideRelationCache.getIfPresent(broadcastHashtableId))
.map(_.pointer)
@@ -85,7 +87,10 @@ object CHBroadcastBuildSideCache extends Logging with
RemovalListener[String, Br
def cleanAll(): Unit = buildSideRelationCache.invalidateAll()
- override def onRemoval(key: String, value: BroadcastHashTable, cause:
RemovalCause): Unit = {
+ override def onRemoval(
+ key: String,
+ value: BroadcastHashTable,
+ cause: RemovalCause): Unit = {
logDebug(s"Remove bhj $key = 0x${value.pointer.toHexString}")
if (value.relation != null) {
value.relation.reset()
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastNestedLoopJoinExecTransformer.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastNestedLoopJoinExecTransformer.scala
index 5b4b8a3bd7..cf39fa5142 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastNestedLoopJoinExecTransformer.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHBroadcastNestedLoopJoinExecTransformer.scala
@@ -58,20 +58,21 @@ case class CHBroadcastNestedLoopJoinExecTransformer(
GlutenDriverEndpoint.collectResources(executionId, buildBroadcastTableId)
} else {
logWarning(
- s"Can't not trace broadcast table data $buildBroadcastTableId" +
+ s"Cannot trace broadcast table data $buildBroadcastTableId" +
s" because execution id is null." +
s" Will clean up until expire time.")
}
val broadcast = buildPlan.executeBroadcast[BuildSideRelation]()
val context =
- BroadCastHashJoinContext(
+ BroadcastJoinContext(
Seq.empty,
finalJoinType,
buildSide == BuildRight,
false,
joinType.isInstanceOf[ExistenceJoin],
buildPlan.output,
- buildBroadcastTableId)
+ buildBroadcastTableId,
+ false)
val broadcastRDD = CHBroadcastBuildSideRDD(sparkContext, broadcast,
context)
streamedRDD :+ broadcastRDD
}
@@ -98,9 +99,9 @@ case class CHBroadcastNestedLoopJoinExecTransformer(
val joinParametersStr = new StringBuffer("JoinParameters:")
joinParametersStr
.append("isBHJ=")
- .append(1)
+ .append(0)
.append("\n")
- .append("buildHashTableId=")
+ .append("buildBroadcastTableId=")
.append(buildBroadcastTableId)
.append("\n")
val message = StringValue
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashJoinExecTransformer.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashJoinExecTransformer.scala
index f646ae19b2..63630d92ab 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashJoinExecTransformer.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashJoinExecTransformer.scala
@@ -187,7 +187,7 @@ case class CHShuffledHashJoinExecTransformer(
case class CHBroadcastBuildSideRDD(
@transient private val sc: SparkContext,
broadcasted: broadcast.Broadcast[BuildSideRelation],
- broadcastContext: BroadCastHashJoinContext)
+ broadcastContext: BroadcastJoinContext)
extends BroadcastBuildSideRDD(sc, broadcasted) {
override def genBroadcastBuildSideIterator(): Iterator[ColumnarBatch] = {
@@ -196,14 +196,15 @@ case class CHBroadcastBuildSideRDD(
}
}
-case class BroadCastHashJoinContext(
+case class BroadcastJoinContext(
buildSideJoinKeys: Seq[Expression],
joinType: JoinType,
buildRight: Boolean,
hasMixedFiltCondition: Boolean,
isExistenceJoin: Boolean,
buildSideStructure: Seq[Attribute],
- buildHashTableId: String,
+ buildTableId: String,
+ isBhj: Boolean,
isNullAwareAntiJoin: Boolean = false)
case class CHBroadcastHashJoinExecTransformer(
@@ -257,7 +258,7 @@ case class CHBroadcastHashJoinExecTransformer(
}
val broadcast = buildPlan.executeBroadcast[BuildSideRelation]()
val context =
- BroadCastHashJoinContext(
+ BroadcastJoinContext(
buildKeyExprs,
joinType,
buildSide == BuildRight,
@@ -265,6 +266,7 @@ case class CHBroadcastHashJoinExecTransformer(
joinType.isInstanceOf[ExistenceJoin],
buildPlan.output,
buildHashTableId,
+ true,
isNullAwareAntiJoin
)
val broadcastRDD = CHBroadcastBuildSideRDD(sparkContext, broadcast,
context)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
b/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
index 97a6175bdb..4965cf1bd3 100644
---
a/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
@@ -95,7 +95,8 @@ class GlutenDriverEndpoint extends IsolatedRpcEndpoint with
Logging {
}
}
-object GlutenDriverEndpoint extends Logging with RemovalListener[String,
util.Set[String]] {
+object GlutenDriverEndpoint extends Logging
+ with RemovalListener[String, util.Set[String]] {
private lazy val executionResourceExpiredTime = SparkEnv.get.conf.getLong(
GlutenConfig.GLUTEN_RESOURCE_RELATION_EXPIRED_TIME.key,
GlutenConfig.GLUTEN_RESOURCE_RELATION_EXPIRED_TIME.defaultValue.get
@@ -126,7 +127,10 @@ object GlutenDriverEndpoint extends Logging with
RemovalListener[String, util.Se
executionResourceRelation.invalidate(executionId)
}
- override def onRemoval(key: String, value: util.Set[String], cause:
RemovalCause): Unit = {
+ override def onRemoval(
+ key: String,
+ value: util.Set[String],
+ cause: RemovalCause): Unit = {
executorDataMap.forEach(
(_, executor) =>
executor.executorEndpointRef.send(GlutenCleanExecutionResource(key, value)))
}
diff --git
a/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
b/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
index 8127c324b7..b61debc3cc 100644
---
a/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
@@ -32,7 +32,9 @@ object GlutenRpcMessages {
case class GlutenExecutorRemoved(executorId: String) extends GlutenRpcMessage
- case class GlutenCleanExecutionResource(executionId: String,
broadcastHashIds: util.Set[String])
+ case class GlutenCleanExecutionResource(
+ executionId: String,
+ broadcastHashIds: util.Set[String])
extends GlutenRpcMessage
// for mergetree cache
diff --git
a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/joins/ClickHouseBuildSideRelation.scala
b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/joins/ClickHouseBuildSideRelation.scala
index 3c6fde8558..8e685ecabb 100644
---
a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/joins/ClickHouseBuildSideRelation.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/joins/ClickHouseBuildSideRelation.scala
@@ -16,7 +16,7 @@
*/
package org.apache.spark.sql.execution.joins
-import org.apache.gluten.execution.{BroadCastHashJoinContext,
ColumnarNativeIterator}
+import org.apache.gluten.execution.{BroadcastJoinContext,
ColumnarNativeIterator}
import org.apache.gluten.utils.{IteratorUtil, PlanNodesUtil}
import org.apache.gluten.vectorized._
@@ -45,24 +45,23 @@ case class ClickHouseBuildSideRelation(
override def asReadOnlyCopy(): ClickHouseBuildSideRelation = this
private var existHashTableData: Long = 0L
- private var existBroadCastHashJoinContext: BroadCastHashJoinContext = null
+ private var existBroadcastHashJoinContext: BroadcastJoinContext = null
- def buildHashTable(
- broadCastContext: BroadCastHashJoinContext): (Long,
ClickHouseBuildSideRelation) =
+ def buildHashTable(broadcastContext: BroadcastJoinContext): (Long,
ClickHouseBuildSideRelation) =
synchronized {
- if (!couldReuseHashTableData(broadCastContext)) {
+ if (!couldReuseHashTableData(broadcastContext)) {
logDebug(
s"BHJ value size: " +
- s"${broadCastContext.buildHashTableId} = ${batches.length}")
+ s"${broadcastContext.buildTableId} = ${batches.length}")
// Build the hash table
existHashTableData = StorageJoinBuilder.build(
batches,
numOfRows,
- broadCastContext,
+ broadcastContext,
newBuildKeys.asJava,
output.asJava,
hasNullKeyValues)
- existBroadCastHashJoinContext = broadCastContext
+ existBroadcastHashJoinContext = broadcastContext
(existHashTableData, this)
} else {
(StorageJoinBuilder.nativeCloneBuildHashTable(existHashTableData),
null)
@@ -71,13 +70,13 @@ case class ClickHouseBuildSideRelation(
def reset(): Unit = synchronized {
existHashTableData = 0
- existBroadCastHashJoinContext = null
+ existBroadcastHashJoinContext = null
}
- private def couldReuseHashTableData(broadCastContext:
BroadCastHashJoinContext): Boolean = {
+ private def couldReuseHashTableData(broadcastContext: BroadcastJoinContext):
Boolean = {
if (existHashTableData != 0) {
- existBroadCastHashJoinContext.joinType == broadCastContext.joinType &&
- existBroadCastHashJoinContext.hasMixedFiltCondition ==
broadCastContext.hasMixedFiltCondition
+ existBroadcastHashJoinContext.joinType == broadcastContext.joinType &&
+ existBroadcastHashJoinContext.hasMixedFiltCondition ==
broadcastContext.hasMixedFiltCondition
}
false
}
@@ -90,7 +89,7 @@ case class ClickHouseBuildSideRelation(
override def transform(key: Expression): Array[InternalRow] = {
// native block reader
val blockReader = new
CHStreamReader(CHShuffleReadStreamFactory.create(batches, true))
- val broadCastIter: Iterator[ColumnarBatch] =
IteratorUtil.createBatchIterator(blockReader)
+ val broadcastIter: Iterator[ColumnarBatch] =
IteratorUtil.createBatchIterator(blockReader)
val transformProjections = mode match {
case HashedRelationBroadcastMode(k, _) => k
@@ -99,7 +98,7 @@ case class ClickHouseBuildSideRelation(
// Expression compute, return block iterator
val expressionEval = new SimpleExpressionEval(
- new ColumnarNativeIterator(broadCastIter.asJava),
+ new ColumnarNativeIterator(broadcastIter.asJava),
PlanNodesUtil.genProjectionsPlanNode(transformProjections, output))
val proj = UnsafeProjection.create(Seq(key))
diff --git
a/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHHashBuildBenchmark.scala
b/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHHashBuildBenchmark.scala
index 2c36b404ec..4e8f388462 100644
---
a/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHHashBuildBenchmark.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHHashBuildBenchmark.scala
@@ -16,7 +16,7 @@
*/
package org.apache.spark.sql.execution.benchmarks
-import org.apache.gluten.execution.BroadCastHashJoinContext
+import org.apache.gluten.execution.BroadcastJoinContext
import org.apache.gluten.vectorized.StorageJoinBuilder
import org.apache.spark.benchmark.Benchmark
@@ -95,7 +95,7 @@ object CHHashBuildBenchmark extends SqlBasedBenchmark with
CHSqlBasedBenchmark w
}
private def createBroadcastRelation(
- child: SparkPlan): (Array[Byte], Long, BroadCastHashJoinContext) = {
+ child: SparkPlan): (Array[Byte], Long, BroadcastJoinContext) = {
val dataSize = SQLMetrics.createSizeMetric(spark.sparkContext, "size of
files read")
val countsAndBytes = child
@@ -105,7 +105,15 @@ object CHHashBuildBenchmark extends SqlBasedBenchmark with
CHSqlBasedBenchmark w
(
countsAndBytes.flatMap(_._2),
countsAndBytes.map(_._1).sum,
- BroadCastHashJoinContext(Seq(child.output.head), Inner, true, false,
false, child.output, "")
+ BroadcastJoinContext(
+ Seq(child.output.head),
+ Inner,
+ true,
+ false,
+ false,
+ child.output,
+ "",
+ true)
)
}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHStorageJoinBenchmark.scala
b/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHStorageJoinBenchmark.scala
index 322c9521e7..b176542d6e 100644
---
a/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHStorageJoinBenchmark.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHStorageJoinBenchmark.scala
@@ -216,8 +216,8 @@ object CHStorageJoinBenchmark extends SqlBasedBenchmark
with CHSqlBasedBenchmark
def iterateBatch(array: Array[Byte], compressed: Boolean): Int = {
val blockReader = new
CHStreamReader(CHShuffleReadStreamFactory.create(array, compressed))
- val broadCastIter: Iterator[ColumnarBatch] =
IteratorUtil.createBatchIterator(blockReader)
- broadCastIter.foldLeft(0) {
+ val broadcastIter: Iterator[ColumnarBatch] =
IteratorUtil.createBatchIterator(blockReader)
+ broadcastIter.foldLeft(0) {
case (acc, batch) =>
acc + batch.numRows
}
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala
b/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala
index 64a61598b4..06f59acf03 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala
@@ -105,7 +105,7 @@ case class BroadcastHashJoinExecTransformer(
right,
isNullAwareAntiJoin) {
- // Unique ID for built table
+ // Unique ID for the build side.
lazy val buildBroadcastTableId: String = buildPlan.id.toString
override protected lazy val substraitJoinType: JoinRel.JoinType = joinType
match {
@@ -139,7 +139,7 @@ case class BroadcastHashJoinExecTransformer(
GlutenDriverEndpoint.collectResources(executionId, buildBroadcastTableId)
} else {
logWarning(
- s"Can not trace broadcast table data $buildBroadcastTableId" +
+ s"Cannot trace broadcast table data $buildBroadcastTableId" +
s" because execution id is null." +
s" Will clean up until expire time.")
}
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 f394c7314a..451c633be4 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
@@ -78,7 +78,7 @@ object VeloxBroadcastBuildSideCache
)
}
- /** This is callback from c++ backend. */
+ /** This is called from c++ side. */
def get(broadcastHashtableId: String): Long = {
Option(buildSideRelationCache.getIfPresent(broadcastHashtableId))
.map(_.pointer)
@@ -95,7 +95,10 @@ object VeloxBroadcastBuildSideCache
def cleanAll(): Unit = buildSideRelationCache.invalidateAll()
- override def onRemoval(key: String, value: BroadcastHashTable, cause:
RemovalCause): Unit = {
+ override def onRemoval(
+ key: String,
+ value: BroadcastHashTable,
+ cause: RemovalCause): Unit = {
synchronized {
if (value.relation != null) {
value.relation match {
diff --git
a/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
b/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
index be0701ea59..6503bbe6ad 100644
---
a/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenDriverEndpoint.scala
@@ -94,7 +94,8 @@ class GlutenDriverEndpoint extends IsolatedRpcEndpoint with
Logging {
}
}
-object GlutenDriverEndpoint extends Logging with RemovalListener[String,
util.Set[String]] {
+object GlutenDriverEndpoint extends Logging
+ with RemovalListener[String, util.Set[String]] {
private lazy val executionResourceExpiredTime = SparkEnv.get.conf.getLong(
GlutenConfig.GLUTEN_RESOURCE_RELATION_EXPIRED_TIME.key,
GlutenConfig.GLUTEN_RESOURCE_RELATION_EXPIRED_TIME.defaultValue.get
@@ -125,7 +126,10 @@ object GlutenDriverEndpoint extends Logging with
RemovalListener[String, util.Se
executionResourceRelation.invalidate(executionId)
}
- override def onRemoval(key: String, value: util.Set[String], cause:
RemovalCause): Unit = {
+ override def onRemoval(
+ key: String,
+ value: util.Set[String],
+ cause: RemovalCause): Unit = {
executorDataMap.forEach(
(_, executor) =>
executor.executorEndpointRef.send(GlutenCleanExecutionResource(key, value)))
}
diff --git
a/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
b/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
index dec67eed78..cfb1abd395 100644
--- a/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
+++ b/backends-velox/src/main/scala/org/apache/spark/rpc/GlutenRpcMessages.scala
@@ -32,6 +32,8 @@ object GlutenRpcMessages {
case class GlutenExecutorRemoved(executorId: String) extends GlutenRpcMessage
- case class GlutenCleanExecutionResource(executionId: String,
broadcastHashIds: util.Set[String])
+ case class GlutenCleanExecutionResource(
+ executionId: String,
+ broadcastHashIds: util.Set[String])
extends GlutenRpcMessage
}
diff --git a/cpp-ch/local-engine/Join/BroadCastJoinBuilder.cpp
b/cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
similarity index 94%
rename from cpp-ch/local-engine/Join/BroadCastJoinBuilder.cpp
rename to cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
index 838a0e5af5..ca9f729184 100644
--- a/cpp-ch/local-engine/Join/BroadCastJoinBuilder.cpp
+++ b/cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "BroadCastJoinBuilder.h"
+#include "BroadcastJoinBuilder.h"
#include <Compression/CompressedReadBuffer.h>
#include <Interpreters/TableJoin.h>
@@ -41,7 +41,7 @@ extern const int UNKNOWN_TYPE;
namespace local_engine
{
-namespace BroadCastJoinBuilder
+namespace BroadcastJoinBuilder
{
using namespace DB;
static jclass Java_CHBroadcastBuildSideCache = nullptr;
@@ -92,7 +92,7 @@ void cleanBuildHashTable(const std::string & hash_table_id,
jlong instance)
/// Record memory usage in Total Memory Tracker
ThreadFromGlobalPoolNoTracingContextPropagation thread(clean_join);
thread.join();
- LOG_DEBUG(&Poco::Logger::get("BroadCastJoinBuilder"), "Broadcast hash
table {} is cleaned", hash_table_id);
+ LOG_DEBUG(&Poco::Logger::get("BroadcastJoinBuilder"), "Broadcast hash
table {} is cleaned", hash_table_id);
}
std::shared_ptr<StorageJoinFromReadBuffer> getJoin(const std::string & key)
@@ -109,12 +109,6 @@ std::shared_ptr<StorageJoinFromReadBuffer> getJoin(const
std::string & key)
return wrapper;
}
-// A join in cross rel.
-static bool isCrossRelJoin(const std::string & key)
-{
- return key.starts_with("BuiltBNLJBroadcastTable-");
-}
-
static void collectBlocksForCountingRows(NativeReader & block_stream, Block &
header, Blocks & result)
{
ProfileInfo profile;
@@ -160,6 +154,7 @@ std::shared_ptr<StorageJoinFromReadBuffer> buildJoin(
jlong row_count,
const std::string & join_keys,
jint join_type,
+ bool is_bhj,
bool has_mixed_join_condition,
bool is_existence_join,
const std::string & named_struct,
@@ -173,14 +168,12 @@ std::shared_ptr<StorageJoinFromReadBuffer> buildJoin(
DB::JoinKind kind;
DB::JoinStrictness strictness;
- bool is_cross_rel_join = isCrossRelJoin(key);
- if (is_cross_rel_join) assert(key_names.empty()); // cross rel join should
not have join keys
+ if (!is_bhj) assert(key_names.empty()); // cross rel join should not have
join keys
- if (is_cross_rel_join)
- std::tie(kind, strictness) =
JoinUtil::getCrossJoinKindAndStrictness(static_cast<substrait::CrossRel_JoinType>(join_type));
- else
+ if (is_bhj)
std::tie(kind, strictness) =
JoinUtil::getJoinKindAndStrictness(static_cast<substrait::JoinRel_JoinType>(join_type),
is_existence_join);
-
+ else
+ std::tie(kind, strictness) =
JoinUtil::getCrossJoinKindAndStrictness(static_cast<substrait::CrossRel_JoinType>(join_type));
substrait::NamedStruct substrait_struct;
substrait_struct.ParseFromString(named_struct);
@@ -202,7 +195,7 @@ std::shared_ptr<StorageJoinFromReadBuffer> buildJoin(
// For not cross join, we need to add a constant join key column
// to make it behavior like a normal join.
- if (is_cross_rel_join && kind != JoinKind::Cross)
+ if (!is_bhj && kind != JoinKind::Cross)
{
auto data_type_u8 = std::make_shared<DataTypeUInt8>();
UInt8 const_key_val = 0;
diff --git a/cpp-ch/local-engine/Join/BroadCastJoinBuilder.h
b/cpp-ch/local-engine/Join/BroadcastJoinBuilder.h
similarity index 95%
rename from cpp-ch/local-engine/Join/BroadCastJoinBuilder.h
rename to cpp-ch/local-engine/Join/BroadcastJoinBuilder.h
index 52db2e7e75..e57bcd32a3 100644
--- a/cpp-ch/local-engine/Join/BroadCastJoinBuilder.h
+++ b/cpp-ch/local-engine/Join/BroadcastJoinBuilder.h
@@ -15,6 +15,7 @@
* limitations under the License.
*/
#pragma once
+#include <cstdint>
#include <memory>
#include <jni.h>
#include <substrait/algebra.pb.h>
@@ -27,25 +28,26 @@ class ReadBuffer;
namespace local_engine
{
class StorageJoinFromReadBuffer;
-namespace BroadCastJoinBuilder
+namespace BroadcastJoinBuilder
{
-
std::shared_ptr<StorageJoinFromReadBuffer> buildJoin(
const std::string & key,
DB::ReadBuffer & input,
jlong row_count,
const std::string & join_keys,
jint join_type,
+ bool is_bhj,
bool has_mixed_join_condition,
bool is_existence_join,
const std::string & named_struct,
bool is_null_aware_anti_join,
bool has_null_key_values);
void cleanBuildHashTable(const std::string & hash_table_id, jlong instance);
-std::shared_ptr<StorageJoinFromReadBuffer> getJoin(const std::string &
hash_table_id);
+std::shared_ptr<StorageJoinFromReadBuffer> getJoin(const std::string &
hash_table_id);
void init(JNIEnv *);
+
void destroy(JNIEnv *);
}
}
diff --git a/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
b/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
index d65f5b5a50..ac20e9e572 100644
--- a/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
+++ b/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
@@ -138,6 +138,7 @@ JoinOptimizationInfo JoinOptimizationInfo::parse(const
String & advance)
tryAssign(kvs, "isBHJ", info.is_broadcast);
tryAssign(kvs, "isSMJ", info.is_smj);
tryAssign(kvs, "buildHashTableId", info.storage_join_key);
+ tryAssign(kvs, "buildBroadcastTableId", info.storage_join_key);
tryAssign(kvs, "isNullAwareAntiJoin", info.is_null_aware_anti_join);
tryAssign(kvs, "isExistenceJoin", info.is_existence_join);
tryAssign(kvs, "isAnyJoin", info.is_any_join);
diff --git a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
index 716cd22734..6442abb3ed 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
@@ -22,7 +22,7 @@
#include <Interpreters/GraceHashJoin.h>
#include <Interpreters/HashJoin/HashJoin.h>
#include <Interpreters/TableJoin.h>
-#include <Join/BroadCastJoinBuilder.h>
+#include <Join/BroadcastJoinBuilder.h>
#include <Join/StorageJoinFromReadBuffer.h>
#include <Parser/AdvancedParametersParseUtil.h>
#include <Parser/ExpressionParser.h>
@@ -171,7 +171,7 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const
substrait::CrossRel & join, DB:
optimization_info.ParseFromString(join.advanced_extension().optimization().value());
auto join_opt_info =
JoinOptimizationInfo::parse(optimization_info.value());
const auto & storage_join_key = join_opt_info.storage_join_key;
- auto storage_join = join_opt_info.is_broadcast ?
BroadCastJoinBuilder::getJoin(storage_join_key) : nullptr;
+ auto storage_join = !storage_join_key.empty() ?
BroadcastJoinBuilder::getJoin(storage_join_key) : nullptr;
if (storage_join)
renamePlanColumns(*left, *right, *storage_join);
auto table_join = createCrossTableJoin(join.type());
@@ -179,7 +179,7 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const
substrait::CrossRel & join, DB:
addConvertStep(*table_join, *left, *right);
// Add a check to find error easily.
- if (!blocksHaveEqualStructure(right_header_before_convert_step,
*right->getCurrentHeader()))
+ if (storage_join &&
!blocksHaveEqualStructure(right_header_before_convert_step,
*right->getCurrentHeader()))
{
throw DB::Exception(
DB::ErrorCodes::LOGICAL_ERROR,
diff --git a/cpp-ch/local-engine/Parser/RelParsers/JoinRelParser.cpp
b/cpp-ch/local-engine/Parser/RelParsers/JoinRelParser.cpp
index 47615fe18d..605f3aea5f 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/JoinRelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/JoinRelParser.cpp
@@ -25,7 +25,7 @@
#include <Interpreters/GraceHashJoin.h>
#include <Interpreters/HashJoin/HashJoin.h>
#include <Interpreters/TableJoin.h>
-#include <Join/BroadCastJoinBuilder.h>
+#include <Join/BroadcastJoinBuilder.h>
#include <Join/StorageJoinFromReadBuffer.h>
#include <Operator/EarlyStopStep.h>
#include <Parser/AdvancedParametersParseUtil.h>
@@ -207,7 +207,7 @@ DB::QueryPlanPtr JoinRelParser::parseJoin(const
substrait::JoinRel & join, DB::Q
optimization_info.ParseFromString(join.advanced_extension().optimization().value());
auto join_opt_info =
JoinOptimizationInfo::parse(optimization_info.value());
LOG_DEBUG(getLogger("JoinRelParser"), "optimization info:{}",
optimization_info.value());
- auto storage_join = join_opt_info.is_broadcast ?
BroadCastJoinBuilder::getJoin(join_opt_info.storage_join_key) : nullptr;
+ auto storage_join = join_opt_info.is_broadcast ?
BroadcastJoinBuilder::getJoin(join_opt_info.storage_join_key) : nullptr;
if (storage_join)
renamePlanColumns(*left, *right, *storage_join);
diff --git a/cpp-ch/local-engine/local_engine_jni.cpp
b/cpp-ch/local-engine/local_engine_jni.cpp
index d70131181f..654078717f 100644
--- a/cpp-ch/local-engine/local_engine_jni.cpp
+++ b/cpp-ch/local-engine/local_engine_jni.cpp
@@ -22,7 +22,7 @@
#include <Columns/ColumnReplicated.h>
#include <Compression/CompressedReadBuffer.h>
#include <DataTypes/DataTypeNullable.h>
-#include <Join/BroadCastJoinBuilder.h>
+#include <Join/BroadcastJoinBuilder.h>
#include <Parser/CHColumnToSparkRow.h>
#include <Parser/LocalExecutor.h>
#include <Parser/ParserContext.h>
@@ -161,7 +161,7 @@ JNIEXPORT jint JNI_OnLoad(JavaVM * vm, void * /*reserved*/)
local_engine::SparkRowToCHColumn::spark_row_iterator_nextBatch =
local_engine::GetMethodID(
env, local_engine::SparkRowToCHColumn::spark_row_interator_class,
"nextBatch", "()Ljava/nio/ByteBuffer;");
- local_engine::BroadCastJoinBuilder::init(env);
+ local_engine::BroadcastJoinBuilder::init(env);
local_engine::CacheManager::initJNI(env);
local_engine::SparkMergeTreeWriterJNI::init(env);
local_engine::SparkRowInfoJNI::init(env);
@@ -206,7 +206,7 @@ JNIEXPORT void
Java_org_apache_gluten_vectorized_ExpressionEvaluatorJniWrapper_n
local_engine::BackendFinalizerUtil::finalizeGlobally();
local_engine::JniErrorsGlobalState::instance().destroy(env);
- local_engine::BroadCastJoinBuilder::destroy(env);
+ local_engine::BroadcastJoinBuilder::destroy(env);
local_engine::SparkMergeTreeWriterJNI::destroy(env);
local_engine::SparkRowInfoJNI::destroy(env);
@@ -1126,6 +1126,7 @@ JNIEXPORT jlong
Java_org_apache_gluten_vectorized_StorageJoinBuilder_nativeBuild
jlong row_count_,
jstring join_key_,
jint join_type_,
+ jboolean is_bhj,
jboolean has_mixed_join_condition,
jboolean is_existence_join,
jbyteArray named_struct,
@@ -1142,12 +1143,13 @@ JNIEXPORT jlong
Java_org_apache_gluten_vectorized_StorageJoinBuilder_nativeBuild
local_engine::ReadBufferFromByteArray read_buffer_from_java_array(in,
length);
DB::CompressedReadBuffer input(read_buffer_from_java_array);
local_engine::configureCompressedReadBuffer(input);
- const auto * obj =
make_wrapper(local_engine::BroadCastJoinBuilder::buildJoin(
+ const auto * obj =
make_wrapper(local_engine::BroadcastJoinBuilder::buildJoin(
hash_table_id,
input,
row_count_,
join_key,
join_type_,
+ is_bhj,
has_mixed_join_condition,
is_existence_join,
struct_string,
@@ -1171,7 +1173,7 @@
Java_org_apache_gluten_vectorized_StorageJoinBuilder_nativeCleanBuildHashTable(J
{
LOCAL_ENGINE_JNI_METHOD_START
auto hash_table_id = jstring2string(env, hash_table_id_);
- local_engine::BroadCastJoinBuilder::cleanBuildHashTable(hash_table_id,
instance);
+ local_engine::BroadcastJoinBuilder::cleanBuildHashTable(hash_table_id,
instance);
LOCAL_ENGINE_JNI_METHOD_END(env, )
}
diff --git a/cpp/velox/benchmarks/data/plan/q17_joins.json
b/cpp/velox/benchmarks/data/plan/q17_joins.json
index 4174b997c3..efb1c50239 100644
--- a/cpp/velox/benchmarks/data/plan/q17_joins.json
+++ b/cpp/velox/benchmarks/data/plan/q17_joins.json
@@ -179,7 +179,7 @@
"advancedExtension": {
"optimization": {
"@type":
"/google.protobuf.StringValue",
- "value":
"inParameters:isBHJ=1\nisNullAwareAntiJoin=0\nbuildHashTableId=BuildedHashTable-421\n"
+ "value":
"inParameters:isBHJ=1\nisNullAwareAntiJoin=0\nbuildHashTableId=421\n"
}
}
}
diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc
index 317b9e56b1..a06c948c0f 100644
--- a/cpp/velox/jni/VeloxJniWrapper.cc
+++ b/cpp/velox/jni/VeloxJniWrapper.cc
@@ -960,7 +960,7 @@ JNIEXPORT jobject JNICALL
Java_org_apache_gluten_execution_IcebergWriteJniWrappe
JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_nativeBuild( // NOLINT
JNIEnv* env,
jobject wrapper,
- jstring tableId,
+ jstring /*tableId*/,
jlongArray batchHandles,
jobjectArray joinKeys,
jobjectArray filterBuildColumns,
@@ -985,8 +985,6 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
queryConf.get<uint32_t>(kAbandonDedupHashMapMinRows,
kAbandonDedupHashMapMinRowsDefault);
const auto abandonHashBuildDedupMinPct =
queryConf.get<uint32_t>(kAbandonDedupHashMapMinPct,
kAbandonDedupHashMapMinPctDefault);
- const auto hashTableId = jStringToCString(env, tableId);
-
// Convert Java String array to C++ vector<string>
std::vector<std::string> hashJoinKeys;
jsize joinKeysCount = env->GetArrayLength(joinKeys);
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
index 32a5793678..8c0f7a8d2a 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
@@ -54,8 +54,8 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
private lazy val substraitJoinType: CrossRel.JoinType =
SubstraitUtil.toCrossRelSubstrait(joinType)
- // Unique ID for builded table
- lazy val buildBroadcastTableId: String = "BuiltBNLJBroadcastTable-" +
buildPlan.id
+ // Unique ID for the build side.
+ lazy val buildBroadcastTableId: String = buildPlan.id.toString
// Hint substrait to switch the left and right,
// since we assume always build right side in substrait.
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala
index 149a0ca729..eac9e6ab81 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala
@@ -301,9 +301,8 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with
TransformSupport {
BackendsApiManager.getTransformerApiInstance.packPBMessage(message)
}
- def genJoinParametersInternal(): (Int, Int, String) = {
- (0, 0, "")
- }
+ // isBHJ, isNullAwareAntiJoin, buildTableId.
+ def genJoinParametersInternal(): (Int, Int, String)
}
object HashJoinLikeExecTransformer {
@@ -354,6 +353,10 @@ abstract class ShuffledHashJoinExecTransformerBase(
override def columnarInputRDDs: Seq[RDD[ColumnarBatch]] = {
getColumnarInputRDDs(streamedPlan) ++ getColumnarInputRDDs(buildPlan)
}
+
+ override def genJoinParametersInternal(): (Int, Int, String) = {
+ (0, 0, buildPlan.id.toString)
+ }
}
abstract class BroadcastHashJoinExecTransformerBase(
@@ -391,8 +394,7 @@ abstract class BroadcastHashJoinExecTransformerBase(
override def joinBuildSide: BuildSide = buildSide
override def hashJoinType: JoinType = joinType
- // Unique ID for builded hash table
- lazy val buildHashTableId: String = "BuiltHashTable-" + buildPlan.id
+ lazy val buildHashTableId: String = buildPlan.id.toString
override def genJoinParametersInternal(): (Int, Int, String) = {
(1, if (isNullAwareAntiJoin) 1 else 0, buildHashTableId)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
