[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,17 @@ abstract class JdbcDialect extends SupportServiceLoader with Logging {
 
   def getSchemaHelper(): SchemaHelper
 
+  /**
+   * Mirrors the standard `Connection#setSchema(String)` JDBC API with a leading
+   * `conn` parameter, allowing dialects to customize the call. The default forwards
+   * to the standard API; dialects whose driver does not support it (e.g. Phoenix,
+   * Impala) should override this to a no-op. Callers should treat this as if it
+   * always takes effect.
+   */
+  def setSchema(conn: Connection, schema: String): Unit = {
+    conn.setSchema(schema)
+  }
+
   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,17 @@ class MySQLDialect extends JdbcDialect {
     query.toString()
   }
 
+  override def setSchema(conn: Connection, schema: String): Unit = {
+    // The MySQL Connector/J `databaseTerm` connection property selects which of
+    // setCatalog / setSchema actually switches the session database; the other
+    // call returns without effect for that mode (verified by black-box testing
+    // and consistent with the reference manual). Invoking both ensures the
+    // database is switched regardless of how the user configured the URL.
+    // StarRocksDialect / DorisDialect inherit this via MySQLDialect.
+    conn.setCatalog(schema)
+    conn.setSchema(schema)
+  }
+
   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

@@ -25,8 +25,9 @@ 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 +46,16 @@ class JdbcSessionImpl(
 
   private var databaseMetaData: DatabaseMetaData = _
 
+  /**
+   * The database the client requested via `USE_DATABASE` at session open. Used by
+   * `JdbcOperationManager` as the metadata schema filter. `None` when no `USE_DATABASE`
+   * was requested, or when the request was the literal `"default"`. The Hive JDBC
+   * driver sends `"default"` as a protocol stub when the user did not specify a
+   * database in the URL, so it cannot be distinguished from a genuine request and
+   * is treated as "no scope".
+   */
+  private[jdbc] var effectiveDatabase: Option[String] = None
+
   val sessionConf: KyuubiConf = normalizeConf
 
   private def normalizeConf: KyuubiConf = {
@@ -67,6 +78,14 @@ class JdbcSessionImpl(
       sessionConf,
       sessionConnection,
       sessionConf.get(ENGINE_JDBC_SESSION_INITIALIZE_SQL))
+    // The Hive JDBC driver sends `"default"` as USE_DATABASE when the user did not
+    // specify a database in the URL: a protocol stub indistinguishable from a real
+    // request. Filter it here so we don't push a non-existent / unintended schema
+    // filter into metadata operations on backends without a `default` database.
+    conf.get(USE_DATABASE).filter(_ != "default").foreach { database =>
+      JdbcDialects.get(sessionConf).setSchema(sessionConnection, database)
+      effectiveDatabase = Some(database)
+    }
     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)
+  }
 }

externals/kyuubi-jdbc-engine/src/test/scala/org/apache/kyuubi/engine/jdbc/starrocks/StarRocksOperationSuite.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.kyuubi.engine.jdbc.starrocks
 
-import java.sql.ResultSet
+import java.sql.{DriverManager, ResultSet}
 
 import scala.collection.mutable.ArrayBuffer
 
@@ -258,4 +258,86 @@ abstract class StarRocksOperationSuite extends WithStarRocksEngine with HiveJDBC
       statement.execute("drop database db2")
     }
   }
+
+  test("starrocks - 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=OLAP DISTRIBUTED BY HASH(`id`) BUCKETS 32 " +
+        "PROPERTIES ('replication_num' = '1')")
+      statement.execute("create database if not exists db7305b")
+      statement.execute("create table db7305b.tb(id bigint)" +
+        "ENGINE=OLAP DISTRIBUTED BY HASH(`id`) BUCKETS 32 " +
+        "PROPERTIES ('replication_num' = '1')")
+    }
+
+    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("starrocks - 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=OLAP DISTRIBUTED BY HASH(`id`) BUCKETS 32 " +
+        "PROPERTIES ('replication_num' = '1')")
+      statement.execute("create database if not exists db7305d")
+      statement.execute("create table db7305d.td(id bigint)" +
+        "ENGINE=OLAP DISTRIBUTED BY HASH(`id`) BUCKETS 32 " +
+        "PROPERTIES ('replication_num' = '1')")
+    }
+
+    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("starrocks - 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;"
+    intercept[java.sql.SQLException] {
+      DriverManager.getConnection(badUrl, user, password).close()
+    }
+  }
 }