ArrowIncremental calls a callable primary_key with the whole table instead of a single row
- Dominant language
- Python
- Stars
- 5.9k
- Forks
- 600
- Avg merge
- 1d 15h
- Merged PRs (30d)
- 38
Description
### dlt version
`devel` (reproduced against current `dlt/extract/incremental/transform.py`)
### Describe the problem
A **callable** `primary_key` is invoked with the wrong argument on Arrow / pandas / polars resources.
dlt's dynamic-hint convention is that a callable hint receives a **single data item** (a dict-like row) — this is how the JSON path (`JsonIncremental` / `resolve_column_value`) calls it. But `ArrowIncremental.__call__` in `dlt/extract/incremental/transform.py` calls `self._primary_key(tbl)` with the **whole arrow table** instead of a row. Two concrete failures follow:
**(a) A standard-style callable takes the wrong branch / crashes.** A callable written the documented way — a membership check like `"id" if "id" in item else ...` — evaluates `"id" in tbl` against a pyarrow `Table`, which is always `False`. It silently returns the wrong key, and if that key doesn't exist the run fails with `IncrementalPrimaryKeyMissing`. The equivalent JSON/object resource works correctly, so the same hint behaves differently depending on the item format.
**(b) A callable resolving to `()` crashes.** `()` is the documented "disable dedup" sentinel, but on the Arrow path it raises `UnboundLocalError: cannot access local variable 'unique_columns'` because `unique_columns` is only assigned inside `if`/`elif` branches with no `else`.
### Expected behavior
A callable `primary_key` should be called with a single representative row (consistent with the JSON path), and resolving it to `()` should disable dedup rather than crash.
### Steps to reproduce
```python
import duckdb
import dlt
# a callable primary_key written the standard way: a membership check on a single row
def dynamic_pk(item):
return "id" if "id" in item else "column_that_does_not_exist"
@dlt.resource
def some_data(created_at=dlt.sources.incremental("created_at")):
# yield an arrow table (or pandas frame) rather than a list of dicts
import pyarrow as pa
yield pa.Table.from_pylist(
[{"created_at": 1, "id": "a"}, {"created_at": 2, "id": "b"}]
)
r = some_data()
r.incremental.primary_key = dynamic_pk
p = dlt.pipeline(destination=dlt.destinations.duckdb(duckdb.connect(":memory:")))
p.run(r) # `"id" in ` is False -> resolves to the missing column -> IncrementalPrimaryKeyMissing
```
The same resource yielding a plain `list[dict]` loads fine, confirming the Arrow-path-specific behavior. Separately, `r.incremental.primary_key = lambda item: ()` raises `UnboundLocalError` on the Arrow path.
I have a fix ready (call the callable with `tbl.slice(0, 1).to_pylist()[0]`, mirroring the JSON path, plus an `else` branch for `unique_columns`) with regression tests parametrized across object / pandas / arrow-table / arrow-batch formats — happy to open the PR.
Contributor guide
Assessment
This issue has not been assessed yet.