diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6cf583e..8e6c341 100644 --- a/.github/workflows/ci.yml +++ b/.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: # --------------------------------------------------------------------- diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 2dc4aa7..31d75da 100644 --- a/.github/workflows/codeql.yml +++ b/.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: diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index ec3a77d..f7367e0 100644 --- a/.github/workflows/publish.yml +++ b/.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: # --------------------------------------------------------------------- diff --git a/CHANGELOG.md b/CHANGELOG.md index fe4ccf3..842e1f1 100644 --- a/CHANGELOG.md +++ b/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`. diff --git a/src/protowire/pxf.py b/src/protowire/pxf.py index 3bc8649..785c595 100644 --- a/src/protowire/pxf.py +++ b/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 diff --git a/tests/test_pxf_dataset_reader.py b/tests/test_pxf_dataset_reader.py index 6ef9920..d6c79a9 100644 --- a/tests/test_pxf_dataset_reader.py +++ b/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()) diff --git a/tests/test_pxf_decode_limits.py b/tests/test_pxf_decode_limits.py index b9ed908..a911c85 100644 --- a/tests/test_pxf_decode_limits.py +++ b/tests/test_pxf_decode_limits.py @@ -211,8 +211,6 @@ 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"): @@ -220,7 +218,6 @@ def test_dataset_reader_reads_rows_under_its_limits(): 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)