fix: PostgreSQL adapter bugs for multi-backend support

Fixes multiple issues discovered during notebook execution against PostgreSQL:

1. Column comment retrieval (blob codec association)
   - PostgreSQL stores comments separately via COMMENT ON COLUMN
   - Updated get_columns_sql to use col_description() function
   - Updated parse_column_info to extract column_comment
   - This enables proper codec association (e.g., <blob>) from comments

2. ENUM type DDL generation
   - PostgreSQL requires CREATE TYPE for enums (not inline like MySQL)
   - Added enum type name generation based on value hash
   - Added get_pending_enum_ddl() method for pre-CREATE TABLE DDL
   - Updated declare() to return pre_ddl and post_ddl separately

3. Upsert/skip_duplicates syntax
   - MySQL: ON DUPLICATE KEY UPDATE pk=table.pk
   - PostgreSQL: ON CONFLICT (pk_cols) DO NOTHING
   - Added skip_duplicates_clause() method to both adapters
   - Updated table.py to use adapter method

4. String quoting in information_schema queries
   - Dependencies.py had hardcoded MySQL double quotes and concat()
   - Made queries backend-agnostic using adapter.quote_string()
   - Added backend property to adapters for conditional SQL generation

5. Index DDL syntax
   - MySQL: inline INDEX in CREATE TABLE
   - PostgreSQL: separate CREATE INDEX statements
   - Added supports_inline_indexes property
   - Added create_index_ddl() method to base adapter
   - Updated declare() to generate CREATE INDEX for PostgreSQL post_ddl

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: dimitri-yatsenko

src/datajoint/adapters/base.py

@@ -114,6 +114,19 @@ def default_port(self) -> int:
         """
         ...
 
+    @property
+    @abstractmethod
+    def backend(self) -> str:
+        """
+        Backend identifier string.
+
+        Returns
+        -------
+        str
+            Backend name: 'mysql' or 'postgresql'.
+        """
+        ...
+
     @abstractmethod
     def get_cursor(self, connection: Any, as_dict: bool = False) -> Any:
         """
@@ -491,6 +504,83 @@ def upsert_on_duplicate_sql(
         """
         ...
 
+    @abstractmethod
+    def skip_duplicates_clause(
+        self,
+        full_table_name: str,
+        primary_key: list[str],
+    ) -> str:
+        """
+        Generate clause to skip duplicate key insertions.
+
+        For MySQL: ON DUPLICATE KEY UPDATE pk=table.pk (no-op update)
+        For PostgreSQL: ON CONFLICT (pk_cols) DO NOTHING
+
+        Parameters
+        ----------
+        full_table_name : str
+            Fully qualified table name (with quotes).
+        primary_key : list[str]
+            Primary key column names (unquoted).
+
+        Returns
+        -------
+        str
+            SQL clause to append to INSERT statement.
+        """
+        ...
+
+    @property
+    def supports_inline_indexes(self) -> bool:
+        """
+        Whether this backend supports inline INDEX in CREATE TABLE.
+
+        MySQL supports inline index definitions in CREATE TABLE.
+        PostgreSQL requires separate CREATE INDEX statements.
+
+        Returns
+        -------
+        bool
+            True for MySQL, False for PostgreSQL.
+        """
+        return True  # Default for MySQL, override in PostgreSQL
+
+    def create_index_ddl(
+        self,
+        full_table_name: str,
+        columns: list[str],
+        unique: bool = False,
+        index_name: str | None = None,
+    ) -> str:
+        """
+        Generate CREATE INDEX statement.
+
+        Parameters
+        ----------
+        full_table_name : str
+            Fully qualified table name (with quotes).
+        columns : list[str]
+            Column names to index (unquoted).
+        unique : bool, optional
+            If True, create a unique index.
+        index_name : str, optional
+            Custom index name. If None, auto-generate from table/columns.
+
+        Returns
+        -------
+        str
+            CREATE INDEX SQL statement.
+        """
+        quoted_cols = ", ".join(self.quote_identifier(col) for col in columns)
+        # Generate index name from table and columns if not provided
+        if index_name is None:
+            # Extract table name from full_table_name for index naming
+            table_part = full_table_name.split(".")[-1].strip('`"')
+            col_part = "_".join(columns)[:30]  # Truncate for long column lists
+            index_name = f"idx_{table_part}_{col_part}"
+        unique_clause = "UNIQUE " if unique else ""
+        return f"CREATE {unique_clause}INDEX {self.quote_identifier(index_name)} ON {full_table_name} ({quoted_cols})"
+
     # =========================================================================
     # Introspection
     # =========================================================================

src/datajoint/adapters/mysql.py

@@ -155,6 +155,11 @@ def default_port(self) -> int:
         """MySQL default port 3306."""
         return 3306
 
+    @property
+    def backend(self) -> str:
+        """Backend identifier: 'mysql'."""
+        return "mysql"
+
     def get_cursor(self, connection: Any, as_dict: bool = False) -> Any:
         """
         Get a cursor from MySQL connection.
@@ -567,6 +572,32 @@ def upsert_on_duplicate_sql(
         ON DUPLICATE KEY UPDATE {update_clauses}
         """
 
+    def skip_duplicates_clause(
+        self,
+        full_table_name: str,
+        primary_key: list[str],
+    ) -> str:
+        """
+        Generate clause to skip duplicate key insertions for MySQL.
+
+        Uses ON DUPLICATE KEY UPDATE with a no-op update (pk=pk) to effectively
+        skip duplicates without raising an error.
+
+        Parameters
+        ----------
+        full_table_name : str
+            Fully qualified table name (with quotes).
+        primary_key : list[str]
+            Primary key column names (unquoted).
+
+        Returns
+        -------
+        str
+            MySQL ON DUPLICATE KEY UPDATE clause.
+        """
+        quoted_pk = self.quote_identifier(primary_key[0])
+        return f" ON DUPLICATE KEY UPDATE {quoted_pk}={full_table_name}.{quoted_pk}"
+
     # =========================================================================
     # Introspection
     # =========================================================================

src/datajoint/adapters/postgres.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Any
 
 try:
@@ -170,6 +171,11 @@ def default_port(self) -> int:
         """PostgreSQL default port 5432."""
         return 5432
 
+    @property
+    def backend(self) -> str:
+        """Backend identifier: 'postgresql'."""
+        return "postgresql"
+
     def get_cursor(self, connection: Any, as_dict: bool = False) -> Any:
         """
         Get a cursor from PostgreSQL connection.
@@ -292,9 +298,27 @@ def core_type_to_sql(self, core_type: str) -> str:
             return f"numeric{params}"
 
         if core_type.startswith("enum("):
-            # Enum requires special handling - caller must use CREATE TYPE
-            # Return the type name pattern (will be replaced by caller)
-            return "{{enum_type_name}}"  # Placeholder for CREATE TYPE
+            # PostgreSQL requires CREATE TYPE for enums
+            # Extract enum values and generate a deterministic type name
+            enum_match = re.match(r"enum\s*\((.+)\)", core_type, re.I)
+            if enum_match:
+                # Parse enum values: enum('M','F') -> ['M', 'F']
+                values_str = enum_match.group(1)
+                # Split by comma, handling quoted values
+                values = [v.strip().strip("'\"") for v in values_str.split(",")]
+                # Generate a deterministic type name based on values
+                # Use a hash to keep name reasonable length
+                import hashlib
+                value_hash = hashlib.md5("_".join(sorted(values)).encode()).hexdigest()[:8]
+                type_name = f"enum_{value_hash}"
+                # Track this enum type for CREATE TYPE DDL
+                if not hasattr(self, "_pending_enum_types"):
+                    self._pending_enum_types = {}
+                self._pending_enum_types[type_name] = values
+                # Return schema-qualified type reference using placeholder
+                # {database} will be replaced with actual schema name in table.py
+                return '"{database}".' + self.quote_identifier(type_name)
+            return "text"  # Fallback if parsing fails
 
         raise ValueError(f"Unknown core type: {core_type}")
 
@@ -611,6 +635,43 @@ def upsert_on_duplicate_sql(
         ON CONFLICT ({conflict_cols}) DO UPDATE SET {update_clauses}
         """
 
+    def skip_duplicates_clause(
+        self,
+        full_table_name: str,
+        primary_key: list[str],
+    ) -> str:
+        """
+        Generate clause to skip duplicate key insertions for PostgreSQL.
+
+        Uses ON CONFLICT (pk_cols) DO NOTHING to skip duplicates without
+        raising an error.
+
+        Parameters
+        ----------
+        full_table_name : str
+            Fully qualified table name (with quotes). Unused but kept for
+            API compatibility with MySQL adapter.
+        primary_key : list[str]
+            Primary key column names (unquoted).
+
+        Returns
+        -------
+        str
+            PostgreSQL ON CONFLICT DO NOTHING clause.
+        """
+        pk_cols = ", ".join(self.quote_identifier(pk) for pk in primary_key)
+        return f" ON CONFLICT ({pk_cols}) DO NOTHING"
+
+    @property
+    def supports_inline_indexes(self) -> bool:
+        """
+        PostgreSQL does not support inline INDEX in CREATE TABLE.
+
+        Returns False to indicate indexes must be created separately
+        with CREATE INDEX statements.
+        """
+        return False
+
     # =========================================================================
     # Introspection
     # =========================================================================
@@ -639,14 +700,17 @@ def get_table_info_sql(self, schema_name: str, table_name: str) -> str:
         )
 
     def get_columns_sql(self, schema_name: str, table_name: str) -> str:
-        """Query to get column definitions."""
+        """Query to get column definitions including comments."""
+        # Use col_description() to retrieve column comments stored via COMMENT ON COLUMN
+        # The regclass cast allows using schema.table notation to get the OID
         return (
-            f"SELECT column_name, data_type, is_nullable, column_default, "
-            f"character_maximum_length, numeric_precision, numeric_scale "
-            f"FROM information_schema.columns "
-            f"WHERE table_schema = {self.quote_string(schema_name)} "
-            f"AND table_name = {self.quote_string(table_name)} "
-            f"ORDER BY ordinal_position"
+            f"SELECT c.column_name, c.data_type, c.is_nullable, c.column_default, "
+            f"c.character_maximum_length, c.numeric_precision, c.numeric_scale, "
+            f"col_description(({self.quote_string(schema_name)} || '.' || {self.quote_string(table_name)})::regclass, c.ordinal_position) as column_comment "
+            f"FROM information_schema.columns c "
+            f"WHERE c.table_schema = {self.quote_string(schema_name)} "
+            f"AND c.table_name = {self.quote_string(table_name)} "
+            f"ORDER BY c.ordinal_position"
         )
 
     def get_primary_key_sql(self, schema_name: str, table_name: str) -> str:
@@ -761,7 +825,7 @@ def parse_column_info(self, row: dict[str, Any]) -> dict[str, Any]:
         Parameters
         ----------
         row : dict
-            Row from information_schema.columns query.
+            Row from information_schema.columns query with col_description() join.
 
         Returns
         -------
@@ -774,7 +838,7 @@ def parse_column_info(self, row: dict[str, Any]) -> dict[str, Any]:
             "type": row["data_type"],
             "nullable": row["is_nullable"] == "YES",
             "default": row["column_default"],
-            "comment": None,  # PostgreSQL stores comments separately
+            "comment": row.get("column_comment"),  # Retrieved via col_description()
             "key": "",  # PostgreSQL key info retrieved separately
             "extra": "",  # PostgreSQL doesn't have auto_increment in same way
         }
@@ -947,6 +1011,34 @@ def enum_type_ddl(self, type_name: str, values: list[str]) -> str | None:
         quoted_values = ", ".join(f"'{v}'" for v in values)
         return f"CREATE TYPE {self.quote_identifier(type_name)} AS ENUM ({quoted_values})"
 
+    def get_pending_enum_ddl(self, schema_name: str) -> list[str]:
+        """
+        Get DDL statements for pending enum types and clear the pending list.
+
+        PostgreSQL requires CREATE TYPE statements before using enum types in
+        column definitions. This method returns DDL for enum types accumulated
+        during type conversion and clears the pending list.
+
+        Parameters
+        ----------
+        schema_name : str
+            Schema name to qualify enum type names.
+
+        Returns
+        -------
+        list[str]
+            List of CREATE TYPE statements (if any pending).
+        """
+        ddl_statements = []
+        if hasattr(self, "_pending_enum_types") and self._pending_enum_types:
+            for type_name, values in self._pending_enum_types.items():
+                # Generate CREATE TYPE with schema qualification
+                quoted_type = f"{self.quote_identifier(schema_name)}.{self.quote_identifier(type_name)}"
+                quoted_values = ", ".join(f"'{v}'" for v in values)
+                ddl_statements.append(f"CREATE TYPE {quoted_type} AS ENUM ({quoted_values})")
+            self._pending_enum_types = {}
+        return ddl_statements
+
     def job_metadata_columns(self) -> list[str]:
         """
         Return PostgreSQL-specific job metadata column definitions.