v0_1_1.md

Design: v0.1.1 hardening

Status: implemented, and it ships in 0.1.1 from PR #117. This document is the contract the hardening work in 0.1.1 is built against. It closes findings 1–20 and fix-list items F1–F4 and F6–F9 from the 2026-09-30 code review. F9 (Scala Spark) is implemented as the experimental engine.spark module; see Non-goals. F5 (issue tracker cleanup) and F10 (PyPI release) need someone to act on GitHub and PyPI. This contract covers them only as local work, and F10 is done when 0.1.1 is published. The v0.1.1 platform contract builds on this one, and both ship in 0.1.1.

Goals

1. One package. The library lives only in src/local_data_platform. The wheel you build is the code you test.
2. Correct writes. Loads are idempotent when you ask them to be (overwrite, upsert), and partitioned when configured.
3. Checked data. Quality checks run before a write and can stop a bad load.
4. Queryable locally. DuckDB runs SQL over Iceberg tables.
5. Safe by default. No secrets in logs, no hidden costs such as queries run twice, and no import-time side effects.
6. Green CI. Lint, tests (Python 3.12 and 3.13, with both the oldest and newest supported pyiceberg), a wheel smoke test and a docs build.

Foundation (already in place)

Module	Contract
local_data_platform/__init__.py	__version__, SupportedFormat, SupportedEngine, Base, Table, Flow, Worker, Credentials, re-exports Config, resolve_path. Importing it has no side effects.
exceptions.py	LDPError base; ConfigError, TableNotFound, PipelineNotFound, EngineNotFound, PlanNotFound, DataQualityError(message, report), GitHubAPIError
logger.py	get_logger(name) gives a namespaced logger. log() is a legacy alias. configure_cli_logging(level) is for the CLI and demo only. Library code never calls basicConfig.
paths.py	resolve_path(path, base_dir=None) -> Path. Absolute paths are kept, relative paths resolve against base_dir or the cwd, and the legacy leading-slash form falls back with a DeprecationWarning.
config.py	Config(identifier, who, what, where, when, how, metadata, base_dir), with from_json(path) (sets base_dir to the file's folder), from_dict(data, base_dir), validate(), .source, .target, .quality (normalized to {"on_failure", "checks"}) and .resolve(path)
Table(name, path=None, format=None, base_dir=None, **options)	self.path is resolved and self.format is always set. get/put raise TableNotFound unless overridden.

Config schema

{
  "identifier": "nyc_taxi",
  "who": "analyst", "what": "rides", "where": "NYC", "when": "daily", "how": "batch",
  "metadata": {
    "source": {"name": "rides", "format": "CSV", "path": "data/rides.csv"},
    "target": {
      "name": "rides",
      "format": "ICEBERG",
      "catalog": {"identifier": "nyc", "warehouse_path": "warehouse"},
      "write_mode": "overwrite",
      "join_cols": ["ride_id"],
      "partition_by": [{"column": "pickup_ts", "transform": "day"}]
    },
    "quality": {
      "on_failure": "fail",
      "checks": [
        {"check": "row_count", "min": 1},
        {"check": "not_null", "columns": ["ride_id", "pickup_ts"]},
        {"check": "unique", "columns": ["ride_id"]},
        {"check": "accepted_values", "column": "city", "values": ["NYC", "BKK"]},
        {"check": "range", "column": "fare", "min": 0},
        {"check": "freshness", "column": "pickup_ts", "max_age_hours": 48},
        {"check": "schema", "columns": {"ride_id": "int64", "fare": "double"}}
      ]
    }
  }
}

All paths are relative to the config file's folder unless absolute. The BigQuery source keeps its pre-0.1.1 shape: format: "JSON", engine: "BIGQUERY", path to a JSON file holding {"query": ...}, and credentials: {"name", "path"}.

A config may also carry the contract of its table in metadata.docs. It holds the layer, a description, an owner, the grain and primary_key of the table, a description of each of its columns, tags and refresh. A run writes it into the Iceberg table, and a gold table has to have all of it. The spec hash leaves docs out. The contract is in the v0.1.4 design.

Windows and schedules (0.1.3)

Two optional metadata keys say when a config runs and how its runs are windowed. A window is the half-open interval [start, end) that one run is responsible for, and it is the logical window of the run's idempotency key.

{
  "identifier": "daily_orders",
  "when": "daily",
  "metadata": {
    "schedule": "5 2 * * *",
    "window": {"size": "day", "column": "ordered_at", "timezone": "Europe/Paris"},
    "source": {"name": "orders", "format": "CSV", "path": "drops/orders_{window_start:%Y-%m-%d}.csv"},
    "target": {"name": "orders", "format": "ICEBERG", "write_mode": "overwrite",
               "catalog": {"identifier": "shop", "warehouse_path": "warehouse"}}
  }
}

Key	What it holds
metadata.schedule	A 5-field cron expression, or @hourly, @daily, @weekly (Sunday) or @monthly. Without it, a when of hourly, daily, weekly or monthly (any case) means the alias, and ldp plan prints the schedule it derived. Any other when text is a label that nothing reads, and ldp plan says so. The spec hash leaves schedule out, so editing it changes nothing about what a run does.
metadata.window	"none", or an object with a size (day, hour or 15m, the length of the windows of a schedule), an optional column (a date or timestamp column that selects the rows of a window) and an optional timezone (an IANA name, UTC by default). The spec hash keeps window, because the column changes which rows a run reads.
metadata.retries	{"max_attempts": 3, "delay_s": 30, "backoff": 2.0}: how a workflow tries the run again after a failure, as for a workflow. A config that runs as a workflow of one task brings it along. The spec hash leaves retries out, because how often a failed run is tried again does not change what it does.
{window_start:FORMAT}, {window_end:FORMAT}	Placeholders in metadata.source.path, where FORMAT is a Python strftime format, for example drops/orders_{window_start:%Y-%m-%d}.csv. They work in no other key.

How a run reads its window, from ldp run CONFIG --window W or run_config(config, window=W):

• The window. W is START/END (ISO 8601), a (start, end) pair, or one of the shortcuts yesterday, last-week (the last full Monday-to-Monday week) and last-month. A shortcut is counted in the timezone of the config, from the local midnights, so a day of a clock change is 23 or 25 hours long. ldp run, ldp plan and the idempotency key all read a shortcut the same way, and ldp run prints the window it stands for.
• The time zone. The placeholders show the bounds in the timezone, and a timestamp column without a time zone is read as local time in it. A timestamp column with a time zone is compared as an instant, and a date column as a date, which needs a window that starts and ends at local midnight.
• The source path. A path with placeholders is filled in for the window, so a daily drop orders_2026-08-02.csv is read by the run of 2 August. Without a window the run fails with a ConfigError that names the placeholder. ldp plan CONFIG --window W prints the path it resolves, and notes a file that is not there yet.
• The rows. With window.column, the run keeps the rows whose value is at or after the start of the window and before its end, right after the extract, so the transforms and the quality checks see only those rows. A row with no value is in no window. run.extracted reports rows_read and rows_kept, PipelineResult.rows_kept has the same count, and RowCount(equals="source") compares with the rows kept.
• Overwrite. With window.column and the write mode overwrite, an Iceberg target replaces only the rows of the window (the overwrite_filter of Iceberg.put), so one changed day can be redone and the other days stay. The first load of a window has nothing to replace. Without window.column, overwrite still replaces the whole table, and ldp plan notes it when a placeholder picks the file.
• Freshness. In a windowed run a freshness check without a now of its own uses the end of the window as now, so a backfill of last month passes the checks that the first run passed.
• Redoing a window. A window and its key are published once, so running it again is skipped. To replace the rows of a day that changed at the source, run it again with a key of your own, for example ldp run CONFIG --window W --idempotency-key orders-2026-08-02-v2.
• A config that selects no rows. A config with a schedule (its own, or one its when means) that appends to an Iceberg table, and names neither a placeholder in the source path nor a window.column, would load the same rows in every window. ldp plan reports that as an error, and ldp run --window logs a warning. A --window alone, on a config with no schedule, stays what it was: the key of a run, which loads the whole source.
• An empty window. A window that keeps no rows ends with the error No data to write, as any empty batch does. A row_count check with a min holds it back as a data gate instead (exit code 3).

Workflow files (0.1.3)

A workflow file says which pipelines and maintenance run, in what order, for each window. It is a spec of the kind Workflow, and ldp workflow plan and ldp workflow run read it.

{
  "apiVersion": "ldp/v1",
  "kind": "Workflow",
  "identifier": "daily_orders",
  "metadata": {"schedule": "5 2 * * *", "window": {"size": "day", "timezone": "Europe/Paris"}},
  "tasks": [
    {"id": "orders", "pipeline": "config/orders.json"},
    {"id": "customers", "pipeline": "config/customers.json"},
    {"id": "export", "pipeline": "config/export.json", "depends_on": ["orders"]},
    {"id": "tidy", "maintain": {"config": "config/orders.json", "expire": 7, "apply": true}, "depends_on": ["export"]}
  ]
}

Key	What it holds
metadata.schedule, metadata.window	As for a dataset config, but window has only size and timezone, because each config names its own window column. The time zone counts a shortcut such as yesterday, and the days of a backfill
metadata.catchup, metadata.start	ldp schedule tick reads them. catchup says whether it runs every window it missed since start, and not only the latest, and true needs a start. start is an ISO-8601 date or time, read in the time zone of the workflow, and no window that starts before it runs
metadata.retries, tasks[].retries	max_attempts (the attempts in all, 1 by default, which is no retry), delay_s (the longest wait before the first retry, 30) and backoff (what it is multiplied by for each retry after that, 2.0). A key of a task overrides the same key of the workflow
metadata.observability	Sinks for the events of the whole workflow run, besides the ledger
tasks[].id, depends_on	A unique id, and the ids of the tasks that must succeed first. A cycle, an unknown id and a duplicate id are errors
tasks[].pipeline	The path of a dataset config, relative to the workflow file
tasks[].maintain	{"config", ...}: the path of a dataset config with an Iceberg table, and the options of ldp maintain as expire, orphans, apply, retain_last, orphan_age_hours and keep_metadata_versions. It is a dry run unless apply is true
tasks[].dbt	{"project", ...}: the path of a dbt project folder with an ldp_dbt.json, relative to the workflow file, and the options command (build, the default, or test), select, exclude, vars, skip_unchanged and dbt_bin. It does what ldp dbt build does, for the window of the run

A task has exactly one kind, the key that holds what it does. The kinds are a registry (orchestration.tasks.register_task_kind), so later milestones add a gold task and a report task without changing the runner. A dataset config with a metadata.schedule, or a when such as daily, is a workflow of one task wherever a workflow file is accepted.

• Plan. ldp workflow plan FILE [--window W] checks the file, and every config it names with spec.plan, and prints the tasks in the order they run. Each pipeline task that publishes to an Iceberg table shows the idempotency key of the window, which is the key ldp plan CONFIG --window W prints. It exits with 1 when anything is invalid, and warns when a task reads a table that another writes and no depends_on connects them.
• Run. ldp workflow run FILE --window W checks the same things first and starts no task when anything is invalid. It runs the tasks one at a time with graphlib.TopologicalSorter, and tasks that nothing orders run in the order of the file. A pipeline task runs with the window of the workflow, so it publishes under that key. A pipeline task whose target is a file runs directly, because a file is replaced whole by every run and cannot take a key.
• Failure. A task that fails, or whose data a quality check blocks, makes the tasks that depend on it upstream_failed, and they do not run. Tasks that do not depend on it still run. The exit code is 1 when a task failed and 3 when the only problem is a data gate.
• Resume. Running the same window again after fixing the cause skips each pipeline task that already published by its key, without reading its source, and runs the tasks that failed or did not run. A maintain task has no key and runs again.
• Retries. A task that fails with an error that another attempt may fix runs again after a wait, until it succeeds or has used its max_attempts. The errors that are retried are CommitConflict with retriable true, pyiceberg's CommitFailedException and CommitStateUnknownException, and any OSError, which covers a dropped connection, a timeout, a file that has not landed yet and the connection errors of requests. A DataQualityError, a ConfigError, a PipelineNotFound, a CommitConflict with retriable false and every other error are not. The wait is a full jitter, a random time from 0 to delay_s * backoff ** (n - 1) seconds after attempt n and never more than a day. A retry is the next attempt of the same run: the task keeps its run id and the attempt counts up, and a pipeline task keeps its idempotency key, so the new attempt fences every older one and a load that landed before the error is found by its key and not written again (Retries and zombies). Each attempt emits its own events with its attempt number, and ldp runs shows the last attempt of each run. run_workflow takes the function that sleeps and the random generator, so a test never waits.
• dbt tasks. A dbt task registers the Iceberg tables of its project as sources, runs dbt build (or dbt test with "command": "test", which publishes nothing) and writes the models marked ldp_publish to Iceberg, as ldp dbt build does. dbt gets the window as the variables window_start and window_end, UTC instants such as 2026-08-01T00:00:00Z, so a model can read only the rows of its window, and vars adds variables of the task's own. dbt's own output goes to standard error, so ldp workflow run --json keeps standard output for its result, and the result lists the tables the task published.
• How a dbt task ends. succeeded, or blocked_quality when only data tests failed or the quality check of a model blocked its write. That is a data gate, so the workflow exits with 3 and a scheduler does not retry. Any other failure ends it failed.
• Plan and order. The plan checks the project, finds dbt, and shows the variables and the fingerprint. A dbt task names the pipelines that load the tables its project reads in its depends_on, and the plan warns when one is missing. A task that reads the tables the dbt task publishes names the dbt task in its own depends_on, because the plan cannot know what a dbt task writes before dbt has run.
• Running a built window again. A dbt task has no idempotency key, so the runner skips it, with skipped_duplicate, when the last build of the task for the same window succeeded, the project has the same fingerprint as then (its files, the command, the selection and the variables), and no task that ran earlier in this run may have written data. skip_unchanged: false always runs it.
• Events. See Workflow runs. The ledger is .ldp/events.jsonl next to the file.
• In Python. local_data_platform.orchestration.run_workflow(path, window=...) returns a WorkflowResult with the status of every task, and plan_workflow(path, window=...) returns the plan as a dict.

Backfills (0.1.3)

ldp backfill FILE --from START [--to END] runs a workflow, or a dataset config with a window, for every window of a range. local_data_platform.orchestration.backfill(path, start, end, ...) does the same from Python and returns a BackfillResult.

• The windows. The windows of one size are laid end to end from START to END. The size is metadata.window.size, else the one the schedule means: @hourly and 0 * * * * are an hour, @daily and 0 0 * * * a day, and a when of daily or hourly means the same. Any other schedule says when a run starts and not how long its window is, so there the size has to be set. A day runs from one local midnight to the next in metadata.window.timezone, so across a clock change it is 23 or 25 hours long. A config with a metadata.window and no schedule is a workflow of one task for this command only.
• The range. START and END are dates or ISO-8601 datetimes. A value with no offset is read in the time zone of the workflow, so 2026-06-01 is the start of that local day, and a value with an offset is taken as written. Each has to start a window, or the command exits with 1 and says where windows of that size start. END is not part of the range, and it is the end of the last complete window when it is not given. A range with no window, and a range with more than --max-windows windows (1000 by default), exit with 1 before anything runs.
• Before anything runs. The workflow and every config it names are checked, the keys that main holds are read with one walk of the history of each table, and the ledger is read for the last run of each window. Nothing is created: a table or a catalog that does not exist yet holds no key.
• The state of a window. published when every pipeline that writes to Iceberg has its key on main and, if the workflow has any other task but a maintain task, the last run of the window in the ledger succeeded for every task of the workflow. Otherwise failed or blocked when that run failed or a data gate held it back, partial when some of the keys are on main, and missing. A run that stopped before every task ran, such as an interrupted one, is none of these, and the window is missing or partial by its keys. --dry-run prints each window with its key, its state and nothing else.
• A run. A window that is published is skipped without being read, planned or written to the ledger. Every other window is run with run_workflow, so a pipeline task publishes under the key of the window, a task with retries is retried, and the events go to the ledger. A window ends published (something was written or built), skipped (every task found its work done), blocked (a data gate held a task back), or failed. A window that fails, or is blocked, does not stop the others, and --stop-on-failure stops at the first window that failed. The summary counts the windows by outcome. The exit status is 0 when no window failed or was blocked, 3 when a data gate is the only problem, and 1 otherwise.
• Resume. Running the same range again does the work that is left, because a done window is skipped, and a window that failed or was blocked runs again with the tasks it finished skipped by their keys.
• Replace. --replace runs every window with replace=True: a pipeline task that writes to Iceberg runs under the replace key of its window, sha256(["ldp/v1", "replace", key, spec_hash]), in the write mode overwrite, which replaces only the rows of the window, and a dbt task builds again whatever it built. A window that was replaced under the same spec is skipped, and one that was replaced under an older spec is replaced again. Every pipeline that writes to Iceberg needs a metadata.window.column, or the command exits with 1 before it runs anything. A workflow that has no such pipeline has no key to show that a replace was done, so --replace runs every window of it each time.
• Where the search is bounded. The key search of the publish loop is sized for the batch with etl.batch_search, for the windows that are going to run, so the 501st window of a backfill publishes as the first did.
• In Python. BackfillResult.windows has a BackfillWindow for each window, with its state, its keys, its outcome and the WorkflowResult of its run. counts() gives the outcomes, states() the states before the run, status and exit_code the verdict, and to_dict() the JSON that --json prints. on_window(window, number, total) is called when a window is done, for progress.

Schedules (0.1.3)

ldp schedule tick DIR is a scheduler with no services. cron or launchd calls it every 5 minutes, it runs every window that is due and not done for every scheduled workflow and config under DIR, and it exits. ldp schedule list DIR shows what is scheduled and what was missed, and ldp schedule install --print cron|launchd DIR prints the recipe that calls the tick. local_data_platform.orchestration.schedule has tick, list_scheduled and discover for Python.

• What is scheduled. Discovery reads every .json file under DIR, but not in a folder whose name starts with a dot or in node_modules, venv, env, site-packages or __pycache__, and not a file that ends in .metadata.json or is larger than 1 MB. A file is scheduled when it is a workflow file (kind Workflow) or a dataset config (kind Pipeline, which a file with no kind and with metadata.source and metadata.target is) and has a schedule, in metadata.schedule or in a when such as daily. Any other file, and a workflow or config with no schedule, is left out without a word. A scheduled file that is not valid is not left out: list prints it with its problems, tick prints them to stderr, both exit with 1, and the other files still run. A dataset config is a workflow of one task, so its events go to .ldp/events.jsonl next to the config, and those of a workflow file go next to the file. A dataset config has no catchup or start, so it runs the latest window only. Wrap it in a workflow file to catch it up.
• Fires. The schedule is a cron expression read in metadata.window.timezone, UTC without one. A fixed time that a clock change skips fires right after the gap, so 30 2 * * * fires at 03:00 on the spring-forward day. A fixed time that happens twice fires once, at its first time, and a time that repeats, such as 0 * * * *, fires in both passes of the hour. The cron library, cronsim 2.6 or later, finds the fires, and every version agrees on this.
• The window of a fire. The window a fire owns is the interval between the previous fire and this fire. When the workflow has a window size, in metadata.window.size or implied by @hourly and @daily, it is instead the last complete window of that size that ends at or before the fire, counted in the time zone of the workflow. The fire of 5 2 * * * on 5 August owns 4 August, and across a clock change a day is 23 or 25 hours long. Several fires can own one window, as 0 */6 * * * does with day windows, and the window runs once. A window of the interval kind has no size, so ldp backfill cannot take it, and it takes no window.column, because a window object needs a size. Its source path names its file with {window_start} and {window_end}.
• Due windows. A tick at time T looks at the latest fire at or before T. Without catchup, which is the default, the window of that fire is the only one it looks at, and only if it does not start before metadata.start. With metadata.catchup true it looks at the window of every fire after start, oldest first. A tick runs the windows it looks at that are not done, at most --max-windows of them for each workflow (1000 by default), and the windows after those wait for the next tick, which starts with the first window that is not done. A catch-up over more than 100000 windows is refused with a message that says to set a later start, because a start long ago is most likely a typo.
• Done. A window is done by the rules of a backfill. Every pipeline that writes to Iceberg has its key on main, and when the workflow has a task that a key cannot vouch for, such as a dbt task or a pipeline that writes a file, the last run of the window in the ledger succeeded for every task. A done window is skipped without being read, planned or written to the ledger.
• Retries across ticks. A window that is not done, and whose last run in the ledger failed, was blocked by a data gate or never finished, is run as the next attempt of that run. The workflow keeps its run id and each task keeps its run id, and the attempt of the workflow and of each task is one more than the highest in the ledger, so the first commit of the new attempt fences every older attempt of the run, also one that a crashed process left behind, and the window is written once however many ticks it takes. A task that the earlier run did not have starts a run of its own. The events carry the attempt numbers, ldp runs shows the last attempt of each run, and --attempts shows every one. A window whose last run succeeded, and whose keys are gone, starts a run of its own. A tick runs a failed window every time it runs, which is every 5 minutes under the recipes, so a failure that does not pass is tried again and again. metadata.retries are the retries within one tick, counted from the attempt the run starts with. A window that a data gate blocked is run again by each tick too, which is how it publishes once the data is fixed, and the tick exits with 3 until then. ldp backfill does not carry a run on, and a backfill of a failed window starts a run of its own.
• One tick at a time. A tick takes an exclusive lock on DIR/.ldp/locks/schedule.lock that it does not wait for, with local_data_platform.filelock, which the lock of an Iceberg table uses too. A tick that starts while another holds the lock does nothing, prints which lock it found, and exits with 0. The operating system releases the lock when the process ends, so a crash or a kill leaves none behind. list takes no lock.
• Exit status. A tick exits with 0 when no window failed or was blocked and no file is invalid, or when it found another tick running. It exits with 3 when a data gate is the only problem, and with 1 when a window failed, a workflow could not be ticked, or a file with a schedule is not valid. An unexpected error in one workflow is recorded for that workflow and does not stop the others.
• Listing. list prints, for each scheduled file, its kind, schedule, time zone, window size, catch-up setting, next fire in UTC and in the time zone of the machine, the last window the ledger holds and how its last run ended (succeeded, failed, blocked, incomplete when it stopped before every task ran, unfinished when it never finished), the windows that are due, and the windows that were missed. A window is missed when a fire since start, or since the first window the ledger holds, owns it, it is not done, and the next tick does not run it, which means that it is before the latest window and catchup is off. For missed windows it prints the command that fills them, one ldp backfill FILE --from FIRST --to LAST from the first missed window to the last, which skips the windows between them that are done, or an ldp workflow run FILE --window START/END for each when the windows have no size. It names at most --max-windows missed windows, the oldest first, and writes nothing.
• Recipes. install --print cron DIR prints a comment and a crontab line, */5 * * * *, that makes DIR/.ldp, calls ldp schedule tick DIR and appends the output to DIR/.ldp/schedule.log. --print launchd DIR prints a property list that runs the same command every 300 seconds and when it loads, logs to ~/Library/Logs, and has a comment on how to load and unload it. Both hold the absolute path of the ldp executable, which is --ldp PATH, else the one that is running, else the first on the PATH, else the one beside the Python, and the absolute path of DIR. Nothing is written to a system location. cron does not run while a Mac sleeps and Apple has deprecated it, and launchd runs an interval that went by during sleep once when the Mac wakes, so launchd is the recipe for macOS and cron the recipe for Linux.
• Not in 0.1.3. A foreground ldp schedule run, and the systemd and Windows Task Scheduler recipes, are stretch items that the roadmap leaves for later. ldp schedule tick with cron or launchd covers a laptop.

Airflow (0.1.3)

local_data_platform.integrations.airflow runs a dataset config as an Apache Airflow 3 task. The mapping from a task instance to a run is plain Python, and Airflow is imported when LdpRunOperator is first used, so the module imports and its functions run without Airflow installed. The guide shows how to install and run it.

• The window. window_of(context) is (data_interval_start, data_interval_end) of the DAG run, the aware datetimes that Airflow gives, which run_config reads in UTC. A DAG run with no data interval raises NoDataInterval, and so does one whose interval is empty, which has the same start and end. Airflow 3 makes an empty interval for the runs of a cron string as schedule, because it makes a trigger timetable for it unless [scheduler] create_cron_data_intervals is True, and its default is False. The message names CronDataIntervalTimetable("@daily", timezone="UTC") of airflow.timetables.interval, which gives each run the day it covers. NoDataInterval is a ConfigError.
• The window is not checked. The data interval is the window as it is, and run_config does not compare it with metadata.window of the config, as it does not for a window given to ldp run. A DAG whose intervals are not the windows that its config loads, or that overlap, loads some rows under two keys. Give the timetable and the config the same size and the same time zone.
• The key. The idempotency key is spec.idempotency_key(config, window). So the key of a data interval is the one ldp plan CONFIG --window START/END prints for it, whatever the DAG, the DAG run, the task or the try, and the same two instants shown in another time zone give the same key. A second DAG run over the interval, such as a backfill that Airflow starts, finds the key on main and writes nothing. key_for(config, context) returns the key without running.
• The run id. run_id_of(context) is airflow:DAG_ID:RUN_ID:TASK_ID, with :MAP_INDEX after the task id for a mapped task instance. It is the same string for every try of a task instance, and no two task instances share one, because the run id names the staging branch of a run. CommitContext accepts any string as a run id and hashes it into the branch name, and the OpenLineage sink hashes a run id that is not a UUID.
• The attempt. attempt_of(context) is ti.try_number, which has to be a whole number from 1, and anything else is a ConfigError, because a try with no attempt would not fence an older one. The first commit of try N fences tries 1 to N - 1 of the task instance, as docs/exactly_once.md describes under "Retries and zombies", so a try that Airflow gave up on and that is still running cannot publish after a later try has, and a try that follows one that was lost after its commit finds the key and writes nothing.
• The run. run(config, context, mode=None) is etl.run_config(config, mode=, window=, run_id=, attempt=) with those values, so it is a staged run, and run_options(context, mode) gives the four arguments. It raises what run_config raises, DataQualityError for a data gate included.
• The operator. LdpRunOperator(config=..., mode=None, **kwargs) subclasses airflow.sdk.BaseOperator, and the class is built the first time the name is used, with from airflow.sdk import BaseOperator, so it is one class for each Airflow that is imported. An import with no Airflow 3 raises an ImportError that says so. config and mode are template fields, config is kept as a string and should be an absolute path that the worker can read, and kwargs go to BaseOperator.
• The result. execute returns PipelineResult.to_dict(), so the XCom return_value of the task is the result, for published and for skipped_duplicate, and it logs the window, the run id, the attempt, the status, the key and the snapshot.
• Errors. A DataQualityError and a ConfigError, which includes NoDataInterval and a config file that is not found, fail the task with AirflowFailException and the original error as its cause, so Airflow does not retry them, as orchestration.retries.is_retriable does not retry them. Any other error is raised as it is, so Airflow retries it by the retries of the task, and the next try is the next attempt. AirflowFailException is imported from airflow.sdk.exceptions, where it is from Airflow 3.2, and from airflow.exceptions when that has none.
• Packaging. There is no airflow extra, because Airflow has to be installed with its own constraints file, and the module needs nothing that LDP does not already depend on. The constraints files of the Airflow 3 releases before 3.1.7 pin pyarrow, pyiceberg or SQLAlchemy to versions that LDP does not accept, so an Airflow before 3.1.7 runs LDP from the command line instead, with ldp run CONFIG --window START/END --run-id ID --attempt N in a BashOperator, which publishes under the same key and the same run id as the operator.
• Not in 0.1.3. DAGs that are built from workflow files, an LdpTaskOperator, the opt-in Airflow CI workflow and recipes for managed Airflow services are in the 0.2.0 issue "Airflow DAGs from workflow files on managed Airflow".

Module contracts

A. Formats and catalog: format/, catalog/, store/

• Format(Table) is the base for file formats.
• CSV(name, path, base_dir=None) and Parquet(name, path, base_dir=None):
  • get() -> pyarrow.Table raises FileNotFoundError naming the resolved path.
  • put(df, allow_empty=False) -> int returns rows written. It raises ValueError if df is None, or empty when allow_empty is false. It creates parent folders and writes atomically (a temp file, then os.replace).
• Json(name, path, base_dir=None).get() returns the parsed JSON.
• LocalIcebergCatalog(name, path, **properties) subclasses pyiceberg SqlCatalog on sqlite:///<abs path>/<name>_catalog.db with the warehouse at file://<abs path>.
  • It creates the folder and keeps get_dbs() and get_tables(namespace).
  • Errors propagate with their original type.
• Iceberg(name, config=None, *, catalog=None, partition_by=None, write_mode="append", join_cols=None, base_dir=None, schema_evolution=True). Here config is the catalog dict {"identifier", "warehouse_path"}; catalog= is a legacy alias for it.

  • It uses the public create_namespace_if_not_exists, never private pyiceberg APIs.

  • identifier = f"{namespace}.{name}".

  • put(df, mode=None) -> WriteResult supports three modes:

    • append
    • overwrite: replaces the table's current rows
    • upsert: needs join_cols, otherwise it raises ConfigError, and behaves as append when the table is new or empty

    Before writing, put casts nanosecond timestamps to microseconds explicitly. It sets no environment variables and stores no fake table properties. It creates the table with the configured partition spec, and when schema_evolution is on it unions in new columns by name. It raises ValueError on None or empty input.

  • get(snapshot_id=None, row_filter=None, selected_fields=None, limit=None) -> pyarrow.Table supports time travel through snapshot_id.

  • snapshots() -> list[dict] returns snapshot_id, parent_id, timestamp_ms, operation and summary for each snapshot.

  • exists() -> bool, table() returns the pyiceberg table, and row_count() -> int reads it from snapshot metadata where possible.

  • partition_by items are {"column", "transform", "name"?}. The transform is one of identity, year, month, day, hour, bucket[N] or truncate[W]. An unknown transform or column raises ConfigError.

• WriteResult is a dataclass with table_identifier, mode, rows_written, rows_before, rows_after and snapshot_id.
• For backward compatibility, store.target.iceberg.Iceberg and store.source.parquet.Parquet re-export the format classes. This removes the duplicate code flagged in issue #51.

B. Quality: quality/

• CheckResult(name, passed, details, failing_rows=0, metrics={}).

• QualityReport(results) provides:

  • .passed and .failures
  • .raise_for_failures(), which raises DataQualityError(msg, report)
  • .summary() -> str and .to_dict()

• The checks all expose run(df, context=None) -> CheckResult:

  Check	Constructor
  Row count	RowCount(min=None, max=None, equals=None), where equals="source" compares against context["source_rows"]
  Not null	NotNull(columns)
  Unique	Unique(columns) (composite)
  Accepted values	AcceptedValues(column, values)
  Range	Range(column, min=None, max=None)
  Freshness	Freshness(column, max_age_hours, now=None)
  Schema	SchemaHas(columns), which takes a list of names or a dict of name to pyarrow type string

  A missing column is a failed result, never an exception. Checks use pyarrow.compute.

• checks_from_config(items) -> list[Check] maps the config check names. An unknown name raises ConfigError.

• run_checks(df, checks, context=None) -> QualityReport.

C. SQL engine: engine/duckdb/

• DuckDBEngine(connection=None) imports duckdb lazily. If it's missing, it raises EngineNotFound with the hint pip install "local-data-platform[duckdb]".
• register_iceberg(table, alias, snapshot_id=None, row_filter=None) accepts an Iceberg format object or a pyiceberg table.
• It also provides register_arrow(df, alias), query(sql) -> pyarrow.Table and close(), and works as a context manager.

D. External sources: store/source/gcp/, github/, issue/

• GCPCredentials(path, kwargs=None) reads project_id from the key file. It never logs or reprs the key contents.
• BigQuery(name, credentials, path=None, client=None) imports the Google libraries lazily, with an EngineNotFound install hint. get(query) runs the query exactly once and returns job.result().to_arrow(). put raises NotImplementedError. client= exists for tests.
• github.get_items(owner, repo, state) stays. It gains get_item(owner, repo, number) -> Item and logs instead of printing.
• Issue(number, repo_name, owner) uses the GitHub REST API through github.get_item, not HTML scraping. get() returns (title, body), and errors raise GitHubAPIError.

E. Pipelines, CLI and demo: pipeline/, etl.py, cli.py, demo.py

• Pipeline(config=None, *, source=None, target=None, transforms=None, checks=None, on_failure=None, name=None) is built by composition: a pipeline has a source and a target.

  • extract(), transform(df) (applies the callables in order), validate(df, context), run(mode=None) -> PipelineResult, and load(mode=None) as an alias of run.
  • Checks run after the transform and before the write. With on_failure="fail" it raises DataQualityError and writes nothing; with "warn" it logs and continues.

• PipelineResult(name, rows_read, rows_written, write_result, quality, duration_s).

• The registry (pipeline/registry.py) provides:

  • register_pipeline(source_format, target_format, engine=None), a decorator
  • get_pipeline_class(...) and registered_pipelines()
  • create_pipeline(config) -> Pipeline
  • An unknown pair raises PipelineNotFound and lists the pairs that are registered.

• Built-in pipelines:

  Pipeline	Route
  CSVToIceberg	CSV to ICEBERG
  ParquetToIceberg	PARQUET to ICEBERG, now an ingestion pipeline
  IcebergToCSV	ICEBERG to CSV
  IcebergToParquet	ICEBERG to PARQUET
  BigQueryToCSV	JSON to CSV with the BIGQUERY engine

  Two legacy names remain as aliases: pipeline.egression.csv_to_iceberg.CSVToIceberg (deprecated) and pipeline.ingestion.pyarrow.PyArrowLoader, which delegates to create_pipeline.

• etl.run_config(path_or_config, mode=None) -> PipelineResult.

• The ldp CLI:

  Command	What it does
  ldp --version	Print the version
  ldp run CONFIG [--mode]	Run the config's pipeline
  ldp pipelines	List the registered pipelines
  ldp snapshots CONFIG	List the target table's snapshots
  ldp query CONFIG SQL	Run SQL over the target table, registered under its target name
  ldp demo [--workdir DIR]	Run the demo

  It exits with 0 on success and 1 on error, and prints no traceback unless -v is set.

• demo.run_demo(workdir, rows=1000, seed=42) -> dict generates deterministic synthetic taxi rides and walks through the following, using only public APIs:

  1. Config from JSON
  2. Append twice to show the duplicates, then overwrite twice to show the load is idempotent
  3. Upsert
  4. A day-partitioned table
  5. Passing quality checks, then a bad batch blocked before the write
  6. DuckDB SQL
  7. Time travel
  8. Export back to CSV

F. Packaging, CI, docs and examples

• pyproject.toml uses PEP 621 metadata with the poetry-core build backend. Its extras are duckdb, bigquery, dev and docs.

• Makefile targets:

  Target	What it does
  install	Create .venv and install -e .[dev,docs]
  lint	Run flake8
  test	Run the test suite
  demo	Run the demo
  build	Build the wheel
  smoke	Install the built wheel in a fresh venv, then run ldp --version and ldp demo
  docs	Build the docs
  clean	Remove build artefacts

• CI has four jobs:

  • lint
  • a test matrix over Python 3.12 and 3.13, with pyiceberg 0.11.0 and the newest supported release
  • a wheel build plus smoke test
  • mkdocs build --strict

  The publish workflow builds from the repo root.

• README.md, CHANGELOG.md and docs/quickstart.md describe only what exists.

Non-goals of the hardening contract

Some of these arrive later in the same 0.1.1 release, through the platform contract.

• Spark or Scala as a supported engine. F9 is implemented, as the experimental engine.spark module: SparkEngine (PySpark, the spark extra plus a JDK 17+), ScalaSparkJob, which runs spark/IcebergJob.scala with scala-cli, and Spark. Both open the same SQLite catalog as pyiceberg, and pyiceberg reads back what Spark writes. No pipeline uses Spark, and its integration tests are opt-in (LDP_RUN_SPARK=1, or make spark-test), so the four CI jobs don't run them. The opt-in jvm.yml workflow does. The platform contract adds spark_catalog_conf_for for the local, sql and rest catalog types (v0.1.1 platform, C5).
• Remote catalogs (REST, Glue, Hive). The catalog is behind one class, so adding one is a new class, not a rewrite. The platform contract adds sql, rest and glue through catalog/provider.py (v0.1.1 platform, C1).
• Streaming or incremental reads.