fix: dj.Top is honored when a limited expression is used in another query

src/datajoint/condition.py

@@ -167,15 +167,13 @@ def merge(self, other: "Top") -> "Top":
         Top
             New Top with merged limit/offset and preserved ordering.
         """
-        # Compute effective limit (minimum of defined limits)
-        if self.limit is None and other.limit is None:
-            new_limit = None
-        elif self.limit is None:
+        # other skips other.offset rows of self's window and takes up to other.limit of the
+        # rest, so the merged window never extends past self's limit
+        if self.limit is None:
             new_limit = other.limit
-        elif other.limit is None:
-            new_limit = self.limit
         else:
-            new_limit = min(self.limit, other.limit)
+            remaining = max(self.limit - other.offset, 0)
+            new_limit = remaining if other.limit is None else min(remaining, other.limit)
 
         return Top(
             limit=new_limit,
@@ -379,6 +377,15 @@ def combine_conditions(negate, conditions):
         negate = not negate
         condition = condition.restriction
 
+    # restrict() applies dj.Top itself; here it was negated or nested in a list or AndList
+    if isinstance(condition, Top):
+        raise DataJointError(
+            "dj.Top must be applied on its own, as `expression & dj.Top(...)`. It cannot be negated "
+            "(with - or dj.Not) or combined with other restrictions in a list, an AndList, or the "
+            "restrictions passed to populate() or progress(). For the rows after the first n, "
+            "use dj.Top(None, order_by=..., offset=n)."
+        )
+
     # restrict by string
     if isinstance(condition, str):
         columns.update(extract_column_names(condition))
@@ -442,7 +449,9 @@ 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):
+        # A dj.Top limit stays inside a subquery: it applies to the expression's rows before
+        # they are projected to the common attributes, and MySQL rejects LIMIT inside IN (...).
+        if isinstance(condition, Aggregation) or condition._top is not None:
             condition = condition.make_subquery()
         return (
             # without common attributes, any non-empty set matches everything

src/datajoint/diagram.py

@@ -799,8 +799,9 @@ def restrict(self, table_expr):
         if node not in result.nodes():
             raise DataJointError(f"Table {node} is not in the diagram.")
         # Seed restriction (AND accumulation)
-        result._restrict_conditions.setdefault(node, AndList()).extend(table_expr.restriction)
-        result._restriction_attrs.setdefault(node, set()).update(table_expr.restriction_attributes)
+        restriction, restriction_attrs = _seed_restriction(table_expr)
+        result._restrict_conditions.setdefault(node, AndList()).extend(restriction)
+        result._restriction_attrs.setdefault(node, set()).update(restriction_attrs)
         # Propagate downstream
         result._propagate_restrictions(node, mode="restrict")
         return result

src/datajoint/expression.py

@@ -400,18 +400,21 @@ def join(self, other, semantic_check=True, left=False, allow_nullable_pk=False):
         need_subquery1 = need_subquery2 = bool(
             (set(self.original_heading.names) & set(other.original_heading.names)) - join_attributes
         )
-        # need subquery if any of the join attributes are derived
+        # need subquery if any of the join attributes are derived, or if an operand is limited
+        # by dj.Top: its ORDER BY/LIMIT must apply to the operand before the join
         need_subquery1 = (
             need_subquery1
             or isinstance(self, Aggregation)
             or any(n in self.heading.new_attributes for n in join_attributes)
             or isinstance(self, Union)
+            or self._top is not None
         )
         need_subquery2 = (
             need_subquery2
             or isinstance(other, Aggregation)
             or any(n in other.heading.new_attributes for n in join_attributes)
             or isinstance(self, Union)
+            or other._top is not None
         )
         # With USING clause (instead of NATURAL JOIN), we need subqueries when
         # joining with multi-table expressions to ensure correct column matching
@@ -614,6 +617,22 @@ def proj(self, *attributes, **named_attributes):
         if not need_subquery and self.restriction:
             # need a subquery if the restriction applies to attributes that have been renamed
             need_subquery = any(name in self.restriction_attributes for name in self.heading.new_attributes)
+        if not need_subquery and self._top is not None:
+            # need a subquery if dj.Top's ORDER BY names an attribute that the projection redefines,
+            # or a computed or renamed attribute that it drops. A dropped table column needs none.
+            projected = self.heading.select(
+                attributes, rename_map=dict(**rename_map, **replicate_map), compute_map=compute_map
+            )
+            ordering = ", ".join(_flatten_attribute_list(self.primary_key, self._top.order_by or ["KEY"]))
+            need_subquery = any(
+                (
+                    projected[name].attribute_expression != self.heading[name].attribute_expression
+                    if name in projected.names
+                    else self.heading[name].attribute_expression is not None
+                )
+                for name in extract_column_names(ordering)
+                if name in self.heading.names
+            )
 
         result = self.make_subquery() if need_subquery else copy.copy(self)
         result._original_heading = result.original_heading
@@ -1186,6 +1205,8 @@ def __bool__(self):
         bool
             True if the result is not empty.
         """
+        if self._top is not None:
+            return bool(self.make_subquery())  # EXISTS must see the ORDER BY/LIMIT
         return bool(
             self.connection.query(
                 "SELECT EXISTS(SELECT 1 FROM {from_}{where})".format(from_=self.from_clause(), where=self.where_clause())
@@ -1434,15 +1455,17 @@ def make_sql(self):
         if not arg1.heading.secondary_attributes and not arg2.heading.secondary_attributes:
             # no secondary attributes: use UNION DISTINCT
             fields = arg1.primary_key
-            alias_name = f"_u{next(self.__count)}{self.sorting_clauses()}"
-            alias_quoted = self.connection.adapter.quote_identifier(alias_name)
+            alias_quoted = self.connection.adapter.quote_identifier(f"_u{next(self.__count)}")
             sql1 = arg1.make_sql() if isinstance(arg1, Union) else arg1.make_sql(fields)
             sql2 = arg2.make_sql() if isinstance(arg2, Union) else arg2.make_sql(fields)
-            return f"SELECT * FROM (({sql1}) UNION ({sql2})) as {alias_quoted}"
+            return f"SELECT * FROM (({sql1}) UNION ({sql2})) as {alias_quoted}{self.sorting_clauses()}"
         # with secondary attributes, use union of left join with anti-restriction
         fields = self.heading.names
         sql1 = arg1.join(arg2, left=True).make_sql(fields)
         sql2 = (arg2 - arg1).proj(..., **{k: "NULL" for k in arg1.heading.secondary_attributes}).make_sql(fields)
+        if self._top is not None:
+            alias_quoted = self.connection.adapter.quote_identifier(f"_u{next(self.__count)}")
+            return f"SELECT * FROM (({sql1}) UNION ({sql2})) as {alias_quoted}{self.sorting_clauses()}"
         return "({sql1})  UNION ({sql2})".format(sql1=sql1, sql2=sql2)
 
     def from_clause(self):
@@ -1509,6 +1532,8 @@ def __and__(self, other):
             other = other()  # instantiate if a class
         if not isinstance(other, QueryExpression):
             raise DataJointError("Set U can only be restricted with a QueryExpression.")
+        if other._top is not None:
+            other = other.make_subquery()  # apply the limit before DISTINCT
         result = copy.copy(other)
         result._distinct = True
         result._heading = result.heading.set_primary_key(self.primary_key)
@@ -1563,6 +1588,9 @@ def aggr(self, group, **named_attributes):
         if missing:
             raise DataJointError(f"Attributes {missing} not found in the group expression.")
 
+        if group._top is not None:
+            group = group.make_subquery()  # apply the limit before GROUP BY
+
         # Create Aggregation directly without join - just group by U's primary key
         result = Aggregation()
         result._connection = group.connection

src/datajoint/heading.py

@@ -775,5 +775,6 @@ def make_subquery_heading(self):
         """
         return Heading(
             (dict(v.todict(), attribute_expression=None) for v in self.attributes.values()),
+            table_info=self.table_info,  # keep the adapter, so a later rename quotes for the backend
             lineage_available=self._lineage_available,
         )

tests/integration/test_relational_operand.py

@@ -608,12 +608,14 @@ def test_top_errors(self, schema_simp_pop):
             L() & dj.Top(order_by=1)
         with pytest.raises(TypeError) as err5:
             L() & dj.Top(offset="1")
-        assert "datajoint.errors.DataJointError: Invalid restriction type Top(limit=1, order_by=['KEY'], offset=0)" == str(
-            err1.exconly()
-        )
-        assert "datajoint.errors.DataJointError: Invalid restriction type Top(limit=1, order_by=['KEY'], offset=0)" == str(
-            err2.exconly()
+        nested_top = (
+            "datajoint.errors.DataJointError: dj.Top must be applied on its own, as `expression & dj.Top(...)`. "
+            "It cannot be negated (with - or dj.Not) or combined with other restrictions in a list, an AndList, "
+            "or the restrictions passed to populate() or progress(). For the rows after the first n, "
+            "use dj.Top(None, order_by=..., offset=n)."
         )
+        assert nested_top == str(err1.exconly())
+        assert nested_top == str(err2.exconly())
         assert "TypeError: Top limit must be an integer" == str(err3.exconly())
         assert "TypeError: Top order_by attributes must all be strings" == str(err4.exconly())
         assert "TypeError: The offset argument must be an integer" == str(err5.exconly())