adding CoalesceBatches

Author: comphead

native/core/src/execution/planner.rs

@@ -1187,7 +1187,10 @@ impl PhysicalPlanner {
                 if join.filter.is_some() {
                     // SMJ with join filter produces lots of tiny batches
                     let coalesce_batches: Arc<dyn ExecutionPlan> =
-                        Arc::new(CoalesceBatchesExec::new(join.clone(), 8192));
+                        Arc::new(CoalesceBatchesExec::new(
+                            Arc::clone(&join),
+                            self.session_ctx.state().config_options().batch_size(),
+                        ));
                     Ok((
                         scans,
                         Arc::new(SparkPlan::new_with_additional(

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

@@ -21,15 +21,14 @@ package org.apache.comet.exec
 
 import org.scalactic.source.Position
 import org.scalatest.Tag
-
 import org.apache.spark.sql.CometTestBase
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
 import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.Decimal
-
 import org.apache.comet.CometConf
+import org.apache.comet.CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED
 import org.apache.comet.CometSparkSessionExtensions.isSpark34Plus
 
 class CometJoinSuite extends CometTestBase {
@@ -294,8 +293,10 @@ class CometJoinSuite extends CometTestBase {
 
   test("SortMergeJoin without join filter") {
     withSQLConf(
+      CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED.key -> "true",
       SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
-      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1"
+      ) {
       withParquetTable((0 until 10).map(i => (i, i % 5)), "tbl_a") {
         withParquetTable((0 until 10).map(i => (i % 10, i + 2)), "tbl_b") {
           val df1 = sql("SELECT * FROM tbl_a JOIN tbl_b ON tbl_a._2 = tbl_b._1")
@@ -338,8 +339,7 @@ class CometJoinSuite extends CometTestBase {
     }
   }
 
-  // https://github.com/apache/datafusion-comet/issues/398
-  ignore("SortMergeJoin with join filter") {
+  test("SortMergeJoin with join filter") {
     withSQLConf(
       CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED.key -> "true",
       SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",