-
Notifications
You must be signed in to change notification settings - Fork 29.3k
[SPARK-30214][SQL] A new framework to resolve v2 commands #26847
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 32 commits
a005713
7c45c9b
024ab39
e489e62
9b272b6
85617cd
e33a200
3e37941
7df9407
57c83fc
ccc4702
72c01a9
1a7c800
415de11
6ab2228
56037ff
6675e7f
a42e12e
2c458f0
573a66e
f6e742b
0754656
a868420
5a6aa2a
0b89d3a
a41acc5
de054c7
e0836f6
fc555be
7b4f3e3
9f2159f
70028dd
9d10239
c391212
d20b1b2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,7 +25,7 @@ import scala.collection.mutable.ArrayBuffer | |
| import scala.util.Random | ||
|
|
||
| import org.apache.spark.sql.AnalysisException | ||
| import org.apache.spark.sql.catalyst._ | ||
| import org.apache.spark.sql.catalyst.{catalog, _} | ||
| import org.apache.spark.sql.catalyst.catalog._ | ||
| import org.apache.spark.sql.catalyst.encoders.OuterScopes | ||
| import org.apache.spark.sql.catalyst.expressions._ | ||
|
|
@@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.rules._ | |
| import org.apache.spark.sql.catalyst.trees.TreeNodeRef | ||
| import org.apache.spark.sql.catalyst.util.toPrettySQL | ||
| import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, CatalogV2Util, Identifier, LookupCatalog, Table, TableCatalog, TableChange, V1Table} | ||
| import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ | ||
| import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform, Transform} | ||
| import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation | ||
| import org.apache.spark.sql.internal.SQLConf | ||
|
|
@@ -196,6 +197,7 @@ class Analyzer( | |
| new SubstituteUnresolvedOrdinals(conf)), | ||
| Batch("Resolution", fixedPoint, | ||
| ResolveTableValuedFunctions :: | ||
| ResolveNamespace(catalogManager) :: | ||
| new ResolveCatalogs(catalogManager) :: | ||
| ResolveInsertInto :: | ||
| ResolveRelations :: | ||
|
|
@@ -720,6 +722,14 @@ class Analyzer( | |
| } | ||
| } | ||
|
|
||
| case class ResolveNamespace(catalogManager: CatalogManager) | ||
| extends Rule[LogicalPlan] with LookupCatalog { | ||
| def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators { | ||
| case UnresolvedNamespace(CatalogAndNamespace(catalog, ns)) => | ||
| ResolvedNamespace(catalog.asNamespaceCatalog, ns) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Resolve relations to temp views. This is not an actual rule, and is called by | ||
| * [[ResolveTables]] and [[ResolveRelations]]. | ||
|
|
@@ -732,6 +742,11 @@ class Analyzer( | |
| lookupTempView(ident) | ||
| .map(view => i.copy(table = view)) | ||
| .getOrElse(i) | ||
| case u @ UnresolvedTable(ident) => | ||
| lookupTempView(ident).foreach { _ => | ||
| u.failAnalysis(s"${ident.quoted} is a view not table.") | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Shouldn't this be |
||
| } | ||
| u | ||
| } | ||
|
|
||
| def lookupTempView(identifier: Seq[String]): Option[LogicalPlan] = | ||
|
|
@@ -753,6 +768,11 @@ class Analyzer( | |
| lookupV2Relation(u.multipartIdentifier) | ||
| .getOrElse(u) | ||
|
|
||
| case u @ UnresolvedTable(NonSessionCatalogAndIdentifier(catalog, ident)) => | ||
| CatalogV2Util.loadTable(catalog, ident) | ||
| .map(ResolvedTable(catalog.asTableCatalog, ident, _)) | ||
| .getOrElse(u) | ||
|
|
||
| case i @ InsertIntoStatement(u: UnresolvedRelation, _, _, _, _) if i.query.resolved => | ||
| lookupV2Relation(u.multipartIdentifier) | ||
| .map(v2Relation => i.copy(table = v2Relation)) | ||
|
|
@@ -862,6 +882,16 @@ class Analyzer( | |
| } | ||
|
|
||
| case u: UnresolvedRelation => resolveRelation(u) | ||
|
|
||
| case u @ UnresolvedTable(SessionCatalogAndIdentifier(catalog, ident)) => | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we should follow the insert case here
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This needs to go thru |
||
| val newIdent: Identifier = withNewNamespace(ident) | ||
| assert(newIdent.namespace.length == 1) | ||
| CatalogV2Util.loadTable(catalog, newIdent) match { | ||
| case Some(v1Table: V1Table) if v1Table.v1Table.tableType == CatalogTableType.VIEW => | ||
| u.failAnalysis(s"$newIdent is a view not table.") | ||
| case Some(table) => ResolvedTable(catalog.asTableCatalog, newIdent, table) | ||
| case None => u | ||
| } | ||
| } | ||
|
|
||
| // Look up a relation from the given session catalog with the following logic: | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.sql.catalyst.analysis | ||
|
|
||
| import org.apache.spark.sql.catalyst.expressions.Attribute | ||
| import org.apache.spark.sql.catalyst.plans.logical.LeafNode | ||
| import org.apache.spark.sql.connector.catalog.SupportsNamespaces | ||
|
|
||
| case class ResolvedNamespace(catalog: SupportsNamespaces, namespace: Seq[String]) | ||
| extends LeafNode { | ||
| override def output: Seq[Attribute] = Nil | ||
| } | ||
|
|
||
| case class UnresolvedNamespace(multipartIdentifier: Seq[String]) extends LeafNode { | ||
| override lazy val resolved: Boolean = false | ||
|
|
||
| override def output: Seq[Attribute] = Nil | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.sql.catalyst.analysis | ||
|
|
||
| import org.apache.spark.sql.catalyst.expressions.Attribute | ||
| import org.apache.spark.sql.catalyst.plans.logical.LeafNode | ||
| import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} | ||
|
|
||
| case class ResolvedTable(catalog: TableCatalog, identifier: Identifier, table: Table) | ||
|
cloud-fan marked this conversation as resolved.
|
||
| extends LeafNode { | ||
| override def output: Seq[Attribute] = Nil | ||
| } | ||
|
|
||
| case class UnresolvedTable(multipartIdentifier: Seq[String]) extends LeafNode { | ||
| override lazy val resolved: Boolean = false | ||
|
|
||
| override def output: Seq[Attribute] = Nil | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does it mean the
ParsedStatementfrom parser will turn to useUnresolvedNamespace? Currently, the catalogs in statements are resolved atResolveCatalogs. Will we need to refactorResolveCatalogsdue to this change?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, see
#26847 (comment)