Revert "[SPARK-26946][SQL][FOLLOWUP] Require lookup function"

This reverts commit 7e11ca6

Author: jzhuge

sql/catalyst/src/main/scala/org/apache/spark/sql/catalog/v2/LookupCatalog.scala

@@ -26,31 +26,35 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 @Experimental
 trait LookupCatalog {
 
-  protected def lookupCatalog(name: String): CatalogPlugin
+  def lookupCatalog: Option[(String) => CatalogPlugin] = None
 
   type CatalogObjectIdentifier = (Option[CatalogPlugin], Identifier)
 
   /**
    * Extract catalog plugin and identifier from a multi-part identifier.
    */
   object CatalogObjectIdentifier {
-    def unapply(parts: Seq[String]): Some[CatalogObjectIdentifier] = parts match {
-      case Seq(name) =>
-        Some((None, Identifier.of(Array.empty, name)))
-      case Seq(catalogName, tail @ _*) =>
-        try {
-          Some((Some(lookupCatalog(catalogName)), Identifier.of(tail.init.toArray, tail.last)))
-        } catch {
-          case _: CatalogNotFoundException =>
-            Some((None, Identifier.of(parts.init.toArray, parts.last)))
-        }
+    def unapply(parts: Seq[String]): Option[CatalogObjectIdentifier] = lookupCatalog.map { lookup =>
+      parts match {
+        case Seq(name) =>
+          (None, Identifier.of(Array.empty, name))
+        case Seq(catalogName, tail @ _*) =>
+          try {
+            val catalog = lookup(catalogName)
+            (Some(catalog), Identifier.of(tail.init.toArray, tail.last))
+          } catch {
+            case _: CatalogNotFoundException =>
+              (None, Identifier.of(parts.init.toArray, parts.last))
+          }
+      }
     }
   }
 
   /**
    * Extract legacy table identifier from a multi-part identifier.
    *
-   * For legacy support only. Please use [[CatalogObjectIdentifier]] instead on DSv2 code paths.
+   * For legacy support only. Please use
+   * [[org.apache.spark.sql.catalog.v2.LookupCatalog.CatalogObjectIdentifier]] in DSv2 code paths.
    */
   object AsTableIdentifier {
     def unapply(parts: Seq[String]): Option[TableIdentifier] = parts match {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -24,7 +24,7 @@ import scala.collection.mutable.ArrayBuffer
 import scala.util.Random
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalog.v2.{CatalogNotFoundException, CatalogPlugin, LookupCatalog}
+import org.apache.spark.sql.catalog.v2.{CatalogPlugin, LookupCatalog}
 import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.encoders.OuterScopes
@@ -96,15 +96,18 @@ object AnalysisContext {
 class Analyzer(
     catalog: SessionCatalog,
     conf: SQLConf,
-    maxIterations: Int)
+    maxIterations: Int,
+    override val lookupCatalog: Option[(String) => CatalogPlugin] = None)
   extends RuleExecutor[LogicalPlan] with CheckAnalysis with LookupCatalog {
 
   def this(catalog: SessionCatalog, conf: SQLConf) = {
     this(catalog, conf, conf.optimizerMaxIterations)
   }
 
-  override protected def lookupCatalog(name: String): CatalogPlugin =
-    throw new CatalogNotFoundException("No catalog lookup function")
+  def this(lookupCatalog: Option[(String) => CatalogPlugin], catalog: SessionCatalog,
+      conf: SQLConf) = {
+    this(catalog, conf, conf.optimizerMaxIterations, lookupCatalog)
+  }
 
   def executeAndCheck(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
     AnalysisHelper.markInAnalyzer {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/v2/LookupCatalogSuite.scala

@@ -0,0 +1,99 @@
+/*
+ * 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.catalog.v2
+
+import org.scalatest.Matchers._
+
+import org.apache.spark.sql.catalog.v2.{CatalogNotFoundException, CatalogPlugin}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, Analyzer}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+private class TestCatalogPlugin(override val name: String) extends CatalogPlugin {
+
+  override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = Unit
+}
+
+class ResolveMultipartIdentifierSuite extends AnalysisTest {
+  import CatalystSqlParser._
+
+  private val analyzer = makeAnalyzer(caseSensitive = false)
+
+  private val catalogs = Seq("prod", "test").map(name => name -> new TestCatalogPlugin(name)).toMap
+
+  private def lookupCatalog(catalog: String): CatalogPlugin =
+    catalogs.getOrElse(catalog, throw new CatalogNotFoundException("Not found"))
+
+  private def makeAnalyzer(caseSensitive: Boolean) = {
+    val conf = new SQLConf().copy(SQLConf.CASE_SENSITIVE -> caseSensitive)
+    new Analyzer(Some(lookupCatalog _), null, conf)
+  }
+
+  override protected def getAnalyzer(caseSensitive: Boolean) = analyzer
+
+  private def checkResolution(sqlText: String, expectedCatalog: Option[CatalogPlugin],
+      expectedNamespace: Array[String], expectedName: String): Unit = {
+
+    import analyzer.CatalogObjectIdentifier
+    val CatalogObjectIdentifier(catalog, ident) = parseMultipartIdentifier(sqlText)
+    catalog shouldEqual expectedCatalog
+    ident.namespace shouldEqual expectedNamespace
+    ident.name shouldEqual expectedName
+  }
+
+  private def checkTableResolution(sqlText: String,
+      expectedIdent: Option[TableIdentifier]): Unit = {
+
+    import analyzer.AsTableIdentifier
+    parseMultipartIdentifier(sqlText) match {
+      case AsTableIdentifier(ident) =>
+        assert(Some(ident) === expectedIdent)
+      case _ =>
+        assert(None === expectedIdent)
+    }
+  }
+
+  test("resolve multipart identifier") {
+    checkResolution("tbl", None, Array.empty, "tbl")
+    checkResolution("db.tbl", None, Array("db"), "tbl")
+    checkResolution("prod.func", catalogs.get("prod"), Array.empty, "func")
+    checkResolution("ns1.ns2.tbl", None, Array("ns1", "ns2"), "tbl")
+    checkResolution("prod.db.tbl", catalogs.get("prod"), Array("db"), "tbl")
+    checkResolution("test.db.tbl", catalogs.get("test"), Array("db"), "tbl")
+    checkResolution("test.ns1.ns2.ns3.tbl",
+      catalogs.get("test"), Array("ns1", "ns2", "ns3"), "tbl")
+    checkResolution("`db.tbl`", None, Array.empty, "db.tbl")
+    checkResolution("parquet.`file:/tmp/db.tbl`", None, Array("parquet"), "file:/tmp/db.tbl")
+    checkResolution("`org.apache.spark.sql.json`.`s3://buck/tmp/abc.json`", None,
+      Array("org.apache.spark.sql.json"), "s3://buck/tmp/abc.json")
+  }
+
+  test("resolve table identifier") {
+    checkTableResolution("tbl", Some(TableIdentifier("tbl")))
+    checkTableResolution("db.tbl", Some(TableIdentifier("tbl", Some("db"))))
+    checkTableResolution("prod.func", None)
+    checkTableResolution("ns1.ns2.tbl", None)
+    checkTableResolution("prod.db.tbl", None)
+    checkTableResolution("`db.tbl`", Some(TableIdentifier("db.tbl")))
+    checkTableResolution("parquet.`file:/tmp/db.tbl`",
+      Some(TableIdentifier("file:/tmp/db.tbl", Some("parquet"))))
+    checkTableResolution("`org.apache.spark.sql.json`.`s3://buck/tmp/abc.json`",
+      Some(TableIdentifier("s3://buck/tmp/abc.json", Some("org.apache.spark.sql.json"))))
+  }
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/v2/ResolveMultipartIdentifierSuite.scala

@@ -42,7 +42,7 @@ case class DataSourceResolution(
 
   import org.apache.spark.sql.catalog.v2.CatalogV2Implicits._
 
-  override protected def lookupCatalog(name: String): CatalogPlugin = findCatalog(name)
+  override def lookupCatalog: Option[String => CatalogPlugin] = Some(findCatalog)
 
   def defaultCatalog: Option[CatalogPlugin] = conf.defaultV2Catalog.map(findCatalog)