DNM: [VL] Test

backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala

@@ -463,11 +463,20 @@ object VeloxConfig extends ConfigRegistry {
   val HASH_PROBE_DYNAMIC_FILTER_PUSHDOWN_ENABLED =
     buildConf("spark.gluten.sql.columnar.backend.velox.hashProbe.dynamicFilterPushdown.enabled")
       .doc(
-        "Whether hash probe can generate any dynamic filter (including Bloom filter) and push" +
-          " down to upstream operators.")
+        "Whether hash probe can generate any dynamic filter (including Bloom filter) and push " +
+          "down to upstream operators.")
       .booleanConf
       .createWithDefault(true)
 
+  val HASH_PROBE_STRING_DYNAMIC_FILTER_PUSHDOWN_ENABLED =
+    buildConf(
+      "spark.gluten.sql.columnar.backend.velox.hashProbe.stringDynamicFilterPushdown.enabled")
+      .doc(
+        "Whether hash probe can generate any dynamic filter (including Bloom filter) for string " +
+          "types and push down to upstream operators.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED =
     buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled")
       .doc(

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala

@@ -342,4 +342,42 @@ class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite {
       }
     }
   }
+
+  test("Hash probe dynamic filter pushdown - string") {
+    withSQLConf(
+      VeloxConfig.HASH_PROBE_DYNAMIC_FILTER_PUSHDOWN_ENABLED.key -> "true",
+      VeloxConfig.HASH_PROBE_STRING_DYNAMIC_FILTER_PUSHDOWN_ENABLED.key -> "true"
+    ) {
+      withTable("probe_table_str", "build_table_str") {
+
+        spark.sql("""
+        CREATE TABLE probe_table_str USING PARQUET AS
+        SELECT CAST(id AS STRING) AS a
+        FROM range(100000)
+      """)
+
+        spark.sql("""
+        CREATE TABLE build_table_str USING PARQUET AS
+        SELECT CAST(id * 1000 AS STRING) AS b
+        FROM range(10)
+      """)
+
+        runQueryAndCompare(
+          "SELECT a FROM probe_table_str JOIN build_table_str ON a = b"
+        ) {
+          df =>
+            val join = find(df.queryExecution.executedPlan) {
+              case _: BroadcastHashJoinExecTransformer => true
+              case _ => false
+            }
+            assert(join.isDefined)
+
+            val metrics = join.get.metrics
+
+            assert(metrics.contains("hashProbeDynamicFiltersProduced"))
+            assert(metrics("hashProbeDynamicFiltersProduced").value == 1)
+        }
+      }
+    }
+  }
 }

cpp/velox/compute/WholeStageResultIterator.cc

@@ -650,6 +650,8 @@ std::unordered_map<std::string, std::string> WholeStageResultIterator::getQueryC
 
     configs[velox::core::QueryConfig::kHashProbeDynamicFilterPushdownEnabled] =
         std::to_string(veloxCfg_->get<bool>(kHashProbeDynamicFilterPushdownEnabled, true));
+    configs[velox::core::QueryConfig::kHashProbeStringDynamicFilterPushdownEnabled] =
+        std::to_string(veloxCfg_->get<bool>(kHashProbeStringDynamicFilterPushdownEnabled, false));
     configs[velox::core::QueryConfig::kHashProbeBloomFilterPushdownMaxSize] =
         std::to_string(veloxCfg_->get<uint64_t>(kHashProbeBloomFilterPushdownMaxSize, 0));
     // spark.gluten.sql.columnar.backend.velox.SplitPreloadPerDriver takes no effect if

cpp/velox/config/VeloxConfig.h

@@ -74,6 +74,8 @@ const std::string kVeloxSplitPreloadPerDriver = "spark.gluten.sql.columnar.backe
 
 const std::string kHashProbeDynamicFilterPushdownEnabled =
     "spark.gluten.sql.columnar.backend.velox.hashProbe.dynamicFilterPushdown.enabled";
+const std::string kHashProbeStringDynamicFilterPushdownEnabled =
+    "spark.gluten.sql.columnar.backend.velox.hashProbe.stringDynamicFilterPushdown.enabled";
 
 const std::string kHashProbeBloomFilterPushdownMaxSize =
     "spark.gluten.sql.columnar.backend.velox.hashProbe.bloomFilterPushdown.maxSize";