[GLUTEN-11678][VL] Native validation should check CrossRelNode's expression

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

@@ -24,7 +24,7 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
 import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec, ColumnarShuffleExchangeExec, SortExec, SparkPlan}
 import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, AQEShuffleReadExec}
 import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, SortMergeJoinExec}
 import org.apache.spark.utils.GlutenSuiteUtils
 
 import scala.collection.mutable.ArrayBuffer
@@ -359,4 +359,41 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
       }
     }
   }
+
+  test("fallback when nested loop join has unsupported expression") {
+    withSQLConf(GlutenConfig.RAS_ENABLED.key -> "false") {
+      val events = new ArrayBuffer[GlutenPlanFallbackEvent]
+      val listener = new SparkListener {
+        override def onOtherEvent(event: SparkListenerEvent): Unit = {
+          event match {
+            case e: GlutenPlanFallbackEvent => events.append(e)
+            case _ =>
+          }
+        }
+      }
+      spark.sparkContext.addSparkListener(listener)
+
+      try {
+        val df = spark.sql("""
+                             |select tmp1.c1, tmp1.c2 from tmp1
+                             |left join tmp2 on (
+                             |  tmp1.c1 = regexp_extract(tmp2.c1, '(?<=@)[^.]+(?=\.)', 0)
+                             |  or tmp2.c1 > 10
+                             |)
+                             |""".stripMargin)
+        df.collect()
+        GlutenSuiteUtils.waitUntilEmpty(spark.sparkContext)
+
+        val nestedLoopJoin = find(df.queryExecution.executedPlan) {
+          _.isInstanceOf[BroadcastNestedLoopJoinExec]
+        }
+        assert(nestedLoopJoin.isDefined)
+        val fallbackReasons = events.flatMap(_.fallbackNodeToReason.values)
+        assert(fallbackReasons.nonEmpty)
+        assert(fallbackReasons.forall(_.contains("regexp_extract due to Pattern")))
+      } finally {
+        spark.sparkContext.removeSparkListener(listener)
+      }
+    }
+  }
 }

cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc

@@ -1124,6 +1124,9 @@ bool SubstraitToVeloxPlanValidator::validate(const ::substrait::CrossRel& crossR
   auto rowType = std::make_shared<RowType>(std::move(names), std::move(types));
 
   if (crossRel.has_expression()) {
+    if (!validateExpression(crossRel.expression(), rowType)) {
+      return false;
+    }
     auto expression = exprConverter_->toVeloxExpr(crossRel.expression(), rowType);
     exec::ExprSet exprSet({std::move(expression)}, execCtx_.get());
   }