[SPARK-19602][SQL] Support column resolution of fully qualified column name ( 3 part name) #17185
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -104,12 +104,12 @@ case class UnresolvedAttribute(nameParts: Seq[String]) extends Attribute with Un | |
| override def exprId: ExprId = throw new UnresolvedException(this, "exprId") | ||
| override def dataType: DataType = throw new UnresolvedException(this, "dataType") | ||
| override def nullable: Boolean = throw new UnresolvedException(this, "nullable") | ||
| override def qualifier: Seq[String] = throw new UnresolvedException(this, "qualifier") | ||
| override lazy val resolved = false | ||
|
|
||
| override def newInstance(): UnresolvedAttribute = this | ||
| override def withNullability(newNullability: Boolean): UnresolvedAttribute = this | ||
| override def withQualifier(newQualifier: Seq[String]): UnresolvedAttribute = this | ||
| override def withName(newName: String): UnresolvedAttribute = UnresolvedAttribute.quoted(newName) | ||
| override def withMetadata(newMetadata: Metadata): Attribute = this | ||
|
|
||
|
|
@@ -240,7 +240,7 @@ abstract class Star extends LeafExpression with NamedExpression { | |
| override def exprId: ExprId = throw new UnresolvedException(this, "exprId") | ||
| override def dataType: DataType = throw new UnresolvedException(this, "dataType") | ||
| override def nullable: Boolean = throw new UnresolvedException(this, "nullable") | ||
| override def qualifier: Seq[String] = throw new UnresolvedException(this, "qualifier") | ||
| override def toAttribute: Attribute = throw new UnresolvedException(this, "toAttribute") | ||
| override def newInstance(): NamedExpression = throw new UnresolvedException(this, "newInstance") | ||
| override lazy val resolved = false | ||
|
|
@@ -262,17 +262,46 @@ abstract class Star extends LeafExpression with NamedExpression { | |
| */ | ||
| case class UnresolvedStar(target: Option[Seq[String]]) extends Star with Unevaluable { | ||
|
|
||
| /** | ||
| * Returns true if the nameParts match the qualifier of the attribute | ||
| * | ||
| * There are two checks: i) Check if the nameParts match the qualifier fully. | ||
| * E.g. SELECT db.t1.* FROM db1.t1 In this case, the nameParts is Seq("db1", "t1") and | ||
| * qualifier of the attribute is Seq("db1","t1") | ||
| * ii) If (i) is not true, then check if nameParts is only a single element and it | ||
| * matches the table portion of the qualifier | ||
| * | ||
| * E.g. SELECT t1.* FROM db1.t1 In this case nameParts is Seq("t1") and | ||
| * qualifier is Seq("db1","t1") | ||
| * SELECT a.* FROM db1.t1 AS a | ||
| * In this case nameParts is Seq("a") and qualifier for | ||
| * attribute is Seq("a") | ||
| */ | ||
| private def matchedQualifier( | ||
| attribute: Attribute, | ||
|
||
| nameParts: Seq[String], | ||
| resolver: Resolver): Boolean = { | ||
| val qualifierList = attribute.qualifier | ||
|
|
||
| val matched = nameParts.corresponds(qualifierList)(resolver) || { | ||
| // check if it matches the table portion of the qualifier | ||
| if (nameParts.length == 1 && qualifierList.nonEmpty) { | ||
| resolver(nameParts.head, qualifierList.last) | ||
| } else { | ||
| false | ||
| } | ||
| } | ||
| matched | ||
| } | ||
|
|
||
| override def expand( | ||
| input: LogicalPlan, | ||
| resolver: Resolver): Seq[NamedExpression] = { | ||
| // If there is no table specified, use all input attributes. | ||
| if (target.isEmpty) return input.output | ||
|
|
||
| val expandedAttributes = input.output.filter(matchedQualifier(_, target.get, resolver)) | ||
|
|
||
| if (expandedAttributes.nonEmpty) return expandedAttributes | ||
|
|
||
| // Try to resolve it as a struct expansion. If there is a conflict and both are possible, | ||
|
|
@@ -316,8 +345,8 @@ case class UnresolvedRegex(regexPattern: String, table: Option[String], caseSens | |
| // If there is no table specified, use all input attributes that match expr | ||
| case None => input.output.filter(_.name.matches(pattern)) | ||
| // If there is a table, pick out attributes that are part of this table that match expr | ||
| case Some(t) => input.output.filter(a => a.qualifier.nonEmpty && | ||
| resolver(a.qualifier.last, t)).filter(_.name.matches(pattern)) | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -345,7 +374,7 @@ case class MultiAlias(child: Expression, names: Seq[String]) | |
|
|
||
| override def nullable: Boolean = throw new UnresolvedException(this, "nullable") | ||
|
|
||
| override def qualifier: Seq[String] = throw new UnresolvedException(this, "qualifier") | ||
|
|
||
| override def toAttribute: Attribute = throw new UnresolvedException(this, "toAttribute") | ||
|
|
||
|
|
@@ -403,7 +432,7 @@ case class UnresolvedAlias( | |
| extends UnaryExpression with NamedExpression with Unevaluable { | ||
|
|
||
| override def toAttribute: Attribute = throw new UnresolvedException(this, "toAttribute") | ||
| override def qualifier: Seq[String] = throw new UnresolvedException(this, "qualifier") | ||
| override def exprId: ExprId = throw new UnresolvedException(this, "exprId") | ||
| override def nullable: Boolean = throw new UnresolvedException(this, "nullable") | ||
| override def dataType: DataType = throw new UnresolvedException(this, "dataType") | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -152,10 +152,22 @@ package object expressions { | |
| unique(attrs.groupBy(_.name.toLowerCase(Locale.ROOT))) | ||
| } | ||
|
|
||
| /** Map to use for qualified case insensitive attribute lookups with 2 part key */ | ||
| @transient private lazy val qualified: Map[(String, String), Seq[Attribute]] = { | ||
| // key is 2 part: table/alias and name | ||
| val grouped = attrs.filter(_.qualifier.nonEmpty).groupBy { | ||
| a => (a.qualifier.last.toLowerCase(Locale.ROOT), a.name.toLowerCase(Locale.ROOT)) | ||
| } | ||
| unique(grouped) | ||
| } | ||
|
|
||
| /** Map to use for qualified case insensitive attribute lookups with 3 part key */ | ||
| @transient private val qualified3Part: Map[(String, String, String), Seq[Attribute]] = { | ||
| // key is 3 part: database name, table name and name | ||
| val grouped = attrs.filter(_.qualifier.length == 2).groupBy { a => | ||
| (a.qualifier.head.toLowerCase(Locale.ROOT), | ||
| a.qualifier.last.toLowerCase(Locale.ROOT), | ||
| a.name.toLowerCase(Locale.ROOT)) | ||
| } | ||
| unique(grouped) | ||
| } | ||
|
|
@@ -169,25 +181,48 @@ package object expressions { | |
| }) | ||
| } | ||
|
|
||
| // Find matches for the given name assuming that the 1st two parts are qualifier | ||
| // (i.e. database name and table name) and the 3rd part is the actual column name. | ||
| // | ||
| // For example, consider an example where "db1" is the database name, "a" is the table name | ||
| // and "b" is the column name and "c" is the struct field name. | ||
| // If the name parts is db1.a.b.c, then Attribute will match | ||
|
||
| // Attribute(b, qualifier("db1,"a")) and List("c") will be the second element | ||
| var matches: (Seq[Attribute], Seq[String]) = nameParts match { | ||
| case dbPart +: tblPart +: name +: nestedFields => | ||
| val key = (dbPart.toLowerCase(Locale.ROOT), | ||
| tblPart.toLowerCase(Locale.ROOT), name.toLowerCase(Locale.ROOT)) | ||
| val attributes = collectMatches(name, qualified3Part.get(key)).filter { | ||
| a => (resolver(dbPart, a.qualifier.head) && resolver(tblPart, a.qualifier.last)) | ||
| } | ||
| (attributes, nestedFields) | ||
| case all => | ||
| (Seq.empty, Seq.empty) | ||
| } | ||
|
|
||
| // If there are no matches, then find matches for the given name assuming that | ||
| // the 1st part is a qualifier (i.e. table name, alias, or subquery alias) and the | ||
| // 2nd part is the actual name. This returns a tuple of | ||
| // matched attributes and a list of parts that are to be resolved. | ||
| // | ||
| // For example, consider an example where "a" is the table name, "b" is the column name, | ||
| // and "c" is the struct field name, i.e. "a.b.c". In this case, Attribute will be "a.b", | ||
| // and the second element will be List("c"). | ||
| if (matches._1.isEmpty) { | ||
| matches = nameParts match { | ||
| case qualifier +: name +: nestedFields => | ||
| val key = (qualifier.toLowerCase(Locale.ROOT), name.toLowerCase(Locale.ROOT)) | ||
| val attributes = collectMatches(name, qualified.get(key)).filter { a => | ||
| resolver(qualifier, a.qualifier.last) | ||
| } | ||
| (attributes, nestedFields) | ||
| case all => | ||
| (Seq.empty[Attribute], Seq.empty[String]) | ||
| } | ||
| } | ||
|
|
||
| // If none of attributes match database.table.column pattern or | ||
| // `table.column` pattern, we try to resolve it as a column. | ||
| val (candidates, nestedFields) = matches match { | ||
| case (Seq(), _) => | ||
| val name = nameParts.head | ||
|
|
||
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.
what about
SELECT db1.t1.* FROM t1while the current database isdb1?Uh oh!
There was an error while loading. Please reload this page.
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.
This (SELECT db1.t1.* FROM t1) would resolve correctly.
Also a similar scenario (details in Section 2 in Table A # 5 in the design doc)
select db1.t1.i1 from t1 will resolve correctly when the current database is db1. This is tested here
I will explicitly add a test case for the db1.t1.* from t1 as well.