pxf: DatasetReader positions count from the start of the stream (#161)

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
 

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_;

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<int>(pos_)}; }
+  Position CurrentPos() const {
+    return Position{line_, column_, start_.offset + static_cast<int>(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<int>(pos_)) {
       pos_ = 0;
-      line_ = 1;
-      column_ = 1;
+      line_ = start_.line;
+      column_ = start_.column;
     }
     while (static_cast<int>(pos_) < target && pos_ < input_.size()) {
       uint8_t ch = static_cast<uint8_t>(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<std::string> owned_;  // pointer-stable storage
   std::vector<Error>* errors_ = nullptr;

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); }
 };

src/pxf/dataset_reader.cc

@@ -476,10 +476,11 @@ Status DatasetReader::Next(DatasetRow* out) {
                           static_cast<int>(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<size_t>(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<size_t>(end) + 1);
+      Consume(static_cast<size_t>(end) + 1);
       if (!parsed.ok()) {
         err_ = parsed.status();
         return err_;
@@ -538,6 +539,28 @@ std::unique_ptr<std::istream> 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<int>(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<size_t>(header_end) + 1);
+      Consume(static_cast<size_t>(header_end) + 1);
       return Status::OK();
     }
     if (src_eof_) {

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<DatasetDirective> ParseDatasetAt(Lexer* lex, Token* current, int max_de
   return tbl;
 }
 
-StatusOr<DatasetRow> ParseDatasetRow(std::string_view row,
-                                     int expected,
-                                     int max_depth,
-                                     int max_bytes_literal) {
-  Parser p(row, max_depth, max_bytes_literal);
+StatusOr<DatasetRow> 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);
 }

src/pxf/parser_internal.h

@@ -21,11 +21,11 @@ namespace protowire::pxf {
 StatusOr<DatasetDirective> 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<DatasetRow> 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<DatasetRow> 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.

test/pxf_dataset_reader_test.cc

@@ -15,6 +15,7 @@
 #include <string>
 
 #include <google/protobuf/compiler/importer.h>
+#include <google/protobuf/descriptor.pb.h>
 #include <google/protobuf/dynamic_message.h>
 
 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