diff --git a/CHANGELOG.md b/CHANGELOG.md index 92d671a..4c30a5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ Framework `datapipeline` (Gem). Neueste Version oben. +## 2.1.0 – 2026-10-09 + +- Sperre gegen parallele Läufe: `lock 'datei'` in der Pipeline, `etl --lock=datei` für das ganze Skript; ein zweiter + Lauf bricht mit `DataPipeline::LockError` ab, Exit-Code `3` +- postgres, sqlite: `upsert: %w[schlüssel]` (`insert … on conflict … do update`); sqlite legt mit `create: true` den + Unique-Index an +- postgres: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion statt je Zeile + einzeln; `batch_size: :all` = eine Transaktion für den ganzen Stream, nach einem Abbruch zurückgerollt +- csv: `encoding:` (z. B. `'ISO-8859-1'`, `'Windows-1252'`), `replace:` für nicht darstellbare Zeichen, `bom: true`; + ein UTF-8-BOM wird beim Lesen übersprungen + ## 2.0.0 – 2026-10-08 - Gem `datapipeline`: `datapipeline.gemspec`, Executable `etl` (exe/etl), Code unter `lib/datapipeline/` diff --git a/Gemfile.lock b/Gemfile.lock index ac649ff..522b1d0 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - datapipeline (2.0.0) + datapipeline (2.1.0) csv (~> 3.3) logger (~> 1.6) diff --git a/README.md b/README.md index ce5c6d8..1e73441 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # DataPipeline – generisches ETL-Framework in Ruby -Stand: 2026-10-08 · Gem `datapipeline` 2.0.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 +Stand: 2026-10-09 · Gem `datapipeline` 2.1.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 --- @@ -574,6 +574,7 @@ flowchart LR | [lib/datapipeline.rb](lib/datapipeline.rb) | DSL und Ausführung (`run`), `Stats`, `Context` | | [lib/datapipeline/validation.rb](lib/datapipeline/validation.rb) | Regeln für `validate` | | [lib/datapipeline/watermark.rb](lib/datapipeline/watermark.rb) | Wasserzeichen für inkrementelle Läufe | +| [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe | | [lib/datapipeline/layout.rb](lib/datapipeline/layout.rb) | Satzarten-Layouts (Feldliste je Satzart, siehe 2.9) | | [exe/etl](exe/etl) | Kommando `etl` des Gems mit Exit-Codes (siehe 2.7); [bin/etl](bin/etl) startet es im Repository mit dem Bundle | | [datapipeline.gemspec](datapipeline.gemspec), [CHANGELOG.md](CHANGELOG.md) | Gem-Beschreibung, Versionsverlauf | @@ -611,6 +612,7 @@ flowchart LR | `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Datenschritten (siehe 2.7) | | `on_finish { \|stats\| ... }` | nach jedem Lauf, auch nach Abbruch | | `watermark(name, column:, file:, initial:)` | letzter verarbeiteter Wert für inkrementelle Läufe (siehe 2.7) | +| `lock(file)` | Sperre während `run`; läuft schon ein Lauf mit derselben Datei, bricht dieser mit `DataPipeline::LockError` ab (siehe 2.7) | | `dry_run(flag = true)`, `limit(n)` | nichts schreiben bzw. nur die ersten n Quellzeilen | | `dry_run?` | in Schritten mit Nebenwirkungen (API-Aufrufe): `true` im Dry-Run | | `logger(logger)` | eigener Ruby-`Logger`; Default: Warnungen auf `$stderr` | @@ -664,7 +666,7 @@ mit 30 MB statt 614 MB im Batch). Die Datenzeilen sind dieselben wie im Batch, m | `header` | kennt `read` und `written` | wird zuerst geschrieben, `read` und `written` sind `nil` | | `ctx.read` in `map`/`filter` | Anzahl Quellzeilen | `nil` | | mehrere `transform` im Parallelmodus | erlaubt | `ArgumentError`, dafür `chained` setzen | -| Postgres als Ziel | eine Transaktion für alles | je Zeile ein `INSERT` ohne gemeinsame Transaktion | +| Postgres als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen (Default 1000); `batch_size: :all` eine für alles | Postgres als Quelle liest im Streaming über einen serverseitigen Cursor (`fetch_size`, Default 1000). Abfragen und Lookups auf derselben Verbindung bleiben währenddessen möglich. @@ -689,21 +691,31 @@ Alle Plugins liefern Zeilen mit **String-Schlüsseln**. Die mitgelieferten Plugi ### 2.5 Mitgelieferte Plugins **DataPipeline::PostgresStorage** – Optionen `host`, `port` (Default 5432), `database`, `username`, `password`, `query`, `params`, -`table`, `typed`, `fetch_size`, `truncate`. +`table`, `typed`, `fetch_size`, `truncate`, `upsert`, `batch_size`. - `read` / `query` mit Platzhaltern `$1, $2, …` und `params: [...]`; Ergebnis als `Hash` (`NULL` = `nil`), mit `typed: true` als Ruby-Typen (Integer, Float, Date, …). - `read { |row| }` liest über einen Cursor in einer Transaktion, je `fetch_size` Zeilen (Default 1000). - `write(rows)` schreibt in `table:` (oder alte Form `write(table, rows)`), alle Zeilen in **einer Transaktion**, Werte als Bind-Parameter, `nil` = `NULL`, Tabellen-/Spaltennamen gequotet (`schema.tabelle` erlaubt). - `truncate: true` leert die Tabelle vor dem ersten Schreiben (bei `write` in derselben Transaktion). +- `upsert: %w[belnr]` schreibt `insert … on conflict (belnr) do update set …` für alle übrigen Spalten der Zeile (ohne + übrige Spalten `do nothing`). Die Tabelle braucht einen Primärschlüssel oder Unique-Index auf genau diesen Spalten. + Damit kann ein Lauf nach einem Abbruch wiederholt werden, ohne doppelte Zeilen. +- `write_row` (Streaming) sammelt die Zeilen und schreibt je `batch_size` Zeilen (Default 1000) in einer Transaktion; + `finalize` und `close` schreiben den Rest. Mit `batch_size: :all` läuft der ganze Stream in **einer** Transaktion: + `finalize` am Ende eines erfolgreichen Laufs bestätigt sie, nach einem Abbruch rollt `close` sie zurück, auch ein + `truncate`. Dafür braucht das Ziel eine eigene Verbindung (nicht dieselbe wie die Quelle). ```ruby pg = container :postgres, host: 'localhost', database: 'crm', username: 'etl', table: 'kunde_import' read pg, query: 'select * from kunde where geaendert >= $1', params: [Date.today - 7] + +# wiederholbar: vorhandene Belege werden aktualisiert, im Streaming alles oder nichts +write container(:postgres, **db, table: 'beleg', upsert: %w[belnr], batch_size: :all) ``` **DataPipeline::CsvStorage** – Optionen `file` (Pflicht), `csv:` (an Rubys `CSV`, gemergt mit den Defaults `headers: true`, -`col_sep: ','`), `columns`, `counter`, `verbose`. +`col_sep: ','`), `columns`, `encoding`, `replace`, `bom`, `counter`, `verbose`. - `columns: %w[a b c]` legt die Spaltenreihenfolge fest: fehlende Felder bleiben leer, String- und Symbol-Schlüssel werden gefunden, die Kopfzeile kommt aus `columns`. Ohne `columns` gilt die Hash-Reihenfolge (nötig für Dateien mit mehreren Satzarten wie CUDO). @@ -711,13 +723,22 @@ read pg, query: 'select * from kunde where geaendert >= $1', params: [Date.today - Leere Daten: ohne `columns` leere Datei, mit `columns` nur die Kopfzeile. - `layout: DataPipeline::Layout` schreibt Dateien mit mehreren Satzarten in der Feldreihenfolge je Satzart, ohne Kopfzeile; ein Feld, das nicht im Layout steht, bricht ab (siehe 2.9). +- `encoding: 'ISO-8859-1'` (oder `'Windows-1252'`, …) ist der Zeichensatz der Datei, Default UTF-8. Gelesen wird als + UTF-8, geschrieben wird von UTF-8 umkodiert. Ein Zeichen, das der Zeichensatz nicht kennt (z. B. `€` in ISO-8859-1), + bricht mit `Encoding::UndefinedConversionError` ab; mit `replace: '?'` wird es ersetzt. Ein UTF-8-BOM am + Dateianfang wird beim Lesen übersprungen, `bom: true` schreibt eines (für Excel). Ein `encoding` in `csv:` wird wie + bisher unverändert an Ruby weitergegeben. + +```ruby +write container(:csv, file: 'out/cudo.txt', encoding: 'Windows-1252', layout: LAYOUT, csv: { headers: false, col_sep: '|' }) +``` Übersicht der übrigen Plugins (alle mit `read`, `read { }`, `write`, `write_row`, `close`): | `container` | Klasse | Gem (bei Nutzung geladen) | Wichtige Optionen | | --- | --- | --- | --- | | `:json` | `DataPipeline::JsonStorage` | – | `file`, `format: :array \| :lines` (Default aus Endung: `.ndjson`/`.jsonl` → Zeilen), `root: 'data.items'`, `pretty`; ein einzelnes Objekt wird als eine Zeile gelesen | -| `:sqlite`, `:sqlite3` | `DataPipeline::SqliteStorage` | `sqlite3` | `database` (Datei oder `':memory:'`), `query`, `params` (Array für `?`, Hash für `:name`), `table`, `create`, `truncate` | +| `:sqlite`, `:sqlite3` | `DataPipeline::SqliteStorage` | `sqlite3` | `database` (Datei oder `':memory:'`), `query`, `params` (Array für `?`, Hash für `:name`), `table`, `create`, `truncate`, `upsert` | | `:excel`, `:xlsx` | `DataPipeline::ExcelStorage` | `roo` (lesen), `caxlsx` (schreiben) | `file`, `sheet` (Name/Index), `headers`, `columns`, `skip` (Titelzeilen) | | `:rest` | `DataPipeline::RestStorage` | `http` | `url`, `auth`, `headers`, `params`, `root`, Paging `page_param` oder `next: ->(body) { … }`, `retries`, `timeout`, `batch_size`, `write_method`, `write_url` (auch als Lambda je Zeile), `ssl_verify` | @@ -727,7 +748,8 @@ read pg, query: 'select * from kunde where geaendert >= $1', params: [Date.today **DataPipeline::SqliteStorage** – Verhält sich wie Postgres: Bind-Parameter, Schreiben in einer Transaktion, Tabellen- und Spaltennamen gequotet, `nil` = `NULL`, Datum als ISO-Text. Werte kommen mit SQLite-Typen zurück (Integer, Float, String). `create: true` legt die Tabelle beim ersten Schreiben aus der ersten Zeile an (Integer → `integer`, Float → `real`, -sonst `text`), `truncate: true` leert sie vorher. Die Datei wird erst bei der ersten Nutzung geöffnet; ein Dry-Run +sonst `text`), `truncate: true` leert sie vorher. `upsert: %w[belnr]` wie bei Postgres; mit `create: true` wird der +nötige Unique-Index gleich mit angelegt. Die Datei wird erst bei der ersten Nutzung geöffnet; ein Dry-Run legt keine an. **DataPipeline::ExcelStorage** – Lesen von xlsx zeilenweise (`each_row_streaming`), leere Zeilen werden übersprungen. Beim @@ -911,6 +933,20 @@ wm = watermark :rechnungen, column: 'rechnung_id', file: 'data/.state.json', ini read pg, query: 'select * from rechnung where rechnung_id > $1 order by rechnung_id', params: [wm.value] ``` +**Sperre gegen parallele Läufe** – Startet ein Cronjob, während der vorige Lauf noch arbeitet, lesen beide dasselbe +Wasserzeichen und exportieren doppelt. `lock` verhindert das mit einer exklusiven Sperre (`flock`) auf eine Datei: +Ein zweiter Lauf wartet nicht, sondern bricht sofort mit `DataPipeline::LockError` ab (Exit-Code `3`), schreibt nichts +und ruft wie jeder Abbruch `on_finish` auf. Das Betriebssystem gibt die Sperre beim Ende des Prozesses frei, auch nach +einem Absturz; es bleiben also keine verwaisten Sperren. Die Datei bleibt liegen und enthält die PID des letzten Laufs. + +```ruby +lock 'data/.cudo.lock' # in der Pipeline: von run bis zum Ende des Laufs +``` + +```sh +bin/etl run cudo.rb --lock=data/.cudo.lock # für das ganze Skript, z. B. mehrere Pipelines nacheinander +``` + **Dry-Run und Limit** – Im Dry-Run läuft alles, aber nichts wird geschrieben: kein Ziel, keine Fehlerdatei, kein Wasserzeichen; eine CSV-Datei wird nicht angelegt. Am Ende stehen die Zusammenfassung und die ersten Zeilen (`preview`, Default 10) als JSON auf `$stdout`. `limit n` verarbeitet nur die ersten n Quellzeilen, im Streaming @@ -929,10 +965,11 @@ bin/etl run pipeline.rb --param land=DE # im Skript: params[:land] | `--from`, `--till` | in `params[:from]`/`params[:till]` und unverändert an das Skript weitergereicht (eigener `OptionParser` der Importer) | | `--param KEY=VALUE` | `params[:key]`, mehrfach möglich | | `--dry-run`, `--limit=N` | für alle Pipelines des Skripts | +| `--lock=FILE` | Sperre für das ganze Skript; ist sie belegt, endet `etl` sofort mit Exit-Code `3` | | `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default über `bin/etl`: Info) | | jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` | -Exit-Code: `0` ok, `1` Abbruch, `2` fertig, aber mit verworfenen Zeilen. Das `Rakefile` ruft `bin/etl` auf: +Exit-Code: `0` ok, `1` Abbruch, `2` fertig, aber mit verworfenen Zeilen, `3` ein anderer Lauf hält die Sperre. Das `Rakefile` ruft `bin/etl` auf: `rake process:posting[2025-11-01,2025-11-30]`, `rake process:all[2025-11-01,2025-11-30,dry]`. Ein fehlgeschlagener Lauf stoppt Rake. ### 2.8 Tests diff --git a/exe/etl b/exe/etl index 8f7f909..5cac587 100755 --- a/exe/etl +++ b/exe/etl @@ -9,9 +9,10 @@ # --dry-run run everything, write nothing, print the first rows # --limit=N only the first N source rows # --verbose log level debug, --quiet only warnings (default info) +# --lock=FILE exclusive lock for the whole script, exit 3 when another run holds it # --name=value any other option: DataPipeline.params[:name] and passed on to the script # -# exit code: 0 ok, 1 aborted, 2 finished with rejected rows +# exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run # require 'optparse' require 'logger' @@ -31,6 +32,7 @@ parser = OptionParser.new do |opts| opts.on('--limit=N', Integer) { |v| options[:settings][:limit] = v } opts.on('--verbose') { options[:settings][:log_level] = Logger::DEBUG } opts.on('--quiet') { options[:settings][:log_level] = Logger::WARN } + opts.on('--lock=FILE', String) { |v| options[:lock] = v } end # own options for the parser, everything else (also --name=value of the script) goes to the script @@ -38,7 +40,7 @@ own = [] args = [] argv = ARGV.dup while (arg = argv.shift) - if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet)(=|\z)/) + if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock)(=|\z)/) own << arg own << argv.shift if %w[--param --limit --from --till].include?(arg) else @@ -54,6 +56,13 @@ abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script) DataPipeline.settings(options[:settings]) +begin + script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock] # held until the process ends +rescue DataPipeline::LockError => e + warn "#{File.basename(script)}: #{e.message}" + exit 3 +end + # the pipeline script sees only its own arguments ARGV.replace(options[:passthrough] + args) @@ -61,10 +70,11 @@ begin load File.expand_path(script) rescue StandardError => e warn "#{File.basename(script)}: #{e.class}: #{e.message}" - exit 1 + exit(e.is_a?(DataPipeline::LockError) ? 3 : 1) end runs = DataPipeline.runs +exit 3 if runs.any? { |stats| stats.error.is_a?(DataPipeline::LockError) } exit 1 if runs.any?(&:error) exit 2 if runs.any? { |stats| stats.rejected.positive? } exit 0 diff --git a/lib/datapipeline.rb b/lib/datapipeline.rb index 7a970b2..b49538d 100644 --- a/lib/datapipeline.rb +++ b/lib/datapipeline.rb @@ -6,6 +6,7 @@ require_relative 'datapipeline/storage/storage_plugin' require_relative 'datapipeline/validation' require_relative 'datapipeline/layout' require_relative 'datapipeline/watermark' +require_relative 'datapipeline/lock' class DataPipeline # plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel), @@ -58,6 +59,7 @@ class DataPipeline limit: nil, preview: 10, on_error: :abort, + lock: nil, log_level: Logger::WARN }.freeze @@ -204,6 +206,11 @@ class DataPipeline @on_finish << block end + # lock 'data/.cudo.lock' -> held during run; a second run at the same time aborts with LockError + def lock file + @settings[:lock] = file + end + # see DataPipeline::Watermark def watermark name, column:, file:, initial: nil @watermark = Watermark.new(name, column:, file:, initial:) @@ -303,6 +310,7 @@ class DataPipeline @logger.info "start#{" #{@settings[:version]}" if @settings[:version]}#{@settings[:dry_run] ? ' (dry run)' : ''}" @preview = [] raise ArgumentError, 'read storage missing' unless @source + (@lock = Lock.new(@settings[:lock])).acquire if @settings[:lock] raise ArgumentError, 'write storage missing' if @targets.empty? before = @transformations.select { |t| EDGE_BEFORE.include?(t[:kind]) } @@ -342,6 +350,7 @@ class DataPipeline def finish @stats.finished_at = Time.now [@source, *@targets.map { |t| t[:storage] }, @reject_target].compact.uniq.each(&:close) + @lock&.release self.class.runs << @stats print_preview if @stats.dry_run && @stats.error.nil? @logger.info "finish #{@stats}" diff --git a/lib/datapipeline/lock.rb b/lib/datapipeline/lock.rb new file mode 100644 index 0000000..0f86be3 --- /dev/null +++ b/lib/datapipeline/lock.rb @@ -0,0 +1,57 @@ +require 'fileutils' + +class DataPipeline + # another run holds the lock + class LockError < StandardError; end + + # + # exclusive lock on a file, against overlapping runs (cron) + # + # lock 'data/.cudo.lock' -> in a pipeline: held from the start to the end of run + # etl run cudo.rb --lock=data/.lock -> for the whole script + # + # the lock is an flock on the file, released by the operating system when the process ends, + # also after a crash; the file stays and holds the pid of the last holder + # + class Lock + attr_reader :file + + def initialize file + @file = file.to_s + end + + def acquire + return self if @io + + FileUtils.mkdir_p(File.dirname(@file)) + io = File.open(@file, File::RDWR | File::CREAT, 0o644) + unless io.flock(File::LOCK_EX | File::LOCK_NB) + holder = io.read.strip + io.close + raise LockError, "#{@file} is locked#{" by pid #{holder}" unless holder.empty?}: another run is still active" + end + io.truncate(0) + io.write(Process.pid.to_s) + io.flush + @io = io + self + end + + def release + return unless @io + + @io.flock(File::LOCK_UN) + @io.close + @io = nil + end + + def locked? = !@io.nil? + + def synchronize + acquire + yield + ensure + release + end + end +end diff --git a/lib/datapipeline/storage/csv.rb b/lib/datapipeline/storage/csv.rb index af5de55..c45d279 100644 --- a/lib/datapipeline/storage/csv.rb +++ b/lib/datapipeline/storage/csv.rb @@ -11,6 +11,10 @@ class DataPipeline # without columns the values are written in hash order # layout: DataPipeline::Layout, field order per record type (several record types in one file), # unknown fields raise; no header line + # encoding: encoding of the file, e.g. 'ISO-8859-1' or 'Windows-1252' (default UTF-8); + # read as UTF-8 strings, written transcoded from UTF-8; a UTF-8 BOM is skipped when reading + # replace: '?' -> characters the encoding cannot hold are replaced when writing (default: error) + # bom: true -> write a UTF-8 BOM (Excel) # counter: print a dot per written row # verbose: print settings # @@ -44,11 +48,11 @@ class DataPipeline def read(&block) if block_given? - CSV.foreach(@file_name, **csv_options) { |row| yield to_hash(row) } + CSV.foreach(@file_name, **read_options) { |row| yield to_hash(row) } return end - CSV.read(@file_name, **csv_options).map { |row| to_hash(row) } + CSV.read(@file_name, **read_options).map { |row| to_hash(row) } end def write data @@ -93,10 +97,30 @@ class DataPipeline @options[:columns] end + def utf8? + @options[:encoding].nil? || Encoding.find(@options[:encoding]) == Encoding::UTF_8 + end + + # csv: { encoding: ... } is passed unchanged (older scripts) + def read_options + return csv_options if csv_options.key?(:encoding) + + csv_options.merge(encoding: utf8? ? 'bom|utf-8' : "#{@options[:encoding]}:utf-8") + end + + def write_options + return csv_options if csv_options.key?(:encoding) || utf8? + + options = csv_options.merge(encoding: @options[:encoding]) + options.merge!(undef: :replace, invalid: :replace, replace: @options[:replace]) if @options[:replace] + options + end + def open_file return if @file - @file = CSV.open(@file_name, 'w', **csv_options) + @file = CSV.open(@file_name, 'w', **write_options) + @file.to_io.write("\uFEFF") if @options[:bom] && utf8? @header_written = false return unless columns && csv_options[:headers] && !@options[:layout] diff --git a/lib/datapipeline/storage/postgres.rb b/lib/datapipeline/storage/postgres.rb index d39016a..ec38d75 100644 --- a/lib/datapipeline/storage/postgres.rb +++ b/lib/datapipeline/storage/postgres.rb @@ -11,6 +11,9 @@ class DataPipeline # typed: true -> values come back as ruby types (Integer, Float, Date, ...) # false (default) -> all values as String, NULL as nil # fetch_size: rows per fetch when reading with a block (default 1000) + # upsert: %w[belnr] -> insert ... on conflict (belnr) do update; needs a unique index on these columns + # batch_size: write_row collects rows and writes them in one transaction per batch (default 1000); + # :all -> one transaction for the whole stream, committed by finalize, rolled back by close # class PostgresStorage < StoragePlugin def initialize **options, &block @@ -65,25 +68,88 @@ class DataPipeline table_name, data = data.nil? ? [@options[:table], table_or_data] : [table_or_data, data] raise "Table for PostgresStorage#write missing" unless table_name - @conn.transaction do + transaction do truncate_once table_name data.each { |row| insert table_name, row } end end + # rows are collected and written per batch_size in one transaction def write_row row raise "Table for PostgresStorage#write_row missing" unless @options[:table] - truncate_once @options[:table] - insert @options[:table], row + if batch_size == :all + begin_stream + truncate_once @options[:table] + insert @options[:table], row + else + (@pending ||= []) << row + flush if @pending.size >= batch_size + end end + # end of a stream: remaining rows, commit of batch_size: :all + def finalize + flush + if @stream_open + @conn.exec('commit') + @stream_open = false + end + close + end + + # pending rows are written; an open stream with batch_size: :all is rolled back (run aborted) + # an error while flushing does not hide the error that aborted the run def close - @conn.close if @conn && !@conn.finished? + return unless @conn && !@conn.finished? + + aborted_by = $! + begin + flush + rescue StandardError + raise unless aborted_by + end + @conn.exec('rollback') if @stream_open + @stream_open = false + @conn.close end private + def batch_size + size = @options.fetch(:batch_size, 1000) + size == :all ? :all : Integer(size) + end + + def flush + return if @pending.nil? || @pending.empty? + + rows = @pending + @pending = [] + transaction do + truncate_once @options[:table] + rows.each { |row| insert @options[:table], row } + end + end + + def begin_stream + return if @stream_open + + raise 'PostgresStorage: batch_size: :all needs its own connection (no open transaction)' unless idle? + + @conn.exec('begin') + @stream_open = true + end + + # inside a transaction (e.g. a cursor of the same connection) no own transaction is opened + def transaction(&block) + idle? ? @conn.transaction(&block) : yield + end + + def idle? + @conn.transaction_status == PG::PQTRANS_IDLE + end + # rows are fetched in batches inside a transaction; other queries on the # same connection (lookups, subqueries) are possible while reading def each_row sql, params @@ -112,11 +178,24 @@ class DataPipeline columns = row.keys.map { |col| @conn.quote_ident(col.to_s) }.join(',') placeholders = (1..row.size).map { |i| "$#{i}" }.join(',') @conn.exec_params( - "insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})", + "insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})#{upsert_clause(row)}", row.values ) end + # on conflict (keys) do update set = excluded. + def upsert_clause row + keys = Array(@options[:upsert]).map(&:to_s) + return '' if keys.empty? + + others = row.keys.map(&:to_s) - keys + target = keys.map { |key| @conn.quote_ident(key) }.join(',') + return " on conflict (#{target}) do nothing" if others.empty? + + updates = others.map { |col| "#{@conn.quote_ident(col)} = excluded.#{@conn.quote_ident(col)}" }.join(', ') + " on conflict (#{target}) do update set #{updates}" + end + # schema.table -> "schema"."table" def quote_table name name.to_s.split('.').map { |part| @conn.quote_ident(part) }.join('.') diff --git a/lib/datapipeline/storage/sqlite.rb b/lib/datapipeline/storage/sqlite.rb index c92f1d0..5e516fe 100644 --- a/lib/datapipeline/storage/sqlite.rb +++ b/lib/datapipeline/storage/sqlite.rb @@ -10,6 +10,8 @@ class DataPipeline # create: true -> create the table on the first write, columns from the first row # (Integer -> integer, Float -> real, else text) # truncate: true -> delete all rows of the table before the first write + # upsert: %w[belnr] -> insert ... on conflict (belnr) do update; with create: true the unique index + # is created too, otherwise the table needs one on these columns # # rows come back as Hash with sqlite types (Integer, Float, String, nil). # the database file is opened on first use: a dry run creates no file. @@ -86,6 +88,7 @@ class DataPipeline if @options[:create] columns = row.map { |key, value| "#{quote(key)} #{column_type(value)}" }.join(', ') db.execute("create table if not exists #{quote_table(table_name)} (#{columns})") + create_upsert_index table_name if upsert_keys.any? end db.execute("delete from #{quote_table(table_name)}") if @options[:truncate] @prepared[table_name] = true @@ -108,7 +111,29 @@ class DataPipeline columns = row.keys.map { |col| quote(col) }.join(',') placeholders = (['?'] * row.size).join(',') values = row.values.map { |v| v.is_a?(Date) || v.is_a?(Time) ? v.iso8601 : v } - db.execute("insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})", values) + db.execute("insert into #{quote_table(table_name)} (#{columns}) values (#{placeholders})#{upsert_clause(row)}", values) + end + + def upsert_keys + Array(@options[:upsert]).map(&:to_s) + end + + # on conflict (keys) do update set = excluded. + def upsert_clause row + return '' if upsert_keys.empty? + + others = row.keys.map(&:to_s) - upsert_keys + target = upsert_keys.map { |key| quote(key) }.join(',') + return " on conflict (#{target}) do nothing" if others.empty? + + " on conflict (#{target}) do update set #{others.map { |col| "#{quote(col)} = excluded.#{quote(col)}" }.join(', ')}" + end + + # schema.table -> index schema.table_upsert on table (keys) + def create_upsert_index table_name + *schema, table = table_name.to_s.split('.') + index = [*schema, "#{table}_upsert"].map { |part| quote(part) }.join('.') + db.execute("create unique index if not exists #{index} on #{quote(table)} (#{upsert_keys.map { |k| quote(k) }.join(',')})") end def quote name diff --git a/lib/datapipeline/version.rb b/lib/datapipeline/version.rb index ea97cc7..7669791 100644 --- a/lib/datapipeline/version.rb +++ b/lib/datapipeline/version.rb @@ -1,5 +1,5 @@ class DataPipeline - VERSION = '2.0.0'.freeze + VERSION = '2.1.0'.freeze # kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md class Version diff --git a/test/cli/etl_cli_test.rb b/test/cli/etl_cli_test.rb index add3399..3f407d2 100644 --- a/test/cli/etl_cli_test.rb +++ b/test/cli/etl_cli_test.rb @@ -86,4 +86,30 @@ class EtlCliTest < PipelineTest assert_includes err, 'Usage: etl [run] PIPELINE.rb' end end + + def test_lock_exit_code_3 + with_tmpdir do |dir| + path = script(dir, 'locked.rb', <<~RUBY) + DataPipeline.new do + lock '.pipeline.lock' + read container(:csv, file: 'in.csv') + write container(:csv, file: 'out.csv') + end.run + RUBY + holder = DataPipeline::Lock.new("#{dir}/.script.lock").acquire + _out, err, status = etl(dir, 'run', path, '--lock=.script.lock') + assert_equal 3, status.exitstatus, err + assert_includes err, 'another run is still active' + refute File.exist?("#{dir}/out.csv") + holder.release + + holder = DataPipeline::Lock.new("#{dir}/.pipeline.lock").acquire + _out, err, status = etl(dir, 'run', path, '--lock=.script.lock') + assert_equal 3, status.exitstatus, err + holder.release + + _out, err, status = etl(dir, 'run', path, '--lock=.script.lock') + assert_equal 0, status.exitstatus, err + end + end end diff --git a/test/core/lock_test.rb b/test/core/lock_test.rb new file mode 100644 index 0000000..5d70270 --- /dev/null +++ b/test/core/lock_test.rb @@ -0,0 +1,47 @@ +require_relative '../test_helper' + +class LockTest < PipelineTest + def test_second_lock_on_the_same_file_fails + with_tmpdir do |dir| + first = DataPipeline::Lock.new("#{dir}/sub/.lock").acquire + assert first.locked? + error = assert_raises(DataPipeline::LockError) { DataPipeline::Lock.new("#{dir}/sub/.lock").acquire } + assert_includes error.message, "pid #{Process.pid}" + first.release + DataPipeline::Lock.new("#{dir}/sub/.lock").synchronize { assert true } + end + end + + def test_pipeline_holds_the_lock_during_run_and_releases_it + with_tmpdir do |dir| + file = "#{dir}/.lock" + inside = nil + run_pipeline([{ 'a' => 1 }]) do + lock file + map { |row| inside = (DataPipeline::Lock.new(file).acquire rescue $!); row } + end + assert_kind_of DataPipeline::LockError, inside + DataPipeline::Lock.new(file).synchronize { assert true } # released after run + end + end + + def test_locked_pipeline_aborts_with_lock_error_and_writes_nothing + with_tmpdir do |dir| + holder = DataPipeline::Lock.new("#{dir}/.lock").acquire + target = MemoryStorage.new + finished = nil + assert_raises(DataPipeline::LockError) do + DataPipeline.new do + logger PipelineHelper::QUIET + lock "#{dir}/.lock" + read MemoryStorage.new(rows: [{ 'a' => 1 }]) + write target + on_finish { |stats| finished = stats } + end.run + end + assert_nil target.written + assert_kind_of DataPipeline::LockError, finished.error + holder.release + end + end +end diff --git a/test/plugins/csv_storage_test.rb b/test/plugins/csv_storage_test.rb index 643557b..3cca527 100644 --- a/test/plugins/csv_storage_test.rb +++ b/test/plugins/csv_storage_test.rb @@ -86,4 +86,34 @@ class CsvStorageTest < PipelineTest assert_equal "FIRC|0\nDOHD|AR1|\nDOPS|1.50\nLARC|0000000002\n", File.read("#{dir}/c.ok") end end + + def test_encoding_read_and_write + with_tmpdir do |dir| + DataPipeline::CsvStorage.new(file: "#{dir}/l1.csv", encoding: 'ISO-8859-1').write([{ 'name' => 'Müller', 'ort' => 'Zürich' }]) + assert_equal "name,ort\nM\xFCller,Z\xFCrich\n".b, File.binread("#{dir}/l1.csv") + rows = DataPipeline::CsvStorage.new(file: "#{dir}/l1.csv", encoding: 'ISO-8859-1').read + assert_equal [{ 'name' => 'Müller', 'ort' => 'Zürich' }], rows + assert_equal Encoding::UTF_8, rows.first['name'].encoding + end + end + + def test_encoding_error_or_replace + with_tmpdir do |dir| + assert_raises(Encoding::UndefinedConversionError) do + DataPipeline::CsvStorage.new(file: "#{dir}/a.csv", encoding: 'ISO-8859-1').write([{ 'x' => '5 €' }]) + end + DataPipeline::CsvStorage.new(file: "#{dir}/b.csv", encoding: 'ISO-8859-1', replace: '?').write([{ 'x' => '5 €' }]) + assert_equal "x\n5 ?\n", File.read("#{dir}/b.csv") + DataPipeline::CsvStorage.new(file: "#{dir}/c.csv", encoding: 'Windows-1252').write([{ 'x' => '5 €' }]) + assert_equal "x\n5 \x80\n".b, File.binread("#{dir}/c.csv") + end + end + + def test_bom_written_and_skipped + with_tmpdir do |dir| + DataPipeline::CsvStorage.new(file: "#{dir}/bom.csv", bom: true).write([{ 'a' => '1' }]) + assert_equal "\uFEFFa\n1\n", File.read("#{dir}/bom.csv", encoding: 'utf-8') + assert_equal [{ 'a' => '1' }], DataPipeline::CsvStorage.new(file: "#{dir}/bom.csv").read + end + end end diff --git a/test/plugins/postgres_storage_test.rb b/test/plugins/postgres_storage_test.rb index 2d30226..e210137 100644 --- a/test/plugins/postgres_storage_test.rb +++ b/test/plugins/postgres_storage_test.rb @@ -130,4 +130,84 @@ class PostgresStorageTest < PipelineTest error = assert_raises(PG::ConnectionBad) { DataPipeline::PostgresStorage.new(host: 'localhost', port: 1, database: 'x') } assert_includes error.message, 'Error connection to database:' end + + def test_upsert_updates_existing_rows + with_pg_database do |db| + pg_exec db, 'create table u (belnr text primary key, betrag numeric, text text)' + writer = DataPipeline::PostgresStorage.new(**db, table: 'u', upsert: %w[belnr]) + writer.write([{ 'belnr' => 'A', 'betrag' => 1, 'text' => 'alt' }, { 'belnr' => 'B', 'betrag' => 2, 'text' => 'b' }]) + writer.write([{ 'belnr' => 'A', 'betrag' => 5, 'text' => 'neu' }]) + writer.write_row({ 'belnr' => 'B' }) + writer.write_row({ 'belnr' => 'C', 'betrag' => 3 }) + writer.close + assert_equal [%w[A 5 neu], %w[B 2 b], ['C', '3', nil]], + pg_exec(db, 'select belnr, betrag::int::text as betrag, text from u order by belnr').map(&:values) + end + end + + def test_write_row_in_batches + with_pg_database do |db| + pg_exec db, 'create table b (id int)' + writer = DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: 2) + 3.times { |i| writer.write_row({ 'id' => i }) } + assert_equal '2', pg_exec(db, 'select count(*) from b').first['count'] # third row still pending + writer.finalize + assert_equal '3', pg_exec(db, 'select count(*) from b').first['count'] + end + end + + def test_close_writes_pending_rows + with_pg_database do |db| + pg_exec db, 'create table b (id int)' + writer = DataPipeline::PostgresStorage.new(**db, table: 'b') + writer.write_row({ 'id' => 1 }) + writer.close + assert_equal '1', pg_exec(db, 'select count(*) from b').first['count'] + end + end + + def test_batch_size_all_is_one_transaction_for_the_stream + with_pg_database do |db| + pg_exec db, 'create table b (id int); insert into b values (99)' + aborted = DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: :all, truncate: true) + 2.times { |i| aborted.write_row({ 'id' => i }) } + aborted.close # without finalize: rolled back, truncate too + assert_equal ['99'], pg_exec(db, 'select id from b').map { |r| r['id'] } + + done = DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: :all, truncate: true) + 2.times { |i| done.write_row({ 'id' => i }) } + done.finalize + assert_equal %w[0 1], pg_exec(db, 'select id from b order by id').map { |r| r['id'] } + end + end + + def test_streamed_pipeline_aborted_with_batch_size_all_writes_nothing + with_pg_database do |db| + pg_exec db, 'create table b (id int)' + assert_raises(RuntimeError) do + DataPipeline.new do + logger PipelineHelper::QUIET + streamed + read MemoryStorage.new(rows: Array.new(5) { |i| { 'id' => i } }) + map { |row| row['id'] == 3 ? raise('kaputt') : row } + write DataPipeline::PostgresStorage.new(**db, table: 'b', batch_size: :all) + end.run + end + assert_equal '0', pg_exec(db, 'select count(*) from b').first['count'] + end + end + + def test_streamed_copy_on_the_same_connection_as_the_cursor + with_pg_database do |db| + pg_exec db, 'create table k as select i as id from generate_series(1, 25) i; create table c (id int)' + pg = DataPipeline::PostgresStorage.new(**db, query: 'select id from k', table: 'c', fetch_size: 10, batch_size: 7) + DataPipeline.new do + logger PipelineHelper::QUIET + streamed + read pg + write pg + end.run + assert_equal '25', pg_exec(db, 'select count(*) from c').first['count'] + end + end end diff --git a/test/plugins/sqlite_storage_test.rb b/test/plugins/sqlite_storage_test.rb index 632430a..0df2064 100644 --- a/test/plugins/sqlite_storage_test.rb +++ b/test/plugins/sqlite_storage_test.rb @@ -92,4 +92,15 @@ class SqliteStorageTest < PipelineTest assert_equal 'Deutschland', names['DE']['name'] end end + + def test_upsert_with_created_unique_index + with_tmpdir do |dir| + writer = DataPipeline::SqliteStorage.new(database: "#{dir}/u.sqlite3", table: 'u', create: true, upsert: %w[belnr]) + writer.write([{ 'belnr' => 'A', 'betrag' => 1.0 }, { 'belnr' => 'B', 'betrag' => 2.0 }]) + writer.write([{ 'belnr' => 'A', 'betrag' => 5.0 }]) + writer.write_row({ 'belnr' => 'B' }) + assert_equal [['A', 5.0], ['B', 2.0]], writer.query(query: 'select belnr, betrag from u order by belnr').map(&:values) + writer.close + end + end end