This is an automated email from the ASF dual-hosted git repository.
zhouyuan 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 54b1439d2f [GLUTEN-12504][VL] Support dumping and replaying BHJ stages
in the micro benchmark (#12592)
54b1439d2f is described below
commit 54b1439d2f39ccce2d512fb0e140a83b9d13479a
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Fri Oct 23 11:13:42 2026 +0300
[GLUTEN-12504][VL] Support dumping and replaying BHJ stages in the micro
benchmark (#12592)
---
.../backendsapi/velox/VeloxSparkPlanExecApi.scala | 19 ++++++++
.../org/apache/gluten/config/VeloxConfig.scala | 5 +-
.../apache/gluten/config/VeloxConfigSuite.scala | 57 ++++++++++++++++++++++
.../gluten/execution/VeloxHashJoinSuite.scala | 43 +++++++++++++++-
cpp/velox/benchmarks/GenericBenchmark.cc | 10 ++--
cpp/velox/jni/JniHashTable.cc | 20 +++++++-
cpp/velox/jni/JniHashTable.h | 9 +++-
cpp/velox/operators/reader/FileReaderIterator.cc | 3 ++
cpp/velox/operators/reader/FileReaderIterator.h | 16 ++++++
.../operators/writer/VeloxColumnarBatchWriter.cc | 6 ++-
cpp/velox/tests/CMakeLists.txt | 1 +
cpp/velox/tests/JniHashTableTest.cc | 50 +++++++++++++++++++
cpp/velox/tests/VeloxWholeStageDumperTest.cc | 32 ++++++++++++
cpp/velox/utils/VeloxWholeStageDumper.cc | 17 +++++++
docs/backends/velox/developers/MicroBenchmarks.md | 15 +++++-
.../org/apache/gluten/config/GlutenConfig.scala | 4 ++
16 files changed, 296 insertions(+), 11 deletions(-)
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 eae0ff97f6..387215ca66 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
@@ -68,11 +68,13 @@ import org.apache.commons.lang3.ClassUtils
import java.net.URI
import java.util.{ArrayList => JArrayList, List => JList}
import java.util.Locale
+import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging {
+ import VeloxSparkPlanExecApi._
/** Transform GetArrayItem to Substrait. */
override def genGetArrayItemTransformer(
@@ -768,6 +770,17 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with
Logging {
// IdentityBroadcastMode
Seq.empty
}
+ if (
+ mode.isInstanceOf[HashedRelationBroadcastMode] &&
+ VeloxConfig.get.benchmarkDumpRequested &&
+
VeloxConfig.get.getConf(VeloxConfig.VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR)
&&
+ warnedBroadcastBuildOncePerExecutorIgnored.compareAndSet(false, true)
+ ) {
+ logWarning(
+ s"${VeloxConfig.VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR.key}
is ignored " +
+ "while whole stage dumping is requested: the broadcast hash table is
built per task " +
+ "so that the micro benchmark can replay the dumped build side.")
+ }
var offload = true
val (newChild, newOutput, newBuildKeys) =
if (VeloxConfig.get.enableBroadcastBuildOncePerExecutor) {
@@ -1749,3 +1762,9 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with
Logging {
}
}
}
+
+object VeloxSparkPlanExecApi {
+ // A new VeloxSparkPlanExecApi is created per call to
Backend.sparkPlanExecApi(), so state that
+ // must be shared across calls lives here.
+ private val warnedBroadcastBuildOncePerExecutorIgnored = new
AtomicBoolean(false)
+}
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 f2d188e818..3345d615d0 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
@@ -70,8 +70,11 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) {
def broadcastBuildMergeBatches: Boolean =
getConf(VELOX_BROADCAST_BUILD_MERGE_BATCHES)
+ // A hash table built once per executor reaches the native join through a
process local cache
+ // rather than through the build side input iterator, so a dumped stage
could not be replayed by
+ // the micro benchmark. Dumping therefore always builds the hash table per
task.
def enableBroadcastBuildOncePerExecutor: Boolean =
- getConf(VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR)
+ getConf(VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR) &&
!benchmarkDumpRequested
def broadcastNestedLoopJoinFullOuterRewriteThreshold: Long =
getConf(VELOX_BROADCAST_NESTED_LOOP_JOIN_FULL_OUTER_REWRITE_THRESHOLD)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/config/VeloxConfigSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/config/VeloxConfigSuite.scala
new file mode 100644
index 0000000000..f91724dbcc
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/config/VeloxConfigSuite.scala
@@ -0,0 +1,57 @@
+/*
+ * 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.config
+
+import org.apache.spark.sql.internal.SQLConf
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class VeloxConfigSuite extends AnyFunSuite {
+
+ private def veloxConfig(entries: (String, String)*): VeloxConfig = {
+ val conf = new SQLConf
+ entries.foreach { case (k, v) => conf.setConfString(k, v) }
+ new VeloxConfig(conf)
+ }
+
+ private val buildOncePerExecutor =
+ VeloxConfig.VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR.key
+
+ test("build hash table once per executor is enabled by default") {
+ val conf = veloxConfig()
+ assert(!conf.benchmarkDumpRequested)
+ assert(conf.enableBroadcastBuildOncePerExecutor)
+ }
+
+ test("whole stage dumping builds the broadcast hash table per task") {
+ // GLUTEN-12504: the micro benchmark can only replay a broadcast hash join
whose build side
+ // was streamed through the input iterator, and so dumped.
+ Seq(
+ GlutenConfig.BENCHMARK_TASK_STAGEID.key -> "1",
+ GlutenConfig.BENCHMARK_TASK_TASK_ID.key -> "3"
+ ).foreach {
+ dumpTrigger =>
+ val conf = veloxConfig(dumpTrigger, buildOncePerExecutor -> "true")
+ assert(conf.benchmarkDumpRequested, dumpTrigger)
+ assert(!conf.enableBroadcastBuildOncePerExecutor, dumpTrigger)
+ }
+ }
+
+ test("build hash table once per executor can still be disabled explicitly") {
+ assert(!veloxConfig(buildOncePerExecutor ->
"false").enableBroadcastBuildOncePerExecutor)
+ }
+}
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 25e5df6e98..8f837cc506 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
@@ -22,10 +22,11 @@ import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec,
ColumnarSubqueryBroadcastExec, InputIteratorTransformer,
SerializedHashTableBroadcastRelation}
+import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec,
ColumnarBuildSideRelation, ColumnarSubqueryBroadcastExec,
InputIteratorTransformer, SerializedHashTableBroadcastRelation}
import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
import org.apache.spark.sql.execution.joins.BuildSideRelation
import org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode
+import org.apache.spark.sql.execution.unsafe.UnsafeColumnarBuildSideRelation
import org.apache.spark.sql.functions.broadcast
class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite {
@@ -487,6 +488,46 @@ class VeloxHashJoinSuite extends
VeloxWholeStageTransformerSuite {
}
}
+ test("GLUTEN-12504: whole stage dumping streams the broadcast build side") {
+ // The micro benchmark can only replay a broadcast hash join whose build
side reaches the native
+ // join through its input iterator. A task id that matches no task
requests dumping without
+ // dumping anything.
+ withSQLConf(
+ ("spark.sql.autoBroadcastJoinThreshold", "10MB"),
+ ("spark.sql.adaptive.enabled", "false"),
+ (VeloxConfig.VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR.key,
"true"),
+ (VeloxConfig.VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_BUILD.key, "true"),
+ (GlutenConfig.BENCHMARK_TASK_TASK_ID.key, "-1")
+ ) {
+ withTable("dump_fact", "dump_dim") {
+ spark.range(0, 400).selectExpr("id as id", "id % 40 as
k").write.saveAsTable("dump_fact")
+ spark.range(0, 40).selectExpr("id as k", "id * 3 as
v").write.saveAsTable("dump_dim")
+
+ val query =
+ """
+ |SELECT /*+ BROADCAST(d) */ f.id, d.v
+ |FROM dump_fact f
+ |JOIN dump_dim d
+ | ON f.k = d.k
+ |""".stripMargin
+
+ runQueryAndCompare(query) {
+ df =>
+ val plan = df.queryExecution.executedPlan
+ assert(plan.collect { case bhj: BroadcastHashJoinExecTransformer
=> bhj }.size == 1)
+
+ val relations = collectBroadcastRelations(df)
+ assert(relations.size == 1)
+ relations.head match {
+ case columnar: ColumnarBuildSideRelation =>
assert(!columnar.offload)
+ case unsafe: UnsafeColumnarBuildSideRelation =>
assert(!unsafe.isOffload)
+ case other => fail(s"Expected the build side to be streamed, got
${other.getClass}")
+ }
+ }
+ }
+ }
+ }
+
test("Broadcast join with multiple cast expressions in join keys") {
withSQLConf(
("spark.sql.autoBroadcastJoinThreshold", "10MB"),
diff --git a/cpp/velox/benchmarks/GenericBenchmark.cc
b/cpp/velox/benchmarks/GenericBenchmark.cc
index feaf1b551f..e71870f122 100644
--- a/cpp/velox/benchmarks/GenericBenchmark.cc
+++ b/cpp/velox/benchmarks/GenericBenchmark.cc
@@ -443,7 +443,8 @@ auto BM_Generic = [](::benchmark::State& state,
inputIters.end(),
std::back_inserter(inputItersRaw),
[](std::shared_ptr<gluten::ResultIterator> iter) {
- return static_cast<FileReaderIterator*>(iter->getInputIter());
+ // Null for an empty input marker, which reads no file.
+ return dynamic_cast<FileReaderIterator*>(iter->getInputIter());
});
}
*Runtime::localWriteFilesTempPath() = FLAGS_write_path;
@@ -495,7 +496,7 @@ auto BM_Generic = [](::benchmark::State& state,
readInputTime +=
std::accumulate(inputItersRaw.begin(), inputItersRaw.end(), 0,
[](int64_t sum, FileReaderIterator* iter) {
- return sum + iter->getCollectBatchTime();
+ return iter == nullptr ? sum : sum + iter->getCollectBatchTime();
});
auto* rawIter =
static_cast<gluten::WholeStageResultIterator*>(resultIter->getInputIter());
@@ -551,8 +552,9 @@ auto BM_ShuffleWriteRead = [](::benchmark::State& state,
localDirs,
dataFileDir);
- auto reader =
static_cast<FileReaderIterator*>(resultIter->getInputIter());
- readInputTime += reader->getCollectBatchTime();
+ if (auto* reader =
dynamic_cast<FileReaderIterator*>(resultIter->getInputIter())) {
+ readInputTime += reader->getCollectBatchTime();
+ }
}
}
diff --git a/cpp/velox/jni/JniHashTable.cc b/cpp/velox/jni/JniHashTable.cc
index 7db693e3d0..0e5481360a 100644
--- a/cpp/velox/jni/JniHashTable.cc
+++ b/cpp/velox/jni/JniHashTable.cc
@@ -45,14 +45,21 @@ void JniHashTableContext::finalize(JNIEnv* env) {
}
}
-jlong JniHashTableContext::callJavaGet(const std::string& id) const {
+std::optional<jlong> JniHashTableContext::callJavaGet(const std::string& id)
const {
+ if (vm_ == nullptr) {
+ return std::nullopt;
+ }
+
JNIEnv* env;
if (vm_->GetEnv(reinterpret_cast<void**>(&env), jniVersion) != JNI_OK) {
throw gluten::GlutenException("JNIEnv was not attached to current thread");
}
const jstring s = env->NewStringUTF(id.c_str());
+ checkException(env);
auto result = env->CallStaticLongMethod(jniVeloxBroadcastBuildSideCache_,
jniGet_, s);
+ env->DeleteLocalRef(s);
+ checkException(env);
return result;
}
@@ -161,7 +168,16 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
}
long getJoin(const std::string& hashTableId) {
- return JniHashTableContext::getInstance().callJavaGet(hashTableId);
+ const auto handle =
JniHashTableContext::getInstance().callJavaGet(hashTableId);
+ if (!handle.has_value()) {
+ // The JVM side VeloxBroadcastBuildSideCache holding the pre-built hash
tables is unreachable,
+ // e.g. in the standalone micro benchmark. Report a miss, the same way
that cache reports one,
+ // and let the caller build the hash table from the join's build side
input instead.
+ LOG(WARNING) << "No JVM is attached to this process, cannot look up the
pre-built hash table for cache key: "
+ << hashTableId << ". A new hash table will be built from the
build side input.";
+ return 0;
+ }
+ return handle.value();
}
size_t serializedHashTableSize(std::shared_ptr<HashTableBuilder> builder) {
diff --git a/cpp/velox/jni/JniHashTable.h b/cpp/velox/jni/JniHashTable.h
index 48bed2b4f0..531b8c864a 100644
--- a/cpp/velox/jni/JniHashTable.h
+++ b/cpp/velox/jni/JniHashTable.h
@@ -18,6 +18,7 @@
#pragma once
#include <jni.h>
+#include <optional>
#include "memory/ColumnarBatch.h"
#include "memory/VeloxMemoryManager.h"
#include "operators/hashjoin/HashTableBuilder.h"
@@ -52,7 +53,11 @@ class JniHashTableContext {
return hashTableObjStore_.get();
}
- jlong callJavaGet(const std::string& id) const;
+ // Returns the handle registered under the given id by the JVM side
+ // VeloxBroadcastBuildSideCache, or std::nullopt when that cache cannot be
reached at all because
+ // no JVM is attached to this process. The latter is the case for the
standalone micro benchmark
+ // and the native unit tests, where JNI_OnLoad never runs.
+ std::optional<jlong> callJavaGet(const std::string& id) const;
private:
JniHashTableContext() : hashTableObjStore_(ObjectStore::create()) {}
@@ -89,6 +94,8 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
std::vector<std::shared_ptr<ColumnarBatch>>& batches,
std::shared_ptr<facebook::velox::memory::MemoryPool> memoryPool);
+// Returns the handle of the pre-built hash table registered for the given id,
or 0 if there is
+// none. Safe to call from a process with no JVM attached, in which case it
always returns 0.
long getJoin(const std::string& hashTableId);
// Return the exact serialized hash table size for direct buffer allocation.
diff --git a/cpp/velox/operators/reader/FileReaderIterator.cc
b/cpp/velox/operators/reader/FileReaderIterator.cc
index 9528763fbc..10da04159f 100644
--- a/cpp/velox/operators/reader/FileReaderIterator.cc
+++ b/cpp/velox/operators/reader/FileReaderIterator.cc
@@ -31,6 +31,9 @@ std::shared_ptr<gluten::ResultIterator>
FileReaderIterator::getInputIteratorFrom
std::shared_ptr<facebook::velox::memory::MemoryPool> pool) {
std::filesystem::path input{path};
auto suffix = input.extension().string();
+ if (suffix == kEmptyInputMarkerSuffix) {
+ return
std::make_shared<gluten::ResultIterator>(std::make_unique<EmptyInputMarkerIterator>());
+ }
if (suffix == kParquetSuffix) {
if (readerType == FileReaderType::kStream) {
return std::make_shared<gluten::ResultIterator>(
diff --git a/cpp/velox/operators/reader/FileReaderIterator.h
b/cpp/velox/operators/reader/FileReaderIterator.h
index 910746d6a4..474746f018 100644
--- a/cpp/velox/operators/reader/FileReaderIterator.h
+++ b/cpp/velox/operators/reader/FileReaderIterator.h
@@ -17,6 +17,8 @@
#pragma once
+#include <string_view>
+
#include "compute/ResultIterator.h"
#include "memory/ColumnarBatchIterator.h"
#include "velox/common/memory/MemoryPool.h"
@@ -25,6 +27,11 @@ namespace gluten {
enum FileReaderType { kBuffered, kStream, kNone };
+// Suffix of the marker file that the whole stage dumper leaves in place of a
data file when an input
+// iterator produced no batches, e.g. `data_1_2_4_0.parquet.empty`. No parquet
file can be written
+// without a schema, which the writer takes from the first batch.
+inline constexpr std::string_view kEmptyInputMarkerSuffix = ".empty";
+
class FileReaderIterator : public ColumnarBatchIterator {
public:
static std::shared_ptr<ResultIterator> getInputIteratorFromFileReader(
@@ -46,4 +53,13 @@ class FileReaderIterator : public ColumnarBatchIterator {
std::string path_;
};
+// Replays an empty input marker as an input iterator that yields no batches.
It reads no file, so it
+// is not a FileReaderIterator.
+class EmptyInputMarkerIterator final : public ColumnarBatchIterator {
+ public:
+ std::shared_ptr<ColumnarBatch> next() override {
+ return nullptr;
+ }
+};
+
} // namespace gluten
diff --git a/cpp/velox/operators/writer/VeloxColumnarBatchWriter.cc
b/cpp/velox/operators/writer/VeloxColumnarBatchWriter.cc
index 598199f19b..5b94eb2b94 100644
--- a/cpp/velox/operators/writer/VeloxColumnarBatchWriter.cc
+++ b/cpp/velox/operators/writer/VeloxColumnarBatchWriter.cc
@@ -57,7 +57,11 @@ arrow::Status VeloxColumnarBatchWriter::write(const
std::shared_ptr<ColumnarBatc
}
arrow::Status VeloxColumnarBatchWriter::close() {
- writer_->close();
+ // The writer takes its schema from the first batch, so it is only created
once something has been
+ // written. Closing a writer that never received a batch is a no-op rather
than a null dereference.
+ if (writer_) {
+ writer_->close();
+ }
return arrow::Status::OK();
}
} // namespace gluten
diff --git a/cpp/velox/tests/CMakeLists.txt b/cpp/velox/tests/CMakeLists.txt
index 9abc4b6aa5..df00a8db15 100644
--- a/cpp/velox/tests/CMakeLists.txt
+++ b/cpp/velox/tests/CMakeLists.txt
@@ -163,6 +163,7 @@ if(ENABLE_S3)
endif()
add_velox_test(scoped_timer_test SOURCES ScopedTimerTest.cc)
add_velox_test(row_based_checksum_test SOURCES RowBasedChecksumTest.cc)
+add_velox_test(jni_hash_table_test SOURCES JniHashTableTest.cc)
if(BUILD_EXAMPLES)
add_velox_test(my_udf_test SOURCES MyUdfTest.cc)
# UDF loading exercises the production shared library, so use the same
diff --git a/cpp/velox/tests/JniHashTableTest.cc
b/cpp/velox/tests/JniHashTableTest.cc
new file mode 100644
index 0000000000..aa57339de5
--- /dev/null
+++ b/cpp/velox/tests/JniHashTableTest.cc
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+#include "jni/JniHashTable.h"
+
+#include <gtest/gtest.h>
+
+namespace gluten {
+
+// This test binary, like the micro benchmark, runs without a JVM. JNI_OnLoad
is therefore never
+// invoked and JniHashTableContext::vm_ stays null.
+//
+// Looking up the build side of a broadcast hash join used to dereference that
null JavaVM
+// unconditionally, so converting any substrait plan carrying a non-empty
`hashTableId` crashed the
+// process. See https://github.com/apache/gluten/issues/12504.
+class JniHashTableTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ ASSERT_EQ(JniHashTableContext::getInstance().getJavaVM(), nullptr)
+ << "This test asserts behaviour of a process with no JVM attached";
+ }
+};
+
+TEST_F(JniHashTableTest, getJoinReportsMissWithoutJvm) {
+ // Must report a miss rather than segfault. SubstraitToVeloxPlanConverter
treats 0 as "no
+ // pre-built table" and falls back to building one from the build side input.
+ EXPECT_EQ(getJoin("no-such-broadcast-hash-table-id"), 0);
+}
+
+TEST_F(JniHashTableTest, callJavaGetReportsUnreachableCacheWithoutJvm) {
+ // Keeps "the cache says there is no such table" distinguishable from "the
cache cannot be
+ // reached at all", rather than collapsing both into a handle of 0 at this
level.
+
EXPECT_FALSE(JniHashTableContext::getInstance().callJavaGet("any-id").has_value());
+}
+
+} // namespace gluten
diff --git a/cpp/velox/tests/VeloxWholeStageDumperTest.cc
b/cpp/velox/tests/VeloxWholeStageDumperTest.cc
index 1ecb5f376a..7ff2bd4d38 100644
--- a/cpp/velox/tests/VeloxWholeStageDumperTest.cc
+++ b/cpp/velox/tests/VeloxWholeStageDumperTest.cc
@@ -19,11 +19,14 @@
#include "compute/VeloxBackend.h"
#include "memory/VeloxColumnarBatch.h"
#include "memory/VeloxMemoryManager.h"
+#include "operators/reader/FileReaderIterator.h"
#include "velox/vector/tests/utils/VectorTestBase.h"
+#include <fmt/format.h>
#include <gtest/gtest.h>
#include <filesystem>
+#include <vector>
using namespace facebook::velox;
@@ -133,4 +136,33 @@ TEST_F(VeloxWholeStageDumperTest,
dumpedBatchesOutliveReaderIterator) {
ASSERT_EQ(leafPool->usedBytes(), baselineBytes);
}
+// An input iterator that yields nothing used to leave the parquet writer
uninitialized, since the
+// writer takes its schema from the first batch, and closing it dereferenced a
null writer.
+// See https://github.com/apache/gluten/issues/12504.
+TEST_F(VeloxWholeStageDumperTest, dumpEmptyInputIterator) {
+ const SparkTaskInfo taskInfo{/*stageId=*/1, /*partitionId=*/2, /*taskId=*/3,
/*vId=*/4};
+ VeloxWholeStageDumper dumper(taskInfo, saveDir_.string(), 4096, vmm_.get());
+ auto input =
std::make_shared<RowVectorBatchIterator>(std::vector<RowVectorPtr>{});
+
+ std::shared_ptr<ColumnarBatchIterator> dumped;
+ ASSERT_NO_THROW(dumped = dumper.dumpInputIterator(0, input));
+
+ // The task keeps running against what the original iterator produced:
nothing.
+ ASSERT_NE(dumped, nullptr);
+ ASSERT_EQ(dumped->next(), nullptr);
+
+ // No parquet file can be written without a schema, so a marker takes its
place.
+ const auto marker = saveDir_ / "data_1_2_4_0.parquet.empty";
+ ASSERT_FALSE(std::filesystem::exists(saveDir_ / "data_1_2_4_0.parquet"));
+ ASSERT_TRUE(std::filesystem::exists(marker));
+
+ // The benchmark replays the marker passed to --data as an empty input
iterator.
+ for (const auto readerType : {FileReaderType::kStream,
FileReaderType::kBuffered}) {
+ auto replayed = FileReaderIterator::getInputIteratorFromFileReader(
+ readerType, marker.string(), 4096, vmm_->getLeafMemoryPool());
+ ASSERT_NE(replayed, nullptr);
+ ASSERT_FALSE(replayed->hasNext());
+ }
+}
+
} // namespace gluten
diff --git a/cpp/velox/utils/VeloxWholeStageDumper.cc
b/cpp/velox/utils/VeloxWholeStageDumper.cc
index 3f0bb556f8..f1129b82ae 100644
--- a/cpp/velox/utils/VeloxWholeStageDumper.cc
+++ b/cpp/velox/utils/VeloxWholeStageDumper.cc
@@ -119,11 +119,28 @@ std::shared_ptr<ColumnarBatchIterator>
VeloxWholeStageDumper::dumpInputIterator(
batchSize_,
memoryManager_->getAggregateMemoryPool()->addAggregateChild(fmt::format("dump_iterator.{}",
iteratorIndex)));
+ bool wroteBatch = false;
while (auto cb = inputIterator->next()) {
GLUTEN_THROW_NOT_OK(writer->write(cb));
+ wroteBatch = true;
}
GLUTEN_THROW_NOT_OK(writer->close());
+ if (!wroteBatch) {
+ // The writer takes its schema from the first batch, so an iterator that
yielded nothing leaves
+ // no parquet file behind. Leave a marker in its place instead: the
benchmark binds the files
+ // passed to --data to iterator indexes by position, and replays the
marker as an empty input
+ // iterator, so the positions of later iterators still line up.
+ const auto markerName = fileName + std::string(kEmptyInputMarkerSuffix);
+ dumpToStorage(
+ saveDir_,
+ markerName,
+ fmt::format("Input iterator {} of {} produced no batches.\n",
iteratorIndex, taskInfo_.toString()));
+ LOG(INFO) << "Input iterator " << iteratorIndex << " of " << taskInfo_ <<
" produced no batches, dumped as "
+ << markerName << ".";
+ return std::make_shared<EmptyInputMarkerIterator>();
+ }
+
// Velox parquet reader requires leaf memory pool. Do not create a dedicated
child pool here: the reader
// iterator is destroyed as soon as the input stream is exhausted, while
vectors it produced may still be
// referenced by downstream operators or by batches handed to Java. Freeing
them against a destroyed pool
diff --git a/docs/backends/velox/developers/MicroBenchmarks.md
b/docs/backends/velox/developers/MicroBenchmarks.md
index dc395eda0c..c7ff2c1772 100644
--- a/docs/backends/velox/developers/MicroBenchmarks.md
+++ b/docs/backends/velox/developers/MicroBenchmarks.md
@@ -139,7 +139,11 @@ or 4 types of dumped file:
file splits.
- Data file(optional): Parquet formatted, file
name `data_{stageId}_{partitionId}_{vId}_{iteratorIdx}.parquet`. If the
first stage contains one or
- more BHJ operators, there can be one or more input data files. The input
data files of a first
+ more BHJ operators, there can be one or more input data files, see
+ [Broadcast hash join](#broadcast-hash-join). An input iterator that produced
no batches is dumped
+ as an empty marker file
`data_{stageId}_{partitionId}_{vId}_{iteratorIdx}.parquet.empty` instead,
+ since no parquet file can be written without a schema. Pass it to `--data`
like any other data
+ file: the benchmark replays it as an empty input iterator. The input data
files of a first
stage will be loaded as iterators to serve as the inputs for the pipeline:
```
@@ -152,6 +156,15 @@ or 4 types of dumped file:
}
```
+### Broadcast hash join
+
+By default the hash table of a BHJ is built once per executor and handed to
the native join through
+a process-local cache, so its build side never reaches the native operator as
input data. While
+`spark.gluten.sql.benchmark_task.stageId` or
`spark.gluten.sql.benchmark_task.taskId` is set,
+Gluten therefore ignores
`spark.gluten.velox.buildHashTableOncePerExecutor.enabled` for every query
+and builds the hash table in each task from the streamed build side, which is
dumped as a regular
+data file. Note that the replayed stage includes building the hash table in
its measurement.
+
Run benchmark. By default, the result will be printed to stdout. You can use
`--noprint-result` to
suppress this output.
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
index 1459e7fed0..70501f3256 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
@@ -358,6 +358,10 @@ class GlutenConfig(conf: SQLConf) extends
GlutenCoreConfig(conf) {
def benchmarkSaveDir: String = getConf(BENCHMARK_SAVE_DIR)
+ // Whether whole stage dumping for the micro benchmark is requested for some
task. Which tasks
+ // are actually dumped is decided per task, see
DebugUtil.isDumpingEnabledForTask.
+ def benchmarkDumpRequested: Boolean = benchmarkStageId >= 0 ||
benchmarkTaskId.nonEmpty
+
def textInputMaxBlockSize: Long = getConf(TEXT_INPUT_ROW_MAX_BLOCK_SIZE)
def textIputEmptyAsDefault: Boolean = getConf(TEXT_INPUT_EMPTY_AS_DEFAULT)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
