diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 06f1accda..f1bd5d4d2 100644 --- a/CONTRIBUTING.md +++ b/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 diff --git a/src/datajoint/condition.py b/src/datajoint/condition.py index 650015539..7e6d7f23b 100644 --- a/src/datajoint/condition.py +++ b/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]: diff --git a/src/datajoint/connection.py b/src/datajoint/connection.py index 4e6252e0d..3d8378d56 100644 --- a/src/datajoint/connection.py +++ b/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']}" diff --git a/tests/integration/test_autopopulate.py b/tests/integration/test_autopopulate.py index 9bd795efc..744223645 100644 --- a/tests/integration/test_autopopulate.py +++ b/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: diff --git a/tests/integration/test_connection.py b/tests/integration/test_connection.py index ff3940587..199976041 100644 --- a/tests/integration/test_connection.py +++ b/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) == [] diff --git a/tests/integration/test_relational_operand.py b/tests/integration/test_relational_operand.py index eea53288e..7d548c70e 100644 --- a/tests/integration/test_relational_operand.py +++ b/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" diff --git a/tests/integration/test_restriction_complement.py b/tests/integration/test_restriction_complement.py new file mode 100644 index 000000000..499ac7294 --- /dev/null +++ b/tests/integration/test_restriction_complement.py @@ -0,0 +1,311 @@ +""" +The complement of a restriction by a query expression (``A - B``, ``A & dj.Not(B)``) +compiles to a correlated ``NOT EXISTS``. + +Matching uses ``=`` on every shared attribute, so NULL matches nothing and ``A - B`` +is the exact complement of ``A & B``, also when a shared attribute is nullable or the +operand contains NULLs (``NOT IN`` dropped such rows from both sides). +""" + +import time + +import pytest + +import datajoint as dj + + +@pytest.fixture(scope="function") +def schema_by_backend(connection_by_backend, db_creds_by_backend): + backend = db_creds_by_backend["backend"] + schema_name = f"djtest_complement_{backend}_{str(int(time.time() * 1000))[-8:]}" + schema = dj.Schema(schema_name, connection=connection_by_backend) + yield schema + schema.drop(prompt=False) + + +@pytest.fixture +def tables(schema_by_backend): + schema = schema_by_backend + + @schema + class Tag(dj.Lookup): + definition = """ + tag : varchar(8) + --- + color = null : varchar(8) + """ + contents = [("xa", "red"), ("xb", None), ("y%", "blue")] + + @schema + class Subject(dj.Manual): + definition = """ + subject_id : int32 + --- + -> [nullable] Tag + weight : int32 + """ + + @schema + class Session(dj.Manual): + definition = """ + -> Subject + session : int32 + --- + -> [nullable] Tag + """ + + @schema + class Scan(dj.Manual): + definition = """ + -> Session + scan : int32 + """ + + @schema + class SessionCount(dj.Computed): + definition = """ + -> Subject + --- + n_sessions : int32 + """ + + @property + def key_source(self): + return Subject - (Session & "session = 0 AND tag LIKE 'y%'") + + def make(self, key): + self.insert1(dict(key, n_sessions=len(Session & key))) + + tag = [None, "xa", "xb", "y%"] + Subject.insert([(i, tag[i % 4], i % 5) for i in range(24)]) + # subjects 20..23 have no sessions + Session.insert([(i, s, tag[(i + s) % 4]) for i in range(20) for s in range(i % 3 + 1)]) + Scan.insert([(i, s, k) for i, s in zip(*Session.fetch("subject_id", "session")) for k in range(2) if (i + s) % 2]) + return Tag(), Subject(), Session(), Scan(), SessionCount() + + +def rows(query, attrs): + """Distinct tuples of ``attrs`` (in this order) from ``query``.""" + return {tuple(row[a] for a in attrs) for row in query.proj(*attrs).to_dicts()} + + +def expected_anti(a_rows, b_rows): + """Rows of A that match no row of B with SQL equality (NULL matches nothing).""" + return {r for r in a_rows if not any(all(x is not None and x == y for x, y in zip(r, s)) for s in b_rows)} + + +def check_complement(A, B, attrs): + """A & B and A - B partition A, and A - B agrees with SQL equality semantics.""" + keys = A.primary_key + pos, neg = rows(A & B, keys), rows(A - B, keys) + assert pos and neg, "the operand must split A" + assert len(A & B) + len(A - B) == len(A) + assert pos | neg == rows(A, keys) and not pos & neg + assert rows(A & dj.Not(B), keys) == neg + # A - B agrees with a direct evaluation on the matched attributes + b_rows = rows(B, attrs) + expected = {tuple(row[k] for k in keys) for row in A.to_dicts() if expected_anti({tuple(row[a] for a in attrs)}, b_rows)} + assert neg == expected + + +def test_sql_form(tables): + _, Subject, Session, _, _ = tables + sql = (Subject - Session).make_sql() + assert "NOT EXISTS" in sql + assert " not in " not in sql.lower() + # positive restriction still uses IN + sql = (Subject & Session).make_sql() + assert "EXISTS" not in sql and " in (" in sql + + +def test_complement_nullable_namesake(tables): + """Subject and Session share subject_id and the nullable tag.""" + _, Subject, Session, _, _ = tables + common = ["subject_id", "tag"] + assert None in (Session & "tag IS NULL").fetch("tag") + check_complement(Subject, Session, common) + check_complement(Subject, Session.proj("tag"), common) + check_complement(Session, Subject, common) + + +def test_null_in_operand(tables): + """The operand's NULL tag must not empty the anti-restriction (NOT IN returned no rows).""" + _, Subject, Session, _, _ = tables + B = dj.U("tag") & (Session & "subject_id < 4") # every tag, and NULL + assert None in B.fetch("tag") + check_complement(Subject, B, ["tag"]) + assert len(Subject - B) > 0 + # the operand holds every tag, so only subjects with a NULL tag remain + assert rows(Subject & "tag IS NULL", ["subject_id"]) == rows(Subject - B, ["subject_id"]) + + +@pytest.mark.parametrize( + "operand", + ["join", "restricted", "aggregation", "u", "union", "top", "complemented"], +) +def test_operand_shapes(tables, operand): + _, Subject, Session, Scan, _ = tables + B = { + "join": lambda: Session.proj() * Scan, + "restricted": lambda: Session.proj() & "session > 0", + "aggregation": lambda: Subject.aggr(Session.proj(), n="count(*)") & "n > 1", + "u": lambda: dj.U("subject_id") & (Scan & "scan = 1"), + "union": lambda: (Session.proj() & "session = 1") + (Session.proj() & "session = 2"), + "top": lambda: Session.proj() & dj.Top(7, order_by="subject_id DESC"), + "complemented": lambda: Session.proj() - Scan, + }[operand]() + expected = expected_anti(rows(Subject, ["subject_id"]), rows(B, ["subject_id"])) + assert expected and expected != rows(Subject, ["subject_id"]), "the operand must split Subject" + assert rows(Subject - B, ["subject_id"]) == expected + assert len(Subject - B) == len(expected) + if operand == "top" and Subject.connection.adapter.backend == "mysql": + pytest.skip("positive restriction by a LIMIT query fails on MySQL (error 1235), see #1657") + assert len(Subject & B) + len(Subject - B) == len(Subject) + + +def test_union_operand(tables): + """Restriction by a union compiles for both & and - (#1658).""" + _, Subject, Session, _, _ = tables + U = (Session.proj() & "session = 1") + (Session.proj() & "session = 2") + with_sessions = set((Session & "session > 0").fetch("subject_id")) + assert set((Subject & U).fetch("subject_id")) == with_sessions + assert set((Subject - U).fetch("subject_id")) == set(Subject.fetch("subject_id")) - with_sessions + # a union with secondary attributes compiles to a left join UNION an anti-restriction + V = (Subject & "weight < 2").proj(w="weight") + (Subject & "weight > 3").proj(t="tag") + picked = set((Subject & "weight < 2 OR weight > 3").fetch("subject_id")) + assert set((Session & V).fetch("subject_id")) == picked & set(Session.fetch("subject_id")) + assert set((Session - V).fetch("subject_id")) == set(Session.fetch("subject_id")) - picked + assert len(Session & V) + len(Session - V) == len(Session) + + +def test_or_list_and_not(tables): + _, Subject, Session, Scan, _ = tables + # both match on subject_id alone, and the subjects fall into all four regions + B1 = (Session & "session = 1").proj() + B2 = (Scan & "session = 0").proj() + keys = ["subject_id"] + s1, s2 = rows(Subject & B1, keys), rows(Subject & B2, keys) + every = rows(Subject, keys) + assert s1 - s2 and s2 - s1 and s1 & s2 and every - s1 - s2 + assert rows(Subject - [B1, B2], keys) == every - s1 - s2 + assert rows(Subject - [B1, B2], keys) == rows(Subject - B1 - B2, keys) + assert rows(Subject & [B1, B2], keys) == s1 | s2 + assert rows(Subject - dj.AndList([B1, B2]), keys) == every - (s1 & s2) + assert rows(Subject & dj.Not(dj.Not(B1)), keys) == rows(Subject & B1, keys) + assert rows(Subject & [dj.Not(B1), "weight = 0"], keys) == (every - s1) | rows(Subject & "weight = 0", keys) + assert rows(Subject - [B1, "weight = 0"], keys) == (every - s1) - rows(Subject & "weight = 0", keys) + # edge cases of the negated lists + assert len(Subject - []) == len(Subject) + assert len(Subject - dj.AndList([])) == 0 + assert len(Subject - [B1, True]) == 0 + assert rows(Subject - [B1, False], keys) == every - s1 + # complement with nullable namesakes in an OR-list + pos, neg = rows(Subject & [Session, B2], keys), rows(Subject - [Session, B2], keys) + assert pos and neg and not pos & neg and pos | neg == every + + +def test_complement_aggregation(tables): + """A restricted aggregation compiles its restriction to HAVING on MySQL, a wrapper on PostgreSQL.""" + _, Subject, Session, Scan, _ = tables + counts = Subject.aggr(Session.proj(), n="count(*)", exclude_nonmatching=True) + keys = ["subject_id"] + exclude = Scan & "session = 0" + for aggr in (counts, counts & "n > 1"): + expected = rows(aggr, keys) - rows(exclude, keys) + assert expected and expected != rows(aggr, keys), "the operand must split the aggregation" + assert rows(aggr - exclude, keys) == expected + assert len(aggr - exclude) == len(expected) + assert len(aggr & exclude) + len(aggr - exclude) == len(aggr) + # anti-restriction followed by a further restriction of the aggregation + q = (counts - exclude) & "n > 1" + assert rows(q, keys) == rows(counts & "n > 1", keys) - rows(exclude, keys) + + +def test_complement_left_join(tables): + """extend() is a left join; its NULL-filled columns never match.""" + Tag, _, Session, _, _ = tables + E = Session.extend(Tag) + assert len(E) == len(Session) + keys = ["subject_id", "session"] + red = (Tag & "color = 'red'").proj() + expected = rows(E, keys) - rows(E & "tag = 'xa'", keys) + assert rows(E - red, keys) == expected + assert len(E & red) + len(E - red) == len(E) + # match on the left-joined, nullable `color` (no lineage, so without semantic check) + colors = dj.U("color") & Tag + assert None in colors.fetch("color") + q = E.restrict(dj.Not(colors), semantic_check=False) + assert rows(q, keys) == rows(E & "color IS NULL", keys) + q = E.restrict(colors, semantic_check=False) + assert len(q) + len(E.restrict(dj.Not(colors), semantic_check=False)) == len(E) + + +def test_populate_key_source_minus_self(tables): + _, Subject, Session, _, SessionCount = tables + source = SessionCount.key_source + assert "NOT EXISTS" in source.make_sql() + assert SessionCount.progress() == (len(source), len(source)) + SessionCount.populate("subject_id < 10") + remaining, total = SessionCount.progress() + assert total == len(source) + assert remaining == len(source - "subject_id < 10") + assert set(SessionCount.fetch("subject_id")) == set((source & "subject_id < 10").fetch("subject_id")) + SessionCount.populate() + assert SessionCount.progress() == (0, len(source)) + assert len(SessionCount) == len(source) + + +def test_cascade_delete_from_complement(tables): + _, Subject, Session, Scan, _ = tables + keys = ["subject_id"] + session_keys, scan_keys = ["subject_id", "session"], ["subject_id", "session", "scan"] + with_scans = rows(Subject & Scan, keys) + doomed = rows(Subject - Scan, keys) + assert doomed and with_scans + sessions_kept = {s for s in rows(Session, session_keys) if s[:1] in with_scans} + assert sessions_kept != rows(Session, session_keys) + (Subject - Scan).delete() + assert rows(Subject, keys) == with_scans + assert rows(Session, session_keys) == sessions_kept + # cascade through Session to Scan, seeded by an anti-restriction by a restricted operand + second = (Session & "session = 1").proj() + keep = rows(Subject & second, keys) + assert keep and keep != rows(Subject, keys) and len(Scan - (Subject & second).proj()) + scans_kept = {s for s in rows(Scan, scan_keys) if s[:1] in keep} + assert scans_kept + (Subject - second).delete() + assert rows(Subject, keys) == keep + assert rows(Session, keys) == keep + assert rows(Scan, scan_keys) == scans_kept + + +def test_negated_composite_condition(tables): + """NOT applies to the whole conjunction of a multi-attribute dict, record or AndList.""" + _, _, Session, _, _ = tables + key = {"subject_id": 1, "session": 0} + keys = ["subject_id", "session"] + assert len(Session & key) == 1 + # subject 1 has two sessions, so a NOT applied to subject_id = 1 alone would differ + assert len(Session & "subject_id = 1") == 2 + expected = rows(Session, keys) - {(1, 0)} + assert rows(Session - key, keys) == expected + assert rows(Session - [key], keys) == expected + assert rows(Session & dj.Not(key), keys) == expected + assert rows(Session - dj.AndList(["subject_id = 1", "session = 0"]), keys) == expected + record = (Session & key).proj().to_arrays() + assert rows(Session - record[0], keys) == expected + assert rows(Session - record, keys) == expected + + +def test_percent_in_restriction_strings(tables): + """`%` in string restrictions inside and outside the NOT EXISTS operand.""" + _, Subject, Session, _, _ = tables + keys = ["subject_id"] + like_y = Session & "tag LIKE 'y%'" + assert len(like_y) + expected = rows(Subject, keys) - rows(like_y, keys) + assert rows(Subject - like_y.proj(), keys) == expected + outer = Subject & "tag LIKE 'x%'" + assert rows(outer - like_y.proj(), keys) == rows(outer, keys) - rows(like_y, keys) + assert rows(Subject - (Session & {"tag": "y%"}).proj(), keys) == expected + check_complement(Subject, like_y, ["subject_id", "tag"]) diff --git a/tests/integration/test_restriction_complement_reference.py b/tests/integration/test_restriction_complement_reference.py new file mode 100644 index 000000000..1d9e6f453 --- /dev/null +++ b/tests/integration/test_restriction_complement_reference.py @@ -0,0 +1,548 @@ +""" +Restriction by a query expression and its complement, checked against a Python reference. + +The reference matches a row of A to a row of the operand on every shared attribute name, +and ``None`` never matches. The keys of ``A & B`` and ``A - B`` must be disjoint and +together make up the keys of A, and each must agree with the reference. Since the +reference does not depend on the backend, both backends agree with each other wherever +they agree with it. + +The writes at the end (``populate`` with reserved jobs, cascading deletes and +``insert``) take their rows from complements of restrictions. +""" + +import time +from collections import namedtuple + +import pytest + +import datajoint as dj + + +@pytest.fixture(scope="function") +def schema_by_backend(connection_by_backend, db_creds_by_backend): + backend = db_creds_by_backend["backend"] + schema_name = f"djtest_complref_{backend}_{str(int(time.time() * 1000))[-8:]}" + schema = dj.Schema(schema_name, connection=connection_by_backend) + yield schema + schema.drop(prompt=False) + + +def key_set(query, primary_key): + """The primary keys of ``query`` as a set of tuples ordered by ``primary_key``.""" + return {tuple(key[k] for k in primary_key) for key in query.keys()} + + +def assert_partition(A, B): + """ + The keys of ``A & B`` and ``A - B`` are disjoint and together are the keys of A. + + Returns the two key sets. + """ + primary_key = A.primary_key + inside, outside = key_set(A & B, primary_key), key_set(A - B, primary_key) + assert not inside & outside, f"keys in both A & B and A - B: {sorted(inside & outside, key=repr)}" + assert inside | outside == key_set(A, primary_key) + return inside, outside + + +# ---------- Python reference ---------- + +Ref = namedtuple("Ref", "names rows") # attribute names and rows (dicts) of a query expression + + +def ref(query, rows=None): + """Reference operand for ``query``, with its rows from the database unless given.""" + return Ref(frozenset(query.heading.names), query.to_dicts() if rows is None else rows) + + +def ref_matches(row, operand): + """ + Whether ``row`` (a dict) satisfies the reference ``operand``. + + An operand is a ``Ref``, or ``("any", items)``, ``("all", items)`` or ``("not", item)``. + """ + if isinstance(operand, Ref): + shared = [name for name in operand.names if name in row] + if not shared: + return bool(operand.rows) + return any(all(row[a] is not None and row[a] == other[a] for a in shared) for other in operand.rows) + kind, items = operand + if kind == "any": + return any(ref_matches(row, item) for item in items) + if kind == "all": + return all(ref_matches(row, item) for item in items) + assert kind == "not" + return not ref_matches(row, items) + + +def ref_has_null(operand, names): + """Whether any row of the reference ``operand`` holds NULL in one of ``names``.""" + if isinstance(operand, Ref): + return any(row[a] is None for row in operand.rows for a in operand.names & names) + kind, items = operand + return any(ref_has_null(item, names) for item in (items if kind != "not" else [items])) + + +def ref_restrict(rows, operand): + return [row for row in rows if ref_matches(row, operand)] + + +def ref_complement(rows, operand): + return [row for row in rows if not ref_matches(row, operand)] + + +# ---------- Operand matrix ---------- + +TAGS = [None, "a", "b", "c"] +COLORS = [None, "r", "g"] +RIGHT = [ + (0, "a", "r"), + (1, None, "r"), + (2, "b", None), + (3, None, None), + (4, "c", "g"), + (5, "a", None), + (6, "b", "g"), + (7, None, "g"), + (8, "a", "g"), + (9, "b", "r"), +] +MARKS = [(0, 0), (1, None), (2, 3), (3, 4), (4, None), (5, 7), (6, 9), (7, 2)] + + +@pytest.fixture(params=["left", "right", "both"]) +def nulls(request): + """Where the shared nullable attributes ``tag`` and ``color`` hold NULLs.""" + return request.param + + +@pytest.fixture +def matrix(schema_by_backend, nulls): + schema = schema_by_backend + + @schema + class Tag(dj.Lookup): + definition = """ + tag : varchar(8) + """ + contents = [("a",), ("b",), ("c",)] + + @schema + class Color(dj.Lookup): + definition = """ + color : varchar(8) + """ + contents = [("r",), ("g",)] + + @schema + class Left(dj.Manual): + definition = """ + left_id : int32 + --- + -> [nullable] Tag + -> [nullable] Color + """ + + @schema + class Right(dj.Manual): + definition = """ + right_id : int32 + --- + -> [nullable] Tag + -> [nullable] Color + """ + + @schema + class Note(dj.Manual): + definition = """ + -> Right + note : int32 + """ + + @schema + class Mark(dj.Manual): + definition = """ + mark_id : int32 + --- + -> [nullable] Right + """ + + left = [(t, c) for t in TAGS for c in COLORS] + if nulls == "right": + left = [(t, c) for t, c in left if t is not None and c is not None] + Left.insert([(i, t, c) for i, (t, c) in enumerate(left)]) + right = RIGHT + if nulls == "left": + right = [r for r in right if None not in r] + Right.insert(right) + right_ids = {r[0] for r in right} + Note.insert([(i, n) for i in right_ids for n in range(i % 3)]) + # a mark without a right row is NULL-filled by extend(), except when only the left has NULLs + Mark.insert([(m, r) for m, r in MARKS if r in right_ids or (r is None and nulls != "left")]) + return dict(Left=Left, Right=Right, Note=Note, Mark=Mark) + + +def build_operand(name, Right, Note, Mark): + """The operand ``name`` as a query and as its reference.""" + if name == "base": + return Right, ref(Right) + if name == "join": + q = Right * Note + return q, ref(q) + if name == "aggregation": + q = Right.aggr(Note, "tag", "color", n="count(*)") & "n > 0" + return q, ref(q) + if name == "u": + q = dj.U("tag") & (Right & "right_id > 4") + return q, ref(q) + if name == "union": + q = (dj.U("tag", "color") & (Right & "right_id < 4")) + (dj.U("tag", "color") & (Right & "right_id > 6")) + return q, ref(q) + if name == "top": + q = Right & dj.Top(4, order_by="right_id") + return q, ref(q) + if name == "nested": + # A - (B - C), the inner complement matched on the nullable color + C = dj.U("color") & (Right & "right_id in (0, 3)") + inner = Right - C + return inner, ref(inner, rows=ref_complement(Right.to_dicts(), ref(C))) + B, C = Right & "right_id < 5", dj.U("tag") & (Right & "right_id > 5") + if name == "or_list": + return [B, C], ("any", [ref(B), ref(C)]) + if name == "and_list": + return dj.AndList([B, C]), ("all", [ref(B), ref(C)]) + if name == "not_list": + return dj.Not([B, C]), ("not", ("any", [ref(B), ref(C)])) + if name == "extend": + q = Mark.extend(Right) + return q, ref(q) + raise ValueError(name) + + +@pytest.mark.parametrize( + "operand", + ["base", "join", "aggregation", "u", "union", "top", "nested", "or_list", "and_list", "not_list", "extend"], +) +def test_matches_reference(matrix, nulls, operand): + """``A & B`` and ``A - B`` partition A and agree with the Python reference.""" + Left = matrix["Left"] + B, reference = build_operand(operand, matrix["Right"], matrix["Note"], matrix["Mark"]) + rows = Left.to_dicts() + pk = Left.primary_key + # the NULL layout under test is really present on each side + shared = {"tag", "color"} + assert any(r[a] is None for r in rows for a in shared) == (nulls != "right") + assert ref_has_null(reference, shared) == (nulls != "left") + expected_inside = {tuple(r[k] for k in pk) for r in ref_restrict(rows, reference)} + expected_outside = {tuple(r[k] for k in pk) for r in ref_complement(rows, reference)} + assert expected_inside and expected_outside, "the data must exercise both sides" + + assert key_set(Left - B, pk) == expected_outside + assert key_set(Left & dj.Not(B), pk) == expected_outside + if operand == "top" and Left.connection.adapter.backend == "mysql": + pytest.skip("positive restriction by a LIMIT query fails on MySQL (error 1235), see #1657") + inside, outside = assert_partition(Left, B) + assert inside == expected_inside + assert outside == expected_outside + assert key_set(Left - dj.Not(B), pk) == expected_inside + + +# ---------- populate with reserved jobs ---------- + + +def _statuses(jobs, attr): + return {row[attr]: row["status"] for row in jobs.proj("status").to_dicts()} + + +def test_populate_reserved_aggregation_key_source(schema_by_backend): + """Jobs refresh removes jobs whose key left an aggregated key_source.""" + schema = schema_by_backend + + @schema + class Raw(dj.Manual): + definition = """ + raw_id : int32 + """ + + @schema + class Sample(dj.Manual): + definition = """ + -> Raw + sample : int32 + """ + + @schema + class Summary(dj.Computed): + definition = """ + -> Raw + --- + n : int32 + """ + + @property + def key_source(self): + return Raw.aggr(Sample, n="count(*)", exclude_nonmatching=True) & "n > 1" + + def make(self, key): + self.insert1(dict(key, n=len(Sample & key))) + + n_samples = {0: 2, 1: 3, 2: 2, 3: 1, 4: 0, 5: 2} + Raw.insert([(i,) for i in n_samples]) + Sample.insert([(i, s) for i, n in n_samples.items() for s in range(n)]) + assert set(Summary.key_source.fetch("raw_id")) == {0, 1, 2, 5} + + with dj.config.override(jobs={"keep_completed": True, "stale_timeout": 0}): + Summary.populate("raw_id < 2", reserve_jobs=True) + Summary.jobs.refresh() + assert Summary.jobs.reserve({"raw_id": 5}) + Summary.jobs.error({"raw_id": 5}, "failed") + assert _statuses(Summary.jobs, "raw_id") == {0: "success", 1: "success", 2: "pending", 5: "error"} + + (Raw & {"raw_id": 1}).delete(prompt=False) # gone with its Summary row + (Sample & {"raw_id": 2, "sample": 1}).delete(prompt=False) # pending job, key left key_source + (Sample & {"raw_id": 5}).delete(prompt=False) # error record, key left key_source + Sample.insert1((3, 1)) # enters key_source + assert set(Summary.key_source.fetch("raw_id")) == {0, 3} + + result = Summary.populate(reserve_jobs=True) + + assert result["success_count"] == 1 + assert _statuses(Summary.jobs, "raw_id") == {0: "success", 3: "success"} + assert dict(zip(*Summary.fetch("raw_id", "n"))) == {0: 2, 3: 2} + + +def test_populate_reserved_nullable_key_source(schema_by_backend): + """A key_source restricted on a nullable attribute; NULL never matches the operand.""" + schema = schema_by_backend + + @schema + class Tag(dj.Lookup): + definition = """ + tag : varchar(8) + --- + color = null : varchar(8) + """ + contents = [("a", "red"), ("b", None), ("c", "red")] + + @schema + class Raw(dj.Manual): + definition = """ + raw_id : int32 + --- + -> [nullable] Tag + """ + + @schema + class Proc(dj.Computed): + definition = """ + -> Raw + """ + + @property + def key_source(self): + # raws whose tag has no color, by a complement matched on the nullable tag + return Raw - (Tag & "color IS NULL") + + def make(self, key): + self.insert1(key) + + Raw.insert([(0, "a"), (1, "b"), (2, "c"), (3, None), (4, "a"), (5, None)]) + assert set(Proc.key_source.fetch("raw_id")) == {0, 2, 3, 4, 5} + + with dj.config.override(jobs={"stale_timeout": 0}): + Proc.jobs.refresh() + assert Proc.jobs.reserve({"raw_id": 4}) # held by another worker + assert Proc.jobs.reserve({"raw_id": 5}) + Proc.jobs.error({"raw_id": 5}, "failed") + assert _statuses(Proc.jobs, "raw_id") == {0: "pending", 2: "pending", 3: "pending", 4: "reserved", 5: "error"} + + Raw.update1(dict(raw_id=3, tag="b")) # pending job, key left key_source + Raw.update1(dict(raw_id=5, tag="b")) # error record, key left key_source + Raw.update1(dict(raw_id=1, tag=None)) # enters key_source + assert set(Proc.key_source.fetch("raw_id")) == {0, 1, 2, 4} + + result = Proc.populate(reserve_jobs=True) + + assert result["success_count"] == 3 + assert _statuses(Proc.jobs, "raw_id") == {4: "reserved"} + assert set(Proc.fetch("raw_id")) == {0, 1, 2} + + +# ---------- cascading delete and insert ---------- + + +@pytest.fixture +def lab(schema_by_backend): + schema = schema_by_backend + + @schema + class Tag(dj.Lookup): + definition = """ + tag : varchar(8) + --- + color = null : varchar(8) + """ + contents = [("a", "red"), ("b", None)] + + @schema + class Animal(dj.Manual): + definition = """ + animal_id : int32 + --- + -> [nullable] Tag + """ + + @schema + class Site(dj.Manual): + definition = """ + site_id : int32 + --- + -> [nullable] Animal + """ + + @schema + class Recording(dj.Manual): + definition = """ + -> Animal + rec : int32 + """ + + @schema + class Unit(dj.Manual): + definition = """ + -> Recording + -> Site + unit : int32 + """ + + @schema + class Batch(dj.Manual): + definition = """ + batch_id : int32 + --- + -> [nullable] Tag + """ + + class Item(dj.Part): + definition = """ + -> master + item : int32 + """ + + @schema + class Label(dj.Manual): + definition = """ + -> Batch.Item + label : int32 + """ + + Animal.insert(enumerate(["a", "b", None, "a", None, "b"])) + Site.insert(enumerate([0, 1, None, 3, 4, None])) + Recording.insert([(i, r) for i in range(6) for r in range(1 + i % 2)]) + Unit.insert([(i, r, s, 0) for i, r in zip(*Recording.fetch("animal_id", "rec")) for s in {(i + r) % 6, (i + 2) % 6}]) + Batch.insert(enumerate(["a", None, "b", "a", None, "b", "a"])) + Batch.Item.insert([(b, k) for b in range(7) for k in range(b % 3 + 1)]) + Label.insert([(b, k, 0) for b, k in zip(*Batch.Item.fetch("batch_id", "item")) if k != 1]) + return dict(Tag=Tag, Animal=Animal, Site=Site, Recording=Recording, Unit=Unit, Batch=Batch, Label=Label) + + +def test_cascade_to_child_of_two_restricted_parents(lab): + """The cascade restricts Unit by an OR of its two restricted parents, Recording and Site.""" + Tag, Animal, Site, Recording, Unit = (lab[k] for k in ("Tag", "Animal", "Site", "Recording", "Unit")) + red = Tag & "color = 'red'" + keep = {a for (a,) in key_set(Animal & red, ["animal_id"])} + doomed = {r["animal_id"] for r in Animal.to_dicts()} - keep + # NULL tags never match, so animals 2 and 4 are in the complement + assert doomed == {1, 2, 4, 5} + doomed_sites = {r["site_id"] for r in Site.to_dicts() if r["animal_id"] in doomed} + units = Unit.to_dicts() + # each parent alone removes units that the other parent keeps + assert any(u["animal_id"] in keep and u["site_id"] in doomed_sites for u in units) + assert any(u["animal_id"] in doomed and u["site_id"] not in doomed_sites for u in units) + expected_units = { + (u["animal_id"], u["rec"], u["site_id"], u["unit"]) + for u in units + if u["animal_id"] not in doomed and u["site_id"] not in doomed_sites + } + + (Animal - red).delete(prompt=False) + + assert {a for (a,) in key_set(Animal, ["animal_id"])} == keep + assert {r["animal_id"] for r in Recording.to_dicts()} == keep + assert {r["site_id"] for r in Site.to_dicts()} == set(range(6)) - doomed_sites + assert key_set(Unit, ["animal_id", "rec", "site_id", "unit"]) == expected_units + + +def test_delete_complement_with_parts(lab): + """``(A - B).delete()`` deletes exactly the complement and its part and child rows.""" + Animal, Batch, Label = lab["Animal"], lab["Batch"], lab["Label"] + B = dj.U("tag") & (Animal & "animal_id in (0, 2, 3)") # tags a and NULL + assert None in B.fetch("tag") + inside, outside = assert_partition(Batch, B) + # batches with a NULL tag never match, also when the operand holds NULL + assert {b for (b,) in outside} == {1, 2, 4, 5} and {b for (b,) in inside} == {0, 3, 6} + kept = (Batch & B).proj() + items_kept = key_set(Batch.Item & kept, ["batch_id", "item"]) + labels_kept = key_set(Label & kept, ["batch_id", "item", "label"]) + assert len(Batch.Item()) > len(items_kept) and len(Label()) > len(labels_kept) > 0 + + (Batch - B).delete(prompt=False) + + assert key_set(Batch, ["batch_id"]) == inside + assert key_set(Batch.Item, ["batch_id", "item"]) == items_kept + assert key_set(Label, ["batch_id", "item", "label"]) == labels_kept + + +def test_delete_complement_by_same_table(lab): + """The operand may read the table being deleted from (MySQL error 1093 otherwise).""" + Batch = lab["Batch"] + B = dj.U("tag") & (Batch & "batch_id < 2") # tags a and NULL + inside, outside = assert_partition(Batch, B) + assert {b for (b,) in outside} == {1, 2, 4, 5} + (Batch - B).delete(prompt=False) + assert key_set(Batch, ["batch_id"]) == inside + + +def test_insert_missing_rows(schema_by_backend): + """``T.insert(Src - T)`` inserts exactly the rows of Src that T lacks.""" + schema = schema_by_backend + + @schema + class Tag(dj.Lookup): + definition = """ + tag : varchar(8) + """ + contents = [("a",), ("b",)] + + @schema + class Src(dj.Manual): + definition = """ + src_id : int32 + --- + -> [nullable] Tag + """ + + @schema + class Copy(dj.Manual): + definition = """ + -> Src + --- + -> [nullable] Tag + """ + + def contents(table): + return {(r["src_id"], r["tag"]) for r in table.to_dicts()} + + Src.insert(enumerate(["a", None, "b", None, "a", "b"])) + Copy.insert((Src & "src_id in (0, 2)").to_dicts()) + assert len(Src - Copy) == 4 + Copy.insert(Src - Copy) + assert contents(Copy) == contents(Src) + + # Matching on the key alone: the NULL tag of a copied row would never match + (Copy & "src_id > 1").delete(prompt=False) + assert contents(Copy) == {(0, "a"), (1, None)} + assert 1 in (Src - Copy).fetch("src_id") + Copy.insert(Src - Copy.proj()) + assert contents(Copy) == contents(Src) diff --git a/tests/unit/test_connection_warning.py b/tests/unit/test_connection_warning.py index 9eb3cd0eb..4b5bbacdc 100644 --- a/tests/unit/test_connection_warning.py +++ b/tests/unit/test_connection_warning.py @@ -1,10 +1,10 @@ -"""Unit tests for the MariaDB compatibility warning emitted at connect time.""" +"""Unit tests for the server compatibility warnings emitted at connect time.""" import warnings import pytest -from datajoint.connection import _warn_if_mariadb +from datajoint.connection import MIN_MYSQL_VERSION, _warn_if_mariadb, _warn_if_old_mysql @pytest.mark.parametrize( @@ -40,3 +40,55 @@ def test_no_warn_on_mysql(version_str): warnings.simplefilter("always") _warn_if_mariadb(version_str) assert caught == [] + + +@pytest.mark.parametrize( + "version_str", + [ + "8.0.13", + "8.0.16", + "8.0.17", + "8.0.17-log", + "5.7.44", + ], +) +def test_warn_on_old_mysql(version_str): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + _warn_if_old_mysql(version_str) + assert len(caught) == 1 + assert issubclass(caught[0].category, UserWarning) + message = str(caught[0].message) + assert "requires MySQL 8.0.18 or later" in message + assert version_str in message + assert "(A - B)" in message + for context in ("DELETE", "HAVING", "OR condition", "populate(reserve_jobs=True)", "jobs refresh"): + assert context in message + + +@pytest.mark.parametrize( + "version_str", + [ + "8.0.18", + "8.0.18-log", + "8.0.21", + "8.0.40", + "8.0.40-0ubuntu0.22.04.1", + "8.4.2-log", + "8.4.0", + "9.0.0", + "9.1.0", + "10.11.5-MariaDB", # MariaDB has its own warning + "5.5.68-MariaDB", + "unknown", + ], +) +def test_no_warn_on_current_mysql(version_str): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + _warn_if_old_mysql(version_str) + assert caught == [] + + +def test_min_mysql_version(): + assert MIN_MYSQL_VERSION == (8, 0, 18)