[SPARK-32492][SQL] Fulfill missing column meta information COLUMN_SIZE /DECIMAL_DIGITS/NUM_PREC_RADIX/ORDINAL_POSITION for thriftserver client tools

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala

@@ -17,12 +17,10 @@
 
 package org.apache.spark.sql.hive.thriftserver
 
-import java.util.UUID
 import java.util.regex.Pattern
 
 import scala.collection.JavaConverters.seqAsJavaListConverter
 
-import org.apache.commons.lang3.exception.ExceptionUtils
 import org.apache.hadoop.hive.ql.security.authorization.plugin.{HiveOperationType, HivePrivilegeObject}
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrivilegeObject.HivePrivilegeObjectType
 import org.apache.hive.service.cli._
@@ -34,7 +32,7 @@ import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.SessionCatalog
 import org.apache.spark.sql.hive.thriftserver.ThriftserverShimUtils.toJavaSQLType
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types._
 
 /**
  * Spark's own SparkGetColumnsOperation
@@ -126,12 +124,52 @@ private[hive] class SparkGetColumnsOperation(
     HiveThriftServer2.eventManager.onStatementFinish(statementId)
   }
 
+  /**
+   * For numeric and datetime types, it returns the default size of its catalyst type
+   * For struct type, when its elements are fixed-size, the summation of all element sizes will be
+   * returned.
+   * For array, map, string, and binaries, the column size is variable, return null as unknown.
+   */
+  private def getColumnSize(typ: DataType): Option[Int] = typ match {
+    case StringType | BinaryType | _: ArrayType | _: MapType => None
+    case StructType(fields) =>
+      val sizeArr = fields.map(f => getColumnSize(f.dataType))
+      if (sizeArr.contains(None)) {
+        None
+      } else {
+        Some(sizeArr.map(_.get).sum)
+      }
+    case other => Some(other.defaultSize)
+  }
+
+  /**
+   * The number of fractional digits for this type.
+   * Null is returned for data types where this is not applicable.
+   * For boolean and integrals, the decimal digits is 0
+   * For floating types, we follow the IEEE Standard for Floating-Point Arithmetic (IEEE 754)
+   * For timestamp values, we support microseconds
+   * For decimals, it returns the scale
+   */
+  private def getDecimalDigits(typ: DataType) = typ match {
+    case BooleanType | _: IntegerType => Some(0)
+    case FloatType => Some(7)
+    case DoubleType => Some(15)
+    case d: DecimalType => Some(d.scale)
+    case TimestampType => Some(6)
+    case _ => None
+  }
+
+  private def getNumPrecRadix(typ: DataType): Option[Int] = typ match {
+    case _: NumericType => Some(10)
+    case _ => None
+  }
+
   private def addToRowSet(
       columnPattern: Pattern,
       dbName: String,
       tableName: String,
       schema: StructType): Unit = {
-    schema.foreach { column =>
+    schema.zipWithIndex.foreach { case (column, pos) =>
       if (columnPattern != null && !columnPattern.matcher(column.name).matches()) {
       } else {
         val rowData = Array[AnyRef](
@@ -141,17 +179,17 @@ private[hive] class SparkGetColumnsOperation(
           column.name, // COLUMN_NAME
           toJavaSQLType(column.dataType.sql).asInstanceOf[AnyRef], // DATA_TYPE
           column.dataType.sql, // TYPE_NAME
-          null, // COLUMN_SIZE
+          getColumnSize(column.dataType).map(_.asInstanceOf[AnyRef]).orNull, // COLUMN_SIZE
           null, // BUFFER_LENGTH, unused
-          null, // DECIMAL_DIGITS
-          null, // NUM_PREC_RADIX
+          getDecimalDigits(column.dataType).map(_.asInstanceOf[AnyRef]).orNull, // DECIMAL_DIGITS
+          getNumPrecRadix(column.dataType).map(_.asInstanceOf[AnyRef]).orNull, // NUM_PREC_RADIX
           (if (column.nullable) 1 else 0).asInstanceOf[AnyRef], // NULLABLE
           column.getComment().getOrElse(""), // REMARKS
           null, // COLUMN_DEF
           null, // SQL_DATA_TYPE
           null, // SQL_DATETIME_SUB
           null, // CHAR_OCTET_LENGTH
-          null, // ORDINAL_POSITION
+          pos.asInstanceOf[AnyRef], // ORDINAL_POSITION
           "YES", // IS_NULLABLE
           null, // SCOPE_CATALOG
           null, // SCOPE_SCHEMA

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala

@@ -19,9 +19,12 @@ package org.apache.spark.sql.hive.thriftserver
 
 import java.sql.SQLException
 
+import scala.collection.JavaConverters._
+
 import org.apache.hive.service.cli.HiveSQLException
 
 import org.apache.spark.sql.hive.HiveUtils
+import org.apache.spark.sql.types.{ByteType, DecimalType, DoubleType, LongType, ShortType}
 
 trait ThriftServerWithSparkContextSuite extends SharedThriftServer {
 
@@ -101,6 +104,79 @@ trait ThriftServerWithSparkContextSuite extends SharedThriftServer {
       }
     }
   }
+
+  test("SparkGetColumnsOperation") {
+    val schemaName = "default"
+    val tableName = "spark_get_col_operation"
+    val decimalType = DecimalType(10, 2)
+    val ddl =
+      s"""
+         |CREATE TABLE $schemaName.$tableName
+         |  (
+         |    a boolean comment '0',
+         |    b int comment '1',
+         |    c float comment '2',
+         |    d ${decimalType.sql} comment '3',
+         |    e array<long> comment '4',
+         |    f array<string> comment '5',
+         |    g map<smallint, tinyint> comment '6',
+         |    h date comment '7',
+         |    i timestamp comment '8',
+         |    j struct<X: bigint,Y: double> comment '9'
+         |  ) using parquet""".stripMargin
+
+    withCLIServiceClient { client =>
+      val sessionHandle = client.openSession(user, "")
+      val confOverlay = new java.util.HashMap[java.lang.String, java.lang.String]
+      val opHandle = client.executeStatement(sessionHandle, ddl, confOverlay)
+      var status = client.getOperationStatus(opHandle)
+      while (!status.getState.isTerminal) {
+        Thread.sleep(10)
+        status = client.getOperationStatus(opHandle)
+      }
+      val getCol = client.getColumns(sessionHandle, "", schemaName, tableName, null)
+      val rowSet = client.fetchResults(getCol)
+      val columns = rowSet.toTRowSet.getColumns
+      assert(columns.get(0).getStringVal.getValues.asScala.forall(_.isEmpty),
+        "catalog name mismatches")
+
+      assert(columns.get(1).getStringVal.getValues.asScala.forall(_ == schemaName),
+        "schema name mismatches")
+
+      assert(columns.get(2).getStringVal.getValues.asScala.forall(_ == tableName),
+        "table name mismatches")
+
+      // column name
+      columns.get(3).getStringVal.getValues.asScala.zipWithIndex.foreach {
+        case (v, i) => assert(v === ('a' + i).toChar.toString, "column name mismatches")
+      }
+
+      val javaTypes = columns.get(4).getI32Val.getValues
+      assert(javaTypes.get(3).intValue() === java.sql.Types.DECIMAL)
+      assert(javaTypes.get(6).intValue() === java.sql.Types.JAVA_OBJECT)
+
+      val typeNames = columns.get(5).getStringVal.getValues
+      assert(typeNames.get(3) === decimalType.sql)
+
+      val colSize = columns.get(6).getI32Val.getValues
+      assert(colSize.get(3).intValue() === decimalType.defaultSize)
+      assert(colSize.get(6).intValue() === 0)
+      assert(colSize.get(9).intValue() === LongType.defaultSize + DoubleType.defaultSize,
+        "struct column size mismatches")
+
+      val decimalDigits = columns.get(8).getI32Val.getValues
+      assert(decimalDigits.get(1).intValue() === 0)
+      assert(decimalDigits.get(2).intValue() === 7)
+      assert(decimalDigits.get(3).intValue() === decimalType.scale)
+      assert(decimalDigits.get(8).intValue() === 6,
+        "timestamp support 6 digits for second fraction")
+
+      val positions = columns.get(16).getI32Val.getValues
+      positions.asScala.zipWithIndex.foreach { case (pos, idx) =>
+        assert(pos === idx, "the client columns disorder")
+      }
+    }
+  }
 }