Repository navigation
One file-to-database pipeline with a single conversion policy - #46
Conversation
…checks, quote mode, malformed lines
… all dialects, typed time zones
…ing passed as an argument, defaults filled by the processor
… that logs the toggling
…atabase-assigned keys, rename insert_if_empty to insert, move reader options off LoaderContext
…pragma took effect, remove unused FK and connection hooks
…s; evaluate expression defaults once; move materialized views to their own backend interface
…es, let stage_file roll back
gkennos
left a comment
There was a problem hiding this comment.
Changes needed before merge:
- Text null words: keep the new default (it fixes #44), but document the change under Breaking changes and ship the 2.x words as a preset that can be selected or adapted. Details inline on
conversion/policy.py. - Re-export
infer_delimfromorm_loader.loaders; downstream code still imports it from the old path. - Inline, blocking: per-row SQL defaults (
loaders/processor.py) and primary-key presence (loaders/loader.py).
…and column-name handling, report rejects per row - SQL-expression defaults are applied per row by COALESCE in the merge; their staging columns are nullable and evaluate_default is removed - Numeric without precision, or above 76 digits, is read as text; 39-76 digits use decimal256; invalid scale is rejected - Legacy null markers cover the pandas/pyarrow defaults and case variants - Reject counts and reasons are reported per row; the row limit is checked after processing - Primary keys and model columns are keyed by database column name; computed columns are skipped; primary keys stay required without normalisation - infer_delim is exported; staging and conversion docs updated
…he role into the pool relaxed_foreign_keys sets session_replication_role with SET LOCAL and reapplies it whenever a transaction begins on the selected bind, so a commit that switches pooled connections keeps keys relaxed and no connection is returned to the pool in replica. The role found in an open transaction is restored on exit. A scoped_session relaxes only the current thread's session. Docs say to flush ORM inserts inside the block.
…ct limits at the end, keep computed columns out of loads - model_columns() excludes computed columns, so the reader, processor and required_columns agree - a primary key is always required in the file and never takes a default - SQL-expression defaults are applied by COALESCE in the merge and are not written by the processor - a count limit stops a load at once, a fraction is checked against all rows read, and a load that rejected rows logs a warning - Numeric is cast up to 38 digits and kept as validated text beyond that or without a precision - handlers validate the declared column type through validate(), which PassthroughHandler no longer trips over - the review regression tests move into topical test files
… marker_quoting and LEGACY_TEXT_MARKERS are removed - readers take no NullPolicy and hand over text as read, so empty_is_null=False keeps empty strings in every format - text markers match after trimming for every format, and a reader can no longer disagree with the policy - infer_delim and infer_encoding are no longer exported from orm_loader.loaders - docs describe how to restore the 2.x words with text_markers and the raw pass-through of normalise=False
…longer uses - relaxed_foreign_keys sets the replication role per transaction, so disable_fk_check and restore_fk_check are gone from PostgresBackend - the role check accepts only the role names Postgres reports
|
On the Differs from the suggestion: there is no re-export from |
cava-pipelines uses this but I checked and they have a dependency ceiling version set so no issue to migrate - this is fine |
Summary
ConversionPolicy, and not in several loaders.Closes
NULL,None,N/A,NaN) are no longer turned into NULL in text columns, and only empty values are007or1.50are no longer rewrittenWhat changed
stage_file(ctx, reader, staging)runs reader,BatchProcessorand the backend's bulk sinkorm_loader.conversion)ConversionPolicyis passed toload_file(policy=...)and holdsNullPolicy,ErrorPolicyand one handler per kind of columnNullPolicysets what counts as missing: empty values, optional text markers, and whether quoting mattersErrorPolicysetsraise(default) orreject_row, with an optionalmax_rejects. A count stops the load as soon as it is passed, and a fraction is checked against all rows read once the file has been readConvertingHandlersubclass (clean, detect missing, parse, check) added through the policypolicy.columns, then the column type's class hierarchy. A type with no handler raises before any row is read, and passthrough is an explicit choiceNumericcolumns are read by Arrow into a decimal of the declared precision and scale, from plain and exponent forms such as1.5E3, and a value that would round or overflow is an error. Text is checked before it reaches Arrow's parser, which silently wraps on very long digit strings (apache/arrow#49817) and returns0.00for exponents such as5e-50in the versions tested (pyarrow 23.0.1 to 26.0.1)Numericcolumn an integer is widened before the cast, so anint64column loads, 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 numbersorm_loader.loaders.readers)FileReaderis the contract, withDelimitedTextReaderandParquetReader, andopen_readerpicks one from the file suffixload_file(reader=...)accepts a reader, andquote_modeandchunksizeconfigure the default one. Passing either together with a reader raisesValueErrorquote_modedefaults toby_delimiterreject_row, lines with the wrong number of fields are written to the rejects file as well, with their row number in the fileBatchProcessorapplies the handlers, checks required columns and writes rejected rows with their raw text and reasonLoadReportaccounts for every row read as loaded, rejected or deduplicated, and a load that cannot account for every row raisesCOPYwritten from Arrowexecutemanystage_filerolls the session back after its first write (DuckDB commits every few batches, so earlier batches stay committed)load_fileresolves the backend once and holds it onLoaderContext.backendreplacedeletes the matching keys and inserts,upsertskips keys that already exist and the first of a repeated key wins, andinsertis a plain insert whose key collision raises the database'sIntegrityErrorinsertwithdedupe=Trueis fasterIngestErrorDedupeResultload_filereads the state back and raises when the session already has uncommitted writesTable.validate_foreign_keys(session)checks every FK with an anti-join, in any load order, and raisesForeignKeyViolationErrorBreaking changes
Compared with
main.CSVLoadableTableInterfaceis nowFileLoadableTableInterfaceCSVTableProtocolis nowFileTableProtocolload_csvis nowload_file_select_loaderis removed, andload_file(reader=...)replaces itcsv_columnsis removed, and readers map file headers tomodel_columns()LoaderInterface,PandasLoader,ParquetLoaderand theloader=argument ofload_filenormalise_nullCastRule,register_column_cast_rule,TableCastingStatsandColumnCastingStatsinfer_delim,infer_encodingandquick_load_pgfromorm_loader.loaders. The detection functions are inorm_loader.loaders.readers.detectionorm_staging_loadandorm_file_loadautovalue ofquote_modeLoaderContext.staging_table,LoaderContext.staging_schema_tag,LoaderContext.quote_modeandLoaderContext.chunksize. The staging table is passed tostage_fileand the sinks, the context holds the resolvedbackend, and the reader options belong toload_fileand the readerget_staging_table,load_staging,create_staging_tableanddrop_staging_table: use the backend'screate_staging_tableanddrop_staging_tableORMTableBase.pk_valuesandpk_tupleclean_kwargsandfrom_dicthelpers/bulk.pyandhelpers/sqlite.py: useresolve_backend(session).relaxed_foreign_keys(session), andload_filealready handles FKs for file loadsbulk_load_context,merge_context,enable_fk_checkandload_staging_fastBackendCapabilities,capabilities,supports_dialectandrequires_capability.supports_replace_with_kept_indexesis a class attribute of the backendinstall_engine_hooks,configure_dbapi_connection,explain_fk_errorandrestore_journal_mode, and the Postgresengine_with_replica_roleDatabaseBackend: they exist only on backends that inheritMaterializedViewBackend, which is PostgreSQL, andMaterializedViewMixinraisesNotImplementedErroron the othersmerge_replace,merge_upsertandmerge_inserttake astagingparameter, andcreate_staging_tablereturns the table it createdmanage_indicesandmerge_from_stagingtake thebackendinstead of resolving itinsertreplacesinsert_if_empty. It does not check that the target is empty, and a key collision raises the database'sIntegrityErrorreplaceandupsertno longer turn into a plain insert when the target is empty. Upsert into an empty target only logs a hint, and a repeated key keeps the first rowIngestErrorinstead of only being loggedquote_modedefaults toby_delimiter, which reads quotes in tab-delimited files as dataload_fileraises when the session has uncommitted writes, because the foreign key pragma cannot change inside a transactionNULL,None,N/AandNaNstay as text in text columns, and only an empty field is NULL. To treat words as missing again, list them inNullPolicy(text_markers=...)null,none,na,n/a,nan,#n/a,-nanand<na>, case-insensitiveInfinity, and year 0 are errorsEnumcolumn accepts only its labelsNumericcolumn must declare a precision of at most 38 digits. A wider or unconstrainedNumericraises aPolicyErrorbefore any row is read, andcolumns={(table, column): PassthroughHandler()}leaves such a column to the database asmaindidNumericvalue that does not fit the declared scale or precision is an error.mainpassed the text to the database, which rounded it (1.555into scale 2 became1.56) or aborted the whole loadDeferred
Numericcolumns of more than 38 digits, or with no declared precision, in the loaderNumericas a floatPolicyErrorand does not get a separate text path with its own validation. It loads today with aPassthroughHandler, which leaves parsing and rounding to the databaseBenchmark
main(3998ff9), after is this branchvocab_test_duckdbset, tab-delimited:CONCEPT(2.3M rows),CONCEPT_SYNONYM(2.5M rows) andDRUG_STRENGTH(210k rows)auto, the new loader's default isby_delimiter, which is also what omop-alchemy usesCOPYfirst and falls back to pandas when it fails. The path that ran is recorded, and the pandas path is also run on its ownby_delimiterby_delimiterby_delimiterby_delimiterauto(default), newby_delimiter(default)by_delimiterby_delimiterby_delimiterby_delimiterauto(default), newby_delimiter(default)by_delimiterby_delimiterby_delimiterby_delimiterauto(default), newby_delimiter(default)by_delimiterthe oldCOPYpath and the new loader take about the same time, and the old path is faster on the smallest file, where fixed costs dominate. The new loader parses, validates and re-serialises the data instead of copying the fileCOPYpath and its pandas fallback gave different data for the same fileCOPYkeepsNULL,None,N/AandNaNas text and keeps a quoted""as an empty stringCOPYlets the Postgres session parse dates, so02/01/2020is 1 Feb or 2 Jan depending onDateStyle, while the pandas path always reads 2 JanNaNin numeric columns and the pandas path makes it NULL007became7and1.50became1.5onandoffinto NULL without an errorCOPY, and the pandas fallback then drops that row with no errorCOPYfailed onCONCEPTandCONCEPT_SYNONYM, two of the three files hereautosamples the first 2000 rows and chooses CSV quoting for these tab-delimited files, because the first field that starts with a quote ("Sick" headaches) is on line 103,811. In CSV mode it opens a quoted field that swallows the following fieldsCOPYfails, and the loader logs a warning and switches to pandasCONCEPTrows and 1CONCEPT_SYNONYMrow namedNone,N/AorNull, and rewrote 2,319concept_codevaluesby_delimiterthe new loader loads every row unchangedautomode. Its defaultby_delimiterreads quotes in tab-delimited files as data, andcsvis an explicit choice for files that quote fieldsCONCEPTon all three backends