You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2023/02/17 04:41:18 UTC
[spark] branch branch-3.4 updated: [SPARK-39859][SQL] Support v2 DESCRIBE TABLE EXTENDED for columns
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new ca811db31f9 [SPARK-39859][SQL] Support v2 DESCRIBE TABLE EXTENDED for columns
ca811db31f9 is described below
commit ca811db31f983ce75a92c2d918e334610c0f7632
Author: huaxingao <hu...@apple.com>
AuthorDate: Thu Feb 16 20:40:58 2023 -0800
[SPARK-39859][SQL] Support v2 DESCRIBE TABLE EXTENDED for columns
### What changes were proposed in this pull request?
Support v2 DESCRIBE TABLE EXTENDED for columns
### Why are the changes needed?
DS v1/v2 command parity
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT
Closes #40058 from huaxingao/describe_col.
Authored-by: huaxingao <hu...@apple.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit ebab0ef7c8572e1dac41474c5991f482dbe9d253)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../datasources/v2/DataSourceV2Strategy.scala | 13 +++++--
.../datasources/v2/DescribeColumnExec.scala | 42 ++++++++++++++++++++--
.../execution/command/v2/DescribeTableSuite.scala | 11 ++++--
3 files changed, 59 insertions(+), 7 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index 29f0da1158f..757b66e1534 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.connector.catalog.{Identifier, StagingTableCatalog,
import org.apache.spark.sql.connector.catalog.index.SupportsIndex
import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue}
import org.apache.spark.sql.connector.expressions.filter.{And => V2And, Not => V2Not, Or => V2Or, Predicate}
-import org.apache.spark.sql.connector.read.LocalScan
+import org.apache.spark.sql.connector.read.{LocalScan, SupportsReportStatistics}
import org.apache.spark.sql.connector.read.streaming.{ContinuousStream, MicroBatchStream}
import org.apache.spark.sql.connector.write.V1Write
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
@@ -329,10 +329,17 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
}
DescribeTableExec(output, r.table, isExtended) :: Nil
- case DescribeColumn(_: ResolvedTable, column, isExtended, output) =>
+ case DescribeColumn(r: ResolvedTable, column, isExtended, output) =>
column match {
case c: Attribute =>
- DescribeColumnExec(output, c, isExtended) :: Nil
+ val colStats =
+ r.table.asReadable.newScanBuilder(CaseInsensitiveStringMap.empty()).build() match {
+ case s: SupportsReportStatistics =>
+ val stats = s.estimateStatistics()
+ Some(stats.columnStats().get(FieldReference.column(c.name)))
+ case _ => None
+ }
+ DescribeColumnExec(output, c, isExtended, colStats) :: Nil
case nested =>
throw QueryCompilationErrors.commandNotSupportNestedColumnError(
"DESC TABLE COLUMN", toPrettySQL(nested))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DescribeColumnExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DescribeColumnExec.scala
index 3be9b5c5471..491c214080a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DescribeColumnExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DescribeColumnExec.scala
@@ -22,11 +22,13 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.connector.read.colstats.ColumnStatistics
case class DescribeColumnExec(
override val output: Seq[Attribute],
column: Attribute,
- isExtended: Boolean) extends LeafV2CommandExec {
+ isExtended: Boolean,
+ colStats: Option[ColumnStatistics] = None) extends LeafV2CommandExec {
override protected def run(): Seq[InternalRow] = {
val rows = new ArrayBuffer[InternalRow]()
@@ -42,7 +44,43 @@ case class DescribeColumnExec(
CharVarcharUtils.getRawType(column.metadata).getOrElse(column.dataType).catalogString)
rows += toCatalystRow("comment", comment)
- // TODO: The extended description (isExtended = true) can be added here.
+ if (isExtended && colStats.nonEmpty) {
+ if (colStats.get.min().isPresent) {
+ rows += toCatalystRow("min", colStats.get.min().toString)
+ } else {
+ rows += toCatalystRow("min", "NULL")
+ }
+
+ if (colStats.get.max().isPresent) {
+ rows += toCatalystRow("max", colStats.get.max().toString)
+ } else {
+ rows += toCatalystRow("max", "NULL")
+ }
+
+ if (colStats.get.nullCount().isPresent) {
+ rows += toCatalystRow("num_nulls", colStats.get.nullCount().getAsLong.toString)
+ } else {
+ rows += toCatalystRow("num_nulls", "NULL")
+ }
+
+ if (colStats.get.distinctCount().isPresent) {
+ rows += toCatalystRow("distinct_count", colStats.get.distinctCount().getAsLong.toString)
+ } else {
+ rows += toCatalystRow("distinct_count", "NULL")
+ }
+
+ if (colStats.get.avgLen().isPresent) {
+ rows += toCatalystRow("avg_col_len", colStats.get.avgLen().getAsLong.toString)
+ } else {
+ rows += toCatalystRow("avg_col_len", "NULL")
+ }
+
+ if (colStats.get.maxLen().isPresent) {
+ rows += toCatalystRow("max_col_len", colStats.get.maxLen().getAsLong.toString)
+ } else {
+ rows += toCatalystRow("max_col_len", "NULL")
+ }
+ }
rows.toSeq
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DescribeTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DescribeTableSuite.scala
index 334521a96e5..25363dcea69 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DescribeTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DescribeTableSuite.scala
@@ -149,13 +149,14 @@ class DescribeTableSuite extends command.DescribeTableSuiteBase
}
}
- // TODO(SPARK-39859): Support v2 `DESCRIBE TABLE EXTENDED` for columns
test("describe extended (formatted) a column") {
withNamespaceAndTable("ns", "tbl") { tbl =>
sql(s"""
|CREATE TABLE $tbl
|(key INT COMMENT 'column_comment', col STRING)
|$defaultUsing""".stripMargin)
+
+ sql(s"INSERT INTO $tbl values (1, 'aaa'), (2, 'bbb'), (3, 'ccc'), (null, 'ddd')")
val descriptionDf = sql(s"DESCRIBE TABLE EXTENDED $tbl key")
assert(descriptionDf.schema.map(field => (field.name, field.dataType)) === Seq(
("info_name", StringType),
@@ -165,7 +166,13 @@ class DescribeTableSuite extends command.DescribeTableSuiteBase
Seq(
Row("col_name", "key"),
Row("data_type", "int"),
- Row("comment", "column_comment")))
+ Row("comment", "column_comment"),
+ Row("min", "NULL"),
+ Row("max", "NULL"),
+ Row("num_nulls", "1"),
+ Row("distinct_count", "4"),
+ Row("avg_col_len", "NULL"),
+ Row("max_col_len", "NULL")))
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org