[KYUUBI #7305] Fix JDBC engine returning tables from all databases when a database is specified in the URL

When DBeaver (or any Hive-JDBC client) connects to the Kyuubi JDBC engine with
a database path in the URL, DatabaseMetaData.getTables should return only the
tables from that database. Previously, the Hive JDBC driver converted a null
schemaPattern to "%" before sending the request, which the engine forwarded
straight into a WHERE TABLE_SCHEMA LIKE '%' clause on the backend, returning
tables from every database.

- Introduce JdbcDialect.setCurrentDatabase(connection, database) as an
  extension point; the default is a no-op returning false. MySQLDialect
  overrides it with USE `db` so MySQL/StarRocks/Doris switch the backend
  connection on session open.
- JdbcSessionImpl records the successfully applied database in
  effectiveDatabase and only keeps it when setCurrentDatabase returns true.
  "default" (the Hive JDBC driver's fallback when no database is specified in
  the URL) is tolerated on USE failure to stay compatible with backends that
  have no "default" database.
- JdbcOperationManager.newGetTablesOperation treats a blank schema or "%" as
  "use the session's effective database", routing through
  effectiveDatabase.orNull. Backends without an effective database
  (PostgreSQL, Oracle, etc.) keep the previous no-filter behavior.
- MySQLDialect now implements getCatalogsOperation and getSchemasOperation
  (previously featureNotSupported). getSchemasOperation aliases result columns
  to the JDBC-standard TABLE_CATALOG / TABLE_SCHEM so
  DatabaseMetaData.getSchemas works with clients such as DBeaver.

Closes #7305.

Author: wangzhigang1999

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/dialect/JdbcDialect.scala

@@ -82,6 +82,20 @@ abstract class JdbcDialect extends SupportServiceLoader with Logging {
 
   def getSchemaHelper(): SchemaHelper
 
+  /**
+   * Switch the current database on the backend connection.
+   *
+   * Called during session open when the client specified a database in the connection URL
+   * (populated by the Hive JDBC driver into `USE_DATABASE`). Dialects that support an
+   * in-session database context switch (e.g. MySQL's `USE db`) should override this and
+   * return `true` on success; the default is a no-op that returns `false`, meaning the
+   * backend has no notion of "current database" that applies to this session.
+   *
+   * The return value is used by `JdbcOperationManager` to decide whether the requested
+   * database can be treated as the effective schema filter for metadata operations.
+   */
+  def setCurrentDatabase(connection: Connection, database: String): Boolean = false
+
   def cancelStatement(jdbcStatement: Statement): Unit = {
     if (jdbcStatement != null) {
       jdbcStatement.cancel()

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/dialect/MySQLDialect.scala

@@ -36,6 +36,34 @@ class MySQLDialect extends JdbcDialect {
     statement
   }
 
+  override def getCatalogsOperation(): String = {
+    "SELECT CATALOG_NAME FROM INFORMATION_SCHEMA.SCHEMATA GROUP BY CATALOG_NAME"
+  }
+
+  override def getSchemasOperation(catalog: String, schema: String): String = {
+    // Alias the MySQL-native columns to the JDBC-standard names expected by
+    // DatabaseMetaData.getSchemas(): TABLE_CATALOG and TABLE_SCHEM.
+    val query = new StringBuilder(
+      """SELECT CATALOG_NAME AS TABLE_CATALOG, SCHEMA_NAME AS TABLE_SCHEM
+        |FROM INFORMATION_SCHEMA.SCHEMATA
+        |""".stripMargin)
+
+    val filters = ArrayBuffer[String]()
+    if (StringUtils.isNotBlank(catalog)) {
+      filters += s"CATALOG_NAME LIKE '$catalog'"
+    }
+    if (StringUtils.isNotBlank(schema)) {
+      filters += s"SCHEMA_NAME LIKE '$schema'"
+    }
+
+    if (filters.nonEmpty) {
+      query.append(" WHERE ")
+      query.append(filters.mkString(" AND "))
+    }
+
+    query.toString()
+  }
+
   override def getTablesQuery(
       catalog: String,
       schema: String,
@@ -122,6 +150,16 @@ class MySQLDialect extends JdbcDialect {
     query.toString()
   }
 
+  override def setCurrentDatabase(connection: Connection, database: String): Boolean = {
+    val stmt = connection.createStatement()
+    try {
+      stmt.execute(s"USE `$database`")
+      true
+    } finally {
+      stmt.close()
+    }
+  }
+
   override def getTRowSetGenerator(): JdbcTRowSetGenerator = new MySQLTRowSetGenerator
 
   override def getSchemaHelper(): SchemaHelper = {

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

@@ -18,6 +18,8 @@ package org.apache.kyuubi.engine.jdbc.operation
 
 import java.util
 
+import org.apache.commons.lang3.StringUtils
+
 import org.apache.kyuubi.KyuubiSQLException
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.{ENGINE_JDBC_FETCH_SIZE, ENGINE_JDBC_OPERATION_INCREMENTAL_COLLECT}
@@ -91,7 +93,12 @@ class JdbcOperationManager(conf: KyuubiConf) extends OperationManager("JdbcOpera
       schemaName: String,
       tableName: String,
       tableTypes: util.List[String]): Operation = {
-    val query = dialect.getTablesQuery(catalogName, schemaName, tableName, tableTypes)
+    val effectiveSchema = if (StringUtils.isBlank(schemaName) || schemaName == "%") {
+      session.asInstanceOf[JdbcSessionImpl].effectiveDatabase.orNull
+    } else {
+      schemaName
+    }
+    val query = dialect.getTablesQuery(catalogName, effectiveSchema, tableName, tableTypes)
     val normalizedConf = session.asInstanceOf[JdbcSessionImpl].normalizedConf
     val fetchSize = normalizedConf.get(ENGINE_JDBC_FETCH_SIZE.key).map(_.toInt)
       .getOrElse(session.sessionManager.getConf.get(ENGINE_JDBC_FETCH_SIZE))

externals/kyuubi-jdbc-engine/src/main/scala/org/apache/kyuubi/engine/jdbc/session/JdbcSessionImpl.scala

@@ -19,14 +19,16 @@ package org.apache.kyuubi.engine.jdbc.session
 import java.sql.{Connection, DatabaseMetaData}
 
 import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
 
 import org.apache.kyuubi.KyuubiSQLException
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
 import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_SESSION_HANDLE_KEY
 import org.apache.kyuubi.engine.jdbc.connection.ConnectionProvider
+import org.apache.kyuubi.engine.jdbc.dialect.JdbcDialects
 import org.apache.kyuubi.engine.jdbc.util.KyuubiJdbcUtils
-import org.apache.kyuubi.session.{AbstractSession, SessionHandle, SessionManager}
+import org.apache.kyuubi.session.{AbstractSession, SessionHandle, SessionManager, USE_DATABASE}
 import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TGetInfoType, TGetInfoValue, TProtocolVersion}
 
 class JdbcSessionImpl(
@@ -45,6 +47,13 @@ class JdbcSessionImpl(
 
   private var databaseMetaData: DatabaseMetaData = _
 
+  /**
+   * The database that was successfully applied to the backend connection during session open.
+   * `None` when no `USE_DATABASE` was requested, or when the request was for "default"
+   * (the Hive JDBC driver's fallback) and the backend had no such database.
+   */
+  private[jdbc] var effectiveDatabase: Option[String] = None
+
   val sessionConf: KyuubiConf = normalizeConf
 
   private def normalizeConf: KyuubiConf = {
@@ -67,6 +76,20 @@ class JdbcSessionImpl(
       sessionConf,
       sessionConnection,
       sessionConf.get(ENGINE_JDBC_SESSION_INITIALIZE_SQL))
+    conf.get(USE_DATABASE).foreach { database =>
+      try {
+        if (JdbcDialects.get(sessionConf).setCurrentDatabase(sessionConnection, database)) {
+          effectiveDatabase = Some(database)
+          info(s"Switched to database: $database")
+        }
+      } catch {
+        case NonFatal(e) if database == "default" =>
+          // The Hive JDBC driver sends "default" when the user didn't specify a database
+          // in the connection URL. Tolerate USE failure in that case so the session can
+          // still open against backends that have no "default" database.
+          warn(s"Failed to switch to database 'default', ignored.", e)
+      }
+    }
     super.open()
     info(s"The jdbc session is started.")
   }

externals/kyuubi-jdbc-engine/src/test/scala/org/apache/kyuubi/engine/jdbc/mysql/MySQLOperationSuite.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.kyuubi.engine.jdbc.mysql
 
-import java.sql.ResultSet
+import java.sql.{DriverManager, ResultSet}
 
 import scala.collection.mutable.ArrayBuffer
 
@@ -250,4 +250,102 @@ abstract class MySQLOperationSuite extends WithMySQLEngine with HiveJDBCTestHelp
       statement.execute("drop database db2")
     }
   }
+
+  test("mysql - getTables scopes to database from connection URL (KYUUBI #7305)") {
+    withJdbcStatement() { statement =>
+      statement.execute("create database if not exists db7305a")
+      statement.execute("create table db7305a.ta(id bigint)" +
+        "ENGINE=InnoDB DEFAULT CHARSET=utf8;")
+      statement.execute("create database if not exists db7305b")
+      statement.execute("create table db7305b.tb(id bigint)" +
+        "ENGINE=InnoDB DEFAULT CHARSET=utf8;")
+    }
+
+    val hostAndPort = jdbcUrl.stripPrefix("jdbc:hive2://").split("/;").head
+    val scopedUrl = s"jdbc:hive2://$hostAndPort/db7305a;"
+    val conn = DriverManager.getConnection(scopedUrl, user, password)
+    try {
+      val tables = conn.getMetaData.getTables(null, null, "t%", null)
+      val found = ArrayBuffer[(String, String)]()
+      while (tables.next()) {
+        found += ((tables.getString(TABLE_SCHEMA), tables.getString(TABLE_NAME)))
+      }
+      assert(found.contains(("db7305a", "ta")))
+      assert(!found.exists { case (schema, _) => schema == "db7305b" })
+
+      // Hive driver converts schemaPattern=null to "%", and the fix routes "%" back to
+      // the effective database, so the behavior should match the call above.
+      val tablesPct = conn.getMetaData.getTables(null, "%", "t%", null)
+      val foundPct = ArrayBuffer[(String, String)]()
+      while (tablesPct.next()) {
+        foundPct += ((tablesPct.getString(TABLE_SCHEMA), tablesPct.getString(TABLE_NAME)))
+      }
+      assert(foundPct.contains(("db7305a", "ta")))
+      assert(!foundPct.exists { case (schema, _) => schema == "db7305b" })
+    } finally {
+      conn.close()
+    }
+
+    withJdbcStatement() { statement =>
+      statement.execute("drop table db7305a.ta")
+      statement.execute("drop database db7305a")
+      statement.execute("drop table db7305b.tb")
+      statement.execute("drop database db7305b")
+    }
+  }
+
+  test("mysql - getTables returns all tables when no database in URL") {
+    withJdbcStatement() { statement =>
+      statement.execute("create database if not exists db7305c")
+      statement.execute("create table db7305c.tc(id bigint)" +
+        "ENGINE=InnoDB DEFAULT CHARSET=utf8;")
+      statement.execute("create database if not exists db7305d")
+      statement.execute("create table db7305d.td(id bigint)" +
+        "ENGINE=InnoDB DEFAULT CHARSET=utf8;")
+    }
+
+    withJdbcStatement() { statement =>
+      val tables = statement.getConnection.getMetaData.getTables(null, null, "t%", null)
+      val found = ArrayBuffer[(String, String)]()
+      while (tables.next()) {
+        found += ((tables.getString(TABLE_SCHEMA), tables.getString(TABLE_NAME)))
+      }
+      // Without USE_DATABASE, the fix must not filter, so both dbs are visible.
+      assert(found.contains(("db7305c", "tc")))
+      assert(found.contains(("db7305d", "td")))
+
+      statement.execute("drop table db7305c.tc")
+      statement.execute("drop database db7305c")
+      statement.execute("drop table db7305d.td")
+      statement.execute("drop database db7305d")
+    }
+  }
+
+  test("mysql - getSchemas returns all schemas (KYUUBI #7305)") {
+    withJdbcStatement() { statement =>
+      statement.execute("create database if not exists db7305e")
+      statement.execute("create database if not exists db7305f")
+    }
+
+    withJdbcStatement() { statement =>
+      val rs = statement.getConnection.getMetaData.getSchemas
+      val schemas = ArrayBuffer[String]()
+      while (rs.next()) schemas += rs.getString("TABLE_SCHEM")
+      assert(schemas.contains("db7305e"))
+      assert(schemas.contains("db7305f"))
+
+      statement.execute("drop database db7305e")
+      statement.execute("drop database db7305f")
+    }
+  }
+
+  test("mysql - session open fails when URL specifies a non-existent database") {
+    val hostAndPort = jdbcUrl.stripPrefix("jdbc:hive2://").split("/;").head
+    val badUrl = s"jdbc:hive2://$hostAndPort/does_not_exist_db_7305;"
+    val ex = intercept[java.sql.SQLException] {
+      DriverManager.getConnection(badUrl, user, password).close()
+    }
+    // Error surface could come from the engine or JDBC driver; we only assert it is raised.
+    assert(ex != null)
+  }
 }