fix(query): count and restrict expressions with an empty primary key

An expression with an empty primary key, such as `dj.U() & T` or
`dj.U().aggr(T, n="count(*)")`, holds at most one row. Two operations on
such expressions failed:

- `len()` took the single-column branch and generated
  `count(DISTINCT )`, which is invalid SQL. It now returns 0 or 1 by
  checking whether the row exists.
- Restricting a `dj.U()` aggregation hit a bare assert in
  `Aggregation.make_sql`. Removing the assert alone would not have been
  enough: the MySQL branch emitted HAVING only together with GROUP BY,
  so the restriction would have been dropped without notice. MySQL now
  emits HAVING without GROUP BY, and PostgreSQL takes the subquery path
  whenever the aggregation is restricted, with or without grouping
  attributes.

Restriction conditions on an aggregation are now joined with str(), as
the base where_clause already does, so restricting any aggregation by
False no longer raises TypeError.

Closes #1581
Closes #1580

Author: dimitri-yatsenko

src/datajoint/expression.py

@@ -1156,6 +1156,10 @@ def __len__(self):
             Number of elements in the result set.
         """
         result = self.make_subquery() if self._top else copy.copy(self)
+        if not result.primary_key:
+            # An empty primary key (e.g. ``dj.U() & table``) admits at most one row:
+            # count it by checking whether it exists.
+            return int(bool(result))
         has_left_join = any(is_left for is_left, _ in result._joins)
 
         # Build COUNT query - PostgreSQL requires different syntax for multi-column DISTINCT
@@ -1355,41 +1359,38 @@ def where_clause(self):
     def make_sql(self, fields=None):
         adapter = self.connection.adapter
         fields = self.heading.as_sql(fields or self.heading.names, adapter=adapter)
-        assert self._grouping_attributes or not self.restriction
         distinct = set(self.heading.names) == set(self.primary_key)
+        # Without grouping attributes (``dj.U().aggr(...)``), the whole input is one group.
+        group_by = (
+            " GROUP BY {}".format(", ".join(adapter.quote_identifier(col) for col in self._grouping_attributes))
+            if self._grouping_attributes
+            else ""
+        )
+        conditions = ")AND(".join(str(s) for s in self.restriction)
 
         # PostgreSQL doesn't allow column aliases in HAVING clause (SQL standard).
         # For PostgreSQL with restrictions, wrap aggregation in subquery and use WHERE.
-        use_subquery_for_having = adapter.backend == "postgresql" and self.restriction and self._grouping_attributes
-
-        if use_subquery_for_having:
+        if adapter.backend == "postgresql" and self.restriction:
             # Generate inner query without HAVING
             inner_sql = "SELECT {distinct}{fields} FROM {from_}{where}{group_by}".format(
                 distinct="DISTINCT " if distinct else "",
                 fields=fields,
                 from_=self.from_clause(),
                 where=self.where_clause(),
-                group_by=" GROUP BY {}".format(", ".join(adapter.quote_identifier(col) for col in self._grouping_attributes)),
+                group_by=group_by,
             )
             # Wrap in subquery with WHERE for the HAVING conditions
             subquery_alias = adapter.quote_identifier(f"_aggr{next(self._subquery_alias_count)}")
-            outer_where = " WHERE (%s)" % ")AND(".join(self.restriction)
-            return f"SELECT * FROM ({inner_sql}) AS {subquery_alias}{outer_where}{self.sorting_clauses()}"
+            return f"SELECT * FROM ({inner_sql}) AS {subquery_alias} WHERE ({conditions}){self.sorting_clauses()}"
         else:
-            # MySQL path: use HAVING directly
-            return "SELECT {distinct}{fields} FROM {from_}{where}{group_by}{sorting}".format(
+            # MySQL path: use HAVING directly, which MySQL also accepts without GROUP BY
+            return "SELECT {distinct}{fields} FROM {from_}{where}{group_by}{having}{sorting}".format(
                 distinct="DISTINCT " if distinct else "",
                 fields=fields,
                 from_=self.from_clause(),
                 where=self.where_clause(),
-                group_by=(
-                    ""
-                    if not self.primary_key
-                    else (
-                        " GROUP BY {}".format(", ".join(adapter.quote_identifier(col) for col in self._grouping_attributes))
-                        + ("" if not self.restriction else " HAVING (%s)" % ")AND(".join(self.restriction))
-                    )
-                ),
+                group_by=group_by,
+                having=f" HAVING ({conditions})" if self.restriction else "",
                 sorting=self.sorting_clauses(),
             )
 

tests/integration/test_empty_primary_key.py

@@ -0,0 +1,118 @@
+"""
+Queries whose primary key is empty: ``dj.U() & table`` and ``dj.U().aggr(...)``.
+
+An expression with an empty primary key holds at most one row. ``len()`` must
+count that row (#1581), and a restriction on a ``dj.U()`` aggregation must
+filter it (#1580). Both run on MySQL and PostgreSQL.
+"""
+
+import time
+
+import pytest
+
+import datajoint as dj
+
+
+@pytest.fixture(scope="function")
+def schema_by_backend(connection_by_backend, db_creds_by_backend):
+    """Create a fresh schema per test, parameterized across backends."""
+    backend = db_creds_by_backend["backend"]
+    test_id = str(int(time.time() * 1000))[-8:]
+    schema_name = f"djtest_emptypk_{backend}_{test_id}"[:64]
+
+    if connection_by_backend.is_connected:
+        try:
+            connection_by_backend.query(
+                f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}"
+            )
+        except Exception:
+            pass
+
+    schema = dj.Schema(schema_name, connection=connection_by_backend)
+    yield schema
+
+    if connection_by_backend.is_connected:
+        try:
+            connection_by_backend.query(
+                f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}"
+            )
+        except Exception:
+            pass
+
+
+@pytest.fixture
+def File(schema_by_backend):
+    @schema_by_backend
+    class File(dj.Manual):
+        definition = """
+        file_id : int32
+        ---
+        kind : varchar(16)
+        """
+
+    return File
+
+
+def test_len_of_universal_restriction(File):
+    """``dj.U() & table`` has one row when the table is non-empty, none otherwise."""
+    assert len(dj.U() & File) == 0
+    File.insert([(1, "data"), (2, "data"), (3, "meta")])
+    assert len(dj.U() & File) == 1
+    assert len(dj.U() & (File & "kind = 'meta'")) == 1
+    assert len(dj.U() & (File & "kind = 'none'")) == 0
+
+
+def test_len_of_universal_aggregation_subquery(File):
+    """``len()`` of a subquery over a ``dj.U()`` aggregation, restricted or not."""
+    File.insert([(1, "data"), (2, "data"), (3, "meta")])
+    total = dj.U().aggr(File, n="count(*)").make_subquery()
+    assert len(total) == 1
+    assert len(total & "n = 3") == 1
+    assert len(total & "n = 4") == 0
+
+
+def test_restrict_universal_aggregation(File):
+    """A restriction on the aggregated value of a ``dj.U()`` aggregation filters its row."""
+    File.insert([(1, "data"), (2, "data"), (3, "meta")])
+    total = dj.U().aggr(File, n="count(*)")
+
+    assert (total & "n = 3").to_dicts() == [{"n": 3}]
+    assert (total & {"n": 3}).to_dicts() == [{"n": 3}]
+    assert (total - "n = 4").to_dicts() == [{"n": 3}]
+    assert (total & "n = 4").to_dicts() == []
+    assert (total & {"n": 4}).to_dicts() == []
+    assert (total - "n = 3").to_dicts() == []
+    assert (total & False).to_dicts() == []
+
+    assert len(total & "n = 3") == 1
+    assert len(total & "n = 4") == 0
+    assert total & "n = 3"
+    assert not total & "n = 4"
+
+
+def test_restrict_universal_aggregation_of_restricted_group(File):
+    """Conditions on the group (WHERE) and on the aggregate (HAVING) combine."""
+    File.insert([(1, "data"), (2, "data"), (3, "meta")])
+    data_count = dj.U().aggr(File & "kind = 'data'", n="count(*)")
+    assert (data_count & "n = 2").to_dicts() == [{"n": 2}]
+    assert (data_count & "n = 3").to_dicts() == []
+
+
+def test_restrict_by_restricted_universal_aggregation(File):
+    """A restricted ``dj.U()`` aggregation shares no attributes with ``File``:
+    it keeps every row when non-empty and none when empty."""
+    File.insert([(1, "data"), (2, "data"), (3, "meta")])
+    total = dj.U().aggr(File, n="count(*)")
+    assert len(File & (total & "n = 3")) == 3
+    assert len(File & (total & "n = 4")) == 0
+    assert len(File - (total & "n = 4")) == 3
+
+
+def test_restrict_grouped_aggregation(File):
+    """Restricting an aggregation with grouping attributes still filters by group."""
+    File.insert([(1, "data"), (2, "data"), (3, "meta")])
+    per_kind = dj.U("kind").aggr(File, n="count(*)")
+    assert (per_kind & "n > 1").to_dicts() == [{"kind": "data", "n": 2}]
+    assert len(per_kind & "n > 1") == 1
+    assert (per_kind & {"kind": "meta"}).to_dicts() == [{"kind": "meta", "n": 1}]
+    assert (per_kind & False).to_dicts() == []