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

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 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

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    datapipeline (3.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.

README.md

@@ -1,6 +1,6 @@
 # DataPipeline – ETL-Framework in Ruby
 
-Stand: 2026-10-09 · Gem `datapipeline` 3.0.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. [Umstieg](#7-umstieg) – von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
+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
@@ -826,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.
@@ -1013,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
 
@@ -1516,6 +1523,12 @@ schnellen Blick auf das Ergebnis.
 
 ## 7. Umstieg
 
+### 7.0 Von 3.0 auf 3.1
+
+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.
+
 ### 7.1 Von 2.x auf 3.0
 
 3.0 entfernt das alte Ausführungsmodell: `transform`, den Parallelmodus, `chained` und `condition:` als String
@@ -1607,9 +1620,13 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als
 
 - 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 (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.
 
@@ -1623,8 +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.
-4. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`,
+3. **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. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen.
+4. **Weitere Plugins:** SFTP-Ablage.

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']

lib/datapipeline/storage/postgres.rb

@@ -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?
 

lib/datapipeline/storage/sqlite.rb

@@ -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]
 
-      prepare_table @options[:table], row
-      insert @options[:table], row
+      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]

lib/datapipeline/version.rb

@@ -1,5 +1,5 @@
 class DataPipeline
-  VERSION = '3.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/core/connections_totals_history_test.rb

@@ -1,4 +1,5 @@
 require_relative '../test_helper'
+require 'time'
 
 # connections file, totals with reconcile, run history
 class ConnectionsTotalsHistoryTest < PipelineTest

test/plugins/sqlite_storage_test.rb

@@ -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