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

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)
+  }
 }