[KYUUBI #7305] Replace higher-order function operations with concrete classes

Convert the metadata and lifecycle operations from higher-order-function
constructor parameters to concrete subclasses, matching the per-operation
file style of the spark/hive engines. ExecuteMetaDataOperation becomes an
abstract base whose subclasses (GetTypeInfo, GetCatalogs, GetSchemas,
GetTables, GetTableTypes, GetColumns, GetFunctions, GetPrimaryKeys,
GetCrossReference) each fix the metadata RPC; the four lifecycle classes
(SetCurrentCatalog/Database, GetCurrentCatalog/Database) likewise own the
catalog/database parameter directly and call the dialect from runInternal.
JdbcOperationManager's factory methods collapse to one-liners.

Author: wangzhigang1999

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/ExecuteMetaDataOperation.scala

@@ -31,13 +31,12 @@ import org.apache.kyuubi.session.Session
 import org.apache.kyuubi.shaded.hive.service.rpc.thrift.TFetchResultsResp
 
 /**
- * Runs a `DatabaseMetaData.getXxx(...)` call against the session connection and exposes the
- * resulting [[ResultSet]] as the operation output.
+ * Base class that streams the result of a `DatabaseMetaData.getXxx(...)` call against the
+ * session connection. Subclasses fix the particular metadata RPC via [[runMetaDataCall]].
  */
-class ExecuteMetaDataOperation(
-    session: Session,
-    metadataCall: ExecuteMetaDataOperation.MetaDataCall)
-  extends JdbcOperation(session) {
+abstract class ExecuteMetaDataOperation(session: Session) extends JdbcOperation(session) {
+
+  protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet
 
   private val operationLog: OperationLog = OperationLog.createOperationLog(session, getHandle)
   override def getOperationLog: Option[OperationLog] = Option(operationLog)
@@ -48,7 +47,7 @@ class ExecuteMetaDataOperation(
     setState(OperationState.RUNNING)
     try {
       val connection = session.asInstanceOf[JdbcSessionImpl].sessionConnection
-      val resultSet = metadataCall(dialect, connection)
+      val resultSet = runMetaDataCall(dialect, connection)
       val iterator = new ResultSetFetchIterator(resultSet)
       try {
         // Normalise spec-divergent column labels (e.g. Impala HS2 TABLE_MD -> TABLE_SCHEM).
@@ -118,7 +117,3 @@ class ExecuteMetaDataOperation(
     }
   }
 }
-
-object ExecuteMetaDataOperation {
-  type MetaDataCall = (JdbcDialect, Connection) => ResultSet
-}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetCatalogs.scala

@@ -0,0 +1,27 @@
+/*
+ * 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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetCatalogs(session: Session) extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getCatalogs(conn)
+}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetColumns.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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetColumns(
+    session: Session,
+    catalog: String,
+    schema: String,
+    tableName: String,
+    columnName: String)
+  extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getColumns(conn, catalog, schema, tableName, columnName)
+}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetCrossReference.scala

@@ -0,0 +1,42 @@
+/*
+ * 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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetCrossReference(
+    session: Session,
+    primaryCatalog: String,
+    primarySchema: String,
+    primaryTable: String,
+    foreignCatalog: String,
+    foreignSchema: String,
+    foreignTable: String)
+  extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getCrossReference(
+      conn,
+      primaryCatalog,
+      primarySchema,
+      primaryTable,
+      foreignCatalog,
+      foreignSchema,
+      foreignTable)
+}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetCurrentCatalog.scala

@@ -16,22 +16,20 @@
  */
 package org.apache.kyuubi.engine.jdbc.operation
 
-import java.sql.{Connection, Types}
+import java.sql.Types
 
-import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
 import org.apache.kyuubi.engine.jdbc.schema.{Column, Row, Schema}
 import org.apache.kyuubi.engine.jdbc.session.JdbcSessionImpl
 import org.apache.kyuubi.operation.{ArrayFetchIterator, OperationState}
 import org.apache.kyuubi.session.Session
 
-class GetCurrentCatalog(session: Session, fetch: (JdbcDialect, Connection) => String)
-  extends JdbcOperation(session) {
+class GetCurrentCatalog(session: Session) extends JdbcOperation(session) {
 
   override protected def runInternal(): Unit = {
     setState(OperationState.RUNNING)
     try {
       val connection = session.asInstanceOf[JdbcSessionImpl].sessionConnection
-      val catalog = fetch(dialect, connection)
+      val catalog = dialect.getCatalog(connection)
       schema = Schema(List(Column(
         "TABLE_CAT",
         "VARCHAR",

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetCurrentDatabase.scala

@@ -16,22 +16,20 @@
  */
 package org.apache.kyuubi.engine.jdbc.operation
 
-import java.sql.{Connection, Types}
+import java.sql.Types
 
-import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
 import org.apache.kyuubi.engine.jdbc.schema.{Column, Row, Schema}
 import org.apache.kyuubi.engine.jdbc.session.JdbcSessionImpl
 import org.apache.kyuubi.operation.{ArrayFetchIterator, OperationState}
 import org.apache.kyuubi.session.Session
 
-class GetCurrentDatabase(session: Session, fetch: (JdbcDialect, Connection) => String)
-  extends JdbcOperation(session) {
+class GetCurrentDatabase(session: Session) extends JdbcOperation(session) {
 
   override protected def runInternal(): Unit = {
     setState(OperationState.RUNNING)
     try {
       val connection = session.asInstanceOf[JdbcSessionImpl].sessionConnection
-      val database = fetch(dialect, connection)
+      val database = dialect.getCurrentSchema(connection)
       schema = Schema(List(Column(
         "TABLE_SCHEM",
         "VARCHAR",

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetFunctions.scala

@@ -0,0 +1,28 @@
+/*
+ * 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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetFunctions(session: Session, catalog: String, schema: String, functionName: String)
+  extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getFunctions(conn, catalog, schema, functionName)
+}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetPrimaryKeys.scala

@@ -0,0 +1,28 @@
+/*
+ * 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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetPrimaryKeys(session: Session, catalog: String, schema: String, tableName: String)
+  extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getPrimaryKeys(conn, catalog, schema, tableName)
+}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetSchemas.scala

@@ -0,0 +1,28 @@
+/*
+ * 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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetSchemas(session: Session, catalog: String, schema: String)
+  extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getSchemas(conn, catalog, schema)
+}

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/operation/GetTableTypes.scala

@@ -0,0 +1,27 @@
+/*
+ * 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.kyuubi.engine.jdbc.operation
+
+import java.sql.{Connection, ResultSet}
+
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialect
+import org.apache.kyuubi.session.Session
+
+class GetTableTypes(session: Session) extends ExecuteMetaDataOperation(session) {
+  override protected def runMetaDataCall(dialect: JdbcDialect, conn: Connection): ResultSet =
+    dialect.getTableTypes(conn)
+}