add streaming request and save response to tmp file

Author: PNixx

CHANGELOG.md

@@ -1,3 +1,7 @@
+### Version 1.6.2 (Jan 30, 2026)
+
+* Add streaming request and save response to tmp file
+
 ### Version 1.6.1 (Jan 20, 2026)
 
 * Fix prepare rake task on ClickHouse 24.6

README.md

@@ -235,6 +235,14 @@ Action.with(ActionView.select(Arel.sql('min(date)')) => :min_date).where(Arel.sq
 #=> #<ActiveRecord::Relation [#<Action *** >]>
 ```
 
+### Streaming request
+
+```ruby
+path = Action.connection.execute_streaming(Action.where(date: Date.current), format: 'CSVWithNames')
+# Clickhouse Stream (10.3ms)  SELECT actions.* FROM actions WHERE actions.date = '2017-11-29'
+file = File.open(path)
+```
+
 
 ### Migration Data Types
 

lib/active_record/connection_adapters/clickhouse/schema_statements.rb

@@ -53,6 +53,18 @@ def execute_batch(statements, name = nil, **kwargs)
           end
         end
 
+        # @return [ClickhouseActiverecord::StreamResponse]
+        def execute_streaming(sql, name = nil, format: @response_format, settings: {})
+          with_response_format(format) do
+            log(sql, [adapter_name, 'Stream', name].compact.join(' ')) do
+              statement = Statement.new(sql, format: @response_format)
+              request(statement, settings: settings) do |response|
+                return statement.streaming_response(response)
+              end
+            end
+          end
+        end
+
         def exec_insert(sql, name = nil, _binds = [], _pk = nil, _sequence_name = nil, returning: nil)
           new_sql = sql.sub(/ (DEFAULT )?VALUES/, " VALUES")
           with_response_format(nil) { execute(new_sql, name) }
@@ -268,21 +280,22 @@ def has_default_function?(default) # :nodoc:
 
         def raw_execute(sql, settings: {}, except_params: [])
           statement = Statement.new(sql, format: @response_format)
-          statement.response = request(statement, settings: settings, except_params: except_params)
-          statement.processed_response
+          response = request(statement, settings: settings, except_params: except_params)
+          statement.processed_response(response)
         end
 
         # Make HTTP request to ClickHouse server
         # @param [ActiveRecord::ConnectionAdapters::Clickhouse::Statement] statement
         # @param [Hash] settings
         # @param [Array] except_params
         # @return [Net::HTTPResponse]
-        def request(statement, settings: {}, except_params: [])
+        def request(statement, settings: {}, except_params: [], &block)
           @lock.synchronize do
-            @connection.post("/?#{settings_params(settings, except: except_params)}",
-                             statement.formatted_sql,
-                             'Content-Type' => 'application/x-www-form-urlencoded',
-                             'User-Agent' => ClickhouseAdapter::USER_AGENT)
+            req = Net::HTTP::Post.new("/?#{settings_params(settings, except: except_params)}", {
+              'Content-Type' => 'application/x-www-form-urlencoded',
+              'User-Agent' => ClickhouseAdapter::USER_AGENT,
+            })
+            @connection.request(req, statement.formatted_sql, &block)
           end
         end
 

lib/active_record/connection_adapters/clickhouse/statement.rb

@@ -9,19 +9,27 @@ module Clickhouse
       class Statement
 
         attr_reader :format
-        attr_writer :response
 
         def initialize(sql, format:)
           @sql = sql
           @format = format
         end
 
+        # @return [String]
         def formatted_sql
           @formatted_sql ||= FormatManager.new(@sql, format: @format).apply
         end
 
-        def processed_response
-          ResponseProcessor.new(@response, @format, @sql).process
+        # @param [Net::HTTPResponse] response
+        # @return [String, Hash, Array, nil]
+        def processed_response(response)
+          ResponseProcessor.new(response, @format, @sql).process
+        end
+
+        # @param [Net::HTTPResponse] response
+        # @return [String, nil]
+        def streaming_response(response)
+          ResponseProcessor.new(response, @format, @sql).streaming_process
         end
 
       end

lib/active_record/connection_adapters/clickhouse/statement/response_processor.rb

@@ -8,14 +8,18 @@ class ResponseProcessor
 
           DB_EXCEPTION_REGEXP = /\ACode:\s+\d+\.\s+DB::Exception:/.freeze
 
+          # @param [Net::HTTPResponse] raw_response
+          # @param [String, nil] format
+          # @param [String] sql
           def initialize(raw_response, format, sql)
             @raw_response = raw_response
-            @body = raw_response.body
             @format = format
             @sql = sql
           end
 
+          # @return [String, Hash, Array, nil]
           def process
+            @body = @raw_response.body
             if success?
               process_successful_response
             else
@@ -25,12 +29,28 @@ def process
             @body
           end
 
+          # @return [String, nil]
+          def streaming_process
+            file = Tempfile.new('clickhouse-activerecord', binmode: true)
+            if success?
+              @raw_response.read_body do |chunk|
+                file.write(chunk)
+              end
+              file.close
+              file.path
+            else
+              @body = @raw_response.body
+              raise_database_error!
+            end
+          end
+
           private
 
           def success?
             @raw_response.code.to_i == 200
           end
 
+          # @return [String, Hash, Array]
           def process_successful_response
             raise_generic!(@sql) if @body.include?('DB::Exception') && @body.match?(DB_EXCEPTION_REGEXP)
 

lib/clickhouse-activerecord/version.rb

@@ -1,3 +1,3 @@
 module ClickhouseActiverecord
-  VERSION = '1.6.1'
+  VERSION = '1.6.2'
 end

spec/single/streaming_spec.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+RSpec.describe 'Streaming', :migrations do
+  class Model < ActiveRecord::Base
+    self.table_name = 'sample'
+  end
+
+  describe 'sample' do
+    before do
+      migrations_dir = File.join(FIXTURES_PATH, 'migrations', 'add_sample_data')
+      quietly { ActiveRecord::MigrationContext.new(migrations_dir).up }
+    end
+
+    it 'simple' do
+      path = Model.connection.execute_streaming('SELECT count(*) AS count FROM sample')
+      expect(path.is_a?(String)).to be_truthy
+      expect(File.read(path)).to eq("[\"count\"]\n[\"UInt64\"]\n[\"0\"]\n")
+    end
+
+    it 'JSONCompact format' do
+      path = Model.connection.execute_streaming('SELECT count(*) AS count FROM sample', format: 'JSONCompact')
+      data = JSON.parse(File.read(path))
+      expect(data['data'][0][0]).to eq('0')
+    end
+
+    it 'JSONEachRow format' do
+      path = Model.connection.execute_streaming('SELECT count(*) AS count FROM sample', format: 'JSONEachRow')
+      data = JSON.parse(File.read(path))
+      expect(data['count']).to eq('0')
+    end
+
+    it 'multiple rows JSONEachRow format' do
+      path = Model.connection.execute_streaming('SELECT * FROM generate_series(1, 1000000)', format: 'JSONEachRow')
+      lines = File.readlines(path)
+      expect(JSON.parse(lines[0])).to eq('generate_series' => '1')
+      expect(lines.size).to eq(1000000)
+    end
+
+    it 'multiple rows CSVWithNames format' do
+      path = Model.connection.execute_streaming('SELECT * FROM generate_series(1, 1000000)', format: 'CSVWithNames')
+      lines = File.readlines(path)
+      expect(JSON.parse(lines[0])).to eq('generate_series')
+      expect(JSON.parse(lines[1])).to eq(1)
+      expect(lines.size).to eq(1000001)
+    end
+
+    it 'error' do
+      expect { Model.connection.execute_streaming('error request') }.to raise_error(ActiveRecord::ActiveRecordError, include('DB::Exception'))
+    end
+  end
+end