From 3f16d46ccb4b6f3c26a539e521ec1ee4cce936d3 Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Fri, 6 Nov 2026 18:06:10 -0800 Subject: [PATCH] pxf: DatasetReader positions count from the start of the stream (#161) A row was parsed by a fresh lexer starting at 1:1, so a row's pos, its cells' positions and every error from Next or Scan counted from the start of that row. protowire-go starts the row's lexer at the row's place in the stream (newParserAt with posAt(start)) and reports positions against the whole stream. The reader now keeps base_, the stream position of pending_[0], and moves it past each byte it consumes. A row is parsed by a lexer that starts at PosAt(start). pxf::Lexer gains a constructor taking that start: line and column begin there and offsets add its offset. Input(), Mark and RepositionTo still index the piece itself, and RepositionTo going back returns to the start rather than to 1:1. The test replays nine inputs and expects protowire-go 903b4a7's output line for line: the first row, a later row, rows on the header's line, a row's second line, an arity error, a bad token, a row 18000 bytes and several reads in, and two Scan errors. Eight of the nine fail without the fix; the row's second line matched already, since that row starts on line 1. --- CHANGELOG.md | 1 + include/protowire/pxf/dataset_reader.h | 4 + include/protowire/pxf/lexer.h | 17 +++- include/protowire/pxf/token.h | 8 +- src/pxf/dataset_reader.cc | 29 ++++++- src/pxf/parser.cc | 12 ++- src/pxf/parser_internal.h | 10 +-- test/pxf_dataset_reader_test.cc | 111 +++++++++++++++++++++++++ 8 files changed, 170 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 97a4624..96183dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ number says nothing about any other port. See the Releases section of ### Fixed - **`DatasetReader` reads a row under the caller's `max_bytes_literal_length`** (#160), as protowire-go does. A row's cells were parsed under the default `kMaxBytesLiteralLength` whatever the options said. A caller who lowered the limit still got a row whose `b"…"` cell was past it, and a caller who raised it had that cell refused. `0` still means the default. Only the nesting depth used to reach the row parser. The header (a `@header` block and the `@dataset` line) was parsed under the defaults for both limits. It is now parsed under the caller's `max_nesting_depth` and `max_bytes_literal_length`, so `Create` refuses such a header as protowire-go's `NewDatasetReader` does, with the same text and position. +- **`DatasetReader` positions count from the start of the stream** (#161), as protowire-go's do. A row's `pos`, its cells' positions and the line and column of an error from `Next` or `Scan` counted from the start of that row, so every row was at `1:1`. A refused row in a long stream now names its own line. `pxf::Lexer` gains a constructor that starts at a given `Position`, which the reader uses for each row. ## [2.0.0] — 2026-11-05 diff --git a/include/protowire/pxf/dataset_reader.h b/include/protowire/pxf/dataset_reader.h index 37ef919..3d24355 100644 --- a/include/protowire/pxf/dataset_reader.h +++ b/include/protowire/pxf/dataset_reader.h @@ -118,7 +118,11 @@ class DatasetReader { Status Pull(size_t n); std::istream* src_ = nullptr; + Position PosAt(size_t i) const; + void Consume(size_t n); + std::string pending_; // bytes pulled from src_ but not yet consumed + Position base_; // where pending_[0] is in the stream bool src_eof_ = false; bool finished_ = false; Status err_; diff --git a/include/protowire/pxf/lexer.h b/include/protowire/pxf/lexer.h index 7df9f61..de51d1c 100644 --- a/include/protowire/pxf/lexer.h +++ b/include/protowire/pxf/lexer.h @@ -17,6 +17,14 @@ class Lexer { // The input must outlive the lexer; tokens reference into it. explicit Lexer(std::string_view input) : input_(input) {} + // A lexer over a piece of a larger document that starts at `start` in it, + // as a dataset row does in its stream. Positions count from `start`: the + // first byte is at start.line and start.column, and offsets add + // start.offset. Input(), Mark and RepositionTo still index the piece + // itself. + Lexer(std::string_view input, Position start) + : input_(input), line_(start.line), column_(start.column), start_(start) {} + Token Next(); // Bounds the decoded length of a b"…" literal (HARDENING.md @@ -46,7 +54,9 @@ class Lexer { column_ = m.column; } - Position CurrentPos() const { return Position{line_, column_, static_cast(pos_)}; } + Position CurrentPos() const { + return Position{line_, column_, start_.offset + static_cast(pos_)}; + } // Raw input view — used by parseDirective to slice the body bytes // between '{' and '}' once the matching brace has been located. @@ -59,8 +69,8 @@ class Lexer { void RepositionTo(int target) { if (target < static_cast(pos_)) { pos_ = 0; - line_ = 1; - column_ = 1; + line_ = start_.line; + column_ = start_.column; } while (static_cast(pos_) < target && pos_ < input_.size()) { uint8_t ch = static_cast(input_[pos_++]); @@ -109,6 +119,7 @@ class Lexer { size_t pos_ = 0; int line_ = 1; int column_ = 1; + Position start_{1, 1, 0}; // where input_[0] is in the document int max_bytes_literal_ = kMaxBytesLiteralLength; std::deque owned_; // pointer-stable storage std::vector* errors_ = nullptr; diff --git a/include/protowire/pxf/token.h b/include/protowire/pxf/token.h index 4ae6583..726f645 100644 --- a/include/protowire/pxf/token.h +++ b/include/protowire/pxf/token.h @@ -45,10 +45,10 @@ const char* TokenKindName(TokenKind k); struct Position { int line = 1; int column = 1; - // Byte offset into the lexer's input. Used by directive Body - // extraction to slice the raw bytes between '{' and '}'; line/column - // remain the primary user-facing identifier. Zero is the start of - // input. + // Byte offset into the document. Used by directive Body extraction to + // slice the raw bytes between '{' and '}'; line/column remain the + // primary user-facing identifier. Zero is the start of input, and a + // DatasetReader's rows count from the start of its stream. int offset = 0; std::string ToString() const { return std::to_string(line) + ":" + std::to_string(column); } }; diff --git a/src/pxf/dataset_reader.cc b/src/pxf/dataset_reader.cc index 62f536d..455717e 100644 --- a/src/pxf/dataset_reader.cc +++ b/src/pxf/dataset_reader.cc @@ -476,10 +476,11 @@ Status DatasetReader::Next(DatasetRow* out) { static_cast(columns_.size()), opts_.max_nesting_depth > 0 ? opts_.max_nesting_depth : kMaxNestingDepth, opts_.max_bytes_literal_length > 0 ? opts_.max_bytes_literal_length - : kMaxBytesLiteralLength); + : kMaxBytesLiteralLength, + PosAt(static_cast(start))); // Advance past the consumed bytes whether parse succeeded or not // — on failure we don't want to retry the same bad row forever. - pending_.erase(0, static_cast(end) + 1); + Consume(static_cast(end) + 1); if (!parsed.ok()) { err_ = parsed.status(); return err_; @@ -538,6 +539,28 @@ std::unique_ptr DatasetReader::Tail() { // ---- DatasetReader internals ------------------------------------------------ +// PosAt is the stream position of pending_[i], counted as the lexer counts: +// a byte is a column, and a newline starts the next line. +Position DatasetReader::PosAt(size_t i) const { + Position p = base_; + for (size_t k = 0; k < i; ++k) { + if (pending_[k] == '\n') { + ++p.line; + p.column = 1; + } else { + ++p.column; + } + } + p.offset += static_cast(i); + return p; +} + +// Consume drops the first n pending bytes, moving base_ past them. +void DatasetReader::Consume(size_t n) { + base_ = PosAt(n); + pending_.erase(0, n); +} + Status DatasetReader::Pull(size_t n) { if (src_eof_) return Status::OK(); std::string buf(n, '\0'); @@ -575,7 +598,7 @@ Status DatasetReader::ReadHeader() { type_ = std::move(doc->datasets[0].type); columns_ = std::move(doc->datasets[0].columns); directives_ = std::move(doc->directives); - pending_.erase(0, static_cast(header_end) + 1); + Consume(static_cast(header_end) + 1); return Status::OK(); } if (src_eof_) { diff --git a/src/pxf/parser.cc b/src/pxf/parser.cc index d506a6a..6277e20 100644 --- a/src/pxf/parser.cc +++ b/src/pxf/parser.cc @@ -43,8 +43,8 @@ std::string WrittenKey(const std::string& key, std::string_view prec) { // nearest entry or block boundary. Strict mode returns the first error. class Parser { public: - Parser(std::string_view input, int max_depth, int max_bytes_literal) - : lex_(input), max_depth_(max_depth) { + Parser(std::string_view input, int max_depth, int max_bytes_literal, Position start = {}) + : lex_(input, start), max_depth_(max_depth) { lex_.SetMaxBytesLiteral(max_bytes_literal); Advance(); } @@ -1153,11 +1153,9 @@ StatusOr ParseDatasetAt(Lexer* lex, Token* current, int max_de return tbl; } -StatusOr ParseDatasetRow(std::string_view row, - int expected, - int max_depth, - int max_bytes_literal) { - Parser p(row, max_depth, max_bytes_literal); +StatusOr ParseDatasetRow( + std::string_view row, int expected, int max_depth, int max_bytes_literal, Position start) { + Parser p(row, max_depth, max_bytes_literal, start); int end = 0; return p.ParseTableRow(expected, &end); } diff --git a/src/pxf/parser_internal.h b/src/pxf/parser_internal.h index 82afcf9..a457f83 100644 --- a/src/pxf/parser_internal.h +++ b/src/pxf/parser_internal.h @@ -21,11 +21,11 @@ namespace protowire::pxf { StatusOr ParseDatasetAt(Lexer* lex, Token* current, int max_depth); // ParseDatasetRow parses one `( cell, ... )` of expected cells, under the -// caller's nesting and bytes-literal limits. -StatusOr ParseDatasetRow(std::string_view row, - int expected, - int max_depth, - int max_bytes_literal); +// caller's nesting and bytes-literal limits. The row starts at `start` in +// its stream, and the row, its cells and its errors carry positions in the +// stream, as protowire-go's do (#161). +StatusOr ParseDatasetRow( + std::string_view row, int expected, int max_depth, int max_bytes_literal, Position start); // CheckDescriptorSchema runs ValidateDescriptor and joins any violations // into one error, the bind-time check Unmarshal and DatasetReader share. diff --git a/test/pxf_dataset_reader_test.cc b/test/pxf_dataset_reader_test.cc index 08dff5e..c6056ce 100644 --- a/test/pxf_dataset_reader_test.cc +++ b/test/pxf_dataset_reader_test.cc @@ -15,6 +15,7 @@ #include #include +#include #include namespace { @@ -370,4 +371,114 @@ TEST_F(PxfTableReader, BindRowUnknownColumnErrors) { EXPECT_FALSE(s.ok()); } +// ---- Positions count from the start of the stream (#161) -------------------- + +// Each case reads rows until an error and writes one line per row as the Go +// probe in #161's PR does: a row's position and its first cell's span, or the +// error. The expected lines are protowire-go 903b4a7's output for the same +// inputs, under MaxBytesLiteralLength=191. The first `skip` rows are read +// and not written. +std::string ReadAll(const std::string& in, bool scan, int skip) { + std::istringstream src(in); + protowire::pxf::UnmarshalOptions opts; + opts.max_bytes_literal_length = 191; + auto tr = DatasetReader::Create(&src, opts); + if (!tr.ok()) return "create: " + tr.status().ToString() + "\n"; + auto at = [](const protowire::pxf::Position& p) { + return std::to_string(p.line) + ":" + std::to_string(p.column) + "@" + std::to_string(p.offset); + }; + std::string out; + for (int i = 0; i < skip; ++i) { + DatasetRow row; + if (protowire::Status st = (*tr)->Next(&row); !st.ok()) return "skip: " + st.ToString() + "\n"; + } + for (int i = skip; i < skip + 10; ++i) { + const std::string name = "row" + std::to_string(i) + ": "; + if (scan) { + pb::FieldDescriptorProto msg; + protowire::Status st = (*tr)->Scan(&msg); + if (!st.ok()) return out + name + st.ToString() + "\n"; + continue; + } + DatasetRow row; + protowire::Status st = (*tr)->Next(&row); + if (!st.ok()) return out + name + st.ToString() + "\n"; + if (row.cells.empty()) return out + name + "EOF\n"; + out += name + "pos=" + at(row.pos); + if (row.cells[0].has_value()) { + auto [first, end] = protowire::pxf::ValueSpan(*row.cells[0]); + out += " cell0=" + at(first) + " end=" + at(end); + } + out += "\n"; + } + return out; +} + +TEST(DatasetReaderPositions, MatchProtowireGo) { + const std::string big = "b\"" + std::string(256, 'A') + "\""; + const std::string over = ": bytes literal decodes to more than MaxBytesLiteralLength=191 bytes\n"; + struct Case { + const char* name; + std::string in; + bool scan; + std::string want; + int skip = 0; + }; + const Case cases[] = { + {"first row", "@dataset x.Row ( raw )\n( " + big + " )\n", false, "row0: 2:3" + over}, + {"later row", + "@dataset x.Row ( raw )\n( 1 )\n( 2 )\n\n ( " + big + " )\n", + false, + "row0: pos=2:1@23 cell0=2:3@25 end=2:4@26\n" + "row1: pos=3:1@29 cell0=3:3@31 end=3:4@32\n" + "row2: 5:5" + + over}, + {"rows on the header's line", + "@dataset x.Row ( raw ) ( 1 ) ( " + big + " )\n", + false, + "row0: pos=1:24@23 cell0=1:26@25 end=1:27@26\n" + "row1: 1:34" + + over}, + {"a row's second line", + "@dataset x.Row ( a, b ) ( 1,\n " + big + " )\n", + false, + "row0: 2:5" + over}, + {"arity", + "@dataset x.Row ( a, b )\n( 1, 2 )\n ( 1 )\n", + false, + "row0: pos=2:1@24 cell0=2:3@26 end=2:4@27\n" + "row1: 3:4: @dataset row has 1 cells, expected 2 (column count)\n"}, + {"bad token", + "@dataset x.Row ( a )\n( 1 )\n ( 1 2 )\n", + false, + "row0: pos=2:1@21 cell0=2:3@23 end=2:4@24\n" + "row1: 3:7: expected ',' or ')' in @dataset row, got integer\n"}, + // 18000 bytes of rows before the bad one: it arrives several reads in. + {"many reads", + "@dataset x.Row ( raw )\n" + + [] { + std::string rows; + for (int i = 0; i < 3000; ++i) rows += "( 1 )\n"; + return rows; + }() + + " ( 1, 2 )\n", + false, + "row3000: 3002:3: @dataset row has 2 cells, expected 1 (column count)\n", + 3000}, + {"scan, a column the type lacks", + "@dataset google.protobuf.FieldDescriptorProto ( nope )\n\n ( 1 )\n", + true, + "row0: 3:3: @dataset column \"nope\": google.protobuf.FieldDescriptorProto has no field " + "\"nope\"\n"}, + {"scan", + "@dataset google.protobuf.FieldDescriptorProto ( name, number )\n(\"a\", 1)\n (\"b\", " + "\"x\")\n", + true, + "row1: 3:3: column number: expected integer for field \"number\"\n"}, + }; + for (const Case& c : cases) { + EXPECT_EQ(ReadAll(c.in, c.scan, c.skip), c.want) << c.name; + } +} + } // namespace