From a3a28dc0564ea4f01d622183d50e8ebe6e983fa3 Mon Sep 17 00:00:00 2001 From: Peter Adams <18162810+Maxteabag@users.noreply.github.com> Date: Sun, 19 Apr 2026 14:15:00 +0200 Subject: [PATCH] Fix MySQL autocomplete in multi-database setups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two bugs surfaced by driving the TUI's completion engine against a real MySQL server with multiple user databases: 1. Qualified table/view names rendered as `db`.``.`table`. MySQL has no schema-within-database, so the schema slot comes back as ''; the inline name-builder inserted it unconditionally, producing a three-part identifier with an empty middle. Promote the composition rule to a new `Dialect.qualified_name(database, schema, name)` method. Base default skips empty segments, which yields the correct result for every current adapter: MySQL/MariaDB: `db`.`table` PostgreSQL: "schema"."table" SQL Server: [db].[schema].[table] SQLite: "table" Autocomplete's four inline builders (tables/views × sync/async loader) now call `dialect.qualified_name(...)` directly. 2. Completions stuck on "Loading..." for cursor positions like `SELECT * FROM shop.cu`. The completion engine classifies this as an ALIAS_COLUMN with scope 'shop', which isn't a real table. The loader skips unknown keys silently; the caller kept returning Loading... forever because the guard assumed "not yet loaded" meant "in flight". Guard the sentinel on the table key actually existing in `_table_metadata`. Covered by 8 new unit tests pinning `qualified_name` on MySQL, MariaDB, PostgreSQL, SQL Server, empty-segment behavior, quote-char escaping, and the stuck-Loading regression. May fix #151. --- .../connections/providers/adapters/base.py | 16 ++ sqlit/domains/connections/providers/model.py | 12 ++ .../query/ui/mixins/autocomplete_schema.py | 36 +--- .../ui/mixins/autocomplete_suggestions.py | 16 +- tests/unit/test_autocomplete_multidb.py | 155 ++++++++++++++++++ 5 files changed, 203 insertions(+), 32 deletions(-) create mode 100644 tests/unit/test_autocomplete_multidb.py diff --git a/sqlit/domains/connections/providers/adapters/base.py b/sqlit/domains/connections/providers/adapters/base.py index 696c089e..109ea1da 100644 --- a/sqlit/domains/connections/providers/adapters/base.py +++ b/sqlit/domains/connections/providers/adapters/base.py @@ -417,6 +417,22 @@ def quote_identifier(self, name: str) -> str: """Quote an identifier (table name, column name, etc.).""" pass + def qualified_name(self, database: str | None, schema: str | None, name: str) -> str: + """Build a quoted qualified identifier, skipping empty segments. + + Default handles SQL Server-style `[db].[schema].[name]`, PostgreSQL- + style `"schema"."name"`, and single-part `"name"` by omitting any + empty/None component. Dialects that want different composition + (e.g. MySQL, which has no schemas within databases) can override. + """ + parts: list[str] = [] + if database: + parts.append(self.quote_identifier(database)) + if schema: + parts.append(self.quote_identifier(schema)) + parts.append(self.quote_identifier(name)) + return ".".join(parts) + @abstractmethod def build_select_query(self, table: str, limit: int, database: str | None = None, schema: str | None = None) -> str: """Build a SELECT query with limit. diff --git a/sqlit/domains/connections/providers/model.py b/sqlit/domains/connections/providers/model.py index 93b60b09..b104db3c 100644 --- a/sqlit/domains/connections/providers/model.py +++ b/sqlit/domains/connections/providers/model.py @@ -62,6 +62,18 @@ def build_select_query(self, table: str, limit: int, database: str | None = None def format_table_name(self, schema: str | None, table: str) -> str: ... + def qualified_name(self, database: str | None, schema: str | None, name: str) -> str: + """Build a quoted qualified identifier for use in SQL. + + Called when completing tables/views across multiple databases. Each + dialect decides how many segments to emit: + - SQL Server / generic: `[db].[schema].[name]` + - PostgreSQL: `"schema"."name"` (databases are isolated) + - MySQL/MariaDB: `` `db`.`name` `` (no schemas within databases) + - SQLite: `"name"` + """ + ... + @runtime_checkable class SchemaInspector(Protocol): diff --git a/sqlit/domains/query/ui/mixins/autocomplete_schema.py b/sqlit/domains/query/ui/mixins/autocomplete_schema.py index a0e547dd..bc8b7f4b 100644 --- a/sqlit/domains/query/ui/mixins/autocomplete_schema.py +++ b/sqlit/domains/query/ui/mixins/autocomplete_schema.py @@ -406,13 +406,7 @@ def work() -> None: if single_db: full_name = table_name else: - quoted_db = dialect.quote_identifier(database) if database else "" - quoted_schema = dialect.quote_identifier(schema_name) - quoted_table = dialect.quote_identifier(table_name) - if database: - full_name = f"{quoted_db}.{quoted_schema}.{quoted_table}" - else: - full_name = f"{quoted_schema}.{quoted_table}" + full_name = dialect.qualified_name(database, schema_name, table_name) display_name = dialect.format_table_name(schema_name, table_name) metadata = [ @@ -495,13 +489,7 @@ def work() -> None: if single_db: full_name = view_name else: - quoted_db = dialect.quote_identifier(database) if database else "" - quoted_schema = dialect.quote_identifier(schema_name) - quoted_view = dialect.quote_identifier(view_name) - if database: - full_name = f"{quoted_db}.{quoted_schema}.{quoted_view}" - else: - full_name = f"{quoted_schema}.{quoted_view}" + full_name = dialect.qualified_name(database, schema_name, view_name) display_name = dialect.format_table_name(schema_name, view_name) metadata = [ @@ -816,14 +804,8 @@ async def run_db_call(fn: Any, *args: Any, **kwargs: Any) -> Any: # Single database - use simple table name schema_cache["tables"].append(table_name) else: - # Multiple databases - use full qualifier [db].[schema].[table] - quoted_db = dialect.quote_identifier(database) if database else "" - quoted_schema = dialect.quote_identifier(schema_name) - quoted_table = dialect.quote_identifier(table_name) - if database: - full_name = f"{quoted_db}.{quoted_schema}.{quoted_table}" - else: - full_name = f"{quoted_schema}.{quoted_table}" + # Multiple databases - use qualified identifier + full_name = dialect.qualified_name(database, schema_name, table_name) schema_cache["tables"].append(full_name) # Keep metadata for column loading (multiple keys for flexible lookup) display_name = dialect.format_table_name(schema_name, table_name) @@ -843,14 +825,8 @@ async def run_db_call(fn: Any, *args: Any, **kwargs: Any) -> Any: # Single database - use simple view name schema_cache["views"].append(view_name) else: - # Multiple databases - use full qualifier [db].[schema].[view] - quoted_db = dialect.quote_identifier(database) if database else "" - quoted_schema = dialect.quote_identifier(schema_name) - quoted_view = dialect.quote_identifier(view_name) - if database: - full_name = f"{quoted_db}.{quoted_schema}.{quoted_view}" - else: - full_name = f"{quoted_schema}.{quoted_view}" + # Multiple databases - use qualified identifier + full_name = dialect.qualified_name(database, schema_name, view_name) schema_cache["views"].append(full_name) # Keep metadata for column loading (multiple keys for flexible lookup) display_name = dialect.format_table_name(schema_name, view_name) diff --git a/sqlit/domains/query/ui/mixins/autocomplete_suggestions.py b/sqlit/domains/query/ui/mixins/autocomplete_suggestions.py index 1cb1d58b..94ef4e7e 100644 --- a/sqlit/domains/query/ui/mixins/autocomplete_suggestions.py +++ b/sqlit/domains/query/ui/mixins/autocomplete_suggestions.py @@ -74,13 +74,25 @@ def _get_autocomplete_suggestions(self: AutocompleteMixinHost, text: str, cursor alias_map = self._build_alias_map(text) table_refs = extract_table_refs(text) loading: set[str] = getattr(self, "_columns_loading", set()) + table_metadata = getattr(self, "_table_metadata", {}) or {} + + def needs_column_load(key: str) -> bool: + """Return True only if the key names a known table that hasn't + been loaded yet. Returning Loading... for an unknown key would + wedge forever because the loader skips unknown tables silently. + """ + if key in columns: + return False + if key not in table_metadata: + return False + return True for suggestion in suggestions: if suggestion.type == SuggestionType.COLUMN: # Check if any tables need column loading for ref in table_refs: table_key = ref.name.lower() - if table_key not in columns and table_key not in loading: + if needs_column_load(table_key) and table_key not in loading: self._load_columns_for_table(table_key) return ["Loading..."] elif table_key in loading: @@ -92,7 +104,7 @@ def _get_autocomplete_suggestions(self: AutocompleteMixinHost, text: str, cursor scope_lower = scope.lower() table_key = alias_map.get(scope_lower, scope_lower) - if table_key not in columns and table_key not in loading: + if needs_column_load(table_key) and table_key not in loading: self._load_columns_for_table(table_key) return ["Loading..."] elif table_key in loading: diff --git a/tests/unit/test_autocomplete_multidb.py b/tests/unit/test_autocomplete_multidb.py new file mode 100644 index 00000000..e34946cb --- /dev/null +++ b/tests/unit/test_autocomplete_multidb.py @@ -0,0 +1,155 @@ +"""Regression tests for MySQL autocomplete in multi-database scenarios (#151). + +Two narrow bugs are covered: + +1. Qualified identifiers for databases without schemas (MySQL/MariaDB) used + to render as `db`.``.`table` — an empty-backticked middle segment. The + qualifying logic is now a Dialect method so each adapter owns its own + composition rule. + +2. Autocomplete returned a permanent "Loading..." sentinel whenever the + table reference in the query didn't resolve to anything in the schema + cache (e.g. the user typed `SELECT * FROM shop.cu`: `shop` was treated + as an alias and the loader spun forever for an unknown key). +""" + +from __future__ import annotations + +import pytest + + +# -------------------------------------------------------------------------- +# Bug 1: Dialect.qualified_name +# -------------------------------------------------------------------------- + + +def _get_dialect(db_type: str): + from sqlit.domains.connections.providers.catalog import get_provider + + return get_provider(db_type).dialect + + +def test_mysql_qualified_name_is_two_part() -> None: + """MySQL has no schema-within-database; the qualified form is + `db`.`table`, NOT `db`.``.`table`.""" + dialect = _get_dialect("mysql") + assert dialect.qualified_name("shop", "", "customers") == "`shop`.`customers`" + + +def test_mariadb_qualified_name_is_two_part() -> None: + dialect = _get_dialect("mariadb") + assert dialect.qualified_name("shop", "", "customers") == "`shop`.`customers`" + + +def test_postgresql_qualified_name_uses_schema_only() -> None: + """PostgreSQL databases are isolated; only schema.table makes sense + for cross-reference within the connected database.""" + dialect = _get_dialect("postgresql") + # No db segment expected when schema is present. + assert dialect.qualified_name(None, "public", "users") == '"public"."users"' + + +def test_sqlserver_qualified_name_is_three_part() -> None: + """SQL Server explicitly uses [db].[schema].[table].""" + dialect = _get_dialect("mssql") + assert dialect.qualified_name("app", "dbo", "Users") == "[app].[dbo].[Users]" + + +def test_qualified_name_skips_empty_segments_everywhere() -> None: + """Regardless of dialect, empty db/schema segments must be omitted, + never rendered as empty-quoted placeholders.""" + for db_type in ("mysql", "postgresql", "mssql", "sqlite"): + dialect = _get_dialect(db_type) + # bare table name: single quoted segment, no dot joins. + bare = dialect.qualified_name(None, None, "t") + assert "t" in bare, f"{db_type}: {bare}" + assert "." not in bare, f"{db_type} unexpected joins: {bare}" + # empty schema segment must not produce empty quote pair like `` or "" or []. + out = dialect.qualified_name(None, "", "t") + for marker in ("``", '""', "[]"): + assert marker not in out, f"{db_type} emitted empty-quoted segment: {out}" + + +def test_qualified_name_escapes_embedded_quote_chars() -> None: + """Each dialect must escape its own quote char, not leak raw input.""" + for db_type, payload, expected_substr in [ + ("mysql", "app`evil", "app``evil"), + ("postgresql", 'app"evil', 'app""evil'), + ("mssql", "app]evil", "app]]evil"), + ]: + dialect = _get_dialect(db_type) + out = dialect.qualified_name(None, None, payload) + assert expected_substr in out, f"{db_type}: {out}" + + +# -------------------------------------------------------------------------- +# Bug 2: stuck Loading... for unknown table references +# -------------------------------------------------------------------------- + + +class _SchemaHost: + """Minimal stand-in for AutocompleteMixinHost — just enough to drive + `_get_autocomplete_suggestions`.""" + + def __init__(self, tables, metadata, columns=None, loading=None): + self._schema_cache = { + "tables": tables, + "views": [], + "columns": columns or {}, + "procedures": [], + } + self._table_metadata = metadata + self._columns_loading = loading or set() + self.load_calls: list[str] = [] + + def _load_columns_for_table(self, table_name: str) -> None: + self.load_calls.append(table_name) + + def _build_alias_map(self, text: str) -> dict: + return {} + + +def test_unknown_table_ref_does_not_stick_on_loading() -> None: + """`SELECT * FROM shop.cu` parses `shop` as an ALIAS_COLUMN scope. It + isn't a real table. Before the fix the completion engine called + _load_columns_for_table('shop') and returned `Loading...`; the loader + skipped unknown keys silently, so the sentinel never cleared.""" + from sqlit.domains.query.ui.mixins.autocomplete_suggestions import AutocompleteSuggestionsMixin + + host = _SchemaHost( + tables=["customers", "orders", "products"], + metadata={ + "customers": ("", "customers", "shop"), + "shop.customers": ("", "customers", "shop"), + "orders": ("", "orders", "shop"), + "products": ("", "products", "shop"), + }, + ) + + get_suggestions = AutocompleteSuggestionsMixin._get_autocomplete_suggestions.__get__(host) + + text = "SELECT * FROM shop.cu" + result = get_suggestions(text, len(text)) + + assert result != ["Loading..."], ( + "unknown table key must not pin Loading... — loader never clears it" + ) + assert "shop" not in host.load_calls + + +def test_known_table_ref_still_triggers_loading_on_first_call() -> None: + """Sanity check: the fix must not regress legit lazy loading.""" + from sqlit.domains.query.ui.mixins.autocomplete_suggestions import AutocompleteSuggestionsMixin + + host = _SchemaHost( + tables=["customers"], + metadata={"customers": ("", "customers", None)}, + columns={}, + ) + get_suggestions = AutocompleteSuggestionsMixin._get_autocomplete_suggestions.__get__(host) + + text = "SELECT * FROM customers WHERE em" + result = get_suggestions(text, len(text)) + + assert result == ["Loading..."] + assert "customers" in host.load_calls