pxf: Scan's options pinned per option; validator and schema errors carry no row position (#166)

CHANGELOG.md

@@ -22,6 +22,8 @@ number says nothing about any other port. See the Releases section of
 - **A timestamp cell right before a row's `)` parses** (#169), as protowire-go does. `(2024-01-02T03:04:05Z)` was refused with `invalid timestamp: 2024-01-02T03:04:05Z)`, because the lexer read the `)` into the timestamp. A space before the `)`, or a later cell, hid it.
 - **`DatasetReader` reads a long token in time linear in its length** (#168), as protowire-go does. A string, bytes literal, comment or number spread over many reads was lexed again from its start on every 4 KiB read: an unclosed 1 MiB string cost 136 MB of lexing, and 2 MiB cost 540 MB. The scan now lexes a token still open again only once the bytes read since could end it, which is Go's `mayEndToken` rule, and the header scan resumes where it stopped too. The same 1 MiB string now costs 1.05 MB.
 - **`DatasetReader` returns a complete row from a stream that stays open** (#171), as protowire-go does. On a pipe, a socket or a file still being appended to, `Next` could wait for 4 KiB more input, or for the stream to end, before returning a row that had already arrived. The read took a full 4 KiB (`std::istream::read`), and the scan held a token until four more bytes followed it. `Next` now reads what the stream has, blocking only when nothing is buffered. It takes a token as complete as soon as the bytes the lexer looks at past it have arrived. Input that ends reads as before.
+- **A row's validator error and schema violation carry no row position** (#166), as protowire-go gives them. `DatasetReader::Scan` and `BindRow` put the row's position in front of every error, so a validator's refusal read `2:1: check: id: denied (...)` and a schema violation `2:1: PXF schema violations: ...`. Only a decode error is moved to the row now, and a hook's error from `on_secret_field` is one, so it keeps the row's position and its column. `Scan` also checks the column paths before the schema, as Go does, so a row type that breaks both gets the column's error.
+- **`DatasetReader::Scan` honours every option but `max_message_size`, and protowire-go is changing to match** (#166). Go's `Scan` ignored `skip_validate`, `skip_post_decode`, `discard_unknown`, `on_secret_field` and `validator`, while cpp's honoured them. The owner chose to change Go (trendvidia/protowire-go#280) and keep cpp's behaviour. Each option now has a test that binds the same row through `Scan` and `BindRow`, with the texts Go gives, and `dataset_reader.h` lists the options `Scan` uses.
 
 ## [2.0.0] — 2026-11-05
 

include/protowire/pxf/dataset_reader.h

@@ -71,11 +71,13 @@ class DatasetReader {
 
   // Create under options. The bytes held while a row's end is sought are
   // capped at opts.max_message_size, and the header and every row are
-  // parsed under its nesting and bytes-literal limits. Scan binds rows with
-  // the options, but under the default max_message_size, since a row's body
-  // repeats the column names. So a cell holding an Any block needs
-  // opts.type_resolver, and opts.skip_validate turns off Scan's schema
-  // check.
+  // parsed under its nesting and bytes-literal limits. Scan binds each row
+  // as BindRow does under the same options, except that max_message_size
+  // stays at its default, since a row's body repeats the column names. So
+  // the other limits, type_resolver (which a cell holding an Any block
+  // needs), discard_unknown, skip_post_decode, skip_validate,
+  // on_secret_field, validator and report all apply to a streamed row, as
+  // they do in protowire-go (#166).
   static StatusOr<std::unique_ptr<DatasetReader>> Create(std::istream* src, UnmarshalOptions opts);
 
   // Row message type declared by the @dataset header (e.g. "trades.v1.Trade").
@@ -216,8 +218,11 @@ Status BindRow(google::protobuf::Message* msg,
                const std::vector<std::string>& columns,
                const DatasetRow& row);
 
-// BindRow under options: their limits, skip_validate, and type_resolver,
-// which a cell holding an Any block that opens with a type entry needs.
+// BindRow under options: every one of them applies to the row, as
+// Unmarshal applies it to a body. type_resolver is what a cell holding an
+// Any block that opens with a type entry needs. A decode error carries the
+// row's position; a schema violation or a validator's error is returned as
+// it is, as protowire-go returns them.
 Status BindRow(google::protobuf::Message* msg,
                const std::vector<std::string>& columns,
                const DatasetRow& row,

src/pxf/dataset_reader.cc

@@ -13,6 +13,7 @@
 
 #include "dataset_paths.h"
 #include "parser_internal.h"
+#include "protowire/check_seam.h"
 #include "protowire/limits.h"
 #include "protowire/pxf.h"  // Unmarshal, UnmarshalOptions
 #include "protowire/pxf/format.h"
@@ -180,15 +181,31 @@ Status AtRow(const Status& st,
 // UnmarshalFull rather than Unmarshal because its post-decode pass is what
 // gives an empty cell the field's (pxf.default), rejects it for
 // (pxf.required), and lets a null cell satisfy required (draft §3.4.4).
+//
+// Only a decode error is moved to the row, as protowire-go's atRow moves
+// only its *Error. The schema check runs before the decode and the
+// validator after it, so their errors are returned as they are (#166).
 Status BindCheckedRow(google::protobuf::Message* msg,
                       const std::vector<std::string>& columns,
                       const DatasetRow& row,
-                      const UnmarshalOptions& opts) {
+                      UnmarshalOptions opts) {
+  if (!opts.skip_validate) {
+    if (Status s = CheckDescriptorSchema(msg->GetDescriptor()); !s.ok()) return s;
+  }
+  check::Validator* validator = opts.validator;
+  check::Report* report_out = opts.report;
+  opts.skip_validate = true;
+  opts.validator = nullptr;
+  opts.report = nullptr;
   BodyWriter w(row);
   WriteRowBody(columns, row, &w);
   auto r = UnmarshalFull(w.body(), msg, opts);
-  if (r.ok()) return Status::OK();
-  return AtRow(r.status(), row.pos, columns, w.spans());
+  if (!r.ok()) return AtRow(r.status(), row.pos, columns, w.spans());
+  check::Report report;
+  bool has_report = false;
+  Status st = check::Validate(validator, *msg, &report, &has_report);
+  if (has_report && report_out != nullptr) *report_out = std::move(report);
+  return st;
 }
 
 }  // namespace
@@ -271,19 +288,20 @@ Status DatasetReader::Scan(google::protobuf::Message* msg) {
   // one type, which is the usual case, and a caller alternating types pays
   // only for what it does.
   const auto* desc = msg->GetDescriptor();
+  // The paths are checked first, as protowire-go checks them.
   if (desc != checked_type_) {
-    if (!opts_.skip_validate) {
-      if (Status s = CheckDescriptorSchema(desc); !s.ok()) return s;
-    }
     if (std::string path_err = CheckColumnPaths(desc, columns_); !path_err.empty()) {
       return HeaderError(std::move(path_err), row.pos);
     }
+    if (!opts_.skip_validate) {
+      if (Status s = CheckDescriptorSchema(desc); !s.ok()) return s;
+    }
     checked_type_ = desc;
   }
-  // The caller's limits but max_message_size, which is left at its default,
-  // as protowire-go binds a row: the row's bytes were capped as they were
-  // read, and the body it stands for repeats the column names, so it is
-  // wider than the row.
+  // The caller's options but max_message_size, which is left at its
+  // default, as protowire-go binds a row: the row's bytes were capped as
+  // they were read, and the body it stands for repeats the column names, so
+  // it is wider than the row.
   UnmarshalOptions opts = opts_;
   opts.skip_validate = true;
   opts.max_message_size = 0;

test/CMakeLists.txt

@@ -38,7 +38,7 @@ if(TARGET protowire_pxf)
             pxf_null_test pxf_bignum_test pxf_features_test pxf_any_test
             pxf_fast_test pxf_full_roundtrip_test pxf_escapes_test
             pxf_annotations_test pxf_format_test pxf_tolerant_test pxf_check_test pxf_directive_test
-            pxf_schema_test pxf_schema_rules_test pxf_secret_test pxf_element_defaults_test pxf_result_directives_test pxf_dataset_reader_test
+            pxf_schema_test pxf_schema_rules_test pxf_secret_test pxf_element_defaults_test pxf_result_directives_test pxf_dataset_reader_test pxf_dataset_scan_options_test
             pxf_proto_directive_test pxf_map_keys_test pxf_keyed_test pxf_hardening_test pxf_go_parity_test pxf_dataset_writer_test pxf_rewriter_test pxf_wkt_shape_test
             pxf_nan_test pxf_format_directives_test
             pxf_duration_block_test

test/pxf_dataset_scan_options_test.cc

@@ -0,0 +1,258 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2026 TrendVidia, LLC.
+//
+// DatasetReader::Scan binds a row under the options the reader was built
+// with, as BindRow binds it under the same options (#166). protowire-go's
+// Scan ignored skip_validate, skip_post_decode, discard_unknown,
+// on_secret_field and validator; it was changed to honour them
+// (protowire-go#280), so the two ports agree. Each case binds one row
+// through both paths, with the option off and on, and requires the paths
+// to agree. The protos, rows and error texts are protowire-go's
+// (encoding/pxf/dataset_scan_options_test.go).
+
+#include "protowire/check_seam.h"
+#include "protowire/pxf.h"
+#include "protowire/pxf/dataset_reader.h"
+
+#include <gtest/gtest.h>
+#include "fixture_dir.h"
+#include "protoc_compat.h"
+
+#include <fstream>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <google/protobuf/compiler/importer.h>
+#include <google/protobuf/dynamic_message.h>
+#include <google/protobuf/util/message_differencer.h>
+
+namespace {
+
+namespace pb = google::protobuf;
+namespace pxf = protowire::pxf;
+using protowire::Status;
+
+constexpr const char* kSchema = R"src(
+syntax = "proto3";
+package so.v1;
+
+import "pxf/annotations.proto";
+import "pxf/secret.proto";
+
+message Inner { int32 n = 1; }
+message Row {
+  string id = 1;
+  int32 req = 2 [(pxf.required) = true];
+  string dflt = 3 [(pxf.default) = "z"];
+  Inner inner = 4;
+  pxf.Secret pw = 5;
+}
+)src";
+
+// A field named null breaks a bind-time schema rule (draft §3.13). It is in
+// a file of its own, because the schema check reads the whole file.
+constexpr const char* kBadSchema = R"src(
+syntax = "proto3";
+package sobad.v1;
+
+message Bad {
+  string id = 1;
+  string null = 2;
+}
+)src";
+
+class Errors : public pb::compiler::MultiFileErrorCollector {
+ public:
+  PROTOWIRE_PROTOC_RECORD_ERROR(filename, line, column, msg) {
+    last = std::string(filename) + ":" + std::to_string(line) + ":" + std::to_string(column) +
+           ": " + std::string(msg);
+  }
+  std::string last;
+};
+
+class DenyAll : public protowire::check::Validator {
+ public:
+  protowire::StatusOr<protowire::check::Report> Validate(const pb::Message&) override {
+    protowire::check::Report r;
+    r.violations.push_back({"id", "test.deny", "denied"});
+    return r;
+  }
+};
+
+// One row bound both ways: the message, or the error both paths gave.
+struct Bound {
+  std::unique_ptr<pb::Message> msg;
+  std::string err;  // empty when the row bound
+};
+
+class DatasetScanOptions : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    std::ofstream(protowire_test::FixtureDir() + "/so.proto", std::ios::binary) << kSchema;
+    std::ofstream(protowire_test::FixtureDir() + "/sobad.proto", std::ios::binary) << kBadSchema;
+    tree_.MapPath("", protowire_test::FixtureDir());
+    tree_.MapPath("", PROTO_DIR);
+    tree_.MapPath("", WKT_PROTO_DIR);
+    importer_ = std::make_unique<pb::compiler::Importer>(&tree_, &errors_);
+    const pb::FileDescriptor* fd = importer_->Import("so.proto");
+    ASSERT_NE(fd, nullptr) << errors_.last;
+    row_ = fd->FindMessageTypeByName("Row");
+    const pb::FileDescriptor* bad = importer_->Import("sobad.proto");
+    ASSERT_NE(bad, nullptr) << errors_.last;
+    bad_ = bad->FindMessageTypeByName("Bad");
+  }
+
+  // BindUnder binds the one row of table under opts, through Scan and
+  // through Parse + BindRow, and requires the two to agree.
+  Bound BindUnder(const pxf::UnmarshalOptions& opts,
+                  const pb::Descriptor* desc,
+                  const std::string& table) {
+    Bound out;
+    auto doc = pxf::Parse(table);
+    EXPECT_TRUE(doc.ok()) << doc.status().ToString();
+    if (!doc.ok()) return out;
+    const auto& ds = doc->datasets[0];
+    EXPECT_EQ(ds.rows.size(), 1u);
+    std::unique_ptr<pb::Message> via_bind(factory_.GetPrototype(desc)->New());
+    const Status berr = pxf::BindRow(via_bind.get(), ds.columns, ds.rows[0], opts);
+
+    std::istringstream src(table);
+    auto tr = pxf::DatasetReader::Create(&src, opts);
+    EXPECT_TRUE(tr.ok()) << tr.status().ToString();
+    if (!tr.ok()) return out;
+    std::unique_ptr<pb::Message> via_scan(factory_.GetPrototype(desc)->New());
+    const Status serr = (*tr)->Scan(via_scan.get());
+    EXPECT_EQ(berr.ok() ? "" : berr.ToString(), serr.ok() ? "" : serr.ToString())
+        << "BindRow vs Scan";
+    if (!berr.ok()) {
+      out.err = berr.ToString();
+      return out;
+    }
+    EXPECT_FALSE((*tr)->Done()) << "Scan read no row";
+    EXPECT_TRUE(pb::util::MessageDifferencer::Equals(*via_bind, *via_scan))
+        << "BindRow " << via_bind->ShortDebugString() << ", Scan " << via_scan->ShortDebugString();
+    std::unique_ptr<pb::Message> after(factory_.GetPrototype(desc)->New());
+    EXPECT_TRUE((*tr)->Scan(after.get()).ok());
+    EXPECT_TRUE((*tr)->Done()) << "a second row was read";
+    out.msg = std::move(via_scan);
+    return out;
+  }
+
+  static std::string Str(const pb::Message& m, const char* name) {
+    return m.GetReflection()->GetString(m, m.GetDescriptor()->FindFieldByName(name));
+  }
+  static const pb::Message& Sub(const pb::Message& m, const char* name) {
+    return m.GetReflection()->GetMessage(m, m.GetDescriptor()->FindFieldByName(name));
+  }
+
+  pb::compiler::DiskSourceTree tree_;
+  Errors errors_;
+  std::unique_ptr<pb::compiler::Importer> importer_;
+  pb::DynamicMessageFactory factory_;
+  const pb::Descriptor* row_ = nullptr;
+  const pb::Descriptor* bad_ = nullptr;
+};
+
+TEST_F(DatasetScanOptions, SkipPostDecode) {
+  const std::string table = "@dataset so.v1.Row ( id )\n(\"x\")\n";
+  EXPECT_EQ(BindUnder({}, row_, table).err, "2:1: required field \"req\" is absent");
+
+  pxf::UnmarshalOptions o;
+  o.skip_post_decode = true;
+  Bound b = BindUnder(o, row_, table);
+  ASSERT_EQ(b.err, "");
+  // No post-decode pass: the required field is not checked, and the
+  // default is not applied.
+  EXPECT_EQ(Str(*b.msg, "dflt"), "");
+}
+
+TEST_F(DatasetScanOptions, DiscardUnknown) {
+  const std::string table =
+      "@dataset so.v1.Row ( id, req, inner )\n(\"x\", 1, { n = 2 nope = 3 })\n";
+  EXPECT_EQ(BindUnder({}, row_, table).err,
+            "2:1: column inner: unknown field \"nope\" in so.v1.Inner");
+
+  pxf::UnmarshalOptions o;
+  o.discard_unknown = true;
+  Bound b = BindUnder(o, row_, table);
+  ASSERT_EQ(b.err, "");
+  const pb::Message& inner = Sub(*b.msg, "inner");
+  EXPECT_EQ(inner.GetReflection()->GetInt32(inner, inner.GetDescriptor()->FindFieldByName("n")), 2);
+}
+
+TEST_F(DatasetScanOptions, OnSecretField) {
+  const std::string table = "@dataset so.v1.Row ( id, req, pw )\n(\"x\", 1, \"s3cret\")\n";
+  Bound plain = BindUnder({}, row_, table);
+  ASSERT_EQ(plain.err, "");
+  EXPECT_EQ(Str(Sub(*plain.msg, "pw"), "value"), "s3cret");
+
+  // BindUnder binds the row twice, so the hook fires once per path.
+  std::vector<std::string> got;
+  pxf::UnmarshalOptions o;
+  o.on_secret_field = [&got](std::string_view path, std::string_view value) {
+    got.push_back(std::string(path) + "=" + std::string(value));
+    return Status::OK();
+  };
+  Bound b = BindUnder(o, row_, table);
+  ASSERT_EQ(b.err, "");
+  EXPECT_EQ(got, (std::vector<std::string>{"pw=s3cret", "pw=s3cret"}));
+  EXPECT_EQ(Str(Sub(*b.msg, "pw"), "value"), "");
+}
+
+// A hook's error is a decode error: it carries the row's position and names
+// the column, as protowire-go gives it.
+TEST_F(DatasetScanOptions, OnSecretFieldError) {
+  pxf::UnmarshalOptions o;
+  o.on_secret_field = [](std::string_view, std::string_view) {
+    return Status::Error("hook says no");
+  };
+  for (const char* table : {"@dataset so.v1.Row ( id, req, pw )\n(\"x\", 1, \"s3cret\")\n",
+                            "@dataset so.v1.Row ( pw, id, req )\n(\"s3cret\", \"x\", 1)\n"}) {
+    EXPECT_EQ(BindUnder(o, row_, table).err,
+              "2:1: column pw: pxf.Secret hook for \"pw\": hook says no")
+        << table;
+  }
+}
+
+// A validator's error is returned as it is, with no position: protowire-go
+// moves only a decode error to the row.
+TEST_F(DatasetScanOptions, Validator) {
+  const std::string table = "@dataset so.v1.Row ( id, req )\n(\"x\", 1)\n";
+  EXPECT_EQ(BindUnder({}, row_, table).err, "");
+
+  DenyAll deny;
+  protowire::check::Report report;
+  pxf::UnmarshalOptions o;
+  o.validator = &deny;
+  o.report = &report;
+  EXPECT_EQ(BindUnder(o, row_, table).err, "check: id: denied (test.deny)");
+  ASSERT_EQ(report.violations.size(), 1u);
+  EXPECT_EQ(report.violations[0].rule_id, "test.deny");
+}
+
+// A schema violation is returned as it is, with no position, through
+// BindRow as through Scan.
+TEST_F(DatasetScanOptions, SkipValidate) {
+  const std::string table = "@dataset sobad.v1.Bad ( id )\n(\"x\")\n";
+  EXPECT_EQ(BindUnder({}, bad_, table).err,
+            "PXF schema violations:\n  sobad.proto: message field \"sobad.v1.Bad.null\" uses "
+            "PXF-reserved name \"null\" (draft §3.13)");
+
+  pxf::UnmarshalOptions o;
+  o.skip_validate = true;
+  Bound b = BindUnder(o, bad_, table);
+  ASSERT_EQ(b.err, "");
+  EXPECT_EQ(Str(*b.msg, "id"), "x");
+}
+
+// The column paths are checked before the schema, as protowire-go checks
+// them: a row that breaks both gets the column's error.
+TEST_F(DatasetScanOptions, ColumnPathsBeforeSchema) {
+  EXPECT_EQ(BindUnder({}, bad_, "@dataset sobad.v1.Bad ( nope )\n(\"x\")\n").err,
+            "2:1: @dataset column \"nope\": sobad.v1.Bad has no field \"nope\"");
+}
+
+}  // namespace