Implement semantic matching for joins

This implements semantic matching for DataJoint 2.0 joins as specified in
docs/src/design/semantic-matching-spec.md.

Key changes:

1. Lineage tracking:
   - Add `lineage` field to Attribute class (heading.py)
   - Create lineage.py module for ~lineage table management
   - Populate lineage at table declaration time
   - Clean up lineage entries when tables are dropped
   - Load lineage from database when fetching headings

2. Semantic matching in joins:
   - Update assert_join_compatibility() to check for non-homologous namesakes
   - Update join() to only match on homologous namesakes (same name AND lineage)
   - Lineage is preserved through projections and renames

3. API changes:
   - Remove @ operator (raises error directing to .join(semantic_check=False))
   - dj.U * table raises deprecation error (use dj.U & table instead)
   - dj.U - table raises error (infinite set)
   - dj.U is always compatible (contains all possible lineages)

4. Tests:
   - Add comprehensive tests for lineage tracking
   - Test homologous and non-homologous namesake handling
   - Test deprecated operator errors
   - Test dj.U operations with semantic matching

Author: claude

src/datajoint/condition.py

@@ -97,10 +97,13 @@ def __init__(self, restriction):
 
 def assert_join_compatibility(expr1, expr2):
     """
-    Determine if expressions expr1 and expr2 are join-compatible.  To be join-compatible,
-    the matching attributes in the two expressions must be in the primary key of one or the
-    other expression.
-    Raises an exception if not compatible.
+    Check semantic compatibility of two expressions for joining.
+
+    Uses semantic matching: attributes are only matched when they share both
+    the same name AND the same lineage (origin).
+
+    Raises DataJointError if non-homologous namesakes are detected (same name
+    but different lineage).
 
     :param expr1: A QueryExpression object
     :param expr2: A QueryExpression object
@@ -110,14 +113,25 @@ def assert_join_compatibility(expr1, expr2):
     for rel in (expr1, expr2):
         if not isinstance(rel, (U, QueryExpression)):
             raise DataJointError("Object %r is not a QueryExpression and cannot be joined." % rel)
-    if not isinstance(expr1, U) and not isinstance(expr2, U):  # dj.U is always compatible
-        try:
+
+    # dj.U is always compatible - it contains all possible lineages
+    if isinstance(expr1, U) or isinstance(expr2, U):
+        return
+
+    # Find namesake attributes (same name in both expressions)
+    namesakes = set(expr1.heading.names) & set(expr2.heading.names)
+
+    for name in namesakes:
+        lineage1 = expr1.heading[name].lineage
+        lineage2 = expr2.heading[name].lineage
+
+        # Non-homologous namesakes: same name, different lineage
+        if lineage1 != lineage2:
             raise DataJointError(
-                "Cannot join query expressions on dependent attribute `%s`"
-                % next(r for r in set(expr1.heading.secondary_attributes).intersection(expr2.heading.secondary_attributes))
+                f"Cannot join on attribute `{name}`: different lineages "
+                f"({lineage1} vs {lineage2}). "
+                f"Use .proj() to rename one of the attributes."
             )
-        except StopIteration:
-            pass  # all ok
 
 
 def make_condition(query_expression, condition, columns):

src/datajoint/expression.py

@@ -275,30 +275,45 @@ def __mul__(self, other):
 
     def __matmul__(self, other):
         """
-        Permissive join of query expressions `self` and `other` ignoring compatibility check
-            e.g. ``q1 @ q2``.
+        The @ operator has been removed in DataJoint 2.0.
+        Use .join(other, semantic_check=False) for permissive joins.
         """
-        if inspect.isclass(other) and issubclass(other, QueryExpression):
-            other = other()  # instantiate
-        return self.join(other, semantic_check=False)
+        raise DataJointError(
+            "The @ operator has been removed in DataJoint 2.0. "
+            "Use .join(other, semantic_check=False) for permissive joins."
+        )
 
     def join(self, other, semantic_check=True, left=False):
         """
-        create the joined QueryExpression.
-        a * b  is short for A.join(B)
-        a @ b  is short for A.join(B, semantic_check=False)
-        Additionally, left=True will retain the rows of self, effectively performing a left join.
+        Create the joined QueryExpression.
+
+        Uses semantic matching: only attributes with the same name AND the same
+        lineage (homologous namesakes) are used for joining.
+
+        :param other: QueryExpression to join with
+        :param semantic_check: If True (default), raise error on non-homologous namesakes.
+            If False, bypass semantic check (use for legacy compatibility).
+        :param left: If True, perform a left join retaining all rows from self.
+
+        Examples:
+            a * b  is short for a.join(b)
+            a.join(b, semantic_check=False)  for permissive joins
         """
-        # trigger subqueries if joining on renamed attributes
+        # Handle U objects: redirect to U's restriction operation
         if isinstance(other, U):
-            return other * self
+            return other & self
         if inspect.isclass(other) and issubclass(other, QueryExpression):
             other = other()  # instantiate
         if not isinstance(other, QueryExpression):
             raise DataJointError("The argument of join must be a QueryExpression")
         if semantic_check:
             assert_join_compatibility(self, other)
-        join_attributes = set(n for n in self.heading.names if n in other.heading.names)
+        # Only join on homologous namesakes (same name AND same lineage)
+        join_attributes = set(
+            n
+            for n in self.heading.names
+            if n in other.heading.names and self.heading[n].lineage == other.heading[n].lineage
+        )
         # needs subquery if self's FROM clause has common attributes with other's FROM clause
         need_subquery1 = need_subquery2 = bool(
             (set(self.original_heading.names) & set(other.original_heading.names)) - join_attributes
@@ -735,9 +750,9 @@ class U:
     """
     dj.U objects are the universal sets representing all possible values of their attributes.
     dj.U objects cannot be queried on their own but are useful for forming some queries.
-    dj.U('attr1', ..., 'attrn') represents the universal set with the primary key attributes attr1 ... attrn.
-    The universal set is the set of all possible combinations of values of the attributes.
-    Without any attributes, dj.U() represents the set with one element that has no attributes.
+    dj.U() or dj.U('attr1', ..., 'attrn') represents the universal set with the primary key
+    attributes attr1 ... attrn. Without any attributes, dj.U() represents the set with one
+    element that has no attributes.
 
     Restriction:
 
@@ -747,11 +762,15 @@ class U:
 
     >>> dj.U('contrast', 'brightness') & stimulus
 
+    Empty U for distinct primary keys:
+
+    >>> dj.U() & expr
+
     Aggregation:
 
     In aggregation, dj.U is used for summary calculation over an entire set:
 
-    The following expression yields one element with one attribute `s` containing the total number of elements in
+    The following expression yields one element with one attribute `n` containing the total number of elements in
     query expression `expr`:
 
     >>> dj.U().aggr(expr, n='count(*)')
@@ -760,7 +779,7 @@ class U:
     query expression `expr`.
 
     >>> dj.U().aggr(expr, n='count(distinct attr)')
-    >>> dj.U().aggr(dj.U('attr').aggr(expr), 'n=count(*)')
+    >>> dj.U().aggr(dj.U('attr').aggr(expr), n='count(*)')
 
     The following expression yields one element and one attribute `s` containing the sum of values of attribute `attr`
     over entire result set of expression `expr`:
@@ -770,16 +789,13 @@ class U:
     The following expression yields the set of all unique combinations of attributes `attr1`, `attr2` and the number of
     their occurrences in the result set of query expression `expr`.
 
-    >>> dj.U(attr1,attr2).aggr(expr, n='count(*)')
+    >>> dj.U('attr1', 'attr2').aggr(expr, n='count(*)')
 
-    Joins:
+    Homology:
 
-    If expression `expr` has attributes 'attr1' and 'attr2', then expr * dj.U('attr1','attr2') yields the same result
-    as `expr` but `attr1` and `attr2` are promoted to the the primary key.  This is useful for producing a join on
-    non-primary key attributes.
-    For example, if `attr` is in both expr1 and expr2 but not in their primary keys, then expr1 * expr2 will throw
-    an error because in most cases, it does not make sense to join on non-primary key attributes and users must first
-    rename `attr` in one of the operands.  The expression dj.U('attr') * rel1 * rel2 overrides this constraint.
+    Since dj.U conceptually contains all possible lineages, its attributes are homologous to
+    any namesake attribute in other expressions. This makes dj.U always compatible for
+    semantic matching in joins and restrictions.
     """
 
     def __init__(self, *primary_key):
@@ -826,8 +842,22 @@ def join(self, other, left=False):
         return result
 
     def __mul__(self, other):
-        """shorthand for join"""
-        return self.join(other)
+        """
+        dj.U * table is deprecated in DataJoint 2.0.
+        Use dj.U & table instead.
+        """
+        raise DataJointError(
+            "dj.U(...) * table is deprecated in DataJoint 2.0. "
+            "Use dj.U(...) & table instead."
+        )
+
+    def __sub__(self, other):
+        """
+        dj.U - table produces an infinite set and is not supported.
+        """
+        raise DataJointError(
+            "dj.U(...) - table produces an infinite set and is not supported."
+        )
 
     def aggr(self, group, **named_attributes):
         """

src/datajoint/heading.py

@@ -7,6 +7,7 @@
 
 from .attribute_adapter import get_adapter
 from .attribute_type import AttributeType
+from .lineage import get_all_lineages
 from .declare import (
     EXTERNAL_TYPES,
     NATIVE_TYPES,
@@ -73,6 +74,7 @@ def decode(self, stored, *, key=None):
     attribute_expression=None,
     database=None,
     dtype=object,
+    lineage=None,  # Origin of attribute: "schema.table.attribute" or None for native secondary
 )
 
 
@@ -406,6 +408,11 @@ def _init_from_database(self):
                 # restore adapted type name
                 attr["type"] = adapter_name
 
+        # Load lineage data from ~lineage table
+        lineages = get_all_lineages(conn, database, table_name)
+        for attr in attributes:
+            attr["lineage"] = lineages.get(attr["name"])
+
         self._attributes = dict(((q["name"], Attribute(**q)) for q in attributes))
 
         # Read and tabulate secondary indexes

src/datajoint/lineage.py

@@ -0,0 +1,177 @@
+"""
+Lineage tracking for semantic matching in joins.
+
+Lineage identifies the origin of an attribute - where it was first defined.
+It is represented as a string in the format: "schema.table.attribute"
+
+Only attributes WITH lineage are stored in the ~lineage table:
+- Native primary key attributes: lineage is this table
+- FK-inherited attributes: lineage is traced to the origin
+- Native secondary attributes: no lineage (no entry in table)
+"""
+
+import logging
+
+logger = logging.getLogger(__name__.split(".")[0])
+
+LINEAGE_TABLE_NAME = "~lineage"
+
+
+def _lineage_table_sql(database):
+    """Generate SQL to create the ~lineage table."""
+    return f"""
+        CREATE TABLE IF NOT EXISTS `{database}`.`{LINEAGE_TABLE_NAME}` (
+            table_name VARCHAR(64) NOT NULL,
+            attribute_name VARCHAR(64) NOT NULL,
+            lineage VARCHAR(255) NOT NULL,
+            PRIMARY KEY (table_name, attribute_name)
+        ) ENGINE=InnoDB
+    """
+
+
+def ensure_lineage_table(connection, database):
+    """Create the ~lineage table if it doesn't exist."""
+    connection.query(_lineage_table_sql(database))
+
+
+def lineage_table_exists(connection, database):
+    """Check if the ~lineage table exists in the schema."""
+    result = connection.query(
+        """
+        SELECT COUNT(*) FROM information_schema.tables
+        WHERE table_schema = %s AND table_name = %s
+        """,
+        args=(database, LINEAGE_TABLE_NAME),
+    )
+    return result.fetchone()[0] > 0
+
+
+def get_lineage(connection, database, table_name, attribute_name):
+    """
+    Get lineage for an attribute from the ~lineage table.
+
+    Returns the lineage string if found, None otherwise (indicating no lineage
+    or attribute is a native secondary).
+    """
+    if not lineage_table_exists(connection, database):
+        return None
+
+    result = connection.query(
+        f"""
+        SELECT lineage FROM `{database}`.`{LINEAGE_TABLE_NAME}`
+        WHERE table_name = %s AND attribute_name = %s
+        """,
+        args=(table_name, attribute_name),
+    )
+    row = result.fetchone()
+    return row[0] if row else None
+
+
+def get_all_lineages(connection, database, table_name):
+    """
+    Get all lineage entries for a table.
+
+    Returns a dict mapping attribute_name -> lineage.
+    Attributes not in the dict have no lineage (native secondary).
+    """
+    if not lineage_table_exists(connection, database):
+        return {}
+
+    result = connection.query(
+        f"""
+        SELECT attribute_name, lineage FROM `{database}`.`{LINEAGE_TABLE_NAME}`
+        WHERE table_name = %s
+        """,
+        args=(table_name,),
+    )
+    return {row[0]: row[1] for row in result}
+
+
+def delete_lineage_entries(connection, database, table_name):
+    """Delete all lineage entries for a table."""
+    if not lineage_table_exists(connection, database):
+        return
+
+    connection.query(
+        f"""
+        DELETE FROM `{database}`.`{LINEAGE_TABLE_NAME}`
+        WHERE table_name = %s
+        """,
+        args=(table_name,),
+    )
+
+
+def insert_lineage_entries(connection, database, entries):
+    """
+    Insert lineage entries for a table.
+
+    :param entries: list of (table_name, attribute_name, lineage) tuples
+    """
+    if not entries:
+        return
+
+    ensure_lineage_table(connection, database)
+
+    # Use INSERT ... ON DUPLICATE KEY UPDATE to handle re-declarations
+    for table_name, attribute_name, lineage in entries:
+        connection.query(
+            f"""
+            INSERT INTO `{database}`.`{LINEAGE_TABLE_NAME}`
+            (table_name, attribute_name, lineage)
+            VALUES (%s, %s, %s)
+            ON DUPLICATE KEY UPDATE lineage = VALUES(lineage)
+            """,
+            args=(table_name, attribute_name, lineage),
+        )
+
+
+def compute_lineage_from_dependencies(connection, full_table_name, attribute_name, primary_key):
+    """
+    Compute lineage by traversing FK relationships.
+
+    Fallback method when ~lineage table doesn't exist.
+
+    :param connection: database connection
+    :param full_table_name: fully qualified table name like `schema`.`table`
+    :param attribute_name: the attribute to compute lineage for
+    :param primary_key: list of primary key attribute names for this table
+    :return: lineage string or None
+    """
+    connection.dependencies.load(force=False)
+
+    # Parse database and table name
+    parts = full_table_name.replace("`", "").split(".")
+    database = parts[0]
+    table_name = parts[1]
+
+    # Check if attribute is inherited via FK
+    parents = connection.dependencies.parents(full_table_name)
+    for parent_table, props in parents.items():
+        # Skip alias nodes (numeric strings)
+        if parent_table.isdigit():
+            # Get the actual parent through the alias
+            grandparents = connection.dependencies.parents(parent_table)
+            if grandparents:
+                parent_table, props = next(iter(grandparents.items()))
+
+        attr_map = props.get("attr_map", {})
+        if attribute_name in attr_map:
+            parent_attr = attr_map[attribute_name]
+            parent_parts = parent_table.replace("`", "").split(".")
+            parent_db = parent_parts[0]
+            parent_tbl = parent_parts[1]
+
+            # Get parent's primary key
+            parent_pk = connection.dependencies.nodes.get(parent_table, {}).get("primary_key", set())
+
+            # Recursively trace to origin
+            return compute_lineage_from_dependencies(
+                connection, parent_table, parent_attr, list(parent_pk)
+            )
+
+    # Not inherited - check if primary key
+    if attribute_name in primary_key:
+        return f"{database}.{table_name}.{attribute_name}"
+
+    # Native secondary - no lineage
+    return None