pitrou commented on code in PR #50945:
URL: https://github.com/apache/arrow/pull/50945#discussion_r3892912499
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -124,40 +102,89 @@ namespace {
// and uses actual JSON parsing to delimit them.
class ParsingBoundaryFinder : public BoundaryFinder {
public:
- Status FindFirst(string_view partial, string_view block, int64_t* out_pos)
override {
- auto length = ConsumeWholeObject(MultiStringStream({partial, block}));
- if (length == string_view::npos) {
+ Status FindFirst(std::string_view partial, std::string_view block,
+ int64_t* out_pos) override {
+ std::string combined;
+ std::string_view input;
+
+ if (partial.empty()) {
+ input = block;
+ } else if (block.empty()) {
+ input = partial;
+ } else {
+ combined.reserve(partial.size() + block.size());
+ combined.append(partial);
+ combined.append(block);
+ input = combined;
+ }
+
+ const size_t start = ConsumeWhitespace(combined);
+ if (start < combined.size() && combined[start] != '{' && combined[start]
!= '[') {
Review Comment:
Out of curiosity, what happens if we remove the check `combined[start] !=
'{' && combined[start] != '['`?
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -17,105 +17,83 @@
#include "arrow/json/chunker.h"
-#include <algorithm>
#include <string_view>
#include <utility>
-#include <vector>
-#include "arrow/json/rapidjson_defs.h"
-#include "rapidjson/reader.h"
+#include <simdjson.h>
#include "arrow/buffer.h"
#include "arrow/json/options.h"
#include "arrow/util/logging_internal.h"
+#include "arrow/util/simdjson_internal.h"
namespace arrow {
-using std::string_view;
-
namespace json {
-namespace rj = arrow::rapidjson;
-
-static size_t ConsumeWhitespace(string_view view) {
-#ifdef RAPIDJSON_SIMD
- auto data = view.data();
- auto nonws_begin = rj::SkipWhitespace_SIMD(data, data + view.size());
- return nonws_begin - data;
-#else
- auto ws_count = view.find_first_not_of(" \t\r\n");
- if (ws_count == string_view::npos) {
+static size_t ConsumeWhitespace(std::string_view view) {
+ const auto ws_count = view.find_first_not_of(" \t\r\n");
+ if (ws_count == std::string_view::npos) {
return view.size();
- } else {
- return ws_count;
}
-#endif
+ return ws_count;
}
-/// RapidJson custom stream for reading JSON stored in multiple buffers
-/// http://rapidjson.org/md_doc_stream.html#CustomStream
-class MultiStringStream {
- public:
- using Ch = char;
- explicit MultiStringStream(std::vector<string_view> strings)
- : strings_(std::move(strings)) {
- std::reverse(strings_.begin(), strings_.end());
- }
- explicit MultiStringStream(const BufferVector& buffers) :
strings_(buffers.size()) {
- for (size_t i = 0; i < buffers.size(); ++i) {
- strings_[i] = string_view(*buffers[i]);
- }
- std::reverse(strings_.begin(), strings_.end());
- }
- char Peek() const {
- if (strings_.size() == 0) return '\0';
- return strings_.back()[0];
+static size_t ConsumeWholeObject(std::string_view input) {
Review Comment:
Can this take a `padded_string` directly so that the caller can optimize the
number of allocations and copies?
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -124,40 +102,89 @@ namespace {
// and uses actual JSON parsing to delimit them.
class ParsingBoundaryFinder : public BoundaryFinder {
public:
- Status FindFirst(string_view partial, string_view block, int64_t* out_pos)
override {
- auto length = ConsumeWholeObject(MultiStringStream({partial, block}));
- if (length == string_view::npos) {
+ Status FindFirst(std::string_view partial, std::string_view block,
+ int64_t* out_pos) override {
+ std::string combined;
Review Comment:
Can we create a `padded_string` directly to save one allocation and memory
copy?
For example we could use
[`padded_string_builder`](https://simdjson.github.io/simdjson/classsimdjson_1_1padded__string__builder.html)?
##########
cpp/src/arrow/json/chunker_test.cc:
##########
@@ -264,12 +264,17 @@ TEST(ChunkerTest, Errors) {
std::string parts[] = {R"({"a":0})", "}", R"({"a":1})"};
auto chunker = MakeChunker(true);
std::shared_ptr<Buffer> whole, rest, completion;
+
ASSERT_OK(chunker->Process(Buffer::FromString(parts[0] + parts[1]), &whole,
&rest));
- ASSERT_EQ(std::string_view(*whole), parts[0]);
- ASSERT_EQ(std::string_view(*rest), parts[1]);
+
+ // simdjson rejects the malformed stream as a whole, so no complete chunk
+ // is emitted before the trailing invalid data.
+ ASSERT_TRUE(whole);
+ ASSERT_EQ(std::string_view(*whole), "");
+ ASSERT_EQ(std::string_view(*rest), parts[0] + parts[1]);
+
auto status =
chunker->ProcessWithPartial(rest, Buffer::FromString(parts[2]),
&completion, &rest);
- ASSERT_RAISES(Invalid, status);
Review Comment:
Why remove this?
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -124,40 +102,89 @@ namespace {
// and uses actual JSON parsing to delimit them.
class ParsingBoundaryFinder : public BoundaryFinder {
public:
- Status FindFirst(string_view partial, string_view block, int64_t* out_pos)
override {
- auto length = ConsumeWholeObject(MultiStringStream({partial, block}));
- if (length == string_view::npos) {
+ Status FindFirst(std::string_view partial, std::string_view block,
+ int64_t* out_pos) override {
+ std::string combined;
+ std::string_view input;
+
+ if (partial.empty()) {
+ input = block;
+ } else if (block.empty()) {
+ input = partial;
+ } else {
+ combined.reserve(partial.size() + block.size());
+ combined.append(partial);
+ combined.append(block);
+ input = combined;
+ }
+
+ const size_t start = ConsumeWhitespace(combined);
+ if (start < combined.size() && combined[start] != '{' && combined[start]
!= '[') {
+ return Status::Invalid("JSON chunk error: invalid data at end of
document");
+ }
+
+ const auto length = ConsumeWholeObject(combined);
+
+ if (length == std::string_view::npos) {
*out_pos = -1;
} else if (ARROW_PREDICT_FALSE(length < partial.size())) {
return Status::Invalid("JSON chunk error: invalid data at end of
document");
} else {
DCHECK_LE(length, partial.size() + block.size());
*out_pos = static_cast<int64_t>(length - partial.size());
}
+
return Status::OK();
}
Status FindLast(std::string_view block, int64_t* out_pos) override {
const size_t block_length = block.size();
size_t consumed_length = 0;
+
+ if (block_length > 0) {
+ const size_t start = ConsumeWhitespace(block);
+ if (start < block.size() && block[start] != '{' && block[start] != '[') {
+ return Status::Invalid("JSON parse error: Invalid value");
+ }
+ }
+
while (consumed_length < block_length) {
- rj::MemoryStream ms(reinterpret_cast<const char*>(block.data()),
block.size());
- using InputStream = rj::EncodedInputStream<rj::UTF8<>, rj::MemoryStream>;
- auto length = ConsumeWholeObject(InputStream(ms));
- if (length == string_view::npos || length == 0) {
- // found incomplete object or block is empty
+ const auto length = ConsumeWholeObject(block);
Review Comment:
Each time `ConsumeWholeObject` is called in this loop, it will create a new
`padded_string` and a new `parser`.
Why not iterate on the document stream here?
Alternatively, `ConsumeWholeObject` could take a boolean flag to iterate
until the last document in the stream.
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -17,105 +17,83 @@
#include "arrow/json/chunker.h"
-#include <algorithm>
#include <string_view>
#include <utility>
-#include <vector>
-#include "arrow/json/rapidjson_defs.h"
-#include "rapidjson/reader.h"
+#include <simdjson.h>
#include "arrow/buffer.h"
#include "arrow/json/options.h"
#include "arrow/util/logging_internal.h"
+#include "arrow/util/simdjson_internal.h"
namespace arrow {
-using std::string_view;
-
namespace json {
-namespace rj = arrow::rapidjson;
-
-static size_t ConsumeWhitespace(string_view view) {
-#ifdef RAPIDJSON_SIMD
- auto data = view.data();
- auto nonws_begin = rj::SkipWhitespace_SIMD(data, data + view.size());
- return nonws_begin - data;
-#else
- auto ws_count = view.find_first_not_of(" \t\r\n");
- if (ws_count == string_view::npos) {
+static size_t ConsumeWhitespace(std::string_view view) {
+ const auto ws_count = view.find_first_not_of(" \t\r\n");
+ if (ws_count == std::string_view::npos) {
return view.size();
- } else {
- return ws_count;
}
-#endif
+ return ws_count;
}
-/// RapidJson custom stream for reading JSON stored in multiple buffers
-/// http://rapidjson.org/md_doc_stream.html#CustomStream
-class MultiStringStream {
- public:
- using Ch = char;
- explicit MultiStringStream(std::vector<string_view> strings)
- : strings_(std::move(strings)) {
- std::reverse(strings_.begin(), strings_.end());
- }
- explicit MultiStringStream(const BufferVector& buffers) :
strings_(buffers.size()) {
- for (size_t i = 0; i < buffers.size(); ++i) {
- strings_[i] = string_view(*buffers[i]);
- }
- std::reverse(strings_.begin(), strings_.end());
- }
- char Peek() const {
- if (strings_.size() == 0) return '\0';
- return strings_.back()[0];
+static size_t ConsumeWholeObject(std::string_view input) {
+ if (input.empty()) {
+ return 0;
}
- char Take() {
- if (strings_.size() == 0) return '\0';
- char taken = strings_.back()[0];
- if (strings_.back().size() == 1) {
- strings_.pop_back();
- } else {
- strings_.back() = strings_.back().substr(1);
- }
- ++index_;
- return taken;
+
+ const size_t start = ConsumeWhitespace(input);
+ if (start >= input.size()) {
+ return 0;
}
- size_t Tell() { return index_; }
- void Put(char) { ARROW_LOG(FATAL) << "not implemented"; }
- void Flush() { ARROW_LOG(FATAL) << "not implemented"; }
- char* PutBegin() {
- ARROW_LOG(FATAL) << "not implemented";
- return nullptr;
+
+ // Keep the padded buffer alive while iterating the document stream.
+ simdjson::padded_string padded(input);
+ simdjson::ondemand::parser parser;
+ simdjson::ondemand::document_stream stream;
+
+ auto stream_status = internal::ResolveSimdjsonResult(
+ parser.iterate_many(padded), "Failed to create JSON document stream");
+ if (!stream_status.ok()) {
+ return std::string_view::npos;
}
Review Comment:
You don't need to call `ResolveSimdjsonResult` if it's only to return
`npos`, I think?
You could write something like:
```c++
simdjon::ondemand::document_stream stream;
if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
return std::string_view::npos;
}
```
(this could be made even simpler with a helper function that returns a
`std::optional<T>` or a `std::pair<error_code, T>`)
##########
cpp/src/arrow/json/chunker.cc:
##########
@@ -17,105 +17,83 @@
#include "arrow/json/chunker.h"
-#include <algorithm>
#include <string_view>
#include <utility>
-#include <vector>
-#include "arrow/json/rapidjson_defs.h"
-#include "rapidjson/reader.h"
+#include <simdjson.h>
#include "arrow/buffer.h"
#include "arrow/json/options.h"
#include "arrow/util/logging_internal.h"
+#include "arrow/util/simdjson_internal.h"
namespace arrow {
-using std::string_view;
-
namespace json {
-namespace rj = arrow::rapidjson;
-
-static size_t ConsumeWhitespace(string_view view) {
-#ifdef RAPIDJSON_SIMD
- auto data = view.data();
- auto nonws_begin = rj::SkipWhitespace_SIMD(data, data + view.size());
- return nonws_begin - data;
-#else
- auto ws_count = view.find_first_not_of(" \t\r\n");
- if (ws_count == string_view::npos) {
+static size_t ConsumeWhitespace(std::string_view view) {
+ const auto ws_count = view.find_first_not_of(" \t\r\n");
+ if (ws_count == std::string_view::npos) {
return view.size();
- } else {
- return ws_count;
}
-#endif
+ return ws_count;
}
-/// RapidJson custom stream for reading JSON stored in multiple buffers
-/// http://rapidjson.org/md_doc_stream.html#CustomStream
-class MultiStringStream {
- public:
- using Ch = char;
- explicit MultiStringStream(std::vector<string_view> strings)
- : strings_(std::move(strings)) {
- std::reverse(strings_.begin(), strings_.end());
- }
- explicit MultiStringStream(const BufferVector& buffers) :
strings_(buffers.size()) {
- for (size_t i = 0; i < buffers.size(); ++i) {
- strings_[i] = string_view(*buffers[i]);
- }
- std::reverse(strings_.begin(), strings_.end());
- }
- char Peek() const {
- if (strings_.size() == 0) return '\0';
- return strings_.back()[0];
+static size_t ConsumeWholeObject(std::string_view input) {
+ if (input.empty()) {
+ return 0;
}
- char Take() {
- if (strings_.size() == 0) return '\0';
- char taken = strings_.back()[0];
- if (strings_.back().size() == 1) {
- strings_.pop_back();
- } else {
- strings_.back() = strings_.back().substr(1);
- }
- ++index_;
- return taken;
+
+ const size_t start = ConsumeWhitespace(input);
Review Comment:
Why do we have to consume whitespace explicitly? It seems simdjson already
takes care of that? See
https://simdjson.github.io/simdjson/md_doc_2iterate__many.html
> we support any file that contains any amount of valid JSON document,
separated by one or more character that is considered whitespace by the JSON
spec
##########
python/pyarrow/tests/test_json.py:
##########
@@ -150,8 +150,7 @@ def test_block_sizes(self):
for newlines_in_values in [False, True]:
parse_options.newlines_in_values = newlines_in_values
read_options.block_size = 4
- with pytest.raises(ValueError,
- match="try to increase block size"):
+ with pytest.raises(ValueError):
Review Comment:
What is the actual error message in this case?
--
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]
