feather.read_table 150x slower when reading columns in newer versions

[asfimport]

Description

Performance when reading columns using feather.read_table on Arrow 7.0.0-9.0.0 is drastically slower than it was in 6.0.0.

Profiling the code below shows that the bottleneck is somewhere in the read_names function of pyarrow._feather.FeatherReader.

Example

Setup code:

import pandas as pd
from pyarrow import feather

rows, cols = (1_000_000, 10)
data = {f'c{c}': range(rows) for c in range(cols)}
df = pd.DataFrame(data=data)

feather.write_feather(df, 'test.feather', compression="uncompressed")

Benchmarks Arrow 9.0.0:

%timeit feather.read_table('test.feather', memory_map=True)
%timeit feather.read_table('test.feather', columns=list(df.columns), memory_map=True)

> 178 µs ± 1.23 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
33.8 ms ± 964 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

Benchmarks Arrow 6.0.0:

%timeit feather.read_table('test.feather', memory_map=True)
%timeit feather.read_table('test.feather', columns=list(df.columns), memory_map=True)

> 173 µs ± 2.12 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
224 µs ± 12.1 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

Environment: python 3.9, ubuntu 20.04
Reporter: Håkon Magne Holmen

Related issues:

• [C++] Implement a read range process without caching (is related to)

_{Note: This issue was originally created as ARROW-17913. Please see the migration documentation for further details.}

[asfimport]

Joris Van den Bossche / @jorisvandenbossche:
I am not directly sure what <=6.0 did differently, but looking at the current implementation this is somewhat expected (it might still be that it can be implemented in a better way, of course): when specifying columns, it will read each column separately from the MemoryMappedFile (instead doing a single ReadAt call), and copying each read chunk in a single output buffer, and thus because of this copy the memory-mapping basically has no effect in this case (

arrow/cpp/src/arrow/io/file.h

Lines 182 to 185 in ec579df

	// Raw copy of the memory at specified position. Thread-safe, but
	// locks out other readers for the duration of memcpy. Prefer the
	// zero copy method
	Result<int64_t> ReadAt(int64_t position, int64_t nbytes, void* out) override;

)

This can also be seen when you compare timings with and without memory mapping (with memory_map=False, there is no difference anymore between manually selecting all columns or not):

In [5]: %timeit feather.read_table('test.feather', columns=list(df.columns), memory_map=True)
29.4 ms ± 541 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

In [6]: %timeit feather.read_table('test.feather', columns=list(df.columns), memory_map=False)
35.3 ms ± 234 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

In [7]: %timeit feather.read_table('test.feather', memory_map=True)
239 µs ± 1.08 µs per loop (mean ± std. dev. of 7 runs, 1,000 loops each)

In [8]: %timeit feather.read_table('test.feather', memory_map=False)
35 ms ± 428 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

Now, I would have assumed that it is not needed that all buffers of all columns live in a single body, so I am not 100% sure why it is needed to copy each field to a single output.

[asfimport]

Joris Van den Bossche / @jorisvandenbossche:
The ReadFieldsSubset that does this copying of each field into a single body, was added in #11486 / ARROW-12683 (which would match the observation that this changed in 6.0 -> 7.0)

I suppose before that change, we always loaded the full record batch in memory even when only reading a subset of the fields. As long as you are memory mapping, that might actually be OK, but if you are accessing data eg from S3, you want to read only those buffers that you need to avoid too much IO.

cc @david Li

[asfimport]

David Li / @lidavidm:
I guess the issue is that we need to know whether the I/O is expected to be high-latency or low-latency. There isn't a single universal implementation we can have.

Or, we need to introduce something like ReadRangeCache (without the caching part) pervasively. And we should probably figure out a variety of scenarios to test. I think this is just one symptom of a wider problem; but maybe the 'smarts' should be moved into a higher layer, and the core reader reverted to the 'obvious' implementation. Also CC @westonpace

[asfimport]

Antoine Pitrou / @pitrou:
At some point I think we may have to bite the bullet and collect IO stats.

[asfimport]

Håkon Magne Holmen:
Maybe use something akin to the old implementation when memory_map=True, since the I/O is expected to be low latency and zero-copy?

[asfimport]

Weston Pace / @westonpace:
I'm not sure ReadRangeCache is the answer here (it is already being used). I don't think the problem is that we are issuing few vs. many Read calls (these are pretty cheap in a memory mapped file). I think the problem is that reading a subset of fields is causing a copy (which is a problem whether you are using memory mapped I/O or not).

The code was roughly (very liberal pseudo-code here)...

Buffer body = read_everything();
std::vector<Array> arrays = allocate_arrays();
for (arr in arrays) {
 for (buffer_index in expected_buffers_for_array(arr)) {
 arr.buffers[buffer_index] = get_range_for_arr_buffer(arr, buffer_index);
 }
}

In order to keep that latter part unchanged I changed the partial-fields case to...

Buffer body = AllocateBigBuffer();
for (auto buffer_range in needed_ranges) {
 Buffer little_body = read_partial(buffer_range);
 body.set(appropriate_offset).to(little_body);
}
std::vector<Array> arrays = allocate_arrays();
for (arr in arrays) {
 for (buffer_index in expected_buffers_for_array(arr)) {
 arr.buffers[buffer_index] = get_range_for_arr_buffer(arr, buffer_index);
 }
}

That triggers a memcpy (I don't recall realizing this at the time). A better algorithm would be to bite the bullet and change the lower half to get...

std::vector<Buffer> buffers;
for (auto buffer_range in needed_ranges) {
 Buffer little_body = read_partial(buffer_range);
 buffers.push_back(little_body);
}
std::vector<Array> arrays = allocate_arrays();
for (arr in arrays) {
 for (buffer_index in expected_buffers_for_array(arr)) {
 arr.buffers[buffer_index] = buffers[buffer_index + some_offset];
 }
}

[asfimport]

Weston Pace / @westonpace:
As a workaround in the meantime there is always:

 tab = feather.read_table('test.feather', memory_map=True)
 return tab.select(list(df.columns))

[asfimport]

Håkon Magne Holmen:
I'll use it for the time being. Thanks for the help

[asfimport]

David Li / @lidavidm:
I agree ReadRangeCache is not relevant here, I was trying to make the general point (here and in ARROW-17917 , ARROW-17961) that it seems the file format readers need a better I/O abstraction, instead of ad-hoc trying to integrate readahead and coalescing into each and every format (but maybe it's too tightly integrated for that to be possible)

[asfimport]

Weston Pace / @westonpace:
I think we could add a ReadRanges method to the filesystem and then move the read range cache into the filesystem. It would probably simplify the readers a little bit. However, the burden is still on the readers to figure out all the ranges they need first and then issue the reads second. This is probably the hardest part of using a read range cache.

Alternatively, we could do something at the filesystem layer like "if a read is in progress" (or enough reads that the filesystem is saturated) then add the request to a queue and coalesce within the queue. This is very similar to Linux's "plugging".

[asfimport]

David Li / @lidavidm:
(The 'caching' and 'coalescing' parts of ReadRangeCache also probably need to be separated in that scheme or we're going to run into memory issues like in other JIRAs. Or really, we only care about 'coalescing' and not 'caching'.)

[asfimport]

Weston Pace / @westonpace:
Yes, I think caching will always need to be a higher level concern (or we could make a caching filesystem if a user really knows they want it).