Calculate valid partition divisions to avoid parser EOF error when CSV rows contain unescaped newlines

[lmmx]

This follows discussion in #4415 on the parser error encountered (‘EOF within string’) when partitioning a file with unescaped newlines on a row. The partition naively splits on newlines (from what I understand?), which is only a valid assumption for files without unescaped within-row EOL characters.

dask/dask/dataframe/io/csv.py

Lines 704 to 707 in e6b9e89

	It should also be noted that this function may fail if a {file_type} file
	includes quoted strings that contain the line terminator. To get around this
	you can specify ``blocksize=None`` to not split files into multiple partitions,
	at the cost of reduced parallelism.

Would it be possible to ‘assist’ dask in identifying where to partition a CSV somehow?

From the above issue I surmise that this is not currently a feature (but if it is and nobody thought to suggest it there please let me know).

I can imagine a function that:

• opens the file,
• seeks to the would-be partition offset minus some chunk size (expected to cover one or more lines),
• reads in a chunk
• identifies the latest row ending position within that chunk
• passes that chunk to dask as the partition point
• [repeat until file is completely partitioned]

(I recently wrote a function that ‘greps backwards’ within a buffer, here, which seems in the same spirit)

This could be done to identify the partition positions and then pass these into dask, or dask could provide a helper method to generate these positions.

Alternatively, if dask’s partition positions are accessible, these could be modified after calculation, but I presume manually doing so would be undesirable (primarily as if all the partition positions moved backwards then the final one would risk being greater than the max. permitted partition size)

I tried to load the WIT dataset which has newlines within rows (i.e. unescaped free text, from Wikipedia image captions) and cannot get the advantage of cuDF speedup over pandas due to the problem of distributing the computation (in the way that dask achieves with partitions).

I think I’d be able to speed this up if dask could successfully partition with such a routine, but I don’t know where to put it (i.e. I don’t see the “select where to partition” aspect of dask exposed in the API, I presume it is not)

Another alternative I can envision is to try/except reading one row at the start of each partition (skipping the first as this is not post-partition), and if it fails with the EOF error then move that partition to the previous newline and repeat

---

Minimal example:

• First get the sample file (here I also decompressed it)
  • wget https://storage.googleapis.com/gresearch/wit/wit_v1.train.all-1percent_sample.tsv.gz

import dask.dataframe as dd
from pathlib import Path
import time

p = Path("wit_v1.train.all-1percent_sample.tsv")
print(f"Data file: {p.name}")

t0 = time.time()
dtype_dict = {"mime_type": "category"}
fields = [*dtype_dict]
df = dd.read_csv(p, sep="\t", dtype=dtype_dict, usecols=fields)
filetypes = list(df.mime_type)
t1 = time.time()
print(f"dask took: {t1-t0}s")

⇣

Data file: wit_v1.train.all-1percent_sample.tsv
Traceback (most recent call last):
  File "dask_test.py", line 12, in <module>
    filetypes = list(df.mime_type)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/dataframe/core.py", line 579, in __len__
    return self.reduction(
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/base.py", line 286, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/base.py", line 568, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/threaded.py", line 79, in get
    results = get_async(
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/local.py", line 514, in get_async
    raise_exception(exc, tb)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/local.py", line 325, in reraise
    raise exc
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/local.py", line 223, in execute_task
    result = _execute_task(task, data)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/optimization.py", line 969, in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/core.py", line 151, in get
    result = _execute_task(task, cache)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/core.py", line 121, in <genexpr>
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/dataframe/io/csv.py", line 125, in __call__
    df = pandas_read_text(
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/dask/dataframe/io/csv.py", line 178, in pandas_read_text
    df = reader(bio, **kwargs)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/pandas/util/_decorators.py", line 311, in wrapper
    return func(*args, **kwargs)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/pandas/io/parsers/readers.py", line 586, in read_csv
    return _read(filepath_or_buffer, kwds)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/pandas/io/parsers/readers.py", line 488, in _read
    return parser.read(nrows)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/pandas/io/parsers/readers.py", line 1047, in read
    index, columns, col_dict = self._engine.read(nrows)
  File "/home/louis/miniconda3/envs/dask_test/lib/python3.8/site-packages/pandas/io/parsers/c_parser_wrapper.py", line 223, in read
    chunks = self._reader.read_low_memory(nrows)
  File "pandas/_libs/parsers.pyx", line 801, in pandas._libs.parsers.TextReader.read_low_memory
  File "pandas/_libs/parsers.pyx", line 857, in pandas._libs.parsers.TextReader._read_rows
  File "pandas/_libs/parsers.pyx", line 843, in pandas._libs.parsers.TextReader._tokenize_rows
  File "pandas/_libs/parsers.pyx", line 1925, in pandas._libs.parsers.raise_parser_error
pandas.errors.ParserError: Error tokenizing data. C error: EOF inside string starting at row 34111

(The exact row number given varies)

• This refers to an error thrown when reader (i.e. pd.read_csv) reads the bytes buffer bio

  dask/dask/dataframe/io/csv.py

  Lines 173 to 178 in 5cb8961

  	bio = BytesIO()
  	if write_header and not b.startswith(header.rstrip()):
  	bio.write(header)
  	bio.write(b)
  	bio.seek(0)
  	df = reader(bio, **kwargs)

• via

  dask/dask/dataframe/io/csv.py

  Lines 125 to 135 in 5cb8961

  	df = pandas_read_text(
  	self.reader,
  	block,
  	self.header,
  	rest_kwargs,
  	self.dtypes,
  	columns,
  	write_header,
  	self.enforce,
  	path_info,
  	)

• In turn this CsvFunctionWrapper call comes from text_blocks_to_pandas getting called in the return statement of read_pandas (which gets put in the dask.dataframe namespace , from dd.io, from dd.io.csv where it's a wrapper on make_reader using pd.read_csv as its reader arg)

  • i.e. it simplifies to just pd.read_csv, passing a bytes buffer corresponding to the block's bytes within the file rather than the input file

I can't see any way of inspecting the divisions that are computed here (as they're not coming from a pre-existing dataframe but created from the TSV, hence no index to display in the divisions)

>>> df.divisions
(None, None, None, None, None, None, None, None, None, None, None, None)
>>> df.known_divisions
False

but since it's bytes buffers under the hood you could check the divisions by seeking to the end of these buffers and performing the routine described above.

The blocks in question are the values passed as the block_lists arg to text_blocks_to_pandas after being constructed in read_pandas:

dask/dask/dataframe/io/csv.py

Line 620 in 5cb8961

values = [[list(dsk.dask.values()) for dsk in block] for block in values]

so read_pandas is where the change I am suggesting would go.

The values in turn were unpacked from b_out, which comes from read_bytes (from dask.bytes.core):

dask/dask/dataframe/io/csv.py

Lines 547 to 562 in 5cb8961

	b_out = read_bytes(
	urlpath,
	delimiter=b_lineterminator,
	blocksize=blocksize,
	sample=sample,
	compression=compression,
	include_path=include_path_column,
	**(storage_options or {}),
	)
	if include_path_column:
	b_sample, values, paths = b_out
	path = (include_path_column, path_converter)
	else:
	b_sample, values = b_out
	path = None

The offsets are derived from these block sizes without any line delimiter detection:

dask/dask/bytes/core.py

Lines 115 to 116 in 5cb8961

	off = list(range(0, size, blocksize))
	length = [blocksize] * len(off)

and then very importantly the blocks are split according to the delimiter here (the delimiter was passed as the byte-encoded line terminator):

dask/dask/bytes/core.py

Lines 126 to 138 in 5cb8961

	for path, offset, length in zip(paths, offsets, lengths):
	token = tokenize(fs_token, delimiter, path, fs.ukey(path), compression, offset)
	keys = ["read-block-%s-%s" % (o, token) for o in offset]
	values = [
	delayed_read(
	OpenFile(fs, path, compression=compression),
	o,
	l,
	delimiter,
	dask_key_name=key,
	)
	for o, key, l in zip(offset, keys, length)
	]

After which they are ready to be passed on to the delayed_read which then becomes values a.k.a. block_list.

It's that step of splitting the blocks according to the delimiter that I think could be modified to ensure rows are parsed correctly.

2 further aspects to this:

• Within the delayed_read function call is OpenFile (i.e. executed and then passed into delayed_read)
• delayed_read is itself an alias:

  dask/dask/bytes/core.py

  Line 123 in 5cb8961

  delayed_read = delayed(read_block_from_file)

  • delayed is the lazy computation proxy (not relevant to what's done here)
  • read_blocks_from_file means that the computation being delayed is to read the block passed with 4 arguments: lazy_file, off, bs, delimiter

    dask/dask/bytes/core.py

    Lines 168 to 172 in 5cb8961

    	def read_block_from_file(lazy_file, off, bs, delimiter):
    	with copy.copy(lazy_file) as f:
    	if off == 0 and bs is None:
    	return f.read()
    	return read_block(f, off, bs, delimiter)

  • The lazy_file = an OpenFile object via fsspec.core
  • The off offset is the unmodified offset (a multiple of the block size from list(range(0, size, block_size))
  • The bs is the l, an item in lengths, which will always be block_size except for the last block which may be shorter if the file size is not an exact multiple of the block_size
  • read_block is called when l in lengths is set [by virtue of block_size being set, as it is here] to anything other than None, and this function comes from fsspec.utils.
• The signature and docstring show that its default behaviour is split_before=False, hence it starts/stops reading after the delimiter [newline] at offset and offset+length. This is how the newlines are detected.

---

So long story short:

• Roughly speaking [assuming file is evenly split into blocks of equal size], dd.read_csv reduces to

for o in range(0, size, dd.io.csv.AUTO_BLOCK_SIZE):
  fsspec.read_block(f=input_csv, offset=o, length=dd.io.csv.AUTO_BLOCKSIZE,, delimiter="\n", split_before=False)

• This respects line separators but not necessarily row-delimiting line separators
• Extending this procedure to respect row-delimiting line separators would make dask usable with a larger subset of the CSV files pandas can read

---

The problem remains somewhat tricky to specify (how to verify that the newline at a given offset into a CSV is/isn’t the start of a valid row and partition at the next one)

---

I thought about this some more and the three approaches I can think of are:

1. Let the user provide a custom checking function to be used to verify the buffer bytes at a position are valid

• In the example dataset I am using, this would be very simple: the first column must equal a lower case two letter language code followed by the separator, and so on. This would make it very simple to check that the bytes after the line separator at a given offset was indeed the start of a row. (Or rather, to rule out invalid rows)

2. Let the user provide a list of rows which may contain free text (including newlines, i.e. the "separable columns", those which may be separated across lines), so as to validate based on an attempted separation (reading as many bytes as necessary, including line separator/s, until fulfilling the expected number of columns). Alternatively, these may be discerned from the sample.
3. Simply have pandas 'stream' a row, and catch any parsing error as a failure

The 1st option would potentially involve reading fewer bytes (thus faster).

The 2nd option seems the simplest to achieve, to identify the correct "delimiter offset" for the given partition offset (that is, the number of newline delimiters away from the offset at which the next row-delimiting newline is found). If the first newline delimiter after the partition offset was a valid start of row then the 'delimiter offset' would be 0, if it was the one after then it'd be 1, and so on.

The 3rd option would be most 'foolproof' (and perhaps simplest).

I need to think about this some more (I suspect it may need a mix of the 3 approaches). Any suggestions/comments welcome

[martindurant]

As you can see, @lmmx , it's not a simple problem, and there's a reason that it's been around a long time. Your suggestions may be interesting, but hard to evaluate without some case studies. Providing custom offsets for read_block that have been obtained in some other way is certainly doable, and providing custom validation of blocks is not totally unreasonable either. However, the read_csv function is already massively overloaded with arguments, so I fear that any additions would be very carefully scrutinised.

I have an orthogonal suggestion from another project I am working on, fsspec-reference-maker. The idea is, to make a "references" file (JSON), in which each entry corresponds to a specific bytes block of some other URL. This set of reference blocks can be viewed as a filesystem via fsspec - which dask already understands. Thus, you only need to find valid offsets and encode them, and then everything else will work out fine - without touching Dask's code. I imagine you could add a CSV module to fsspec-reference-maker which uses pandas (or csv directly) to stream through the target file and identify valid offsets on the way, with any custom validation. This only need happen once, and once you have the references JSON, Dask need never know that something special is going on.

[lmmx]

I'm with you on all of the above. Regarding not overloading read_csv, I propose that the routine I'm outlining here can reuse the sample_rows=10 argument.

That said, not sure about how else this check could be switched off (if it isn't switched off, then rows would be read from every offset, greedily, which may be undesirable for known-valid CSVs, e.g. those without multiline strings within rows). However I'll see what it looks like, I suspect any buffering 'cost' would be negligible so it wouldn't matter.

If no new options are to be added to read_csv, then partitioning a CSV with 2 or more will incur a reading cost of 10 rows of bytes per partition, using the csv module (not pandas). I hope this can be done efficiently and will need to benchmark to make sure I don't cause a performance regression.

To be able to turn it off, I would suggest adding a sample_partition_rows=10 argument, such that sample_partition_rows=0 would turn off this check (but for now will operate based on reusing sample_rows).

Content warning(?) -- long documentation follows in the rest of this post, perhaps only primarily of interest during implementation. Please don't feel obliged to wade through if the level of detail is too much, I just needed to flesh it out to be sure enough how to implement it.

Please do feel free to pick up on any or none of the discussion below 😺

Part 0: Background on how pandas parses CSVs

Click for background on how pandas parses CSVs

The lineterminator argument to dask.dataframe.io.csv.read_pandas is already copied from pandas.read_csv (see docs) so there is room to also use the arguments -- even if simply extracted [copied] from kwargs -- for quoting:

• quotechar='"'
• quoting=0
• doublequote=True

N.B. doublequote does not toggle the quotechar as being the double quotation mark character(!)

quotechar : str (length 1), optional
The character used to denote the start and end of a quoted item. Quoted
items can include the delimiter and it will be ignored.

quoting : int or csv.QUOTE_* instance, default 0
Control field quoting behavior per csv.QUOTE_* constants. Use one of
QUOTE_MINIMAL (0), QUOTE_ALL (1), QUOTE_NONNUMERIC (2) or QUOTE_NONE (3).

doublequote : bool, default True
When quotechar is specified and quoting is not QUOTE_NONE, indicate
whether or not to interpret two consecutive quotechar elements INSIDE a
field as a single quotechar element.

and the code itself takes the following approach:

• read_csv(file_or_buf, **kwargs) --> _read(file_or_buf, kwds=kwargs) --> TextFileReader(file_or_buf, kwds=kwds)
• [Regardless of engine] the parser_defaults keys are iterated through (including the 3 mentioned above) and used to populate the options dict [initialised as {}] which is returned into the init method of TextFileReader and then assigned on its options attribute after passing through its _clean_options method
  • in the _clean_options method, if the quotechar is non-ASCII then the C engine will fallback to the Python engine

To take the example of PythonParser (which must presumably match the procedure of the C parser, but is readable in the Python code)

• self.quoting is checked not to be csv.QUOTING_NONE in _rows_to_cols,
• which sets alldata in the read method
• which [for whichever parser engine] TextFileReader wraps through its read method

Part 1: Problem statement

I've thought about how it would be achieved, and I suggest sampling from an identical number of rows as sampled for the file head (sample in read_bytes), at the tail of each partition [backwards from each candidate newline delimiter], to consume an identical number of rows up to the row-delimiting newline that first follows the end of the partition (sample_rows=10).

I've also road tested some approaches to going about this, and concluded that pandas will not be able to achieve it (the Python csv module is needed instead, due to the way pandas conflates absent and empty values as "missing"/NaN).

Click for further details on the pandas parser (and its shortcomings)

It's important to go to the lengths of 10 rows, so as to be sure to avoid any potential edge case in which a quoted multiline field value could mimic one or more entire rows. This is already the default for sampling rows at the file head, as in the following edge case:

>>> pd.read_csv(io.BytesIO(b'intA,intB,strC\n1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"'))
   intA  intB                    strC
0     1     2                   hello
1     3     4                   world
2     5     6  foo\n7,8,bar\n9,10,baz

in which the final row [index 2] expands to:

5,6,"foo
7,8,bar
9,10,baz"

which dask could, by some odd chance, land on and, since it fulfils the requirements of the row, assume came from:

>>> pd.read_csv(io.BytesIO(b'intA,intB,strC\n7,8,bar'))
   intA  intB strC
0     7     8  bar

which would quietly induce an error upon parsing at the next partition's start (when the quotechar would be encountered, giving the parser EOF error this issue opened with).

Note that this edge case relies on the validity of inconsistent quoting of a column. This corresponds to the Python csv module docs (“field” means “item in a row”):

csv.QUOTE_MINIMAL¶
Instructs writer objects to only quote those fields which contain special characters such as delimiter, quotechar or any of the characters in lineterminator.

I think sampling 10 rows from the tail of each partition per file is adequate to catch any such edge case and so to validate the next partition.

I think they should be sampled backwards from the first row-delimiting newline after the partition end, as is done for the head [except the head is sampled forwards, only once per file].

This check can be skipped for the final partition -- as of course there is no choice of newline delimiters following the EOF to iterate past so the EOF itself must be accepted as a row-delimiting newline, else the file is corrupt/invalid [out of scope of this feature]

• A file of 1 partition does not have to check a partition tail,
• A file of 2 partitions only has to check the first partition's tail.

It should also fail fast (i.e. move to the next lineseparator delimiter ASAP), by catching the same pandas parser error containing the 'EOF' substring as done when computing the head:

dask/dask/dataframe/io/csv.py

Lines 593 to 596 in e6b9e89

	try:
	head = reader(BytesIO(b_sample), nrows=sample_rows, **head_kwargs)
	except pd.errors.ParserError as e:
	if "EOF" in str(e):

but where the file head routine raises a ValueError from the caught exception, the partition tail routine should instead move onto the next candidate lineseparator delimiter (with seek_delimiter) and retry, only failing if it reaches EOF (i.e. the routine has proceeded through the entirety of the final partition without finding a new row)

• This would mean that what was thought to be the final partition was actually entirely one of the earlier examples of a misleading column. This could only conceivably happen if the partitions were very small and the quoted column value was very long [spanning many lines]. I wouldn't expect this to be cause for concern.

Another major difference between the file head and partition tail routines: the partition tails are to be read backwards [from the first newline after the offset, and if that isn't a row-delimiting newline then the next after that, etc], whereas the file head is simply read forwards.

One aspect which I can't see any problem with (other than it's weird) is that the only sensible way to read backwards is to also reverse the bytes buffer.

Luckily I recently wrote just such a routine for iterating backwards through newlines in a buffer which can be adapted for this task. I'm confident it is correct, and it is well documented and commented for clarity (I will remove the type annotations when reusing it to fit the dask house style).

• I can't think of any of the data types which would not be appropriate to do this for
  • strings (including categoricals), integers, floats, etc are all reversible and still valid
  • datetime dtype cannot be created without converters [to my knowledge/my attempts to produce them]
  • "valid" here only means syntax, not semantics (the correct number of columns per row)

e.g. take the previous example, and pretend that the dataframe there was the sample (giving us the column names)

>>> bytestr = b'intA,intB,strC\n1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"'
>>> pd.read_csv(io.BytesIO(bytestr))
   intA  intB                    strC
0     1     2                   hello
1     3     4                   world
2     5     6  foo\n7,8,bar\n9,10,baz
>>> pd.read_csv(io.BytesIO(bytestr[::-1]))
  zab,01,9\nrab,8,7\noof     6     5
0                  dlrow     4     3
1                  olleh     2     1
2                   Crts  Btni  Atni

• Note here how the first partition in a file runs the risk of including the 1st line of a file (the header). This however can be easily detected: if the tail partition reads back to the start of the file (easy to check from offset and the tail buffer length), then only in this case would it be acceptable to read the 10 partition tail rows going forward instead of reverse (however in that case surely you'd just use the sample instead, as it'd be equivalent)

Now ignore the header row and just look at the first 2 rows of data, and pretend that the dataframe earlier was the sample (so we now have the column names)

>>> rows_bytestr = b'1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"'
>>> rev_rows_b = rows_bytestr[::-1]
>>> pd.read_csv(io.BytesIO(rev_rows_b), header=None).iloc[::-1, ::-1].T.reset_index(drop=True).T.reset_index(drop=True)
   0  1                       2
0  1  2                   olleh
1  3  4                   dlrow
2  5  6  zab,01,9\nrab,8,7\noof

Note that the column names should not be passed to read_csv here: if they are, invalid lines will be salvaged (padded with NaN) whereas without providing column names parser errors are thrown as you would want in a validation step.

>>> bad_rows_bytestr = rows_bytestr + b"\nJUST_ONE_COLUMN_NOT_THREE"
>>> bad_rev_b = bad_rows_bytestr[::-1]
>>> pd.read_csv(io.BytesIO(bad_rev_b), header=None).iloc[::-1, ::-1].T.reset_index(drop=True).T.reset_index(drop=True)
Traceback (most recent call last):
...
  File "pandas/_libs/parsers.pyx", line 1951, in pandas._libs.parsers.raise_parser_error
pandas.errors.ParserError: Error tokenizing data. C error: Expected 1 fields in line 2, saw 3

whereas if you provide names (let's imagine they come from the sample):

>>> sample_df = pd.read_csv(io.BytesIO(bytestr))
>>> cols_rev = sample_df.columns[::-1]
>>> pd.read_csv(io.BytesIO(bad_rev_b), names=cols_rev, header=None).iloc[::-1, ::-1].T.reset_index(drop=True).T.reset_index(drop=True)
     0    1                          2
0  1.0  2.0                      olleh
1  3.0  4.0                      dlrow
2  5.0  6.0     zab,01,9\nrab,8,7\noof
3  NaN  NaN  EERHT_TON_NMULOC_ENO_TSUJ

• The invalid single column line gets padded with NaN rather than throwing an error!
• integers got quietly coerced to floats

So don't pass in column names to this validation step.

Let's say we reach our target number of rows (instead of 10, let's call it 3, so the above example from rev_rows_b is completed), and the next newline only has part of a row (again due to an intra-row linesep):

>>> extra_line_bytestr = b'\n5,6,"foo\n'
>>> overfull_rows_bytestr = rows_bytestr + extra_line_bytestr
>>> overfull_rows_bytestr # 3 valid rows and an invalid (partial) row
b'1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"\n5,6,"foo\n'
>>> pd.read_csv(io.BytesIO(overfull_rows_bytestr), header=None)
Traceback (most recent call last):
...
  File "pandas/_libs/parsers.pyx", line 1951, in pandas._libs.parsers.raise_parser_error
pandas.errors.ParserError: Error tokenizing data. C error: EOF inside string starting at row 3

As expected, there is a parser error, and this can be overcome by stopping early:

>>> pd.read_csv(io.BytesIO(overfull_rows_bytestr), header=None, nrows=3)

In reverse however, the invalid rows are accepted!

>>> rev_over_rows_b = overfull_rows_bytestr[:-1][::-1] + b"\n"
>>> rev_over_rows_b # an invalid (partial) row and 3 valid rows
b'oof",6,5\n"zab,01,9\nrab,8,7\noof",6,5\ndlrow,4,3\nolleh,2,1\n'
>>> pd.read_csv(io.BytesIO(rev_over_rows_b), header=None).iloc[::-1, ::-1].T.reset_index(drop=True).T.reset_index(drop=True)
   0  1                       2
0  1  2                   olleh
1  3  4                   dlrow
2  5  6  zab,01,9\nrab,8,7\noof
3  5  6                    oof"

• This appears to be a bug in pandas: in reverse, you can parse the invalid row (even though it has a lone, unquoted character)
• This could very trivially be checked for (an unescaped quotechar should not be found in any row)
  • I would rather not do this kind of validation when pandas should do it to be honest

Since neither the start nor end delimiter can be known, I don't know if this can be so neatly resolved (TBC)

This is not something unique to the first row, it can be pushed onto the second row

>>> another_valid_row = b"a,0,0\n"
>>> pd.read_csv(io.BytesIO(another_valid_row + rev_over_rows_b), header=None)
                        0  1  2
0                       a  0  0
1                    oof"  6  5
2  zab,01,9\nrab,8,7\noof  6  5
3                   dlrow  4  3
4                   olleh  2  1

It does seem to be due to being directly before a row with [properly paired] quotechar, which perhaps confuses the parser:

>>> spaced_b = rows_bytestr + b"\n" + another_valid_row[:-1] + extra_line_bytestr
>>> pd.read_csv(io.BytesIO(spaced_b), header=None)
Traceback (most recent call last):
...
  File "pandas/_libs/parsers.pyx", line 1951, in pandas._libs.parsers.raise_parser_error
pandas.errors.ParserError: Error tokenizing data. C error: EOF inside string starting at row 4

When parsed in the forward direction, the partial row is forcibly accepted: this means you do not want to decide if a row is valid or not by putting it at the start of a dataframe

I don't really understand why this is not a parser error:

>>> pd.read_csv(io.BytesIO(b'foo"\n'), header=None)                 
      0
0  foo"

However the padding only seems to come after a valid row

>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\n'), header=None)
       0      1    2
0  hello  world  etc
1   foo"    NaN  NaN
>>> pd.read_csv(io.BytesIO(b'foo"\nhello,world,etc\n'), header=None)
Traceback (most recent call last):
...
  File "pandas/_libs/parsers.pyx", line 1951, in pandas._libs.parsers.raise_parser_error
pandas.errors.ParserError: Error tokenizing data. C error: Expected 1 fields in line 2, saw 3

The fact that they're being NaN-padded means they're interpreted as "blank", rather than invalid:

>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\n'), header=None, na_filter=False)
       0      1    2
0  hello  world  etc
1   foo"            
>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\netc,etc,etc\n'), header=None, na_filter=False)
       0      1    2
0  hello  world  etc
1   foo"            
2    etc    etc  etc
>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\netc,etc,etc\n'), header=None, na_filter=False)[2]
0    etc
1       
2    etc
Name: 2, dtype: object
>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\netc,etc,etc\n'), header=None, na_filter=False)[0][1]
'foo"'
>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\netc,etc,etc\n'), header=None, na_filter=False)[0][0]
'hello'
>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\netc,etc,etc\n'), header=None, na_filter=False)[2][0]
'etc'
>>> pd.read_csv(io.BytesIO(b'hello,world,etc\nfoo"\netc,etc,etc\n'), header=None, na_filter=False)[2][1]
''

• This doesn't seem to be something pandas is capable of

---

There are 2 options then:

1. 'Manually' validate the input (check if the quotechar is in the dataframe values: this is always invalid!)
2. Use the Python csv module (I expect it will be more reliable at validation)

The 2nd option isn’t hard:

>>> import csv
>>> r = csv.reader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n')
>>> for row in r: print(row)
['hello', 'world','etc']
['foo']
['etc', 'etc', 'etc']
>>> r = csv.reader(io.StringIO('foo"\n'))
['foo"']
>>> r = csv.reader(io.StringIO('intA,intB,strC\n1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"'))
>>> for row in r: print(row)
['intA', 'intB', 'strC'],
['1', '2', 'hello'],
['3', '4', 'world']
['5', '6', 'foo\n7,8,bar\n9,10,baz']
>>> r = csv.reader(io.StringIO('1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"\n5,6,"foo\n'))
['1', '2', 'hello'],
['3', '4', 'world']
['5', '6', 'foo\n7,8,bar\n9,10,baz']
['5', '6', 'foo\n']

This is not actually any more valid: note that the quotechar that is "left open" on the last line is stripped, whereas pandas leaves it in (where it can at least be identified)

---

I expect this routine will look similar to how the head is made (in the if sample block at the end of read_bytes, which is read by reader through a BytesIO buffer ⇢ head)

The output from this routine is as outlined for fsspec-reference-maker:

So the only task you need to accomplish, is to find a set of offset/size pairs that can load the data.

...and from this generate a filespec to pass to a to-be-written CSV module in fsspec-reference-maker.

The offsets returned from read_byte are a guide as to the positions after which to seek to delimiters. So in that respect nothing has changed except we are refining this "seek to delimiters" part, making it "seek to lineseparator delimiters which are also row delimiters" (or equivalently "lineseparator delimiters which are not within a row column"), and validating that they are indeed row delimiters, by 10 consecutive rows [positive lookbehind condition].

Note that the routine for sampling the head, per file, works by sampling more and more of the buffer. The value of sample (256,000) is the minimum number of bytes to read until finding the lineseparator, which in read_bytes is referred to as the delimiter, [or if None is given], else it's the chunk size read at each iteration until finding the lineseparator delimiter [or the EOF]:

dask/dask/bytes/core.py

Lines 146 to 162 in e6b9e89

	with OpenFile(fs, paths[0], compression=compression) as f:
	# read block without seek (because we start at zero)
	if delimiter is None:
	sample = f.read(sample)
	else:
	sample_buff = f.read(sample)
	while True:
	new = f.read(sample)
	if not new:
	break
	if delimiter in new:
	sample_buff = (
	sample_buff + new.split(delimiter, 1)[0] + delimiter
	)
	break
	sample_buff = sample_buff + new
	sample = sample_buff

The same routine is needed in reverse, per partition, to identify the first valid [i.e. row-delimiting] lineseparator.

---

Part 2 Overview of CSV partition validation

Click for an overview of the CSV partition validation task

The edge case of a shorter row is handled correctly by the CSV module:

• The quotechar is left in, where it can be detected

>>> r = csv.reader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n'))
>>> for row in r: print(row)
... 
['hello', 'world', 'etc']
['foo"']
['etc', 'etc', 'etc']

One possibility is a DictReader, which would presumably fail if one column (key) is not present:

• This correctly supplies None when it finds missing values
• This incorrectly defaults to reading the first line as a header row (when fieldnames=None)

>>> r = csv.DictReader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n'))
>>> for row in r: print(row)
... 
{'hello': 'foo"', 'world': None, 'etc': None}
{'hello': 'etc', 'world': 'etc', 'etc': 'etc'}

This could be avoided however by mandating the use of the sample_df earlier (the head formed from b_sample in dask)

>>> bytestr = b'intA,intB,strC\n1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"'
>>> sample_df = pd.read_csv(io.BytesIO(bytestr))
>>> r = csv.DictReader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n'), fieldnames=sample_df.columns)
>>> for row in r: print(row)
... 
{'intA': 'hello', 'intB': 'world', 'strC': 'etc'}
{'intA': 'foo"', 'intB': None, 'strC': None}
{'intA': 'etc', 'intB': 'etc', 'strC': 'etc'}

Which is correct!

This can also handle the case when the sample does not have a header:

>>> rows_bytestr = b'1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"'
>>> no_header_sample_df = pd.read_csv(io.BytesIO(rows_bytestr), header=None)
>>> r = csv.DictReader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n'), fieldnames=no_header_sample_df.columns)
>>> for row in r: print(row)
... 
{0: 'hello', 1: 'world', 2: 'etc'}
{0: 'foo"', 1: None, 2: None}
{0: 'etc', 1: 'etc', 2: 'etc'}

The keys of the dict are integers rather than strings in that case.

The values are the fields, which can be checked for the unescaped quotechar

>>> r = csv.DictReader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n'), fieldnames=no_header_sample_df.columns)
>>> for row in r:
...    if any(quotechar in v for v in row.values()):
...        raise ValueError
...    print(row)
... 
{0: 'hello', 1: 'world', 2: 'etc'}
Traceback (most recent call last):
  File "<stdin>", line 3, in <module>
ValueError

Actually it has to be a little more precise than just checking if the value contains the quotechar:

• Not preceded by the escapechar (so should use a regex created by an f-string
• If doublequote is True, then two consecutive [again, unescaped] quotechar should be considered as a single quotechar
  • This could be incorporated into the regex or could use str.replace

If the field has quotes around the entirety of it then these should (will) be stripped, so no need to handle that case.

So I think the regex is rf'[^{escapechar}]{quotechar}\{1,{1 + int(doublequote)}\}'

>>> escapechar = r'\\'
>>> quotechar = '"'
>>> n_matchable = 1 + int(doublequote)
>>> re_patt_str = rf'([^{escapechar}]{quotechar})' + r'{1,' + f'{n_matchable}' + r'}'
>>> re_patt_str
'([^\\\\]"){1,2}'
>>> re_patt = re.compile(re_patt_str)

...but you also want:

• to match it within a str (so within .* either side)
• to identify the quotechar itself

You could therefore use a second match group inside the repetition matched group:

>>> re_patt_str = rf'.*([^{escapechar}]({quotechar}))' + r'{1,' + f'{n_matchable}' + r'}.*'
>>> re_patt = re.compile(re_patt_str)
>>> re_patt.match('hello"foo').groups()
('o"', '"')

Note that the outer [repetition] match group includes whichever character precedes. You could also use a negative lookbehind

>>> rf"(?<!({escapechar}))"
'(?<!(\\\\))'
>>> re_patt_str = rf'.*((?<!({escapechar}))({quotechar}))' + r'{1,' + f'{n_matchable}' + r'}.*'
>>> re_patt = re.compile(re_patt_str)
>>> re_patt
re.compile('.*((?<!(\\\\))(")){1,2}.*')

In full, it works as expected, and does not provide an unwanted non-escapechar in the match group (if a negative lookbehind succeeds it provides None as the matched entity rather than what was there instead):

>>> re_patt_str = rf'.*((?<!({escapechar}))({quotechar}))' + r'{1,' + f'{n_matchable}' + r'}.*'
>>> re_patt = re.compile(re_patt_str)
>>> re_patt_str
'.*((?<!(\\\\))(")){1,2}.*'
>>> re_patt.match(r'hello\"\"world').groups()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'NoneType' object has no attribute 'groups'
>>> re_patt.match(r'hello\""world').groups()
('"', None, '"')

We can now specify this to the csv.DictReader, and since a match is a 3-tuple and a non-match is None, we can simply check against the truthiness of the match call:

>>> r = csv.DictReader(io.StringIO('hello,world,etc\nfoo"\netc,etc,etc\n'), fieldnames=no_header_sample_df.columns)
>>> for row in r:
...    if any(re_patt.match(v) for v in row.values()):
...        raise ValueError
...    print(row)
{0: 'hello', 1: 'world', 2: 'etc'}
Traceback (most recent call last):
  File "<stdin>", line 3, in <module>
ValueError

Fully validating would also mean checking that each row has the correct number of columns [i.e. consistent with the head sample]. In fact we can make it a self-contained demo now:

import csv
import io
import re
import pandas as pd

__all__ = ["reader", "validate_dictreader", "validate_str"]

sample_header_bytestr = b'intA,intB,strC\n'
sample_df = pd.read_csv(io.BytesIO(sample_header_bytestr))

def reader(input_str, df=sample_df):
    """
    Args:
      input_str : The string to load into a new :class:`io.StringIO` buffer to create
                  the DictReader from
      df        : The DataFrame whose columns give the fieldnames fro the DictReader
    """
    buf = io.StringIO(input_str)
    return csv.DictReader(buf, fieldnames=df.columns)

def validate_dictreader(r):
    """
    Args:
      r : (:class:`csv.DictReader`) The DictReader which can be iterated to give the rows to be validated
          in the form of :class:`dict` objects (one per row).
    """
    escapechar = r"\\"
    quotechar = '"'
    doublequote = True
    n_matchable = 1 + int(doublequote)
    re_patt_str = rf'.*((?<!({escapechar}))({quotechar}))' + r'{1,' + f'{n_matchable}' + r'}.*'
    re_patt = re.compile(re_patt_str)
    validated_rows = []
    for row in r:
        if None in row.values():
            raise ValueError(f"Absent field (incomplete row) at {row=}")
        if any(re_patt.match(v) for v in row.values()):
            raise ValueError(f"{quotechar=} found in {row=}")
        validated_rows.append(row)
    for row in validated_rows:
        print(row) # Only print rows if entire DataFrame validated

def validate_str(input_str, sample_df=sample_df):
    """
    Validate the string ``input_str`` whose column names (or index if no names) are
    available through the ``sample_df`` DataFrame (sampled from the file's head).
    """
    r = reader(input_str, df=sample_df)
    validate_dictreader(r)

>>> validate_str(input_str='hello,world,etc\nfoo"\netc,etc,etc\n')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "validate_rows.py", line 49, in validate_str
    validate_dictreader(r)
  File "validate_rows.py", line 36, in validate_dictreader
    raise ValueError(f"Absent field (incomplete row) at {row=}")
ValueError: Absent field (incomplete row) at row={'intA': 'foo"', 'intB': None, 'strC': None}

Note that fields being present but empty is valid.

>>> validate_str(input_str='1,2,etc\n3,,\n5,6,etc\n')
{'intA': '1', 'intB': '2', 'strC': 'etc'}
{'intA': '3', 'intB': '', 'strC': ''}
{'intA': '5', 'intB': '6', 'strC': 'etc'}

The other check I wanted of an implementation was that an 'overfull' byte string would be acceptable if sufficient number of rows were read before the extra/partial line was read in. Instead of 10, I suggested to pretend this sufficient number of rows was 3, and then use the following example:

>>> rows_str = '1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"\n'
>>> extra_line_str = '5,6,"foo\n'
>>> overfull_rows_str = rows_str + extra_line_str
>>> overfull_rows_str # 3 valid rows and an invalid (partial) row
'1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"\n5,6,"foo\n'
>>> validate_str(overfull_rows_str)
{'intA': '1', 'intB': '2', 'strC': 'hello'}
{'intA': '3', 'intB': '4', 'strC': 'world'}
{'intA': '5', 'intB': '6', 'strC': 'foo\n7,8,bar\n9,10,baz'}
{'intA': '5', 'intB': '6', 'strC': 'foo\n'}

Unfortunately this does not validate correctly

---

Part 3: How to validate correctly

Click to show the conclusions on partitioned CSV validation

As shown in the previous section, the csv module's DictReader for some reason cleans up the lone quotechar in the example overfull_rows_str

'1,2,hello\n3,4,world\n5,6,"foo\n7,8,bar\n9,10,baz"\n5,6,"foo\n'

Which should be split as:

1,2,hello
3,4,world
5,6,"foo\n7,8,bar\n9,10,baz"
5,6,"foo\n

I suspect the only way that it can be ensured to do so is to read the rows one at a time, 'greedily', not lazily (e.g. the regex ? quantifier). The error here arises from trying to read lazily, and arriving at an invalid row as a result.

>>> r = csv.reader(io.StringIO(overfull_rows_str))
>>> for row in r: print(row)
...
['1', '2', 'hello']
['3', '4', 'world']
['5', '6', 'foo\n7,8,bar\n9,10,baz']
['5', '6', 'foo\n']

The real take-away here is that you can't trust the last row (it might be unfinished), and for this reason you might want to read in 11 rows and drop the last (to get 10 rows).

However: note that the presence of the open quotechar is implicitly indicated by the presence of the newline within the row value ('foo\n')

• If the row value ends in a newline, it should therefore be considered as (and validated as if) starting with a quotechar

This means we do not need to consider reading in an 11th line, only to check for the newline (and always making sure that the line we read in is lineterminator-terminated, or else we would not get that newline within the row value to check).

If we extend the previous example accordingly:

>>> further_extension_str = 'one,two,three\n'
>>> further_extended_rows_str = overfull_rows_str + further_extension_str
>>> r = csv.reader(io.StringIO(further_extended_rows_str))
>>> for row in r: print(row)
... 
['1', '2', 'hello']
['3', '4', 'world']
['5', '6', 'foo\n7,8,bar\n9,10,baz']
['5', '6', 'foo\none,two,three\n']

Note that even though the lone quotechar on the 3rd row appeared to have been 'missed' (not shown in the row value), it was not missed from the row parsing, and so when further rows are given, the parser is not 'tricked' into believing the values to be on another row, but [correctly] conjoins them with an intra-row newline ('foo\n' ⇒ 'foo\none,two,three\n').

>>> overfull_recovery_str = 'etc"'
>>> recovered_overfull_rows_str = further_extended_rows_str + overfull_extension_str
>>> r = csv.reader(io.StringIO(recovered_overfull_rows_str))
>>> for row in r: print(row)
...
['1', '2', 'hello']
['3', '4', 'world']
['5', '6', 'foo\n7,8,bar\n9,10,baz']
['5', '6', 'foo\none,two,three\none,two,three\netc']

Note that here, the final value has no newline (lineseparator) at the end of it, indicating that the multiline string value is closed.

• Also note that naturally, this will still apply when reading in reverse (since from either direction a single quotechar will be picked up). The same problems would be encountered with multiline field values, just in reverse.

The last issue to address is the possibility of entering within a multiline string, e.g. consider if the parser picked up after the first of these lines (becoming 'tricked' into treating the lines in all-caps as rows when in fact they are all a continuation of the line '1,2,"hello\n'):

1,2,"hello
A,B,C
D,E,F
G,H,I
J,K,L"
3,4,etc

Of course when all of these lines are present we can validate it:

>>> full_str = '1,2,"hello\nA,B,C\nD,E,F\nG,H,I\nJ,K,L"\n3,4,etc\n'
>>> validate_str(full_str)
{'intA': '1', 'intB': '2', 'strC': 'hello\nA,B,C\nD,E,F\nG,H,I\nJ,K,L'}
{'intA': '3', 'intB': '4', 'strC': 'etc'}

but if we omit the first line:

>>> trick_str = full_str[11:]
>>> trick_str
'A,B,C\nD,E,F\nG,H,I\nJ,K,L"\n3,4,etc\n'
>>> validate_str(trick_str)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "validate_rows.py", line 49, in validate_str
    validate_dictreader(r)
  File "validate_rows.py", line 38, in validate_dictreader
    raise ValueError(f"{quotechar=} found in {row=}")
ValueError: quotechar='"' found in row={'intA': 'J', 'intB': 'K', 'strC': 'L"'}

Then we need to read through 4 lines until meeting the lone [closing] quotechar which raises the parser error.
This is as expected, and something to bear in mind when interpreting the meaning of this error. It also gives further reason not to reduce the number of rows to sample from the partition tails, as too few risk not exitting the end of the multiline quoted string the parser started within.

To illustrate, here's the case when only 3 lines are iterated through (not reaching the quotechar on the 4th):

>>> r = reader(trick_str)
>>> escapechar = r"\\"
>>> quotechar = '"'
>>> doublequote = True
>>> n_matchable = 1 + int(doublequote)
>>> re_patt_str = rf'.*((?<!({escapechar}))({quotechar}))' + r'{1,' + f'{n_matchable}' + r'}.*'
>>> re_patt = re.compile(re_patt_str)
>>> validated_rows = []
>>> for row_idx, row in enumerate(r):
>>>     if row_idx == 3: break
>>>     if None in row.values():
>>>         raise ValueError(f"Absent field (incomplete row) at {row=}")
>>>     if any(re_patt.match(v) for v in row.values()):
>>>         raise ValueError(f"{quotechar=} found in {row=}")
>>>     validated_rows.append(row)
>>>
>>> for row in validated_rows:
>>>     print(row) # Only print rows if entire DataFrame validated
{'intA': 'A', 'intB': 'B', 'strC': 'C'}
{'intA': 'D', 'intB': 'E', 'strC': 'F'}
{'intA': 'G', 'intB': 'H', 'strC': 'I'}
{'intA': '3', 'intB': '4', 'strC': 'etc'}

i.e. it validates the lines incorrectly into rows, as it couldn't distinguish. The only way to avoid this is a sufficiently large (and an assumption that this will be a rare occurrence in real world data).

---

The csv module therefore provides solutions to all of the problems relevant to this task:

1. Ability to indicate when more bytes are needed to complete a row (which can be read in from a buffer iteratively)

• Indicated by a newline at the end of the field value [always at the end of a row]

2. Ability to indicate when the start position was within a row

• Indicated by a quotechar within a field value (anywhere)

3. Ability to indicate when the row is not a valid row

• Indicated by any None values in the DictReader field values

The function given earlier (validate_str) is sufficient to do this, except it doesn't check for newlines (as in a real implementation, instead of throwing a ValueError you'd want to read in more bytes, which could be done without discarding the rows you already have(?)

If the row value ends in a newline, it should therefore be considered as (and validated as if) starting with a quotechar

[martindurant]

This is quite a lot of information to take in, so before attempting that, a quick question, which of these does your procedure do:

• given a random offset in a CSV file, find the location of valid newline character that is truly a row terminator
• scan a CSV from the beginning to find the locations of all true row-terminating newlines

[lmmx]

Apologies for any information overload! I just wanted it in one place after having reached a conclusion I was confident in.

I misread your question, it does the former. It does not read an entire file!

It would only end up reading an entire file if:

• the file was a single row extending over multiple lines (if it was a single row on one line or otherwise in one partition then it would be assumed valid and nothing computed: [0, EOF) is assumed a valid partition range).
• the file contained 10 or fewer rows per partition and more than 1 partition

If a sample_partition_rows=10 argument were added, note that reducing the value to 0 would turn it off. My impression is dask is not made for such small file partitions however?

It does not need to proceed from the start (the algorithm is not sequential/recursive). It can take a random offset in a CSV file (the only other required info is the number of columns).

• These offsets would be in range(blocksize, size, blocksize) (i.e. the ends of all the partitions except the last)

  • e.g. size=100, blocksize=50 --> 2 partitions, only check end of the first
    • range(blocksize, size, blocksize) = [50]
  • e.g. size=100, blocksize=20 --> 5 partitions, only check ends of first 4
    • range(blocksize, size, blocksize) = [20, 40, 60, 80]

• Above I assumed the number of columns would require the head sample DataFrame, but in fact we could reduce that requirement to just the number of columns.

In practice however it will run once on an entire file due to the [unlikely] possibility of a partition to extend through one or more entire subsequent partitions (thus 'collapsing'/consuming them). To have confidence in the uniqueness of the row-terminator offsets you should compute them all (not just one)

• Potentially in parallel but I don't know if I can assume multiprocessing is permitted so would just do it sequentially.

I plan to reuse the node visitor/translator callback pattern from the fsspec_reference_maker.hdf module to keep consistency across the library.

(P.S. Forgot to mention: routine described above will be done by the fsspec_reference_maker.csv module so that dask "need never know", as you recommended. But I left the notes here as the purpose of that routine is dask-specific.)

[lmmx]

[Comment removed and moved to the fsspec-reference-maker thread]