How ClickHouse Parses CSV in Parallel

TL;DR: A CSV file cannot be split at arbitrary byte offsets, because a quoted field may contain the delimiter. So ClickHouse first runs a segmentation engine — a byte-level scanner that knows only where rows end — cuts the file into row-aligned chunks, and then hands each chunk to a separate thread. A ring of “processing units” numbered by ticket keeps the output in file order.

Parsing CSV is embarrassingly parallel in theory and surprisingly awkward in practice. The work per row is small — find fields, convert to a column, append — but the loop is strictly sequential, so a 10-core box uses one core to import a 10 GB file.

The naive way to parallelise it is to cut the file into N equal byte ranges and parse each range independently. That is wrong for CSV, and wrong in a way that is easy to miss, because it only fails on some files:

text
  a,b,c            <-- cut here
  1,2,3
  4,"hello, wor   <-- cut here: the field is split in half
  ld",6

A cut in the middle of a quoted field means one chunk sees 4,"hello, wor and the next sees ld",6. The first chunk’s parser will happily report a malformed row; the second will report three columns instead of two.

So the first question is: how do you find a safe cut point? And the answer that ClickHouse uses is instructive, because it draws a sharp line between two different kinds of knowledge about CSV.

Two kinds of knowledge

A CSV parser needs to know:

Syntactic knowledge — where does this row end? Which bytes are inside a quoted field?
Semantic knowledge — is 2024-01-01 a date, a string, or an integer? Is this column Nullable(Int64)?

Only the first kind is needed to find cut points. And the first kind is much simpler: it is a byte-level state machine that tracks whether you are inside quotes, and nothing else. It never needs to know the schema, the column types, or even how many columns the file has.

ClickHouse puts that state machine in a separate object — the file segmentation engine — that is registered per format:

cpp
// src/Processors/Formats/Impl/CSVRowInputFormat.cpp:898
void registerFileSegmentationEngineCSV(FormatFactory & factory);
void registerFileSegmentationEngineCSV(FormatFactory & factory)
{
    auto register_func = [&](const String & format_name, bool, bool)
    {
        static constexpr size_t min_rows = 3; /// Make it 3 for header auto detection (first 3 rows must be always in the same segment).
        factory.registerFileSegmentationEngineCreator(format_name, [](const FormatSettings & settings) -> FormatFactory::FileSegmentationEngine
        {
            return [settings] (ReadBuffer & in, DB::Memory<> & memory, size_t min_bytes, size_t max_rows)
            {
                return fileSegmentationEngineCSVImpl(in, memory, min_bytes, min_rows, max_rows, settings);
            };
        });
    };

    registerWithNamesAndTypes("CSV", register_func);
    markFormatWithNamesAndTypesSupportsSamplingColumns("CSV", factory);
}

That min_rows = 3 is a small but important detail: header auto-detection needs to look at the first two rows to decide whether they are a header, and the header detection rolls back if it decides they are not. If the first two rows were split across two segments, the second segment’s parser would have no header to detect and would parse the header row as data. So the first segment must contain at least three rows.

The segmentation engine

The engine itself is a quote-aware scanner over raw bytes:

cpp
// src/Processors/Formats/Impl/CSVRowInputFormat.cpp:826
std::pair<bool, size_t> fileSegmentationEngineCSVImpl(ReadBuffer & in, DB::Memory<> & memory, size_t min_bytes, size_t min_rows, size_t max_rows, const FormatSettings & settings)
{
    char * pos = in.position();
    bool quotes = false;
    bool need_more_data = true;
    size_t number_of_rows = 0;

    if (max_rows && (max_rows < min_rows))
        max_rows = min_rows;

    while (loadAtPosition(in, memory, pos) && need_more_data)
    {
        if (quotes)
        {
            pos = find_first_symbols<'"'>(pos, in.buffer().end());
            if (pos > in.buffer().end())
                throw Exception(ErrorCodes::LOGICAL_ERROR, "Position in buffer is out of bounds. There must be a bug.");
            if (pos == in.buffer().end())
                continue;
            if (*pos == '"')
            {
                ++pos;
                if (loadAtPosition(in, memory, pos) && *pos == '"')
                    ++pos;
                else
                    quotes = false;
            }
        }
        else
        {
            pos = find_first_symbols<'"', '\r', '\n'>(pos, in.buffer().end());
            if (pos > in.buffer().end())
                throw Exception(ErrorCodes::LOGICAL_ERROR, "Position in buffer is out of bounds. There must be a bug.");
            if (pos == in.buffer().end())
                continue;

            if (*pos == '"')
            {
                quotes = true;
                ++pos;
                continue;
            }

            if (*pos == '\n')
            {
                ++pos;
                if (loadAtPosition(in, memory, pos) && *pos == '\r')
                    ++pos;
            }
            else if (*pos == '\r')
            {
                ++pos;
                if (settings.csv.allow_cr_end_of_line)
                    continue;
                if (loadAtPosition(in, memory, pos) && *pos == '\n')
                    ++pos;
                else
                    continue;
            }

            ++number_of_rows;
            if ((number_of_rows >= min_rows)
                && ((memory.size() + static_cast<size_t>(pos - in.position()) >= min_bytes) || (number_of_rows == max_rows)))
                need_more_data = false;

        }
    }

    saveUpToPosition(in, memory, pos);
    return {loadAtPosition(in, memory, pos), number_of_rows};
}

Read it as a two-state machine:

Not in quotes. Scan forward for the next ", \r or \n. A " opens a quoted field. A \n (or \r\n) ends the row.
In quotes. Scan forward for the next ". If the byte after it is another ", that is an escaped quote and the field continues; otherwise the field ends and we go back to “not in quotes”.

The continue statements are the “I ran out of buffer, get more data and resume” path. That is what loadAtPosition does — it copies the bytes scanned so far out of the read buffer into a Memory<> and refills:

cpp
// src/IO/ReadHelpers.cpp:2316
bool loadAtPosition(ReadBuffer & in, Memory<> & memory, char * & current)
{
    chassert(current <= in.buffer().end());

    if (current < in.buffer().end())
        return true;

    saveUpToPosition(in, memory, current);

    bool loaded_more = !in.eof();
    // A sanity check. Buffer position may be in the beginning of the buffer
    // (normal case), or have some offset from it (AIO).
    chassert(in.position() >= in.buffer().begin());
    chassert(in.position() <= in.buffer().end());
    current = in.position();

    return loaded_more;
}

Note the asymmetry: the scanner consumes bytes from the read buffer and accumulates them into memory, which becomes the segment handed to a parser thread. At the end, saveUpToPosition flushes the last chunk of scanned bytes into memory and advances the read buffer’s cursor past them:

cpp
// src/IO/ReadHelpers.cpp:2297
void saveUpToPosition(ReadBuffer & in, Memory<> & memory, char * current)
{
    chassert(current >= in.position());
    chassert(current <= in.buffer().end());

    const size_t old_bytes = memory.size();
    const size_t additional_bytes = current - in.position();
    const size_t new_bytes = old_bytes + additional_bytes;

    /// Nothing to append, and `in.position()` may be null.
    if (additional_bytes == 0)
        return;

    chassert(in.position() + additional_bytes <= in.buffer().end());
    memory.resizeAmortized(new_bytes);
    memcpy(memory.data() + old_bytes, in.position(), additional_bytes);
    in.position() = current;
}

So the segmentation engine is also the thing that reads the file. The segmentator thread does the I/O and the byte scanning; the parser threads only ever see a Memory<> that is already a whole number of rows.

One more subtlety: the quoted-field handling above does not check whether the quote is at the start of a field. a"b,c"d is not valid CSV, but the scanner treats the " as opening a quoted region. That is deliberate — the scanner’s job is to find a safe cut point, and being conservative (treating more bytes as quoted) is safe. Being liberal would be not.

The three-thread pipeline

With cut points found, the parallel parsing machinery is a small pipeline. ParallelParsingInputFormat has three kinds of threads, and the class comment draws it:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.h:33
 *                    Parsers
 *      |   |   |   |   |   |   |   |   |   |
 *      v   v   v   v   v   v   v   v   v   v
 *    |---|---|---|---|---|---|---|---|---|---|
 *    | 1 | 2 | 3 | 4 | 5 | . | . | . | . | N | <-- Processing units
 *    |---|---|---|---|---|---|---|---|---|---|
 *      ^               ^
 *      |               |
 *   readImpl        Segmentator

and the operating procedure:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.h:54
 * The outline of what the threads do is as follows:
 * segmentator thread:
 *  1) wait for the next processing unit to become empty
 *  2) fill it with a part of input file
 *  3) start a parser thread
 *  4) repeat until eof
 * parser thread:
 *  1) parse the given raw buffer without any synchronization
 *  2) signal that the given unit is ready to read
 *  3) finish
 * readImpl():
 *  1) wait for the next processing unit to become ready to read
 *  2) take the blocks from the processing unit to return them to the caller
 *  3) signal that the processing unit is empty
 *  4) repeat until it encounters unit that is marked as "past_the_end"

The “processing units” are a ring buffer. Each unit holds a Memory<> of raw bytes plus the chunks parsed out of it:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.h:264
    struct ProcessingUnit
    {
        ProcessingUnit()
            : status(ProcessingUnitStatus::READY_TO_INSERT)
        {
        }

        ChunkExt chunk_ext;
        Memory<> segment;
        size_t original_segment_size{};
        std::atomic<ProcessingUnitStatus> status;
        /// Needed for better exception message.
        size_t offset = 0;
        bool is_last{false};
    };

with three states — READY_TO_INSERT (empty, waiting for the segmentator), READY_TO_PARSE (filled, waiting for a parser), READY_TO_READ (parsed, waiting for the reader). The unit has three possible owners at any time, and the state is the ownership token. The ring has a couple of extra slots so the segmentator does not bump into the reader on wraparound:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.h:103
        // One unit for each thread, including segmentator and reader, plus a
        // couple more units so that the segmentation thread doesn't spuriously
        // bump into reader thread on wraparound.
        processing_units.resize(params.max_threads + 2);

The segmentator

The segmentator is a single thread that owns the input stream:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:11
void ParallelParsingInputFormat::segmentatorThreadFunction(ThreadGroupPtr thread_group)
{
    ThreadGroupSwitcher switcher(thread_group, ThreadName::PARALLEL_PARSING_SEGMENTATOR);

    try
    {
        while (!parsing_finished)
        {
            const auto segmentator_unit_number = segmentator_ticket_number % processing_units.size();
            auto & unit = processing_units[segmentator_unit_number];

            {
                std::unique_lock<std::mutex> lock(mutex);
                segmentator_condvar.wait(lock, [&] { return unit.status == READY_TO_INSERT || parsing_finished; });
            }

            if (parsing_finished)
                break;

            chassert(unit.status == READY_TO_INSERT);

            // Segmentating the original input.
            unit.segment.resize(0);

            size_t segment_start = getDataOffsetMaybeCompressed(*in);
            auto file_segmentation_engine = file_segmentation_engine_creator(format_settings);
            auto [have_more_data, currently_read_rows] = file_segmentation_engine(*in, unit.segment, min_chunk_bytes, max_block_size);

            unit.original_segment_size = getDataOffsetMaybeCompressed(*in) - segment_start;
            unit.offset = successfully_read_rows_count;
            successfully_read_rows_count += currently_read_rows;

            unit.is_last = !have_more_data;
            unit.status = READY_TO_PARSE;
            scheduleParserThreadForUnitWithNumber(segmentator_ticket_number);
            ++segmentator_ticket_number;

The ticket number is the ordering key. Unit k is filled from the k-th segment of the file, and the reader consumes units in ticket order, so the output order matches the input order regardless of which parser finishes first.

The parser

Each parser thread gets a fresh input format over the segment’s memory:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:75
        ReadBuffer read_buffer(unit.segment.data(), unit.segment.size(), 0);

        InputFormatPtr input_format = internal_parser_creator(read_buffer);
        input_format->setRowsReadBefore(unit.offset);
        input_format->setErrorsLogger(errors_logger);
        input_format->setSerializationHints(serialization_hints);

        InternalParser parser(input_format);

and then drains it into a vector of chunks:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:92
        // We don't know how many blocks will be. So we have to read them all
        // until an empty block occurred.
        Chunk chunk;
        while (!parsing_finished && (chunk = parser.getChunk()))
        {
            /// Variable chunk is moved, but it is not really used in the next iteration.
            /// NOLINTNEXTLINE(bugprone-use-after-move, hicpp-invalid-access-moved)
            unit.chunk_ext.chunk.emplace_back(std::move(chunk));

No synchronisation at all inside the parser: it owns its unit exclusively, and the only shared state it touches at the end is the status flag.

There is one piece of state that genuinely must be shared: the column mapping. For a CSVWithNames file, the header is in the first segment, and every other segment needs to know which file column maps to which result column. So the first parser extracts it and publishes it:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:113
        /// Extract column_mapping from first parser to propagate it to others
        if (current_ticket_number == 0)
        {
            column_mapping = input_format->getColumnMapping();
            column_mapping->is_set = true;
            first_parser_finished.set();
        }

and the segmentator explicitly waits for that before it lets the other parsers run:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.h:290
    void scheduleParserThreadForUnitWithNumber(size_t ticket_number)
    {
        runner.enqueueAndKeepTrack([this, ticket_number]()
        {
            parserThreadFunction(ticket_number);
        });
        /// We have to wait here to possibly extract ColumnMappingPtr from the first parser.
        if (ticket_number == 0)
            first_parser_finished.wait();
    }

This is a real serialisation point in the pipeline, but it costs one row’s worth of parsing.

The reader

The reader is the thread that calls readImpl(), and it is where the ordering is enforced:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:174
    const auto inserter_unit_number = reader_ticket_number % processing_units.size();
    auto * unit = &processing_units[inserter_unit_number];

    if (!next_block_in_current_unit.has_value())
    {
        while (true)
        {
            // We have read out all the Blocks from the previous Processing Unit,
            // wait for the current one to become ready.
            std::unique_lock<std::mutex> lock(mutex);
            reader_condvar.wait(lock, [&]() { return unit->status == READY_TO_READ || parsing_finished; });

The reader waits for unit reader_ticket_number % N specifically — not for any unit to be ready. If unit 3 finishes before unit 2, the reader still waits for unit 2. That is the whole ordering mechanism: a ring indexed by ticket number, plus a condition variable.

When a unit is drained, it is handed back to the segmentator:

cpp
// src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:233
    if (*next_block_in_current_unit == unit->chunk_ext.chunk.size())
    {
        // parsing_finished reading this Processing Unit, move to the next one.
        next_block_in_current_unit.reset();
        ++reader_ticket_number;

        if (unit->is_last)
        {
            // It it was the last unit, we're parsing_finished.
            parsing_finished = true;
        }
        else
        {
            // Pass the unit back to the segmentator.
            std::lock_guard lock(mutex);
            unit->status = READY_TO_INSERT;
            segmentator_condvar.notify_all();
        }
    }

This is also the backpressure mechanism. The segmentator can only produce segments as fast as the reader consumes them, because the ring is finite: with max_threads + 2 units, at most that many segments are in flight, so memory use is bounded by (max_threads + 2) * min_chunk_bytes.

When parallel parsing is used at all

The decision is made once, in FormatFactory:

cpp
// src/Formats/FormatFactory.cpp:608
    // Decide whether to use ParallelParsingInputFormat.

    size_t max_parsing_threads = parser_shared_resources->getParsingThreadsPerReader();
    bool parallel_parsing = max_parsing_threads > 1 && settings[Setting::input_format_parallel_parsing]
        && creators.file_segmentation_engine_creator && !(creators.random_access_input_creator || creators.random_access_input_creator_with_metadata)
        && !need_only_count;

    if (settings[Setting::max_memory_usage]
        && settings[Setting::min_chunk_bytes_for_parallel_parsing] * max_parsing_threads * 2 > settings[Setting::max_memory_usage])
        parallel_parsing = false;

Three conditions kill it: the format has no segmentation engine, the format needs random access (Parquet, Arrow — they have their own readers), or the memory budget cannot hold 2 * threads * chunk_bytes. And the chunk size itself is a setting:

cpp
// src/Core/Settings.cpp:1861
    DECLARE(NonZeroUInt64, min_chunk_bytes_for_parallel_parsing, (10 * 1024 * 1024), R"(
- Type: unsigned int
- Default value: 1 MiB

The minimum chunk size in bytes, which each thread will parse in parallel.
)", 0) \

10 MiB per thread by default. That is a lot of memory for a chunk, and it is deliberate: the per-chunk overhead (thread wakeup, buffer allocation, the first rows of the segment being a header-detection warm-up) is amortised over the chunk, and the segment is parsed into many blocks, so the reader gets a steady supply of small blocks even though the parser is chewing on 10 MiB.

(Amusingly, the setting’s own description says “Default value: 1 MiB” — the description is stale; the code says 10 * 1024 * 1024.)

All of this is reachable from SQL:

sql
SELECT count()
FROM file('events.csv', CSVWithNames)
SETTINGS input_format_parallel_parsing = 1,
         max_parsing_threads = 8,
         min_chunk_bytes_for_parallel_parsing = 1048576;

The field scanner

Now the second half of the story: how a chunk is parsed. The row loop is row-major over a column-major store, and the field loop is where the bytes are actually examined.

cpp
// src/Processors/Formats/RowInputFormatWithNamesAndTypes.cpp:297
    else
    {
        for (size_t file_column = 0; file_column < column_mapping->column_indexes_for_input_fields.size(); ++file_column)
        {
            if (file_column != 0)
                format_reader->skipFieldDelimiter();

            const auto & column_index = column_mapping->column_indexes_for_input_fields[file_column];
            const bool is_last_file_column = file_column + 1 == column_mapping->column_indexes_for_input_fields.size();
            if (column_index)
                ext.read_columns[*column_index] = format_reader->readField(
                    *columns[*column_index],
                    data_types[*column_index],
                    serializations[*column_index],
                    is_last_file_column,
                    column_mapping->names_of_columns[file_column],
                    *column_index);
            else
                format_reader->skipField(file_column);
        }
    }

The CSV reader’s readField does the whitespace trimming and then hands the buffer to the type’s serialiser:

cpp
// src/Processors/Formats/Impl/CSVRowInputFormat.cpp:413
bool CSVFormatReader::readFieldImpl(ReadBuffer & istr, DB::IColumn & column, const DB::DataTypePtr & type, const DB::SerializationPtr & serialization)
{
    if (format_settings.null_as_default && !isNullableOrLowCardinalityNullable(type)
        && !isCSVSeparateColumnsTuple(type, format_settings))
    {
        /// If value is null but type is not nullable then use default value instead.
        return SerializationNullable::deserializeNullAsDefaultOrNestedTextCSV(column, istr, format_settings, serialization);
    }

    /// Read the column normally.
    serialization->deserializeTextCSV(column, istr, format_settings);
    return true;
}

This is the division of labour that makes the whole thing work: the format layer knows about quoting and delimiters; the type layer knows how to turn text into a column value. Int64 parses digits, DateTime parses a date string, LowCardinality(String) interns the string it just read.

The actual byte scanning lives in readCSVStringInto, and it has two paths. The unquoted path is a SIMD scan for the delimiter, CR or LF:

cpp
// src/IO/ReadHelpers.cpp:1124
        /// Unquoted case. Look for delimiter or \r (followed by '\n') or \n.
        while (!buf.eof())
        {
            char * next_pos = buf.position();

            [&]()
            {
#ifdef __SSE2__
                auto rc = _mm_set1_epi8('\r');
                auto nc = _mm_set1_epi8('\n');
                auto dc = _mm_set1_epi8(delimiter);
                for (; next_pos + 15 < buf.buffer().end(); next_pos += 16)
                {
                    __m128i bytes = _mm_loadu_si128(reinterpret_cast<const __m128i *>(next_pos));
                    auto eq = _mm_or_si128(_mm_or_si128(_mm_cmpeq_epi8(bytes, rc), _mm_cmpeq_epi8(bytes, nc)), _mm_cmpeq_epi8(bytes, dc));
                    uint16_t bit_mask = static_cast<uint16_t>(_mm_movemask_epi8(eq));
                    if (bit_mask)
                    {
                        next_pos += std::countr_zero(bit_mask);
                        return;
                    }
                }

16 bytes per iteration, three comparisons OR-ed together, and movemask + countr_zero to find the first interesting byte. (There is an NEON version right below it with the same shape.)

The quoted path is different in kind: it looks for the closing quote, and it has to handle the doubled-quote escape:

cpp
// src/IO/ReadHelpers.cpp:1045
        /// The quoted case. We are looking for the next quotation mark.
        while (!buf.eof())
        {
            char * next_pos = reinterpret_cast<char *>(memchr(buf.position(), maybe_quote, buf.buffer().end() - buf.position()));

            if (nullptr == next_pos)
                next_pos = buf.buffer().end();

            appendToStringOrVector(s, buf, next_pos);
            buf.position() = next_pos;

            if (!buf.hasPendingData())
                continue;

            if constexpr (include_quotes)
                s.push_back(maybe_quote);

            /// Now there is a quotation mark under the cursor. Is there any following?
            ++buf.position();

            if (buf.eof())
                return;

            if (*buf.position() == maybe_quote)
            {
                s.push_back(maybe_quote);
                ++buf.position();
                continue;
            }

            return;
        }

memchr for the quote, then a one-byte look-ahead to distinguish "" (an escaped quote, stay inside) from " (the end of the field). Note that this path uses memchr and not the SIMD loop — a quoted field is expected to be long, but the terminator is a single byte, and memchr is already vectorised in libc.

Whitespace trimming, decided once

A subtle performance detail: CSV allows spaces and tabs around fields, and ClickHouse has a setting for whether to trim them. But trimming must not apply to a String column (where the space might be data). So the decision is per column, and it is computed once per block rather than per field:

cpp
// src/Processors/Formats/Impl/CSVRowInputFormat.cpp:359
void CSVFormatReader::setDataTypes(const DataTypes & types)
{
    skip_whitespaces_before_field.resize(types.size());
    for (size_t i = 0; i != types.size(); ++i)
        skip_whitespaces_before_field[i]
            = format_settings.csv.trim_whitespaces || !isStringOrFixedString(removeNullable(types[i]));
}

and then consulted as a one-byte lookup:

cpp
// src/Processors/Formats/Impl/CSVRowInputFormat.cpp:367
bool CSVFormatReader::readField(
    IColumn & column,
    const DataTypePtr & type,
    const SerializationPtr & serialization,
    bool is_last_file_column,
    const String & /*column_name*/,
    size_t column_index)
{
    chassert(column_index < skip_whitespaces_before_field.size());
    if (skip_whitespaces_before_field[column_index]) [[likely]]
        skipWhitespacesAndTabs(*buf, format_settings.csv.allow_whitespace_or_tab_as_delimiter);

The rollback, one more time

Header auto-detection is the other place where the peek buffer earns its keep. The reader wants to know whether the first row is a header. It reads it, checks whether the values look like column names, and if not, unreads it:

cpp
// src/Processors/Formats/RowInputFormatWithNamesAndTypes.cpp:157
template <typename FormatReaderImpl>
void RowInputFormatWithNamesAndTypes<FormatReaderImpl>::tryDetectHeader(std::vector<String> & column_names_out, std::vector<String> & type_names_out)
{
    auto & read_buf = getReadBuffer();
    PeekableReadBuffer * peekable_buf = dynamic_cast<PeekableReadBuffer *>(&read_buf);
    if (!peekable_buf)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Header detection is supported only for formats that use PeekableReadBuffer");

    /// Empty data.
    if (unlikely(format_reader->checkForSuffix()))
    {
        end_of_stream = true;
        return;
    }

    /// Make a checkpoint before reading the first row.
    peekable_buf->setCheckpoint();
    auto first_row_values = format_reader->readRowForHeaderDetection();

and if the first row turns out to be data after all:

cpp
// src/Processors/Formats/RowInputFormatWithNamesAndTypes.cpp:182
    if (!isSubsetOf(first_row_values_set, column_names_set) && !isSubsetOf(column_names_set, first_row_values_set))
    {
        /// Rollback to the beginning of the first row to parse it as data later.
        peekable_buf->rollbackToCheckpoint(true);
        return;
    }

The true argument means “drop the checkpoint as well”, so the parser continues from the restored position. The second row is treated the same way (it might be a row of type names). This is the PeekableReadBuffer trick in its purest form: speculate, and undo.

Trade-offs

What the design buys:

Linear scaling with cores, for the part of the work that is genuinely parallel. Parsing a 10 GB CSV on a 16-core box goes roughly 10x faster than on one core.
Order preservation for free. The ticket ring means no reordering buffer, no sorting, no sequence numbers in the output.
Bounded memory. max_threads + 2 units, each capped by min_chunk_bytes, and the reader hands units back.
A clean separation of concerns. The segmentation engine knows about quotes and newlines; the type serialisers know about dates and integers; neither knows about the other.

What it costs:

The segmentation engine is a second, redundant parser. It re-scans the bytes looking for quotes and newlines — work the real parser does again. For a file of quoted fields, that is a meaningful constant factor.
A serialisation point. The first parser must publish the column mapping before the others can proceed, so the pipeline starts with a small stall.
Memory. 10 MiB per thread by default, times the number of threads. The FormatFactory explicitly disables parallel parsing when the memory budget cannot absorb that.
The scanner is deliberately conservative. It treats any " as opening a quoted region, so a file with stray quotes produces segments that are larger than necessary — but never smaller, which is the direction that matters.
Only formats with a segmentation engine benefit. Parquet, Arrow and friends have their own parallel readers; JSONEachRow has its own; the machinery applies to the row-oriented text formats.

And one thing worth calling out: the whole scheme rests on a distinction that is easy to get wrong. Row boundaries are a syntactic property of the file and can be found by looking at bytes. Column types are a semantic property and cannot. A parallel CSV parser that tries to be clever about types at the cutting stage is trying to do two things at once, and will get one of them wrong.