Repository: ibis-project/ibis

Issue:
bug(postgres): cross-schema column mixing
### What happened?

On `con.table("accounts").to_pandas()` I receive an error `ValueError: 10 columns passed, passed data had 5 columns`. I believe ibis is picking up columns from an `accounts` table in a different schema (`raw`).

I expected the default `con.table("accounts")` to only pick columns from the default `accounts` table (`public` schema/database).

I can avoid this by specifying `database`, but figured I'd post this ticket in case this is unintended behavior.

```sql
CREATE TABLE accounts (
    account_id INTEGER PRIMARY KEY NOT NULL UNIQUE
    ,account_name TEXT NOT NULL
    ,account_timezone TEXT
    ,account_notes TEXT
    ,account_active BOOLEAN NOT NULL
);

CREATE TABLE IF NOT EXISTS raw.accounts (
    id TEXT
    ,name TEXT
    ,timezone TEXT
    ,notes TEXT
    ,active TEXT
);
```

```python
sc.con.table("accounts")
Out[88]: 
DatabaseTable: accounts
  account_id       !int32
  id               string
  account_name     !string
  name             string
  account_timezone string
  timezone         string
  notes            string
  account_notes    string
  account_active   !boolean
  active           string
```

Also, sorry for the raw_sql. (:

### What version of ibis are you using?

9.1

### What backend(s) are you using, if any?

Postgres

### Relevant log output

```sh
sc.con.table("accounts")
Out[93]: 
DatabaseTable: accounts
  account_id       !int32
  id               string
  account_name     !string
  name             string
  account_timezone string
  timezone         string
  notes            string
  account_notes    string
  account_active   !boolean
  active           string

sc.con.raw_sql("DROP TABLE accounts")
Out[94]: <cursor object at 0x71512acb55d0; closed: 0>

sc.con.table("accounts")
Out[95]: 
DatabaseTable: accounts
  id       string
  name     string
  timezone string
  notes    string
  active   string

sc.con.table("accounts", schema="public")
/home/patrick/anaconda3/envs/.../lib/python3.12/site-packages/ibis/backends/sql/__init__.py:36: FutureWarning: `schema` is deprecated as of v9.0, removed in v10.0; Use the `database` kwarg with one of the following patterns:
database="database"
database=("catalog", "database")
database="catalog.database"
  util.warn_deprecated(
Traceback (most recent call last):

  Cell In[96], line 1
    sc.con.table("accounts", schema="public")

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/backends/sql/__init__.py:137 in table
    table_schema = self.get_schema(name, catalog=catalog, database=database)

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/backends/postgres/__init__.py:600 in get_schema
    raise com.IbisError(f"Table not found: {name!r}")

IbisError: Table not found: 'accounts'


sc.con.raw_sql("""
CREATE TABLE accounts (
    account_id INTEGER PRIMARY KEY NOT NULL UNIQUE
    ,account_name TEXT NOT NULL
    ,account_timezone TEXT
    ,account_notes TEXT
    ,account_active BOOLEAN NOT NULL
);
""")
Out[97]: <cursor object at 0x71512acb4e50; closed: 0>

sc.con.table("accounts", database="public")
Out[99]: 
DatabaseTable: public.accounts
  account_id       !int32
  account_name     !string
  account_timezone string
  account_notes    string
  account_active   !boolean

sc.con.table("accounts")
Out[100]: 
DatabaseTable: accounts
  account_id       !int32
  id               string
  account_name     !string
  name             string
  account_timezone string
  timezone         string
  notes            string
  account_notes    string
  account_active   !boolean
  active           string
```


### Code of Conduct

- [X] I agree to follow this project's Code of Conduct

Discussion:
cpcloud: It looks like `temp` takes precedence if the names collide, at least in DuckDB:

```
D create table t (x int);
D create temp table t (y int);
D table t;
┌────────┐
│   y    │
│ int32  │
├────────┤
│ 0 rows │
└────────┘
```
cpcloud: Pushed up the additional test. It passed without changes 🥳 
gforsyth: > It looks like `temp` takes precedence if the names collide, at least in DuckDB:

Ok, I guess we need to decide on what convention we want to enforce?
gforsyth: lol, I'm going to have to relax the cross-backend exception check a fair bit until we get Naty's `TableNotFound` PR in, but I think this should go in before 10.0
gforsyth: ok, `datafusion` and `clickhouse` are failing b/c the exception doesn't match.

MSSQL is failing the way postgres and duckdb did before the fix.
gforsyth: A hilarious error message from MySQL:
```
        if database is not None and database != self.current_database:
>           raise com.UnsupportedOperationError(
                "Creating tables in other databases is not supported by Postgres"
            )
E           ibis.common.exceptions.UnsupportedOperationError: Creating tables in other databases is not supported by Postgres
```

Need to fix that, too...
gforsyth: Ok, this turned into a bit of a nightmare.  I'll pick it back up tomorrow or Monday
cpcloud: Exasol does indeed allow creating tables in other databases, but there was a bug in our code (we weren't quoting identifiers).
cpcloud: The only remaining task is to decide how we want to scope temp tables versus non-temp tables in backends that support them. We can defer that to a follow-up IMO.
cpcloud: BigQuery and Snowflake are failing because the test assumes the backend can differentiate between int32 and int64, which isn't true for both BigQuery and Snowflake. All integers are at least int64 for both, Snowflake is actually int128 (by way of using `decimal(38, 0)`).
cpcloud: I adjusted the test to use `int64` and `string` because of the issue described in the previous comment.
gforsyth: I raised https://github.com/apache/datafusion-python/issues/796 in re: a better exception for missing tables, at least

Patch:
diff --git a/.github/workflows/set-milestone-on-issue.yml b/.github/workflows/set-milestone-on-issue.yml
new file mode 100644
index 000000000..36dfdcb71
--- /dev/null
+++ b/.github/workflows/set-milestone-on-issue.yml
@@ -0,0 +1,24 @@
+name: Set milestone on issues closed by PR
+
+on:
+  push:
+    branches:
+      - main
+
+concurrency:
+  group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}
+
+permissions:
+  issues: write
+  pull-requests: read
+
+jobs:
+  set-milestone:
+    if: github.repository == 'ibis-project/ibis'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: set milestone
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: ./ci/set-milestone-on-issue.sh "${GITHUB_SHA}"
diff --git a/ci/linked-issues.gql b/ci/linked-issues.gql
new file mode 100644
index 000000000..7e71ab04b
--- /dev/null
+++ b/ci/linked-issues.gql
@@ -0,0 +1,11 @@
+query ($owner: String!, $repo: String!, $pr: Int!) {
+  repository(owner: $owner, name: $repo) {
+    pullRequest(number: $pr) {
+      closingIssuesReferences(first: 100) {
+        nodes {
+          number
+        }
+      }
+    }
+  }
+}
diff --git a/ci/set-milestone-on-issue.sh b/ci/set-milestone-on-issue.sh
new file mode 100755
index 000000000..8434d63ee
--- /dev/null
+++ b/ci/set-milestone-on-issue.sh
@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+top="$(dirname "$(readlink -f "$0")")"
+
+# find all pull requests associated with commit
+linked_issues_query='.data.repository.pullRequest.closingIssuesReferences.nodes[].number'
+
+gh pr list --search "$1" --state merged --json number --jq '.[].number' |
+  sed '/^$/d' |
+  while read -r pr; do
+    milestone="$(gh pr view "${pr}" --json milestone --jq '.milestone.title')"
+
+    if [ -n "${milestone}" ]; then
+      # find all issues associated with said pull requests
+      # taken from https://github.com/cli/cli/discussions/7097#discussioncomment-5229031
+      readarray -t issues < <(
+        gh api graphql -F owner=ibis-project -F repo=ibis -F pr="${pr}" -F query="@${top}/linked-issues.gql" \
+        --jq "${linked_issues_query}" | sed '/^$/d')
+
+      if [ "${#issues[@]}" -gt 0 ]; then
+        gh issue edit "${issues[@]}" --milestone "${milestone}"
+      fi
+    fi
+  done
diff --git a/docker/mysql/startup.sql b/docker/mysql/startup.sql
index 29982e2f7..38d0d54b9 100644
--- a/docker/mysql/startup.sql
+++ b/docker/mysql/startup.sql
@@ -1,5 +1,4 @@
 CREATE USER 'ibis'@'localhost' IDENTIFIED BY 'ibis';
 CREATE SCHEMA IF NOT EXISTS test_schema;
-GRANT CREATE, DROP ON *.* TO 'ibis'@'%';
-GRANT CREATE,SELECT,DROP ON `test_schema`.* TO 'ibis'@'%';
+GRANT CREATE,SELECT,DROP ON *.* TO 'ibis'@'%';
 FLUSH PRIVILEGES;
diff --git a/ibis/backends/duckdb/__init__.py b/ibis/backends/duckdb/__init__.py
index c59feade2..4b60f87a1 100644
--- a/ibis/backends/duckdb/__init__.py
+++ b/ibis/backends/duckdb/__init__.py
@@ -337,41 +337,29 @@ class Backend(SQLBackend, CanCreateDatabase, CanCreateSchema, UrlFromPath):
         -------
         sch.Schema
             Ibis schema
-
         """
-        conditions = [sg.column("table_name").eq(sge.convert(table_name))]
-
-        if catalog is not None:
-            conditions.append(sg.column("table_catalog").eq(sge.convert(catalog)))
-
-        if database is not None:
-            conditions.append(sg.column("table_schema").eq(sge.convert(database)))
-
-        query = (
-            sg.select(
-                "column_name",
-                "data_type",
-                sg.column("is_nullable").eq(sge.convert("YES")).as_("nullable"),
+        query = sge.Describe(
+            this=sg.table(
+                table_name, db=database, catalog=catalog, quoted=self.compiler.quoted
             )
-            .from_(sg.table("columns", db="information_schema"))
-            .where(sg.and_(*conditions))
-            .order_by("ordinal_position")
-        )
+        ).sql(self.dialect)
 
-        with self._safe_raw_sql(query) as cur:
-            meta = cur.fetch_arrow_table()
-
-        if not meta:
+        try:
+            result = self.con.sql(query)
+        except duckdb.CatalogException:
             raise exc.IbisError(f"Table not found: {table_name!r}")
+        else:
+            meta = result.fetch_arrow_table()
 
         names = meta["column_name"].to_pylist()
-        types = meta["data_type"].to_pylist()
-        nullables = meta["nullable"].to_pylist()
+        types = meta["column_type"].to_pylist()
+        nullables = meta["null"].to_pylist()
 
+        type_mapper = self.compiler.type_mapper
         return sch.Schema(
             {
-                name: self.compiler.type_mapper.from_string(typ, nullable=nullable)
-                for name, typ, nullable in zip(names, types, nullables)
+                name: type_mapper.from_string(typ, nullable=null == "YES")
+                for name, typ, null in zip(names, types, nullables)
             }
         )
 
@@ -512,7 +500,7 @@ class Backend(SQLBackend, CanCreateDatabase, CanCreateSchema, UrlFromPath):
         query = (
             sg.select(f.anon.unnest(f.list_append(C.aliases, C.extension_name)))
             .from_(f.duckdb_extensions())
-            .where(sg.and_(C.installed, C.loaded))
+            .where(C.installed, C.loaded)
         )
         with self._safe_raw_sql(query) as cur:
             installed = map(itemgetter(0), cur.fetchall())
diff --git a/ibis/backends/exasol/__init__.py b/ibis/backends/exasol/__init__.py
index 5b1acd584..89ce6dec9 100644
--- a/ibis/backends/exasol/__init__.py
+++ b/ibis/backends/exasol/__init__.py
@@ -345,16 +345,9 @@ class Backend(SQLBackend, CanCreateDatabase, CanCreateSchema):
 
         if temp:
             raise com.UnsupportedOperationError(
-                "Creating temp tables is not supported by Exasol."
+                f"Creating temp tables is not supported by {self.name}"
             )
 
-        if database is not None and database != self.current_database:
-            raise com.UnsupportedOperationError(
-                "Creating tables in other databases is not supported by Exasol"
-            )
-        else:
-            database = None
-
         quoted = self.compiler.quoted
 
         temp_memtable_view = None
@@ -435,7 +428,11 @@ class Backend(SQLBackend, CanCreateDatabase, CanCreateSchema):
             raise NotImplementedError(
                 "`catalog` argument is not supported for the Exasol backend"
             )
-        drop_schema = sg.exp.Drop(kind="SCHEMA", this=name, exists=force)
+        drop_schema = sg.exp.Drop(
+            kind="SCHEMA",
+            this=sg.to_identifier(name, quoted=self.compiler.quoted),
+            exists=force,
+        )
         with self.begin() as con:
             con.execute(drop_schema.sql(dialect=self.dialect))
 
@@ -446,7 +443,11 @@ class Backend(SQLBackend, CanCreateDatabase, CanCreateSchema):
             raise NotImplementedError(
                 "`catalog` argument is not supported for the Exasol backend"
             )
-        create_database = sg.exp.Create(kind="SCHEMA", this=name, exists=force)
+        create_database = sg.exp.Create(
+            kind="SCHEMA",
+            this=sg.to_identifier(name, quoted=self.compiler.quoted),
+            exists=force,
+        )
         open_database = self.current_database
         with self.begin() as con:
             con.execute(create_database.sql(dialect=self.dialect))
diff --git a/ibis/backends/mssql/__init__.py b/ibis/backends/mssql/__init__.py
index bb8c5ed72..a4e711b93 100644
--- a/ibis/backends/mssql/__init__.py
+++ b/ibis/backends/mssql/__init__.py
@@ -178,19 +178,15 @@ class Backend(SQLBackend, CanCreateCatalog, CanCreateDatabase, CanCreateSchema,
         if name.startswith("ibis_cache_"):
             catalog, database = ("tempdb", "dbo")
             name = "##" + name
-        conditions = [sg.column("table_name").eq(sge.convert(name))]
-
-        if database is not None:
-            conditions.append(sg.column("table_schema").eq(sge.convert(database)))
 
         query = (
             sg.select(
-                "column_name",
-                "data_type",
-                "is_nullable",
-                "numeric_precision",
-                "numeric_scale",
-                "datetime_precision",
+                C.column_name,
+                C.data_type,
+                C.is_nullable,
+                C.numeric_precision,
+                C.numeric_scale,
+                C.datetime_precision,
             )
             .from_(
                 sg.table(
@@ -199,8 +195,11 @@ class Backend(SQLBackend, CanCreateCatalog, CanCreateDatabase, CanCreateSchema,
                     catalog=catalog or self.current_catalog,
                 )
             )
-            .where(*conditions)
-            .order_by("ordinal_position")
+            .where(
+                C.table_name.eq(sge.convert(name)),
+                C.table_schema.eq(sge.convert(database or self.current_database)),
+            )
+            .order_by(C.ordinal_position)
         )
 
         with self._safe_raw_sql(query) as cur:
@@ -487,13 +486,9 @@ GO"""
         """
         table_loc = self._warn_and_create_table_loc(database, schema)
         catalog, db = self._to_catalog_db_tuple(table_loc)
-        conditions = []
-
-        if db:
-            conditions.append(C.table_schema.eq(sge.convert(db)))
 
         sql = (
-            sg.select("table_name")
+            sg.select(C.table_name)
             .from_(
                 sg.table(
                     "TABLES",
@@ -501,12 +496,10 @@ GO"""
                     catalog=catalog if catalog is not None else self.current_catalog,
                 )
             )
+            .where(C.table_schema.eq(sge.convert(db or self.current_database)))
             .distinct()
         )
 
-        if conditions:
-            sql = sql.where(*conditions)
-
         sql = sql.sql(self.dialect)
 
         with self._safe_raw_sql(sql) as cur:
diff --git a/ibis/backends/mysql/__init__.py b/ibis/backends/mysql/__init__.py
index 02045d76e..9d1e39263 100644
--- a/ibis/backends/mysql/__init__.py
+++ b/ibis/backends/mysql/__init__.py
@@ -397,13 +397,6 @@ class Backend(SQLBackend, CanCreateDatabase):
         if obj is None and schema is None:
             raise ValueError("Either `obj` or `schema` must be specified")
 
-        if database is not None and database != self.current_database:
-            raise com.UnsupportedOperationError(
-                "Creating tables in other databases is not supported by Postgres"
-            )
-        else:
-            database = None
-
         properties = []
 
         if temp:
diff --git a/ibis/backends/postgres/__init__.py b/ibis/backends/postgres/__init__.py
index e1963503b..801514d32 100644
--- a/ibis/backends/postgres/__init__.py
+++ b/ibis/backends/postgres/__init__.py
@@ -303,6 +303,20 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase, CanCreateSchema):
         with self.begin() as cur:
             cur.execute("SET TIMEZONE = UTC")
 
+    @property
+    def _session_temp_db(self) -> str | None:
+        # Postgres doesn't assign the temporary table database until the first
+        # temp table is created in a given session.
+        # Before that temp table is created, this will return `None`
+        # After a temp table is created, it will return `pg_temp_N` where N is
+        # some integer
+        res = self.raw_sql(
+            "select nspname from pg_namespace where oid = pg_my_temp_schema()"
+        ).fetchone()
+        if res is not None:
+            return res[0]
+        return res
+
     def list_tables(
         self,
         like: str | None = None,
@@ -458,7 +472,7 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase, CanCreateSchema):
                 on=n.oid.eq(p.pronamespace),
                 join_type="LEFT",
             )
-            .where(sg.and_(*predicates))
+            .where(*predicates)
         )
 
         def split_name_type(arg: str) -> tuple[str, dt.DataType]:
@@ -571,6 +585,16 @@ $$""".format(**self._get_udf_source(udf_node))
 
         format_type = self.compiler.f["pg_catalog.format_type"]
 
+        # If no database is specified, assume the current database
+        db = database or self.current_database
+
+        dbs = [sge.convert(db)]
+
+        # If a database isn't specified, then include temp tables in the
+        # returned values
+        if database is None and (temp_table_db := self._session_temp_db) is not None:
+            dbs.append(sge.convert(temp_table_db))
+
         type_info = (
             sg.select(
                 a.attname.as_("column_name"),
@@ -591,7 +615,7 @@ $$""".format(**self._get_udf_source(udf_node))
             .where(
                 a.attnum > 0,
                 sg.not_(a.attisdropped),
-                n.nspname.eq(sge.convert(database)) if database is not None else TRUE,
+                n.nspname.isin(*dbs),
                 c.relname.eq(sge.convert(name)),
             )
             .order_by(a.attnum)
diff --git a/ibis/backends/risingwave/__init__.py b/ibis/backends/risingwave/__init__.py
index 960ab9522..d43ca2b51 100644
--- a/ibis/backends/risingwave/__init__.py
+++ b/ibis/backends/risingwave/__init__.py
@@ -586,3 +586,9 @@ class Backend(PostgresBackend):
         )
         with self._safe_raw_sql(src):
             pass
+
+    @property
+    def _session_temp_db(self) -> str | None:
+        # Return `None`, because RisingWave does not implement temp tables like
+        # Postgres
+        return None
diff --git a/ibis/backends/snowflake/__init__.py b/ibis/backends/snowflake/__init__.py
index 448776ccb..64d570281 100644
--- a/ibis/backends/snowflake/__init__.py
+++ b/ibis/backends/snowflake/__init__.py
@@ -646,6 +646,13 @@ $$"""
         catalog: str | None = None,
         database: str | None = None,
     ) -> Iterable[tuple[str, dt.DataType]]:
+        # this will always show temp tables with the same name as a non-temp
+        # table first
+        #
+        # snowflake puts temp tables in the same catalog and database as
+        # non-temp tables and differentiates between them using a different
+        # mechanism than other database that often put temp tables in a hidden
+        # or intentionall-difficult-to-access catalog/database
         table = sg.table(
             table_name, db=database, catalog=catalog, quoted=self.compiler.quoted
         )
diff --git a/ibis/backends/trino/__init__.py b/ibis/backends/trino/__init__.py
index c86bf33d2..0bbd32045 100644
--- a/ibis/backends/trino/__init__.py
+++ b/ibis/backends/trino/__init__.py
@@ -139,20 +139,18 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase, CanCreateSchema):
             Ibis schema
 
         """
-        conditions = [sg.column("table_name").eq(sge.convert(table_name))]
-
-        if database is not None:
-            conditions.append(sg.column("table_schema").eq(sge.convert(database)))
-
         query = (
             sg.select(
-                "column_name",
-                "data_type",
-                sg.column("is_nullable").eq(sge.convert("YES")).as_("nullable"),
+                C.column_name,
+                C.data_type,
+                C.is_nullable.eq(sge.convert("YES")).as_("nullable"),
             )
             .from_(sg.table("columns", db="information_schema", catalog=catalog))
-            .where(sg.and_(*conditions))
-            .order_by("ordinal_position")
+            .where(
+                C.table_name.eq(sge.convert(table_name)),
+                C.table_schema.eq(sge.convert(database or self.current_database)),
+            )
+            .order_by(C.ordinal_position)
         )
 
         with self._safe_raw_sql(query) as cur:
@@ -162,9 +160,11 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase, CanCreateSchema):
             fqn = sg.table(table_name, db=database, catalog=catalog).sql(self.name)
             raise com.IbisError(f"Table not found: {fqn}")
 
+        type_mapper = self.compiler.type_mapper
+
         return sch.Schema(
             {
-                name: self.compiler.type_mapper.from_string(typ, nullable=nullable)
+                name: type_mapper.from_string(typ, nullable=nullable)
                 for name, typ, nullable in meta
             }
         )


Test patch:
diff --git a/ibis/backends/duckdb/tests/test_client.py b/ibis/backends/duckdb/tests/test_client.py
index ac55ba940..c6729e13a 100644
--- a/ibis/backends/duckdb/tests/test_client.py
+++ b/ibis/backends/duckdb/tests/test_client.py
@@ -345,3 +345,33 @@ def test_hugging_face(con, url, method_name):
     method = getattr(con, method_name)
     t = method(url)
     assert t.count().execute() > 0
+
+
+def test_multiple_tables_with_the_same_name(tmp_path):
+    # check within the same database
+    path = tmp_path / "test1.ddb"
+    with duckdb.connect(str(path)) as con:
+        con.execute("CREATE TABLE t (x INT)")
+        con.execute("CREATE SCHEMA s")
+        con.execute("CREATE TABLE s.t (y STRING)")
+
+    con = ibis.duckdb.connect(path)
+    t1 = con.table("t")
+    t2 = con.table("t", database="s")
+    assert t1.schema() == ibis.schema({"x": "int32"})
+    assert t2.schema() == ibis.schema({"y": "string"})
+
+    path = tmp_path / "test2.ddb"
+    with duckdb.connect(str(path)) as c:
+        c.execute("CREATE TABLE t (y DOUBLE[])")
+
+    # attach another catalog and check that too
+    con.attach(path, name="w")
+    t1 = con.table("t")
+    t2 = con.table("t", database="s")
+    assert t1.schema() == ibis.schema({"x": "int32"})
+    assert t2.schema() == ibis.schema({"y": "string"})
+
+    t3 = con.table("t", database="w.main")
+
+    assert t3.schema() == ibis.schema({"y": "array<float64>"})
diff --git a/ibis/backends/tests/errors.py b/ibis/backends/tests/errors.py
index 044b85dc9..b13de521b 100644
--- a/ibis/backends/tests/errors.py
+++ b/ibis/backends/tests/errors.py
@@ -55,8 +55,9 @@ except ImportError:
 
 try:
     from google.api_core.exceptions import BadRequest as GoogleBadRequest
+    from google.api_core.exceptions import NotFound as GoogleNotFound
 except ImportError:
-    GoogleBadRequest = None
+    GoogleBadRequest = GoogleNotFound = None
 
 try:
     from polars.exceptions import ColumnNotFoundError as PolarsColumnNotFoundError
diff --git a/ibis/backends/tests/test_client.py b/ibis/backends/tests/test_client.py
index ae446850e..8ae35e635 100644
--- a/ibis/backends/tests/test_client.py
+++ b/ibis/backends/tests/test_client.py
@@ -27,12 +27,16 @@ import ibis.expr.datatypes as dt
 import ibis.expr.operations as ops
 from ibis.backends.conftest import ALL_BACKENDS
 from ibis.backends.tests.errors import (
+    ClickHouseDatabaseError,
     ExaQueryError,
+    GoogleNotFound,
     ImpalaHiveServer2Error,
+    MySQLProgrammingError,
     OracleDatabaseError,
     PsycoPg2InternalError,
     PsycoPg2UndefinedObject,
     PyODBCProgrammingError,
+    PySparkAnalysisException,
     SnowflakeProgrammingError,
     TrinoUserError,
 )
@@ -1392,13 +1396,8 @@ def test_create_catalog(con_create_catalog):
 def test_create_database(con_create_database):
     database = gen_name("test_create_database")
     con_create_database.create_database(database)
-    if con_create_database.name == "exasol":
-        database = database.upper()
     assert database in con_create_database.list_databases()
-    database = database.lower()
     con_create_database.drop_database(database)
-    if con_create_database.name == "exasol":
-        database = database.upper()
     assert database not in con_create_database.list_databases()
 
 
@@ -1421,15 +1420,11 @@ def test_create_schema(con_create_database):
     with pytest.warns(FutureWarning):
         con_create_database.create_schema(schema)
     with pytest.warns(FutureWarning):
-        if con_create_database.name == "exasol":
-            schema = schema.upper()
         assert schema in con_create_database.list_schemas()
         schema = schema.lower()
     with pytest.warns(FutureWarning):
         con_create_database.drop_schema(schema)
     with pytest.warns(FutureWarning):
-        if con_create_database.name == "exasol":
-            schema = schema.upper()
         assert schema not in con_create_database.list_schemas()
 
 
@@ -1629,3 +1624,62 @@ def test_from_connection(con, top_level):
     new_con = backend.from_connection(getattr(con, CON_ATTR.get(con.name, "con")))
     result = int(new_con.execute(ibis.literal(1, type="int")))
     assert result == 1
+
+
+@pytest.mark.notimpl(
+    ["flink"], raises=com.IbisError, reason="not yet implemented for Flink"
+)
+def test_no_accidental_cross_database_table_load(con_create_database):
+    con = con_create_database
+
+    # Create an extra database
+    con.create_database(dbname := gen_name("dummy_db"))
+
+    # Create table with same name in current db and dummy db
+    con.create_table(
+        table := gen_name("table"), schema=(sch1 := ibis.schema({"a": "int"}))
+    )
+
+    con.create_table(table, schema=ibis.schema({"b": "string"}), database=dbname)
+
+    # Can grab table object from current db:
+    t = con.table(table)
+    assert t.schema().equals(sch1)
+
+    con.drop_table(table)
+
+    # NOTE: this entire block of exception type munging goes away once we unify
+    # table-not-found exceptions
+
+    # always allowed to raise
+    always_allowed = (com.IbisError,)
+
+    # these exception types are None when the backend dependency that
+    # defines them is not installed
+    allowed_when_installed = filter(
+        None,
+        (
+            ClickHouseDatabaseError,
+            PySparkAnalysisException,
+            MySQLProgrammingError,
+            ExaQueryError,
+            SnowflakeProgrammingError,
+            GoogleNotFound,
+        ),
+    )
+
+    # we only want to allow base Exception when we're testing datafusion
+    # otherwise any exceptions, including those that are unrelated to the
+    # problem under test will be considered correctly raising
+    datafusion_only = (Exception,) * (con.name == "datafusion")
+
+    # Now attempting to load same table name without specifying db should fail
+    with pytest.raises((*always_allowed, *allowed_when_installed, *datafusion_only)):
+        t = con.table(table)
+
+    # But can load if specify other db
+    t = con.table(table, database=dbname)
+
+    # Clean up
+    con.drop_table(table, database=dbname)
+    con.drop_database(dbname)