Copilot commented on code in PR #13115: URL: https://github.com/apache/gluten/pull/13115#discussion_r4278643773
########## gluten-ut/spark35/src/test/backends-velox/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala: ########## @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.spark.sql.execution.joins + +import org.apache.gluten.execution.BroadcastHashJoinExecTransformer + +import org.apache.spark.SparkConf +import org.apache.spark.sql.GlutenSQLTestsTrait +import org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec + +class GlutenParallelBroadcastJoinSuite extends GlutenSQLTestsTrait { + + override def sparkConf: SparkConf = super.sparkConf + .set("spark.sql.adaptive.enabled", "false") + .set("spark.gluten.sql.columnar.backend.velox.IOThreads", "0") + .set("spark.gluten.velox.broadcast.build.targetBytesPerThread", "8KB") + + for (buildOnDriver <- Seq(false, true)) { + testGluten(s"parallel broadcast hash table build with IO disabled, driver=$buildOnDriver") { + withSQLConf( + "spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild" -> + buildOnDriver.toString, + "spark.gluten.velox.minTableRowsForParallelJoinBuild" -> "1", + "spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads" -> "1" Review Comment: The test forces `broadcastHashTableBuildThreads` to `1` but then asserts `buildThreads > 1`. With a single-thread pool, the build cannot legitimately use more than one thread, so this assertion is expected to be flaky or consistently fail depending on how the metric is computed. Either (a) set `broadcastHashTableBuildThreads` to a value > 1 to validate parallelism, or (b) keep it at 1 and assert `buildThreads == 1` (or `<= 1` if the metric can be 0 when parallel build is not triggered). ########## cpp/velox/compute/VeloxBackend.cc: ########## @@ -254,6 +255,10 @@ void VeloxBackend::init( std::make_unique<folly::CPUThreadPoolExecutor>(ioThreads, folly::CPUThreadPoolExecutor::makeLifoSemQueue()); } + const auto broadcastHashTableBuildThreads = + std::max<int32_t>(1, backendConf_->get<int32_t>(kBroadcastHashTableBuildThreads, numTaskSlotsPerExecutor)); + broadcastHashTableBuildExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(broadcastHashTableBuildThreads); Review Comment: This creates an additional `CPUThreadPoolExecutor` unconditionally during backend init. With the default equal to `numTaskSlotsPerExecutor`, this can noticeably increase thread count (and idle thread overhead) even for workloads that never use broadcast hash-table parallel build. Consider making creation conditional (e.g., only when parallel join build is enabled/configured) or lazily initializing on first use; the tradeoff is a small first-use latency vs. reduced steady-state resource usage. ########## docs/velox-configuration.md: ########## @@ -16,6 +16,7 @@ nav_order: 16 | 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.broadcastHashTableBuildThreads | ⚓ Static | <undefined> | Size of the dedicated CPU thread pool for the parallel broadcast hash-table build. By default, the thread pool is sized to the maximum task slots per Spark executor on executors, or to spark.driver.cores on the driver (used when driver-side broadcast hash-table build is enabled), with a minimum of one thread. | Review Comment: The default column is set to `<undefined>`, but the description specifies a concrete defaulting behavior (task slots per executor / `spark.driver.cores`, min 1). Consider updating the default column to reflect this (e.g., `auto (task slots per executor; driver: spark.driver.cores; min 1)`) to avoid confusing users into thinking there is no default. ########## gluten-ut/spark35/src/test/backends-velox/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala: ########## @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.spark.sql.execution.joins + +import org.apache.gluten.execution.BroadcastHashJoinExecTransformer + +import org.apache.spark.SparkConf +import org.apache.spark.sql.GlutenSQLTestsTrait +import org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec + +class GlutenParallelBroadcastJoinSuite extends GlutenSQLTestsTrait { + + override def sparkConf: SparkConf = super.sparkConf + .set("spark.sql.adaptive.enabled", "false") + .set("spark.gluten.sql.columnar.backend.velox.IOThreads", "0") + .set("spark.gluten.velox.broadcast.build.targetBytesPerThread", "8KB") + + for (buildOnDriver <- Seq(false, true)) { + testGluten(s"parallel broadcast hash table build with IO disabled, driver=$buildOnDriver") { + withSQLConf( + "spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild" -> + buildOnDriver.toString, + "spark.gluten.velox.minTableRowsForParallelJoinBuild" -> "1", + "spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads" -> "1" + ) { + withTable("parallel_broadcast_build") { + spark.range(4096) + .selectExpr("id", "sha2(cast(id as string), 256) AS value") + .write.saveAsTable("parallel_broadcast_build") + + val result = spark.sql(""" + |SELECT /*+ BROADCAST(b) */ p.id, b.value + |FROM parallel_broadcast_build p JOIN parallel_broadcast_build b + |ON p.id = b.id + |""".stripMargin) + val expected = spark.table("parallel_broadcast_build").collect().toSeq + checkAnswer(result, expected) + + val plan = result.queryExecution.executedPlan + assert(plan.collect { case j: BroadcastHashJoinExecTransformer => j }.nonEmpty) + val exchanges = plan.collect { case e: ColumnarBroadcastExchangeExec => e } + assert(exchanges.size == 1) + assert(exchanges.head.metrics("buildThreads").value > 1) Review Comment: The test forces `broadcastHashTableBuildThreads` to `1` but then asserts `buildThreads > 1`. With a single-thread pool, the build cannot legitimately use more than one thread, so this assertion is expected to be flaky or consistently fail depending on how the metric is computed. Either (a) set `broadcastHashTableBuildThreads` to a value > 1 to validate parallelism, or (b) keep it at 1 and assert `buildThreads == 1` (or `<= 1` if the metric can be 0 when parallel build is not triggered). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
