Build on protowire-cpp v2.1.0; DatasetReader refuses a broken row

.github/workflows/ci.yml

@@ -19,7 +19,7 @@ env:
   # Pin the sibling C++ checkout to a tagged protowire-cpp release. Python
   # decodes and encodes PXF and SBE through it, so a spec fix reaches this
   # port only when the pin moves. Bump in lockstep with cpp release cuts.
-  PROTOWIRE_CPP_REF: v2.0.0
+  PROTOWIRE_CPP_REF: v2.1.0
 
 jobs:
   # ---------------------------------------------------------------------

.github/workflows/codeql.yml

@@ -17,7 +17,7 @@ permissions:
 
 env:
   # See ci.yml for the rationale on this pin.
-  PROTOWIRE_CPP_REF: v2.0.0
+  PROTOWIRE_CPP_REF: v2.1.0
 
 jobs:
   analyze:

.github/workflows/publish.yml

@@ -26,7 +26,7 @@ env:
   #
   # Pinned to a tagged C++ release: wheels are built from this ref, so it
   # must never be a branch. Bump in lockstep with cpp release cuts.
-  PROTOWIRE_CPP_REF: v2.0.0
+  PROTOWIRE_CPP_REF: v2.1.0
 
 jobs:
   # ---------------------------------------------------------------------

CHANGELOG.md

@@ -13,6 +13,9 @@ format changes.
 
 ### Changed
 
+- **Built on protowire-cpp v2.1.0** (#45), up from v2.0.0. Its fixes to the
+  `@dataset` streaming reader reach `pxf.DatasetReader`; they are listed under
+  Fixed below. See protowire-cpp's [2.1.0 release notes](https://github.com/trendvidia/protowire-cpp/releases/tag/v2.1.0).
 - **Built on protowire-cpp v2.0.0** (#31), up from v1.2.0. This port decodes
   and encodes PXF and SBE through it, so the changes below reach Python
   callers with no Python API change. Several narrow what decodes, encodes or
@@ -76,6 +79,25 @@ format changes.
 
 ### Fixed
 
+- **`DatasetReader` refuses a row cut short instead of ending the dataset
+  there** (#45). A last row whose `)` never came, a broken row followed by
+  more rows, or junk between rows read as a shorter, complete dataset, with no
+  error. Each is now the error `unmarshal_full` gives, at the same position,
+  as protowire-go's reader does. The rows end only at the end of the input or
+  at the next directive.
+- **`DatasetReader` reads rows under its `max_bytes_literal_length`**
+  (protowire-cpp#160). Rows were parsed under the default, so a lowered limit
+  was only enforced when `scan` bound the row, and `next_or_none` returned it.
+- **`DatasetReader` caps the bytes it holds while looking for a row's end at
+  `max_message_size`,** as protowire-go does: `dataset input holds N bytes
+  without a row boundary, MaxMessageSize=M`. The cap counts every byte the
+  reader holds, the header included, and the reader reads 4 KiB at a time.
+- **`DatasetReader.scan` no longer applies `max_message_size` to the row it
+  binds** (#45), as protowire-go binds it. The body a row is bound as repeats
+  the column names, so it is wider than the row: a row the reader accepted
+  could still be refused when bound. The other limits apply as before.
+- **`DatasetReader` positions count from the start of the input.** Every
+  row's error was at `1:1`.
 - **`bind_row` and `DatasetReader.scan` escape a line feed in a string
   cell.** On protowire-cpp v2.0.0 a cell holding one failed with `newline in
   string literal`.

src/protowire/pxf.py

@@ -446,7 +446,8 @@ class DatasetReader:
     def __init__(self, native: object, limits: dict[str, int] | None = None) -> None:
         # Private — construct via `from_bytes`. We keep the native handle
         # only; type/columns/directives are forwarded on demand. `limits` are
-        # the from_bytes keywords, which scan passes on to bind_row.
+        # the from_bytes keywords, which scan passes on to bind_row (all but
+        # max_message_size).
         self._native = native
         self._limits = limits or {}
 
@@ -463,7 +464,8 @@ def from_bytes(
         max_big_float_exponent: int = 0,
     ) -> "DatasetReader":
         """Read `data`'s `@dataset` header. The limits are described under
-        "Decode limits" above; `scan` binds each row under the same ones."""
+        "Decode limits" above. `scan` binds each row under the same ones,
+        except `max_message_size`, which caps the bytes of a row as it is read."""
         if isinstance(data, str):
             data = data.encode("utf-8")
         limits = dict(
@@ -539,7 +541,11 @@ def scan(self, msg: Message, *, skip_validate: bool = True) -> bool:
         cells = self.next_or_none()
         if cells is None:
             return False
-        bind_row(msg, self.columns, cells, skip_validate=skip_validate, **self._limits)
+        # Every limit but max_message_size, as protowire-go binds a row: the
+        # body a row is bound as repeats the column names, so it is wider
+        # than the row, whose bytes the reader has already capped.
+        limits = dict(self._limits, max_message_size=0)
+        bind_row(msg, self.columns, cells, skip_validate=skip_validate, **limits)
         return True
 
 

tests/test_pxf_dataset_reader.py

@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+import re
+
 import pytest
 
 from protowire import pxf
@@ -245,3 +247,93 @@ def test_bind_row_string_with_line_feed(all_types_cls):
     msg = all_types_cls()
     pxf.bind_row(msg, tr.columns, row)
     assert msg.string_field == "a\nb\r\tc"
+
+
+# ---- appending: where a row sequence may end (spec testdata/dataset) -------
+#
+# Four of the spec's fixtures, copied verbatim from testdata/dataset/ of
+# trendvidia/protowire (CI has no checkout of it). The errors are protowire-go
+# v1.17.0's, streamed through its DatasetReader: same text, same position. On
+# protowire-cpp v2.0.0 the three broken ones read as one clean row, then EOF.
+
+_FIXTURES = {
+    "truncated-row-eof": """\
+# §3.4.4 appending. The last row is cut short by end of input — the shape
+# an interrupted append leaves behind. MUST be rejected, by the streaming
+# API as well as the materializing one: a row whose closing ")" never
+# arrives is a decode error, not the end of the row sequence.
+
+@dataset test.v1.AllTypes (string_field, int32_field)
+
+("row-1", 1)
+("row-2", 2
+""",
+    "truncated-row-then-rows": """\
+# §3.4.4 appending. A row cut short mid-number, followed by complete rows
+# appended later. MUST be rejected — a decoder that treated the broken row
+# as the end of the sequence would silently drop every row after it.
+
+@dataset test.v1.AllTypes (string_field, int32_field)
+
+("row-1", 1)
+("row-2", 2
+("row-3", 3)
+("row-4", 4)
+""",
+    "junk-between-rows": """\
+# §3.4.4 appending. Something other than a row or a directive where the next
+# row could begin. MUST be rejected — the row sequence ends only at end of
+# input or at the next directive.
+
+@dataset test.v1.AllTypes (string_field, int32_field)
+
+("row-1", 1)
+xyz
+("row-2", 2)
+""",
+    "rows-end-at-directive": """\
+# §3.4.4 appending. Two datasets in one document. The first one's rows end
+# at the second "@dataset" directive, which is a legitimate end of the
+# sequence. MUST be accepted: two datasets, two rows and one row.
+
+@dataset test.v1.AllTypes (string_field, int32_field)
+
+("row-1", 1)
+("row-2", 2)
+
+@dataset test.v1.AllTypes (string_field, enum_field)
+
+("row-3", STATUS_ACTIVE)
+""",
+}
+
+
+@pytest.mark.parametrize("fixture, error", [
+    ("truncated-row-eof", "10:1: expected ',' or ')' in @dataset row, got EOF"),
+    ("truncated-row-then-rows", "9:1: expected ',' or ')' in @dataset row, got ("),
+    ("junk-between-rows",
+     "8:1: @dataset directive cannot coexist with top-level field entries"),
+])
+def test_a_broken_row_is_an_error_not_the_end(all_types_cls, fixture, error):
+    reader = pxf.DatasetReader.from_bytes(_FIXTURES[fixture])
+    msg = all_types_cls()
+    assert reader.scan(msg)
+    assert (msg.string_field, msg.int32_field) == ("row-1", 1)
+    with pytest.raises(ValueError, match=re.escape(error)):
+        reader.scan(all_types_cls())
+
+
+def test_rows_end_at_the_next_directive(all_types_cls):
+    first = pxf.DatasetReader.from_bytes(_FIXTURES["rows-end-at-directive"])
+    got = []
+    msg = all_types_cls()
+    while first.scan(msg):
+        got.append((msg.string_field, msg.int32_field))
+        msg = all_types_cls()
+    assert got == [("row-1", 1), ("row-2", 2)]
+    second = pxf.DatasetReader.from_bytes(first.tail())
+    assert second.columns == ("string_field", "enum_field")
+    msg = all_types_cls()
+    assert second.scan(msg)
+    assert (msg.string_field, msg.enum_field) == ("row-3", 1)
+    assert not second.scan(all_types_cls())

tests/test_pxf_decode_limits.py

@@ -211,16 +211,13 @@ def test_a_negative_limit_is_an_error(name):
 _BYTES_ROW = "@dataset lim.H (b)\n(" + _bytes_doc(192).split(" = ", 1)[1].strip() + ")\n"
 
 
-@pytest.mark.xfail(strict=True, reason="protowire-cpp#160: rows are read under the default "
-                   "MaxBytesLiteralLength")
 def test_dataset_reader_reads_rows_under_its_limits():
     reader = pxf.DatasetReader.from_bytes(_BYTES_ROW, max_bytes_literal_length=191)
     with pytest.raises(ValueError, match="MaxBytesLiteralLength=191"):
         reader.next_or_none()
 
 
 def test_scan_refuses_a_cell_past_the_readers_limit():
-    # Until protowire-cpp#160, the reader returns the row and binding refuses it.
     with pytest.raises(ValueError, match="MaxBytesLiteralLength=191"):
         pxf.DatasetReader.from_bytes(_BYTES_ROW, max_bytes_literal_length=191).scan(H())
     msg = H()
@@ -238,3 +235,39 @@ def test_scan_binds_under_the_readers_limits():
     msg = H()
     assert pxf.DatasetReader.from_bytes(data, max_numeric_literal_digits=5000).scan(msg)
     assert msg.HasField("i")
+
+
+def _wide_schema():
+    # Forty int32 fields with 200-character names: a row of forty 1s is
+    # 121 bytes, but the body it is bound as repeats every name, 8200 bytes.
+    fd = descriptor_pb2.FileDescriptorProto(name="wide.proto", package="w", syntax="proto3")
+    w = fd.message_type.add(name="W")
+    for i in range(40):
+        w.field.add(name=f"f{i:02d}_" + "x" * 196, number=i + 1, type=F.TYPE_INT32)
+    pool = descriptor_pool.DescriptorPool()
+    pool.Add(fd)
+    return message_factory.GetMessageClass(pool.FindMessageTypeByName("w.W"))
+
+
+W = _wide_schema()
+
+
+def test_scan_does_not_apply_max_message_size_to_the_bound_row():
+    # protowire-go binds a row under every limit but MaxMessageSize, which
+    # caps the bytes the reader holds instead. The input is 8457 bytes and
+    # passes MaxMessageSize=8192 (the reader holds at most two 4 KiB reads);
+    # each row's body is 8200 bytes and would not. Go reads all three rows.
+    names = [f.name for f in W.DESCRIPTOR.fields]
+    data = (f"@dataset w.W ({', '.join(names)})\n"
+            + f"({', '.join('1' for _ in names)})\n" * 3)
+    assert len(data) == 8457
+    reader = pxf.DatasetReader.from_bytes(data, max_message_size=8192)
+    rows = 0
+    msg = W()
+    while reader.scan(msg):
+        rows += 1
+        assert getattr(msg, names[-1]) == 1
+    assert rows == 3
+    # The reader's cap still holds: below one read's worth, the header fails.
+    with pytest.raises(ValueError, match="MaxMessageSize=4096"):
+        pxf.DatasetReader.from_bytes(data, max_message_size=4096)