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..feb14ce5 100644 --- a/include/internal/parser/mmap.cpp +++ b/include/internal/parser/mmap.cpp @@ -72,11 +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); - return; + + // Issue #337: The head is capped at 500KB regardless of chunk_size, + // 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 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") {