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

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

@@ -26,35 +26,31 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 @Experimental
 trait LookupCatalog {
 
-  def lookupCatalog: Option[(String) => CatalogPlugin] = None
+  protected def lookupCatalog(name: String): CatalogPlugin
 
   type CatalogObjectIdentifier = (Option[CatalogPlugin], Identifier)
 
   /**
    * Extract catalog plugin and identifier from a multi-part identifier.
    */
   object CatalogObjectIdentifier {
-    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))
-          }
-      }
+    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)))
+        }
     }
   }
 
   /**
    * Extract legacy table identifier from a multi-part identifier.
    *
-   * For legacy support only. Please use
-   * [[org.apache.spark.sql.catalog.v2.LookupCatalog.CatalogObjectIdentifier]] in DSv2 code paths.
+   * For legacy support only. Please use [[CatalogObjectIdentifier]] instead on 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.{CatalogPlugin, LookupCatalog}
+import org.apache.spark.sql.catalog.v2.{CatalogNotFoundException, CatalogPlugin, LookupCatalog}
 import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.encoders.OuterScopes
@@ -52,7 +52,11 @@ object SimpleAnalyzer extends Analyzer(
     new SQLConf().copy(SQLConf.CASE_SENSITIVE -> true)) {
     override def createDatabase(dbDefinition: CatalogDatabase, ignoreIfExists: Boolean) {}
   },
-  new SQLConf().copy(SQLConf.CASE_SENSITIVE -> true))
+  new SQLConf().copy(SQLConf.CASE_SENSITIVE -> true)) {
+
+  override protected def lookupCatalog(name: String): CatalogPlugin =
+    throw new CatalogNotFoundException("No catalog lookup function")
+}
 
 /**
  * Provides a way to keep state during the analysis, this enables us to decouple the concerns
@@ -93,22 +97,16 @@ object AnalysisContext {
  * Provides a logical query plan analyzer, which translates [[UnresolvedAttribute]]s and
  * [[UnresolvedRelation]]s into fully typed objects using information in a [[SessionCatalog]].
  */
-class Analyzer(
+abstract class Analyzer(
     catalog: SessionCatalog,
     conf: SQLConf,
-    maxIterations: Int,
-    override val lookupCatalog: Option[(String) => CatalogPlugin] = None)
+    maxIterations: Int)
   extends RuleExecutor[LogicalPlan] with CheckAnalysis with LookupCatalog {
 
   def this(catalog: SessionCatalog, conf: SQLConf) = {
     this(catalog, conf, conf.optimizerMaxIterations)
   }
 
-  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 {
       val analyzed = executeAndTrack(plan, tracker)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisExternalCatalogSuite.scala

@@ -43,7 +43,7 @@ class AnalysisExternalCatalogSuite extends AnalysisTest with Matchers {
         CatalogStorageFormat.empty,
         StructType(Seq(StructField("a", IntegerType, nullable = true)))),
       ignoreIfExists = false)
-    new Analyzer(catalog, conf)
+    new TestAnalyzer(catalog, conf)
   }
 
   test("query builtin functions don't call the external catalog") {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisTest.scala

@@ -41,7 +41,7 @@ trait AnalysisTest extends PlanTest {
     catalog.createTempView("TaBlE", TestRelations.testRelation, overrideIfExists = true)
     catalog.createTempView("TaBlE2", TestRelations.testRelation2, overrideIfExists = true)
     catalog.createTempView("TaBlE3", TestRelations.testRelation3, overrideIfExists = true)
-    new Analyzer(catalog, conf) {
+    new TestAnalyzer(catalog, conf) {
       override val extendedResolutionRules = EliminateSubqueryAliases :: Nil
     }
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecisionSuite.scala

@@ -31,7 +31,7 @@ import org.apache.spark.sql.types._
 
 class DecimalPrecisionSuite extends AnalysisTest with BeforeAndAfter {
   private val catalog = new SessionCatalog(new InMemoryCatalog, EmptyFunctionRegistry, conf)
-  private val analyzer = new Analyzer(catalog, conf)
+  private val analyzer = new TestAnalyzer(catalog, conf)
 
   private val relation = LocalRelation(
     AttributeReference("i", IntegerType)(),

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/LookupFunctionsSuite.scala

@@ -36,7 +36,7 @@ class LookupFunctionsSuite extends PlanTest {
       catalog.createDatabase(
         CatalogDatabase("default", "", new URI("loc"), Map.empty),
         ignoreIfExists = false)
-      new Analyzer(catalog, conf)
+      new TestAnalyzer(catalog, conf)
     }
 
     def table(ref: String): LogicalPlan = UnresolvedRelation(TableIdentifier(ref))
@@ -63,7 +63,7 @@ class LookupFunctionsSuite extends PlanTest {
       catalog.createDatabase(
         CatalogDatabase("default", "", new URI("loc"), Map.empty),
         ignoreIfExists = false)
-      new Analyzer(catalog, conf)
+      new TestAnalyzer(catalog, conf)
     }
 
     def table(ref: String): LogicalPlan = UnresolvedRelation(TableIdentifier(ref))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TestAnalyzer.scala

@@ -0,0 +1,30 @@
+/*
+ * 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.catalog.v2.{CatalogNotFoundException, CatalogPlugin}
+import org.apache.spark.sql.catalyst.catalog._
+import org.apache.spark.sql.internal.SQLConf
+
+class TestAnalyzer(
+    catalog: SessionCatalog,
+    conf: SQLConf) extends Analyzer(catalog, conf) {
+
+  override protected def lookupCatalog(name: String): CatalogPlugin =
+    throw new CatalogNotFoundException("No catalog lookup function")
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -1429,7 +1429,7 @@ abstract class SessionCatalogSuite extends AnalysisTest {
       val catalog = new SessionCatalog(newBasicCatalog(), new SimpleFunctionRegistry, conf)
       catalog.setCurrentDatabase("db1")
       try {
-        val analyzer = new Analyzer(catalog, conf)
+        val analyzer = new TestAnalyzer(catalog, conf)
 
         // The analyzer should report the undefined function rather than the undefined table first.
         val cause = intercept[AnalysisException] {

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

@@ -0,0 +1,88 @@
+/*
+ * 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.Inside
+import org.scalatest.Matchers._
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalog.v2.{CatalogNotFoundException, CatalogPlugin, Identifier, LookupCatalog}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+private case class TestCatalogPlugin(override val name: String) extends CatalogPlugin {
+
+  override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = Unit
+}
+
+class LookupCatalogSuite extends SparkFunSuite with LookupCatalog with Inside {
+  import CatalystSqlParser._
+
+  private val catalogs = Seq("prod", "test").map(x => x -> new TestCatalogPlugin(x)).toMap
+
+  override def lookupCatalog(name: String): CatalogPlugin =
+    catalogs.getOrElse(name, throw new CatalogNotFoundException(s"$name not found"))
+
+  test("catalog object identifier") {
+    Seq(
+      ("tbl", None, Seq.empty, "tbl"),
+      ("db.tbl", None, Seq("db"), "tbl"),
+      ("prod.func", catalogs.get("prod"), Seq.empty, "func"),
+      ("ns1.ns2.tbl", None, Seq("ns1", "ns2"), "tbl"),
+      ("prod.db.tbl", catalogs.get("prod"), Seq("db"), "tbl"),
+      ("test.db.tbl", catalogs.get("test"), Seq("db"), "tbl"),
+      ("test.ns1.ns2.ns3.tbl", catalogs.get("test"), Seq("ns1", "ns2", "ns3"), "tbl"),
+      ("`db.tbl`", None, Seq.empty, "db.tbl"),
+      ("parquet.`file:/tmp/db.tbl`", None, Seq("parquet"), "file:/tmp/db.tbl"),
+      ("`org.apache.spark.sql.json`.`s3://buck/tmp/abc.json`", None,
+        Seq("org.apache.spark.sql.json"), "s3://buck/tmp/abc.json")).foreach {
+      case (sql, expectedCatalog, namespace, name) =>
+        inside(parseMultipartIdentifier(sql)) {
+          case CatalogObjectIdentifier(catalog, ident) =>
+            catalog shouldEqual expectedCatalog
+            ident shouldEqual Identifier.of(namespace.toArray, name)
+        }
+    }
+  }
+
+  test("table identifier") {
+    Seq(
+      ("tbl", "tbl", None),
+      ("db.tbl", "tbl", Some("db")),
+      ("`db.tbl`", "db.tbl", None),
+      ("parquet.`file:/tmp/db.tbl`", "file:/tmp/db.tbl", Some("parquet")),
+      ("`org.apache.spark.sql.json`.`s3://buck/tmp/abc.json`", "s3://buck/tmp/abc.json",
+        Some("org.apache.spark.sql.json"))).foreach {
+      case (sql, table, db) =>
+        inside (parseMultipartIdentifier(sql)) {
+          case AsTableIdentifier(ident) =>
+            ident shouldEqual TableIdentifier(table, db)
+        }
+    }
+    Seq(
+      "prod.func",
+      "prod.db.tbl",
+      "ns1.ns2.tbl").foreach { sql =>
+      parseMultipartIdentifier(sql) match {
+        case AsTableIdentifier(_) =>
+          fail(s"$sql should not be resolved as TableIdentifier")
+        case _ =>
+      }
+    }
+  }
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/AggregateOptimizeSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.optimizer
 
-import org.apache.spark.sql.catalyst.analysis.{Analyzer, EmptyFunctionRegistry}
+import org.apache.spark.sql.catalyst.analysis.{EmptyFunctionRegistry, TestAnalyzer}
 import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
@@ -31,7 +31,7 @@ import org.apache.spark.sql.internal.SQLConf.{CASE_SENSITIVE, GROUP_BY_ORDINAL}
 class AggregateOptimizeSuite extends PlanTest {
   override val conf = new SQLConf().copy(CASE_SENSITIVE -> false, GROUP_BY_ORDINAL -> false)
   val catalog = new SessionCatalog(new InMemoryCatalog, EmptyFunctionRegistry, conf)
-  val analyzer = new Analyzer(catalog, conf)
+  val analyzer = new TestAnalyzer(catalog, conf)
 
   object Optimize extends RuleExecutor[LogicalPlan] {
     val batches = Batch("Aggregate", FixedPoint(100),

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/BooleanSimplificationSuite.scala

@@ -189,7 +189,7 @@ class BooleanSimplificationSuite extends PlanTest with ExpressionEvalHelper with
   }
 
   private val caseInsensitiveConf = new SQLConf().copy(SQLConf.CASE_SENSITIVE -> false)
-  private val caseInsensitiveAnalyzer = new Analyzer(
+  private val caseInsensitiveAnalyzer = new TestAnalyzer(
     new SessionCatalog(new InMemoryCatalog, EmptyFunctionRegistry, caseInsensitiveConf),
     caseInsensitiveConf)