Address gatorsmile's comments

Author: janewangfb

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1239,8 +1239,8 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     val attr = ctx.fieldName.getText
     expression(ctx.base) match {
       case unresolved_attr @ UnresolvedAttribute(nameParts) =>
-        matchEscapedIdentifier(ctx.fieldName.getStart.getText) match {
-          case Some(i) if conf.supportQuotedIdentifiers =>
+        ctx.fieldName.getStart.getText match {
+          case escapedIdentifier(i) if conf.supportQuotedRegexColumnName =>
             UnresolvedRegex(i, Some(unresolved_attr.name))
           case _ =>
             UnresolvedAttribute(nameParts :+ attr)
@@ -1255,8 +1255,8 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    * quoted in ``
    */
   override def visitColumnReference(ctx: ColumnReferenceContext): Expression = withOrigin(ctx) {
-    matchEscapedIdentifier(ctx.getStart.getText) match {
-      case Some(i) if conf.supportQuotedIdentifiers =>
+    ctx.getStart.getText match {
+      case escapedIdentifier(i) if conf.supportQuotedRegexColumnName =>
         UnresolvedRegex(i, None)
       case _ =>
         UnresolvedAttribute.quoted(ctx.getText)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala

@@ -177,17 +177,11 @@ object ParserUtils {
     sb.toString()
   }
 
+  /** the column name pattern in quoted regex without qualifier */
   val escapedIdentifier = "`(.+)`".r
 
-  /**
-   * Return the substring extracted using regex
-   */
-  def matchEscapedIdentifier(b: String): Option[String] = {
-    b match {
-      case escapedIdentifier(i) => Some(i)
-      case _ => None
-    }
-  }
+  /** the column name pattern in quoted regex with qualifier */
+  val qualifiedEscapedIdentifier = ("(.+)" + """.""" + "`(.+)`").r
 
   /** Some syntactic sugar which makes it easier to work with optional clauses for LogicalPlans. */
   implicit class EnhancedLogicalPlan(val plan: LogicalPlan) extends AnyVal {

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -795,9 +795,9 @@ object SQLConf {
       .intConf
       .createWithDefault(UnsafeExternalSorter.DEFAULT_NUM_ELEMENTS_FOR_SPILL_THRESHOLD.toInt)
 
-  val SUPPORT_QUOTED_IDENTIFIERS = buildConf("spark.sql.support.quoted.identifiers")
+  val SUPPORT_QUOTED_REGEX_COLUMN_NAME = buildConf("spark.sql.parser.quotedRegexColumnNames")
     .internal()
-    .doc("When true, identifiers specified by regex patterns will be expanded.")
+    .doc("When true, column names specified by quoted regex pattern will be expanded.")
     .booleanConf
     .createWithDefault(false)
 
@@ -1057,7 +1057,7 @@ class SQLConf extends Serializable with Logging {
 
   def starSchemaFTRatio: Double = getConf(STARSCHEMA_FACT_TABLE_RATIO)
 
-  def supportQuotedIdentifiers: Boolean = getConf(SUPPORT_QUOTED_IDENTIFIERS)
+  def supportQuotedRegexColumnName: Boolean = getConf(SUPPORT_QUOTED_REGEX_COLUMN_NAME)
 
   /** ********************** SQLConf functionality methods ************ */
 

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -25,9 +25,7 @@ import scala.collection.JavaConverters._
 import scala.language.implicitConversions
 import scala.reflect.runtime.universe.TypeTag
 import scala.util.control.NonFatal
-
 import org.apache.commons.lang3.StringUtils
-
 import org.apache.spark.annotation.{DeveloperApi, Experimental, InterfaceStability}
 import org.apache.spark.api.java.JavaRDD
 import org.apache.spark.api.java.function._
@@ -40,13 +38,13 @@ import org.apache.spark.sql.catalyst.catalog.CatalogRelation
 import org.apache.spark.sql.catalyst.encoders._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
-import org.apache.spark.sql.catalyst.json.{JacksonGenerator, JSONOptions}
+import org.apache.spark.sql.catalyst.json.{JSONOptions, JacksonGenerator}
 import org.apache.spark.sql.catalyst.optimizer.CombineUnions
-import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.parser.{ParseException, ParserUtils}
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, PartitioningCollection}
-import org.apache.spark.sql.catalyst.util.{usePrettyExpression, DateTimeUtils}
+import org.apache.spark.sql.catalyst.util.{DateTimeUtils, usePrettyExpression}
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.command._
 import org.apache.spark.sql.execution.datasources.LogicalRelation
@@ -1188,6 +1186,11 @@ class Dataset[T] private[sql](
   def col(colName: String): Column = colName match {
     case "*" =>
       Column(ResolvedStar(queryExecution.analyzed.output))
+    case ParserUtils.escapedIdentifier(i) if sqlContext.conf.supportQuotedRegexColumnName =>
+      Column(UnresolvedRegex(i, None))
+    case ParserUtils.qualifiedEscapedIdentifier(i, j)
+      if sqlContext.conf.supportQuotedRegexColumnName =>
+      Column(UnresolvedRegex(j, Some(i)))
     case _ =>
       val expr = resolve(colName)
       Column(expr)

sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala

@@ -245,7 +245,8 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
   }
 
   test("select 3, regex") {
-    val ds = Seq(("a", 1), ("b", 2), ("c", 3)).toDF()
+    val ds = Seq(("a", 1), ("b", 2), ("c", 3)).toDS()
+
     intercept[AnalysisException] {
       ds.select(expr("`(_1)?+.+`").as[Int])
     }
@@ -254,7 +255,37 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
       ds.select(expr("`(_1|_2)`").as[Int])
     }
 
-    withSQLConf(SQLConf.SUPPORT_QUOTED_IDENTIFIERS.key -> "true") {
+    intercept[AnalysisException] {
+      ds.select(ds("`(_1)?+.+`"))
+    }
+
+    intercept[AnalysisException] {
+      ds.select(ds("`(_1|_2)`"))
+    }
+
+    withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "true") {
+      checkDataset(
+        ds.select(ds("`(_1|_2)`"))
+          .select(expr("named_struct('a', _1, 'b', _2)").as[ClassData]),
+        ClassData("a", 1), ClassData("b", 2), ClassData("c", 3))
+
+      checkDataset(
+        ds.alias("g")
+          .select(ds("g.`(_1|_2)`"))
+          .select(expr("named_struct('a', _1, 'b', _2)").as[ClassData]),
+        ClassData("a", 1), ClassData("b", 2), ClassData("c", 3))
+
+      checkDataset(
+        ds.select(ds("`(_1)?+.+`"))
+          .select(expr("_2").as[Int]),
+        1, 2, 3)
+
+      checkDataset(
+        ds.alias("g")
+          .select(ds("g.`(_1)?+.+`"))
+          .select(expr("_2").as[Int]),
+        1, 2, 3)
+
       checkDataset(
         ds.select(expr("`(_1)?+.+`").as[Int]),
         1, 2, 3)

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -2664,7 +2664,7 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
     }
 
     // now, turn on hive.support.quoted.identifiers
-    withSQLConf(SQLConf.SUPPORT_QUOTED_IDENTIFIERS.key -> "true") {
+    withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "true") {
       checkAnswer(
         sql(
           """