Support flink engine under the select statement, the results can be read in a stream

Author: waywtdcc

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/SchemaHelper.scala

@@ -142,7 +142,12 @@ object SchemaHelper {
           .contains(dt.getClass.getSimpleName) => Some(dt.defaultSize)
     case dt @ (BooleanType | _: NumericType | DateType | TimestampType |
         CalendarIntervalType | NullType) =>
-      Some(dt.defaultSize)
+      // decimal type
+      if (dt.isInstanceOf[DecimalType]) {
+        Some(dt.asInstanceOf[DecimalType].precision)
+      } else {
+        Some(dt.defaultSize)
+      }
     case StructType(fields) =>
       val sizeArr = fields.map(f => getColumnSize(f.dataType))
       if (sizeArr.contains(None)) {

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala

@@ -154,6 +154,7 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
           val colSize = rowSet.getInt(COLUMN_SIZE)
           schema(pos).dataType match {
             case StringType | BinaryType | _: ArrayType | _: MapType => assert(colSize === 0)
+            case d: DecimalType => assert(colSize === d.precision)
             case StructType(fields) if fields.length == 1 => assert(colSize === 0)
             case o => assert(colSize === o.defaultSize)
           }