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

datapipeline 3.0.0: transform, Parallelmodus, chained und String-Bedingungen entfernt
2026-10-09 22:11:24 +02:00 · 2026-10-09 22:04:12 +02:00
19 changed files with 301 additions and 358 deletions
@@ -2,6 +2,21 @@
 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
@@ -1,7 +1,7 @@
 PATH
  remote: .
  specs:
-    datapipeline (2.1.0)
+    datapipeline (3.1.0)
      csv (~> 3.3)
      logger (~> 1.6)
@@ -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.
@@ -1,6 +1,6 @@
 # DataPipeline – ETL-Framework in Ruby
-Stand: 2026-10-09 · Gem `datapipeline` 2.1.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4
+Stand: 2026-10-09 · Gem `datapipeline` 3.1.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE))
 `DataPipeline` liest Daten aus einer Quelle, schickt jede Zeile durch eine Kette von Schritten (prüfen, filtern,
 umbauen, anreichern, abgleichen) und schreibt das Ergebnis in ein oder mehrere Ziele. Eine Pipeline ist ein kurzes
@@ -15,7 +15,7 @@ Festbreiten-Dateien und REST-APIs.
 4. [Plugins](#4-plugins) – mitgelieferte Quellen und Ziele, eigene Plugins, Abhängigkeiten
 5. [Satzarten-Layouts](#5-satzarten-layouts) – Dateien mit Kopf-, Positions- und Schlusssätzen
 6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigung, Fehlersuche
-7. [Veraltetes und Umstieg](#7-veraltetes-und-umstieg) – von 2.0 auf 2.1, `transform` und Parallelmodus
+7. [Umstieg](#7-umstieg) – auf 3.1, von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
 8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen
 9. [Ausblick](#9-ausblick)
@@ -54,10 +54,13 @@ In einer anderen Anwendung:
 ```ruby
 # Gemfile
-gem 'datapipeline', path: '../datapipeline'        # oder git: '…'
+gem 'datapipeline', '~> 3.1', git: 'https://git.wicom.li/wkh/datapipeline.git', branch: 'main'
 gem 'pg'                                            # nur die Gems der benötigten Plugins (4.10)
 ```
 `'~> 3.1'` verhindert, dass `bundle update` unbemerkt eine neue Hauptversion mit Breaking Changes holt (8.3). Für die
 Entwicklung gegen einen lokalen Checkout: `bundle config set local.datapipeline ../datapipeline`.
 ```sh
 bundle install
 bundle exec etl pipeline.rb --dry-run               # das Kommando etl kommt mit dem Gem
@@ -783,7 +786,6 @@ Tippfehler fallen also sofort auf.
 | --- | --- |
 | `streamed(flag = true)` | Zeile für Zeile statt Batch (3.3) |
 | `verbose(flag = true)` | Quelle, Ergebnis und Ziel im Batch als JSON auf `$stdout` |
 | `chained(flag = true)` | nur für `transform` nötig (7.2) |
 | `configure(opts)` | mehrere Einstellungen als Hash, z. B. `configure(streamed: true, version: '1.0')` |
 | `run` | führt die Pipeline aus und liefert `DataPipeline::Stats` |
 | `DataPipeline.new(**settings) { … }` | Einstellungen nur für diese Pipeline, z. B. `new(limit: 10)` |
@@ -827,7 +829,7 @@ Unterschieden:
 | `header` | kennt `read` und `written` | wird zuerst geschrieben, `read` und `written` sind `nil` |
 | `ctx.read` in `map`/`filter` | Anzahl Quellzeilen | `nil` |
 | `reconcile` | vor dem Schreiben | nach der letzten Zeile; die Zeilen sind schon geschrieben |
-| Postgres als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen, mit `batch_size: :all` eine für alles (4.2) |
+| Postgres, SQLite als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen, mit `batch_size: :all` eine für alles (4.2, 4.3) |
 | Postgres als Quelle | eine Abfrage | serverseitiger Cursor, je `fetch_size` Zeilen |
 `limit n` beendet im Streaming das Lesen nach n Zeilen.
@@ -918,8 +920,8 @@ Dry-Run wird die Fehlerdatei nicht geschrieben.
 | `to_s` | `read=9 written=4 rejected=4 0.097s`; mit Warnungen `warnings=n`, mit `totals` `spalte=quelle/ergebnis` |
 | `to_h` | alles als Hash, wie in der Laufhistorie |
-Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung), veraltete Aufrufe
+Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung) und einen
-einmal je Prozess (Warnung, 7.2) und einen Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem
+Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem
 Abbruch; ein Fehler im Hook wird nur geloggt und ändert weder Ergebnis noch Exit-Code.
 ```ruby
@@ -1014,9 +1016,13 @@ write container(:sqlite, database: 'tmp/kunden.sqlite3', table: 'kunden', create
 | `create: true` | Tabelle beim ersten Schreiben aus der ersten Zeile anlegen (Integer → `integer`, Float → `real`, sonst `text`) |
 | `truncate: true` | Tabelle vor dem ersten Schreiben leeren |
 | `upsert: %w[belnr]` | wie bei Postgres; mit `create: true` wird der nötige Unique-Index mit angelegt |
 | `batch_size` | Streaming: Zeilen je Transaktion (Default 1000) oder `:all` |
 Werte kommen mit SQLite-Typen zurück (Integer, Float, String, `nil`); `Date` und `Time` werden als ISO-Text
-geschrieben. `write` schreibt in einer Transaktion.
+geschrieben. `write` schreibt in einer Transaktion. Im Streaming gilt dasselbe wie bei Postgres (4.2): `write_row`
 schreibt je `batch_size` Zeilen in einer Transaktion, `finalize` und `close` den Rest; mit `batch_size: :all` ist der
 ganze Stream eine Transaktion, die nach einem Abbruch zurückgerollt wird. Eine Abfrage (`query`, `lookup`) auf
 demselben Storage schreibt noch gesammelte Zeilen vorher, sie sieht also alles, was schon übergeben wurde.
 ### 4.4 CSV
@@ -1507,53 +1513,62 @@ gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0
 | `LayoutError: TEXT: "…" longer than 30` | Wert passt nicht in das Feld einer Festbreiten-Datei | vorher `validate LAYOUT` mit `on_error`, oder den Wert kürzen |
 | `Encoding::UndefinedConversionError: U+20AC from UTF-8 to ISO-8859-1` | Zeichen gibt es im Zielzeichensatz nicht | `Windows-1252` statt ISO-8859-1, oder `replace: '?'` |
 | `connection :x not in config/connections.yml (a, b)` | Name falsch oder Datei nicht gefunden | Namen prüfen; Pfad relativ zum Arbeitsverzeichnis, sonst `connections_file` setzen |
-| `streamed: several transformations need chained (or map/filter)` | mehrere `transform` im Streaming | `map` verwenden (7.2) |
+| `NoMethodError: undefined method 'transform'` (oder `chained`) | Skript für 2.x | auf `map`, `header`, `trailer` umstellen (7.1) |
-| `WARN deprecated: …` | veralteter Aufruf | umstellen, siehe 7.2 |
+| `map: condition must be a lambda ->(row) { ... }` | `condition:` als String aus 2.x | `condition: ->(row) { … }` (7.1) |
 Mehr sehen: `--verbose` (Log-Level Debug), `verbose` in der Pipeline (Daten als JSON), `--dry-run --limit=20` für einen
 schnellen Blick auf das Ergebnis.
 ---
-## 7. Veraltetes und Umstieg
+## 7. Umstieg
-### 7.1 Von 2.0 auf 2.1
+### 7.0 Von 3.0 auf 3.1
-2.1 ist abwärtskompatibel; bestehende Skripte laufen ohne Änderung. Anders als bisher:
+3.1 ist abwärtskompatibel. Anders als in 3.0: SQLite schreibt im Streaming (`write_row`) nicht mehr jede Zeile
 einzeln, sondern je 1000 Zeilen in einer Transaktion (4.3). Wer während eines Laufs mit einer zweiten Verbindung
 auf die Zieldatei schaut, sieht die Zeilen also paketweise. Das Gem steht jetzt unter der BSD-3-Clause-Lizenz.
-| Thema | 2.0 | 2.1 |
+### 7.1 Von 2.x auf 3.0
 3.0 entfernt das alte Ausführungsmodell: `transform`, den Parallelmodus, `chained` und `condition:` als String
 (`eval`). Alle Schritte laufen jetzt verkettet, jeder bekommt die Ausgabe des vorigen. Alles andere ist unverändert;
 ein Skript, das unter 2.1 ohne `WARN deprecated` lief, läuft auch unter 3.0.
 | 2.x | 3.0 |
 | --- | --- |
 | `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
 | `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
 | `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }`, Anzahl als `ctx.written` |
 | `transform(condition: "row['a'] == 1") { … }` | `map(condition: ->(row) { row['a'] == 1 }) { … }` |
 | `chained`, `configure(chained: true)` | weglassen |
 Worauf beim Umstellen zu achten ist:
 - **Parallelmodus:** Ohne `map`, `filter`, `validate` und `chained` bekam unter 2.x jeder `transform` eine eigene
  Kopie der Quellzeilen, und die Ergebnisse wurden hintereinander angehängt. Zwei `transform` auf 2 Quellzeilen
  ergaben 4 Zeilen, zwei `map` ergeben 2 Zeilen mit beiden Änderungen. Wer wirklich jede Zeile mehrfach ausgeben will,
  gibt in einem `map` ein Array zurück: `map { |r| [mit_b(r), mit_c(r)] }`.
 - **Kopfsatz:** `transform(processing: :prepend)` lief vor den Daten, `header` läuft im Batch erst danach (steht aber
  vorne). Das zählt nur, wenn der Block etwas anderes tut, als eine Zeile zu liefern.
 - **Schlusssatz:** `processing: :append` bekam im Parallelmodus die Anzahl der Quellzeilen; `ctx.written` ist die
  Anzahl der Datenzeilen nach allen Schritten, `ctx.read` die der Quellzeilen.
 - **Bedingung:** Ein String in `condition:` bricht schon beim Aufbau der Pipeline mit `ArgumentError` ab.
 Prüfen lässt sich die Umstellung mit `etl run skript.rb --dry-run` vorher und nachher: Die Zusammenfassung und die
 ersten Zeilen sollten gleich sein.
 ### 7.2 Von 2.0 auf 2.1
 2.1 war abwärtskompatibel. Anders als in 2.0:
 | Thema | 2.0 | ab 2.1 |
 | --- | --- | --- |
 | Postgres, `write_row` im Streaming | jede Zeile sofort, ohne Transaktion | je 1000 Zeilen in einer Transaktion; der Rest in `finalize`/`close`, auch nach einem Abbruch |
 | CSV lesen mit UTF-8-BOM | BOM steht im ersten Spaltennamen | BOM wird übersprungen |
 | `Stats#to_s` | – | mit `totals` zusätzlich `spalte=quelle/ergebnis` |
 | `transform`, Parallelmodus, `condition:` als String | ohne Hinweis | `WARN deprecated: …` einmal je Prozess |
 | `etl` ohne `name` | – | `stats.name` = Skriptname |
 ### 7.2 `transform` und Parallelmodus
 `transform`, der Parallelmodus und `condition:` als String (`eval`) sind veraltet und entfallen in 3.0. Ab 3.0
 laufen alle Schritte verkettet, wie heute schon, sobald `map`, `filter` oder `validate` vorkommt.
 | Alt | Neu |
 | --- | --- |
 | `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
 | `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
 | `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }` mit `ctx.written` |
 | `transform(condition: "row['a'] == 1")` | `map(condition: ->(row) { row['a'] == 1 })` |
 | mehrere `transform` ohne `chained` (Parallelmodus) | mehrere `map`; wer wirklich jede Zeile mehrfach ausgeben will, gibt in einem `map` ein Array zurück |
 Bis dahin gilt für alte Skripte:
 - **Parallelmodus** (Default ohne `map`/`filter`/`validate` und ohne `chained`): Jeder `transform` bekommt eine eigene
  Kopie der Quellzeilen, die Ergebnisse aller Schritte werden hintereinander angehängt. Beispiel mit 2 Quellzeilen und
  `transform { |r| r.merge('b' => 1) }` + `transform { |r| r.merge('c' => 2) }`: parallel 4 Zeilen (2× mit `b`, dann
  2× mit `c`), verkettet 2 Zeilen mit `b` und `c`. Im Streaming sind mehrere `transform` nur verkettet erlaubt.
 - **`transform(processing: :prepend)`** wird mit `(nil, 0)` vor den Daten aufgerufen, **`:append`** mit `(nil, n)`
  nach den Daten (parallel n = Quellzeilen, verkettet n = Datenzeilen).
 - **`condition:` als String** wird mit `eval` ausgewertet, die Variable `row` ist sichtbar.
 `test/core/compat_test.rb` hält die Ausgabe des alten Modells fest, damit es sich bis 3.0 nicht ändert.
 ---
 ## 8. Entwicklung
@@ -1577,13 +1592,13 @@ Bis dahin gilt für alte Skripte:
 ### 8.2 Tests
-`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 205 Fälle. Einzelne Datei:
+`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 190 Fälle. Einzelne Datei:
 `bundle exec rake test TEST=test/core/steps_test.rb`.
 | Ordner | Inhalt |
 | --- | --- |
 | `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST gegen einen lokalen Testserver; Laden der Gems erst bei Nutzung |
-| `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen; `compat_test.rb` hält das alte `transform`-Modell fest |
+| `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen |
 | `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Exit-Codes |
 | `test/examples` | die Beispiele aus Kapitel 2: Code in der Doku = Datei, Lauf und Ergebnis; REST gegen die Demo-API |
@@ -1604,10 +1619,14 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als
 ### 8.3 Gem und Versionen
 - Version in [lib/datapipeline/version.rb](lib/datapipeline/version.rb), Änderungen in [CHANGELOG.md](CHANGELOG.md)
-  (neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (3.0 entfernt `transform`).
+  (neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (wie 3.0, 7.1).
 - Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.1.0 -m "datapipeline 3.1.0" && git push origin v3.1.0`).
  Anwendungen binden das Gem mit einer Versionsgrenze ein (`'~> 3.1'`, 1.2) oder fest auf ein Tag
  (`tag: 'v3.1.0'`; dann geht `local.datapipeline` nicht).
 - Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`,
  `caxlsx`, `http`) und `test`; `Gemfile.lock` ist versioniert. Ohne Plugin-Gems: `bundle config set without plugins`.
 - Gem bauen: `gem build datapipeline.gemspec`.
 - Lizenz: BSD-3-Clause, siehe [LICENSE](LICENSE); die Datei ist Teil des Gems.
 - Beispiele in Kapitel 2 werden mit einem Marker `<!-- beispiel: examples/datei.rb -->` vor dem Codeblock eingebunden;
  der Doku-Test vergleicht Block und Datei, Änderungen also immer in beiden machen.
@@ -1621,9 +1640,6 @@ Offene Punkte, Reihenfolge = empfohlene Umsetzung:
   schreiben), `etl list` (Pipelines, Wasserzeichen, letzter Lauf aus der Historie), `--json` für die Stats.
 2. **Qualität:** CI auf git.wicom.li mit `rake test` und Postgres-Service, RuboCop, `DataPipeline::Testing` mit
   `MemoryStorage` als öffentliche Testhilfe für Anwendungen.
-3. **SQLite im Streaming:** `write_row` wie bei Postgres in Paketen schreiben.
+3. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`,
 4. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`,
   Maskierung von `&`, `<`, `>`, Incoming Webhooks.
-5. **Weitere Plugins:** SFTP-Ablage.
+4. **Weitere Plugins:** SFTP-Ablage.
 6. **3.0:** `transform`, Parallelmodus und `condition:` als String entfernen.
 7. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen.
@@ -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']
@@ -64,28 +64,15 @@ class DataPipeline
  # raised when a reconcile check fails
  class ReconcileError < StandardError; end
  EDGE_BEFORE = %i[prepend header].freeze
  EDGE_AFTER  = %i[append trailer].freeze
  CHAINING    = %i[map filter validate].freeze
  ON_ERROR    = %i[abort skip collect].freeze
  MAX_ERRORS  = 100
  # removed in 3.0, logged once per process as a warning
  DEPRECATIONS = {
    transform: 'transform is deprecated and will be removed in 3.0: use map/filter, ' \
               'header/trailer instead of processing: :prepend/:append',
    parallel: 'parallel mode (several transform steps without chained) is deprecated and will be removed in 3.0: ' \
              'set chained or use map',
    condition: 'condition: as string (eval) is deprecated and will be removed in 3.0: use a lambda ->(row) { ... }'
  }.freeze
  # 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,
@@ -100,7 +87,6 @@ class DataPipeline
  SETTINGS_LOCK = Mutex.new
  @defaults = DEFAULT_SETTINGS
  @plugins = {}
  @deprecations_logged = []
  #
  # defaults for every following DataPipeline.new, e.g. from bin/etl
@@ -166,18 +152,6 @@ class DataPipeline
    @runs ||= []
  end
  # true the first time a deprecation is reported in this process
  def self.deprecation_once key
    SETTINGS_LOCK.synchronize do
      next false if @deprecations_logged.include?(key)
      @deprecations_logged << key
      true
    end
  end
  def self.reset_deprecations! = SETTINGS_LOCK.synchronize { @deprecations_logged = [] }
  # see DataPipeline::Connections
  def self.connections_file = Connections.file
  def self.connections_file=(path)
@@ -204,7 +178,7 @@ class DataPipeline
  # 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')
@@ -232,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
@@ -343,29 +312,19 @@ class DataPipeline
    klass.new(**options, &block)
  end
-  #
+  # map { |row, ctx| ... }    -> hash, array of hashes or nil
-  # transform(options) { |row, index| ... }
+  # condition: ->(row) { ... } on map and filter: the step only for matching rows, the others are dropped
  #   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
  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 (turns chaining on)
+  # 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
@@ -419,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)
  #
@@ -427,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 = {}
@@ -450,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 }
@@ -472,10 +431,9 @@ class DataPipeline
    (@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]) }
+    before = @steps.select { |t| t[:kind] == :header }
-    after  = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) }
+    after  = @steps.select { |t| t[:kind] == :trailer }
-    body   = @transformations - before - after
+    body   = @steps - before - after
    log_deprecations body
    @settings[:streamed] ? run_streamed(before, body, after) : run_batch(before, body, after)
@@ -493,8 +451,11 @@ class DataPipeline
  def add_step kind, options, &block
    raise ArgumentError, "#{kind}: block missing" unless block
    if options.key?(:condition) && !options[:condition].respond_to?(:call)
      raise ArgumentError, "#{kind}: condition must be a lambda ->(row) { ... }"
    end
-    @transformations << { kind:, block:, options: }
+    @steps << { kind:, block:, options: }
  end
  def value_of row, key
@@ -511,21 +472,6 @@ class DataPipeline
    raise ArgumentError, "#{key}: #{value.inspect} is not #{type}"
  end
  # each deprecated feature once per process, with the logger of the pipeline
  def log_deprecations body
    used = []
    used << :transform if @transformations.any? { |t| %i[prepend normal append].include?(t[:kind]) }
    used << :parallel if !chained? && body.size > 1
    used << :condition if @transformations.any? { |t| t[:options].is_a?(Hash) && t[:options][:condition].is_a?(String) }
    used.each { |key| @logger.warn "deprecated: #{DEPRECATIONS[key]}" if DataPipeline.deprecation_once(key) }
  end
  # without any transformation the source is copied unchanged
  def chained?
    @settings[:chained] || @transformations.empty? ||
      @transformations.any? { |t| CHAINING.include?(t[:kind]) }
  end
  def write_row row
    @targets.each { |t| t[:storage].write_row(row) if t[:only].nil? || t[:only].call(row) }
  end
@@ -607,33 +553,21 @@ class DataPipeline
    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
    @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)
    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] }
+    before.each { |t| add_rows new_data, t[:block].call(ctx) }
    new_data.concat rows
-    after.each { |t| add_rows new_data, edge(t, ctx, count) }
+    after.each { |t| add_rows new_data, t[:block].call(ctx) }
    @stats.written = rows.length
    dump 'Target#data', new_data
    output(new_data) do |all|
      @targets.each { |t| t[:storage].write(t[:only] ? all.select(&t[:only]) : all) }
@@ -642,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)
@@ -673,7 +603,7 @@ class DataPipeline
    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
@@ -688,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)
@@ -715,7 +640,7 @@ class DataPipeline
  def step_name step
    name = step[:options].is_a?(Hash) && step[:options][:name] || step[:kind]
-    "#{name}##{@transformations.index(step) + 1}"
+    "#{name}##{@steps.index(step) + 1}"
  end
  # rows produced by one step for one input row
@@ -743,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
+  # condition: ->(row) { ... } on map and filter
    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
  def condition? step, row
-    condition = step[:options].fetch(:condition, true)
+    condition = step[:options][:condition]
-    return true if condition == true
+    condition.nil? || condition.call(row)
    return condition.call(row) if condition.respond_to?(:call)
    eval(condition)
  end
  # a hash is one row, an array are several rows, nil is no row
@@ -48,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?
@@ -12,6 +12,8 @@ class DataPipeline
  #   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.
@@ -33,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?
@@ -58,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
@@ -81,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]
@@ -1,5 +1,5 @@
 class DataPipeline
-  VERSION = '2.1.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
@@ -14,8 +14,8 @@ class EtlCliTest < PipelineTest
    "#{dir}/#{name}"
  end
-  # like import.*.rb: own OptionParser for --from/--till, transform api
+  # scripts with an own OptionParser for --from/--till
-  def test_legacy_script_gets_from_till_and_limit
+  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' } }
+          header { { 'h' => 'HEAD' } }
-          transform { |r, i| r.merge('i' => i) }
+          map { |r, ctx| r.merge('i' => ctx.index) }
          write container(:csv, file: 'out.csv', csv: { headers: false, col_sep: '|' })
        end.run
      RUBY
@@ -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
+  def test_each_step_gets_the_previous_output
-    out = output(ROWS) { chained; transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
+    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
+  def test_several_headers_and_trailers_in_registration_order
-    out = output(ROWS) { header { { 'h' => 1 } }; transform(processing: :prepend) { { 'p' => 1 } }; map { |r| r } }
+    out = output(ROWS) { header { { 'h' => 1 } }; trailer { { 't' => 1 } }; header { { 'h' => 2 } }; trailer { { 't' => 2 } } }
-    assert_equal [{ 'h' => 1 }, { 'p' => 1 }], out.first(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
+  def test_condition_must_be_a_lambda
-    assert_equal 6, output(ROWS) { chained false; transform { |r| r }; transform { |r| r } }.size
+    assert_includes assert_raises(ArgumentError) { output(ROWS) { map(condition: "row['a'] == 1") { |r| r } } }.message, 'lambda'
  end
  def test_csv_end_to_end
@@ -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
@@ -1,4 +1,5 @@
 require_relative '../test_helper'
 require 'time'
 # connections file, totals with reconcile, run history
 class ConnectionsTotalsHistoryTest < PipelineTest
@@ -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
+  def test_skip_in_map
-    target, stats = run_pipeline(ROWS) { on_error :skip; transform { |r| r['id'] == '1' ? raise('x') : r } }
+    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
+  def test_validate_gets_the_output_of_the_previous_step
-    target, = run_pipeline(ROWS) { on_error :skip; transform { |r| r.merge('x' => 1) }; validate RULES }
+    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
@@ -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
+  def test_removed_dsl_of_version_2
-    out = output(ROWS) { transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
+    %i[transform chained].each { |name| assert_raises(NoMethodError) { DataPipeline.new { send(name) { |r| r } } } }
    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 } }
  end
  def test_copy_without_steps
@@ -1,6 +1,6 @@
 require_relative '../test_helper'
-# rename, select, cast, default and the deprecation warnings
+# 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
@@ -44,33 +44,4 @@ class StepsTest < PipelineTest
    end
    assert_equal [{ 'id' => '1', 'betrag' => 12.5 }, { 'id' => '2', 'betrag' => 0.0 }], rows
  end
  def deprecation_log(&block)
    log = StringIO.new
    DataPipeline.new do
      logger Logger.new(log)
      read MemoryStorage.new(rows: [{ 'a' => 1 }])
      write MemoryStorage.new
      instance_eval(&block)
    end.run
    log.string.lines.grep(/deprecated/)
  end
  def test_deprecations_once_per_process
    lines = deprecation_log do
      transform(condition: 'true') { |r| r }
      transform { |r| r }
    end
    assert_equal 3, lines.size
    assert lines.any? { |l| l.include?('transform is deprecated') }
    assert lines.any? { |l| l.include?('parallel mode') }
    assert lines.any? { |l| l.include?('condition: as string') }
    assert_empty deprecation_log { transform { |r| r } }
  end
  def test_no_deprecation_for_map_filter_and_single_transform_in_chained_mode
    assert_empty deprecation_log { map { |r| r }; filter(condition: ->(_r) { true }) { true } }
    lines = deprecation_log { chained; transform { |r| r }; transform { |r| r } }
    assert_equal 1, lines.size # only transform, no parallel mode
  end
 end
@@ -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
+      map { |r| [r, r] }
-      transform(processing: :prepend) { |_, n| { 'p' => n } }
+      map { |r, ctx| r.merge('i' => ctx.index) }
-      transform { |r| [r, r] }
+      trailer { |ctx| { 'n' => ctx.written } }
      transform(processing: :append) { |_, n| { 'n' => n } }
    end
    assert_equal({ 'p' => 0 }, out.first)
    assert_equal({ 'n' => 6 }, out.last)
-    assert_equal 8, out.size
+    assert_equal (0...6).to_a, out.first(6).map { |r| r['i'] }
  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'
  end
  def test_stream_condition_and_copy
@@ -103,4 +103,60 @@ class SqliteStorageTest < PipelineTest
      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
@@ -55,7 +55,6 @@ module PipelineHelper
    DataPipeline.reset_settings!
    DataPipeline.runs.clear
    DataPipeline.params.clear
    DataPipeline.reset_deprecations!
  end
  # runs a pipeline from rows into a MemoryStorage, returns [target, stats]
Author	SHA1	Message	Date
wkh	ee47baa22c	datapipeline 3.1.0: SQLite schreibt im Streaming in Paketen, Lizenz BSD-3-Clause	2026-10-09 22:11:24 +02:00
wkh	a8b8381293	datapipeline 3.0.0: transform, Parallelmodus, chained und String-Bedingungen entfernt	2026-10-09 22:04:12 +02:00