From fc52b77bfaeb1a7d7f92bbc140b560e5c3a6b9b4 Mon Sep 17 00:00:00 2001
From: Kaifei Yi <yikaifei@apache.org>
Date: Thu, 3 Sep 2026 16:14:43 +0800
Subject: [PATCH] [MINOR][CORE] Make ColumnarInputAdapter inherit from
 GlutenPlan

---
 .../GlutenClickHouseTPCDSMetricsSuite.scala     |  7 +++++--
 .../GlutenClickHouseTPCHMetricsSuite.scala      | 10 ++++++++--
 .../ColumnarCollapseTransformStages.scala       | 17 ++++++-----------
 3 files changed, 19 insertions(+), 15 deletions(-)

diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCDSMetricsSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCDSMetricsSuite.scala
index af025d80846..42093f3ff5c 100644
--- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCDSMetricsSuite.scala
+++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCDSMetricsSuite.scala
@@ -21,7 +21,7 @@ import org.apache.gluten.execution.GlutenPlan
 
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.execution.InputIteratorTransformer
+import org.apache.spark.sql.execution.{ColumnarInputAdapter, InputIteratorTransformer}
 import org.apache.spark.task.TaskResources
 
 import scala.collection.JavaConverters._
@@ -87,7 +87,10 @@ class GlutenClickHouseTPCDSMetricsSuite extends GlutenClickHouseTPCDSAbstractSui
     ) {
       () =>
         val allGlutenPlans = wholeStageTransformer.collect {
-          case g: GlutenPlan if !g.isInstanceOf[InputIteratorTransformer] => g
+          case g: GlutenPlan
+              if !g.isInstanceOf[InputIteratorTransformer] &&
+                !g.isInstanceOf[ColumnarInputAdapter] =>
+            g
         }
 
         assert(allGlutenPlans.size == 30)
diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCHMetricsSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCHMetricsSuite.scala
index f5a70731d56..021ea6ae0dd 100644
--- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCHMetricsSuite.scala
+++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/metrics/GlutenClickHouseTPCHMetricsSuite.scala
@@ -246,7 +246,10 @@ class GlutenClickHouseTPCHMetricsSuite extends ParquetTPCHSuite {
         ) {
           () =>
             val allGlutenPlans = wholeStageTransformer1.collect {
-              case g: GlutenPlan if !g.isInstanceOf[InputIteratorTransformer] => g
+              case g: GlutenPlan
+                  if !g.isInstanceOf[InputIteratorTransformer] &&
+                    !g.isInstanceOf[ColumnarInputAdapter] =>
+                g
             }
 
             val scanPlan = allGlutenPlans(9)
@@ -283,7 +286,10 @@ class GlutenClickHouseTPCHMetricsSuite extends ParquetTPCHSuite {
         ) {
           () =>
             val allGlutenPlans = wholeStageTransformer2.collect {
-              case g: GlutenPlan if !g.isInstanceOf[InputIteratorTransformer] => g
+              case g: GlutenPlan
+                  if !g.isInstanceOf[InputIteratorTransformer] &&
+                    !g.isInstanceOf[ColumnarInputAdapter] =>
+                g
             }
 
             assert(allGlutenPlans.size == 58)
diff --git a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseTransformStages.scala b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseTransformStages.scala
index cf9fcdfd978..1cfca3f27ec 100644
--- a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseTransformStages.scala
+++ b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseTransformStages.scala
@@ -20,7 +20,7 @@ import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution._
 import org.apache.gluten.extension.ApplyStageInputStatsRule
-import org.apache.gluten.extension.columnar.transition.{Convention, ConventionReq}
+import org.apache.gluten.extension.columnar.transition.Convention
 import org.apache.gluten.metrics.MetricsUpdater
 import org.apache.gluten.substrait.SubstraitContext
 import org.apache.gluten.substrait.rel.{InputIteratorRelNode, RelBuilder}
@@ -215,21 +215,16 @@ case class ColumnarCollapseTransformStages(glutenConf: GlutenConfig) extends Rul
   }
 }
 
-// TODO: Make this inherit from GlutenPlan.
 case class ColumnarInputAdapter(child: SparkPlan)
-  extends InputAdapterGenerateTreeStringShim
-  with Convention.KnownBatchType
-  with Convention.KnownRowType
-  with ConventionReq.KnownChildConvention {
+  extends GlutenPlan
+  with InputAdapterGenerateTreeStringShim {
   override def output: Seq[Attribute] = child.output
-  final override val supportsColumnar: Boolean = true
-  final override val supportsRowBased: Boolean = false
+  // Row output is unsupported, so GlutenPlan derives supportsRowBased = false.
   override def rowType(): Convention.RowType = Convention.RowType.None
+  // Columnar output only, so GlutenPlan derives supportsColumnar = true. GlutenPlan's default
+  // requiredChildConvention then requires the same batch type from the child.
   override def batchType(): Convention.BatchType =
     BackendsApiManager.getSettings.primaryBatchType
-  override def requiredChildConvention(): Seq[ConventionReq] = Seq(
-    ConventionReq.ofBatch(
-      ConventionReq.BatchType.Is(BackendsApiManager.getSettings.primaryBatchType)))
   override protected def doExecute(): RDD[InternalRow] = throw new UnsupportedOperationException()
   override protected def doExecuteColumnar(): RDD[ColumnarBatch] = child.executeColumnar()
   override def outputPartitioning: Partitioning = child.outputPartitioning