datapipeline 3.1.0: SQLite schreibt im Streaming in Paketen, Lizenz BSD-3-Clause

CHANGELOG.md

@@ -2,6 +2,48 @@
 
 Framework `datapipeline` (Gem). Neueste Version oben.
 
+## 3.1.0 – 2026-10-09
+
+- sqlite: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion;
+  `batch_size: :all` = eine Transaktion für den ganzen Stream, nach einem Abbruch zurückgerollt
+- postgres, sqlite: `query` schreibt noch gesammelte Zeilen von `write_row` vorher
+- Lizenz BSD-3-Clause (`LICENSE`, gemspec)
+
+## 3.0.0 – 2026-10-09
+
+- **Breaking:** `transform` (auch `processing: :prepend`/`:append`), der Parallelmodus, `chained` und `condition:` als
+  String (`eval`) entfernt; alle Schritte laufen verkettet. Umstellung: `map`, `header`, `trailer`,
+  `condition: ->(row) { … }` (README 7.1)
+- `condition:` muss ein Lambda sein, sonst `ArgumentError` beim Aufbau der Pipeline
+- Hinweise `WARN deprecated` und `test/core/compat_test.rb` entfernt
+
+## 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
+- Layout: `format: true` bringt Werte vor Prüfen und Schreiben ins Format ihres Typs (`N` mit Nachkommastellen,
+  kaufmännisch gerundet; `DAT`/`TIM` aus `Date`, `Time` und ISO-Text)
+- neues Plugin `:fixed` (`DataPipeline::FixedStorage`): Festbreiten-Dateien nach Layout schreiben und lesen
+- csv: mit `layout:` und `headers: false` werden Dateien mit Satzarten gelesen, Schlüssel aus den Feldnamen
+- DSL: Kurzschritte `rename`, `select`, `cast` (`:integer`, `:decimal`, `:date`, `:string`) und `default`
+- veraltet, entfällt in 3.0: `transform`, der Parallelmodus und `condition:` als String; je Prozess einmal
+  `WARN deprecated: …` im Log
+- Verbindungen in `config/connections.yml` (mit ERB für ENV): `connection(:name, **opts)` und
+  `container(:typ, connection: :name)`
+- `totals 'spalte'` (exakte Summen über Quelle und Ergebnis, `stats.total`) und `reconcile { |stats| … }`: Abgleich
+  vor dem Schreiben, eine Meldung bricht mit `DataPipeline::ReconcileError` ab
+- Laufhistorie: `history 'datei'` bzw. `etl --history=datei`, eine JSON-Zeile je Lauf; `DataPipeline.history` liest
+  sie; `name` des Laufs (über `etl` der Skriptname), `Stats#to_h`
+- README neu gegliedert (Überblick, Beispiele, Pipelines schreiben, Plugins, Layouts, Betrieb, Umstieg,
+  Entwicklung) mit Inhaltsverzeichnis, Fehlersuche und Umstiegshinweisen; neues Beispiel
+  `examples/buchungen_export.rb` für einen produktiven Lauf
+
 ## 2.0.0 – 2026-10-08
 
 - Gem `datapipeline`: `datapipeline.gemspec`, Executable `etl` (exe/etl), Code unter `lib/datapipeline/`

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    datapipeline (2.0.0)
+    datapipeline (3.1.0)
       csv (~> 3.3)
       logger (~> 1.6)
 

LICENSE

@@ -0,0 +1,28 @@
+BSD 3-Clause License
+
+Copyright (c) 2026, Karl-Heinz Wild
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its
+   contributors may be used to endorse or promote products derived from
+   this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

datapipeline.gemspec

@@ -12,13 +12,13 @@ Gem::Specification.new do |spec|
     Fehlerdateien, Dry-Run, inkrementellen Läufen und dem Kommando etl mit Exit-Codes.
   TEXT
   spec.homepage      = 'https://git.wicom.li/wkh/datapipeline'
-  spec.license       = 'Nonstandard'
+  spec.license       = 'BSD-3-Clause'
   spec.required_ruby_version = '>= 3.4'
 
   spec.metadata['changelog_uri'] = "#{spec.homepage}/src/branch/main/CHANGELOG.md"
   spec.metadata['rubygems_mfa_required'] = 'true'
 
-  spec.files         = Dir['lib/**/*.rb', 'exe/*', 'README.md', 'CHANGELOG.md']
+  spec.files         = Dir['lib/**/*.rb', 'exe/*', 'README.md', 'CHANGELOG.md', 'LICENSE']
   spec.bindir        = 'exe'
   spec.executables   = ['etl']
   spec.require_paths = ['lib']

examples/buchungen_export.rb

@@ -0,0 +1,59 @@
+#
+# Produktiver Lauf: Buchungen prüfen, Summe und Saldo abgleichen, wiederholbar in die Buchhaltungs-Datenbank
+# und als CSV in Windows-1252 für das Zielsystem schreiben; mit Sperre, Laufhistorie und Slack bei Problemen
+#
+#   bin/etl run examples/buchungen_export.rb [--out=tmp] [--dry-run]
+#
+require 'fileutils'
+require 'datapipeline'
+
+DATA = File.expand_path('data', __dir__)
+out = DataPipeline.params.fetch(:out, 'tmp')
+FileUtils.mkdir_p(out)
+
+# Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml)
+DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__)
+
+# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf
+slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']
+
+DataPipeline.new do
+  name 'buchungen_export'
+  lock "#{out}/.buchungen.lock"              # läuft schon ein Export, bricht dieser ab (Exit-Code 3)
+  history "#{out}/runs.ndjson"               # eine JSON-Zeile je Lauf, auch nach Abbruch
+
+  read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' })
+
+  # 1. prüfen: ohne on_error bricht ein Fehler den Lauf ab, es entsteht keine halbe Buchung
+  validate 'belnr' => { required: true, format: /\A\d{10}\z/ },
+           'konto' => { required: true, format: /\A\d+\z/ },
+           'betrag' => { required: true, type: :decimal }
+
+  # 2. umbauen mit Kurzschritten und einem map
+  default 'text' => 'ohne Text'
+  rename 'text' => 'buchungstext'
+  cast 'buzei' => :integer, 'datum' => :date, 'betrag' => :decimal
+  map do |row|
+    row.merge('shkzg' => row['betrag'].negative? ? 'H' : 'S', 'dmbtr' => format('%.2f', row['betrag'].abs))
+  end
+  select %w[belnr buzei datum konto shkzg dmbtr betrag buchungstext]
+
+  # 3. abgleichen, bevor etwas geschrieben wird
+  totals 'betrag'
+  reconcile do |stats|
+    quelle, ergebnis = stats.total('betrag', :read), stats.total('betrag')
+    next "Summe geändert: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ergebnis)}" if quelle != ergebnis
+
+    "Soll und Haben ungleich, Saldo #{DataPipeline.decimal(ergebnis)}" unless ergebnis.zero?
+  end
+
+  # 4. schreiben: Upsert macht den Lauf wiederholbar; die CSV-Datei im Zeichensatz des Zielsystems
+  write connection(:buchhaltung)
+  write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' },
+                        columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext])
+
+  on_finish do |stats|
+    puts "#{stats.name}: #{stats}"
+    slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok?
+  end
+end.run

examples/config/connections.yml

@@ -0,0 +1,8 @@
+# Verbindungen der Beispiele. Die Datei wird mit ERB gelesen: Zugangsdaten und Pfade kommen aus der Umgebung,
+# bei Postgres z. B.  password: <%= ENV['PDS_PASSWORD'] %>
+buchhaltung:
+  type: sqlite
+  database: <%= ENV.fetch('BUCHHALTUNG_DB', 'tmp/buchhaltung.sqlite3') %>
+  table: buchungen
+  create: true
+  upsert: [belnr, buzei]

examples/data/buchungen.csv

@@ -0,0 +1,8 @@
+belnr;buzei;datum;konto;betrag;text
+5100000001;1;2026-10-01;1020;1250.00;Zahlung Müller GmbH
+5100000001;2;2026-10-01;1100;-1250.00;Zahlung Müller GmbH
+5100000002;1;2026-10-02;6500;380.50;Büromaterial
+5100000002;2;2026-10-02;1170;29.30;Vorsteuer
+5100000002;3;2026-10-02;2000;-409.80;Papeterie Zürcher AG
+5100000003;1;2026-10-03;3200;99.95;
+5100000003;2;2026-10-03;1020;-99.95;Gutschrift

exe/etl

@@ -9,9 +9,11 @@
 #   --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
+#   --history=FILE            one json line per pipeline run (DataPipeline.history)
 #   --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 +33,8 @@ 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 }
+  opts.on('--history=FILE', String) { |v| options[:settings][:history] = v }
 end
 
 # own options for the parser, everything else (also --name=value of the script) goes to the script
@@ -38,9 +42,9 @@ 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|history)(=|\z)/)
     own << arg
-    own << argv.shift if %w[--param --limit --from --till].include?(arg)
+    own << argv.shift if %w[--param --limit --from --till --lock --history].include?(arg)
   else
     args << arg
   end
@@ -52,7 +56,14 @@ DataPipeline.params.merge!(DataPipeline.parse_params(args)) { |_key, own_value,
 abort parser.help if script.nil?
 abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script)
 
-DataPipeline.settings(options[:settings])
+DataPipeline.settings(options[:settings].merge(name: File.basename(script, '.rb')))
+
+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 +72,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

lib/datapipeline.rb

@@ -6,6 +6,8 @@ require_relative 'datapipeline/storage/storage_plugin'
 require_relative 'datapipeline/validation'
 require_relative 'datapipeline/layout'
 require_relative 'datapipeline/watermark'
+require_relative 'datapipeline/lock'
+require_relative 'datapipeline/connections'
 
 class DataPipeline
   # plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel),
@@ -16,6 +18,7 @@ class DataPipeline
   autoload :SqliteStorage,   File.expand_path('datapipeline/storage/sqlite', __dir__)
   autoload :ExcelStorage,    File.expand_path('datapipeline/storage/excel', __dir__)
   autoload :RestStorage,     File.expand_path('datapipeline/storage/rest', __dir__)
+  autoload :FixedStorage,    File.expand_path('datapipeline/storage/fixed', __dir__)
   autoload :Slack,           File.expand_path('datapipeline/notification/slack', __dir__)
 
   # passed to header, trailer, map and filter
@@ -28,36 +31,56 @@ class DataPipeline
   #   read: source rows, written: body rows, output: rows to the target incl. header/trailer,
   #   rejected: rows dropped by on_error :skip/:collect, errors: first 100 of them,
   #   warnings: rows that failed a validate step with warn: true (kept), error: exception of an aborted run
-  Stats = Struct.new(:version, :read, :written, :output, :rejected, :errors, :warnings, :started_at, :finished_at,
-                     :dry_run, :error, keyword_init: true) do
+  #   totals: { 'column' => { read: Rational, written: Rational } } of the columns named by totals
+  Stats = Struct.new(:name, :version, :read, :written, :output, :rejected, :errors, :warnings, :started_at,
+                     :finished_at, :dry_run, :error, :totals, keyword_init: true) do
     def duration = finished_at && started_at ? (finished_at - started_at).round(3) : nil
     def ok? = error.nil? && rejected.zero?
 
+    # stats.total('dmbtr') -> sum after all steps, stats.total('dmbtr', :read) -> sum of the source rows
+    def total column, side = :written
+      totals.to_h.dig(column.to_s, side)
+    end
+
     def to_s
       "read=#{read} written=#{written} rejected=#{rejected}#{warnings.to_i.positive? ? " warnings=#{warnings}" : ''}" \
+        "#{totals.to_h.map { |col, sum| " #{col}=#{DataPipeline.decimal(sum[:read])}/#{DataPipeline.decimal(sum[:written])}" }.join} " \
         "#{duration}s#{dry_run ? ' (dry run)' : ''}" \
         "#{error ? " ABORTED: #{error.class}: #{error.message}" : ''}"
     end
+
+    # one line of the run history
+    def to_h
+      { name:, version:, started_at: started_at&.iso8601(3), finished_at: finished_at&.iso8601(3), duration:,
+        read:, written:, output:, rejected:, warnings:, dry_run: dry_run ? true : false, ok: ok?,
+        error: error && "#{error.class}: #{error.message}",
+        totals: totals.to_h.transform_values { |sum| sum.transform_values { |v| DataPipeline.decimal(v) } } }
+    end
   end
 
   # stops reading at the limit, not an error
   class LimitReached < StandardError; end
 
-  EDGE_BEFORE = %i[prepend header].freeze
-  EDGE_AFTER  = %i[append trailer].freeze
-  CHAINING    = %i[map filter validate].freeze
+  # raised when a reconcile check fails
+  class ReconcileError < StandardError; end
+
   ON_ERROR    = %i[abort skip collect].freeze
   MAX_ERRORS  = 100
 
+  # value conversions for cast, empty values become nil
+  CASTS = Validation::TYPES.merge(string: ->(v) { v.to_s }).freeze
+
   DEFAULT_SETTINGS = {
     version: nil,
     streamed: false,
-    chained: false,
     verbose: false,
     dry_run: false,
     limit: nil,
     preview: 10,
     on_error: :abort,
+    lock: nil,
+    name: nil,
+    history: nil,
     log_level: Logger::WARN
   }.freeze
 
@@ -129,10 +152,33 @@ class DataPipeline
     @runs ||= []
   end
 
+  # see DataPipeline::Connections
+  def self.connections_file = Connections.file
+  def self.connections_file=(path)
+    Connections.file = path
+  end
+
+  # entries of a run history file (see history), oldest first; name: only the runs of this pipeline
+  def self.history file, name: nil
+    return [] unless File.exist?(file)
+
+    entries = File.foreach(file).filter_map { |line| JSON.parse(line) unless line.strip.empty? }
+    name ? entries.select { |entry| entry['name'] == name.to_s } : entries
+  end
+
+  # exact decimal text of a sum: 1234.5 -> '1234.5', 1/3 -> '0.333333'
+  def self.decimal value
+    return value.to_s unless value.is_a?(Rational)
+    return value.to_i.to_s if value.denominator == 1
+
+    places = (1..10).find { |n| (value * 10**n).denominator == 1 } || 6
+    format("%.#{places}f", value)
+  end
+
   # DataPipeline.new(verbose: true) { ... } overrides the defaults for this pipeline only
   def initialize(**settings, &block)
     @settings = DataPipeline.settings.merge(settings)
-    @transformations = []
+    @steps = []
     @targets = []
     @on_finish = []
     @logger = Logger.new($stderr, level: @settings[:log_level], progname: 'etl')
@@ -160,11 +206,6 @@ class DataPipeline
     @settings[:streamed] = flag
   end
 
-  # each body step gets the output of the previous step instead of the source rows
-  def chained flag = true
-    @settings[:chained] = flag
-  end
-
   # run everything, write nothing (target, rejects, watermark); prints the first rows
   def dry_run flag = true
     @settings[:dry_run] = flag
@@ -204,6 +245,48 @@ 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
+
+  # name of the pipeline in the stats and the run history; etl sets the script name as default
+  def name value
+    @settings[:name] = value
+  end
+
+  # history 'data/runs.ndjson' -> one json line per run (also aborted and dry runs), see DataPipeline.history
+  def history file
+    @settings[:history] = file
+  end
+
+  #
+  # totals 'dmbtr', 'menge'   -> sums of these columns over the source rows and over the rows after all steps
+  #                              (stats.totals, stats.total('dmbtr')); empty and non numeric values are left out
+  #
+  def totals *columns
+    @totals = columns.flatten.map(&:to_s)
+  end
+
+  #
+  # reconcile { |stats| 'Summe weicht ab' if stats.total('dmbtr', :read) != stats.total('dmbtr') }
+  #
+  # runs after all steps, before writing (streaming: after the last row); a message (or array of messages)
+  # aborts the run with ReconcileError, nil means ok
+  #
+  def reconcile(&block)
+    raise ArgumentError, 'reconcile: block missing' unless block
+
+    (@reconcile ||= []) << block
+  end
+
+  # connection(:pds, query: '...') -> container of the type in the connections file, options of the call win
+  def connection name, **options, &block
+    settings = Connections.fetch(name)
+    type = settings.fetch(:type) { raise ArgumentError, "connection #{name.inspect}: type missing" }
+    container(type, **settings.except(:type), **options, &block)
+  end
+
   # see DataPipeline::Watermark
   def watermark name, column:, file:, initial: nil
     @watermark = Watermark.new(name, column:, file:, initial:)
@@ -211,34 +294,75 @@ class DataPipeline
 
   # only a missing plugin is reported as unsupported, errors while loading it
   # (missing gem) or from the plugin itself (connection, options) are passed through
+  #
+  # container :postgres, connection: :pds, table: 'x' -> options from the connections file, the call wins
+  #
   def container type, **options, &block
     klass = DataPipeline.plugin(type)
     raise "Unsupported storage type or class missing: #{type.to_s.capitalize}Storage" unless klass
 
+    if (name = options.delete(:connection))
+      settings = Connections.fetch(name)
+      if settings[:type] && settings[:type].to_s != type.to_s
+        raise ArgumentError, "connection #{name.inspect} is #{settings[:type]}, not #{type}"
+      end
+
+      options = settings.except(:type).merge(options)
+    end
     klass.new(**options, &block)
   end
 
-  #
-  # transform(options) { |row, index| ... }
-  #   processing: :prepend | :normal (default) | :append
-  #   condition:  ruby expression as string (row is visible) or lambda { |row| ... }
-  #
-  def transform(options = {}, &block)
-    kind = options.fetch(:processing, :normal)
-    kind = :normal unless %i[prepend append].include?(kind)
-    add_step kind, options, &block
-  end
-
-  # map { |row, ctx| ... }    -> hash, array of hashes or nil; turns chaining on
+  # map { |row, ctx| ... }    -> hash, array of hashes or nil
+  # condition: ->(row) { ... } on map and filter: the step only for matching rows, the others are dropped
   def map(options = {}, &block)
     add_step :map, options, &block
   end
 
-  # filter { |row, ctx| ... } -> keeps the row when the block is truthy; turns chaining on
+  # filter { |row, ctx| ... } -> keeps the row when the block is truthy
   def filter(options = {}, &block)
     add_step :filter, options, &block
   end
 
+  #
+  # short steps, each one a map step
+  #
+  #   rename 'kunnr' => 'kundennr'               -> new key at the same position
+  #   select %w[belnr betrag]                    -> only these keys in this order, missing ones nil
+  #   cast 'betrag' => :decimal                  -> :integer, :decimal (Float), :date, :string; empty -> nil
+  #   default 'land' => 'CH'                     -> for nil or blank values; a lambda gets the row
+  #
+  def rename mapping
+    mapping = mapping.to_h { |from, to| [from.to_s, to.to_s] }
+    add_step(:map, { name: 'rename' }) { |row| row.to_h { |key, value| [mapping.fetch(key.to_s, key), value] } }
+  end
+
+  def select *keys
+    keys = keys.flatten.map(&:to_s)
+    add_step(:map, { name: 'select' }) { |row| keys.to_h { |key| [key, value_of(row, key)] } }
+  end
+
+  def cast types
+    types = types.to_h { |key, type| [key.to_s, type.to_sym] }
+    unknown = types.values - CASTS.keys
+    raise ArgumentError, "cast: #{unknown.join(', ')} unknown, use #{CASTS.keys.join(', ')}" if unknown.any?
+
+    add_step(:map, { name: 'cast' }) do |row|
+      types.each_with_object(row.dup) do |(key, type), out|
+        value = value_of(row, key)
+        out[key] = blank?(value) ? nil : cast_value(key, value, type)
+      end
+    end
+  end
+
+  def default values
+    values = values.to_h { |key, value| [key.to_s, value] }
+    add_step(:map, { name: 'default' }) do |row|
+      values.each_with_object(row.dup) do |(key, value), out|
+        out[key] = value.respond_to?(:call) ? value.call(row) : value if blank?(value_of(row, key))
+      end
+    end
+  end
+
   # header { |ctx| ... } -> rows in front of the body, called after the body with full counters
   def header(&block)
     add_step :header, {}, &block
@@ -254,7 +378,7 @@ class DataPipeline
   # validate layout                                   -> fields of a DataPipeline::Layout
   # validate { |row| 'soll ungleich haben' if ... }   -> nil, message or array of messages
   #
-  # a failing row raises ValidationError, handled by on_error; turns chaining on
+  # a failing row raises ValidationError, handled by on_error
   # warn: true      -> only a warning in the log and stats.warnings, the row is kept
   # ignore: %w[A B] -> messages for these fields are left out (known gaps)
   #
@@ -262,7 +386,7 @@ class DataPipeline
     warn = options.delete(:warn) || false
     ignore = Array(options.delete(:ignore)).map(&:to_s)
     rules ||= options # validate 'konto' => { ... } arrives as keywords
-    @transformations << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
+    @steps << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
   end
 
   def read storage, options = {}
@@ -285,7 +409,7 @@ class DataPipeline
   end
 
   #
-  # loads a table once, for lookups in map/transform instead of one query per row
+  # loads a table once, for lookups in map instead of one query per row
   #
   #   lookup pg, query: 'select ...', params: [...], key: 'rechnung_dbid'  -> { key => [rows] }
   #   lookup pg, query: 'select ...', key: 'id', unique: true              -> { key => row }
@@ -298,16 +422,18 @@ class DataPipeline
   end
 
   def run
-    @stats = Stats.new(version: @settings[:version], read: 0, written: 0, output: 0, rejected: 0, errors: [], warnings: 0,
-                       started_at: Time.now, dry_run: @settings[:dry_run])
+    @stats = Stats.new(name: @settings[:name], version: @settings[:version], read: 0, written: 0, output: 0, rejected: 0,
+                       errors: [], warnings: 0, started_at: Time.now, dry_run: @settings[:dry_run],
+                       totals: @totals&.to_h { |col| [col, { read: 0r, written: 0r }] })
     @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]) }
-    after  = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) }
-    body   = @transformations - before - after
+    before = @steps.select { |t| t[:kind] == :header }
+    after  = @steps.select { |t| t[:kind] == :trailer }
+    body   = @steps - before - after
 
     @settings[:streamed] ? run_streamed(before, body, after) : run_batch(before, body, after)
 
@@ -325,14 +451,25 @@ class DataPipeline
 
   def add_step kind, options, &block
     raise ArgumentError, "#{kind}: block missing" unless block
-
-    @transformations << { kind:, block:, options: }
+    if options.key?(:condition) && !options[:condition].respond_to?(:call)
+      raise ArgumentError, "#{kind}: condition must be a lambda ->(row) { ... }"
     end
 
-  # without any transformation the source is copied unchanged
-  def chained?
-    @settings[:chained] || @transformations.empty? ||
-      @transformations.any? { |t| CHAINING.include?(t[:kind]) }
+    @steps << { kind:, block:, options: }
+  end
+
+  def value_of row, key
+    row.key?(key) ? row[key] : row[key.to_sym]
+  end
+
+  def blank? value
+    value.nil? || (value.is_a?(String) && value.strip.empty?)
+  end
+
+  def cast_value key, value, type
+    CASTS.fetch(type).call(value)
+  rescue ArgumentError, TypeError, Date::Error
+    raise ArgumentError, "#{key}: #{value.inspect} is not #{type}"
   end
 
   def write_row row
@@ -342,9 +479,11 @@ 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}"
+    append_history
     @on_finish.each do |hook|
       hook.call(@stats)
     rescue StandardError => e
@@ -352,6 +491,41 @@ class DataPipeline
     end
   end
 
+  # one json line per run, appended under an flock (parallel runs of other pipelines)
+  def append_history
+    return unless @settings[:history]
+
+    FileUtils.mkdir_p(File.dirname(@settings[:history]))
+    File.open(@settings[:history], 'a') do |file|
+      file.flock(File::LOCK_EX)
+      file.puts JSON.generate(@stats.to_h)
+    end
+  rescue StandardError => e
+    @logger.error "history: #{e.class}: #{e.message}"
+  end
+
+  def add_total side, row
+    @stats.totals&.each do |col, sum|
+      value = row.key?(col) ? row[col] : row[col.to_sym]
+      number = to_number(value)
+      sum[side] += number if number
+    end
+  end
+
+  def to_number value
+    case value
+    when Float then value.finite? ? value.to_s.to_r : nil
+    when Numeric then value.to_r
+    when String then value.strip.match?(Layout::NUMBER) ? value.strip.to_r : nil
+    end
+  end
+
+  # messages of the reconcile blocks abort the run
+  def check_reconcile
+    messages = Array(@reconcile).flat_map { |block| Array(block.call(@stats)).compact }
+    raise ReconcileError, messages.join('; ') if messages.any?
+  end
+
   def source_rows
     data = @source.read
     data = data.first(@settings[:limit]) if @settings[:limit]
@@ -376,32 +550,24 @@ class DataPipeline
     data = source_rows
     @stats.read = data.length
     data.each { |row| @watermark.track(row) } if @watermark
+    data.each { |row| add_total(:read, row) } if @totals
     dump 'Source#data', data
 
-    # prepend keeps its old timing: called before the body
-    results = {}.compare_by_identity
-    before.select { |t| t[:kind] == :prepend }.each { |t| results[t] = edge(t, nil, 0) }
-
     rows = []
-    if chained?
     counters = Array.new(body.size, 0)
     data.each { |row| push(row, body, counters, data.length) { |out| rows << out } }
-    else
-      body.each { |step| rows.concat apply(step, data) }
-    end
     dump 'Transformation#data', rows
-
-    ctx = Context.new(read: data.length, written: rows.length)
-    before.select { |t| t[:kind] == :header }.each { |t| results[t] = edge(t, ctx, 0) }
-
-    # parallel mode keeps the old count of source rows for append
-    count = chained? ? rows.length : data.length
-    new_data = []
-    before.each { |t| add_rows new_data, results[t] }
-    new_data.concat rows
-    after.each { |t| add_rows new_data, edge(t, ctx, count) }
-
     @stats.written = rows.length
+    rows.each { |row| add_total(:written, row) } if @totals
+    check_reconcile
+
+    # header is computed after the body, so it knows the counters
+    ctx = Context.new(read: data.length, written: rows.length)
+    new_data = []
+    before.each { |t| add_rows new_data, t[:block].call(ctx) }
+    new_data.concat rows
+    after.each { |t| add_rows new_data, t[:block].call(ctx) }
+
     dump 'Target#data', new_data
     output(new_data) do |all|
       @targets.each { |t| t[:storage].write(t[:only] ? all.select(&t[:only]) : all) }
@@ -410,13 +576,9 @@ class DataPipeline
 
   # row by row from source.read { } to target.write_row, header only knows nothing yet
   def run_streamed before, body, after
-    if !chained? && body.size > 1
-      raise ArgumentError, 'streamed: several transformations need chained (or map/filter)'
-    end
-
     emit = ->(result) { output(add_rows([], result)) { |rows| rows.each { |row| write_row(row) } } }
 
-    before.each { |t| emit.call edge(t, Context.new, 0) }
+    before.each { |t| emit.call t[:block].call(Context.new) }
 
     read = written = 0
     counters = Array.new(body.size, 0)
@@ -426,8 +588,10 @@ class DataPipeline
 
         read += 1
         @watermark&.track(row)
+        add_total(:read, row) if @totals
         push(row, body, counters, nil) do |out|
           written += 1
+          add_total(:written, out) if @totals
           emit.call out
         end
       end
@@ -436,9 +600,10 @@ class DataPipeline
     end
     @stats.read = read
     @stats.written = written
+    check_reconcile
 
     ctx = Context.new(read:, written:)
-    after.each { |t| emit.call edge(t, ctx, chained? ? written : read) }
+    after.each { |t| emit.call t[:block].call(ctx) }
     @targets.each { |t| t[:storage].finalize }
   end
 
@@ -453,11 +618,6 @@ class DataPipeline
     end
   end
 
-  # parallel mode: every step reads its own copy of the source rows
-  def apply step, rows
-    rows.each_with_index.flat_map { |row, index| safe_step_rows(step, row.dup, index, rows.length) }
-  end
-
   # errors of a step for one row go to on_error
   def safe_step_rows step, row, index, read
     step_rows(step, row, index, read)
@@ -479,7 +639,8 @@ class DataPipeline
   end
 
   def step_name step
-    "#{step[:kind]}##{@transformations.index(step) + 1}"
+    name = step[:options].is_a?(Hash) && step[:options][:name] || step[:kind]
+    "#{name}##{@steps.index(step) + 1}"
   end
 
   # rows produced by one step for one input row
@@ -507,26 +668,13 @@ class DataPipeline
       step[:block].call(row, Context.new(read:, index:)) ? [row] : []
     when :map
       add_rows [], step[:block].call(row, Context.new(read:, index:))
-    else
-      add_rows [], step[:block].call(row, index)
     end
   end
 
-  def edge step, ctx, count
-    case step[:kind]
-    when :prepend then step[:block].call(nil, 0)
-    when :append  then step[:block].call(nil, count)
-    else step[:block].call(ctx)
-    end
-  end
-
-  # string conditions are evaluated with the local variable row
+  # condition: ->(row) { ... } on map and filter
   def condition? step, row
-    condition = step[:options].fetch(:condition, true)
-    return true if condition == true
-    return condition.call(row) if condition.respond_to?(:call)
-
-    eval(condition)
+    condition = step[:options][:condition]
+    condition.nil? || condition.call(row)
   end
 
   # a hash is one row, an array are several rows, nil is no row

lib/datapipeline/connections.rb

@@ -0,0 +1,59 @@
+require 'erb'
+require 'yaml'
+
+class DataPipeline
+  #
+  # connections in one yaml file instead of credentials in every script
+  #
+  #   # config/connections.yml (ERB allowed, secrets from ENV)
+  #   pds:
+  #     type: postgres
+  #     host: <%= ENV.fetch('PDS_HOST', 'localhost') %>
+  #     database: pds
+  #     username: etl
+  #     password: <%= ENV['PDS_PASSWORD'] %>
+  #
+  #   read connection(:pds, query: 'select ...')          -> container :postgres with these options
+  #   container :postgres, connection: :pds, table: 'x'   -> same, options of the call win
+  #
+  # file: DataPipeline.connections_file = '...', else ENV['DATAPIPELINE_CONNECTIONS'], else config/connections.yml
+  #
+  module Connections
+    DEFAULT_FILE = 'config/connections.yml'.freeze
+
+    module_function
+
+    def file
+      @file || ENV['DATAPIPELINE_CONNECTIONS'] || DEFAULT_FILE
+    end
+
+    def file= path
+      @file = path
+      @all = nil
+    end
+
+    # options of one connection with symbol keys, type included
+    def fetch name
+      all.fetch(name.to_s) do
+        raise ArgumentError, "connection #{name.inspect} not in #{file} (#{all.keys.join(', ')})"
+      end
+    end
+
+    def all
+      @all = nil if @loaded_from != file
+      @loaded_from = file
+      @all ||= begin
+        raise ArgumentError, "connections file #{file} missing" unless File.exist?(file)
+
+        data = YAML.safe_load(ERB.new(File.read(file)).result, aliases: true) || {}
+        raise ArgumentError, "#{file}: a mapping of connection names expected" unless data.is_a?(Hash)
+
+        data.to_h { |name, options| [name.to_s, (options || {}).to_h { |key, value| [key.to_sym, value] }] }
+      end
+    end
+
+    def reset!
+      @all = nil
+    end
+  end
+end

lib/datapipeline/layout.rb

@@ -18,8 +18,15 @@ class DataPipeline
   # keys of a row are matched without regard to case, string or symbol.
   #
   #   write container(:csv, file: ..., layout: layout)   -> values in field order, unknown keys raise
+  #   write container(:fixed, file: ..., layout: layout) -> fixed width lines (see FixedStorage)
+  #   read container(:csv, ..., layout: layout)          -> rows with the field names as keys
   #   validate layout                                    -> required, length, format per field
   #
+  # format: true converts values to the text format of the field type, before checking and writing:
+  #   N   numbers with the decimals of the length (13.2 -> 2), rounded half up; numeric texts too
+  #   DAT Date, Time or 'YYYY-MM-DD' -> YYYYMMDD
+  #   TIM Time or 'HH:MM:SS'         -> HHMMSS
+  #
   class Layout
     Field = Struct.new(:name, :type, :length, :required) do
       def key = name.downcase
@@ -30,13 +37,30 @@ class DataPipeline
         after ||= 0
         [before - after, after]
       end
+
+      # characters in a fixed width file: N 13.2 -> 14 (11 digits, point, 2 decimals), DAT 8, TIM 6
+      def width
+        case type
+        when 'DAT' then 8
+        when 'TIM' then 6
+        when 'N'
+          return unless length
+
+          before, after = digits
+          before + after + (after.positive? ? 1 : 0)
+        else length&.to_i
         end
+      end
+    end
+
+    NUMBER = /\A[+-]?\d+(\.\d+)?\z/
 
     attr_reader :records
 
-    def initialize type_field: nil, strict: true, &block
+    def initialize type_field: nil, strict: true, format: false, &block
       @type_field = type_field&.to_s&.downcase
       @strict = strict
+      @format = format
       @records = {}
       instance_eval(&block) if block
     end
@@ -56,23 +80,148 @@ class DataPipeline
     def values row
       fields = fields_for(row)
       normalized = normalize(row)
-      if @strict
-        unknown = normalized.keys - fields.map(&:key)
-        raise LayoutError, "#{record_type(row)}: fields not in layout: #{unknown.join(', ')}" if unknown.any?
-      end
-      fields.map { |field| normalized[field.key] }
+      unknown!(row, fields, normalized)
+      fields.map { |field| output(field, normalized[field.key]) }
     end
 
     def errors row
       fields = fields_for(row)
       normalized = normalize(row)
-      fields.flat_map { |field| check(field, normalized[field.key]) }
+      fields.flat_map { |field| check(field, output(field, normalized[field.key])) }
     rescue LayoutError => e
       [e.message]
     end
 
+    # value as text in the format of the field type (see format: true)
+    def format_value field, value
+      case value
+      when nil then nil
+      when Numeric then field.type == 'N' ? format_number(field, value) : value.to_s
+      when Date, Time then format_time(field, value)
+      else format_text(field, value.to_s)
+      end
+    end
+
+    # one line of a fixed width file: text left aligned, numbers right aligned with zeros, empty fields blank
+    def line row
+      fields = fields_for(row)
+      normalized = normalize(row)
+      unknown!(row, fields, normalized)
+      fields.map { |field| fixed(field, format_value(field, normalized[field.key])) }.join
+    end
+
+    # values of a line in field order (csv without header) -> row with the field names as keys
+    def parse values
+      type, fields = find_record { |fields| values[type_index(fields)].to_s.strip }
+      raise LayoutError, "record type #{values[0].inspect} not in layout (#{@records.keys.join(', ')})" unless fields
+      raise LayoutError, "#{type}: #{values.size} values, layout has #{fields.size} fields" if @strict && values.size > fields.size
+
+      fields.each_with_index.to_h { |field, i| [field.key, values[i]] }
+    end
+
+    # a line of a fixed width file -> row with the field names as keys, values without padding, blank -> nil
+    def parse_line line
+      type, fields = find_record { |fields| slices(line, fields)[type_index(fields)].to_s.strip }
+      raise LayoutError, "record type of #{line[0, 20].inspect} not in layout (#{@records.keys.join(', ')})" unless fields
+
+      fields.zip(slices(line, fields)).to_h do |field, text|
+        text = field.type == 'AN' ? text.to_s.rstrip : text.to_s.strip
+        [field.key, text.empty? ? nil : text]
+      end
+    end
+
+    # fixed width needs a length for every AN and N field
+    def check_widths!
+      missing = @records.flat_map { |type, fields| fields.reject(&:width).map { |f| "#{type}.#{f.name}" } }
+      raise LayoutError, "fixed width needs a length for: #{missing.join(', ')}" if missing.any?
+
+      self
+    end
+
     private
 
+    def output field, value
+      @format ? format_value(field, value) : value
+    end
+
+    def unknown! row, fields, normalized
+      return unless @strict
+
+      unknown = normalized.keys - fields.map(&:key)
+      raise LayoutError, "#{record_type(row)}: fields not in layout: #{unknown.join(', ')}" if unknown.any?
+    end
+
+    def find_record
+      @records.find { |type, fields| yield(fields) == type }
+    end
+
+    def type_index fields
+      @type_field ? fields.index { |f| f.key == @type_field } || 0 : 0
+    end
+
+    def slices line, fields
+      offset = 0
+      fields.map do |field|
+        text = line[offset, field.width]
+        offset += field.width
+        text
+      end
+    end
+
+    # 12.345 -> '12.35' for N 11.2; floats by their shortest text, so 12.345 is not 12.3449999...
+    def format_number field, value
+      _before, after = field.length ? field.digits : [nil, 0]
+      number = value.is_a?(Float) ? value.to_s.to_r : value.to_r
+      number = number.round(after, half: :up)
+      sign = number.negative? ? '-' : ''
+      int, frac = number.abs.divmod(1)
+      return "#{sign}#{int}" if after.zero?
+
+      "#{sign}#{int}.#{(frac * 10**after).round.to_s.rjust(after, '0')}"
+    end
+
+    def format_time field, value
+      case field.type
+      when 'DAT' then value.strftime('%Y%m%d')
+      when 'TIM' then value.respond_to?(:hour) ? value.strftime('%H%M%S') : value.to_s
+      else value.respond_to?(:hour) ? value.iso8601 : value.to_s
+      end
+    end
+
+    # numeric texts get the decimals of the field, ISO dates and times lose their separators
+    def format_text field, text
+      case field.type
+      when 'N'
+        _before, after = field.length ? field.digits : [nil, 0]
+        return text unless text.match?(NUMBER) && after.positive?
+
+        decimals = text.split('.')[1].to_s
+        decimals.length > after ? text : "#{text.split('.')[0]}.#{decimals.ljust(after, '0')}"
+      when 'DAT' then text.match?(/\A\d{4}-\d{2}-\d{2}\z/) ? text.delete('-') : text
+      when 'TIM' then text.match?(/\A\d{2}:\d{2}:\d{2}\z/) ? text.delete(':') : text
+      else text
+      end
+    end
+
+    def fixed field, text
+      width = field.width
+      return ' ' * width if text.nil? || text.empty?
+      raise LayoutError, "#{field.name}: #{text.inspect} longer than #{width}" if text.length > width
+
+      case field.type
+      when 'N' then raise LayoutError, "#{field.name}: #{text.inspect} is not a number" unless text.match?(NUMBER)
+      when 'DAT', 'TIM' then raise LayoutError, "#{field.name}: #{text.inspect} needs #{width} digits" unless text.match?(/\A\d{#{width}}\z/)
+      else return text.ljust(width)
+      end
+      return text if field.type != 'N'
+
+      sign = text.start_with?('-') ? '-' : ''
+      digits = text.delete('+-')
+      raise LayoutError, "#{field.name}: #{text.inspect} longer than #{width}" if sign.length + digits.length > width
+
+      sign + digits.rjust(width - sign.length, '0')
+    end
+
     def record_type row
       @type_field ? normalize(row)[@type_field] : row.values.first
     end
@@ -95,7 +244,7 @@ class DataPipeline
     end
 
     def check_number field, text
-      return ["#{field.name}: not a number"] unless text.match?(/\A[+-]?\d+(\.\d+)?\z/)
+      return ["#{field.name}: not a number"] unless text.match?(NUMBER)
       return [] unless field.length
 
       before, after = field.digits

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

lib/datapipeline/storage/csv.rb

@@ -10,7 +10,12 @@ class DataPipeline
   #   columns:  fixed column order for writing (and keys for reading without headers)
   #             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
+  #             unknown fields raise; no header line. reading with headers: false: keys from the fields
+  #             of the record type
+  #   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 +49,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 +98,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]
 
@@ -107,6 +132,7 @@ class DataPipeline
     # rows with headers are CSV::Row, without headers plain arrays
     def to_hash row
       return row.to_h if row.is_a?(CSV::Row)
+      return @options[:layout].parse(row) if @options[:layout]
 
       keys = columns&.map(&:to_s) || (0...row.length).map(&:to_s)
       keys.zip(row).to_h

lib/datapipeline/storage/fixed.rb

@@ -0,0 +1,77 @@
+require_relative 'storage_plugin'
+
+class DataPipeline
+  #
+  # fixed width file, one line per row, field widths from a DataPipeline::Layout
+  #
+  # options
+  #   file:     path of the file
+  #   layout:   DataPipeline::Layout, every AN and N field needs a length
+  #   encoding: encoding of the file (default UTF-8), read as UTF-8, written transcoded
+  #   replace:  '?' -> characters the encoding cannot hold are replaced when writing (default: error)
+  #   line_end: "\n" (default) or "\r\n"
+  #
+  # writing: values formatted as with Layout format: true, text left aligned with blanks, numbers right
+  # aligned with zeros (sign in front), empty fields blank; a value longer than its field raises LayoutError
+  # reading: record type from the type field, keys are the field names in lower case, values without padding,
+  # blank fields nil
+  #
+  class FixedStorage < StoragePlugin
+    def initialize **options, &block
+      @options = options
+      @file_name = options.fetch(:file) { raise ArgumentError, 'FixedStorage: option file missing' }
+      @layout = options.fetch(:layout) { raise ArgumentError, 'FixedStorage: option layout missing' }.check_widths!
+      @line_end = options.fetch(:line_end, "\n")
+
+      yield if block_given?
+    end
+
+    def read(&block)
+      return each_line.map { |line| @layout.parse_line(line) } unless block_given?
+
+      each_line { |line| yield @layout.parse_line(line) }
+      nil
+    end
+
+    def write data
+      data.each { |row| write_row row }
+    ensure
+      close
+    end
+
+    def write_row row
+      @file ||= File.open(@file_name, 'w', **write_options)
+      @file.write(@layout.line(row), @line_end)
+    end
+
+    def close
+      @file&.close
+      @file = nil
+    end
+
+    private
+
+    def utf8?
+      @options[:encoding].nil? || Encoding.find(@options[:encoding]) == Encoding::UTF_8
+    end
+
+    # empty lines are skipped
+    def each_line
+      return enum_for(:each_line) unless block_given?
+
+      encoding = utf8? ? 'bom|utf-8' : "#{@options[:encoding]}:utf-8"
+      File.foreach(@file_name, encoding:) do |line|
+        line = line.chomp
+        yield line unless line.strip.empty?
+      end
+    end
+
+    def write_options
+      return { encoding: 'utf-8' } if utf8?
+
+      options = { encoding: @options[:encoding] }
+      options.merge!(undef: :replace, invalid: :replace, replace: @options[:replace]) if @options[:replace]
+      options
+    end
+  end
+end

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
@@ -45,9 +48,12 @@ class DataPipeline
       each_row(@options[:query], params, &block)
     end
 
+    # pending rows of write_row are written first, so a query sees them
     def query options = {}
       raise "Query for PostgresStorage:query missing" unless options[:query]
 
+      flush
+
       result = @conn.exec_params(options[:query], options.fetch(:params, []))
       return result.to_a unless block_given?
 
@@ -65,25 +71,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]
 
+      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 +181,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 <other columns> = excluded.<column>
+    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('.')

lib/datapipeline/storage/sqlite.rb

@@ -10,6 +10,10 @@ 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
+  #   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
   #
   # rows come back as Hash<String, value> with sqlite types (Integer, Float, String, nil).
   # the database file is opened on first use: a dry run creates no file.
@@ -31,9 +35,11 @@ class DataPipeline
       query(query: @options[:query], params: @options.fetch(:params, []), &block)
     end
 
+    # pending rows of write_row are written first, so a query sees them
     def query options = {}
       raise 'Query for SqliteStorage:query missing' unless options[:query]
 
+      flush
       statement = db.prepare(options[:query])
       result = statement.execute(*bind_args(options.fetch(:params, [])))
       return result.map(&:to_h) unless block_given?
@@ -56,20 +62,47 @@ class DataPipeline
 
       return if data.empty?
 
-      db.transaction do
+      transaction do
         prepare_table table_name, data.first
         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 SqliteStorage#write_row missing' unless @options[:table]
 
+      if batch_size == :all
+        begin_stream
         prepare_table @options[:table], row
         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
+        db.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
+      aborted_by = $!
+      begin
+        flush
+      rescue StandardError
+        raise unless aborted_by
+      end
+      @db.rollback if @stream_open && @db.transaction_active?
+      @stream_open = false
       @db.close if @db && !@db.closed?
     end
 
@@ -79,6 +112,38 @@ class DataPipeline
       @db ||= SQLite3::Database.new(@database).tap { |d| d.results_as_hash = true }
     end
 
+    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
+        rows.each do |row|
+          prepare_table @options[:table], row
+          insert @options[:table], row
+        end
+      end
+    end
+
+    def begin_stream
+      return if @stream_open
+
+      raise 'SqliteStorage: batch_size: :all needs its own connection (no open transaction)' if db.transaction_active?
+
+      db.transaction
+      @stream_open = true
+    end
+
+    # inside an open transaction no own transaction is opened
+    def transaction(&block)
+      db.transaction_active? ? yield : db.transaction(&block)
+    end
+
     # create / truncate once per table, before its first row
     def prepare_table table_name, row
       return if @prepared[table_name]
@@ -86,6 +151,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 +174,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 <other columns> = excluded.<column>
+    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

lib/datapipeline/version.rb

@@ -1,5 +1,5 @@
 class DataPipeline
-  VERSION = '2.0.0'.freeze
+  VERSION = '3.1.0'.freeze
 
   # kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md
   class Version

test/cli/etl_cli_test.rb

@@ -14,8 +14,8 @@ class EtlCliTest < PipelineTest
     "#{dir}/#{name}"
   end
 
-  # like import.*.rb: own OptionParser for --from/--till, transform api
-  def test_legacy_script_gets_from_till_and_limit
+  # scripts with an own OptionParser for --from/--till
+  def test_script_with_own_option_parser_gets_from_till_and_limit
     with_tmpdir do |dir|
       path = script(dir, 'legacy.rb', <<~RUBY)
         require 'optparse/date'
@@ -24,8 +24,8 @@ class EtlCliTest < PipelineTest
         File.write('args.txt', "\#{opts[:from]} \#{opts[:till]}")
         DataPipeline.new do
           read container(:csv, file: 'in.csv')
-          transform(processing: :prepend) { |_, _| { 'h' => 'HEAD' } }
-          transform { |r, i| r.merge('i' => i) }
+          header { { 'h' => 'HEAD' } }
+          map { |r, ctx| r.merge('i' => ctx.index) }
           write container(:csv, file: 'out.csv', csv: { headers: false, col_sep: '|' })
         end.run
       RUBY
@@ -86,4 +86,45 @@ 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
+
+  def test_history_with_script_name
+    with_tmpdir do |dir|
+      path = script(dir, 'export_kunden.rb', <<~RUBY)
+        DataPipeline.new do
+          read container(:csv, file: 'in.csv')
+          write container(:csv, file: 'out.csv')
+        end.run
+      RUBY
+      _out, err, status = etl(dir, 'run', path, '--history', 'runs.ndjson')
+      assert_equal 0, status.exitstatus, err
+      entry = JSON.parse(File.read("#{dir}/runs.ndjson"))
+      assert_equal ['export_kunden', 3, true], entry.values_at('name', 'read', 'ok')
+    end
+  end
 end

test/core/chained_test.rb

@@ -4,16 +4,22 @@ require_relative '../test_helper'
 class ChainedTest < PipelineTest
   ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
 
-  def test_chained_transform_gets_previous_output
-    out = output(ROWS) { chained; transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
+  def test_each_step_gets_the_previous_output
+    out = output(ROWS) { map { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 2) } }
     assert_equal ROWS.map { |r| r.merge('b' => 1, 'c' => 2) }, out
   end
 
   def test_one_to_many_then_index
-    out = output(ROWS) { chained; transform { |r| [r, r] }; transform { |r, i| r.merge('i' => i) } }
+    out = output(ROWS) { map { |r| [r, r] }; map { |r, ctx| r.merge('i' => ctx.index) } }
     assert_equal (0...6).to_a, out.map { |r| r['i'] }
   end
 
+  def test_nil_and_empty_array_drop_rows
+    assert_equal [{ 'a' => 1 }], output(ROWS) { map { |r| r['a'] == 1 ? r : nil } }
+    assert_equal ROWS, output(ROWS) { map { |r| [r, nil] } }
+    assert_equal [], output(ROWS) { map { [] } }
+  end
+
   def test_map_and_filter
     out = output(ROWS) do
       map { |r| r.merge('d' => r['a'] * 2) }
@@ -23,12 +29,6 @@ class ChainedTest < PipelineTest
     assert_equal [{ 'a' => 2, 'd' => 4, 'i' => 0 }, { 'a' => 3, 'd' => 6, 'i' => 1 }], out
   end
 
-  def test_map_turns_chaining_on_for_transform
-    out = output(ROWS) { transform { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 1) } }
-    assert_equal 3, out.size
-    assert(out.all? { |r| r['b'] && r['c'] })
-  end
-
   def test_header_and_trailer_with_context
     out = output(ROWS) do
       header { |c| { 'H' => c.read, 'W' => c.written } }
@@ -41,9 +41,14 @@ class ChainedTest < PipelineTest
     assert_equal 6, out.size
   end
 
-  def test_edge_order_follows_registration
-    out = output(ROWS) { header { { 'h' => 1 } }; transform(processing: :prepend) { { 'p' => 1 } }; map { |r| r } }
-    assert_equal [{ 'h' => 1 }, { 'p' => 1 }], out.first(2)
+  def test_several_headers_and_trailers_in_registration_order
+    out = output(ROWS) { header { { 'h' => 1 } }; trailer { { 't' => 1 } }; header { { 'h' => 2 } }; trailer { { 't' => 2 } } }
+    assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
+  end
+
+  def test_stable_order_of_many_steps
+    out = output(ROWS) { 20.times { |k| map { |r| r.merge("k#{k}" => k) } } }
+    assert_equal (0...20).to_a, out.first.values.drop(1)
   end
 
   def test_header_and_trailer_nil_and_arrays
@@ -51,12 +56,13 @@ class ChainedTest < PipelineTest
     assert_equal ROWS + [{ 'x' => 1 }, { 'y' => 2 }], out
   end
 
-  def test_condition_on_map
+  def test_condition_on_map_and_filter
     assert_equal [{ 'a' => 1 }], output(ROWS) { map(condition: ->(r) { r['a'] == 1 }) { |r| r } }
+    assert_equal [{ 'a' => 3 }], output(ROWS) { filter(condition: ->(r) { r['a'] > 1 }) { |r| r['a'] == 3 } }
   end
 
-  def test_chained_false
-    assert_equal 6, output(ROWS) { chained false; transform { |r| r }; transform { |r| r } }.size
+  def test_condition_must_be_a_lambda
+    assert_includes assert_raises(ArgumentError) { output(ROWS) { map(condition: "row['a'] == 1") { |r| r } } }.message, 'lambda'
   end
 
   def test_csv_end_to_end

test/core/compat_test.rb

@@ -1,81 +0,0 @@
-require_relative '../test_helper'
-
-#
-# parallel mode (transform without chained/map/filter) must keep the output of v1.2.1,
-# the version the CUDO importers were built on. expected values were recorded with v1.2.1.
-#
-class CompatTest < PipelineTest
-  ROWS = (1..6).map { |i| { 'id' => i, 'land' => i.even? ? 'DE' : 'AT' } }
-
-  def assert_output expected_json, &block
-    assert_equal expected_json, JSON.generate(output(ROWS, &block))
-  end
-
-  # pattern of import.customer.rb
-  def test_customer_pattern
-    assert_output '[{"SATYP":"FIRC","LFDNR":0},{"id":1,"land":"AT","lfdnr":"0000000001"},' \
-                  '{"id":2,"land":"DE","lfdnr":"0000000002"},{"id":3,"land":"AT","lfdnr":"0000000003"},' \
-                  '{"id":4,"land":"DE","lfdnr":"0000000004"},{"id":5,"land":"AT","lfdnr":"0000000005"},' \
-                  '{"id":6,"land":"DE","lfdnr":"0000000006"},{"topic":"LARC","counter":"0000000006"}]' do
-      transform(processing: :prepend) { |_row, n| { 'SATYP': 'FIRC', 'LFDNR': n } }
-      transform(processing: :append) { |_row, n| { 'topic': 'LARC', 'counter': format('%010d', n) } }
-      transform { |row, _i| row.merge('lfdnr' => format('%010d', row['id'])) }
-    end
-  end
-
-  # pattern of import.posting.rb: closure counter, 1:n
-  def test_posting_pattern
-    expected = [{ SATYP: 'FIRC', LFDNR: 0 }]
-    lfdnr = 0
-    ROWS.each do |row|
-      expected << row.merge('lfdnr' => lfdnr += 1)
-      row['id'].times { |k| expected << { 'pos' => k, 'lfdnr' => lfdnr += 1 } }
-    end
-    expected << { SATYPE: 'LARC', SUMME: '0000000027' }
-
-    assert_output JSON.generate(expected) do
-      cnt = 1
-      transform(processing: :prepend) { |_row, _n| { 'SATYP': 'FIRC', 'LFDNR': 0 } }
-      transform do |row|
-        out = [row.merge('lfdnr' => cnt)]
-        cnt += 1
-        row['id'].times do |k|
-          out << { 'pos' => k, 'lfdnr' => cnt }
-          cnt += 1
-        end
-        out
-      end
-      transform(processing: :append) { |_row| { 'SATYPE': 'LARC', 'SUMME': format('%010d', cnt - 1) } }
-    end
-  end
-
-  def test_append_registered_before_prepend
-    assert_output '[{"h":0},{"id":1,"land":"AT"},{"id":2,"land":"DE"},{"id":3,"land":"AT"},' \
-                  '{"id":4,"land":"DE"},{"id":5,"land":"AT"},{"id":6,"land":"DE"},{"t":6}]' do
-      transform(processing: :append) { |_, n| { 't' => n } }
-      transform { |r| r }
-      transform(processing: :prepend) { |_, n| { 'h' => n } }
-    end
-  end
-
-  def test_condition_string
-    assert_output '[{"id":2,"land":"DE","i":1},{"id":4,"land":"DE","i":3},{"id":6,"land":"DE","i":5}]' do
-      transform(condition: "row['land'] == 'DE'") { |r, i| r.merge('i' => i) }
-    end
-  end
-
-  def test_two_parallel_transforms
-    expected = ROWS.map { |r| r.merge('a' => 1) } + ROWS.map { |r| r.merge('b' => 1) }
-    assert_output JSON.generate(expected) do
-      transform { |r| r.merge('a' => 1) }
-      transform { |r| [r.merge('b' => 1), nil] }
-    end
-  end
-
-  def test_nil_and_empty_array
-    assert_output '[{"id":1,"land":"AT"}]' do
-      transform { |r| r['id'] > 3 ? nil : [] }
-      transform { |r| r['id'] == 1 ? r : nil }
-    end
-  end
-end

test/core/connections_totals_history_test.rb

@@ -0,0 +1,98 @@
+require_relative '../test_helper'
+require 'time'
+
+# connections file, totals with reconcile, run history
+class ConnectionsTotalsHistoryTest < PipelineTest
+  def teardown
+    DataPipeline.connections_file = nil
+    ENV.delete('ETL_TEST_SECRET')
+    super
+  end
+
+  def write_connections dir
+    File.write("#{dir}/connections.yml", <<~YAML)
+      mem:
+        type: memory
+        rows: [{ a: 1 }]
+      secret:
+        type: csv
+        file: <%= ENV.fetch('ETL_TEST_SECRET', 'default.csv') %>
+    YAML
+    DataPipeline.connections_file = "#{dir}/connections.yml"
+  end
+
+  def test_connection_and_container_with_connection
+    with_tmpdir do |dir|
+      write_connections(dir)
+      ENV['ETL_TEST_SECRET'] = "#{dir}/from_env.csv"
+      pipeline = DataPipeline.new { logger PipelineHelper::QUIET }
+      assert_kind_of MemoryStorage, pipeline.connection(:mem)
+      csv = pipeline.container(:csv, connection: 'secret')
+      assert_equal "#{dir}/from_env.csv", csv.instance_variable_get(:@file_name)
+      csv = pipeline.connection(:secret, file: "#{dir}/override.csv")
+      assert_equal "#{dir}/override.csv", csv.instance_variable_get(:@file_name)
+
+      assert_includes assert_raises(ArgumentError) { pipeline.container(:json, connection: :secret) }.message, 'is csv, not json'
+      assert_includes assert_raises(ArgumentError) { pipeline.connection(:nope) }.message, 'mem, secret'
+      DataPipeline.connections_file = "#{dir}/missing.yml"
+      assert_includes assert_raises(ArgumentError) { pipeline.connection(:mem) }.message, 'missing.yml missing'
+    end
+  end
+
+  ROWS = [{ 'beleg' => '1', 'dmbtr' => '10.10' }, { 'beleg' => '2', 'dmbtr' => 5.05 },
+          { 'beleg' => '3', 'dmbtr' => '' }, { 'beleg' => '4', 'dmbtr' => 'x' }].freeze
+
+  def test_totals_batch_and_streaming
+    [false, true].each do |streamed|
+      _target, stats = run_pipeline(ROWS, streamed:) do
+        totals 'dmbtr'
+        filter { |row| row['beleg'] != '2' }
+      end
+      assert_equal Rational(1515, 100), stats.total('dmbtr', :read)
+      assert_equal Rational(1010, 100), stats.total('dmbtr')
+      assert_includes stats.to_s, 'dmbtr=15.15/10.1'
+    end
+  end
+
+  def test_reconcile_aborts_before_writing
+    target = nil
+    error = assert_raises(DataPipeline::ReconcileError) do
+      target, = run_pipeline(ROWS) do
+        totals 'dmbtr'
+        filter { |row| row['beleg'] != '2' }
+        reconcile { |stats| "Summe: #{stats.total('dmbtr', :read).to_f} statt #{stats.total('dmbtr').to_f}" if stats.total('dmbtr', :read) != stats.total('dmbtr') }
+        reconcile { |stats| ['zu wenig Zeilen'] if stats.written < 4 }
+      end
+    end
+    assert_equal 'Summe: 15.15 statt 10.1; zu wenig Zeilen', error.message
+    assert_nil target
+
+    _target, stats = run_pipeline(ROWS) { totals 'dmbtr'; reconcile { |s| nil if s.total('dmbtr', :read) == s.total('dmbtr') } }
+    assert stats.ok?
+  end
+
+  def test_history_lines_also_for_aborted_runs
+    with_tmpdir do |dir|
+      file = "#{dir}/log/runs.ndjson"
+      run_pipeline(ROWS) { name 'buchungen'; version '1.0'; history file; totals 'dmbtr' }
+      assert_raises(RuntimeError) { run_pipeline(ROWS) { name 'buchungen'; history file; map { raise 'kaputt' } } }
+      run_pipeline(ROWS) { name 'andere'; history file }
+
+      entries = DataPipeline.history(file, name: 'buchungen')
+      assert_equal 2, entries.size
+      first, second = entries
+      assert_equal({ 'name' => 'buchungen', 'version' => '1.0', 'read' => 4, 'written' => 4, 'rejected' => 0, 'ok' => true,
+                     'error' => nil, 'totals' => { 'dmbtr' => { 'read' => '15.15', 'written' => '15.15' } } },
+                   first.slice('name', 'version', 'read', 'written', 'rejected', 'ok', 'error', 'totals'))
+      assert_equal [false, 'RuntimeError: kaputt'], second.values_at('ok', 'error')
+      assert Time.iso8601(first['started_at'])
+      assert_equal 3, DataPipeline.history(file).size
+      assert_equal [], DataPipeline.history("#{dir}/none.ndjson")
+    end
+  end
+
+  def test_decimal
+    assert_equal ['1234.5', '0.333333', '7', '12'], [DataPipeline.decimal(Rational(12_345, 10)), DataPipeline.decimal(Rational(1, 3)),
+                                                     DataPipeline.decimal(7r), DataPipeline.decimal(12)]
+  end
+end

test/core/layout_test.rb

@@ -71,4 +71,54 @@ class LayoutTest < PipelineTest
     assert_equal [{ 'a' => '1' }], target.output
     assert_equal 1, stats.rejected
   end
+
+  FORMATTED = DataPipeline::Layout.new(format: true) do
+    record 'K', [['SATYP', 'AN', 2, true], ['DATUM', 'DAT'], ['ZEIT', 'TIM'], ['BETRAG', 'N', '11.2'], ['NR', 'N', 5], ['TEXT', 'AN', 6]]
+  end
+
+  def test_format_values_by_type
+    row = { 'satyp' => 'K', 'datum' => Date.new(2026, 10, 9), 'zeit' => Time.new(2026, 10, 9, 7, 5, 3), 'betrag' => 12.345, 'nr' => 7, 'text' => 'abc' }
+    assert_equal ['K', '20261009', '070503', '12.35', '7', 'abc'], FORMATTED.values(row)
+    assert_equal ['K', '20261009', '070503', '3.50', nil, nil], FORMATTED.values({ 'satyp' => 'K', 'datum' => '2026-10-09', 'zeit' => '07:05:03', 'betrag' => '3.5' })
+    assert_equal '-0.01', FORMATTED.values({ 'satyp' => 'K', 'betrag' => -0.005 })[3]
+    assert_equal '0.33', FORMATTED.values({ 'satyp' => 'K', 'betrag' => Rational(1, 3) })[3]
+    assert_equal '1.234', FORMATTED.values({ 'satyp' => 'K', 'betrag' => '1.234' })[3] # texts are not rounded
+    assert_equal ['K', nil, nil, nil, nil, nil], FORMATTED.values({ 'satyp' => 'K' })
+    assert_equal [nil, 12.345], [LAYOUT.values({ 'satyp' => 'HEAD' })[1], DataPipeline::Layout.new { record 'K', [['S'], ['B', 'N', '5.2']] }.values({ 's' => 'K', 'b' => 12.345 })[1]]
+  end
+
+  def test_errors_check_the_formatted_values
+    assert_empty FORMATTED.errors({ 'satyp' => 'K', 'datum' => Date.today, 'betrag' => 0.1 + 0.2 })
+    assert_equal ['BETRAG: more than 2 decimals'], DataPipeline::Layout.new { record 'K', [['S'], ['BETRAG', 'N', '11.2']] }.errors({ 's' => 'K', 'betrag' => 0.1 + 0.2 })
+  end
+
+  def test_fixed_line_and_parse_line
+    row = { 'satyp' => 'K', 'datum' => Date.new(2026, 10, 9), 'betrag' => -12.5, 'nr' => 7, 'text' => 'Müller' }
+    line = FORMATTED.line(row)
+    assert_equal 'K 20261009      -00000012.5000007Müller', line
+    assert_equal 2 + 8 + 6 + 12 + 5 + 6, line.length
+    assert_equal({ 'satyp' => 'K', 'datum' => '20261009', 'zeit' => nil, 'betrag' => '-00000012.50', 'nr' => '00007', 'text' => 'Müller' },
+                 FORMATTED.parse_line(line))
+    assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.line(row.merge('text' => 'zu lang')) }.message, 'longer than 6'
+    assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.line(row.merge('nr' => 'x')) }.message, 'not a number'
+    assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.line(row.merge('extra' => 1)) }.message, 'extra'
+    assert_includes assert_raises(DataPipeline::LayoutError) { FORMATTED.parse_line('XX') }.message, 'not in layout'
+  end
+
+  def test_check_widths
+    layout = DataPipeline::Layout.new { record 'A', [['SATYP', 'AN', 1], ['TEXT'], ['NR', 'N'], ['D', 'DAT']] }
+    assert_equal 'fixed width needs a length for: A.TEXT, A.NR', assert_raises(DataPipeline::LayoutError) { layout.check_widths! }.message
+    assert_same FORMATTED, FORMATTED.check_widths!
+  end
+
+  def test_parse_values_and_csv_read_with_layout
+    assert_equal({ 'satyp' => 'POS', 'nr' => '1', 'text' => nil, 'betrag' => nil }, LAYOUT.parse(%w[POS 1]))
+    assert_includes assert_raises(DataPipeline::LayoutError) { LAYOUT.parse(%w[HEAD 1 2 3]) }.message, '4 values'
+    with_tmpdir do |dir|
+      File.write("#{dir}/in.txt", "HEAD|20251103|\nPOS|1|abc|1.50\n")
+      rows = DataPipeline::CsvStorage.new(file: "#{dir}/in.txt", layout: LAYOUT, csv: { headers: false, col_sep: '|' }).read
+      assert_equal [{ 'satyp' => 'HEAD', 'datum' => '20251103', 'zeit' => nil },
+                    { 'satyp' => 'POS', 'nr' => '1', 'text' => 'abc', 'betrag' => '1.50' }], rows
+    end
+  end
 end

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

test/core/operations_test.rb

@@ -41,8 +41,8 @@ class OperationsTest < PipelineTest
     assert_raises(RuntimeError) { run_pipeline(ROWS) { on_error :skip; header { raise 'h' }; map { |r| r } } }
   end
 
-  def test_skip_in_parallel_mode
-    target, stats = run_pipeline(ROWS) { on_error :skip; transform { |r| r['id'] == '1' ? raise('x') : r } }
+  def test_skip_in_map
+    target, stats = run_pipeline(ROWS) { on_error :skip; map { |r| r['id'] == '1' ? raise('x') : r } }
     assert_equal 4, target.output.size
     assert_equal 1, stats.rejected
   end
@@ -67,8 +67,8 @@ class OperationsTest < PipelineTest
     assert_empty DataPipeline::Validation.errors({ 'a' => '' }, 'a' => { max: 1, type: :integer })
   end
 
-  def test_validate_turns_chaining_on
-    target, = run_pipeline(ROWS) { on_error :skip; transform { |r| r.merge('x' => 1) }; validate RULES }
+  def test_validate_gets_the_output_of_the_previous_step
+    target, = run_pipeline(ROWS) { on_error :skip; map { |r| r.merge('x' => 1) }; validate RULES }
     assert_equal 3, target.output.size
     assert(target.output.all? { |r| r['x'] })
   end

test/core/pipeline_test.rb

@@ -1,17 +1,11 @@
 require_relative '../test_helper'
 
-# dsl, settings, plugin lookup, parallel mode
+# dsl, settings, plugin lookup
 class PipelineCoreTest < PipelineTest
   ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
 
   def settings_of(pipeline) = pipeline.instance_variable_get(:@settings)
 
-  def test_nil_filters_rows
-    assert_equal [{ 'a' => 1 }], output(ROWS) { transform { |r| r['a'] == 1 ? r : nil } }
-    assert_equal ROWS, output(ROWS) { transform { |r| [r, nil] } }
-    assert_equal ROWS, output(ROWS) { transform(processing: :prepend) { nil }; transform { |r| r }; transform(processing: :append) { nil } }
-  end
-
   def test_dsl_switches
     assert_equal false, settings_of(DataPipeline.new { verbose false })[:verbose]
     assert_equal true, settings_of(DataPipeline.new { verbose })[:verbose]
@@ -73,39 +67,8 @@ class PipelineCoreTest < PipelineTest
     assert_equal Array.new(8) { |i| (i % 3) + 1 }, reads
   end
 
-  def test_parallel_mode_is_side_by_side
-    out = output(ROWS) { transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
-    assert_equal 6, out.size
-  end
-
-  def test_parallel_mode_copies_rows_per_step
-    out = output(ROWS) { transform { |r| r['x'] = 1; r }; transform { |r| r.key?('x') ? nil : r } }
-    assert_equal 6, out.size
-    assert_equal [{ 'a' => 10 }, { 'a' => 20 }, { 'a' => 30 }], output(ROWS) { transform { |r| [r.transform_values! { |v| v * 10 }] } }
-  end
-
-  def test_stable_order_and_all_prepends_appends
-    out = output(ROWS) { 20.times { |k| transform { |r| r['a'] == 1 ? { 'k' => k } : nil } } }
-    assert_equal (0...20).to_a, out.map { |r| r['k'] }
-
-    out = output(ROWS) do
-      transform(processing: :prepend) { { 'h' => 1 } }
-      transform(processing: :prepend) { { 'h' => 2 } }
-      transform { |r| r }
-      transform(processing: :append) { { 't' => 1 } }
-      transform(processing: :append) { { 't' => 2 } }
-    end
-    assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
-  end
-
-  def test_append_count_parallel_vs_chained
-    assert_equal({ 'n' => 3 }, output(ROWS) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
-    assert_equal({ 'n' => 6 }, output(ROWS) { chained; transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
-  end
-
-  def test_conditions
-    assert_equal [{ 'a' => 1 }, { 'a' => 3 }], output(ROWS) { transform(condition: ->(r) { r['a'] != 2 }) { |r| r } }
-    assert_equal [{ 'a' => 2 }], output(ROWS) { transform(condition: "row['a'] == 2") { |r| r } }
+  def test_removed_dsl_of_version_2
+    %i[transform chained].each { |name| assert_raises(NoMethodError) { DataPipeline.new { send(name) { |r| r } } } }
   end
 
   def test_copy_without_steps

test/core/steps_test.rb

@@ -0,0 +1,47 @@
+require_relative '../test_helper'
+
+# rename, select, cast, default
+class StepsTest < PipelineTest
+  ROWS = [{ 'kunnr' => '1', 'betrag' => '12.50', 'datum' => '2026-10-09', 'land' => nil },
+          { 'kunnr' => '2', 'betrag' => '', 'datum' => ' ', 'land' => 'DE' }].freeze
+
+  def test_rename_keeps_the_position
+    assert_equal [%w[kundennr betrag datum land]] * 2, output(ROWS) { rename 'kunnr' => 'kundennr' }.map(&:keys)
+    assert_equal [{ 'b' => 1 }], output([{ a: 1 }]) { rename a: 'b' }
+  end
+
+  def test_select_order_and_missing_keys
+    assert_equal [{ 'betrag' => '12.50', 'kunnr' => '1', 'x' => nil }], output(ROWS.first(1)) { select %w[betrag kunnr x] }
+    assert_equal [{ 'kunnr' => '1' }], output([{ kunnr: '1', b: 2 }]) { select 'kunnr' }
+  end
+
+  def test_cast_types_and_empty_values
+    rows = output(ROWS) { cast 'betrag' => :decimal, 'datum' => :date, 'kunnr' => :integer }
+    assert_equal [1, 12.5, Date.new(2026, 10, 9)], rows[0].values_at('kunnr', 'betrag', 'datum')
+    assert_equal [2, nil, nil], rows[1].values_at('kunnr', 'betrag', 'datum')
+    assert_equal [{ 'a' => '5' }], output([{ 'a' => 5 }]) { cast 'a' => :string }
+    assert_includes assert_raises(ArgumentError) { output(ROWS) { cast 'a' => :money } }.message, 'money unknown'
+  end
+
+  def test_cast_error_goes_to_on_error
+    target, stats = run_pipeline([{ 'n' => 'x' }, { 'n' => '3' }]) { on_error :skip; cast 'n' => :integer }
+    assert_equal [{ 'n' => 3 }], target.output
+    assert_equal [{ step: 'cast#1', error: 'n: "x" is not integer', row: { 'n' => 'x' } }], stats.errors
+  end
+
+  def test_default_value_and_lambda
+    rows = output(ROWS) { default 'land' => 'CH', 'datum' => ->(row) { "neu #{row['kunnr']}" }, 'neu' => 0 }
+    assert_equal [['CH', '2026-10-09', 0], ['DE', 'neu 2', 0]], rows.map { |r| r.values_at('land', 'datum', 'neu') }
+  end
+
+  def test_steps_chain_with_map_and_validate
+    rows = output(ROWS) do
+      rename 'kunnr' => 'id'
+      default 'betrag' => '0'
+      cast 'betrag' => :decimal
+      validate 'id' => { required: true }
+      select %w[id betrag]
+    end
+    assert_equal [{ 'id' => '1', 'betrag' => 12.5 }, { 'id' => '2', 'betrag' => 0.0 }], rows
+  end
+end

test/core/streaming_test.rb

@@ -21,23 +21,14 @@ class StreamingTest < PipelineTest
     assert_equal({ 'T' => 4, 'R' => 3 }, stream.last)
   end
 
-  def test_stream_prepend_append
+  def test_stream_several_steps_one_to_many
     out = output(ROWS, streamed: true) do
-      chained
-      transform(processing: :prepend) { |_, n| { 'p' => n } }
-      transform { |r| [r, r] }
-      transform(processing: :append) { |_, n| { 'n' => n } }
+      map { |r| [r, r] }
+      map { |r, ctx| r.merge('i' => ctx.index) }
+      trailer { |ctx| { 'n' => ctx.written } }
     end
-    assert_equal({ 'p' => 0 }, out.first)
     assert_equal({ 'n' => 6 }, out.last)
-    assert_equal 8, out.size
-  end
-
-  def test_stream_parallel_single_step_and_error_for_several
-    out = output(ROWS, streamed: true) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }
-    assert_equal({ 'n' => 3 }, out.last)
-    error = assert_raises(ArgumentError) { output(ROWS, streamed: true) { transform { |r| r }; transform { |r| r } } }
-    assert_includes error.message, 'chained'
+    assert_equal (0...6).to_a, out.first(6).map { |r| r['i'] }
   end
 
   def test_stream_condition_and_copy

test/examples/doc_examples_test.rb

@@ -4,15 +4,15 @@ require 'open3'
 require 'fileutils'
 
 #
-# the examples of README.md section 1: the code in the documentation is the file,
+# the examples of README.md section 2: the code in the documentation is the file,
 # and every example runs and produces the documented result
 #
 class DocExamplesTest < PipelineTest
   DOC = File.read(File.join(ROOT, 'README.md'))
   EXAMPLES = DOC.scan(/<!-- beispiel: (\S+) -->\n```ruby\n(.*?)\n```/m)
 
-  def run_example dir, name, *args
-    env = { 'BUNDLE_GEMFILE' => File.join(ROOT, 'Gemfile') }
+  def run_example dir, name, *args, env: {}
+    env = { 'BUNDLE_GEMFILE' => File.join(ROOT, 'Gemfile') }.merge(env)
     Open3.capture3(env, RbConfig.ruby, File.join(ROOT, 'bin/etl'), 'run', File.join(ROOT, name), "--out=#{dir}", *args,
                    chdir: dir)
   end
@@ -20,7 +20,7 @@ class DocExamplesTest < PipelineTest
   def test_documented_code_equals_files
     assert_equal %w[examples/kunden_import.rb examples/rechnungen_satzarten.rb examples/bewegungen_inkrementell.rb
                     examples/eigenes_plugin.rb examples/rest_lesen.rb examples/rest_schreiben.rb
-                    examples/rest_zu_rest.rb], EXAMPLES.map(&:first)
+                    examples/rest_zu_rest.rb examples/buchungen_export.rb], EXAMPLES.map(&:first)
     EXAMPLES.each do |file, code|
       assert_equal File.read(File.join(ROOT, file)).rstrip, code.rstrip, "#{file} differs from README.md"
     end
@@ -152,4 +152,33 @@ class DocExamplesTest < PipelineTest
       end
     end
   end
+
+  def test_buchungen_export
+    with_tmpdir do |dir|
+      env = { 'BUCHHALTUNG_DB' => "#{dir}/buchhaltung.sqlite3", 'SLACK_TOKEN' => nil }
+      out, err, status = run_example(dir, 'examples/buchungen_export.rb', '--dry-run', env:)
+      assert_equal 0, status.exitstatus, err
+      refute File.exist?("#{dir}/buchhaltung.sqlite3")
+
+      2.times do
+        out, err, status = run_example(dir, 'examples/buchungen_export.rb', env:)
+        assert_equal 0, status.exitstatus, err
+      end
+      assert_includes out, 'buchungen_export: read=7 written=7 rejected=0 betrag=0/0'
+
+      db = DataPipeline::SqliteStorage.new(database: "#{dir}/buchhaltung.sqlite3", query: 'select * from buchungen order by belnr, buzei')
+      rows = db.read
+      db.close
+      assert_equal 7, rows.size # upsert: the second run updates instead of inserting
+      assert_equal({ 'belnr' => '5100000002', 'buzei' => 3, 'datum' => '2026-10-02', 'konto' => '2000', 'shkzg' => 'H',
+                     'dmbtr' => '409.80', 'betrag' => -409.8, 'buchungstext' => 'Papeterie Zürcher AG' }, rows[4])
+      assert_equal 'ohne Text', rows[5]['buchungstext']
+
+      csv = File.read("#{dir}/buchungen_sap.csv", encoding: 'Windows-1252').encode('UTF-8')
+      assert_includes csv, "5100000001;1;2026-10-01;1020;S;1250.00;Zahlung Müller GmbH\n"
+      history = DataPipeline.history("#{dir}/runs.ndjson", name: 'buchungen_export')
+      assert_equal [true, true, true], history.map { |run| run['ok'] }
+      assert_equal [true, false, false], history.map { |run| run['dry_run'] }
+    end
+  end
 end

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

test/plugins/fixed_storage_test.rb

@@ -0,0 +1,63 @@
+require_relative '../test_helper'
+
+class FixedStorageTest < PipelineTest
+  LAYOUT = DataPipeline::Layout.new do
+    record 'KOPF', [['SATYP', 'AN', 4, true], ['BELNR', 'AN', 6, true], ['DATUM', 'DAT', 8]]
+    record 'POS',  [['SATYP', 'AN', 4, true], ['TEXT', 'AN', 10], ['BETRAG', 'N', '7.2']]
+  end
+
+  ROWS = [{ 'satyp' => 'KOPF', 'belnr' => 'R1', 'datum' => Date.new(2026, 10, 9) },
+          { 'satyp' => 'POS', 'text' => 'Müller', 'betrag' => 12.5 },
+          { 'satyp' => 'POS', 'betrag' => -3 }].freeze
+
+  def test_write_and_read_back
+    with_tmpdir do |dir|
+      DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT).write(ROWS)
+      assert_equal "KOPFR1    20261009\nPOS Müller    00012.50\nPOS           -0003.00\n", File.read("#{dir}/out.txt")
+      assert_equal [{ 'satyp' => 'KOPF', 'belnr' => 'R1', 'datum' => '20261009' },
+                    { 'satyp' => 'POS', 'text' => 'Müller', 'betrag' => '00012.50' },
+                    { 'satyp' => 'POS', 'text' => nil, 'betrag' => '-0003.00' }],
+                   DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT).read
+    end
+  end
+
+  def test_encoding_line_end_and_streaming_read
+    with_tmpdir do |dir|
+      DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT, encoding: 'ISO-8859-1', line_end: "\r\n").write(ROWS.first(2))
+      assert_equal "KOPFR1    20261009\r\nPOS M\xFCller    00012.50\r\n".b, File.binread("#{dir}/out.txt")
+      rows = []
+      DataPipeline::FixedStorage.new(file: "#{dir}/out.txt", layout: LAYOUT, encoding: 'ISO-8859-1').read { |row| rows << row }
+      assert_equal 'Müller', rows.last['text']
+    end
+  end
+
+  def test_pipeline_with_container_and_dry_run
+    with_tmpdir do |dir|
+      rows = ROWS.map(&:dup)
+      out = capture_stdout do
+        DataPipeline.new do
+          logger PipelineHelper::QUIET
+          dry_run
+          read MemoryStorage.new(rows:)
+          write container(:fixed, file: "#{dir}/dry.txt", layout: LAYOUT)
+        end.run
+      end
+      refute File.exist?("#{dir}/dry.txt")
+      assert_includes out, 'DRY RUN'
+
+      DataPipeline.new do
+        logger PipelineHelper::QUIET
+        read MemoryStorage.new(rows:)
+        write container(:fixed, file: "#{dir}/out.txt", layout: LAYOUT)
+      end.run
+      assert_equal 3, File.readlines("#{dir}/out.txt").size
+    end
+  end
+
+  def test_options_and_lengths_required
+    assert_raises(ArgumentError) { DataPipeline::FixedStorage.new(layout: LAYOUT) }
+    assert_raises(ArgumentError) { DataPipeline::FixedStorage.new(file: 'x') }
+    layout = DataPipeline::Layout.new { record 'A', [['SATYP', 'AN']] }
+    assert_raises(DataPipeline::LayoutError) { DataPipeline::FixedStorage.new(file: 'x', layout:) }
+  end
+end

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

test/plugins/sqlite_storage_test.rb

@@ -92,4 +92,71 @@ 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
+
+  def count(file) = DataPipeline::SqliteStorage.new(database: file).then { |db| db.query(query: 'select count(*) as n from b').first['n'].tap { db.close } }
+
+  def test_write_row_in_batches_and_query_sees_pending_rows
+    with_tmpdir do |dir|
+      file = "#{dir}/b.sqlite3"
+      writer = DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true, batch_size: 2)
+      3.times { |i| writer.write_row({ 'id' => i }) }
+      assert_equal 2, count(file) # third row still pending
+      assert_equal 3, writer.query(query: 'select count(*) as n from b').first['n']
+      writer.write_row({ 'id' => 3 })
+      writer.finalize
+      assert_equal 4, count(file)
+    end
+  end
+
+  def test_close_writes_pending_rows
+    with_tmpdir do |dir|
+      writer = DataPipeline::SqliteStorage.new(database: "#{dir}/b.sqlite3", table: 'b', create: true)
+      writer.write_row({ 'id' => 1 })
+      writer.close
+      assert_equal 1, count("#{dir}/b.sqlite3")
+    end
+  end
+
+  def test_batch_size_all_is_one_transaction_for_the_stream
+    with_tmpdir do |dir|
+      file = "#{dir}/b.sqlite3"
+      DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true).tap { |w| w.write([{ 'id' => 99 }]); w.close }
+      aborted = DataPipeline::SqliteStorage.new(database: file, table: 'b', truncate: true, batch_size: :all)
+      2.times { |i| aborted.write_row({ 'id' => i }) }
+      aborted.close # without finalize: rolled back, truncate too
+      assert_equal 1, count(file)
+
+      done = DataPipeline::SqliteStorage.new(database: file, table: 'b', truncate: true, batch_size: :all)
+      2.times { |i| done.write_row({ 'id' => i }) }
+      done.finalize
+      assert_equal 2, count(file)
+    end
+  end
+
+  def test_streamed_pipeline_aborted_with_batch_size_all_writes_nothing
+    with_tmpdir do |dir|
+      file = "#{dir}/b.sqlite3"
+      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::SqliteStorage.new(database: file, table: 'b', create: true, batch_size: :all)
+        end.run
+      end
+      refute DataPipeline::SqliteStorage.new(database: file).query(query: "select name from sqlite_master where name = 'b'").any?
+    end
+  end
 end