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)