test(e2e): cover hive_metastore and spark_catalog with escaped catalog names

Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>

Author: vuanhphung

tests/e2e/test_driver.py

@@ -740,6 +740,32 @@ def test_escaped_catalog_name_matches_exact_catalog(self):
             escaped = sorted(tuple(r) for r in cursor.fetchall())
         assert escaped == unescaped
 
+    def test_hive_metastore_metadata_unchanged_by_escaping(self):
+        # hive_metastore is the most common catalog name with `_`, so escaping must
+        # not change its results; spark_catalog (its alias) must not lose rows.
+        with self.connection({}) as escaped, self.connection(
+            {"escape_catalog_name": False}
+        ) as raw:
+
+            def rows(conn, call, **filters):
+                with conn.cursor() as cursor:
+                    getattr(cursor, call)(**filters)
+                    return {tuple(r) for r in cursor.fetchall()}
+
+            hms = {"catalog_name": "hive_metastore", "schema_name": "default"}
+            expected = rows(raw, "schemas", **hms)
+            if not expected:
+                pytest.skip("hive_metastore.default is not visible")
+            assert rows(escaped, "schemas", **hms) == expected
+
+            tables = rows(raw, "tables", **hms)
+            if tables:
+                one = dict(hms, table_name=min(t[2] for t in tables))
+                assert rows(escaped, "tables", **one) == rows(raw, "tables", **one)
+
+            spark = {"catalog_name": "spark_catalog", "schema_name": "default"}
+            assert rows(raw, "schemas", **spark) <= rows(escaped, "schemas", **spark)
+
     @skipUnless(pysql_supports_arrow(), "arrow test need arrow support")
     @pytest.mark.parametrize(
         "extra_params",

tests/e2e/test_kernel_backend.py

@@ -379,6 +379,21 @@ def test_tables_catalog_name_underscore_is_literal_by_default(
     assert all(row[0] == "system" for row in rows)
 
 
+def test_tables_hive_metastore_unchanged_by_escaping(kernel_conn_params):
+    # Kernel tables() honours escape_catalog_name; hive_metastore is the most
+    # common catalog name with `_`, so escaping must not change its results.
+    def rows(extra_params, **filters):
+        with sql.connect(**kernel_conn_params, **extra_params) as c, c.cursor() as cur:
+            cur.tables(catalog_name="hive_metastore", schema_name="default", **filters)
+            return {tuple(r) for r in cur.fetchall()}
+
+    listing = rows({"escape_catalog_name": False})
+    if not listing:
+        pytest.skip("hive_metastore.default is not visible")
+    one = {"table_name": min(t[2] for t in listing)}
+    assert rows({}, **one) == rows({"escape_catalog_name": False}, **one)
+
+
 def test_schemas_with_empty_string_filter_matches_nothing(conn):
     """An empty string is a real pattern, distinct from absent ``None``."""
     with conn.cursor() as cur: