[VL] Add schema validation for all operators

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

@@ -258,6 +258,16 @@ class TestOperator extends VeloxWholeStageTransformerSuite with AdaptiveSparkPla
     checkLengthAndPlan(df, 5)
   }
 
+  testWithSpecifiedSparkVersion("coalesce validation", Some("3.4")) {
+    withTempPath {
+      path =>
+        val data = "2019-09-09 01:02:03.456789"
+        val df = Seq(data).toDF("strTs").selectExpr(s"CAST(strTs AS TIMESTAMP_NTZ) AS ts")
+        df.coalesce(1).write.format("parquet").save(path.getCanonicalPath)
+        spark.read.parquet(path.getCanonicalPath).collect
+    }
+  }
+
   test("groupby") {
     val df = runQueryAndCompare(
       "select l_orderkey, sum(l_partkey) as sum from lineitem " +

gluten-core/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala

@@ -311,15 +311,6 @@ case class ColumnarUnionExec(children: Seq[SparkPlan]) extends SparkPlan with Gl
   }
 
   override protected def doExecuteColumnar(): RDD[ColumnarBatch] = columnarInputRDD
-
-  override protected def doValidateInternal(): ValidationResult = {
-    BackendsApiManager.getValidatorApiInstance
-      .doSchemaValidate(schema)
-      .map {
-        reason => ValidationResult.notOk(s"Found schema check failure for $schema, due to: $reason")
-      }
-      .getOrElse(ValidationResult.ok)
-  }
 }
 
 /**

gluten-core/src/main/scala/org/apache/gluten/extension/GlutenPlan.scala

@@ -65,7 +65,13 @@ trait GlutenPlan extends SparkPlan with Convention.KnownBatchType with LogLevelU
   final def doValidate(): ValidationResult = {
     try {
       TransformerState.enterValidation
-      val res = doValidateInternal()
+      val res = BackendsApiManager.getValidatorApiInstance
+        .doSchemaValidate(schema)
+        .map {
+          reason =>
+            ValidationResult.notOk(s"Found schema check failure for $schema, due to: $reason")
+        }
+        .getOrElse(doValidateInternal())
       if (!res.isValid) {
         TestStats.addFallBackClassName(this.getClass.toString)
       }

gluten-core/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala

@@ -17,7 +17,7 @@
 package org.apache.spark.sql.execution
 
 import org.apache.gluten.backendsapi.BackendsApiManager
-import org.apache.gluten.extension.{GlutenPlan, ValidationResult}
+import org.apache.gluten.extension.GlutenPlan
 import org.apache.gluten.metrics.GlutenTimeMetric
 import org.apache.gluten.sql.shims.SparkShimLoader
 
@@ -133,19 +133,6 @@ case class ColumnarBroadcastExchangeExec(mode: BroadcastMode, child: SparkPlan)
     ColumnarBroadcastExchangeExec(canonicalized, child.canonicalized)
   }
 
-  override protected def doValidateInternal(): ValidationResult = {
-    BackendsApiManager.getValidatorApiInstance
-      .doSchemaValidate(schema)
-      .map {
-        reason =>
-          {
-            ValidationResult.notOk(
-              s"Unsupported schema in broadcast exchange: $schema, reason: $reason")
-          }
-      }
-      .getOrElse(ValidationResult.ok)
-  }
-
   override def doPrepare(): Unit = {
     // Materialize the future.
     relationFuture