fix(query): compile the complement of a restriction to NOT EXISTS; MySQL 8.0.18 minimum

CONTRIBUTING.md

@@ -56,7 +56,7 @@ export DOCKER_HOST=unix://$HOME/.docker/run/docker.sock
 
 ### PostgreSQL Backend
 
-DataJoint supports MySQL 8.0.13+ and PostgreSQL 15+ as production database backends. To install the PostgreSQL driver:
+DataJoint supports MySQL 8.0.18+ and PostgreSQL 15+ as production database backends. On connecting to an older MySQL server, DataJoint emits a warning (it does not refuse the connection): older servers can be orders of magnitude slower for the complement of a restriction (`A - B`) inside a `DELETE`, a `HAVING` clause or an `OR` condition. To install the PostgreSQL driver:
 
 ```bash
 pip install -e ".[postgres]"    # Installs psycopg2-binary

src/datajoint/condition.py

@@ -291,6 +291,28 @@ def assert_join_compatibility(
             )
 
 
+def _not_exists(adapter, operand: QueryExpression, attributes: list[str]) -> str:
+    """
+    Compile the complement of the restriction by ``operand`` on ``attributes`` as a correlated
+    NOT EXISTS.
+
+    The operand's columns are renamed inside the subquery, so the unqualified outer names
+    cannot resolve to the operand and always refer to the restricted query. The aliases
+    start with ``$``, which no attribute name can (hidden attributes start with ``_``).
+    Each column is matched with its own ``=``: NULL matches nothing, so ``A - B`` is the
+    exact complement of ``A & B``. Unlike ``NOT IN``, both MySQL and PostgreSQL decorrelate
+    this form, also when a matched column is nullable.
+    """
+    q = adapter.quote_identifier
+    keys = [q(f"$k{i}") for i in range(len(attributes))]
+    renamed = ", ".join(f"{q(a)} AS {k}" for a, k in zip(attributes, keys))
+    match = " AND ".join(f"{q('$r')}.{k} = {q(a)}" for a, k in zip(attributes, keys))
+    return (
+        f"NOT EXISTS (SELECT 1 FROM (SELECT {renamed} FROM ({operand.make_sql(attributes)}) AS {q('$s')}) "
+        f"AS {q('$r')} WHERE {match})"
+    )
+
+
 def make_condition(
     query_expression: QueryExpression,
     condition: Any,
@@ -330,7 +352,7 @@ def make_condition(
     - ``numpy.void``: Like dict, from record array
     - ``pandas.DataFrame``: Converted to records, then OR-ed
     """
-    from .expression import Aggregation, QueryExpression, U
+    from .expression import Aggregation, QueryExpression, U, Union
 
     # Get adapter for backend-agnostic SQL generation
     adapter = query_expression.connection.adapter
@@ -372,7 +394,9 @@ def prep_value(k, v):
         return f"{k}={v}"
 
     def combine_conditions(negate, conditions):
-        return f"{'NOT ' if negate else ''} ({')AND('.join(conditions)})"
+        combined = f" ({')AND('.join(conditions)})"
+        # NOT binds tighter than AND: negate the whole conjunction, not its first term
+        return f"NOT ({combined})" if negate else combined
 
     negate = False
     while isinstance(condition, Not):
@@ -386,6 +410,9 @@ def combine_conditions(negate, conditions):
 
     # restrict by AndList
     if isinstance(condition, AndList):
+        if negate:
+            # NOT (a AND b) == (NOT a) OR (NOT b): push the negation down to each item
+            return make_condition(query_expression, [Not(cond) for cond in condition], columns, semantic_check)
         # omit all conditions that evaluate to True
         items = [
             item
@@ -442,19 +469,16 @@ def combine_conditions(negate, conditions):
         # Match on all non-hidden namesakes (hidden attributes excluded)
         common_attributes = [q for q in condition.heading.names if q in query_expression.heading.names]
         columns.update(common_attributes)
-        if isinstance(condition, Aggregation):
+        if isinstance(condition, (Aggregation, Union)):
             condition = condition.make_subquery()
-        return (
+        if not common_attributes:
             # without common attributes, any non-empty set matches everything
-            (not negate if condition else negate)
-            if not common_attributes
-            else (
-                "({fields}) {not_}in ({subquery})".format(
-                    fields=", ".join(adapter.quote_identifier(a) for a in common_attributes),
-                    not_="not " if negate else "",
-                    subquery=condition.make_sql(common_attributes),
-                )
-            )
+            return not negate if condition else negate
+        if negate:
+            return _not_exists(adapter, condition, common_attributes)
+        return "({fields}) in ({subquery})".format(
+            fields=", ".join(adapter.quote_identifier(a) for a in common_attributes),
+            subquery=condition.make_sql(common_attributes),
         )
 
     # restrict by pandas.DataFrames
@@ -463,14 +487,18 @@ def combine_conditions(negate, conditions):
 
     # if iterable (but not a string, a QueryExpression, or an AndList), treat as an OrList
     try:
-        or_list = [make_condition(query_expression, q, columns, semantic_check) for q in condition]
+        items = list(condition)
     except TypeError:
         raise DataJointError("Invalid restriction type %r" % condition)
-    else:
-        or_list = [item for item in or_list if item is not False]  # ignore False conditions
-        if any(item is True for item in or_list):  # if any item is True, entirely True
-            return not negate
-        return f"{'NOT ' if negate else ''} ({' OR '.join(or_list)})" if or_list else negate
+    if negate:
+        # NOT (a OR b) == (NOT a) AND (NOT b): push the negation down to each item so that
+        # query expressions compile to NOT EXISTS rather than NOT (... IN ...)
+        return make_condition(query_expression, AndList(Not(q) for q in items), columns, semantic_check)
+    or_list = [make_condition(query_expression, q, columns, semantic_check) for q in items]
+    or_list = [item for item in or_list if item is not False]  # ignore False conditions
+    if any(item is True for item in or_list):  # if any item is True, entirely True
+        return True
+    return f" ({' OR '.join(or_list)})" if or_list else False
 
 
 def extract_column_names(sql_expression: str) -> set[str]:

src/datajoint/connection.py

@@ -120,6 +120,34 @@ def _warn_if_mariadb(version_str: str) -> None:
         )
 
 
+MIN_MYSQL_VERSION = (8, 0, 18)
+
+
+def _warn_if_old_mysql(version_str: str) -> None:
+    """
+    Emit a UserWarning if `version_str` is a MySQL release older than MIN_MYSQL_VERSION.
+
+    DataJoint warns rather than refuses: older servers return correct results, but
+    cannot decorrelate the NOT EXISTS subquery that DataJoint emits for the complement
+    of a restriction (A - B) in some positions.  MariaDB and unparsable version strings
+    are ignored.
+    """
+    match = re.match(r"(\d+)\.(\d+)\.(\d+)", version_str)
+    if "MariaDB" in version_str or match is None:
+        return
+    if tuple(int(v) for v in match.groups()) < MIN_MYSQL_VERSION:
+        minimum = ".".join(str(v) for v in MIN_MYSQL_VERSION)
+        warnings.warn(
+            f"DataJoint requires MySQL {minimum} or later (server reports {version_str}). "
+            f"Older servers return correct results, but the complement of a restriction "
+            f"(A - B) inside a DELETE, a HAVING clause or an OR condition can be orders of "
+            f"magnitude slower. This affects deletes restricted by A - B, jobs refresh and "
+            f"populate(reserve_jobs=True). Upgrade the server to MySQL {minimum} or later.",
+            UserWarning,
+            stacklevel=3,
+        )
+
+
 class EmulatedCursor:
     """acts like a cursor"""
 
@@ -234,7 +262,9 @@ def __init__(
             )
             self.connection_id = self.adapter.get_connection_id(self._conn)
             if self.adapter.backend == "mysql":
-                _warn_if_mariadb(self.query("SELECT @@version").fetchone()[0])
+                version = self.query("SELECT @@version").fetchone()[0]
+                _warn_if_mariadb(version)
+                _warn_if_old_mysql(version)
         else:
             raise errors.LostConnectionError(
                 f"Connection failed {self.conn_info['user']}@{self.conn_info['host']}:{self.conn_info['port']}"

tests/integration/test_autopopulate.py

@@ -112,10 +112,10 @@ def test_allow_insert(clean_autopopulate, subject, experiment):
         experiment.insert1(key)
 
 
-def test_populate_antijoin_with_secondary_attrs(clean_autopopulate, subject, experiment):
-    """Test that populate correctly computes pending keys via antijoin.
+def test_populate_pending_keys_with_secondary_attrs(clean_autopopulate, subject, experiment):
+    """Test that populate correctly computes pending keys as the complement key_source - self.
 
-    Verifies that partial populate + antijoin gives correct pending counts.
+    Verifies that partial populate + the complement gives correct pending counts.
     Note: Experiment.make() inserts fake_experiments_per_subject rows per key.
     """
     assert subject, "root tables are empty"
@@ -130,21 +130,21 @@ def test_populate_antijoin_with_secondary_attrs(clean_autopopulate, subject, exp
 
     # key_source - target must return only unpopulated keys
     pending = experiment.key_source - experiment
-    assert len(pending) == total_keys - 2, f"Antijoin returned {len(pending)} pending keys, expected {total_keys - 2}."
+    assert len(pending) == total_keys - 2, f"{len(pending)} pending keys, expected {total_keys - 2}."
 
     # Verify progress() reports correct counts
     remaining, total = experiment.progress()
     assert total == total_keys
     assert remaining == total_keys - 2
 
-    # Populate the rest and verify antijoin returns 0
+    # Populate the rest and verify that no keys are pending
     experiment.populate()
     pending_after = experiment.key_source - experiment
-    assert len(pending_after) == 0, f"Antijoin returned {len(pending_after)} pending keys after full populate, expected 0."
+    assert len(pending_after) == 0, f"{len(pending_after)} pending keys after full populate, expected 0."
 
 
-def test_populate_antijoin_overlapping_attrs(prefix, connection_test):
-    """Regression test: antijoin with overlapping secondary attribute names.
+def test_populate_pending_keys_overlapping_attrs(prefix, connection_test):
+    """Regression test: key_source - self with overlapping secondary attribute names.
 
     This reproduces the bug where `key_source - self` returns ALL keys instead
     of just unpopulated ones. The condition is:
@@ -161,7 +161,7 @@ def test_populate_antijoin_overlapping_attrs(prefix, connection_test):
     quality, processing_datetime as secondary attrs. InitialContainer (target)
     also has those same-named columns with different values.
     """
-    test_schema = dj.Schema(f"{prefix}_antijoin_overlap", connection=connection_test)
+    test_schema = dj.Schema(f"{prefix}_pending_overlap", connection=connection_test)
 
     @test_schema
     class Sensor(dj.Lookup):
@@ -196,7 +196,7 @@ def make(self, key):
             # Fetch source data (key only contains PK after projection)
             source = (Sensor() & key).fetch1()
             # Values intentionally differ from source — this is what triggers
-            # the bug: the antijoin tries to match on num_samples and quality
+            # the bug: the anti-restriction tries to match on num_samples and quality
             # too, and since values differ, no match is found.
             self.insert1(
                 dict(
@@ -229,7 +229,7 @@ def make(self, key):
         assert remaining == 0, f"Expected 0 remaining, got {remaining}"
         assert total == 4
 
-        # Verify antijoin with .proj() is correct
+        # Verify the anti-restriction by .proj() is correct
         pending = ProcessedSensor().key_source - ProcessedSensor().proj()
         assert len(pending) == 0
     finally:

tests/integration/test_connection.py

@@ -136,3 +136,37 @@ def test_cancel(schema_tx, connection_test):
     connection_test.cancel_transaction()
     assert len(Subjects()) == 1, "Length is not 1. Expected because rollback should have happened."
     assert len(Subjects & "subject_id = 2") == 0, "Length is not 0. Expected because rollback should have happened."
+
+
+def test_connect_warns_below_mysql_floor(db_creds_root, monkeypatch):
+    """Connecting to a MySQL server older than MIN_MYSQL_VERSION warns but still connects."""
+    import warnings
+
+    from datajoint import connection as dj_connection
+
+    real_server = dj.Connection(**db_creds_root)
+    server_version = real_server.query("SELECT @@version").fetchone()[0]
+    real_server.close()
+
+    def connect_reporting(version_str):
+        original_query = dj_connection.Connection.query
+
+        def query(self, sql, *args, **kwargs):
+            if sql == "SELECT @@version":
+                return original_query(self, "SELECT %s", (version_str,))
+            return original_query(self, sql, *args, **kwargs)
+
+        monkeypatch.setattr(dj_connection.Connection, "query", query)
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            conn = dj.Connection(**db_creds_root)
+        monkeypatch.setattr(dj_connection.Connection, "query", original_query)
+        assert conn.is_connected
+        conn.close()
+        return [str(w.message) for w in caught if "requires MySQL" in str(w.message)]
+
+    floor = ".".join(str(v) for v in dj_connection.MIN_MYSQL_VERSION)
+    below = connect_reporting("8.0.17")
+    assert len(below) == 1 and f"requires MySQL {floor} or later" in below[0]
+    assert connect_reporting("8.0.18") == []
+    assert connect_reporting(server_version) == []

tests/integration/test_relational_operand.py

@@ -121,7 +121,7 @@ def test_join(schema_simp_pop):
     y = D()
     rel = x * y
     assert len(rel) >= len(x) and len(rel) >= len(y), "incorrect join"
-    assert not rel - cond, "incorrect join, restriction, or antijoin"
+    assert not rel - cond, "incorrect join, restriction, or anti-restriction"
     assert set(x.heading.names).union(y.heading.names) == set(rel.heading.names), "incorrect join heading"
     assert set(x.primary_key).union(y.primary_key) == set(rel.primary_key), "incorrect join primary_key"
 
@@ -176,7 +176,7 @@ def test_project(schema_simp_pop):
 
     # projection after restriction
     cond = L() & "cond_in_l"
-    assert len(D() & cond) + len(D() - cond) == len(D()), "failed semijoin or antijoin"
+    assert len(D() & cond) + len(D() - cond) == len(D()), "restriction and its complement do not partition D"
     assert len((D() & cond).proj()) == len((D() & cond)), "projection failed: altered its arguments cardinality"
 
 
@@ -293,9 +293,9 @@ def test_aggr(schema_simp_pop):
             assert np.isclose(max_, values.max(), rtol=1e-4, atol=1e-5), "aggregation failed (max)"
 
 
-def test_semijoin(schema_simp_pop):
+def test_restriction_by_query(schema_simp_pop):
     """
-    test that semijoins and antijoins are formed correctly
+    test that restrictions by a query and their complements are formed correctly
     """
     x = IJ()
     y = JI()
@@ -341,9 +341,9 @@ def test_restrictions_by_lists(schema_simp_pop):
 
     lenx = len(x)
     assert lenx > 0 and len(y) > 0 and len(x & y) < len(x), "incorrect test setup"
-    assert len(x & y) == len(D * L & "cond_in_l"), "incorrect semijoin"
-    assert len(x - y) == len(x) - len(x & y), "incorrect antijoin"
-    assert len(y - x) == len(y) - len(y & x), "incorrect antijoin"
+    assert len(x & y) == len(D * L & "cond_in_l"), "incorrect restriction"
+    assert len(x - y) == len(x) - len(x & y), "incorrect complement"
+    assert len(y - x) == len(y) - len(y & x), "incorrect complement"
     assert len(x & []) == 0, "incorrect restriction by an empty list"
     assert len(x & ()) == 0, "incorrect restriction by an empty tuple"
     assert len(x & set()) == 0, "incorrect restriction by an empty set"