[VL] Add validation for coalesce operator

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.5")) {
+    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/ColumnarCoalesceExec.scala

@@ -16,7 +16,9 @@
  */
 package org.apache.gluten.execution
 
+import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.extension.GlutenPlan
+import org.apache.gluten.extension.ValidationResult
 
 import org.apache.spark.{Partition, SparkContext, TaskContext}
 import org.apache.spark.rdd.RDD
@@ -32,6 +34,16 @@ case class ColumnarCoalesceExec(numPartitions: Int, child: SparkPlan)
 
   override def supportsColumnar: Boolean = true
 
+  override protected def doValidateInternal(): ValidationResult = {
+    BackendsApiManager.getValidatorApiInstance
+      .doSchemaValidate(child.schema)
+      .map {
+        reason =>
+          ValidationResult.notOk(s"Found schema check failure for ${child.schema}, due to: $reason")
+      }
+      .getOrElse(ValidationResult.ok)
+  }
+
   override def output: Seq[Attribute] = child.output
 
   override def outputPartitioning: Partitioning = {