Fix MySQL autocomplete in multi-database setups

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
   name-builder inserted it unconditionally between backticks, producing
   a three-part identifier with an empty middle. Add a small
   `_qualify_table_name` helper that skips empty db/schema segments, and
   route the four inline builders (tables/views × sync/async loader)
   through it.

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 7 new unit tests pinning the name-builder across several
dialects and the stuck-Loading regression.

Closes #151.

Author: Maxteabag

sqlit/domains/query/ui/mixins/autocomplete_schema.py

@@ -12,6 +12,22 @@
 SCHEMA_PROCESS_BATCH_SIZE = 200
 
 
+def _qualify_table_name(dialect: Any, database: str | None, schema: str, name: str) -> str:
+    """Build a quoted qualified identifier, skipping empty db/schema segments.
+
+    MySQL/MariaDB have no schema-within-database, so `schema` comes back as
+    an empty string. Joining `db`.`` empty ``.`table` unconditionally produces
+    garbage like `db`.``.`table`; skip empties instead.
+    """
+    parts: list[str] = []
+    if database:
+        parts.append(dialect.quote_identifier(database))
+    if schema:
+        parts.append(dialect.quote_identifier(schema))
+    parts.append(dialect.quote_identifier(name))
+    return ".".join(parts)
+
+
 class AutocompleteSchemaMixin:
     """Mixin providing schema loading and caching for autocomplete."""
 
@@ -406,13 +422,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 = _qualify_table_name(dialect, database, schema_name, table_name)
 
                     display_name = dialect.format_table_name(schema_name, table_name)
                     metadata = [
@@ -495,13 +505,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 = _qualify_table_name(dialect, database, schema_name, view_name)
 
                     display_name = dialect.format_table_name(schema_name, view_name)
                     metadata = [
@@ -816,14 +820,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 = _qualify_table_name(dialect, 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 +841,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 = _qualify_table_name(dialect, 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)

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:

tests/unit/test_autocomplete_multidb.py

@@ -0,0 +1,130 @@
+"""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.
+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
+
+from unittest.mock import MagicMock
+
+from sqlit.domains.query.ui.mixins.autocomplete_schema import _qualify_table_name
+
+
+class _FakeDialect:
+    def quote_identifier(self, name: str) -> str:
+        escaped = name.replace("`", "``")
+        return f"`{escaped}`"
+
+
+def test_qualify_skips_empty_schema_segment() -> None:
+    """MySQL/MariaDB have no schemas within databases; the schema slot
+    arrives as ''. The qualified name must be `db`.`table`, not the
+    pre-fix `db`.``.`table`."""
+    result = _qualify_table_name(_FakeDialect(), "shop", "", "customers")
+    assert result == "`shop`.`customers`"
+
+
+def test_qualify_skips_empty_database_segment() -> None:
+    result = _qualify_table_name(_FakeDialect(), None, "public", "users")
+    assert result == "`public`.`users`"
+
+
+def test_qualify_full_three_part() -> None:
+    """Normal case: both db and schema present (e.g. SQL Server)."""
+    result = _qualify_table_name(_FakeDialect(), "app", "dbo", "Users")
+    assert result == "`app`.`dbo`.`Users`"
+
+
+def test_qualify_only_table_when_both_empty() -> None:
+    result = _qualify_table_name(_FakeDialect(), None, "", "customers")
+    assert result == "`customers`"
+
+
+def test_qualify_escapes_embedded_backticks() -> None:
+    result = _qualify_table_name(_FakeDialect(), "app`evil", "", "tbl")
+    # Embedded backticks are doubled by the dialect quoter.
+    assert result == "`app``evil`.`tbl`"
+
+
+class _SchemaHost:
+    """Minimal stand-in for the AutocompleteMixin protocol — 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)
+
+    # _get_autocomplete_suggestions depends on these helpers from the real
+    # mixin; stub them with minimal plausible behavior for these tests.
+    def _build_alias_map(self, text: str) -> dict:
+        return {}
+
+
+def test_unknown_table_ref_does_not_stick_on_loading() -> None:
+    """Issue #151: `SELECT * FROM shop.cu` parses `shop` as an ALIAS_COLUMN
+    scope, which isn't in table_metadata. 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"),
+            "shop.orders": ("", "orders", "shop"),
+            "products": ("", "products", "shop"),
+            "shop.products": ("", "products", "shop"),
+        },
+    )
+
+    # Bind the mixin method to the host so `self` resolves correctly.
+    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"
+    )
+    # Should not have triggered a load for 'shop' (not a real table).
+    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.
+    When the user references a real table whose columns aren't cached yet,
+    we still return Loading... and schedule the loader."""
+    from sqlit.domains.query.ui.mixins.autocomplete_suggestions import AutocompleteSuggestionsMixin
+
+    host = _SchemaHost(
+        tables=["customers"],
+        metadata={"customers": ("", "customers", None)},
+        columns={},  # not loaded yet
+    )
+    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