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

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