fix+refactor: backend-agnostic quoting, table name construction, and adapter abstractions

src/datajoint/adapters/base.py

@@ -238,6 +238,39 @@ def parameter_placeholder(self) -> str:
         """
         ...
 
+    def make_full_table_name(self, database: str, table_name: str) -> str:
+        """
+        Construct a fully-qualified table name for this backend.
+
+        Default implementation produces a two-part name (``schema.table``).
+        Backends that require additional namespace levels can override.
+
+        Parameters
+        ----------
+        database : str
+            Schema/database name.
+        table_name : str
+            Table name (including tier prefix).
+
+        Returns
+        -------
+        str
+            Fully-qualified, quoted table name.
+        """
+        return f"{self.quote_identifier(database)}.{self.quote_identifier(table_name)}"
+
+    @property
+    def max_table_name_length(self) -> int:
+        """
+        Maximum length of a table name for this backend.
+
+        Returns
+        -------
+        int
+            Maximum allowed characters in a table identifier.
+        """
+        return 64  # safe default (MySQL limit)
+
     # =========================================================================
     # Type Mapping
     # =========================================================================

src/datajoint/adapters/postgres.py

@@ -249,6 +249,11 @@ def quote_identifier(self, name: str) -> str:
         """
         return f'"{name}"'
 
+    @property
+    def max_table_name_length(self) -> int:
+        """PostgreSQL NAMEDATALEN-1 = 63."""
+        return 63
+
     def split_full_table_name(self, full_table_name: str) -> tuple[str, str]:
         """Split ``"schema"."table"`` into ``('schema', 'table')``."""
         schema, table = full_table_name.replace('"', "").split(".")

src/datajoint/autopopulate.py

@@ -776,9 +776,10 @@ def _update_job_metadata(self, key, start_time, duration, version):
         from .condition import make_condition
 
         pk_condition = make_condition(self, key, set())
+        q = self.connection.adapter.quote_identifier
         self.connection.query(
             f"UPDATE {self.full_table_name} SET "
-            "`_job_start_time`=%s, `_job_duration`=%s, `_job_version`=%s "
+            f"{q('_job_start_time')}=%s, {q('_job_duration')}=%s, {q('_job_version')}=%s "
             f"WHERE {pk_condition}",
             args=(start_time, duration, version[:64] if version else ""),
         )

src/datajoint/connection.py

@@ -486,19 +486,25 @@ def start_transaction(self) -> None:
         """
         if self.in_transaction:
             raise errors.DataJointError("Nested connections are not supported.")
-        self.query(self.adapter.start_transaction_sql())
+        sql = self.adapter.start_transaction_sql()
+        if sql:
+            self.query(sql)
         self._in_transaction = True
         logger.debug("Transaction started")
 
     def cancel_transaction(self) -> None:
         """Cancel the current transaction and roll back all changes."""
-        self.query(self.adapter.rollback_sql())
+        sql = self.adapter.rollback_sql()
+        if sql:
+            self.query(sql)
         self._in_transaction = False
         logger.debug("Transaction cancelled. Rolling back ...")
 
     def commit_transaction(self) -> None:
         """Commit all changes and close the transaction."""
-        self.query(self.adapter.commit_sql())
+        sql = self.adapter.commit_sql()
+        if sql:
+            self.query(sql)
         self._in_transaction = False
         logger.debug("Transaction committed and closed.")
 

src/datajoint/declare.py

@@ -52,7 +52,6 @@
 # Get SQL mapping for core types
 CORE_TYPE_SQL = {name: sql_type for name, (_, sql_type) in CORE_TYPES.items()}
 
-MAX_TABLE_NAME_LENGTH = 64
 CONSTANT_LITERALS = {
     "CURRENT_TIMESTAMP",
     "NULL",
@@ -296,7 +295,7 @@ def compile_foreign_key(
     parent_full_name = ref.support[0]
     # Parse as database.table using the adapter's quoting convention
     parts = adapter.split_full_table_name(parent_full_name)
-    ref_table_name = f"{adapter.quote_identifier(parts[0])}.{adapter.quote_identifier(parts[1])}"
+    ref_table_name = adapter.make_full_table_name(parts[0], parts[1])
 
     foreign_key_sql.append(
         f"FOREIGN KEY ({fk_cols}) REFERENCES {ref_table_name} ({pk_cols}) ON UPDATE CASCADE ON DELETE RESTRICT"
@@ -432,21 +431,13 @@ def declare(
     DataJointError
         If table name exceeds max length or has no primary key.
     """
-    # Parse table name without assuming quote character
-    # Extract schema.table from quoted name using adapter
-    quote_char = adapter.quote_identifier("x")[0]  # Get quote char from adapter
-    parts = full_table_name.split(".")
-    if len(parts) == 2:
-        schema_name = parts[0].strip(quote_char)
-        table_name = parts[1].strip(quote_char)
-    else:
-        schema_name = None
-        table_name = parts[0].strip(quote_char)
+    # Parse table name using adapter (handles backend-specific quoting)
+    schema_name, table_name = adapter.split_full_table_name(full_table_name)
 
-    if len(table_name) > MAX_TABLE_NAME_LENGTH:
+    if len(table_name) > adapter.max_table_name_length:
         raise DataJointError(
             "Table name `{name}` exceeds the max length of {max_length}".format(
-                name=table_name, max_length=MAX_TABLE_NAME_LENGTH
+                name=table_name, max_length=adapter.max_table_name_length
             )
         )
 

src/datajoint/lineage.py

@@ -79,14 +79,12 @@ def lineage_table_exists(connection, database):
     bool
         True if the table exists, False otherwise.
     """
-    result = connection.query(
-        """
-        SELECT COUNT(*) FROM information_schema.tables
-        WHERE table_schema = %s AND table_name = '~lineage'
-        """,
-        args=(database,),
-    ).fetchone()
-    return result[0] > 0
+    try:
+        result = connection.query(connection.adapter.get_table_info_sql(database, "~lineage")).fetchone()
+        return result is not None
+    except Exception:
+        # Schema or catalog query may fail on some backends
+        return False
 
 
 def get_lineage(connection, database, table_name, attribute_name):

src/datajoint/schemas.py

@@ -341,10 +341,11 @@ def make_classes(self, into: dict[str, Any] | None = None) -> None:
                 frame = inspect.currentframe().f_back
                 into = frame.f_locals
                 del frame
+        adapter = self.connection.adapter
         tables = [
             row[0]
-            for row in self.connection.query(self.connection.adapter.list_tables_sql(self.database))
-            if lookup_class_name("`{db}`.`{tab}`".format(db=self.database, tab=row[0]), into, 0) is None
+            for row in self.connection.query(adapter.list_tables_sql(self.database))
+            if lookup_class_name(adapter.make_full_table_name(self.database, row[0]), into, 0) is None
         ]
         master_classes = (Lookup, Manual, Imported, Computed)
         part_tables = []
@@ -502,7 +503,7 @@ def jobs(self) -> list[Job]:
         # Iterate over auto-populated tables and check if their job table exists
         for table_name in self.list_tables():
             adapter = self.connection.adapter
-            full_name = f"{adapter.quote_identifier(self.database)}." f"{adapter.quote_identifier(table_name)}"
+            full_name = adapter.make_full_table_name(self.database, table_name)
             table = FreeTable(self.connection, full_name)
             tier = _get_tier(table.full_table_name)
             if tier in (Computed, Imported):
@@ -602,8 +603,7 @@ def get_table(self, name: str) -> FreeTable:
         if table_name is None:
             raise DataJointError(f"Table `{name}` does not exist in schema `{self.database}`.")
 
-        adapter = self.connection.adapter
-        full_name = f"{adapter.quote_identifier(self.database)}.{adapter.quote_identifier(table_name)}"
+        full_name = self.connection.adapter.make_full_table_name(self.database, table_name)
         return FreeTable(self.connection, full_name)
 
     def __getitem__(self, name: str) -> FreeTable:

src/datajoint/table.py

@@ -474,7 +474,7 @@ def full_table_name(self):
                 f"Class {self.__class__.__name__} is not associated with a schema. "
                 "Apply a schema decorator or use schema() to bind it."
             )
-        return f"{self.adapter.quote_identifier(self.database)}.{self.adapter.quote_identifier(self.table_name)}"
+        return self.adapter.make_full_table_name(self.database, self.table_name)
 
     @property
     def adapter(self):

src/datajoint/user_tables.py

@@ -106,8 +106,7 @@ def full_table_name(cls):
         """The fully qualified table name (quoted per backend)."""
         if cls.database is None:
             return None
-        adapter = cls._connection.adapter
-        return f"{adapter.quote_identifier(cls.database)}.{adapter.quote_identifier(cls.table_name)}"
+        return cls._connection.adapter.make_full_table_name(cls.database, cls.table_name)
 
 
 class UserTable(Table, metaclass=TableMeta):
@@ -186,8 +185,7 @@ def full_table_name(cls):
         """The fully qualified table name (quoted per backend)."""
         if cls.database is None or cls.table_name is None:
             return None
-        adapter = cls._connection.adapter
-        return f"{adapter.quote_identifier(cls.database)}.{adapter.quote_identifier(cls.table_name)}"
+        return cls._connection.adapter.make_full_table_name(cls.database, cls.table_name)
 
     @property
     def master(cls):