Review comments

Author: jzhuge

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -140,7 +140,7 @@ statement
     | ALTER TABLE tableIdentifier partitionSpec? SET locationSpec      #setTableLocation
     | ALTER TABLE tableIdentifier RECOVER PARTITIONS                   #recoverPartitions
     | DROP TABLE (IF EXISTS)? multipartIdentifier PURGE?               #dropTable
-    | DROP VIEW (IF EXISTS)? multipartIdentifier                       #dropTable
+    | DROP VIEW (IF EXISTS)? multipartIdentifier                       #dropView
     | CREATE (OR REPLACE)? (GLOBAL? TEMPORARY)?
         VIEW (IF NOT EXISTS)? tableIdentifier
         identifierCommentList?

sql/catalyst/src/main/java/org/apache/spark/sql/catalog/v2/CatalogManager.java

@@ -23,8 +23,10 @@
 import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 import org.apache.spark.util.Utils;
 
+import java.util.HashMap;
 import java.util.Map;
-import java.util.stream.Collectors;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import static scala.collection.JavaConverters.mapAsJavaMapConverter;
 
@@ -33,23 +35,6 @@ public class Catalogs {
   private Catalogs() {
   }
 
-  public static String classKey(String name) {
-    return "spark.sql.catalog." + name;
-  }
-
-  public static String optionKeyPrefix(String name) {
-    return "spark.sql.catalog." + name + ".";
-  }
-
-  public static boolean isOptionKey(String name, String keyName) {
-    return keyName.startsWith(optionKeyPrefix(name));
-  }
-
-  public static String optionName(String name, String keyName) {
-    assert(isOptionKey(name, keyName));
-    return keyName.substring(optionKeyPrefix(name).length());
-  }
-
   /**
    * Load and configure a catalog by name.
    * <p>
@@ -64,10 +49,10 @@ public static String optionName(String name, String keyName) {
    */
   public static CatalogPlugin load(String name, SQLConf conf)
       throws CatalogNotFoundException, SparkException {
-    String pluginClassName = conf.getConfString(classKey(name), null);
+    String pluginClassName = conf.getConfString("spark.sql.catalog." + name, null);
     if (pluginClassName == null) {
       throw new CatalogNotFoundException(String.format(
-          "Catalog '%s' plugin class not found: %s is not defined", name, classKey(name)));
+          "Catalog '%s' plugin class not found: spark.sql.catalog.%s is not defined", name, name));
     }
 
     ClassLoader loader = Utils.getContextOrSparkClassLoader();
@@ -111,12 +96,17 @@ public static CatalogPlugin load(String name, SQLConf conf)
    * @return a case insensitive string map of options starting with spark.sql.catalog.(name).
    */
   private static CaseInsensitiveStringMap catalogOptions(String name, SQLConf conf) {
-    Map<String, String> options =
-        mapAsJavaMapConverter(conf.getAllConfs()).asJava().entrySet().stream()
-            .filter(e -> isOptionKey(name, e.getKey()))
-            .collect(Collectors.toMap(
-                e -> optionName(name, e.getKey()),
-                e -> e.getValue()));
+    Map<String, String> allConfs = mapAsJavaMapConverter(conf.getAllConfs()).asJava();
+    Pattern prefix = Pattern.compile("^spark\\.sql\\.catalog\\." + name + "\\.(.+)");
+
+    HashMap<String, String> options = new HashMap<>();
+    for (Map.Entry<String, String> entry : allConfs.entrySet()) {
+      Matcher matcher = prefix.matcher(entry.getKey());
+      if (matcher.matches() && matcher.groupCount() > 0) {
+        options.put(matcher.group(1), entry.getValue());
+      }
+    }
+
     return new CaseInsensitiveStringMap(options);
   }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/catalog/v2/Catalogs.java

@@ -22,8 +22,6 @@
 
 import java.util.Arrays;
 import java.util.Objects;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 /**
  *  An {@link Identifier} implementation.
@@ -51,21 +49,6 @@ public String name() {
     return name;
   }
 
-  private String quote(String part) {
-    if (part.contains("`")) {
-      return part.replace("`", "``");
-    } else {
-      return part;
-    }
-  }
-
-  @Override
-  public String toString() {
-    return Stream.concat(Stream.of(namespace), Stream.of(name))
-        .map(part -> '`' + quote(part) + '`')
-        .collect(Collectors.joining("."));
-  }
-
   @Override
   public boolean equals(Object o) {
     if (this == o) {

sql/catalyst/src/main/java/org/apache/spark/sql/catalog/v2/IdentifierImpl.java

@@ -26,7 +26,6 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 
 /**
@@ -179,17 +178,4 @@ public double getDouble(String key, double defaultValue) {
   public Map<String, String> asCaseSensitiveMap() {
     return Collections.unmodifiableMap(original);
   }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o) return true;
-    if (o == null || getClass() != o.getClass()) return false;
-    CaseInsensitiveStringMap that = (CaseInsensitiveStringMap) o;
-    return delegate.equals(that.delegate);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(delegate);
-  }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java

@@ -2195,4 +2195,22 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     }
   }
 
+  /**
+   * Create a [[sql.DropTableStatement]] command.
+   */
+  override def visitDropTable(ctx: DropTableContext): LogicalPlan = withOrigin(ctx) {
+    sql.DropTableStatement(
+      visitMultipartIdentifier(ctx.multipartIdentifier()),
+      ctx.EXISTS != null,
+      ctx.PURGE != null)
+  }
+
+  /**
+   * Create a [[sql.DropViewStatement]] command.
+   */
+  override def visitDropView(ctx: DropViewContext): AnyRef = withOrigin(ctx) {
+    sql.DropViewStatement(
+      visitMultipartIdentifier(ctx.multipartIdentifier()),
+      ctx.EXISTS != null)
+  }
 }

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

@@ -26,7 +26,6 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 case class DropTableStatement(
     tableName: Seq[String],
     ifExists: Boolean,
-    isView: Boolean,
     purge: Boolean) extends ParsedStatement {
 
   override def output: Seq[Attribute] = Seq.empty

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/sql/DropTableStatement.scala

@@ -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.plans.logical.sql
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+/**
+ * A DROP VIEW statement, as parsed from SQL.
+ */
+case class DropViewStatement(
+    tableName: Seq[String],
+    ifExists: Boolean) extends ParsedStatement {
+
+  override def output: Seq[Attribute] = Seq.empty
+
+  override def children: Seq[LogicalPlan] = Seq.empty
+}