feat: add backend parameter to Instance and cross-connection validation

Instance now accepts backend="mysql"|"postgresql" to explicitly set the
database backend, with automatic port default derivation (3306 vs 5432).

Join, restriction, and union operators now validate that both operands
use the same connection, raising DataJointError with a clear message
when expressions from different Instances are combined.

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

Author: dimitri-yatsenko

src/datajoint/condition.py

@@ -244,6 +244,13 @@ def assert_join_compatibility(
     if isinstance(expr1, U) or isinstance(expr2, U):
         return
 
+    # Check that both expressions use the same connection
+    if expr1.connection is not expr2.connection:
+        raise DataJointError(
+            "Cannot operate on expressions from different connections. "
+            "Ensure both operands use the same dj.Instance or global connection."
+        )
+
     if semantic_check:
         # Check if lineage tracking is available for both expressions
         if not expr1.heading.lineage_available or not expr2.heading.lineage_available:

src/datajoint/expression.py

@@ -1414,8 +1414,11 @@ def create(cls, arg1, arg2):
             arg2 = arg2()  # instantiate if a class
         if not isinstance(arg2, QueryExpression):
             raise DataJointError("A QueryExpression can only be unioned with another QueryExpression")
-        if arg1.connection != arg2.connection:
-            raise DataJointError("Cannot operate on QueryExpressions originating from different connections.")
+        if arg1.connection is not arg2.connection:
+            raise DataJointError(
+                "Cannot operate on expressions from different connections. "
+                "Ensure both operands use the same dj.Instance or global connection."
+            )
         if set(arg1.primary_key) != set(arg2.primary_key):
             raise DataJointError("The operands of a union must share the same primary key.")
         if set(arg1.heading.secondary_attributes) & set(arg2.heading.secondary_attributes):

src/datajoint/instance.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 from .connection import Connection
 from .errors import ThreadSafetyError
@@ -54,9 +54,11 @@ class Instance:
     password : str
         Database password.
     port : int, optional
-        Database port. Default from config or 3306.
+        Database port. Defaults to 3306 for MySQL, 5432 for PostgreSQL.
     use_tls : bool or dict, optional
         TLS configuration.
+    backend : str, optional
+        Database backend: ``"mysql"`` or ``"postgresql"``. Default from config.
     **kwargs : Any
         Additional config overrides applied to this instance's config.
 
@@ -81,11 +83,19 @@ def __init__(
         password: str,
         port: int | None = None,
         use_tls: bool | dict | None = None,
+        backend: Literal["mysql", "postgresql"] | None = None,
         **kwargs: Any,
     ) -> None:
         # Create fresh config with defaults loaded from env/file
         self.config = _create_config()
 
+        # Apply backend override before other kwargs (port default depends on it)
+        if backend is not None:
+            self.config.database.backend = backend
+            # Re-derive port default since _create_config resolved it before backend was set
+            if port is None and "database__port" not in kwargs:
+                self.config.database.port = 5432 if backend == "postgresql" else 3306
+
         # Apply any config overrides from kwargs
         for key, value in kwargs.items():
             if hasattr(self.config, key):

tests/unit/test_thread_safe.py

@@ -155,6 +155,123 @@ def test_instance_always_allowed_in_thread_safe_mode(self, monkeypatch):
         assert callable(Instance)
 
 
+class TestInstanceBackend:
+    """Test Instance backend parameter."""
+
+    def test_instance_backend_sets_config(self, monkeypatch):
+        """Instance(backend=...) sets config.database.backend."""
+        monkeypatch.setenv("DJ_THREAD_SAFE", "false")
+        from datajoint.instance import Instance
+        from unittest.mock import patch
+
+        with patch("datajoint.instance.Connection"):
+            inst = Instance(
+                host="localhost", user="root", password="secret",
+                backend="postgresql",
+            )
+            assert inst.config.database.backend == "postgresql"
+
+    def test_instance_backend_default_from_config(self, monkeypatch):
+        """Instance without backend uses config default."""
+        monkeypatch.setenv("DJ_THREAD_SAFE", "false")
+        from datajoint.instance import Instance
+        from unittest.mock import patch
+
+        with patch("datajoint.instance.Connection"):
+            inst = Instance(
+                host="localhost", user="root", password="secret",
+            )
+            assert inst.config.database.backend == "mysql"
+
+    def test_instance_backend_affects_port_default(self, monkeypatch):
+        """Instance(backend='postgresql') uses port 5432 by default."""
+        monkeypatch.setenv("DJ_THREAD_SAFE", "false")
+        from datajoint.instance import Instance
+        from unittest.mock import patch, call
+
+        with patch("datajoint.instance.Connection") as MockConn:
+            Instance(
+                host="localhost", user="root", password="secret",
+                backend="postgresql",
+            )
+            # Connection should be called with port 5432 (PostgreSQL default)
+            args, kwargs = MockConn.call_args
+            assert args[3] == 5432  # port is the 4th positional arg
+
+
+class TestCrossConnectionValidation:
+    """Test that cross-connection operations are rejected."""
+
+    def test_join_different_connections_raises(self):
+        """Join of expressions from different connections raises DataJointError."""
+        from datajoint.expression import QueryExpression
+        from datajoint.errors import DataJointError
+        from unittest.mock import MagicMock
+
+        expr1 = QueryExpression()
+        expr1._connection = MagicMock()
+        expr1._heading = MagicMock()
+        expr1._heading.names = []
+
+        expr2 = QueryExpression()
+        expr2._connection = MagicMock()  # different connection object
+        expr2._heading = MagicMock()
+        expr2._heading.names = []
+
+        with pytest.raises(DataJointError, match="different connections"):
+            expr1 * expr2
+
+    def test_join_same_connection_allowed(self):
+        """Join of expressions from the same connection does not raise."""
+        from datajoint.condition import assert_join_compatibility
+        from datajoint.expression import QueryExpression
+        from unittest.mock import MagicMock
+
+        shared_conn = MagicMock()
+
+        expr1 = QueryExpression()
+        expr1._connection = shared_conn
+        expr1._heading = MagicMock()
+        expr1._heading.names = []
+        expr1._heading.lineage_available = False
+
+        expr2 = QueryExpression()
+        expr2._connection = shared_conn
+        expr2._heading = MagicMock()
+        expr2._heading.names = []
+        expr2._heading.lineage_available = False
+
+        # Should not raise
+        assert_join_compatibility(expr1, expr2)
+
+    def test_restriction_different_connections_raises(self):
+        """Restriction by expression from different connection raises DataJointError."""
+        from datajoint.expression import QueryExpression
+        from datajoint.errors import DataJointError
+        from unittest.mock import MagicMock
+
+        expr1 = QueryExpression()
+        expr1._connection = MagicMock()
+        expr1._heading = MagicMock()
+        expr1._heading.names = ["a"]
+        expr1._heading.__getitem__ = MagicMock()
+        expr1._heading.new_attributes = set()
+        expr1._support = ["`db`.`t1`"]
+        expr1._restriction = []
+        expr1._restriction_attributes = set()
+        expr1._joins = []
+        expr1._top = None
+        expr1._original_heading = expr1._heading
+
+        expr2 = QueryExpression()
+        expr2._connection = MagicMock()  # different connection
+        expr2._heading = MagicMock()
+        expr2._heading.names = ["a"]
+
+        with pytest.raises(DataJointError, match="different connections"):
+            expr1 & expr2
+
+
 class TestThreadSafetyError:
     """Test ThreadSafetyError exception."""