From cbc3f55f0db1486b97c901fa200e003d728d4b4f Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Sat, 7 Nov 2026 13:50:10 -0800 Subject: [PATCH] pxf: bind a dataset row through protowire-cpp's BindRow bind_row wrote each cell back as PXF text and decoded the body with unmarshal. That skipped (pxf.default) and (pxf.required), refused list and block cells and dotted columns, and gave errors at the body's positions. scan did the same on cells already turned into Python tuples. bind_row now rebuilds the row's AST cells in the bridge and calls pxf::BindRow. scan keeps the row native and binds it there, as DatasetReader::Scan does: under the reader's limits but max_message_size, with the schema checked once per type per reader. On all 43 fixtures in the spec's testdata/dataset/, scan gives protowire-go v1.17.0's message or error, text and position included (testdata/dataset-scan.go.txt). As in Go, decided on #24: - a row binds into what the message holds, which is no longer cleared; - skip_validate defaults to False for both, as Go and unmarshal check. Both take discard_unknown. Cells read with next_or_none still lose list and block values and precision suffixes; that is #47. Fixes #24 --- CHANGELOG.md | 20 ++ src/_protowire/module.cc | 145 +++++++++++++++ src/protowire/pxf.py | 152 +++++++-------- testdata/dataset-scan.go.txt | 119 ++++++++++++ tests/test_pxf_bind_row.py | 305 +++++++++++++++++++++++++++++++ tests/test_pxf_dataset_reader.py | 32 +++- 6 files changed, 696 insertions(+), 77 deletions(-) create mode 100644 testdata/dataset-scan.go.txt create mode 100644 tests/test_pxf_bind_row.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 842e1f1..2d6f444 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,26 @@ format changes. ### Changed +- **`pxf.bind_row` and `DatasetReader.scan` bind a row as protowire-go does** + (#24), through protowire-cpp's `BindRow` rather than writing the row out as + text and decoding it with `unmarshal`. On all 43 of the spec's + `testdata/dataset/` fixtures, `scan` now gives protowire-go v1.17.0's + message or error, text and position included. Before, most failed or bound + differently. What changes: + - An absent cell takes the field's `(pxf.default)`. An absent + `(pxf.required)` field is an error, as in `unmarshal_full`. + - A cell may hold a list or a block (`scan` only; see #47 for rows read as + cells), and a column may be a dotted path into a sub-message. + - **A row binds into what the message already holds,** which is no longer + cleared first. In a reused message, a field an earlier row set keeps its + value under an empty or `null` cell, as in Go. Bind each row into a new or + cleared message. + - **The schema is checked by default** (`skip_validate=False`), as Go and + `unmarshal` check it. `scan` checks it on the first row of each type; + `bind_row` on every call. Pass `skip_validate=True` for the old behaviour. + - A `scan` decode error carries the row's line and column in the input, + and names the column (`8:1: column int32_field: ...`). + - Both take `discard_unknown`. - **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). diff --git a/src/_protowire/module.cc b/src/_protowire/module.cc index a319fe8..097fe73 100644 --- a/src/_protowire/module.cc +++ b/src/_protowire/module.cc @@ -157,6 +157,105 @@ void ApplyLimits(const PxfLimits& l, protowire::pxf::UnmarshalOptions& opts) { opts.max_big_float_exponent = v[5]; } +// PyCellToValue is CellToPyTuple's inverse, for a cell pxf.py has already +// normalized: None for absent, else (kind, value) with value a str, a bool +// ("bool") or bytes ("bytes"). A value cell that crossed to Python as +// ("unknown", None), a list or a block, cannot come back. +std::optional PyCellToValue(nb::handle cell) { + using namespace protowire::pxf; + if (cell.is_none()) return std::nullopt; + auto t = nb::cast(cell); + if (t.size() != 2) throw nb::value_error("pxf.bind_row: a cell is None or a (kind, value) pair"); + const auto kind = nb::cast(t[0]); + nb::handle v = t[1]; + if (kind == "null") return ValuePtr(std::make_unique()); + if (kind == "string") { + auto p = std::make_unique(); + p->value = nb::cast(v); + return ValuePtr(std::move(p)); + } + if (kind == "int") { + auto p = std::make_unique(); + p->raw = nb::cast(v); + return ValuePtr(std::move(p)); + } + if (kind == "float") { + auto p = std::make_unique(); + p->raw = nb::cast(v); + return ValuePtr(std::move(p)); + } + if (kind == "bool") { + auto p = std::make_unique(); + p->value = nb::cast(v); + return ValuePtr(std::move(p)); + } + if (kind == "bytes") { + auto b = nb::cast(v); + auto p = std::make_unique(); + const auto* d = reinterpret_cast(b.c_str()); + p->value.assign(d, d + b.size()); + return ValuePtr(std::move(p)); + } + if (kind == "ident") { + auto p = std::make_unique(); + p->name = nb::cast(v); + return ValuePtr(std::move(p)); + } + if (kind == "timestamp") { + auto p = std::make_unique(); + p->raw = nb::cast(v); + return ValuePtr(std::move(p)); + } + if (kind == "duration") { + auto p = std::make_unique(); + p->raw = nb::cast(v); + return ValuePtr(std::move(p)); + } + throw nb::value_error(("pxf.bind_row: unknown cell kind '" + kind + "'").c_str()); +} + +// NewMessageFrom makes a message of desc holding msg_bytes. A row binds into +// what the message already holds, as protowire-go's BindRow and Scan do: they +// do not reset it first. +std::unique_ptr NewMessageFrom(const SchemaBundle& schema, + const pbuf::Descriptor* desc, + nb::bytes msg_bytes, const char* what) { + std::unique_ptr msg(schema.factory->GetPrototype(desc)->New()); + if (!msg->ParsePartialFromArray(msg_bytes.c_str(), static_cast(msg_bytes.size()))) { + throw nb::value_error((std::string(what) + ": proto parse failed").c_str()); + } + return msg; +} + +nb::bytes SerializeMessage(const pbuf::Message& msg, const char* what) { + std::string out; + if (!msg.SerializePartialToString(&out)) { + throw nb::value_error((std::string(what) + ": proto serialization failed").c_str()); + } + return nb::bytes(out.data(), out.size()); +} + +// pxf::BindRow under the caller's options. msg_bytes is what the message +// holds now; the result is what it holds after the row. +nb::bytes PxfBindRow(nb::bytes msg_bytes, const std::vector& columns, + nb::list cells, nb::bytes fds_bytes, const std::string& full_name, + bool discard_unknown, bool skip_validate, const PxfLimits& limits) { + protowire::pxf::UnmarshalOptions opts; + ApplyLimits(limits, opts); + opts.discard_unknown = discard_unknown; + opts.skip_validate = skip_validate; + protowire::pxf::DatasetRow row; + row.cells.reserve(cells.size()); + for (nb::handle c : cells) row.cells.push_back(PyCellToValue(c)); + auto schema = BuildSchema(std::string_view(fds_bytes.c_str(), fds_bytes.size())); + const auto* desc = FindDescriptor(schema, full_name); + auto msg = NewMessageFrom(schema, desc, msg_bytes, "pxf.bind_row"); + if (auto st = protowire::pxf::BindRow(msg.get(), columns, row, opts); !st.ok()) { + throw nb::value_error(("pxf.bind_row: " + st.ToString()).c_str()); + } + return SerializeMessage(*msg, "pxf.bind_row"); +} + // PXF text -> binary proto bytes. nb::bytes PxfUnmarshal(nb::bytes text, nb::bytes fds_bytes, const std::string& full_name, bool discard_unknown, @@ -311,6 +410,7 @@ class PyDatasetReader { throw nb::value_error(("pxf.DatasetReader: " + tr.status().ToString()).c_str()); } out->reader_ = std::move(*tr); + out->opts_ = opts; // Marshal the side-channel directives once at construction; they're // fixed for the reader's lifetime. for (const auto& d : out->reader_->Directives()) { @@ -352,6 +452,40 @@ class PyDatasetReader { return RowToList(row); } + // Reads the next row and binds it into a message holding msg_bytes, as + // pxf::DatasetReader::Scan does: under the reader's options but + // max_message_size, with discard_unknown and skip_validate from the call. + // The schema and the column paths are checked once per type. Returns the + // bound message's bytes, or None at EOF. + nb::object Scan(nb::bytes msg_bytes, nb::bytes fds_bytes, const std::string& full_name, + bool discard_unknown, bool skip_validate) { + if (reader_->Done()) return nb::none(); + protowire::pxf::DatasetRow row; + if (auto s = reader_->Next(&row); !s.ok()) { + throw nb::value_error(("pxf.DatasetReader.scan: " + s.ToString()).c_str()); + } + if (reader_->Done()) return nb::none(); + const std::string_view fds(fds_bytes.c_str(), fds_bytes.size()); + if (!schema_.pool || fds != fds_) { + schema_ = BuildSchema(fds); + fds_ = std::string(fds); + checked_type_ = nullptr; + } + const auto* desc = FindDescriptor(schema_, full_name); + auto msg = NewMessageFrom(schema_, desc, msg_bytes, "pxf.DatasetReader.scan"); + protowire::pxf::UnmarshalOptions opts = opts_; + // A row's bytes were capped as they were read; the body it is bound as + // repeats the column names, so it is wider than the row. + opts.max_message_size = 0; + opts.discard_unknown = discard_unknown; + opts.skip_validate = skip_validate || desc == checked_type_; + if (auto s = protowire::pxf::BindRow(msg.get(), reader_->Columns(), row, opts); !s.ok()) { + throw nb::value_error(("pxf.DatasetReader.scan: " + s.ToString()).c_str()); + } + if (!skip_validate) checked_type_ = desc; + return SerializeMessage(*msg, "pxf.DatasetReader.scan"); + } + // Drains the remaining buffered + underlying bytes. Only meaningful // after Done(); the Python wrapper exposes this as a method that // returns bytes so callers can chain a second DatasetReader on @@ -375,6 +509,12 @@ class PyDatasetReader { std::unique_ptr stream_; std::unique_ptr reader_; std::vector directives_; + protowire::pxf::UnmarshalOptions opts_; + // Scan's schema, rebuilt when the caller's descriptor set changes, and the + // type whose schema it has checked. + std::string fds_; + SchemaBundle schema_; + const pbuf::Descriptor* checked_type_ = nullptr; }; // Binary proto bytes -> PXF text. max_nesting_depth is @@ -591,6 +731,9 @@ NB_MODULE(_protowire, m) { m.def("pxf_marshal", &PxfMarshal, "msg_bytes"_a, "fds"_a, "full_name"_a, "max_nesting_depth"_a = 0); m.def("pxf_validate_descriptor", &PxfValidateDescriptor, "fds"_a, "full_name"_a); + m.def("pxf_bind_row", &PxfBindRow, "msg_bytes"_a, "columns"_a, "cells"_a, "fds"_a, + "full_name"_a, "discard_unknown"_a = false, "skip_validate"_a = false, + "limits"_a = no_limits); nb::class_(m, "PxfDatasetReader") .def_static("from_bytes", &PyDatasetReader::FromBytes, "data"_a, "limits"_a = no_limits) @@ -599,6 +742,8 @@ NB_MODULE(_protowire, m) { .def_prop_ro("directives", &PyDatasetReader::Directives) .def_prop_ro("done", &PyDatasetReader::Done) .def("next_or_none", &PyDatasetReader::NextOrNone) + .def("scan", &PyDatasetReader::Scan, "msg_bytes"_a, "fds"_a, "full_name"_a, + "discard_unknown"_a = false, "skip_validate"_a = false) .def("tail", &PyDatasetReader::Tail) .def("__iter__", [](PyDatasetReader& self) -> PyDatasetReader& { return self; }) .def("__next__", &PyDatasetReader::Next); diff --git a/src/protowire/pxf.py b/src/protowire/pxf.py index 785c595..57dac74 100644 --- a/src/protowire/pxf.py +++ b/src/protowire/pxf.py @@ -8,7 +8,6 @@ from __future__ import annotations -import base64 from dataclasses import dataclass, field from typing import Literal, Union @@ -443,13 +442,13 @@ class DatasetReader: file-like / chunked-IO bridge is a possible follow-up. """ - def __init__(self, native: object, limits: dict[str, int] | None = None) -> None: + def __init__(self, native: object) -> 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 (all but - # max_message_size). + # only; type/columns/directives are forwarded on demand. The native + # reader keeps the from_bytes limits, which scan binds each row under. self._native = native - self._limits = limits or {} + # scan's descriptor set, for the last message type it bound. + self._fds: tuple[object, bytes] | None = None @classmethod def from_bytes( @@ -468,16 +467,11 @@ def from_bytes( 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( - max_message_size=max_message_size, - max_nesting_depth=max_nesting_depth, - max_numeric_literal_digits=max_numeric_literal_digits, - max_bytes_literal_length=max_bytes_literal_length, - max_repeated_count=max_repeated_count, - max_big_float_exponent=max_big_float_exponent, - ) - native = _protowire.PxfDatasetReader.from_bytes(bytes(data), _limits(**limits)) - return cls(native, limits) + native = _protowire.PxfDatasetReader.from_bytes(bytes(data), _limits( + max_message_size, max_nesting_depth, max_numeric_literal_digits, + max_bytes_literal_length, max_repeated_count, max_big_float_exponent, + )) + return cls(native) @property def type(self) -> str: @@ -534,18 +528,36 @@ def tail(self) -> bytes: """ return self._native.tail() - def scan(self, msg: Message, *, skip_validate: bool = True) -> bool: - """Read the next row and bind its cells to `msg`'s fields by - column name. Returns True on success; returns False at EOF - (callers check `reader.done`).""" - cells = self.next_or_none() - if cells is None: + def scan( + self, msg: Message, *, skip_validate: bool = False, discard_unknown: bool = False, + ) -> bool: + """Read the next row and bind it to `msg`, as `bind_row` binds it. + Returns True on success; returns False at EOF (callers check + `reader.done`). + + This is protowire-cpp's `DatasetReader::Scan`, which binds a row as + protowire-go's `Scan` does: under the limits given to `from_bytes` + except `max_message_size`, which caps the bytes of a row as it is + read (the body a row is bound as repeats the column names, so it is + wider than the row). A decode error carries the row's position in + the input. Unless `skip_validate` is set, `msg`'s schema is checked + against the bind-time rules, as `unmarshal` checks it, on the first + row of its type rather than on every row. + + As with `bind_row`, the row binds into what `msg` already holds: pass + a new or cleared message for each row. + """ + desc = type(msg).DESCRIPTOR + if self._fds is None or self._fds[0] is not desc: + self._fds = (desc, fds_for_message(msg)) + raw = self._native.scan( + msg.SerializePartialToString(), self._fds[1], desc.full_name, + discard_unknown, skip_validate, + ) + if raw is None: return False - # 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) + msg.Clear() + msg.MergeFromString(raw) return True @@ -557,7 +569,8 @@ def bind_row( columns: tuple[str, ...] | list[str], row: tuple[Cell, ...] | list[Cell], *, - skip_validate: bool = True, + skip_validate: bool = False, + discard_unknown: bool = False, max_message_size: int = 0, max_nesting_depth: int = 0, max_numeric_literal_digits: int = 0, @@ -567,62 +580,57 @@ def bind_row( ) -> None: """Bind a `@dataset` row's cells to `msg`'s fields by column name. - `columns` and `row` MUST have the same length. Cell-state semantics: + This is protowire-cpp's `BindRow`, which binds a row as protowire-go's + does: as the body ` = ` for each non-empty cell, decoded + as `unmarshal_full` decodes it. A column may be a dotted path into a + sub-message. `columns` and `row` MUST have the same length. Cell-state + semantics: - `None` — field absent. (pxf.default) applies if declared; (pxf.required) errors if neither default nor value is present. - - `("null", _)` — field cleared (draft §3.9). + - `("null", _)` — field present-but-null (draft §3.9): it stays unset + and keeps (pxf.default) off. Only a field with + presence may be null. - any other — field set to the cell's value. - Strategy: render the row as a synthetic PXF body (` = ` - per non-None cell) and run it through `unmarshal`. This mirrors - `protowire-cpp`'s `BindRow` and reuses every branch of the existing - decoder — WKT timestamps / durations, wrapper-type nullability, - enum-by-name resolution, oneof handling — instead of growing a - parallel Cell→FieldDescriptor switch. - - `skip_validate` defaults to True: the descriptor was validated once - when the caller constructed the message factory, and re-running the - reserved-name check per row is wasteful in tight loops. The limits apply - to the synthetic body, as for `unmarshal`. + The row binds into what `msg` already holds, as in protowire-go: `msg` + is not cleared first. So bind each row into a new or cleared message. In + a reused one, a field an earlier row set keeps its value under an empty + or `null` cell. + + Unless `skip_validate` is set, `msg`'s schema is checked against the + bind-time rules on every call, as `unmarshal` checks it. A caller + binding many rows of one type can check once and pass + `skip_validate=True`; `DatasetReader.scan` does that itself. A decode + error is positioned in the body the row is bound as, since a row passed + as cells has no position in its input. The limits apply to the body, as + for `unmarshal`. """ - if len(columns) != len(row): - raise ValueError( - f"bind_row: {len(columns)} columns vs {len(row)} cells" - ) - parts: list[str] = [] - for col, cell in zip(columns, row): - if cell is None: - continue - parts.append(f"{col} = {_cell_to_pxf(cell)}\n") - unmarshal( - "".join(parts), msg, skip_validate=skip_validate, - max_message_size=max_message_size, - max_nesting_depth=max_nesting_depth, - max_numeric_literal_digits=max_numeric_literal_digits, - max_bytes_literal_length=max_bytes_literal_length, - max_repeated_count=max_repeated_count, - max_big_float_exponent=max_big_float_exponent, + raw = _protowire.pxf_bind_row( + msg.SerializePartialToString(), list(columns), [_native_cell(c) for c in row], + fds_for_message(msg), msg.DESCRIPTOR.full_name, discard_unknown, skip_validate, + _limits( + max_message_size, max_nesting_depth, max_numeric_literal_digits, + max_bytes_literal_length, max_repeated_count, max_big_float_exponent, + ), ) + msg.Clear() + msg.MergeFromString(raw) -def _cell_to_pxf(cell: tuple[CellKind, object]) -> str: +def _native_cell(cell: Cell | None): + """A cell in the shape the native bind_row takes: None, or (kind, value) + with value a str, or a bool for "bool" and bytes for "bytes".""" + if cell is None: + return None kind, value = cell if kind == "null": - return "null" - if kind == "string": - # Escape `\`, `"` and the line feed, which the lexer rejects raw in a - # string (protowire#350); protowire-cpp's BindRow writes it as `\n` - # too. Other characters round-trip verbatim. - s = str(value).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") - return f'"{s}"' - if kind in ("int", "float", "timestamp", "duration"): - return str(value) # raw text + return ("null", None) + if kind in ("string", "int", "float", "ident", "timestamp", "duration"): + return (kind, str(value)) if kind == "bool": - return "true" if value else "false" + return (kind, bool(value)) if kind == "bytes": - return 'b"' + base64.b64encode(value).decode("ascii") + '"' - if kind == "ident": - return str(value) + return (kind, bytes(value)) raise ValueError(f"bind_row: unknown cell kind {kind!r}") diff --git a/testdata/dataset-scan.go.txt b/testdata/dataset-scan.go.txt new file mode 100644 index 0000000..a94d082 --- /dev/null +++ b/testdata/dataset-scan.go.txt @@ -0,0 +1,119 @@ +== arity-long.pxf + row 1 err: 6:1: @dataset row has 3 cells, expected 2 (column count) +== arity-short.pxf + row 1 err: 6:1: @dataset row has 2 cells, expected 3 (column count) +== block-cell.pxf + row 1: 0a05726f772d315a090a05696e6e65721063 + table test.v1.AllTypes: 1 rows +== block-on-scalar.pxf + row 1 err: 7:1: column int32_field: expected integer for field "int32_field" +== cell-framing.pxf + row 1: 0a05726f772d315a0b0a0761292c2028622c10016202292c6202282c + row 2: 0a05726f772d325a0e0a0a7829202279222c20287a10026203656e64 + table test.v1.AllTypes: 2 rows +== dotted-bad-header-empty-row.pxf + row 1 err: 8:1: @dataset column "picks.name": "picks" is a repeated field, and a column path cannot go through one; use a value cell for "picks" (draft §3.4.4) +== dotted-bad-header-no-rows.pxf + table dataset.v1.Event: 0 rows +== dotted-column.pxf + row 1: 0a05726f772d315a070a05696e6e6572 + table test.v1.AllTypes: 1 rows +== dotted-empty-segment.pxf + header err: 4:32: @dataset column "desc..energy": a column path has an empty segment (draft §3.4.4) +== dotted-grouping.pxf + row 1: 0a027231121209000000000000e03f1204083c10781a01784801 + row 2: 0a027232120909000000000000d03f4802 + row 3: 0a0272334803 + row 4: 0a0272341204120210094804 + table dataset.v1.Event: 4 rows +== dotted-into-any.pxf + row 1: 0a01786a2a0a23747970652e676f6f676c65617069732e636f6d2f646174617365742e76312e5069636b12030a0178 + table dataset.v1.Event: 1 rows +== dotted-into-bignum.pxf + row 1 err: 7:1: @dataset column "big.negative": "big" is a pxf.BigInt, which is written as a literal; a column path cannot go into it (draft §3.4.4) +== dotted-into-duration.pxf + row 1 err: 7:1: @dataset column "dur.seconds": "dur" is a google.protobuf.Duration, which is written as a literal; a column path cannot go into it (draft §3.4.4) +== dotted-into-secret.pxf + row 1: 0a017872150a06733363726574120b64622070617373776f7264 + table dataset.v1.Event: 1 rows +== dotted-into-timestamp.pxf + row 1 err: 7:1: @dataset column "ts.seconds": "ts" is a google.protobuf.Timestamp, which is written as a literal; a column path cannot go into it (draft §3.4.4) +== dotted-into-wrapper.pxf + row 1 err: 7:1: @dataset column "wrap.value": "wrap" is a google.protobuf.Int32Value, which is written as a literal; a column path cannot go into it (draft §3.4.4) +== dotted-null-leaf.pxf + row 1: 0a016e1200 + table dataset.v1.Event: 1 rows +== dotted-oneof-two-members.pxf + row 1 err: 7:1: column b.name: oneof "choice": field "b" conflicts with already-set field "a" +== dotted-oneof.pxf + row 1: 32050a01781001 + row 2: 3a030a0179 + table dataset.v1.Event: 2 rows +== dotted-overlap-deep.pxf + header err: 6:44: @dataset column "desc.tempo.min_bpm" overlaps column "desc.tempo": a field and a path inside it cannot both be columns (draft §3.4.4) +== dotted-overlap.pxf + header err: 5:38: @dataset column "desc.energy" overlaps column "desc": a field and a path inside it cannot both be columns (draft §3.4.4) +== dotted-required-missing.pxf + row 1 err: 7:1: required field "req.n" is absent +== dotted-required.pxf + row 1: 0a06616273656e74 + row 2: 0a0373657442080805120464666c74 + table dataset.v1.Event: 2 rows +== dotted-through-map.pxf + row 1 err: 6:1: @dataset column "by_name.name": "by_name" is a map field, and a column path cannot go through one; use a value cell for "by_name" (draft §3.4.4) +== dotted-through-repeated.pxf + row 1 err: 7:1: @dataset column "picks.name": "picks" is a repeated field, and a column path cannot go through one; use a value cell for "picks" (draft §3.4.4) +== dotted-through-scalar.pxf + row 1 err: 7:1: @dataset column "plain.x": "plain" is not a message field +== dotted-trailing-dot.pxf + header err: 4:32: @dataset column "desc.": a column path has an empty segment (draft §3.4.4) +== dotted-unknown-segment.pxf + row 1 err: 6:1: @dataset column "desc.nope": dataset.v1.Desc has no field "nope" +== dotted-with-value-cells.pxf + row 1: 0a016d121309000000000000e03f1202083c1a01611a01621a030a0170 + table dataset.v1.Event: 1 rows +== duplicate-column.pxf + header err: 4:42: @dataset column "string_field" appears twice (draft §3.4.4) +== empty-rows.pxf + table test.v1.AllTypes: 0 rows +== junk-between-rows.pxf + row 1: 0a05726f772d311001 + row 2 err: 8:1: @dataset directive cannot coexist with top-level field entries; the document's payload is the @dataset rows (draft §3.4.4) +== list-cell.pxf + row 1: 0a05726f772d316205616c706861620462657461 + table test.v1.AllTypes: 1 rows +== nested-cells.pxf + row 1: 0a05726f772d3110015a060a026e3110016201616201626a030a01786a050a0179100272060a016b120176 + row 2: 0a05726f772d3210025a00 + row 3: 0a05726f772d331003 + row 4: 0a05726f772d341004 + table test.v1.AllTypes: 4 rows +== null-on-optional-cell.pxf + row 1: 0a01617801 + row 2: 0a0162 + table dataset.v1.Event: 2 rows +== null-on-required-cell.pxf + row 1: 0a01611001 + row 2 err: 9:1: column o: null is not allowed on required field "o": a required field holds a value +== null-on-scalar-cell.pxf + row 1: 0a01611001 + row 2 err: 8:1: column int32_field: null is not allowed on scalar field "int32_field": declare it optional, or as a wrapper type (google.protobuf.Int32Value or similar), to make it nullable +== rows-end-at-directive.pxf + row 1: 0a05726f772d311001 + row 2: 0a05726f772d321002 + table test.v1.AllTypes: 2 rows + row 1: 0a05726f772d335001 + table test.v1.AllTypes: 1 rows +== string-newline-in-cell.pxf + row 1 err: 7:2: newline in string literal: write it as \n, or use a """triple-quoted""" string +== truncated-row-eof.pxf + row 1: 0a05726f772d311001 + row 2 err: 10:1: expected ',' or ')' in @dataset row, got EOF +== truncated-row-then-rows.pxf + row 1: 0a05726f772d311001 + row 2 err: 9:1: expected ',' or ')' in @dataset row, got ( +== with-body.pxf + row 1: 0a05726f772d311001 + row 2 err: 9:1: @dataset directive cannot coexist with top-level field entries; the document's payload is the @dataset rows (draft §3.4.4) +== with-type.pxf + header err: 6:1: @dataset directive cannot coexist with @type; the @dataset header declares the document's type (draft §3.4.4) diff --git a/tests/test_pxf_bind_row.py b/tests/test_pxf_bind_row.py new file mode 100644 index 0000000..87ac797 --- /dev/null +++ b/tests/test_pxf_bind_row.py @@ -0,0 +1,305 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""pxf.bind_row and DatasetReader.scan bind a row through protowire-cpp's +BindRow, as protowire-go binds it (#24). + +Expected messages and errors are protowire-go v1.17.0's, from its +DatasetReader.Scan on the same input, each message as its deterministic +binary encoding. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +import pytest +from google.protobuf import descriptor_pb2, descriptor_pool, message_factory + +from protowire import pxf + +# --- value cells and column paths: spec fixtures (testdata/dataset/) ---------- +# +# Copied verbatim from trendvidia/protowire (CI has no checkout of it). Built +# on the earlier bind_row, which wrote each cell back as text, every one of +# these but null-on-scalar-cell failed: a list or block cell was refused as +# kind 'unknown', and a dotted column as an unknown field. + +_FIXTURES = { + "list-cell": ( + '# §3.4.4 value cells. A list in a cell on a repeated column. MUST be\n' + '# accepted: a cell means what `repeated_string = ["alpha", "beta"]` means\n' + '# in a body. Rejected before v1.13, when cells were scalar-only.\n' + '\n' + '@dataset test.v1.AllTypes (string_field, repeated_string)\n' + '\n' + '("row-1", ["alpha", "beta"])\n' + ), + "block-cell": ( + '# §3.4.4 value cells. A block in a cell on a message column. MUST be\n' + '# accepted: a cell means what `nested_field = { name = "inner" value = 99 }`\n' + '# means in a body. Rejected before v1.13, when cells were scalar-only.\n' + '\n' + '@dataset test.v1.AllTypes (string_field, nested_field)\n' + '\n' + '("row-1", { name = "inner" value = 99 })\n' + ), + "nested-cells": ( + '# §3.4.4 value cells. MUST be accepted. Each row binds to the message its\n' + '# body would: a non-empty cell in column c means `c = `.\n' + '#\n' + '# row-1: nested_field {name "n1" value 1}; repeated_string [a, b];\n' + '# repeated_nested [{x}, {y 2}]; string_map {k: v}; int32_field 1\n' + '# row-2: nested_field present and empty ({}); repeated_string empty ([]);\n' + '# repeated_nested and string_map absent (empty cells); int32_field 2\n' + '# row-3: every nested column absent; int32_field 3\n' + '# row-4: nested_field null (present-but-null, as `nested_field = null`\n' + '# is in a body); the other nested columns absent; int32_field 4\n' + '\n' + '@dataset test.v1.AllTypes (string_field, nested_field, repeated_string, repeated_nested, string_map, int32_field)\n' + '\n' + '("row-1", { name = "n1" value = 1 }, ["a", "b"], [{ name = "x" }, { name = "y" value = 2 }], { "k": "v" }, 1)\n' + '("row-2", {}, [], , , 2)\n' + '("row-3", , , , , 3)\n' + '("row-4", null, , , , 4)\n' + ), + "cell-framing": ( + '# §3.4.4 value cells. Row framing. MUST be accepted: two rows. A ")" or ","\n' + '# inside a string, a triple-quoted string or a comment within a block cell\n' + '# neither ends the row nor splits the cell. A streaming reader finds the\n' + '# row by matching parentheses outside strings and comments, so it MUST\n' + '# yield the same two rows as a materializing parse.\n' + '\n' + '@dataset test.v1.AllTypes (string_field, nested_field, repeated_string)\n' + '\n' + '("row-1", { name = "a), (b," value = 1 }, ["),", "(,"])\n' + '("row-2", { name = """x) "y", (z""" /* ), ( */ value = 2 }, [ # ) ,\n' + ' "end" ])\n' + ), + "dotted-column": ( + '# §3.4.4 column paths. A column may be a dotted path into a sub-message.\n' + '# MUST be accepted. The row binds as\n' + '# string_field = "row-1" nested_field { name = "inner" }\n' + '# Before v1.13 a dotted column was rejected.\n' + '\n' + '@dataset test.v1.AllTypes (string_field, nested_field.name)\n' + '\n' + '("row-1", "inner")\n' + ), + "block-on-scalar": ( + '# §3.4.4 value cells. A block in a cell on a scalar column. MUST be\n' + '# rejected, binding the row: the cell means `int32_field = { value = 1 }`,\n' + '# and a block on a scalar field is an error in a body too.\n' + '\n' + '@dataset test.v1.AllTypes (string_field, int32_field)\n' + '\n' + '("row-1", { value = 1 })\n' + ), + "null-on-scalar-cell": ( + '# §3.4.4 + § Booleans, Null, and Identifier Values (#339). Row type\n' + '# test.v1.AllTypes. MUST be rejected: a cell applies the singular-field\n' + '# null rules, and int32_field is a plain scalar.\n' + '\n' + '@dataset test.v1.AllTypes (string_field, int32_field)\n' + '\n' + '("a", 1)\n' + '("b", null)\n' + ), +} + +_ACCEPT = { + "list-cell": ["0a05726f772d316205616c706861620462657461"], + "block-cell": ["0a05726f772d315a090a05696e6e65721063"], + "nested-cells": [ + "0a05726f772d3110015a060a026e3110016201616201626a030a01786a050a0179100272060a016b120176", + "0a05726f772d3210025a00", + "0a05726f772d331003", + "0a05726f772d341004", + ], + "cell-framing": [ + "0a05726f772d315a0b0a0761292c2028622c10016202292c6202282c", + "0a05726f772d325a0e0a0a7829202279222c20287a10026203656e64", + ], + "dotted-column": ["0a05726f772d315a070a05696e6e6572"], +} + +_REJECT = { + "block-on-scalar": ([], '7:1: column int32_field: expected integer for field "int32_field"'), + "null-on-scalar-cell": ( + ["0a01611001"], + '8:1: column int32_field: null is not allowed on scalar field "int32_field": declare it ' + "optional, or as a wrapper type (google.protobuf.Int32Value or similar), to make it " + "nullable", + ), +} + + +@pytest.mark.parametrize("fixture", sorted(_ACCEPT)) +def test_scan_binds_value_cells_and_paths_as_go(all_types_cls, fixture): + reader = pxf.DatasetReader.from_bytes(_FIXTURES[fixture]) + got = [] + msg = all_types_cls() + while reader.scan(msg): + got.append(msg.SerializeToString(deterministic=True).hex()) + msg = all_types_cls() + assert got == _ACCEPT[fixture] + + +@pytest.mark.parametrize("fixture", sorted(_REJECT)) +def test_scan_refuses_a_row_at_its_position(all_types_cls, fixture): + rows, error = _REJECT[fixture] + reader = pxf.DatasetReader.from_bytes(_FIXTURES[fixture]) + for want in rows: + msg = all_types_cls() + assert reader.scan(msg) + assert msg.SerializeToString(deterministic=True).hex() == want + with pytest.raises(ValueError, match=re.escape(error)): + reader.scan(all_types_cls()) + + +def test_bind_row_takes_a_dotted_column(all_types_cls): + msg = all_types_cls() + pxf.bind_row(msg, ("string_field", "nested_field.name"), + (("string", "row-1"), ("string", "inner"))) + assert msg.SerializeToString(deterministic=True).hex() == _ACCEPT["dotted-column"][0] + + +# --- (pxf.required) and (pxf.default) on an absent cell (#24) ----------------- +# +# The schema is built from raw option bytes at the registered numbers, as in +# test_pxf_required_default.py. Go v1.17.0 streams this input as: row 1 +# f "a", g "x" (the default), h "z"; row 2 `3:1: required field "f" is absent`. + +_REQUIRED, _DEFAULT = 1314, 1315 +_STRING = descriptor_pb2.FieldDescriptorProto.TYPE_STRING + + +def _varint(n: int) -> bytes: + out = bytearray() + while n > 0x7F: + out.append(n & 0x7F | 0x80) + n >>= 7 + out.append(n) + return bytes(out) + + +def _schema(*fields): + fd = descriptor_pb2.FileDescriptorProto(name="rd.proto", package="t", syntax="proto3") + m = fd.message_type.add(name="M") + for number, (name, options) in enumerate(fields, 1): + f = m.field.add(name=name, number=number, type=_STRING) + if options: + f.options.MergeFromString(options) + pool = descriptor_pool.DescriptorPool() + pool.Add(fd) + return message_factory.GetMessageClass(pool.FindMessageTypeByName("t.M")) + + +REQUIRED = _varint(_REQUIRED << 3) + _varint(1) +DEFAULT = _varint(_DEFAULT << 3 | 2) + _varint(1) + b"x" +M = _schema(("f", REQUIRED), ("g", DEFAULT), ("h", b"")) +_ROWS = '@dataset t.M (f, g, h)\n("a", , "z")\n( , , "z")\n' + + +def test_an_absent_cell_takes_its_default(): + msg = M() + pxf.bind_row(msg, ("f", "g", "h"), (("string", "a"), None, ("string", "z"))) + assert (msg.f, msg.g, msg.h) == ("a", "x", "z") + msg = M() + assert pxf.DatasetReader.from_bytes(_ROWS).scan(msg) + assert msg.SerializeToString(deterministic=True).hex() == "0a01611201781a017a" + + +def test_an_absent_required_cell_is_an_error(): + with pytest.raises(ValueError, match='required field "f" is absent'): + pxf.bind_row(M(), ("f", "g", "h"), (None, None, ("string", "z"))) + reader = pxf.DatasetReader.from_bytes(_ROWS) + assert reader.scan(M()) + with pytest.raises(ValueError, match=re.escape('3:1: required field "f" is absent')): + reader.scan(M()) + + +def test_the_schema_is_checked_unless_skip_validate(): + # (pxf.required) with (pxf.default) on one field breaks a bind-time rule. + # Go validates by default, and so do bind_row and scan now. + bad = _schema(("f", REQUIRED + DEFAULT)) + row = (("string", "y"),) + with pytest.raises(ValueError, match="PXF schema violations"): + pxf.bind_row(bad(), ("f",), row) + with pytest.raises(ValueError, match="PXF schema violations"): + pxf.DatasetReader.from_bytes('@dataset t.M (f)\n("y")\n').scan(bad()) + msg = bad() + pxf.bind_row(msg, ("f",), row, skip_validate=True) + assert msg.f == "y" + + +def test_discard_unknown_reaches_the_row(all_types_cls): + # An unknown field inside a block cell. Go v1.17.0: refused as + # `2:1: column nested_field: unknown field "nope" in test.v1.Nested`, + # and with DiscardUnknown bound as below. + src = '@dataset test.v1.AllTypes (string_field, nested_field)\n("a", { name = "n" nope = 1 })\n' + with pytest.raises(ValueError, match=re.escape( + '2:1: column nested_field: unknown field "nope" in test.v1.Nested')): + pxf.DatasetReader.from_bytes(src).scan(all_types_cls()) + msg = all_types_cls() + assert pxf.DatasetReader.from_bytes(src).scan(msg, discard_unknown=True) + assert msg.SerializeToString(deterministic=True).hex() == "0a01615a030a016e" + +# --- every spec dataset fixture, against protowire-go ------------------------- +# +# testdata/dataset-scan.go.txt is what protowire-go v1.17.0 gives for each +# fixture in the spec's testdata/dataset/ (protowire dd8e221), streamed through +# DatasetReader.Scan into a fresh message per row: each row's message as hex, +# or the error. This runs where the spec repository is checked out beside this +# one; CI has no checkout, and the tests above copy the fixtures they need. + +SPEC = Path(__file__).parent.parent / ".." / "protowire" +GOLDEN = Path(__file__).parent.parent / "testdata" / "dataset-scan.go.txt" + + +def _stream_all(pool) -> str: + out = [] + for p in sorted((SPEC / "testdata" / "dataset").glob("*.pxf")): + out.append(f"== {p.name}") + data = p.read_bytes() + while True: + try: + reader = pxf.DatasetReader.from_bytes(data) + except ValueError as e: + out.append(f" header err: {str(e).removeprefix('pxf.DatasetReader: ')}") + break + cls = message_factory.GetMessageClass(pool.FindMessageTypeByName(reader.type)) + n, failed = 0, False + while True: + msg = cls() + try: + if not reader.scan(msg): + break + except ValueError as e: + out.append(f" row {n + 1} err: {str(e).removeprefix('pxf.DatasetReader.scan: ')}") + failed = True + break + n += 1 + out.append(f" row {n}: {msg.SerializeToString(deterministic=True).hex()}") + if failed: + break + out.append(f" table {reader.type}: {n} rows") + data = reader.tail() + if not data.strip(): + break + return "\n".join(out) + "\n" + + +def test_every_spec_dataset_fixture_streams_as_go(): + if not (SPEC / "testdata" / "dataset").is_dir(): + pytest.skip(f"spec repo testdata not found at {SPEC}") + from conftest import _protoc_to_fds + + fds = descriptor_pb2.FileDescriptorSet.FromString(_protoc_to_fds( + ["test.proto", "paths.proto"], + [str(SPEC / "testdata"), str(SPEC / "testdata" / "dataset"), str(SPEC / "proto")], + )) + pool = descriptor_pool.DescriptorPool() + for f in fds.file: + pool.Add(f) + assert _stream_all(pool) == GOLDEN.read_text() diff --git a/tests/test_pxf_dataset_reader.py b/tests/test_pxf_dataset_reader.py index d6c79a9..680ce80 100644 --- a/tests/test_pxf_dataset_reader.py +++ b/tests/test_pxf_dataset_reader.py @@ -192,19 +192,41 @@ def test_bind_row_absent_cell_leaves_default(all_types_cls): assert msg.int32_field == 7 -def test_bind_row_null_clears_wrapper(all_types_cls): - # A `null` cell on a wrapper field clears it (draft §3.9). +def test_bind_row_null_leaves_a_wrapper_unset(all_types_cls): + # A `null` cell on a wrapper field leaves it unset and present-but-null + # (draft §3.9). src = b'@dataset test.v1.AllTypes ( nullable_string )\n( null )\n' tr = pxf.DatasetReader.from_bytes(src) (row,) = list(tr) msg = all_types_cls() - msg.nullable_string.value = "stale" # populate to confirm clear - assert msg.HasField("nullable_string") pxf.bind_row(msg, tr.columns, row) - # nullable_string is a StringValue — after `null`, HasField is False. assert not msg.HasField("nullable_string") +def test_a_row_binds_into_what_the_message_holds(all_types_cls): + # protowire-go does not clear the message before a row, so in a reused + # message a field set by an earlier row keeps its value under an empty or + # null cell. Measured with protowire-go v1.17.0's Scan on this input: row + # 2 kept int32_field 1 and nullable_string "x". + src = ('@dataset test.v1.AllTypes (string_field, nullable_string, int32_field)\n' + '("a", "x", 1)\n("b", null, )\n') + reader = pxf.DatasetReader.from_bytes(src) + msg = all_types_cls() + assert reader.scan(msg) + assert reader.scan(msg) + assert (msg.string_field, msg.int32_field, msg.nullable_string.value) == ("b", 1, "x") + # A fresh message per row gives each row alone. + reader = pxf.DatasetReader.from_bytes(src) + assert reader.scan(all_types_cls()) + fresh = all_types_cls() + assert reader.scan(fresh) + assert (fresh.string_field, fresh.int32_field) == ("b", 0) + assert not fresh.HasField("nullable_string") + # bind_row binds the same way. + msg = all_types_cls(int32_field=7) + pxf.bind_row(msg, ("string_field", "int32_field"), (("string", "c"), None)) + assert (msg.string_field, msg.int32_field) == ("c", 7) + def test_bind_row_bytes_cell(all_types_cls): src = b'@dataset test.v1.AllTypes ( bytes_field )\n( b"YWJj" )\n' # "abc" tr = pxf.DatasetReader.from_bytes(src)