filesystem: read_csv_duckdb silently truncates every file when the glob matches more than files_per_page

[jtcurlin]

dlt version

1.30.0

Describe the problem

read_csv_duckdb reads each file off DuckDB's single global default connection and streams the result across yields. As soon as the extractor has two of these generators alive, they clobber each other's result set and every file but one is silently truncated to its first chunk.

The relation is built with the module-level function (sources/filesystem/readers.py:132 in 1.30.0):

file_data = duckdb.from_csv_auto(f, **duckdb_kwargs)
yield from helper(file_data, chunk_size)

and both helpers stream off that relation between yields (sources/filesystem/helpers.py):

def fetch_json(file_data, chunk_size):
    while True:
        batch = file_data.fetchmany(chunk_size)
        if not batch:
            break
        yield add_columns(file_data.columns, batch)

def fetch_arrow(file_data, chunk_size):
    yield from file_data.fetch_arrow_reader(batch_size=chunk_size)

A DuckDB connection holds one implicit result slot. Creating and fetching a second relation on the same connection invalidates the first relation's pending result, and its next fetchmany returns [].

That happens whenever the glob matches more than files_per_page files (default 100): filesystem yields file items in pages, the transformer gets one _read_csv_duckdb generator per page, and PipeIterator round-robins between them. This is not a threading problem — it reproduces single-threaded, and dlt is single-threaded here.

The failure is entirely silent. fetch_json's loop exits on if not batch: break, and an invalidated result is indistinguishable from end-of-file, so the reader "successfully" emits only the first chunk of each interleaved file. No warning, no exception, load package marked complete, load_info clean. The truncation point lands on a DuckDB vector-size boundary (2048), which is the only visible fingerprint: with chunk_size=10_000 we got exactly 10,240 rows per file.

Both fetch paths are affected (use_pyarrow=False and use_pyarrow=True).

Impact. In our production pipeline a 584-file / 11,166,412-row S3 CSV feed loaded 4,286,655 rows — 62% silently missing — across months of daily runs. We only found it because someone downloaded one CSV by hand and noticed rows absent from the warehouse.

Root cause in isolation (no dlt)

import duckdb, tempfile
from pathlib import Path

p = Path(tempfile.mkdtemp()) / "f.csv"
p.write_text("a\n" + "".join(f"{i}\n" for i in range(10_000)))

con = duckdb.connect()
rel_a = con.from_csv_auto(str(p))
rel_b = con.from_csv_auto(str(p))

def drain(rel):                        # what fetch_json does
    total = 0
    while True:
        batch = rel.fetchmany(2048)
        if not batch:
            break
        total += len(batch)
    return total

rel_a.fetchmany(2048)                  # start streaming A
rel_b.fetchmany(2048)                  # starting B invalidates A's pending result
print("A total:", 2048 + drain(rel_a), " B total:", 2048 + drain(rel_b))

A total: 2048  B total: 10000  (expected 10000 each)

Expected behavior

Every file is read in full, regardless of how many files the glob matches or how the extractor interleaves pages. Row counts must not depend on files_per_page.

Failing that, an invalidated result set should raise rather than read as end-of-file — silently returning [] turns a recoverable error into undetectable data loss.

The fix is to stop sharing one connection across concurrently-open readers: con.cursor() off a single module-scoped connection gives each file its own result slot while keeping the shared buffer pool, catalog and settings (duckdb.connect() per file also works but builds a fresh in-memory database each time). Verified both restore full row counts. I'd like to open the PR for this.

Steps to reproduce

Self-contained, local filesystem, ~30s:

import shutil, tempfile
from pathlib import Path
import dlt
from dlt.sources.filesystem import filesystem, read_csv_duckdb


def run(n_files, n_rows, chunk, files_per_page, use_pyarrow):
    tmp = Path(tempfile.mkdtemp())
    for i in range(n_files):
        rows = "\n".join(f"{i},{j}" for j in range(n_rows))
        (tmp / f"part_{i:04d}.csv").write_text(f"file_id,row_id\n{rows}\n")

    p = dlt.pipeline(pipeline_name="repro", destination="duckdb", dataset_name="r", dev_mode=True)
    p.run(
        (
            filesystem(bucket_url=tmp.as_uri(), file_glob="*.csv", files_per_page=files_per_page)
            | read_csv_duckdb(chunk_size=chunk, use_pyarrow=use_pyarrow)
        ).with_name("data")
    )
    with p.sql_client() as c:
        loaded = c.execute_sql("select count(*) from data")[0][0]

    expected = n_files * n_rows
    print(
        f"files={n_files:<4} files_per_page={files_per_page:<5} use_pyarrow={use_pyarrow!s:<5}"
        f" expected={expected:>9,} loaded={loaded:>9,} lost={100 * (expected - loaded) / expected:5.1f}%"
    )
    shutil.rmtree(p.working_dir, ignore_errors=True)
    shutil.rmtree(tmp, ignore_errors=True)


run(120, 3000, 1000, 1000, False)   # control: single page, no interleaving
run(120, 3000, 1000, 100, False)    # default paging
run(600, 3000, 1000, 100, False)
run(600, 3000, 1000, 100, True)

Output (identical on 1.20.0 and 1.30.0):

files=120  files_per_page=1000  use_pyarrow=False expected=  360,000 loaded=  360,000 lost=  0.0%
files=120  files_per_page=100   use_pyarrow=False expected=  360,000 loaded=  340,960 lost=  5.3%
files=600  files_per_page=100   use_pyarrow=False expected=1,800,000 loaded=1,229,752 lost= 31.7%
files=600  files_per_page=100   use_pyarrow=True  expected=1,800,000 loaded=  602,000 lost= 66.6%

The first row is the control: raise files_per_page above the file count so only one generator is ever alive, and the loss disappears. Loss scales with the number of pages, i.e. with how much the extractor interleaves.

Operating system

macOS

Runtime environment

Local

Python version

3.13

dlt data source

filesystem(bucket_url=..., file_glob="fans/**/*.csv") | read_csv_duckdb(chunk_size=10_000, header=True)

Local files in the reproducer; S3 (s3fs) in production. 584 matching CSVs, ~1.1 GB.

dlt destination

DuckDB

Other deployment details

• duckdb 1.4.3, pyarrow 18.1.0
• Reproducer run in a plain venv via uv; no threading, no parallelize()
• readers.py is the only module-level duckdb.* call site in dlt

Additional information

• Other readers are unaffected. read_csv uses pandas.read_csv per file (its own parser state), and read_jsonl / read_parquet materialize before yielding. Only read_csv_duckdb holds a live cursor across yields.
• Any glob over 100 files is exposed, which makes this most likely to bite exactly the large-backfill case where per-file row counts are least likely to be checked by hand.
• Suggested defensive change alongside the fix: have the CSV readers assert that a file's emitted row count is non-decreasing to completion, or at minimum log at WARNING when a relation returns an empty
  batch on its first fetch. The current code cannot distinguish "file finished" from "someone stole my result set".

[stadlera]

Describe the problem

I encountered the same problem today in a related scenario:
When performing a backfill for multiple resources grouped inside one source, each file seems to yield only one chunk.
The behaviour can be observed irrespective of use_pyarrow.

Steps to reproduce

Similar to what @jtcurlin reported, as a local executable example:

import shutil, tempfile
from pathlib import Path
import dlt
from dlt.sources.filesystem import filesystem, read_csv_duckdb


def run(desc, n_resources, n_rows, chunk, use_pyarrow):
    print(f"--- {desc} ---")
    tmp = Path(tempfile.mkdtemp())
    resources = []
    for i in range(n_resources):
        res_dir = tmp / f"res_{i}"
        res_dir.mkdir()
        rows = "\n".join(f"{i},{j}" for j in range(n_rows))
        (res_dir / "part.csv").write_text(f"file_id,row_id\n{rows}\n")
        reader = filesystem(bucket_url=res_dir.as_uri(), file_glob="*.csv") | read_csv_duckdb(
            chunk_size=chunk, use_pyarrow=use_pyarrow
        )
        resources.append(reader.with_name(f"data_{i}"))

    p = dlt.pipeline(pipeline_name="repro_multi", destination="duckdb", dataset_name="r", dev_mode=True)
    p.run(resources)

    with p.sql_client() as c:
        for i in range(n_resources):
            loaded = c.execute_sql(f"select count(*) from data_{i}")[0][0]
            print(
                f"resources={n_resources:<3} chunk={chunk:<6} use_pyarrow={use_pyarrow!s:<5} "
                f"table=data_{i:<2} expected={n_rows:>9,} loaded={loaded:>9,} "
                f"lost={100 * (n_rows - loaded) / n_rows:5.1f}%"
            )
    shutil.rmtree(p.working_dir, ignore_errors=True)
    shutil.rmtree(tmp, ignore_errors=True)


run("control: single resource, no interleaving", 1, 30_000, 1000, False)
run("two resources loaded together", 2, 30_000, 1000, False)
run("five resources loaded together", 5, 30_000, 1000, False)
run("five resources loaded together, pyarrow", 5, 30_000, 1000, True)

Output:

--- control: single resource, no interleaving ---
resources=1   chunk=1000   use_pyarrow=False table=data_0  expected=   30,000 loaded=   30,000 lost=  0.0%
--- two resources loaded together ---
resources=2   chunk=1000   use_pyarrow=False table=data_0  expected=   30,000 loaded=    2,048 lost= 93.2%
resources=2   chunk=1000   use_pyarrow=False table=data_1  expected=   30,000 loaded=   30,000 lost=  0.0%
--- five resources loaded together ---
resources=5   chunk=1000   use_pyarrow=False table=data_0  expected=   30,000 loaded=    2,048 lost= 93.2%
resources=5   chunk=1000   use_pyarrow=False table=data_1  expected=   30,000 loaded=    2,048 lost= 93.2%
resources=5   chunk=1000   use_pyarrow=False table=data_2  expected=   30,000 loaded=    2,048 lost= 93.2%
resources=5   chunk=1000   use_pyarrow=False table=data_3  expected=   30,000 loaded=    2,048 lost= 93.2%
resources=5   chunk=1000   use_pyarrow=False table=data_4  expected=   30,000 loaded=   30,000 lost=  0.0%
--- five resources loaded together, pyarrow ---
resources=5   chunk=1000   use_pyarrow=True  table=data_0  expected=   30,000 loaded=    1,000 lost= 96.7%
resources=5   chunk=1000   use_pyarrow=True  table=data_1  expected=   30,000 loaded=    1,000 lost= 96.7%
resources=5   chunk=1000   use_pyarrow=True  table=data_2  expected=   30,000 loaded=    1,000 lost= 96.7%
resources=5   chunk=1000   use_pyarrow=True  table=data_3  expected=   30,000 loaded=    1,000 lost= 96.7%
resources=5   chunk=1000   use_pyarrow=True  table=data_4  expected=   30,000 loaded=   30,000 lost=  0.0%

Only the last resource is fully extracted, all others suffer from high loss. In comparison to the loss caused by the file pagination, this seems to consistently only extract the very first chunk for every source expect the last one in the run.

Operating system

Linux & Windows

Runtime environment

Local & Kubernetes

Python version

Reproduced on 3.12 & 3.14

[stadlera]

Expanding slightly on my earlier comment, after I took a look at possible solutions:

@jtcurlin's root-cause analysis holds for the multi-resource case I posted above using the same shared default connection, same invalidation, just triggered by PipeIterator round-robining separate resources instead of file pages.

Some supporting references and one correction to the proposed solution.

Docs and upstream

The default connection being process-wide is explicit in the Python DB API docs: "By default we create an (unnamed) in-memory-database that lives inside the duckdb module. Every method of DuckDBPyConnection is also available on the duckdb module." - so duckdb.from_csv_auto(...) puts every reader in the process on one connection. Same page on why sharing it doesn't give independent readers: "A single connection is thread-safe but is locked for the duration of each query, effectively serializing database access."

Upstream has this exact failure mode: duckdb#13705 "intermediate queries on a connection make the original result fetch zero rows" and duckdb#12713 for the Arrow streaming variant.

Using cursor() won't work here

.cursor() is DuckDB's general recommendation - the multi-threading guide says "Each thread must use the .cursor() method to create a thread-local connection to the same DuckDB file based on the original connection" - but it fails for this reader specifically.

Passing an fsspec file object instead of a path makes DuckDB register an internal filesystem implicitly. The root of it is duckdb/filesystem.py, which defines a ModifiedMemoryFileSystem with a single hardcoded protocol name DUCKDB_INTERNAL_OBJECTSTORE, described in its own docstring as "Not for external use. May change at any moment." Nothing in the code asks for this; it happens as a side effect of handing DuckDB a file-like object.

The name is registered per database instance, which is the important distinction here: .cursor() returns a new handle on the same instance and therefore shares the registry, while each duckdb.connect() creates a separate instance with a registry of its own. So the second reader on a cursor throws:

InvalidInputException: Invalid Input Error: Filesystem with name DUCKDB_INTERNAL_OBJECTSTORE
has already been registered, cannot re-register!

The registration state shows this directly — registering through one connection makes the name visible to a cursor taken from it, but not to a second duckdb.connect():

before:       c1=False  c2=False  cur(of c1)=False
after c1 reg: c1=True   c2=False  cur(of c1)=True

Since the registration is implicit and the name is a fixed singleton, there's no clean way to avoid the collision from the outside - unregistering it first only pulls the handles out from under the other readers. Separate duckdb.connect() instances are the contained option.

Proposed solution: One connection per generator

Per-file connect() works but pays a fresh in-memory instance every file (~7 ms each here, several seconds on the 600-file case). Hoisting it out of the loop is enough: files within one generator are drained strictly sequentially - the inner loop fully consumes file N before creating the relation for file N+1 - so only separate generators interleave, and each now owns its own instance. Since the transformer is invoked once per page, that's one connection per page or per resource rather than per file: 6 instead of 600 in the example above.

    # a connection holds a single pending result, so this generator needs one of its own:
    # resources and file pages are advanced in turn and would invalidate each other's reads
    with duckdb.connect() as conn:
        for item in items:
            with item.open() as f:
                file_data = conn.from_csv_auto(f, **duckdb_kwargs)

                for batch in helper(file_data, chunk_size):
                    if add_filename:
                        for record in batch:
                            record["filename"] = item["file_name"]
                    yield batch

Contained to _read_csv_duckdb, no signature or config change, default connection untouched. The with also closes the connection if a consumer abandons the generator mid-file.

Verification

All four cases from the original report go to zero loss:

files=120  files_per_page=1000  use_pyarrow=False expected=  360,000 loaded=  360,000 lost=0.0%
files=120  files_per_page=100   use_pyarrow=False expected=  360,000 loaded=  360,000 lost=0.0%
files=600  files_per_page=100   use_pyarrow=False expected=1,800,000 loaded=1,800,000 lost=0.0%
files=600  files_per_page=100   use_pyarrow=True  expected=1,800,000 loaded=1,800,000 lost=0.0%

My multi-resource cases are likewise 0.0% across the board. I added a regression test locally that round-robins two _read_csv_duckdb generators the way PipeIterator does, reproducing the truncation without needing a pipeline; existing tests pass unchanged, including test_standard_readers and its exact csv_duckdb_example row count.

Happy to open a PR with this if the approach looks right.

[bunnysayzz]

Root cause: read_csv_duckdb used DuckDB's single global default connection for every file. When multiple generators are alive simultaneously (interleaved by the extractor across pages), starting a new relation on the shared connection silently invalidates all other pending results. The next fetchmany returns [], which is indistinguishable from end-of-file.

Fix: duckdb.connect() per file gives each generator its own isolated result slot.

Regression test added: opens 5 generators simultaneously and round-robins between them, verifying every file is read in full. This test fails on the old code and passes with the fix.

PR: #4421

[jtcurlin]

The default connection being process-wide is explicit in the Python DB API docs: "By default we create an (unnamed) in-memory-database that lives inside the duckdb module. Every method of DuckDBPyConnection is also available on the duckdb module." - so duckdb.from_csv_auto(...) puts every reader in the process on one connection. Same page on why sharing it doesn't give independent readers: "A single connection is thread-safe but is locked for the duration of each query, effectively serializing database access."

Thanks for the additional analysis. My original description of the bug implied that paging was a requirement to trigger it -- this framing was overly constrained, the root cause here is two or more _read_csv_duckdb generators alive at the same time, which happens both under multi-resource extraction as well as when the number of pages exceeds files_per_page.

Root cause: read_csv_duckdb used DuckDB's single global default connection for every file. When multiple generators are alive simultaneously (interleaved by the extractor across pages), starting a new relation on the shared connection silently invalidates all other pending results. The next fetchmany returns [], which is indistinguishable from end-of-file.

Fix: duckdb.connect() per file gives each generator its own isolated result slot.

Regression test added: opens 5 generators simultaneously and round-robins between them, verifying every file is read in full. This test fails on the old code and passes with the fix.

PR: #4421

Thanks for the additional contribution. I'm new here, but the contributors guide seemed to suggest that PR's needed to be preceded by an agreed upon high level implementation plan in this issue thread. I believe that @stadlera and were both awaiting a maintainer.

More importantly, your proposed regression test does not fail on the current 1.30.0 code in my local environment. Your test's comment about "enough rows to span multiple chunks at default chunk_size" conveys a misunderstanding of the bug.

Concretely — running your test body verbatim against the pre-fix reader, parametrized only on n_rows:

test_read_csv_duckdb_multiple_files_no_truncation[200]   PASSED   <- the value in the PR
test_read_csv_duckdb_multiple_files_no_truncation[3000]  FAILED   AssertionError: File 0: expected 3000 rows but got 2048

The threshold is not chunk_size. It's DuckDB's STANDARD_VECTOR_SIZE — 2048 rows. The engine delivers results in DataChunks of up to that many rows, the first fetchmany materialises a whole chunk, and every subsequent read is served out of that already-materialised memory without re-entering the execution pipeline. The invalidated result set only becomes observable when the reader goes back to the engine for the next chunk.