diff --git a/CHANGELOG.md b/CHANGELOG.md index ab29c46..9657c53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ Framework `datapipeline` (Gem). Neueste Version oben. +## 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 diff --git a/Gemfile.lock b/Gemfile.lock index 522b1d0..778562e 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - datapipeline (2.1.0) + datapipeline (3.0.0) csv (~> 3.3) logger (~> 1.6) diff --git a/README.md b/README.md index 903fd79..9f6c883 100644 --- a/README.md +++ b/README.md @@ -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.0.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 `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) – 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) @@ -783,7 +783,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)` | @@ -918,8 +917,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 -einmal je Prozess (Warnung, 7.2) und einen Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem +Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung) und einen +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 @@ -1507,53 +1506,56 @@ 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) | -| `WARN deprecated: …` | veralteter Aufruf | umstellen, siehe 7.2 | +| `NoMethodError: undefined method 'transform'` (oder `chained`) | Skript für 2.x | auf `map`, `header`, `trailer` umstellen (7.1) | +| `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.1 Von 2.x auf 3.0 -2.1 ist abwärtskompatibel; bestehende Skripte laufen ohne Änderung. Anders als bisher: +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. -| Thema | 2.0 | 2.1 | +| 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 +1579,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,7 +1606,7 @@ 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). - 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`. @@ -1625,5 +1627,4 @@ Offene Punkte, Reihenfolge = empfohlene Umsetzung: 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. -6. **3.0:** `transform`, Parallelmodus und `condition:` als String entfernen. -7. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen. +6. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen. diff --git a/lib/datapipeline.rb b/lib/datapipeline.rb index 97432d0..1c9ca49 100644 --- a/lib/datapipeline.rb +++ b/lib/datapipeline.rb @@ -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 - # - # 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 (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]) } - after = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) } - body = @transformations - before - after - log_deprecations body + 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) @@ -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 + counters = Array.new(body.size, 0) + data.each { |row| push(row, body, counters, data.length) { |out| rows << out } } 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 - 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 diff --git a/lib/datapipeline/version.rb b/lib/datapipeline/version.rb index 7669791..aebf5d6 100644 --- a/lib/datapipeline/version.rb +++ b/lib/datapipeline/version.rb @@ -1,5 +1,5 @@ class DataPipeline - VERSION = '2.1.0'.freeze + VERSION = '3.0.0'.freeze # kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md class Version diff --git a/test/cli/etl_cli_test.rb b/test/cli/etl_cli_test.rb index 69f24b0..c4608cf 100644 --- a/test/cli/etl_cli_test.rb +++ b/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 diff --git a/test/core/chained_test.rb b/test/core/chained_test.rb index 820b201..11c243a 100644 --- a/test/core/chained_test.rb +++ b/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 diff --git a/test/core/compat_test.rb b/test/core/compat_test.rb deleted file mode 100644 index baeb4e4..0000000 --- a/test/core/compat_test.rb +++ /dev/null @@ -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 diff --git a/test/core/operations_test.rb b/test/core/operations_test.rb index 65878e6..9d8b0a4 100644 --- a/test/core/operations_test.rb +++ b/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 diff --git a/test/core/pipeline_test.rb b/test/core/pipeline_test.rb index 63040a3..33efa5e 100644 --- a/test/core/pipeline_test.rb +++ b/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 diff --git a/test/core/steps_test.rb b/test/core/steps_test.rb index 8c2285b..c1fff3c 100644 --- a/test/core/steps_test.rb +++ b/test/core/steps_test.rb @@ -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 diff --git a/test/core/streaming_test.rb b/test/core/streaming_test.rb index 850f7ea..54e56b9 100644 --- a/test/core/streaming_test.rb +++ b/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 diff --git a/test/test_helper.rb b/test/test_helper.rb index eb5c3d2..175d92c 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -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]