taepper commented on code in PR #50945:
URL: https://github.com/apache/arrow/pull/50945#discussion_r3978640434
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
+ // XXX Hopefully this upholds for all std::string implementations
+ DCHECK_GE(buffer_.capacity() - buffer_.size(), simdjson::SIMDJSON_PADDING);
+ auto view = simdjson::padded_string_view(buffer_);
+ DCHECK(view.has_sufficient_padding());
+ return view;
+ }
+
+ // Consume the first or last JSON object (depending on `until_end`)
+ // and return the consumed JSON byte length, or 0 if no valid document
+ // can be parsed.
+ Result<size_t> ConsumeWholeObject(simdjson::padded_string_view input, bool
until_end) {
+ simdjson::ondemand::document_stream stream;
+ // XXX Should be pass a specific batch_size?
+ // The default value used by simdjson is 1MB, probably enough for most
purposes.
+ RETURN_NOT_OK(ToStatus(parser_.iterate_many(input).get(stream)));
+ auto it = stream.begin();
+ if (it == stream.end()) {
+ // Empty input (only whitespace?)
+ return 0;
+ }
+
+ int64_t consumed_length = 0;
+ if (!until_end) {
+ // Parsing the first document only.
+ if (!ConsumeDocument(it).ok()) {
+ // Could be either a partial document or invalid JSON, we'll let
+ // followup chunker or parser calls decide.
+ return 0;
+ }
+ // current_index() is the start of the current document;
+ // source() is the complete source span of the current document.
+ consumed_length = it.current_index() + it.source().size();
+ } else {
+ while (it != stream.end()) {
+ if (!ConsumeDocument(it).ok()) {
+ // Could be either a partial document or invalid JSON, we'll let
+ // followup chunker or parser calls decide.
+ break;
+ }
+ consumed_length = it.current_index() + it.source().size();
+ ++it;
+ }
+ }
+ if (consumed_length > 0) {
+ // If we found at least one document, also consume its trailing
whitespace
+ // to avoid stray bytes at the end of the stream.
+ consumed_length += ConsumeWhitespace(input.substr(consumed_length));
+ }
Review Comment:
Why do we care about this? We can just emit boundaries with whitespace, it
is valid json
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
+ // XXX Hopefully this upholds for all std::string implementations
+ DCHECK_GE(buffer_.capacity() - buffer_.size(), simdjson::SIMDJSON_PADDING);
Review Comment:
I do not think we should rely on this? At least in the other `simdjson` code
I reviewed we did not want to crash in this case
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
+ // XXX Hopefully this upholds for all std::string implementations
+ DCHECK_GE(buffer_.capacity() - buffer_.size(), simdjson::SIMDJSON_PADDING);
+ auto view = simdjson::padded_string_view(buffer_);
+ DCHECK(view.has_sufficient_padding());
+ return view;
+ }
+
+ // Consume the first or last JSON object (depending on `until_end`)
+ // and return the consumed JSON byte length, or 0 if no valid document
+ // can be parsed.
+ Result<size_t> ConsumeWholeObject(simdjson::padded_string_view input, bool
until_end) {
+ simdjson::ondemand::document_stream stream;
+ // XXX Should be pass a specific batch_size?
+ // The default value used by simdjson is 1MB, probably enough for most
purposes.
Review Comment:
I do not think we care about custom batch sizing
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
Review Comment:
Ugh, `simdjson` does not have RapidJson's `MultiStringStream` so the current
api is a rough fit. I see that call sites want this "partial", "continuation"
split to consume streams efficiently
We might want to at least check for emptiness of `partial`. After checking
call sites this seems to be the case a few times
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
+ // XXX Hopefully this upholds for all std::string implementations
+ DCHECK_GE(buffer_.capacity() - buffer_.size(), simdjson::SIMDJSON_PADDING);
+ auto view = simdjson::padded_string_view(buffer_);
+ DCHECK(view.has_sufficient_padding());
+ return view;
+ }
+
+ // Consume the first or last JSON object (depending on `until_end`)
+ // and return the consumed JSON byte length, or 0 if no valid document
+ // can be parsed.
+ Result<size_t> ConsumeWholeObject(simdjson::padded_string_view input, bool
until_end) {
+ simdjson::ondemand::document_stream stream;
+ // XXX Should be pass a specific batch_size?
+ // The default value used by simdjson is 1MB, probably enough for most
purposes.
+ RETURN_NOT_OK(ToStatus(parser_.iterate_many(input).get(stream)));
+ auto it = stream.begin();
+ if (it == stream.end()) {
+ // Empty input (only whitespace?)
+ return 0;
+ }
+
+ int64_t consumed_length = 0;
+ if (!until_end) {
+ // Parsing the first document only.
+ if (!ConsumeDocument(it).ok()) {
+ // Could be either a partial document or invalid JSON, we'll let
+ // followup chunker or parser calls decide.
+ return 0;
+ }
+ // current_index() is the start of the current document;
+ // source() is the complete source span of the current document.
+ consumed_length = it.current_index() + it.source().size();
+ } else {
+ while (it != stream.end()) {
+ if (!ConsumeDocument(it).ok()) {
+ // Could be either a partial document or invalid JSON, we'll let
+ // followup chunker or parser calls decide.
+ break;
+ }
+ consumed_length = it.current_index() + it.source().size();
+ ++it;
+ }
+ }
+ if (consumed_length > 0) {
+ // If we found at least one document, also consume its trailing
whitespace
+ // to avoid stray bytes at the end of the stream.
+ consumed_length += ConsumeWhitespace(input.substr(consumed_length));
+ }
Review Comment:
Or is the contract that `consumed_length` is the same as `input.size()` if
we consume the last element of the input?
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
+ // XXX Hopefully this upholds for all std::string implementations
+ DCHECK_GE(buffer_.capacity() - buffer_.size(), simdjson::SIMDJSON_PADDING);
+ auto view = simdjson::padded_string_view(buffer_);
+ DCHECK(view.has_sufficient_padding());
+ return view;
+ }
+
+ // Consume the first or last JSON object (depending on `until_end`)
+ // and return the consumed JSON byte length, or 0 if no valid document
+ // can be parsed.
+ Result<size_t> ConsumeWholeObject(simdjson::padded_string_view input, bool
until_end) {
Review Comment:
I find the name `until_end` a little confusing. Why did we change from the
previous design where `FindLast` would repeatedly call `ConsumeWholeObject`?
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -165,6 +91,78 @@ class ParsingBoundaryFinder : public BoundaryFinder {
int64_t* out_pos, int64_t* num_found) override {
return Status::NotImplemented("ParsingBoundaryFinder::FindNth");
}
+
+ private:
+ simdjson::ondemand::parser parser_;
+ // A persistent buffer to keep padded contents for simdjson.
+ // This should be more efficient than allocating a new padded_string
everytime.
+ std::string buffer_;
+
+ simdjson::padded_string_view GetPaddedStringView(std::string_view partial,
+ std::string_view block =
{}) {
+ // Adjust buffer size without copying old contents.
+ buffer_.clear();
+ buffer_.reserve(partial.size() + block.size() +
simdjson::SIMDJSON_PADDING);
+ buffer_.append(partial);
+ buffer_.append(block);
+ // XXX Hopefully this upholds for all std::string implementations
+ DCHECK_GE(buffer_.capacity() - buffer_.size(), simdjson::SIMDJSON_PADDING);
Review Comment:
`simdjson::padded_input` will provide a copy-free version, but it is not yet
in a released version:
https://github.com/simdjson/simdjson/commit/85cadf4074b78c7dfa7dabe89bef373b84cf0ba6
Also note that their code is more involved than the checking you do here:
```
const size_t len = s.size();
const size_t cap = s.capacity();
// Here we have the string content from data() to data() + size(),
// but the memory is accessible from data() to data() + capacity().
const size_t needed_padding = (cap - len) < simdjson::SIMDJSON_PADDING
? simdjson::SIMDJSON_PADDING - (cap - len) : 0;
if (needed_padding > 0 && needs_allocation(s.data(), cap, needed_padding))
{
storage = simdjson::padded_string(s);
} else {
storage = simdjson::padded_string_view(
s.data(), len, len + simdjson::SIMDJSON_PADDING);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
