Skip to content

GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Reranko05 wants to merge 14 commits into
base:main
Choose a base branch
from
+149 −91
Open

GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker #50945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
be4c2a1
Replace RapidJSON with simdjson in JSON chunker
Reranko05 Aug 21, 2026
aa6e20d
WIP: preserve chunker behavior before iterate_many
Reranko05 Aug 24, 2026
a5b89a7
WIP: Iterate Many
Reranko05 Aug 25, 2026
3f15314
use iterate_many approach
Reranko05 Aug 27, 2026
b9af8ce
Change expectation of python test
Reranko05 Aug 27, 2026
182daa4
Address Feedback
Reranko05 Aug 29, 2026
9b70128
use ResolveSimdjsonResult
Reranko05 Aug 29, 2026
9a5cb38
Remove ConsumeWhitespaces
Reranko05 Aug 31, 2026
e1a55e7
Revert "Remove ConsumeWhitespaces"
Reranko05 Aug 31, 2026
97da813
Address Feedback
Reranko05 Aug 31, 2026
1da3d4d
Use padded_string_builder
Reranko05 Aug 31, 2026
9935752
cleanup
Reranko05 Aug 31, 2026
1ae1290
Add ASSERT_RAISES
Reranko05 Aug 31, 2026
4772832
Lint fix
Reranko05 Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
227 changes: 140 additions & 87 deletions cpp/src/arrow/json/chunker.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
// Force parsing of the current document.
auto document_status =
internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
if (!document_status.ok()) {
return false;
}
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];

auto document = std::move(document_status).ValueUnsafe();

auto value_status =
internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
if (!value_status.ok()) {
return false;
}
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;

auto value = std::move(value_status).ValueUnsafe();
auto consume_status = internal::ConsumeJsonValue(value);
return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
if (input.size() == 0) {
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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
return std::string_view::npos;
}
size_t PutEnd(char*) {
ARROW_LOG(FATAL) << "not implemented";

auto it = stream.begin();
if (it == stream.end()) {
return 0;
}

private:
size_t index_ = 0;
std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
rj::kParseStopWhenDoneFlag |
rj::kParseNumbersAsStringsFlag;
rj::BaseReaderHandler<rj::UTF8<>> handler;
rj::Reader reader;
// parse a single JSON object
switch (reader.Parse<parse_flags>(stream, handler).Code()) {
case rj::kParseErrorNone:
return stream.Tell();
case rj::kParseErrorDocumentEmpty:
return 0;
default:
// rapidjson emitted an error, the most recent object was partial
return string_view::npos;
// Force parsing of the first document.
if (!ConsumeDocument(it)) {
return std::string_view::npos;
}

// current_index() is the start of this document. source() is the
// complete source span of the current document.
const size_t document_start = it.current_index();
const size_t document_length = it.source().size();

return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
simdjson::padded_string input;

if (partial.empty()) {
input = simdjson::padded_string(block);
} else if (block.empty()) {
input = simdjson::padded_string(partial);
} else {
simdjson::padded_string_builder builder(partial.size() + block.size());
builder.append(partial);
builder.append(block);
input = builder.convert();
}

const std::string_view input_view(input.data(), input.size());
const size_t start = ConsumeWhitespace(input_view);
if (start < input_view.size() && input_view[start] != '{' &&
input_view[start] != '[') {
return Status::Invalid("JSON chunk error: invalid data at end of document");
}

const auto length = ConsumeWholeObject(input);

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

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

// Keep the padded buffer alive while iterating the document stream.
simdjson::padded_string padded(block);
simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
*out_pos = -1;
return Status::OK();
}

auto it = stream.begin();
if (it == stream.end()) {
*out_pos = -1;
return Status::OK();
}

while (it != stream.end()) {
if (!ConsumeDocument(it)) {
break;
}
consumed_length += length;
block = block.substr(length);

consumed_length = it.current_index() + it.source().size();
++it;
}

if (consumed_length == 0) {
const size_t start = ConsumeWhitespace(block);

if (start < block.size()) {
const char first_char = block[start];

// An incomplete object/array is valid here because it may continue
// in the next block. However, non-object/array data cannot start a
// JSON record, except for a lone closing delimiter which may be the
// remainder of an incomplete value.
if (first_char != '{' && first_char != '[') {
const size_t remaining_len = block.size() - start;

if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
return Status::Invalid("JSON parse error: Invalid value");
}
}
}

*out_pos = -1;
} else {
consumed_length += ConsumeWhitespace(block);
// Check the suffix after the last complete document. This is the part
// that may contain an incomplete document spanning into the next block.
const auto remaining = block.substr(consumed_length);
const size_t start = ConsumeWhitespace(remaining);

if (start < remaining.size()) {
const char first_char = remaining[start];

if (first_char != '{' && first_char != '[') {
const size_t remaining_len = remaining.size() - start;

if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
return Status::Invalid("JSON parse error: Invalid value");
}
}
}

consumed_length += ConsumeWhitespace(remaining);
DCHECK_LE(consumed_length, block_length);
*out_pos = static_cast<int64_t>(consumed_length);
}

return Status::OK();
}

Expand Down
10 changes: 8 additions & 2 deletions cpp/src/arrow/json/chunker_test.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.
Expand Down
3 changes: 1 addition & 2 deletions python/pyarrow/tests/test_json.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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):

@pitroupitrouAug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?

@Reranko05Reranko05Aug 31, 2026

Copy link
Copy Markdown
CollaboratorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document

self.read_bytes(data, read_options=read_options,
parse_options=parse_options)

Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker by Reranko05 · Pull Request #50945 · apache/arrow · GitHub
Skip to content
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

Reranko05
wants to merge
14
commits into

base:main
Choose a base branch

from


+149

−91

 Open


GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker
#50945
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
be4c2a1

Replace RapidJSON with simdjson in JSON chunker


Reranko05 Aug 21, 2026
aa6e20d

WIP: preserve chunker behavior before iterate_many


Reranko05 Aug 24, 2026
a5b89a7

WIP: Iterate Many


Reranko05 Aug 25, 2026
3f15314

use iterate_many approach


Reranko05 Aug 27, 2026
b9af8ce

Change expectation of python test


Reranko05 Aug 27, 2026
182daa4

Address Feedback


Reranko05 Aug 29, 2026
9b70128

use ResolveSimdjsonResult


Reranko05 Aug 29, 2026
9a5cb38

Remove ConsumeWhitespaces


Reranko05 Aug 31, 2026
e1a55e7

Revert "Remove ConsumeWhitespaces"


Reranko05 Aug 31, 2026
97da813

Address Feedback


Reranko05 Aug 31, 2026
1da3d4d

Use padded_string_builder


Reranko05 Aug 31, 2026
9935752

cleanup


Reranko05 Aug 31, 2026
1ae1290

Add ASSERT_RAISES


Reranko05 Aug 31, 2026
4772832

Lint fix


Reranko05 Aug 31, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


227 changes: 140 additions & 87 deletions
cpp/src/arrow/json/chunker.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
 // Force parsing of the current document.
 auto document_status =
 internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
 if (!document_status.ok()) {
 return false;
 }
 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];

 auto document = std::move(document_status).ValueUnsafe();

 auto value_status =
 internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
 if (!value_status.ok()) {
 return false;
 }
 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;

 auto value = std::move(value_status).ValueUnsafe();
 auto consume_status = internal::ConsumeJsonValue(value);
 return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
 if (input.size() == 0) {
 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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
 return std::string_view::npos;
 }
 size_t PutEnd(char*) {
 ARROW_LOG(FATAL) << "not implemented";

 auto it = stream.begin();
 if (it == stream.end()) {
 return 0;
 }

 private:
 size_t index_ = 0;
 std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
 static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
 rj::kParseStopWhenDoneFlag |
 rj::kParseNumbersAsStringsFlag;
 rj::BaseReaderHandler<rj::UTF8<>> handler;
 rj::Reader reader;
 // parse a single JSON object
 switch (reader.Parse<parse_flags>(stream, handler).Code()) {
 case rj::kParseErrorNone:
 return stream.Tell();
 case rj::kParseErrorDocumentEmpty:
 return 0;
 default:
 // rapidjson emitted an error, the most recent object was partial
 return string_view::npos;
 // Force parsing of the first document.
 if (!ConsumeDocument(it)) {
 return std::string_view::npos;
 }

 // current_index() is the start of this document. source() is the
 // complete source span of the current document.
 const size_t document_start = it.current_index();
 const size_t document_length = it.source().size();

 return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
 simdjson::padded_string input;

 if (partial.empty()) {
 input = simdjson::padded_string(block);
 } else if (block.empty()) {
 input = simdjson::padded_string(partial);
 } else {
 simdjson::padded_string_builder builder(partial.size() + block.size());
 builder.append(partial);
 builder.append(block);
 input = builder.convert();
 }

 const std::string_view input_view(input.data(), input.size());
 const size_t start = ConsumeWhitespace(input_view);
 if (start < input_view.size() && input_view[start] != '{' &&
 input_view[start] != '[') {
 return Status::Invalid("JSON chunk error: invalid data at end of document");
 }

 const auto length = ConsumeWholeObject(input);

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

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

 // Keep the padded buffer alive while iterating the document stream.
 simdjson::padded_string padded(block);
 simdjson::ondemand::parser parser;
 simdjson::ondemand::document_stream stream;

 if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
 *out_pos = -1;
 return Status::OK();
 }

 auto it = stream.begin();
 if (it == stream.end()) {
 *out_pos = -1;
 return Status::OK();
 }

 while (it != stream.end()) {
 if (!ConsumeDocument(it)) {
 break;
 }
 consumed_length += length;
 block = block.substr(length);

 consumed_length = it.current_index() + it.source().size();
 ++it;
 }

 if (consumed_length == 0) {
 const size_t start = ConsumeWhitespace(block);

 if (start < block.size()) {
 const char first_char = block[start];

 // An incomplete object/array is valid here because it may continue
 // in the next block. However, non-object/array data cannot start a
 // JSON record, except for a lone closing delimiter which may be the
 // remainder of an incomplete value.
 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = block.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 *out_pos = -1;
 } else {
 consumed_length += ConsumeWhitespace(block);
 // Check the suffix after the last complete document. This is the part
 // that may contain an incomplete document spanning into the next block.
 const auto remaining = block.substr(consumed_length);
 const size_t start = ConsumeWhitespace(remaining);

 if (start < remaining.size()) {
 const char first_char = remaining[start];

 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = remaining.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 consumed_length += ConsumeWhitespace(remaining);
 DCHECK_LE(consumed_length, block_length);
 *out_pos = static_cast<int64_t>(consumed_length);
 }

 return Status::OK();
 }

Expand Down

10 changes: 8 additions & 2 deletions
cpp/src/arrow/json/chunker_test.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.

Expand Down

3 changes: 1 addition & 2 deletions
python/pyarrow/tests/test_json.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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):
@pitroupitrouAug 31, 2026

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?
@Reranko05Reranko05Aug 31, 2026

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document
 self.read_bytes(data, read_options=read_options,
 parse_options=parse_options)

Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker by Reranko05 · Pull Request #50945 · apache/arrow · GitHub
Skip to content
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

Reranko05
wants to merge
14
commits into

base:main
Choose a base branch

from


+149

−91

 Open


GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker
#50945
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
be4c2a1

Replace RapidJSON with simdjson in JSON chunker


Reranko05 Aug 21, 2026
aa6e20d

WIP: preserve chunker behavior before iterate_many


Reranko05 Aug 24, 2026
a5b89a7

WIP: Iterate Many


Reranko05 Aug 25, 2026
3f15314

use iterate_many approach


Reranko05 Aug 27, 2026
b9af8ce

Change expectation of python test


Reranko05 Aug 27, 2026
182daa4

Address Feedback


Reranko05 Aug 29, 2026
9b70128

use ResolveSimdjsonResult


Reranko05 Aug 29, 2026
9a5cb38

Remove ConsumeWhitespaces


Reranko05 Aug 31, 2026
e1a55e7

Revert "Remove ConsumeWhitespaces"


Reranko05 Aug 31, 2026
97da813

Address Feedback


Reranko05 Aug 31, 2026
1da3d4d

Use padded_string_builder


Reranko05 Aug 31, 2026
9935752

cleanup


Reranko05 Aug 31, 2026
1ae1290

Add ASSERT_RAISES


Reranko05 Aug 31, 2026
4772832

Lint fix


Reranko05 Aug 31, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


227 changes: 140 additions & 87 deletions
cpp/src/arrow/json/chunker.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
 // Force parsing of the current document.
 auto document_status =
 internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
 if (!document_status.ok()) {
 return false;
 }
 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];

 auto document = std::move(document_status).ValueUnsafe();

 auto value_status =
 internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
 if (!value_status.ok()) {
 return false;
 }
 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;

 auto value = std::move(value_status).ValueUnsafe();
 auto consume_status = internal::ConsumeJsonValue(value);
 return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
 if (input.size() == 0) {
 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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
 return std::string_view::npos;
 }
 size_t PutEnd(char*) {
 ARROW_LOG(FATAL) << "not implemented";

 auto it = stream.begin();
 if (it == stream.end()) {
 return 0;
 }

 private:
 size_t index_ = 0;
 std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
 static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
 rj::kParseStopWhenDoneFlag |
 rj::kParseNumbersAsStringsFlag;
 rj::BaseReaderHandler<rj::UTF8<>> handler;
 rj::Reader reader;
 // parse a single JSON object
 switch (reader.Parse<parse_flags>(stream, handler).Code()) {
 case rj::kParseErrorNone:
 return stream.Tell();
 case rj::kParseErrorDocumentEmpty:
 return 0;
 default:
 // rapidjson emitted an error, the most recent object was partial
 return string_view::npos;
 // Force parsing of the first document.
 if (!ConsumeDocument(it)) {
 return std::string_view::npos;
 }

 // current_index() is the start of this document. source() is the
 // complete source span of the current document.
 const size_t document_start = it.current_index();
 const size_t document_length = it.source().size();

 return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
 simdjson::padded_string input;

 if (partial.empty()) {
 input = simdjson::padded_string(block);
 } else if (block.empty()) {
 input = simdjson::padded_string(partial);
 } else {
 simdjson::padded_string_builder builder(partial.size() + block.size());
 builder.append(partial);
 builder.append(block);
 input = builder.convert();
 }

 const std::string_view input_view(input.data(), input.size());
 const size_t start = ConsumeWhitespace(input_view);
 if (start < input_view.size() && input_view[start] != '{' &&
 input_view[start] != '[') {
 return Status::Invalid("JSON chunk error: invalid data at end of document");
 }

 const auto length = ConsumeWholeObject(input);

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

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

 // Keep the padded buffer alive while iterating the document stream.
 simdjson::padded_string padded(block);
 simdjson::ondemand::parser parser;
 simdjson::ondemand::document_stream stream;

 if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
 *out_pos = -1;
 return Status::OK();
 }

 auto it = stream.begin();
 if (it == stream.end()) {
 *out_pos = -1;
 return Status::OK();
 }

 while (it != stream.end()) {
 if (!ConsumeDocument(it)) {
 break;
 }
 consumed_length += length;
 block = block.substr(length);

 consumed_length = it.current_index() + it.source().size();
 ++it;
 }

 if (consumed_length == 0) {
 const size_t start = ConsumeWhitespace(block);

 if (start < block.size()) {
 const char first_char = block[start];

 // An incomplete object/array is valid here because it may continue
 // in the next block. However, non-object/array data cannot start a
 // JSON record, except for a lone closing delimiter which may be the
 // remainder of an incomplete value.
 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = block.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 *out_pos = -1;
 } else {
 consumed_length += ConsumeWhitespace(block);
 // Check the suffix after the last complete document. This is the part
 // that may contain an incomplete document spanning into the next block.
 const auto remaining = block.substr(consumed_length);
 const size_t start = ConsumeWhitespace(remaining);

 if (start < remaining.size()) {
 const char first_char = remaining[start];

 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = remaining.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 consumed_length += ConsumeWhitespace(remaining);
 DCHECK_LE(consumed_length, block_length);
 *out_pos = static_cast<int64_t>(consumed_length);
 }

 return Status::OK();
 }

Expand Down

10 changes: 8 additions & 2 deletions
cpp/src/arrow/json/chunker_test.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.

Expand Down

3 changes: 1 addition & 2 deletions
python/pyarrow/tests/test_json.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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):
@pitroupitrouAug 31, 2026

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?
@Reranko05Reranko05Aug 31, 2026

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document
 self.read_bytes(data, read_options=read_options,
 parse_options=parse_options)

Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker by Reranko05 · Pull Request #50945 · apache/arrow · GitHub
Skip to content
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

Reranko05
wants to merge
14
commits into

base:main
Choose a base branch

from


+149

−91

 Open


GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker
#50945
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
be4c2a1

Replace RapidJSON with simdjson in JSON chunker


Reranko05 Aug 21, 2026
aa6e20d

WIP: preserve chunker behavior before iterate_many


Reranko05 Aug 24, 2026
a5b89a7

WIP: Iterate Many


Reranko05 Aug 25, 2026
3f15314

use iterate_many approach


Reranko05 Aug 27, 2026
b9af8ce

Change expectation of python test


Reranko05 Aug 27, 2026
182daa4

Address Feedback


Reranko05 Aug 29, 2026
9b70128

use ResolveSimdjsonResult


Reranko05 Aug 29, 2026
9a5cb38

Remove ConsumeWhitespaces


Reranko05 Aug 31, 2026
e1a55e7

Revert "Remove ConsumeWhitespaces"


Reranko05 Aug 31, 2026
97da813

Address Feedback


Reranko05 Aug 31, 2026
1da3d4d

Use padded_string_builder


Reranko05 Aug 31, 2026
9935752

cleanup


Reranko05 Aug 31, 2026
1ae1290

Add ASSERT_RAISES


Reranko05 Aug 31, 2026
4772832

Lint fix


Reranko05 Aug 31, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


227 changes: 140 additions & 87 deletions
cpp/src/arrow/json/chunker.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
 // Force parsing of the current document.
 auto document_status =
 internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
 if (!document_status.ok()) {
 return false;
 }
 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];

 auto document = std::move(document_status).ValueUnsafe();

 auto value_status =
 internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
 if (!value_status.ok()) {
 return false;
 }
 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;

 auto value = std::move(value_status).ValueUnsafe();
 auto consume_status = internal::ConsumeJsonValue(value);
 return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
 if (input.size() == 0) {
 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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
 return std::string_view::npos;
 }
 size_t PutEnd(char*) {
 ARROW_LOG(FATAL) << "not implemented";

 auto it = stream.begin();
 if (it == stream.end()) {
 return 0;
 }

 private:
 size_t index_ = 0;
 std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
 static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
 rj::kParseStopWhenDoneFlag |
 rj::kParseNumbersAsStringsFlag;
 rj::BaseReaderHandler<rj::UTF8<>> handler;
 rj::Reader reader;
 // parse a single JSON object
 switch (reader.Parse<parse_flags>(stream, handler).Code()) {
 case rj::kParseErrorNone:
 return stream.Tell();
 case rj::kParseErrorDocumentEmpty:
 return 0;
 default:
 // rapidjson emitted an error, the most recent object was partial
 return string_view::npos;
 // Force parsing of the first document.
 if (!ConsumeDocument(it)) {
 return std::string_view::npos;
 }

 // current_index() is the start of this document. source() is the
 // complete source span of the current document.
 const size_t document_start = it.current_index();
 const size_t document_length = it.source().size();

 return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
 simdjson::padded_string input;

 if (partial.empty()) {
 input = simdjson::padded_string(block);
 } else if (block.empty()) {
 input = simdjson::padded_string(partial);
 } else {
 simdjson::padded_string_builder builder(partial.size() + block.size());
 builder.append(partial);
 builder.append(block);
 input = builder.convert();
 }

 const std::string_view input_view(input.data(), input.size());
 const size_t start = ConsumeWhitespace(input_view);
 if (start < input_view.size() && input_view[start] != '{' &&
 input_view[start] != '[') {
 return Status::Invalid("JSON chunk error: invalid data at end of document");
 }

 const auto length = ConsumeWholeObject(input);

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

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

 // Keep the padded buffer alive while iterating the document stream.
 simdjson::padded_string padded(block);
 simdjson::ondemand::parser parser;
 simdjson::ondemand::document_stream stream;

 if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
 *out_pos = -1;
 return Status::OK();
 }

 auto it = stream.begin();
 if (it == stream.end()) {
 *out_pos = -1;
 return Status::OK();
 }

 while (it != stream.end()) {
 if (!ConsumeDocument(it)) {
 break;
 }
 consumed_length += length;
 block = block.substr(length);

 consumed_length = it.current_index() + it.source().size();
 ++it;
 }

 if (consumed_length == 0) {
 const size_t start = ConsumeWhitespace(block);

 if (start < block.size()) {
 const char first_char = block[start];

 // An incomplete object/array is valid here because it may continue
 // in the next block. However, non-object/array data cannot start a
 // JSON record, except for a lone closing delimiter which may be the
 // remainder of an incomplete value.
 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = block.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 *out_pos = -1;
 } else {
 consumed_length += ConsumeWhitespace(block);
 // Check the suffix after the last complete document. This is the part
 // that may contain an incomplete document spanning into the next block.
 const auto remaining = block.substr(consumed_length);
 const size_t start = ConsumeWhitespace(remaining);

 if (start < remaining.size()) {
 const char first_char = remaining[start];

 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = remaining.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 consumed_length += ConsumeWhitespace(remaining);
 DCHECK_LE(consumed_length, block_length);
 *out_pos = static_cast<int64_t>(consumed_length);
 }

 return Status::OK();
 }

Expand Down

10 changes: 8 additions & 2 deletions
cpp/src/arrow/json/chunker_test.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.

Expand Down

3 changes: 1 addition & 2 deletions
python/pyarrow/tests/test_json.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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):
@pitroupitrouAug 31, 2026

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?
@Reranko05Reranko05Aug 31, 2026

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document
 self.read_bytes(data, read_options=read_options,
 parse_options=parse_options)

Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker by Reranko05 · Pull Request #50945 · apache/arrow · GitHub
Skip to content
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

Reranko05
wants to merge
14
commits into

base:main
Choose a base branch

from


+149

−91

 Open


GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker
#50945
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
be4c2a1

Replace RapidJSON with simdjson in JSON chunker


Reranko05 Aug 21, 2026
aa6e20d

WIP: preserve chunker behavior before iterate_many


Reranko05 Aug 24, 2026
a5b89a7

WIP: Iterate Many


Reranko05 Aug 25, 2026
3f15314

use iterate_many approach


Reranko05 Aug 27, 2026
b9af8ce

Change expectation of python test


Reranko05 Aug 27, 2026
182daa4

Address Feedback


Reranko05 Aug 29, 2026
9b70128

use ResolveSimdjsonResult


Reranko05 Aug 29, 2026
9a5cb38

Remove ConsumeWhitespaces


Reranko05 Aug 31, 2026
e1a55e7

Revert "Remove ConsumeWhitespaces"


Reranko05 Aug 31, 2026
97da813

Address Feedback


Reranko05 Aug 31, 2026
1da3d4d

Use padded_string_builder


Reranko05 Aug 31, 2026
9935752

cleanup


Reranko05 Aug 31, 2026
1ae1290

Add ASSERT_RAISES


Reranko05 Aug 31, 2026
4772832

Lint fix


Reranko05 Aug 31, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


227 changes: 140 additions & 87 deletions
cpp/src/arrow/json/chunker.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
 // Force parsing of the current document.
 auto document_status =
 internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
 if (!document_status.ok()) {
 return false;
 }
 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];

 auto document = std::move(document_status).ValueUnsafe();

 auto value_status =
 internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
 if (!value_status.ok()) {
 return false;
 }
 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;

 auto value = std::move(value_status).ValueUnsafe();
 auto consume_status = internal::ConsumeJsonValue(value);
 return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
 if (input.size() == 0) {
 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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
 return std::string_view::npos;
 }
 size_t PutEnd(char*) {
 ARROW_LOG(FATAL) << "not implemented";

 auto it = stream.begin();
 if (it == stream.end()) {
 return 0;
 }

 private:
 size_t index_ = 0;
 std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
 static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
 rj::kParseStopWhenDoneFlag |
 rj::kParseNumbersAsStringsFlag;
 rj::BaseReaderHandler<rj::UTF8<>> handler;
 rj::Reader reader;
 // parse a single JSON object
 switch (reader.Parse<parse_flags>(stream, handler).Code()) {
 case rj::kParseErrorNone:
 return stream.Tell();
 case rj::kParseErrorDocumentEmpty:
 return 0;
 default:
 // rapidjson emitted an error, the most recent object was partial
 return string_view::npos;
 // Force parsing of the first document.
 if (!ConsumeDocument(it)) {
 return std::string_view::npos;
 }

 // current_index() is the start of this document. source() is the
 // complete source span of the current document.
 const size_t document_start = it.current_index();
 const size_t document_length = it.source().size();

 return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
 simdjson::padded_string input;

 if (partial.empty()) {
 input = simdjson::padded_string(block);
 } else if (block.empty()) {
 input = simdjson::padded_string(partial);
 } else {
 simdjson::padded_string_builder builder(partial.size() + block.size());
 builder.append(partial);
 builder.append(block);
 input = builder.convert();
 }

 const std::string_view input_view(input.data(), input.size());
 const size_t start = ConsumeWhitespace(input_view);
 if (start < input_view.size() && input_view[start] != '{' &&
 input_view[start] != '[') {
 return Status::Invalid("JSON chunk error: invalid data at end of document");
 }

 const auto length = ConsumeWholeObject(input);

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

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

 // Keep the padded buffer alive while iterating the document stream.
 simdjson::padded_string padded(block);
 simdjson::ondemand::parser parser;
 simdjson::ondemand::document_stream stream;

 if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
 *out_pos = -1;
 return Status::OK();
 }

 auto it = stream.begin();
 if (it == stream.end()) {
 *out_pos = -1;
 return Status::OK();
 }

 while (it != stream.end()) {
 if (!ConsumeDocument(it)) {
 break;
 }
 consumed_length += length;
 block = block.substr(length);

 consumed_length = it.current_index() + it.source().size();
 ++it;
 }

 if (consumed_length == 0) {
 const size_t start = ConsumeWhitespace(block);

 if (start < block.size()) {
 const char first_char = block[start];

 // An incomplete object/array is valid here because it may continue
 // in the next block. However, non-object/array data cannot start a
 // JSON record, except for a lone closing delimiter which may be the
 // remainder of an incomplete value.
 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = block.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 *out_pos = -1;
 } else {
 consumed_length += ConsumeWhitespace(block);
 // Check the suffix after the last complete document. This is the part
 // that may contain an incomplete document spanning into the next block.
 const auto remaining = block.substr(consumed_length);
 const size_t start = ConsumeWhitespace(remaining);

 if (start < remaining.size()) {
 const char first_char = remaining[start];

 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = remaining.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 consumed_length += ConsumeWhitespace(remaining);
 DCHECK_LE(consumed_length, block_length);
 *out_pos = static_cast<int64_t>(consumed_length);
 }

 return Status::OK();
 }

Expand Down

10 changes: 8 additions & 2 deletions
cpp/src/arrow/json/chunker_test.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.

Expand Down

3 changes: 1 addition & 2 deletions
python/pyarrow/tests/test_json.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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):
@pitroupitrouAug 31, 2026

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?
@Reranko05Reranko05Aug 31, 2026

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document
 self.read_bytes(data, read_options=read_options,
 parse_options=parse_options)

Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker by Reranko05 · Pull Request #50945 · apache/arrow · GitHub
Skip to content
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

Reranko05
wants to merge
14
commits into

base:main
Choose a base branch

from


+149

−91

 Open


GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker
#50945
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
be4c2a1

Replace RapidJSON with simdjson in JSON chunker


Reranko05 Aug 21, 2026
aa6e20d

WIP: preserve chunker behavior before iterate_many


Reranko05 Aug 24, 2026
a5b89a7

WIP: Iterate Many


Reranko05 Aug 25, 2026
3f15314

use iterate_many approach


Reranko05 Aug 27, 2026
b9af8ce

Change expectation of python test


Reranko05 Aug 27, 2026
182daa4

Address Feedback


Reranko05 Aug 29, 2026
9b70128

use ResolveSimdjsonResult


Reranko05 Aug 29, 2026
9a5cb38

Remove ConsumeWhitespaces


Reranko05 Aug 31, 2026
e1a55e7

Revert "Remove ConsumeWhitespaces"


Reranko05 Aug 31, 2026
97da813

Address Feedback


Reranko05 Aug 31, 2026
1da3d4d

Use padded_string_builder


Reranko05 Aug 31, 2026
9935752

cleanup


Reranko05 Aug 31, 2026
1ae1290

Add ASSERT_RAISES


Reranko05 Aug 31, 2026
4772832

Lint fix


Reranko05 Aug 31, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


227 changes: 140 additions & 87 deletions
cpp/src/arrow/json/chunker.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
 // Force parsing of the current document.
 auto document_status =
 internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
 if (!document_status.ok()) {
 return false;
 }
 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];

 auto document = std::move(document_status).ValueUnsafe();

 auto value_status =
 internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
 if (!value_status.ok()) {
 return false;
 }
 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;

 auto value = std::move(value_status).ValueUnsafe();
 auto consume_status = internal::ConsumeJsonValue(value);
 return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
 if (input.size() == 0) {
 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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
 return std::string_view::npos;
 }
 size_t PutEnd(char*) {
 ARROW_LOG(FATAL) << "not implemented";

 auto it = stream.begin();
 if (it == stream.end()) {
 return 0;
 }

 private:
 size_t index_ = 0;
 std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
 static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
 rj::kParseStopWhenDoneFlag |
 rj::kParseNumbersAsStringsFlag;
 rj::BaseReaderHandler<rj::UTF8<>> handler;
 rj::Reader reader;
 // parse a single JSON object
 switch (reader.Parse<parse_flags>(stream, handler).Code()) {
 case rj::kParseErrorNone:
 return stream.Tell();
 case rj::kParseErrorDocumentEmpty:
 return 0;
 default:
 // rapidjson emitted an error, the most recent object was partial
 return string_view::npos;
 // Force parsing of the first document.
 if (!ConsumeDocument(it)) {
 return std::string_view::npos;
 }

 // current_index() is the start of this document. source() is the
 // complete source span of the current document.
 const size_t document_start = it.current_index();
 const size_t document_length = it.source().size();

 return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
 simdjson::padded_string input;

 if (partial.empty()) {
 input = simdjson::padded_string(block);
 } else if (block.empty()) {
 input = simdjson::padded_string(partial);
 } else {
 simdjson::padded_string_builder builder(partial.size() + block.size());
 builder.append(partial);
 builder.append(block);
 input = builder.convert();
 }

 const std::string_view input_view(input.data(), input.size());
 const size_t start = ConsumeWhitespace(input_view);
 if (start < input_view.size() && input_view[start] != '{' &&
 input_view[start] != '[') {
 return Status::Invalid("JSON chunk error: invalid data at end of document");
 }

 const auto length = ConsumeWholeObject(input);

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

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

 // Keep the padded buffer alive while iterating the document stream.
 simdjson::padded_string padded(block);
 simdjson::ondemand::parser parser;
 simdjson::ondemand::document_stream stream;

 if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
 *out_pos = -1;
 return Status::OK();
 }

 auto it = stream.begin();
 if (it == stream.end()) {
 *out_pos = -1;
 return Status::OK();
 }

 while (it != stream.end()) {
 if (!ConsumeDocument(it)) {
 break;
 }
 consumed_length += length;
 block = block.substr(length);

 consumed_length = it.current_index() + it.source().size();
 ++it;
 }

 if (consumed_length == 0) {
 const size_t start = ConsumeWhitespace(block);

 if (start < block.size()) {
 const char first_char = block[start];

 // An incomplete object/array is valid here because it may continue
 // in the next block. However, non-object/array data cannot start a
 // JSON record, except for a lone closing delimiter which may be the
 // remainder of an incomplete value.
 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = block.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 *out_pos = -1;
 } else {
 consumed_length += ConsumeWhitespace(block);
 // Check the suffix after the last complete document. This is the part
 // that may contain an incomplete document spanning into the next block.
 const auto remaining = block.substr(consumed_length);
 const size_t start = ConsumeWhitespace(remaining);

 if (start < remaining.size()) {
 const char first_char = remaining[start];

 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = remaining.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 consumed_length += ConsumeWhitespace(remaining);
 DCHECK_LE(consumed_length, block_length);
 *out_pos = static_cast<int64_t>(consumed_length);
 }

 return Status::OK();
 }

Expand Down

10 changes: 8 additions & 2 deletions
cpp/src/arrow/json/chunker_test.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.

Expand Down

3 changes: 1 addition & 2 deletions
python/pyarrow/tests/test_json.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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):
@pitroupitrouAug 31, 2026

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?
@Reranko05Reranko05Aug 31, 2026

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document
 self.read_bytes(data, read_options=read_options,
 parse_options=parse_options)

Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
})();
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker by Reranko05 · Pull Request #50945 · apache/arrow · GitHub
Skip to content
GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker#50945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

Reranko05
wants to merge
14
commits into

base:main
Choose a base branch

from


+149

−91

 Open


GH-50944: [C++] Replace RapidJSON with simdjson in JSON chunker
#50945
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
be4c2a1

Replace RapidJSON with simdjson in JSON chunker


Reranko05 Aug 21, 2026
aa6e20d

WIP: preserve chunker behavior before iterate_many


Reranko05 Aug 24, 2026
a5b89a7

WIP: Iterate Many


Reranko05 Aug 25, 2026
3f15314

use iterate_many approach


Reranko05 Aug 27, 2026
b9af8ce

Change expectation of python test


Reranko05 Aug 27, 2026
182daa4

Address Feedback


Reranko05 Aug 29, 2026
9b70128

use ResolveSimdjsonResult


Reranko05 Aug 29, 2026
9a5cb38

Remove ConsumeWhitespaces


Reranko05 Aug 31, 2026
e1a55e7

Revert "Remove ConsumeWhitespaces"


Reranko05 Aug 31, 2026
97da813

Address Feedback


Reranko05 Aug 31, 2026
1da3d4d

Use padded_string_builder


Reranko05 Aug 31, 2026
9935752

cleanup


Reranko05 Aug 31, 2026
1ae1290

Add ASSERT_RAISES


Reranko05 Aug 31, 2026
4772832

Lint fix


Reranko05 Aug 31, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


227 changes: 140 additions & 87 deletions
cpp/src/arrow/json/chunker.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,105 +17,77 @@

#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());
static bool ConsumeDocument(simdjson::ondemand::document_stream::iterator& it) {
 // Force parsing of the current document.
 auto document_status =
 internal::ResolveSimdjsonResult(*it, "Failed to get JSON document");
 if (!document_status.ok()) {
 return false;
 }
 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];

 auto document = std::move(document_status).ValueUnsafe();

 auto value_status =
 internal::ResolveSimdjsonResult(document.get_value(), "Failed to get JSON value");
 if (!value_status.ok()) {
 return false;
 }
 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;

 auto value = std::move(value_status).ValueUnsafe();
 auto consume_status = internal::ConsumeJsonValue(value);
 return consume_status.ok();
}

static size_t ConsumeWholeObject(const simdjson::padded_string& input) {
 if (input.size() == 0) {
 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;

simdjson::ondemand::parser parser;
simdjson::ondemand::document_stream stream;

if (parser.iterate_many(input).get(stream) != simdjson::SUCCESS) {
 return std::string_view::npos;
 }
 size_t PutEnd(char*) {
 ARROW_LOG(FATAL) << "not implemented";

 auto it = stream.begin();
 if (it == stream.end()) {
 return 0;
 }

 private:
 size_t index_ = 0;
 std::vector<string_view> strings_;
};

template <typename Stream>
static size_t ConsumeWholeObject(Stream&& stream) {
 static constexpr unsigned parse_flags = rj::kParseIterativeFlag |
 rj::kParseStopWhenDoneFlag |
 rj::kParseNumbersAsStringsFlag;
 rj::BaseReaderHandler<rj::UTF8<>> handler;
 rj::Reader reader;
 // parse a single JSON object
 switch (reader.Parse<parse_flags>(stream, handler).Code()) {
 case rj::kParseErrorNone:
 return stream.Tell();
 case rj::kParseErrorDocumentEmpty:
 return 0;
 default:
 // rapidjson emitted an error, the most recent object was partial
 return string_view::npos;
 // Force parsing of the first document.
 if (!ConsumeDocument(it)) {
 return std::string_view::npos;
 }

 // current_index() is the start of this document. source() is the
 // complete source span of the current document.
 const size_t document_start = it.current_index();
 const size_t document_length = it.source().size();

 return document_start + document_length;
}

namespace {
Expand All@@ -124,40 +96,121 @@ 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 {
 simdjson::padded_string input;

 if (partial.empty()) {
 input = simdjson::padded_string(block);
 } else if (block.empty()) {
 input = simdjson::padded_string(partial);
 } else {
 simdjson::padded_string_builder builder(partial.size() + block.size());
 builder.append(partial);
 builder.append(block);
 input = builder.convert();
 }

 const std::string_view input_view(input.data(), input.size());
 const size_t start = ConsumeWhitespace(input_view);
 if (start < input_view.size() && input_view[start] != '{' &&
 input_view[start] != '[') {
 return Status::Invalid("JSON chunk error: invalid data at end of document");
 }

 const auto length = ConsumeWholeObject(input);

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

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

 // Keep the padded buffer alive while iterating the document stream.
 simdjson::padded_string padded(block);
 simdjson::ondemand::parser parser;
 simdjson::ondemand::document_stream stream;

 if (parser.iterate_many(padded).get(stream) != simdjson::SUCCESS) {
 *out_pos = -1;
 return Status::OK();
 }

 auto it = stream.begin();
 if (it == stream.end()) {
 *out_pos = -1;
 return Status::OK();
 }

 while (it != stream.end()) {
 if (!ConsumeDocument(it)) {
 break;
 }
 consumed_length += length;
 block = block.substr(length);

 consumed_length = it.current_index() + it.source().size();
 ++it;
 }

 if (consumed_length == 0) {
 const size_t start = ConsumeWhitespace(block);

 if (start < block.size()) {
 const char first_char = block[start];

 // An incomplete object/array is valid here because it may continue
 // in the next block. However, non-object/array data cannot start a
 // JSON record, except for a lone closing delimiter which may be the
 // remainder of an incomplete value.
 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = block.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 *out_pos = -1;
 } else {
 consumed_length += ConsumeWhitespace(block);
 // Check the suffix after the last complete document. This is the part
 // that may contain an incomplete document spanning into the next block.
 const auto remaining = block.substr(consumed_length);
 const size_t start = ConsumeWhitespace(remaining);

 if (start < remaining.size()) {
 const char first_char = remaining[start];

 if (first_char != '{' && first_char != '[') {
 const size_t remaining_len = remaining.size() - start;

 if (remaining_len > 1 || (first_char != '}' && first_char != ']')) {
 return Status::Invalid("JSON parse error: Invalid value");
 }
 }
 }

 consumed_length += ConsumeWhitespace(remaining);
 DCHECK_LE(consumed_length, block_length);
 *out_pos = static_cast<int64_t>(consumed_length);
 }

 return Status::OK();
 }

Expand Down

10 changes: 8 additions & 2 deletions
cpp/src/arrow/json/chunker_test.cc

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -264,9 +264,15 @@ 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);
Comment thread
Reranko05 marked this conversation as resolved.

Expand Down

3 changes: 1 addition & 2 deletions
python/pyarrow/tests/test_json.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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):
@pitroupitrouAug 31, 2026

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the actual error message in this case?
@Reranko05Reranko05Aug 31, 2026

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON chunk error: invalid data at end of document
 self.read_bytes(data, read_options=read_options,
 parse_options=parse_options)

Expand Down
Loading