airflow-provider-tablefile

English (this file) · Русский

Powered by Claude 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:

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

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

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:

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

{
  "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:

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:

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 EXCLUDEd) — 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 ALLs 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):

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:

# (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:

  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.