Metadata-Version: 2.4
Name: airflow-provider-tablefile
Version: 0.1.0
Summary: Apache Airflow provider that parses tabular files (xlsx/csv/json) with DuckDB SQL and writes partitioned jsonl to S3-compatible storage
Author-email: Michael Kozhin <michael@kozhin.cc>
License-Expression: MIT
Project-URL: Homepage, https://github.com/mkozhin/airflow-provider-tablefile
Project-URL: Documentation, https://github.com/mkozhin/airflow-provider-tablefile#readme
Project-URL: Repository, https://github.com/mkozhin/airflow-provider-tablefile
Project-URL: Changelog, https://github.com/mkozhin/airflow-provider-tablefile/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/mkozhin/airflow-provider-tablefile/issues
Keywords: airflow,duckdb,provider,xlsx,csv,s3,etl
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Apache Airflow
Classifier: Framework :: Apache Airflow :: Provider
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: apache-airflow<3,>=2.9
Requires-Dist: duckdb<2,>=1.5.4
Provides-Extra: s3
Requires-Dist: apache-airflow-providers-amazon==8.20.0; extra == "s3"
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: airflow-provider-gmail>=0.3; extra == "dev"
Provides-Extra: packaging
Requires-Dist: build; extra == "packaging"
Requires-Dist: twine>=6.1; extra == "packaging"
Requires-Dist: packaging>=24.2; extra == "packaging"
Dynamic: license-file

# airflow-provider-tablefile

[English (this file)](README.md) · [Русский](README_RU.md)

*Powered by [Claude Code](https://claude.ai/code)*

An Apache Airflow provider whose `TableFileToS3Operator` reads tabular files
(xlsx / csv / json) from S3 or a local disk, applies a **DuckDB** SQL
transformation, and writes the result as hive-partitioned jsonl objects to an
S3-compatible object store (`s3://<bucket>/<prefix>/<col>=<val>/<filename>`). It
targets **Airflow 2.9.1** (`>=2.9,<3`) and **Python 3.10+**.

Every emailed report used to need its own ad-hoc parsing script. The operator
folds the repeating parts — listing the input files, reading them, path
conventions, idempotent writes, error translation, the XCom contract — into code,
and leaves only what changes from report to report as a short piece of SQL.

## What it is (and is not): the layer-2 parser

The end-to-end task "emailed file → table → warehouse" is split into three layers
by **rate of change**:

| Layer | What it does | Where it lives |
|---|---|---|
| 1 | Gmail → S3 / local disk | [`airflow-provider-gmail`](https://github.com/mkozhin/airflow-provider-gmail) |
| 2 | **Parse file → partitioned jsonl on S3** | **this provider** |
| 3 | jsonl → ClickHouse / PostgreSQL / … | standard providers, a downstream loader |

**This provider is layer 2.** It knows about tabular files (xlsx / csv / json),
DuckDB SQL, and S3 path conventions. It knows nothing about Gmail or the
`_manifest.json` schema — integration with layer 1 lives entirely on the gmail
side (its resolver hands over a flat list of `s3://` URIs, see *Integrating with
the gmail provider*). Downstream, a layer-3 loader reads the operator's XCom (the
list of partitions written) and reloads only those.

The operator is fully generic: its **only** input is `input_paths` (a list of
paths / URIs, an `s3://` prefix, or a local directory). Nothing about the source
of the files leaks into it.

## How it works

```
input_paths ──▶ expand + filename_pattern ──▶ download (ObjectStore)
    ──▶ read → CREATE TEMP TABLE src ──▶ your sql → result
    ──▶ group by canonical partition value ──▶ staging jsonl (one per partition)
    ──▶ build + validate every key ──▶ upload (ObjectStore) ──▶ XCom
```

Internally (ADR-0003 / ADR-0004): all S3 I/O sits behind a small `ObjectStore`
seam (an adapter over `S3Hook`); the parsing itself is an S3-agnostic DuckDB
`pipeline.py`; the operator is a thin orchestrator. DuckDB reads only **local**
files — remote inputs are downloaded first, the result is uploaded through the
same seam. httpfs and DuckDB secrets are never used, so the endpoint / region /
credentials all come from the Airflow Connection, and every S3 path is covered by
an in-memory fake in tests.

## Operator signature

```python
from airflow_provider_tablefile.operators.tablefile import TableFileToS3Operator

TableFileToS3Operator(
    task_id="parse",
    source="acme",                # optional trace label (in error texts / logs)
    aws_conn_id="ya_s3",                # reads inputs AND writes the result
    # the only input — a list of paths/URIs, an s3:// prefix, or a local dir;
    # from gmail this is resolve.output (the resolver's flat list of attachment URIs):
    input_paths=[...] | "s3://bucket/prefix/" | resolve.output,
    filename_pattern=r"^Acme.*\.xlsx$",  # optional regex on basename → files of THIS report
    input_format="xlsx",                # "xlsx" | "csv" | "json" — one format per call
    read_options={"sheet": "Report", "range": "A1:F10000", "stop_at_empty": True},
    sql_file="sql/acme.sql",      # optional; the query in a file (inline: `sql=...`)
    partition_by="dt",                  # result column → <col>=<val>/
    dest_bucket="example-bucket",
    dest_prefix="parsed/acme",
    filename="data.json",               # template; jsonl content
    fail_on_empty=False,
)
```

### Parameters

| Parameter | Type | Default | Meaning |
|---|---|---|---|
| `aws_conn_id` | `str` | *required* | The Amazon/S3 Connection used for **both** listing/downloading inputs and writing the result. Its `endpoint_url` / region / credentials make this work against Yandex Object Storage / MinIO / AWS alike. |
| `input_paths` | `str \| Sequence[str]` | *required* | The **only** input: a non-empty `str` (an `s3://bucket/prefix/` or a local directory to list) **or** a non-empty `Sequence[str]` of concrete paths / URIs (the usual case: a gmail resolver's `resolve.output`). Templated. Type / emptiness is checked on the **rendered** value in `execute()` — `__init__` rejects only a literal `None`. The `s3://` container form may carry **no prefix at all** (`s3://bucket/` or `s3://bucket` lists the whole bucket); an `s3://` element of the *sequence* names one object, so it still requires a key. ⚠️ Because an empty prefix is legal, a **templated** container whose prefix renders to nothing (`"s3://bucket/{{ params.prefix }}"` with `params.prefix == ""`) would otherwise list the WHOLE bucket instead of failing — the rendered value is byte-identical to a deliberate `s3://bucket/`, so **no rule on the rendered value can tell the two apart**. The operator compares against the value the DAG *declared* instead: if `input_paths` was written as an `s3://` URI **with** a key and renders to one **without** a key, that is a `TableFileValidationError` before any download. Two shapes stay outside that rule, deliberately: a **whole-URI** template (`input_paths="{{ var.value.inbox_uri }}"`) declares no key at all — and is also how one deliberately templates the whole-bucket form — and an `input_paths` **reassigned after construction** is judged on what the constructor saw. Both are made **loud** instead: any listing with no prefix logs a `WARNING` naming the value and the number of objects it found, so a whole-bucket ingest is visible in the task log and not only in the data. A **local** element of the sequence must be an existing **file**: a directory (or an empty string, which means the current directory) is rejected — pass it as the single-`str` form to have it listed. |
| `filename_pattern` | `str \| None` | `None` | Optional regex (`re.search` on each basename) selecting the files of *this* report (ADR-0001). Left unset → keep all (a *set* field that renders to `None` is an error — see below); a pattern matching nothing → error; a pattern that is the **empty string** → error too, because `re.compile("")` matches every basename and would silently keep every file (whitespace is *not* stripped first — `" "` is a real pattern). Templated. Keep it simple and **never nest one quantifier inside another** — see *Limitations*. |
| `input_format` | `str` | *required* | `"xlsx"` \| `"csv"` \| `"json"`, chosen once for all files. **Not** templated; validated at DAG parse *and* when the task starts (the field can be assigned after construction). |
| `read_options` | `dict \| None` | `None` | Reader options, validated against a per-format allowlist (see *Read options*). Templated (partially, like a rendered field); a *set* field that renders to `None` is an error — see below. An **empty dict is accepted**: it means what omitting the parameter means (the reader's own defaults, a complete working configuration), so unlike an empty `filename_pattern` there is nothing to silently fall back *from*. |
| `sql` | `str \| None` | `None` | DuckDB SQL over the table `src` (default `SELECT * FROM src`), **always** as an inline query — the value is never read as a path. Templated (a *set* field that renders to `None` is an error — see below). Mutually exclusive with `sql_file`. |
| `sql_file` | `str \| None` | `None` | Path to a file holding that SQL — **always** a path, never a query. Templated (a *set* field that renders to `None` is an error — see below); resolved at run time through Airflow's Jinja loader, and its content is rendered as a template. See *`sql` and `sql_file`* below. |
| `partition_by` | `str` | *required* | Result column whose day-grained value becomes the `<col>=<val>` hive segment. **Not** templated; validated as a bare SQL identifier — and against the reserved names `path` / `rows` — at DAG parse *and* when the task starts (the field can be assigned after construction); matched against the result column **case-sensitively** (see below). |
| `dest_bucket` | `str` | *required* | Target bucket. Templated; validated (non-empty, no control chars, none of `/ ? # % @ : [ ] \`) before the first write. |
| `dest_prefix` | `str` | *required* | Base key prefix, e.g. `parsed/acme`. **One `(dest_bucket, dest_prefix)` = one report = one writer.** Templated; path-safety validated before the first write. |
| `filename` | `str` | *required* | The object name inside each partition; the jsonl content is written under it. Templated; path-safety validated before the first write. |
| `source` | `str \| None` | `None` | Optional free-form trace label carried into error messages and logs. Templated; the one field a rendered `None` is still accepted for (both states mean the same thing). |
| `fail_on_empty` | `bool` | `False` | On a 0-row transform: `False` → green task + WARNING + `{"partitions": [], ...}`; `True` → fail the task. **Not** templated; must be a real `bool`, validated at DAG parse *and* when the task starts (the field can be assigned after construction) — the value is used by truthiness, so a string `"False"` would mean `True`. |

### `None` is how four fields spell *not given*

`sql`, `sql_file`, `read_options` and `filename_pattern` all spell *not given* as
`None`, and passing `None` explicitly means exactly that. `sql=None` is the same
as leaving `sql` out, so the usual authoring shapes just work:

```python
def make_parse_task(sql=None, filename_pattern=None, **kwargs):   # DAG factory
    return TableFileToS3Operator(sql=sql, filename_pattern=filename_pattern, **kwargs)

with DAG(..., default_args={"sql": None}):                        # default_args
    ...
```

What the operator *does* refuse is a field that was given a **real value** and
whose template then evaluated to `None` — see the next section.

⚠️ **Residual, and the reason the two cannot be told apart:** a **dynamically
mapped** `.expand(sql=[q1, None])` is unmapped by calling `__init__` with the
*already-resolved* expansion values, so the second instance arrives as
`sql=None` — identical to a DAG that never mentioned `sql`. That task therefore
falls back to the default (`SELECT * FROM src`, or no pattern) instead of
failing. If you map a field, map a real value for every element (a `dict`
lookup with a default, an `.expand()` over the values that actually exist)
rather than a `None` placeholder.

### A field you set must not render to `None`

The same four fields are templated. Under a DAG with
`render_template_as_native_obj=True` Jinja evaluates the rendered value as a
Python literal, so a nullable expression — `"{{ dag_run.conf.get('sql_path') }}"`,
a Variable or an XCom holding `None` — renders to the **object** `None`, not to
the string `"None"`. That is indistinguishable from "never given", and the run
would otherwise fall back to the default with a **green task**: the query
`SELECT * FROM src` instead of your transform, no reader options at all, every
expanded file instead of only this report's.

So a field that was **set** and rendered to `None` is a `TableFileValidationError`
naming the field and what the fallback would have done, raised before any
download. "Set" means *assigned a real value* — in the constructor **or after
it** (`op.sql = "{{ params.q }}"`, the DAG-factory / post-hoc configuration
idiom), since Airflow writes the rendered value back through the same
assignment. Leaving a field out — or passing `None`, which means the same — is
still perfectly fine: the check rejects a value that *disappeared*, never the
absence itself. `source` is exempt: it is a
free-form trace label, so both states mean the same thing there and no data
changes either way. To make a field genuinely optional per run, give the template
a default (`{{ dag_run.conf.get('pattern', '.*') }}`) rather than letting it yield
`None`.

⚠️ A template that renders to the **empty string** is a different shape, and it
survives Airflow's *default* (non-native) rendering, where every rendered value
is a `str`: a Variable holding `""` renders to `""`, never to `None`. Those are
rejected field by field, where "empty" has a meaning — a blank `sql` / `sql_file`
(*Below*), an empty `dest_prefix` / `filename` / `dest_bucket`, an empty
`input_paths` or a local element of `""`, and an empty `filename_pattern` (which
is not "no pattern" but one matching every basename).

### `sql` and `sql_file`: the query, or the file holding it

- **There is no guessing.** `sql` is **always** an inline query
  (`sql="SELECT ... FROM src"`); `sql_file` is **always** a path to a file holding
  that query (`sql_file="sql/acme.sql"`). Neither value is ever inspected to
  decide which it is, so a query ending in `-- see report.sql` is just a query and
  a path is a path because you said so. Passing **both** is a
  `TableFileValidationError` at DAG parse — and also at run time, which is what
  catches the second one being assigned *after* construction; passing neither
  runs the default `SELECT * FROM src`.
- **`sql_file` takes any path Airflow's Jinja loader resolves** — there is
  deliberately **no** suffix check (the parameter itself is the declaration of
  intent), so `queries/report.txt` and `my report.sql` work exactly like
  `report.sql`. The file is read when the task *runs*.
- **Neither value is trimmed.** A blank (or whitespace-only) `sql` or `sql_file`
  is its own `TableFileValidationError` — that is the usual "the template
  rendered to nothing" case, and it must not reach DuckDB (an empty query) or
  Jinja's loader (`TemplateNotFound: ''`), whose errors explain nothing. But
  surrounding whitespace is **not stripped** from a value that is otherwise
  non-blank: `sql_file="sql/q.sql\n"` fails with `Could not load SQL file
  'sql/q.sql\n'`, not by silently resolving `sql/q.sql`. Trimming would be
  exactly the guessing `sql_file` exists to remove; the `!r` in the message keeps
  the stray newline visible.
- **Prefer a path relative to the DAG folder** — `sql_file="sql/acme.sql"`,
  as in the example DAG. That is the portable form: the DAG folder is always part
  of the loader's search path. A DAG-level `template_searchpath` also works while
  the DAG is parsed from its file, but it is **not** kept in the serialized DAG,
  so do not rely on it where a task executes a DAG straight from the database
  (`airflow tasks run --read-from-db`, `DebugExecutor`).
- **Jinja *inside* the file is rendered** with the usual task context
  (`{{ ds }}` → the logical date). The **path itself** is templated too, so
  `sql_file="{{ params.name }}.sql"` works — note that this lets whoever can set
  that value (a `dag_run.conf`, a Variable) choose *which* file under the loader's
  search path is executed as SQL. Jinja's loader blocks `..` traversal, so the
  blast radius is the search path itself; prefer a fixed path when the template
  value is not fully under your control.
- The **path** of the loaded file goes to the task log at `INFO`, the rendered SQL
  **text** only at `DEBUG` — Jinja can pull sensitive values (Variables,
  Connections) into the query, and task logs have different access rights and
  retention than the UI. That `DEBUG` dump keeps the query readable as SQL, with
  every line **indented** so that no line of it can start at column 0 and imitate
  a task-log record. For the same reason the *Rendered Template* tab shows the
  path, not the SQL; an inline `sql` is of course shown in full. A **failing**
  query honours the same rule as far as it can: DuckDB echoes the offending
  statement in its own error text (`LINE 1: ...`), so the operator strips that
  echo from the error it raises — and from the chained cause — before Airflow logs
  it at `ERROR` (the same stripping applies to read errors, whose echo would
  otherwise repeat the rendered `read_options`). The error still names the reason
  (`Binder Error: Referenced column "x" not found`) and the `DESCRIBE src` schema
  — in full: the echo is recognised only as the **last** thing in the message,
  which is where DuckDB puts it, so a source data row that happens to look like
  one (a quoted CSV cell may contain a newline) does not cut the reason short.
  ⚠️ **This removes the statement and nothing else — it is not redaction.**
  DuckDB's reason text is left intact, and it reproduces a fair amount of the
  input: the value it is complaining about (`Conversion Error: Could not convert
  string 'hunter2' to INT32`), and — for CSV failures — a dump of the **user-set
  reader options** it is *not* complaining about (`date_format = hunter2 (Set By
  User)`, `delimiter`, `quote`, `escape`, `header`, `skip_rows`, `sample_size`,
  …), the `columns` option in full on a sniffing failure, the local staging path
  of the file, and the offending source data row. Redacting the reason would
  leave an error that says nothing useful, so it is kept as is. **The rule is
  therefore: never interpolate a secret into `sql` or `read_options`** — the DEBUG
  gating and the echo stripping reduce the exposure, they do not make an
  interpolated secret safe.
  What *is* handled is the reason's **shape**: a quoted CSV cell may legitimately
  contain a newline, and the reason reproduces such a cell verbatim, so every line
  of the reason after the first is indented — in the raised message and in the
  chained cause alike. Nothing is removed or rewritten; the point is only that no
  line of a report file's own content can start at column 0 and imitate a
  task-log record.
- A missing or unreadable file fails the task with `TableFileValidationError`
  **before** any S3 access — a typo in the path costs nothing. So does a
  `sql_file` that renders to an empty / whitespace-only value (a Variable that is
  unset or blank), which would otherwise reach the loader as a request for the
  search directory itself.
- **A file whose *content* renders blank is an error too** — an empty file, or
  one whose whole body is `{% if … %}`-guarded away. It would otherwise reach
  DuckDB as `CREATE TEMP TABLE result AS ` and come back as `Parser Error: syntax
  error at end of input`, whose statement echo is stripped on purpose, leaving
  nothing that says the query was empty. ⚠️ **Blank only**: a *comment-only* file
  (`-- todo`) is non-blank, so it is executed and fails as DuckDB's own parser
  error. Telling comments from SQL means parsing SQL — the guessing `sql_file`
  exists to remove — and the file name is in the `INFO` line right above the
  failure.
- On a DAG with `render_template_as_native_obj=True`, Jinja evaluates a rendered
  template as a Python literal, so a SQL file whose whole content is one
  (a bare number, or a single `{{ … }}` expression yielding a list / dict) comes
  back as that object rather than a string. That is a
  `TableFileValidationError` naming the file — keep the file's content SQL text.
- **A `.sql` suffix in any *other* parameter is just an ordinary value.** File
  semantics belong to `sql_file` alone, and no container is searched for one:
  `input_paths=["s3://b/report.sql"]`, `input_paths=("sql/q.sql",)` and
  `read_options={"sheet": "q.sql"}` are passed through untouched — nothing is read
  from disk, and no `TemplateNotFound`.

## What goes to XCom

```json
{
  "partitions": [
    {"dt": "2026-07-03", "path": "s3://example-bucket/parsed/acme/dt=2026-07-03/data.json", "rows": 80}
  ],
  "source_files": ["s3://example-bucket/gmail/acme/dt=2026-07-02/18c.../report.xlsx"]
}
```

- `partitions` — one entry per partition written, each with the partition value
  (under the `partition_by` key), the full `s3://<bucket>/<key>` `path`, and the
  `rows` count. On an empty result it is `[]` — **never `None`**, so a downstream
  loader can always iterate it.
- `source_files` — the original input paths (as the user / resolver gave them),
  for tracing back to the source email.
- The `path` is a full URI, but a consumer still needs its own `aws_conn_id`: the
  URI names the object within a store, while the store's `endpoint_url` and
  credentials live in the Connection.

## Partitioning and canonical values (ADR-0002)

The partition is **day-grained** — there is no sub-day partitioning — and the
operator **never converts timezones or mutates data**: the original column value
stays in the jsonl untouched; a separate *canonical* string is derived **only**
for the S3 key and XCom. Rows are grouped by the canonical day
(`CAST(<col> AS DATE)` for temporal columns), so two timestamps on the same day
land in one partition rather than overwriting each other.

Only **two** `partition_by` column types are accepted — a day-grained date and an
integer — and each is canonicalised as follows:

| Result column type | Canonical value |
|---|---|
| `DATE` | the date, usually `YYYY-MM-DD` (see the note on BC dates below) |
| `TIMESTAMP` (naive) | the date part (time dropped — no tz math) |
| all integer types (`TINYINT`…`HUGEINT` + unsigned, incl. `UBIGINT` from JSON inference) | decimal string |
| `VARCHAR`, `BOOLEAN`, `TIMESTAMPTZ`, `FLOAT`, `DOUBLE`, `DECIMAL`, `BLOB`, `INTERVAL`, `TIME`, `UUID`, `TIMESTAMP_S/_MS/_NS`, nested | **`TableFileValidationError`** |

**A coarser partition is still expressible — in SQL.** Narrowing the types does
not mean "a monthly partition is impossible", it means the *column* must already
be a date or an integer. A `VARCHAR` such as `strftime(ts, '%Y-%m')` is rejected,
while both of these work:

```sql
SELECT ..., CAST(date_trunc('month', ts) AS DATE) AS dt FROM src  -- dt=2026-07-01
SELECT ..., CAST(strftime(ts, '%Y%m') AS INTEGER) AS dt FROM src  -- dt=202607
```

Notes:

- **The column name is matched case-sensitively.** DuckDB itself resolves
  identifiers case-insensitively, but `partition_by` is also joined verbatim into
  the object key (`<partition_by>=<value>`) and is the field name of the emitted
  XCom record, so it must be spelled exactly as the result column: with
  `SELECT … AS dt`, `partition_by="DT"` is an error (a loud one — the message
  lists the available columns), not a silent match under a different spelling.
- **`path` and `rows` are reserved column names.** The XCom partition record is
  `{<partition_by>: <canonical value>, "path": <s3:// uri>, "rows": <n>}`, so a
  `partition_by` of either name would simply overwrite the service key and the
  record would lose its partition value. Both are rejected — at DAG parse and
  again when the task starts, since the field can also be assigned on an
  already-built operator.
- **`VARCHAR` and `BOOLEAN` are rejected** — and `VARCHAR` not because a text
  column has no day value, but because of what the value *becomes*. The canonical
  string is joined verbatim into the object key, and a text column holds whatever
  the report holds: `Скидка 50%`, `Brand #1`, `A|B`. Those are ordinary data
  characters but URI delimiters, so such a value used to abort the run at
  key validation — *after* the whole download, the SQL and every staging write —
  fixable only by editing the report SQL. With the types narrowed, a canonical
  value is a date string or a decimal integer, i.e. URL-safe by construction.
  Cast the column in your SQL (`CAST(… AS DATE)`, `CAST(… AS INTEGER)`), or see
  the coarser-partition idiom above. `CAST(<text> AS DATE)` works when the text
  is already ISO (`YYYY-MM-DD`); any other spelling has to be parsed first
  (`CAST(strptime(col, '%d.%m.%Y') AS DATE)`). This is what is supported *right now* —
  nothing forecloses widening it later.
- **`TIMESTAMPTZ` is rejected**: "which day" a tz-aware instant falls on depends
  on a timezone the operator refuses to pick. Cast it to `DATE` in your SQL, where
  the file's timezone is known.
- **Non-finite temporal values are rejected**: a `DATE` / `TIMESTAMP` `'infinity'`
  or `'-infinity'` in `partition_by` raises `TableFileValidationError` (like
  `TIMESTAMPTZ`). DuckDB fetches these back as `date.max` / `date.min` yet they do
  not match their own `CAST(col AS DATE)`, so leaving them in would silently drop
  those rows — cast or filter them out in your SQL.
- **A date value is date-only, but not always `YYYY-MM-DD`.** DuckDB renders a BC
  date — reachable by date arithmetic over bad source data — as `0002-01-01 (BC)`,
  with a space and parentheses, and a year beyond 9999 as `10000-01-01`. Both are
  valid key segments and are written as-is; do not assume a fixed 10-character
  shape when parsing the hive path downstream.
- **`FLOAT` / `DOUBLE` are rejected** because they are the *default* numeric type
  DuckDB infers from an xlsx cell — a partition key must not be a float. Cast the
  column to the integer / date type you actually mean in your SQL (the error
  message hints at this). This is not a restriction on other columns of the
  result — only on `partition_by`.
- **`MAP` / `STRUCT` / `LIST` columns in the result are fine** (they are sortable,
  and the operator's `ORDER BY ALL` handles them) — they simply cannot be the
  `partition_by` column.
- **The type is validated even when the result is empty.** A 0-row result is not
  a special case: with `fail_on_empty=False` the task still goes green and writes
  `{"partitions": []}`, but an unsupported `partition_by` type (`DOUBLE`,
  `TIMESTAMPTZ`, nested, …) raises `TableFileValidationError` all the same. The
  rejection therefore cannot depend on whether the report happened to be empty
  today — a run that is green now would otherwise fail on the first non-empty day.
- The `partition_by` column **stays inside the jsonl** (it is not `EXCLUDE`d) —
  rows are self-contained and a reader never has to parse the hive path.
- More than **1000** distinct partitions in one run logs a WARNING ("probably not
  the column you meant?") but never stops.

## Path safety

Every S3 key is built and validated **before the first upload** (two-phase write
is structural, ADR-0004) — a read / SQL / `COPY` / key-validation failure leaves
**zero** objects on S3. Keys follow the same URL-safe contract as the gmail
provider: the rendered `dest_prefix`, the canonical partition value, and
`filename` may not contain `FORBIDDEN_KEY_CHARS` (`?#%{}^[]<>~|"` and the
backtick), a backslash, any character that breaks a line (every ASCII C0
control, `DEL`, and the Unicode line separators `U+0085` / `U+2028` / `U+2029`),
or a lone surrogate (`U+D800`–`U+DFFF`, the only thing a Python string can hold
that UTF-8 cannot encode — it would otherwise abort the run with a raw
`UnicodeEncodeError` at the byte-length check instead of a named field error);
the partition value and
`filename` additionally may not contain `/`, be empty, or be exactly `.` / `..`
(a substring `..` such as `report..json` is fine). `dest_prefix` may not contain
a `.` / `..` **segment** either (`parsed/../etc` is rejected: a filesystem-backed
S3 gateway would normalise it and write outside the prefix) and may not render
**empty** (that would silently move the whole output to the bucket root). The
whole key is capped at **1024 UTF-8 bytes** (S3 has no per-component limit) — and
that cap is enforced **twice**: on each composed key, and up front on the
*shortest key the configuration could ever produce*
(`<dest_prefix>/<partition_by>=/<filename>` plus one byte for the value), so a
prefix that cannot fit fails on an empty report exactly as it does on a full one
instead of passing silently the day there are no partitions. A violation raises
`TableFileValidationError` before any PUT.

**Input** locations get three narrow checks of their own. A **local** element of an
`input_paths` sequence must be an existing *file*: DuckDB's readers accept a
directory and expand it into every file underneath it, recursively, and
`Path("")` is the current directory — so an empty or directory-valued element
would read the worker's working tree and upload it under this report's key. And
an `s3://` bucket or key
carrying a **line-breaking character** is rejected when `input_paths` is
expanded — both for an element of a list and for every key found under a prefix.
S3 accepts any UTF-8 in a key, and the Amazon provider's `S3Hook` logs the bucket
and key of each download with a plain `%s` into Airflow's line-oriented task log,
so an object *name* chosen by whoever writes into the input prefix could forge
task-log records. The rejected set is exactly what Python's `str.splitlines()`
splits on — every ASCII C0 control and `DEL`, **plus** `U+0085` (NEL), `U+2028`
and `U+2029` — because Airflow's own log reader is built on that method and
re-joins the pieces with `\n`, so those three become real records just as a `\n`
does — **plus** a lone surrogate, for the other half of the same rule: UTF-8
cannot encode one, so it would surface as a raw `UnicodeEncodeError` from inside
`S3Hook` or the basename length check rather than as a domain error. The rest of
the Unicode C1 block is accepted: it cannot break a line and it encodes fine.
Finally, the **basename** of an `s3://` input must be usable as
a local file name — the download preserves it (`preserve_file_name=True`), so a
basename of `..` or one over **255 UTF-8 bytes** (S3 caps only the whole key, at
1024 bytes) would fail deep inside `S3Hook` with a bare `IsADirectoryError` /
"File name too long" half-way through the transfer; it is rejected up front
instead. Nothing else about a foreign key is restricted — `? # % +` and spaces
stay allowed and are addressed back verbatim, and an odd-but-workable basename
(`.`, or a key ending in `/`) is accepted.

Note the contract's precise scope: a space and `+` are *allowed* — both are
legitimate in a `dest_prefix` / `filename` the DAG author chose and S3 accepts
them in a key, so rejecting them would fail an otherwise correct run. (A space
still reaches a *partition* value too, even though `partition_by` is narrowed to
a date or an integer: DuckDB renders a BC date as `0002-01-01 (BC)`.) So the
emitted `s3://` URI still needs percent-encoding before it is used as a URL. What
the check guarantees is that no character can silently *drop or re-target* a key
(control chars, `#`, `?`, `%`).

`dest_bucket` is validated separately and deliberately more narrowly: a
non-empty `str`, no control characters and none of `/ ? # % @ : [ ] \` — the
characters with which a URL parser would read a *different* bucket or key out of
the emitted URI (`s3://a#b/key` parses as bucket `a` with `b/key` as the
fragment). The rest of the key-hostile set (`{ } ^ < > ~ | " \``) is *not*
applied to a bucket: bucket naming is the storage provider's own namespace (AWS,
Yandex Object Storage, MinIO), and this check only promises that the URI cannot
be silently re-targeted.

## filename semantics and vanished partitions

The operator **only ever writes (PUT); it never deletes.** The `filename`
template decides the semantics:

- a **fixed** name (`data.json`) → the partition is **overwritten** on every run
  (`replace=True`), so a re-run/retry is idempotent;
- a **unique** name (`data_{{ ts_nodash }}.json`) → each run **accumulates** a new
  slice, keeping history.

Only partitions present in the *new* result are (re)written. **A partition that
vanishes from fresh data is not removed anywhere** — not on S3 (no delete), and
not in a downstream DB (a layer-3 loader reloads only the partitions in the new
XCom, so rows from that partition's previous load stay in the DB). This is a
documented contract, not a bug — no tombstones or state tracking are added until
deletion is confirmed as a real requirement.

## Limitations and operational contract

- **One instance = one report (ADR-0001).** One operator handles exactly one
  report: one schema, one `sql`, one output namespace. Several different reports
  in one delivery are several operator instances, each selecting its own files
  with `filename_pattern` and writing to its own `dest_prefix`. (Airflow dynamic
  task mapping stays available as a *DAG-level* pattern for partition-disjoint
  files, but is not built into the operator — that would lose cross-file SQL.)
- **One `input_format` per call.** All files of one call share a single reader;
  their schemas must match by position/name/type (a `UNION ALL` mismatch raises
  `TableFileReadError` naming the differing file). Mixed formats → separate
  operators.
- **A glob metacharacter in a *file name* is read literally.** DuckDB's readers
  glob-expand the path they are given (even one bound as a parameter), so a
  report file legitimately named `a?.csv` or `a[1].csv` would otherwise pull in
  its neighbours — or read a different file altogether. Every path handed to the
  reader is glob-escaped first, so exactly the expanded files are read, once
  each. **One shape cannot be escaped and is rejected instead:** a file whose
  path holds *both* a backslash and one of `* ? [` (say `x\?.csv`). DuckDB
  splits a globbed path on `\` as well as on `/` — on Linux too — so such a
  path can only ever address a different file; it fails with
  `TableFileReadError` asking you to rename the file. A backslash on its own is
  fine. Note the flip side: `input_paths` is **not** a glob either — a `*` in it
  is a literal character, never a pattern. Use the prefix / directory form plus
  `filename_pattern` to select files.
- **`max_active_runs=1` is required for an overwriting layer** — but it only
  serialises **one** DAG's runs. It does **not** serialise two different DAGs, nor
  two tasks writing the same `(dest_bucket, dest_prefix, partition, filename)`.
  Idempotent overwrite rests on unique **ownership** of the output namespace: one
  `(dest_bucket, dest_prefix)` has exactly one writer. This is an operational
  requirement, not something code can enforce.
- **The upload series is not one transaction.** A read / SQL / `COPY` /
  key-validation failure yields zero PUTs, but a failure *during* the uploads may
  leave the first objects up. Because the keys are deterministic, a retry / re-run
  reaches the full correct result — it is not "all-or-nothing".
- **`filename_pattern` is a trusted regex, and it is not sandboxed.** It is
  compiled with Python's backtracking `re` and matched against file names this
  provider does **not** control (a mail attachment, a foreign object under an
  input prefix). A pattern with nested quantifiers (`^(a+)+\.xlsx$` and
  friends) takes *exponential* time on a name that nearly matches, so a crafted
  file name would hold the worker slot until the task times out. This is
  accepted, not guarded: the pattern comes from the DAG author — the same trust
  level as `sql`, which runs arbitrary DuckDB — and Python's `re` has no timeout.
  Keep patterns to a literal prefix / suffix plus `.*`, and never nest one
  quantifier inside another.
- **The DAG author is trusted — including the Python *types* they pass.** Drawn
  on the same line as the trusted regex above: values originating in the DAG file
  (parameters, and whatever a template renders to) are checked with `isinstance`,
  not `type(x) is str`, so a deliberately hostile **subclass** of a builtin —
  a `str` whose `__str__` returns something else, an `int` whose `__str__`
  injects SQL into a reader option — can pass a validation and reach a log line
  or the reader call; that is accepted, because writing one takes arbitrary
  Python in the DAG file and that author already has `sql=`, which runs arbitrary
  DuckDB. What the provider does **not** trust — report file contents (a column
  name from a header row, a quoted cell), S3 object names, and DuckDB's own
  message text — is always a plain string, and *is* validated and escaped on
  every path.
- **`try_cast` → NULL is the report's responsibility, not an operator error** — a
  NULL in any column *except* `partition_by` is allowed. A NULL in `partition_by`
  is always a `TableFileValidationError`, whatever produced it.
- **Non-finite floats (`NaN` / `Infinity`) serialize as non-standard JSON — and
  the task WARNs.** The writer never reshapes data columns (ADR-0002): DuckDB's
  `COPY … (FORMAT json)` emits a `NaN` / `Infinity` / `-Infinity` value as a bare
  token, which is **not** valid JSON per RFC 8259 and is rejected by strict
  parsers (ClickHouse, Postgres JSON, `json.loads` with a `parse_constant` guard).
  Since `FLOAT` / `DOUBLE` is xlsx's default numeric type, a result column holding
  a non-finite value produces non-strict jsonl. The task still **succeeds** and
  uploads the object — but before the first write the operator counts non-finite
  values in the `FLOAT` / `DOUBLE` columns of the result and logs a `WARNING`
  naming each offending column and its count, so the problem surfaces in the task
  log instead of only in the downstream loader. Filtering / casting them stays the
  report's SQL responsibility (e.g. `NULLIF`, or a `CASE … WHEN isnan(x) OR
  isinf(x) THEN NULL END`) when a strict-JSON consumer is downstream.
  ⚠️ **The check covers top-level `FLOAT` / `DOUBLE` columns only.** A non-finite
  value nested inside a `LIST` / `STRUCT` / `MAP` is written as the same bare
  token but is **not detected** and produces no `WARNING` — unnesting would mean
  parsing DuckDB's type strings and scanning the data recursively, which ADR-0002
  keeps out. Nested values are the report SQL's responsibility, exactly as
  filtering the top-level ones is. The `WARNING` says so itself.

### Errors

Exactly three exception classes. All three are raised **before the first PUT** —
they come from the expand / download / parse / key-building phases, so a run that
fails with one of them wrote nothing. They are not the only way a task can fail:
an *infrastructure* error from `S3Hook` / `botocore` propagates unwrapped (it is
retryable, not a statement about the data), and one raised inside the upload loop
leaves the partitions already PUT in place — see *Limitations* on the upload
series not being a transaction.

- **`TableFileError`** — the base and "other" pipeline errors, including a user
  SQL error (DuckDB's reason text + context: the files and `DESCRIBE src`). Its
  echo of the failing statement is stripped from the message and the chained
  cause — see *`sql` and `sql_file`* for what that does and does not guarantee.
- **`TableFileReadError`** — a broken/foreign file (`original_path` + `source`), a
  schema mismatch across files of one report, a missing worksheet (the text lists
  the sheets that actually exist), a bad read option, a non-dict `read_options`,
  or a DuckDB `excel` extension that cannot be loaded on an `input_format="xlsx"`
  run (the message names the `install_extension('excel')` fix — see
  *Installation*).
- **`TableFileValidationError`** — a missing/NULL `partition_by`, an unsafe
  partition value / `dest_prefix` / `filename` / `dest_bucket`, an empty/invalid
  rendered `input_paths`, a non-string rendered `sql` / `sql_file`, a local
  `input_paths`
  element that is not an existing file, an input `s3://` bucket
  or key carrying a line-breaking character, an input `s3://` basename that
  cannot be a local file name (`..` or over 255 bytes), an empty listing, a
  `filename_pattern` matching nothing or failing to compile, a `sql_file` that
  cannot be loaded (an empty path, an operator with no DAG to resolve it against,
  or a file that renders to a non-string under `render_template_as_native_obj=True`
  — see *`sql` and `sql_file`*), `sql` and `sql_file` given together, an
  unsupported `partition_by` type, or `fail_on_empty`
  on an empty result.

Infrastructure errors from S3 (listing / download / upload — network, access,
credentials) are **not** wrapped: the `S3Hook`/botocore exceptions propagate as
they are, so Airflow retries see retryable infrastructure directly rather than a
domain error.

## Read options

`read_options` is validated against a per-format allowlist (unknown option or
wrong type → `TableFileReadError` naming the option):

- **xlsx** (`read_xlsx`): `header`, `sheet`, `range`, `stop_at_empty`,
  `all_varchar`, `ignore_errors`, `empty_as_varchar`, `normalize_names`.
- **csv** (`read_csv`): `delim`/`sep`, `header`, `columns`, `types`, `names`,
  `skip`, `quote`, `escape`, `nullstr`, `dateformat`, `timestampformat`,
  `sample_size`, `all_varchar`, `normalize_names`, `ignore_errors`,
  `null_padding`.
- **json** (`read_json`): `format` (`auto`/`array`/`newline_delimited`/
  `unstructured`), `records` (`auto`/`true`/`false`), `columns`,
  `maximum_object_size`, `ignore_errors`, `dateformat`, `timestampformat`,
  `sample_size`.

> **xlsx trap:** `read_xlsx` with a glob silently reads only the first file (and
> does not accept a list), so the operator iterates the files itself and
> `UNION ALL`s them. Do not pass a glob.

## Integrating with the gmail provider

The recommended chain wires the gmail resolver between download and parse
(requires `airflow-provider-gmail>=0.3`):

```python
download = GmailAttachmentsToS3Operator(...)          # → [s3://.../_manifest.json, ...]
resolve = GmailResolveAttachmentsOperator(
    task_id="resolve",
    manifests=download.output,      # the download returns MANIFEST URIs
    pick="latest",                  # "latest" is the mail domain's job, not tablefile's
    aws_conn_id="ya_s3",            # MUST be explicit — see below
)                                                     # → [s3://.../report.xlsx, ...]
parse = TableFileToS3Operator(input_paths=resolve.output, ...)  # never sees a manifest
download >> resolve >> parse
```

Two gotchas that silently break the chain:

- The resolver takes **`manifests=download.output`** — the download's XCom is
  *manifest* URIs, not attachment URIs; the resolver reads each manifest and
  expands it into the flat list of *attachment* URIs that becomes `input_paths`.
- The resolver reads the manifests from S3 **itself**, and its `aws_conn_id`
  defaults to `"aws_default"`. If your S3 Connection is `ya_s3`, the resolver
  **must** be given `aws_conn_id="ya_s3"` explicitly (the same Connection as
  download and parse). With the default it reads through a non-existent
  `aws_default`, `resolve.output` comes back empty, `input_paths` is empty, and
  the emailed file is "not found" — the single most likely failure of the chain.

## Example DAGs

See `example_dags/`:

- **`example_gmail_tablefile_chain.py`** — the full emailed-report pipeline:
  `GmailAttachmentsToS3Operator` → `GmailResolveAttachmentsOperator` →
  `TableFileToS3Operator`, `max_active_runs=1`. Requires `airflow-provider-gmail>=0.3`.
- **`example_tablefile_standalone.py`** — tablefile on its own: `input_paths` is a
  single `s3://bucket/prefix/` string, no gmail dependency.

## Installation

The install is **two-phase**. The Airflow 2.9.1 constraints file pins
`duckdb==0.10.2`, which conflicts with this package's `duckdb>=1.5.4,<2` and makes
a single constrained `pip install` unresolvable. So install Airflow (and the
Amazon provider) **with** constraints, then this package **without** constraints,
repeating the exact `apache-airflow==2.9.1` pin so 2.11 is not pulled:

```bash
# (1) Airflow + Amazon provider — WITH constraints
pip install "apache-airflow==2.9.1" "apache-airflow-providers-amazon==8.20.0" \
  --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.9.1/constraints-3.12.txt

# (2) this package [s3] + DuckDB — WITHOUT constraints, repeating the Airflow pin
pip install "airflow-provider-tablefile[s3]" "apache-airflow==2.9.1"
```

(Match the `-3.12` suffix to the environment's Python: `3.10` / `3.11` as needed.)

- **The `s3` extra is required to *run* the operator.** The package imports
  without it (the `S3Hook` import is lazy), but every run — even one with a local
  input — writes the result through `S3Hook`. Without `[s3]` you get import only,
  not execution. The floor version is **exactly `duckdb>=1.5.4,<2`** (not `>=1.5`).
- **Pre-install the DuckDB `excel` extension** (needed by `read_xlsx`, i.e. only
  for `input_format="xlsx"`) into the image after installing the engine, so runs
  are offline:

  ```bash
  python -c "import duckdb; duckdb.install_extension('excel')"
  ```

  Extensions live in a per-user, per-version directory — install them as the same
  user Airflow runs as (or set a shared `extension_directory`) and verify an
  offline `LOAD excel`.
- **Yandex / S3-compatible endpoints:** point the Amazon Connection's `extra` at a
  custom `endpoint_url` (e.g. `storage.yandexcloud.net`) — DuckDB never touches
  S3, so all endpoint/region/credential handling is the Connection's, and no
  httpfs or DuckDB secret configuration is needed.
