From 7f2bd4d39ed12c1ff9608d6e7b3d3ee4227fae2f Mon Sep 17 00:00:00 2001 From: Vincent La Date: Sat, 10 Oct 2026 11:33:27 -0700 Subject: [PATCH 1/2] Fix mmap reader yielding zero rows when the first row exceeds the head buffer MmapParser parsed its 500KB head buffer as the first window regardless of chunk_size. When that window held no complete row, trim_header() ran against an empty queue, marked the header as trimmed, and the reader silently produced no column names and no rows. - MmapParser::next() now falls through to a full read window after parsing the head, so the first read covers the head plus one window like StreamParser. - trim_header() tracks how many header rows it has popped and only marks the header trimmed once the header row itself is consumed. - initial_read() keeps reading until the header is consumed or EOF, throwing the existing "row larger than chunk size" error if a read makes no progress. This also fixes a header row that starts after the first read window ends, which affected both the mmap and stream paths. Fixes #337 Co-Authored-By: Claude Opus 5.5 --- include/internal/csv_reader.cpp | 28 +++- include/internal/csv_reader.hpp | 13 +- include/internal/parser/mmap.cpp | 9 +- tests/test_edge_cases_large_rows.cpp | 183 +++++++++++++++++++++++++++ 4 files changed, 223 insertions(+), 10 deletions(-) diff --git a/include/internal/csv_reader.cpp b/include/internal/csv_reader.cpp index 56c61a00..378cd1f0 100644 --- a/include/internal/csv_reader.cpp +++ b/include/internal/csv_reader.cpp @@ -71,18 +71,40 @@ namespace csv { return new_format; } + CSV_INLINE void CSVReader::initial_read() { + // Issue #337: A single read window may end before the header row when + // earlier rows are long, so keep reading until the header is consumed. + do { + const int trimmed_before = this->header_rows_trimmed_; + this->read_scheduler_.run([this] { this->read_csv(this->_chunk_size); }); + this->read_scheduler_.join(); + this->read_scheduler_.rethrow_exception_if_any(); + + // trim_header() drains the queue until the header is consumed, so an + // untrimmed header with no progress means the read produced no rows. + if (!this->header_trimmed && !this->parser->eof() + && this->header_rows_trimmed_ == trimmed_before) { + internals::throw_row_too_large_for_chunk(this->_chunk_size); + } + } while (!this->header_trimmed && !this->parser->eof()); + } + CSV_INLINE void CSVReader::trim_header() { if (!this->header_trimmed) { - for (int i = 0; i <= this->_format.header && !this->records->empty(); i++) { - if (i == this->_format.header && this->col_names->empty()) { + // The rows up to and including the header may arrive over several + // reads, so track progress instead of assuming one read holds them all. + while (this->header_rows_trimmed_ <= this->_format.header && !this->records->empty()) { + if (this->header_rows_trimmed_ == this->_format.header && this->col_names->empty()) { this->set_col_names(this->records->pop_front()); } else { this->records->pop_front(); } + + this->header_rows_trimmed_++; } - this->header_trimmed = true; + this->header_trimmed = this->header_rows_trimmed_ > this->_format.header; } } diff --git a/include/internal/csv_reader.hpp b/include/internal/csv_reader.hpp index 699ae076..3d8e5d6c 100644 --- a/include/internal/csv_reader.hpp +++ b/include/internal/csv_reader.hpp @@ -369,6 +369,9 @@ namespace csv { private: /** Whether or not rows before header were trimmed */ bool header_trimmed = false; + + /** How many rows up to and including the header have been trimmed */ + int header_rows_trimmed_ = 0; /** @name Reader Scheduling: Flags and State */ ///@{ @@ -386,6 +389,7 @@ namespace csv { this->n_cols = other.n_cols; this->_n_rows = other._n_rows; this->header_trimmed = other.header_trimmed; + this->header_rows_trimmed_ = other.header_rows_trimmed_; this->_chunk_size = other._chunk_size; this->_read_requested = other._read_requested; this->read_scheduler_.set_threading_enabled(this->_format.is_threading_enabled()); @@ -397,6 +401,7 @@ namespace csv { this->n_cols = 0; this->_n_rows = 0; this->header_trimmed = false; + this->header_rows_trimmed_ = 0; this->_read_requested = false; this->_chunk_size = internals::CSV_CHUNK_SIZE_DEFAULT; } @@ -421,12 +426,8 @@ namespace csv { ); } - /** Read initial chunk to get metadata */ - void initial_read() { - this->read_scheduler_.run([this] { this->read_csv(this->_chunk_size); }); - this->read_scheduler_.join(); - this->read_scheduler_.rethrow_exception_if_any(); - } + /** Read until the header row is consumed (or EOF) to get metadata */ + void initial_read(); void trim_header(); ///@} diff --git a/include/internal/parser/mmap.cpp b/include/internal/parser/mmap.cpp index 5dd32b00..3e6f1551 100644 --- a/include/internal/parser/mmap.cpp +++ b/include/internal/parser/mmap.cpp @@ -76,7 +76,14 @@ namespace csv { this->mmap_pos += length; this->finalize_loaded_chunk(*head_owner, head_owner, length, bytes); - return; + if (this->eof_) { + return; + } + + // Issue #337: The head is capped at 500KB regardless of chunk_size, + // so it may end before the first row does. Fall through and map a + // full read window from the rewound position so the first read + // covers the head plus one window, matching StreamParser. } // Create memory map diff --git a/tests/test_edge_cases_large_rows.cpp b/tests/test_edge_cases_large_rows.cpp index 460b414c..e4191d63 100644 --- a/tests/test_edge_cases_large_rows.cpp +++ b/tests/test_edge_cases_large_rows.cpp @@ -236,6 +236,189 @@ TEST_CASE("Edge case: CSV rows larger than default chunk size", "[edge_cases_lar } } +TEST_CASE("Issue #337 - First row longer than the mmap head buffer", "[issue_337]") { + // MmapParser parses its pre-read head buffer (capped at 500KB regardless of + // chunk_size) as the first window. When that window held no complete row, + // trim_header() ran against an empty queue, marked the header as trimmed, + // and the reader silently produced zero columns and zero rows. The stream + // path was unaffected because its first window is the head plus a full read. + const size_t n_rows = 1000; + std::string body; + for (size_t i = 0; i < n_rows; ++i) { + body += std::to_string(i * 3 + 0) + ',' + + std::to_string(i * 3 + 1) + ',' + + std::to_string(i * 3 + 2) + '\n'; + } + + // Compare without letting Catch2 expand the huge column name on failure. + auto require_col_names = [](CSVReader& reader, const std::string& first) { + const auto names = reader.get_col_names(); + REQUIRE(names.size() == 3); + REQUIRE(names[0].size() == first.size()); + REQUIRE((names[0] == first)); + REQUIRE(names[1] == "B"); + REQUIRE(names[2] == "C"); + }; + + SECTION("Header row fits in chunk_size") { + const std::string long_name = "A" + std::string(600 * 1024, 'x'); + const std::string data = long_name + ",B,C\n" + body; + + CSVFormat format; + format.delimiter(',').header_row(0).chunk_size(2 * 1024 * 1024).speculative_parallel_threads(1); + + auto validate_reader = [&](CSVReader& reader) { + require_col_names(reader, long_name); + + size_t i = 0; + for (auto& row : reader) { + REQUIRE(row.size() == 3); + for (size_t col = 0; col < 3; ++col) { + REQUIRE(row[col].get() == i * 3 + col); + } + ++i; + } + REQUIRE(i == n_rows); + }; + + SECTION("stream path") { + std::istringstream in(data); + CSVReader reader(in, format); + validate_reader(reader); + } + + SECTION("mmap path") { + FileGuard cleanup("./tests/data/tmp_issue_337_long_header.csv"); + { + std::ofstream out(cleanup.filename, std::ios::binary); + out << data; + } + CSVReader reader(cleanup.filename, format); + validate_reader(reader); + } + } + + SECTION("Header row after a short preamble row") { + // header_row(1): the head window holds the complete preamble row but + // not the long header row, so header trimming must resume on a later read. + const std::string long_name = "A" + std::string(600 * 1024, 'x'); + const std::string data = "preamble\n" + long_name + ",B,C\n" + body; + + CSVFormat format; + format.delimiter(',').header_row(1).chunk_size(2 * 1024 * 1024).speculative_parallel_threads(1); + + auto validate_reader = [&](CSVReader& reader) { + require_col_names(reader, long_name); + + size_t i = 0; + for (auto& row : reader) { + REQUIRE(row.size() == 3); + REQUIRE(row[0].get() == i * 3); + ++i; + } + REQUIRE(i == n_rows); + }; + + SECTION("stream path") { + std::istringstream in(data); + CSVReader reader(in, format); + validate_reader(reader); + } + + SECTION("mmap path") { + FileGuard cleanup("./tests/data/tmp_issue_337_preamble.csv"); + { + std::ofstream out(cleanup.filename, std::ios::binary); + out << data; + } + CSVReader reader(cleanup.filename, format); + validate_reader(reader); + } + } + + SECTION("Header row starts after the first read window ends") { + // header_row(2) with long preamble rows: the first read window ends + // inside the header row on both paths, so trim_header() must not mark + // the header as trimmed after popping only the preamble rows. + // The 700KB window starting at byte 0 (stream) or after "short\n" + // (mmap) ends inside the 150KB header row. + const std::string preamble = "preamble," + std::string(600 * 1024, 'p') + '\n'; + const std::string long_name = "A" + std::string(150 * 1024, 'x'); + const std::string data = "short\n" + preamble + long_name + ",B,C\n" + body; + + CSVFormat format; + format.delimiter(',') + .header_row(2) + .chunk_size(internals::CSV_CHUNK_SIZE_FLOOR + 200 * 1024) + .speculative_parallel_threads(1); // Keep the stream window at chunk_size + + auto validate_reader = [&](CSVReader& reader) { + require_col_names(reader, long_name); + + size_t i = 0; + for (auto& row : reader) { + REQUIRE(row.size() == 3); + REQUIRE(row[2].get() == i * 3 + 2); + ++i; + } + REQUIRE(i == n_rows); + }; + + SECTION("stream path") { + std::istringstream in(data); + CSVReader reader(in, format); + validate_reader(reader); + } + + SECTION("mmap path") { + FileGuard cleanup("./tests/data/tmp_issue_337_late_header.csv"); + { + std::ofstream out(cleanup.filename, std::ios::binary); + out << data; + } + CSVReader reader(cleanup.filename, format); + validate_reader(reader); + } + } + + SECTION("Header row larger than chunk_size still throws") { + const std::string data = "A" + std::string(1200 * 1024, 'x') + ",B,C\n" + body; + + CSVFormat format; + format.delimiter(',').header_row(0).chunk_size(internals::CSV_CHUNK_SIZE_FLOOR).speculative_parallel_threads(1); + + auto read_all = [](CSVReader& reader) { + for (auto& row : reader) { (void)row; } + }; + + SECTION("stream path") { + REQUIRE_THROWS_WITH( + [&]() { + std::istringstream in(data); + CSVReader reader(in, format); + read_all(reader); + }(), + Catch::Matchers::ContainsSubstring("chunk size") + ); + } + + SECTION("mmap path") { + FileGuard cleanup("./tests/data/tmp_issue_337_too_long.csv"); + { + std::ofstream out(cleanup.filename, std::ios::binary); + out << data; + } + REQUIRE_THROWS_WITH( + [&]() { + CSVReader reader(cleanup.filename, format); + read_all(reader); + }(), + Catch::Matchers::ContainsSubstring("chunk size") + ); + } + } +} + TEST_CASE("Issue #218 - Infinite read loop detection", "[issue_218]") { SECTION("Detects when row exceeds chunk size and file doesn't end") { From f890eca7e4d31ef26c6c69c707973fd47dc87475 Mon Sep 17 00:00:00 2001 From: Vincent La Date: Sat, 10 Oct 2026 11:49:36 -0700 Subject: [PATCH 2/2] Only read past the mmap head when it holds no complete row The unconditional fallthrough after parsing the head made construction parse an extra full read window even when the head already produced rows, which penalized metadata queries and early-stop readers (500K-row file: ~60ms vs ~12ms to construct). Fall through only when the head made no progress; CSVReader::initial_read() keeps reading when the header row is still pending. Co-Authored-By: Claude Opus 5.5 --- include/internal/parser/mmap.cpp | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/include/internal/parser/mmap.cpp b/include/internal/parser/mmap.cpp index 3e6f1551..feb14ce5 100644 --- a/include/internal/parser/mmap.cpp +++ b/include/internal/parser/mmap.cpp @@ -72,18 +72,21 @@ namespace csv { // for delimiter/header guessing. if (!head_.empty()) { auto head_owner = std::make_shared(std::move(head_)); + const size_t head_start = this->mmap_pos; const size_t length = head_owner->size(); this->mmap_pos += length; this->finalize_loaded_chunk(*head_owner, head_owner, length, bytes); - if (this->eof_) { - return; - } // Issue #337: The head is capped at 500KB regardless of chunk_size, - // so it may end before the first row does. Fall through and map a - // full read window from the rewound position so the first read - // covers the head plus one window, matching StreamParser. + // so it may end before the first row does. Only in that case fall + // through and map a full read window from the rewound position; + // otherwise return so construction doesn't parse an extra window + // (CSVReader::initial_read() keeps reading if the header row is + // still pending). + if (this->eof_ || this->mmap_pos != head_start) { + return; + } } // Create memory map