Define the quad row layout and the predicate stage's term count

dau_core/operators.py

@@ -203,6 +203,10 @@ def matches(self, row: tuple[int, int, int, int]) -> bool:
         return _compare(self.op, _wrap_int32(row[self.field]), self.constant)
 
 
+# the terms one predicate stage compares: term0, term1 and term2 below
+ROW_PREDICATE_TERMS = 3
+
+
 class RowPredicateConfig(CheckedModel):
     """Compound row predicate: up to three terms combined by the two-level
     tree ``(term0 <combine01> term1) <combine2> term2``. ``SINGLE`` at a

dau_core/records.py

@@ -1,5 +1,5 @@
-"""Byte layouts of the rows and records the grouped and bar tiles exchange
-with the host.
+"""Byte layouts of the rows and records the row, grouped and bar tiles
+exchange with the host.
 
 A layout is a tuple of ``(field, kind)`` pairs in wire order, little-endian,
 with ``kind`` a numpy-style code (``i4``, ``u4``, ``i8``, ``u8``). The tiles'
@@ -25,6 +25,8 @@
     "GROUP_RECORD_FIELDS",
     "KEYED_ROW_BYTES",
     "KEYED_ROW_FIELDS",
+    "QUAD_ROW_BYTES",
+    "QUAD_ROW_FIELDS",
     "numpy_dtype_fields",
     "record_bytes",
     "struct_format",
@@ -34,6 +36,8 @@
 
 _STRUCT_CODES = {"i4": "i", "u4": "I", "i8": "q", "u8": "Q"}
 
+# the row tiles' input: four int32 fields in ``RowField`` order
+QUAD_ROW_FIELDS: Layout = (("key", "i4"), ("value", "i4"), ("aux0", "i4"), ("aux1", "i4"))
 # grouped-aggregation input: one (key, value) pair per row
 KEYED_ROW_FIELDS: Layout = (("key", "i4"), ("value", "u4"))
 # grouped-aggregation output: one closed group
@@ -78,6 +82,7 @@ def record_bytes(fields: Layout) -> int:
     return struct.calcsize(struct_format(fields))
 
 
+QUAD_ROW_BYTES = record_bytes(QUAD_ROW_FIELDS)
 KEYED_ROW_BYTES = record_bytes(KEYED_ROW_FIELDS)
 GROUP_RECORD_BYTES = record_bytes(GROUP_RECORD_FIELDS)
 BAR_ROW_BYTES = record_bytes(BAR_ROW_FIELDS)

dau_core/tests/test_operators.py

@@ -1357,3 +1357,9 @@ def test_field_sum_combine_refuses_a_malformed_record_and_an_overflowing_union()
 def test_field_sum_decode_refuses_non_integer_words() -> None:
     with pytest.raises(ValueError, match="unsigned 64-bit integers"):
         FieldSumAggregate.from_record_words(cast(Any, [0, 2.5]), field_mask=0b0001)
+
+
+def test_row_predicate_terms_counts_the_config_terms() -> None:
+    from dau_core.operators import ROW_PREDICATE_TERMS
+
+    assert ROW_PREDICATE_TERMS == len([name for name in RowPredicateConfig.model_fields if name.startswith("term")])

dau_core/tests/test_records.py

@@ -8,7 +8,7 @@
 
 import pytest
 
-from dau_core.operators import BarAggregate, GroupAggregate
+from dau_core.operators import BarAggregate, GroupAggregate, RowField
 from dau_core.records import (
     BAR_RECORD_BYTES,
     BAR_RECORD_FIELDS,
@@ -18,6 +18,8 @@
     GROUP_RECORD_FIELDS,
     KEYED_ROW_BYTES,
     KEYED_ROW_FIELDS,
+    QUAD_ROW_BYTES,
+    QUAD_ROW_FIELDS,
     numpy_dtype_fields,
     struct_format,
 )
@@ -47,3 +49,11 @@ def test_the_struct_and_numpy_views_agree() -> None:
     for fields in (KEYED_ROW_FIELDS, GROUP_RECORD_FIELDS, BAR_ROW_FIELDS, BAR_RECORD_FIELDS):
         assert numpy.dtype(numpy_dtype_fields(fields)).itemsize == struct.calcsize(struct_format(fields))
     assert struct_format(BAR_RECORD_FIELDS) == "<iiiiiiqqQ"
+
+
+def test_the_quad_row_is_the_row_tiles_input_in_field_order() -> None:
+    assert [name for name, _kind in QUAD_ROW_FIELDS] == [field.name.lower() for field in RowField]
+    for tile in ("row-predicate-filter", "row-map-alu", "field-sum-aggregation"):
+        record_words = core(tile).record_words
+        assert record_words is not None
+        assert QUAD_ROW_BYTES == record_words * WORD_BYTES