feat(codegen): struct field access + container equality — TASK-29 Phase B (Tasks 3–4 of 4)

sql_transform/_codegen/engine.py

@@ -115,6 +115,8 @@ def _emit_expr(e: Any, env: dict) -> str:
         return f"{{{items}}}"
     if isinstance(e, cp.ListExpr):
         return f"[{', '.join(_emit_expr(x, env) for x in e.items)}]"
+    if isinstance(e, cp.FieldAccess):
+        return f"rt.getfield({_emit_expr(e.base, env)}, {e.field!r})"
     raise UnsupportedInCodegen(f"cannot compile {type(e).__name__}")
 
 

sql_transform/_codegen/plan.py

@@ -199,6 +199,12 @@ class ListExpr:
     items: list  # list[expr]
 
 
+@dataclass
+class FieldAccess:
+    base: Any
+    field: str
+
+
 @dataclass
 class Case:
     arms: list  # list[tuple[cond, result]]
@@ -303,21 +309,30 @@ def _convert_expr(e: exp.Expression) -> Any:
     if isinstance(e, (exp.Paren, exp.Alias)):
         return _convert_expr(e.this)
     if isinstance(e, exp.Dot):
-        raise UnsupportedInCodegen(
-            "struct field access is not supported in codegen yet"
-        )
+        return FieldAccess(_convert_expr(e.this), _fold(e.expression))
     if isinstance(e, exp.Column):
-        # `s.a.b` parses as a 3-part Column carrying `db` (NOT exp.Dot). Reading
-        # only .table/.name would silently misread it as Column('a', 'b') and drop
-        # `s` -- a wrong answer rather than an error. It is struct field access,
-        # which is deferred.
-        if e.args.get("db") or e.args.get("catalog"):
-            raise UnsupportedInCodegen(
-                "struct field access is not supported in codegen yet"
-            )
-        # Fold the column part (`.this`); leave the qualifier (`.table`) raw --
-        # it names a relation (`__THIS__`/generated), never a data column.
-        return Column(table=e.table or None, name=_fold(e.this))
+        # Dotted parts, outer->inner: [catalog?, db?, table?] + leaf name. Mirror
+        # native expr_build.rs: Column{table:parts[0], name:parts[1]} for the
+        # first two parts, layer FieldAccess for the rest. `s.a.b` -> parts
+        # [s,a,b] -> FieldAccess(Column(s,a), b); the 2-part `s.x` stays a Column
+        # (table.column vs struct.field is ambiguous until schemas resolve, in
+        # _validate_expr). The qualifier (parts[0]) names a relation and stays
+        # raw; column/field parts fold. No `catalog` raise: deliberately dropped
+        # from the plan so `t.s.a.b` (table + nested struct) resolves via layered
+        # FieldAccess rather than erroring -- only a true 4-part catalog.db.tbl.col
+        # (never emitted in __THIS__-scoped SQL) would misparse.
+        quals = [q for q in (e.args.get("catalog"), e.args.get("db")) if q is not None]
+        if e.args.get("table"):
+            quals.append(e.args["table"])
+        if not quals:
+            return Column(table=None, name=_fold(e.this))
+        node: Any = Column(
+            table=quals[0].name, name=_fold(quals[1] if len(quals) > 1 else e.this)
+        )
+        rest = ([*quals[2:], e.this]) if len(quals) > 1 else []
+        for part in rest:
+            node = FieldAccess(node, _fold(part))
+        return node
     if isinstance(e, exp.Null):
         return Literal(None)
     if isinstance(e, exp.Boolean):
@@ -655,8 +670,10 @@ def validate_columns(
         effective_schemas[effective] = schema
 
     used: dict = {}
-    for _, e in plan.projection:
-        _validate_expr(e, resolved, row_schemas, static_schemas, used)
+    plan.projection[:] = [
+        (alias, _validate_expr(e, resolved, row_schemas, static_schemas, used))
+        for alias, e in plan.projection
+    ]
     _validate_rel(plan.input, resolved, row_schemas, static_schemas, used)
     return ColumnValidation({k: sorted(v) for k, v in used.items()}, effective_schemas)
 
@@ -693,43 +710,53 @@ def _resolve_tables(
 
 def _validate_rel(node: Any, resolved, row_schemas, static_schemas, used) -> None:
     if isinstance(node, Filter):
-        _validate_expr(node.predicate, resolved, row_schemas, static_schemas, used)
+        node.predicate = _validate_expr(
+            node.predicate, resolved, row_schemas, static_schemas, used
+        )
         _validate_rel(node.input, resolved, row_schemas, static_schemas, used)
     elif isinstance(node, CrossJoin):
         _validate_rel(node.left, resolved, row_schemas, static_schemas, used)
         _validate_rel(node.right, resolved, row_schemas, static_schemas, used)
     elif isinstance(node, Join):
-        for left, right in node.on:
-            _validate_expr(left, resolved, row_schemas, static_schemas, used)
-            _validate_expr(right, resolved, row_schemas, static_schemas, used)
+        node.on = [
+            (
+                _validate_expr(left, resolved, row_schemas, static_schemas, used),
+                _validate_expr(right, resolved, row_schemas, static_schemas, used),
+            )
+            for left, right in node.on
+        ]
         _validate_rel(node.left, resolved, row_schemas, static_schemas, used)
         _validate_rel(node.right, resolved, row_schemas, static_schemas, used)
     elif isinstance(node, SubqueryAlias):
         _validate_rel(node.input, resolved, row_schemas, static_schemas, used)
     elif isinstance(node, LookupJoin):
-        for k in node.keys:
+        node.keys = [
             _validate_expr(k, resolved, row_schemas, static_schemas, used)
+            for k in node.keys
+        ]
         _validate_rel(node.input, resolved, row_schemas, static_schemas, used)
 
 
-def _validate_expr(e: Any, resolved, row_schemas, static_schemas, used) -> None:
+def _validate_expr(e: Any, resolved, row_schemas, static_schemas, used) -> Any:
+    # Returns the (possibly rewritten) node -- a 2-part Column whose qualifier
+    # is not a relation alias is reinterpreted as struct field access, so
+    # callers must reassign with the return value.
     if isinstance(e, Column):
         if e.table is not None:
             entry = resolved.get(e.table)
             if entry is None:
-                # `s.x` (2-part struct field access, e.g. s is struct{x,y})
-                # parses identically to a table-qualified column -- with no
-                # `db`/`catalog` to flag it, unlike the 3-part `t.s.x` case in
-                # _convert_expr. Without this check it falls to the generic
-                # "unknown table" ValueError instead of the UnsupportedInCodegen
-                # the harness expects for deferred container ops.
-                for schema in (*row_schemas.values(), *static_schemas.values()):
-                    ft = (schema or {}).get(e.table)
-                    if ft is not None and is_container(ft.base):
-                        raise UnsupportedInCodegen(
-                            "struct field access is not supported in codegen yet"
-                        )
-                raise ValueError(f"Unknown table: {e.table}")
+                # `e.table` isn't a relation alias, so the "table.column" parse
+                # was wrong: reinterpret as struct field access `e.table`.`e.name`
+                # and re-validate (mirrors native plan.rs). A relation alias
+                # always wins, so this only runs after the alias lookup fails;
+                # the base column resolves below, erroring if it isn't real.
+                # Fold the qualifier as a column here (the relation lookup above
+                # matched it raw): an unquoted `MyStruct.x` base folds to
+                # `mystruct` like any column, matching DataFusion. (Quotedness is
+                # lost by this point, so a quoted mixed-case struct base is
+                # unsupported -- exotic and untested.)
+                fa = FieldAccess(Column(table=None, name=e.table.lower()), e.name)
+                return _validate_expr(fa, resolved, row_schemas, static_schemas, used)
             real, is_row = entry
             schema = (row_schemas if is_row else static_schemas).get(real)
             if schema is None:
@@ -738,7 +765,7 @@ def _validate_expr(e: Any, resolved, row_schemas, static_schemas, used) -> None:
                 raise ValueError(f"Unknown column: {real}.{e.name}")
             if is_row:
                 used.setdefault(real, set()).add(e.name)
-            return
+            return e
         matches = [
             (effective, real, is_row)
             for effective, (real, is_row) in resolved.items()
@@ -752,27 +779,43 @@ def _validate_expr(e: Any, resolved, row_schemas, static_schemas, used) -> None:
         e.table = effective  # codegen emits a direct subscript off this
         if is_row:
             used.setdefault(real, set()).add(e.name)
+        return e
+    elif isinstance(e, FieldAccess):
+        e.base = _validate_expr(e.base, resolved, row_schemas, static_schemas, used)
     elif isinstance(e, BinaryOp):
-        _validate_expr(e.left, resolved, row_schemas, static_schemas, used)
-        _validate_expr(e.right, resolved, row_schemas, static_schemas, used)
+        e.left = _validate_expr(e.left, resolved, row_schemas, static_schemas, used)
+        e.right = _validate_expr(e.right, resolved, row_schemas, static_schemas, used)
     elif isinstance(e, Not):
-        _validate_expr(e.inner, resolved, row_schemas, static_schemas, used)
+        e.inner = _validate_expr(e.inner, resolved, row_schemas, static_schemas, used)
     elif isinstance(e, Cast):
-        _validate_expr(e.expr, resolved, row_schemas, static_schemas, used)
+        e.expr = _validate_expr(e.expr, resolved, row_schemas, static_schemas, used)
     elif isinstance(e, Case):
-        for cond, result in e.arms:
-            _validate_expr(cond, resolved, row_schemas, static_schemas, used)
-            _validate_expr(result, resolved, row_schemas, static_schemas, used)
-        _validate_expr(e.default, resolved, row_schemas, static_schemas, used)
+        e.arms = [
+            (
+                _validate_expr(cond, resolved, row_schemas, static_schemas, used),
+                _validate_expr(result, resolved, row_schemas, static_schemas, used),
+            )
+            for cond, result in e.arms
+        ]
+        e.default = _validate_expr(
+            e.default, resolved, row_schemas, static_schemas, used
+        )
     elif isinstance(e, Func):
-        for a in e.args:
+        e.args = [
             _validate_expr(a, resolved, row_schemas, static_schemas, used)
+            for a in e.args
+        ]
     elif isinstance(e, StructExpr):
-        for _, v in e.fields:
-            _validate_expr(v, resolved, row_schemas, static_schemas, used)
+        e.fields = [
+            (name, _validate_expr(v, resolved, row_schemas, static_schemas, used))
+            for name, v in e.fields
+        ]
     elif isinstance(e, ListExpr):
-        for x in e.items:
+        e.items = [
             _validate_expr(x, resolved, row_schemas, static_schemas, used)
+            for x in e.items
+        ]
+    return e
 
 
 _STR_FUNCS = frozenset({"upper", "lower", "trim", "substr", "substring"})
@@ -788,19 +831,22 @@ def infer_type(e: Any, schemas: dict) -> FieldType:
     if isinstance(e, BinaryOp):
         left, right = infer_type(e.left, schemas), infer_type(e.right, schemas)
         nullable = left.nullable or right.nullable
+        if is_container(left.base) or is_container(right.base):
+            # Equality is the only op defined on whole containers (rt.eq/neq deep
+            # type-tagged compare). Every other op -- arithmetic, dpipe, ordering
+            # -- has no container semantics; defer rather than fall through to the
+            # scalar arms below (which would mistype struct || x as STR, etc.).
+            # This check must precede those arms.
+            if e.op in ("eq", "neq"):
+                return FieldType(BOOL, nullable)
+            raise UnsupportedInCodegen(
+                "only equality is supported on struct/list values in codegen yet"
+            )
         if e.op in ("add", "sub", "mul", "div", "mod"):
             base = INT if left.base == INT and right.base == INT else FLOAT
             return FieldType(base, nullable)
         if e.op == "dpipe":
             return FieldType(STR, nullable)
-        if is_container(left.base) or is_container(right.base):
-            # Comparing/combining structs or lists needs deep equality, which
-            # the runtime doesn't implement -- without this it silently reaches
-            # the scalar-only arithmetic comparison path and crashes instead of
-            # being reported as a deferred surface.
-            raise UnsupportedInCodegen(
-                "struct/list comparison is not supported in codegen yet"
-            )
         return FieldType(BOOL, nullable)
     if isinstance(e, Not):
         return FieldType(BOOL, infer_type(e.inner, schemas).nullable)
@@ -830,6 +876,16 @@ def infer_type(e: Any, schemas: dict) -> FieldType:
                 _common_base(elem_types), any(t.nullable for t in elem_types)
             )
         return FieldType(ListBase(elem), False)
+    if isinstance(e, FieldAccess):
+        base_ty = infer_type(e.base, schemas)
+        if not isinstance(base_ty.base, StructBase):
+            raise ValueError(f"cannot access field {e.field!r} on a non-struct")
+        for name, ft in base_ty.base.fields:
+            if name == e.field:
+                # A NULL struct makes the field NULL too, so nullability is the
+                # field's OR the struct's (mirrors native types.rs FieldAccess).
+                return FieldType(ft.base, ft.nullable or base_ty.nullable)
+        raise ValueError(f"unknown struct field: {e.field}")
     if isinstance(e, Func):
         return _function_type(e.name, [infer_type(a, schemas) for a in e.args])
     raise UnsupportedInCodegen(f"cannot infer the type of {type(e).__name__}")

sql_transform/_codegen/runtime.py

@@ -37,6 +37,16 @@ def type_name(v: Any) -> str:
     return "object"
 
 
+def getfield(v: Any, name: str) -> Any:
+    """Struct field access: a NULL struct yields NULL; a missing field is an
+    error (mirrors native FieldAccess in expr.rs)."""
+    if v is None:
+        return None
+    if isinstance(v, dict) and name in v:
+        return v[name]
+    raise ValueError(f"cannot access field {name!r} on a {type_name(v)} value")
+
+
 def _fmt_float(f: float) -> str:
     """Render a float the way DataFusion does (measured 2026-07-17).
 
@@ -208,12 +218,33 @@ def _cmp(l: Any, r: Any) -> int:  # noqa: E741
     return -1 if a < b else (0 if a == b else 1)
 
 
+def _veq(a: Any, b: Any) -> bool:
+    """Type-tagged structural equality mirroring native Value::PartialEq: dicts
+    (structs) by key order + values, lists elementwise, scalars by variant tag +
+    value (Int(1) != Float(1.0), via val_eq)."""
+    if isinstance(a, dict) and isinstance(b, dict):
+        return list(a.keys()) == list(b.keys()) and all(_veq(a[k], b[k]) for k in a)
+    if isinstance(a, list) and isinstance(b, list):
+        return len(a) == len(b) and all(_veq(x, y) for x, y in zip(a, b, strict=True))
+    if isinstance(a, (dict, list)) or isinstance(b, (dict, list)):
+        return False  # container vs scalar (or struct vs list) never equal
+    return val_eq(a, b)
+
+
 def eq(l: Any, r: Any) -> Any:  # noqa: E741
-    return None if l is None or r is None else _cmp(l, r) == 0
+    if l is None or r is None:
+        return None
+    if isinstance(l, (dict, list)) or isinstance(r, (dict, list)):
+        return _veq(l, r)
+    return _cmp(l, r) == 0
 
 
 def neq(l: Any, r: Any) -> Any:  # noqa: E741
-    return None if l is None or r is None else _cmp(l, r) != 0
+    if l is None or r is None:
+        return None
+    if isinstance(l, (dict, list)) or isinstance(r, (dict, list)):
+        return not _veq(l, r)
+    return _cmp(l, r) != 0
 
 
 def lt(l: Any, r: Any) -> Any:  # noqa: E741

tests/test_codegen_coverage.py

@@ -95,6 +95,11 @@
         "SELECT array(x, y) AS l FROM t",
         {"t": rows({"x": "int", "y": "int"}, [{"x": 1, "y": 2}])},
     ),
+    ("SELECT s.x AS v FROM t", {"t": rows({"s": "struct{x:int}"}, [{"s": {"x": 1}}])}),
+    (
+        "SELECT (s = s) AS x FROM t",
+        {"t": rows({"s": "struct{x:int}"}, [{"s": {"x": 1}}])},
+    ),
 ]
 
 
@@ -110,10 +115,11 @@ def test_committed_surface_is_never_deferred(query, tables):
 
 _DEFERRED = [
     ("SELECT unnest(l) AS x FROM t", {"t": rows({"l": "list[int]"}, [{"l": [1]}])}),
-    # A struct in a comparison must defer, not silently mis-evaluate: DataFusion
-    # supports it, codegen does not yet, so it must raise (not return a wrong bool).
+    # Equality is the ONLY op defined on containers; a struct in any other op
+    # (here dpipe) must still defer, not fall through to the scalar STR/arith
+    # path and render "<object>" -- pins the container-operand guard.
     (
-        "SELECT (s = s) AS x FROM t",
+        "SELECT s || 'x' AS r FROM t",
         {"t": rows({"s": "struct{x:int}"}, [{"s": {"x": 1}}])},
     ),
 ]

tests/test_diff_types.py

@@ -148,6 +148,46 @@ def test_qualified_struct_field_access():
     )
 
 
+def test_uppercase_qualifier_field_access(xfail_on_native):
+    # An unquoted struct-column qualifier folds like any column: `S.x` -> `s.x`,
+    # matching DataFusion. Native doesn't fold the qualifier (raises "Unknown
+    # column: S") -- parity gap, wants its own ticket.
+    xfail_on_native("native does not fold an unquoted struct-column qualifier (S.x)")
+    check(
+        "SELECT S.x AS v FROM t",
+        {"t": rows({"s": "struct{x:int}"}, [{"s": {"x": 7}}])},
+    )
+
+
+def test_struct_equality_true():
+    check(
+        "SELECT (s = s) AS eq FROM t",
+        {"t": rows({"s": "struct{x:int,y:int}"}, [{"s": {"x": 1, "y": 2}}])},
+    )
+
+
+def test_struct_equality_false():
+    # Deep, type-tagged: differs in one field -> not equal.
+    check(
+        "SELECT (s = named_struct('x', 1, 'y', 99)) AS eq FROM t",
+        {"t": rows({"s": "struct{x:int,y:int}"}, [{"s": {"x": 1, "y": 2}}])},
+    )
+
+
+def test_list_equality():
+    check(
+        "SELECT (l = l) AS eq FROM t",
+        {"t": rows({"l": "list[int]"}, [{"l": [1, 2, 3]}])},
+    )
+
+
+def test_list_inequality():
+    check(
+        "SELECT (l = [9, 9, 9]) AS eq FROM t",
+        {"t": rows({"l": "list[int]"}, [{"l": [1, 2, 3]}])},
+    )
+
+
 def test_unnest_struct_expands_columns():
     check(
         "SELECT unnest(named_struct('x', a, 'y', b)) FROM t",