One file-to-database pipeline with a single conversion policy

.gitignore

@@ -212,4 +212,6 @@ OMOP_CDM*.csv
 .vscode/
 .DS_Store
 _temp/
-notebooks/
+notebooks/
+# Local benchmarks
+benchmarks/

README.md

@@ -40,25 +40,26 @@ class MyTable(ORMTableBase, Base):
 ```
 You can inherit from it directly or pick it up through one of the higher-level mixins.
 
-2. CSV-based ingestion mixins
+2. File ingestion mixins
 
-`CSVLoadableTableInterface` adds staged file loading to ORM tables. It can use pandas or PyArrow loaders, and on PostgreSQL it can use a fast `COPY` path when the input is clean enough.
+`FileLoadableTableInterface` adds staged file loading to ORM tables. Files are read with PyArrow on every backend, interpreted under a `ConversionPolicy`, and written to staging through each backend's bulk path.
 
 Features include:
 * staging table creation and cleanup
-* chunked loading for large files
-* optional casting and deduplication before insert
+* files streamed in bounded batches, so memory does not grow with the row count
+* optional conversion to column types and deduplication of the staging table
 * backend-specific merge behaviour
-* PostgreSQL fast-path loading with ORM fallback
+* Arrow-based loading with a configurable conversion policy on PostgreSQL, DuckDB and SQLite
+* delimited text (tab or comma delimiter, detected encoding) and Parquet input
 * backend-aware index handling during merge
 
 ```python
-class MyTable(CSVLoadableTableInterface, ORMTableBase, Base):
+class MyTable(FileLoadableTableInterface, ORMTableBase, Base):
     __tablename__ = "my_table"
 
 ```
 
-The main extension points here are loader choice, column mapping, and the normal SQLAlchemy model definitions themselves. Most downstream projects do not need to override much beyond `csv_columns()` and the model schema.
+The main extension points here are reader choice and the normal SQLAlchemy model definitions themselves. Most downstream projects do not need to override much beyond the model schema.
 
 3. Structured serialisation and hashing
 

docs/loaders/context.md

@@ -1,4 +1,4 @@
-# Loader Context and Diagnostics
+# Loader Context and Report
 
 This page documents the shared data structures used by the loaders
 subsystem.
@@ -21,48 +21,27 @@ on globals or implicit configuration.
 | `tableclass` | Target ORM table class |
 | `session` | Active SQLAlchemy session |
 | `path` | Path to the input file |
-| `staging_table` | SQLAlchemy Table used for staging |
-| `chunksize` | Optional chunk size |
-| `normalise` | Whether to cast values to ORM types |
-| `dedupe` | Whether to deduplicate incoming data |
-| `quote_mode` | CSV quoting mode, resolved to the same concrete mode on both the PostgreSQL COPY fast-path and the pandas ORM fallback |
+| `backend` | The session's database backend, resolved once for the load |
+| `normalise` | Whether to interpret values under the policy; when false, values are written as read |
+| `dedupe` | Whether the staging table keeps only the first row for each primary key |
+| `policy` | `ConversionPolicy`: how values are interpreted and what happens to unreadable ones |
+| `rejects_dir` | Directory that rejected rows are written to |
+| `report` | `LoadReport` row accounting |
 
-::: orm_loader.loaders.data_classes.LoaderContext
+::: orm_loader.loaders.loader.LoaderContext
 
 ---
 
-## Casting statistics
+## LoadReport
 
-Loaders track **casting failures** during normalisation to support
-debugging and auditability.
+`LoadReport` accounts for every row of a load: rows read, loaded, rejected and removed as duplicates. `DedupeResult` holds the key columns and the rows removed. A load that cannot account for every row raises.
 
-### ColumnCastingStats
-
-Tracks casting failures for a single column:
-
-- total failure count
-- representative example values
-
-### TableCastingStats
-
-Aggregates column-level statistics for a table.
-
-This enables loaders to emit warnings such as:
-
-- unexpected nulls in required columns
-- values that could not be cast to target types
-
-Statistics are logged, not raised as exceptions.
-
-::: orm_loader.loaders.data_classes.ColumnCastingStats
-::: orm_loader.loaders.data_classes.TableCastingStats
+::: orm_loader.loaders.loader.LoadReport
+::: orm_loader.backends.base.DedupeResult
 
 ---
 
 ## Design notes
 
-- Casting failures do **not** abort loads
-- Rows violating required constraints are dropped explicitly
-- Examples are capped to avoid log flooding
-
-This design favours **observability over strict enforcement**.
+- Unreadable values and missing required values raise by default, so no row disappears without a trace
+- With `ErrorPolicy(on_error="reject_row")` each rejected row is written to a rejects file with its raw text and the reason

docs/loaders/conversion.md

@@ -0,0 +1,117 @@
+# Conversion
+
+`orm_loader.conversion` interprets raw values for every file load, whatever the file format.
+
+```mermaid
+flowchart LR
+    A[Arrow column] --> B[Policy picks the column's handler]
+    B --> C[clean]
+    C --> D[detect missing]
+    D --> E[parse]
+    E --> F[check]
+    F --> G[values and invalid mask]
+```
+
+---
+
+## ConversionPolicy
+
+`ConversionPolicy` is a frozen settings object passed to `load_file(policy=...)`. It holds the settings shared by every column and one handler for each kind of column type.
+
+| Field | Meaning |
+|------|------|
+| `nulls` | `NullPolicy`: what counts as a missing value |
+| `errors` | `ErrorPolicy`: what happens to unreadable and missing required values |
+| `text`, `integer`, `floating`, `numeric`, `boolean`, `date`, `datetime` | The handler for that kind of column, each with its own settings |
+| `handlers` | Handlers for further column types, or replacements, by SQLAlchemy type class |
+| `columns` | Handlers for single columns by `(table_name, column_name)` |
+
+```python
+policy = ConversionPolicy(
+    nulls=NullPolicy(text_markers=frozenset({"NULL"})),
+    errors=ErrorPolicy(on_error="reject_row", max_rejects=0.01),
+    date=DateHandler(formats=("dd/mm/yyyy",)),
+    columns={("concept", "concept_code"): TextHandler(strip=False)},
+)
+```
+
+::: orm_loader.conversion.ConversionPolicy
+
+---
+
+## NullPolicy
+
+- Text columns keep every nonempty value verbatim. Only empty and whitespace-only text is null.
+- Words such as `NULL`, `None`, `N/A` and `NaN` stay as text in text columns. Version 2 turned them into NULL, which dropped concepts with those names. To treat words as missing again, list them in `text_markers`, for example `NullPolicy(text_markers=frozenset({"NULL", "None", "N/A", "NaN", "nan", "null", "NA", "n/a"}))`. Set operations add or remove words from a set you keep.
+- `text_markers` are exact, case-sensitive words that are null in every text column of the load. Matching ignores surrounding whitespace and applies the same way to every file format. To treat a word as missing in some columns only, subclass `TextHandler`, override `missing()` and register it in `ConversionPolicy.columns`.
+- `empty_is_null=False` keeps empty text fields as empty strings. Typed columns still treat empty fields as missing.
+- Typed columns accept a valid value, a missing word (`null`, `none`, `na`, `n/a`, `nan` and similar) or a real null, and fail otherwise.
+
+::: orm_loader.conversion.NullPolicy
+
+---
+
+## ErrorPolicy
+
+`on_error` is `raise` (default) or `reject_row`. `max_rejects` stops a `reject_row` load before the merge. A count stops the load as soon as it is passed, and a fraction is checked once all rows have been read, against the total number of rows read. A load that rejected rows logs a warning with their number and the rejects file.
+
+::: orm_loader.conversion.ErrorPolicy
+
+---
+
+## Handlers
+
+The handler for a column is the first match of: the entry in `ConversionPolicy.columns` and the handler for the most specific class of the column type. A `TypeDecorator` is handled as the type it stores. A column type with no handler is an error that names the table, column and type, raised before any row is processed. Loading such a column unchanged is an explicit choice: `columns={("table", "column"): PassthroughHandler()}`.
+
+| Handler | Column types | Accepted text | Settings |
+|------|------|------|------|
+| `TextHandler` | `String`, `Text` | Any nonempty text; longer than the column is an error | `strip` |
+| `EnumHandler` | `Enum` | One of the labels the column stores | `strip` |
+| `IntegerHandler` | `Integer`, `SmallInteger`, `BigInteger` | Optional sign and digits, with an optional all-zero fraction (`1.0`); a value outside the 16, 32 or 64 bits of the column type is an error | none |
+| `FloatHandler` | `Float` | Decimal or exponent form; `Infinity` and values that overflow to infinity are errors | none |
+| `NumericHandler` | `Numeric` | Plain and exponent decimal forms such as `1.5`, `.05` and `1.5E3`, converted to an Arrow decimal of the declared precision and scale without rounding; a precision without a scale has scale 0. A value that does not fit the declared digits is an error. The declaration must have a precision of at most 38 digits, see below | none |
+| `BooleanHandler` | `Boolean` | `true`, `t`, `yes`, `y`, `1`, `on` and their opposites | `true_words`, `false_words` |
+| `DateHandler` | `Date` | One of `formats`, validated against the calendar; year 0 is an error | `formats` |
+| `DateTimeHandler` | `DateTime` | ISO date and time, or a date in one of `formats`; see time zones below | `formats` |
+
+Dates default to `yyyymmdd`, `dd-mon-yyyy` and `yyyy-mm-dd`. Ambiguous forms such as `02/01/2020` are never guessed; declare `DateHandler(formats=("dd/mm/yyyy",))` or `("mm/dd/yyyy",)`. A timestamp column reads date-only values with its own `formats`.
+
+### Time zones
+
+The column type decides, so no value is converted by a convention the schema did not ask for.
+
+- A `DateTime(timezone=True)` column requires an offset (`Z` or `+hh:mm`) and stores UTC. A value without an offset and a date-only value are errors.
+- A `DateTime` column without a time zone rejects a value with an offset.
+
+### Typed input
+
+Input that already has a type, such as a Parquet column, is cast to the column type and checked like parsed text. A value that does not come back unchanged when cast back to its input type is an error, so an integer `2` in a `Boolean` column, a timestamp with a time in a `Date` column and a decimal that loses digits are all rejected. Float `NaN` is missing.
+
+In a `Numeric` column an integer is read exactly and a float is read as its shortest decimal text, so a Parquet float column loads the same values as a CSV of the same numbers. Infinity is an error.
+
+### Numeric columns
+
+The loader converts only what an Arrow decimal can hold, which is a precision of at most 38 digits. That is also the widest decimal DuckDB stores, and SQLite stores `Numeric` as a float. A `Numeric` column with a wider precision or no precision raises a `PolicyError` before any row is read. To load it anyway, leave the value to the database with `columns={("table", "column"): PassthroughHandler()}`: the database then parses the text and rounds a value that exceeds the scale, and a value it cannot parse aborts the load.
+
+### Adding a column type
+
+Subclass `ConvertingHandler`, implement `arrow_type` and `parse`, and pass the handler in `ConversionPolicy.handlers` for the type class, or in `ConversionPolicy.columns` for a single column. `clean`, `missing` and `check` can be overridden.
+
+```python
+class HexHandler(ConvertingHandler):
+    def arrow_type(self, sa_type):
+        return pa.int64()
+
+    def parse(self, cleaned, sa_type):
+        ...  # return Interpretation(values, invalid)
+
+policy = ConversionPolicy(handlers={MyHexType: HexHandler()})
+```
+
+### ConvertingHandler
+
+::: orm_loader.conversion.ConvertingHandler
+
+### ColumnHandler
+
+::: orm_loader.conversion.ColumnHandler