You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2023/04/13 21:58:22 UTC
[spark] branch master updated: [SPARK-43085][SQL] Support column DEFAULT assignment for multi-part table names
This is an automated email from the ASF dual-hosted git repository.
gengliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fa4978a6601 [SPARK-43085][SQL] Support column DEFAULT assignment for multi-part table names
fa4978a6601 is described below
commit fa4978a66014889dc9894e122ad712dc1e2c4328
Author: Daniel Tenedorio <da...@databricks.com>
AuthorDate: Thu Apr 13 14:58:07 2023 -0700
[SPARK-43085][SQL] Support column DEFAULT assignment for multi-part table names
### What changes were proposed in this pull request?
This PR adds support for column DEFAULT assignment for multi-part table names.
### Why are the changes needed?
Spark SQL workloads that refer to tables with multi-part names may want to use column DEFAULT functionality. This PR enables this.
### Does this PR introduce _any_ user-facing change?
Yes, column DEFAULT assignment now works with multi-part table names.
### How was this patch tested?
This PR adds unit tests.
Closes #40732 from dtenedor/fix-three-part-name-bug.
Authored-by: Daniel Tenedorio <da...@databricks.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 7 ++-
.../catalyst/analysis/ResolveDefaultColumns.scala | 63 +++++++---------------
.../analysis/ResolveDefaultColumnsSuite.scala | 21 +++++++-
3 files changed, 46 insertions(+), 45 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 8821e652a31..126b0618727 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -313,7 +313,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
TimeWindowing ::
SessionWindowing ::
ResolveWindowTime ::
- ResolveDefaultColumns(v1SessionCatalog) ::
+ ResolveDefaultColumns(ResolveRelations.resolveRelationOrTempView) ::
ResolveInlineTables ::
ResolveLambdaVariables ::
ResolveTimeZone ::
@@ -1274,6 +1274,11 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
}
+
+ /** Consumes an unresolved relation and resolves it to a v1 or v2 relation or temporary view. */
+ def resolveRelationOrTempView(u: UnresolvedRelation): LogicalPlan = {
+ EliminateSubqueryAliases(resolveRelation(u).getOrElse(u))
+ }
}
/** Handle INSERT INTO for DSv2 */
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
index ca1b2731436..d028d4ff596 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
@@ -19,14 +19,14 @@ package org.apache.spark.sql.catalyst.analysis
import scala.collection.mutable
-import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.{SessionCatalog, UnresolvedCatalogRelation}
+import org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -47,9 +47,11 @@ import org.apache.spark.sql.types._
* (1, 5)
* (4, 6)
*
- * @param catalog the catalog to use for looking up the schema of INSERT INTO table objects.
+ * @param resolveRelation function to resolve relations from the catalog. This should generally map
+ * to the 'resolveRelationOrTempView' method of the ResolveRelations rule.
*/
-case class ResolveDefaultColumns(catalog: SessionCatalog) extends Rule[LogicalPlan] {
+case class ResolveDefaultColumns(
+ resolveRelation: UnresolvedRelation => LogicalPlan) extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = {
plan.resolveOperatorsWithPruning(
(_ => SQLConf.get.enableDefaultColumns), ruleId) {
@@ -578,44 +580,19 @@ case class ResolveDefaultColumns(catalog: SessionCatalog) extends Rule[LogicalPl
* Returns the schema for the target table of a DML command, looking into the catalog if needed.
*/
private def getSchemaForTargetTable(table: LogicalPlan): Option[StructType] = {
- // First find the source relation. Note that we use 'collectFirst' to descend past any
- // SubqueryAlias nodes that may be present.
- val source: Option[LogicalPlan] = table.collectFirst {
- case r: NamedRelation if !r.skipSchemaResolution =>
- // Here we only resolve the default columns in the tables that require schema resolution
- // during write operations.
- r
- case r: UnresolvedCatalogRelation => r
- }
- // Check if the target table is already resolved. If so, return the computed schema.
- source.foreach { r =>
- if (r.schema.fields.nonEmpty) {
- return Some(r.schema)
- }
- }
- // Lookup the relation from the catalog by name. This either succeeds or returns some "not
- // found" error. In the latter cases, return out of this rule without changing anything and let
- // the analyzer return a proper error message elsewhere.
- val tableName: TableIdentifier = source match {
- case Some(r: UnresolvedRelation) => TableIdentifier(r.name)
- case Some(r: UnresolvedCatalogRelation) => r.tableMeta.identifier
- case _ => return None
- }
- // First try to get the table metadata directly. If that fails, check for views below.
- if (catalog.tableExists(tableName)) {
- return Some(catalog.getTableMetadata(tableName).schema)
- }
- val lookup: LogicalPlan = try {
- catalog.lookupRelation(tableName)
- } catch {
- case _: AnalysisException => return None
- }
- lookup match {
- case SubqueryAlias(_, r: UnresolvedCatalogRelation) =>
- Some(r.tableMeta.schema)
- case SubqueryAlias(_, r: View) if r.isTempView =>
- Some(r.desc.schema)
- case _ => None
+ val resolved = table match {
+ case r: UnresolvedRelation if !r.skipSchemaResolution && !r.isStreaming =>
+ resolveRelation(r)
+ case other =>
+ other
+ }
+ resolved.collectFirst {
+ case r: UnresolvedCatalogRelation =>
+ r.tableMeta.schema
+ case d: DataSourceV2Relation if !d.skipSchemaResolution && !d.isStreaming =>
+ CatalogV2Util.v2ColumnsToStructType(d.table.columns())
+ case v: View if v.isTempViewStoringAnalyzedPlan =>
+ v.schema
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumnsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumnsSuite.scala
index fc540e65593..ea0dc9d603d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumnsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumnsSuite.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{StructField, StructType, TimestampType}
class ResolveDefaultColumnsSuite extends QueryTest with SharedSparkSession {
- val rule = ResolveDefaultColumns(catalog = null)
+ val rule = ResolveDefaultColumns(null)
// This is the internal storage for the timestamp 2020-12-31 00:00:00.0.
val literal = Literal(1609401600000000L, TimestampType)
val table = UnresolvedInlineTable(
@@ -74,4 +74,23 @@ class ResolveDefaultColumnsSuite extends QueryTest with SharedSparkSession {
sql("select null, timestamp'2020-12-31'").collect().head)
}
}
+
+ test("SPARK-43085: Column DEFAULT assignment for target tables with multi-part names") {
+ withDatabase("demos") {
+ sql("create database demos")
+ withTable("demos.test_ts") {
+ sql("create table demos.test_ts (id int, ts timestamp) using parquet")
+ sql("insert into demos.test_ts(ts) values (timestamp'2023-01-01')")
+ checkAnswer(spark.table("demos.test_ts"),
+ sql("select null, timestamp'2023-01-01'"))
+ }
+ withTable("demos.test_ts") {
+ sql("create table demos.test_ts (id int, ts timestamp) using parquet")
+ sql("use database demos")
+ sql("insert into test_ts(ts) values (timestamp'2023-01-01')")
+ checkAnswer(spark.table("demos.test_ts"),
+ sql("select null, timestamp'2023-01-01'"))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org