fix(duckdb): cap read_csv buffer to prevent OOM under low memory

DuckDB sizes its read_csv buffer as 16 × max_line_size and allocates it
eagerly. The 256MB max_line_size raise (issue #787) thus demanded a
4 GiB block, causing OOM errors on hosts with a smaller memory_limit.

Extract the read_csv from-expression into a new ReadCsvExpr helper that
sets an explicit buffer_size equal to max_line_size, capping the
allocation at the minimum needed to fit one line.

Also add support for the memory_limit DuckDB session property, plus a
regression unit test and replication config exercising a 5MB line under
a 1GB memory limit.

Author: flarco

core/dbio/iop/duckdb.go

@@ -463,6 +463,10 @@ func (duck *DuckDb) getSessionSettingsSQL() (sql string) {
 	}
 
 	sql += fmt.Sprintf("SET http_timeout = %d;", httpTimeout)
+
+	if limit := duck.GetProp("memory_limit"); limit != "" {
+		sql += fmt.Sprintf("SET memory_limit = '%s';", limit)
+	}
 	return
 }
 
@@ -1934,10 +1938,8 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP
 				// Create a pipe to stream data through
 				pipeR, pipeW := io.Pipe()
 
-				maxLineSize := duck.MaxLineSize(batchR.Columns)
-
 				// can use this as a from table
-				fromExpr := g.F(`read_csv('%s', delim=',', header=True, columns=%s, max_line_size=%d, parallel=false, quote='"', escape='"', nullstr='\N', auto_detect=false)`, httpURL, duck.GenerateCsvColumns(batchR.Columns), maxLineSize)
+				fromExpr := duck.ReadCsvExpr(httpURL, batchR.Columns)
 
 				select {
 				case streamPartChn <- HttpStreamPart{
@@ -2029,6 +2031,16 @@ func (duck *DuckDb) MaxLineSize(columns Columns) int {
 	return DuckDbDefaultMaxLineSize
 }
 
+// ReadCsvExpr returns the read_csv from-expression for the CSV bridge.
+// DuckDB sizes its read buffer as 16 × max_line_size and allocates it
+// eagerly, so the 256MB raise would demand a 4 GiB block and OOM hosts
+// with a smaller memory_limit. An explicit buffer_size caps the
+// allocation at max_line_size, the minimum that fits one line.
+func (duck *DuckDb) ReadCsvExpr(uri string, columns Columns) string {
+	maxLineSize := duck.MaxLineSize(columns)
+	return g.F(`read_csv('%s', delim=',', header=True, columns=%s, max_line_size=%d, buffer_size=%d, parallel=false, quote='"', escape='"', nullstr='\N', auto_detect=false)`, uri, duck.GenerateCsvColumns(columns), maxLineSize, maxLineSize)
+}
+
 func (duck *DuckDb) GenerateCsvColumns(columns Columns) (colStr string) {
 	// {'FlightDate': 'DATE', 'UniqueCarrier': 'VARCHAR', 'OriginCityName': 'VARCHAR', 'DestCityName': 'VARCHAR'}
 

core/dbio/iop/duckdb_test.go

@@ -3,9 +3,11 @@ package iop
 import (
 	"context"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/flarco/g"
 	"github.com/slingdata-io/sling-cli/core/dbio"
 	"github.com/spf13/cast"
 	"github.com/stretchr/testify/assert"
@@ -636,6 +638,49 @@ func TestDuckDbMaxLineSize(t *testing.T) {
 	})
 }
 
+// regression guard for the v1.5.25 OOM: DuckDB sizes its read_csv buffer as
+// 16 × max_line_size and allocates it eagerly. The 256MB raise thus demands a
+// 4 GiB block, which fails on hosts with memory_limit below ~4 GiB. The bridge
+// expression must cap the buffer so it works under small memory limits.
+func TestDuckDbReadCsvExprLowMemory(t *testing.T) {
+	cols := NewColumnsFromFields("id", "payload")
+	cols[0].Type = IntegerType
+	cols[1].Type = TextType
+
+	// a ~5MB line exceeds the 2 MB default limit, so this also guards the
+	// #787 raise: the line must still load with the capped buffer_size
+	csvPath := os.TempDir() + "/duckdb_low_mem_test.csv"
+	payload := strings.Repeat("x", 5*1024*1024)
+	err := os.WriteFile(csvPath, []byte("id,payload\n1,"+payload+"\n"), 0644)
+	if !assert.NoError(t, err) {
+		return
+	}
+	defer os.Remove(csvPath)
+
+	duck := NewDuckDb(context.Background(), "memory_limit=1GB")
+	err = duck.Open()
+	if !assert.NoError(t, err) {
+		return
+	}
+	defer duck.Close()
+
+	expr := duck.ReadCsvExpr(csvPath, cols)
+	assert.Contains(t, expr, cast.ToString(DuckDbLargeMaxLineSize))
+
+	// same COPY shape the fabric/parquet bridge submits in production
+	parquetPath := os.TempDir() + "/duckdb_low_mem_test.parquet"
+	defer os.Remove(parquetPath)
+	_, err = duck.Exec(g.F("COPY (select * from %s) TO '%s' (format 'parquet', overwrite true)", expr, parquetPath))
+	if !assert.NoError(t, err) {
+		return
+	}
+
+	data, err := duck.Query(g.F("select count(*) cnt from read_parquet('%s')", parquetPath))
+	if assert.NoError(t, err) && assert.Equal(t, 1, len(data.Rows)) {
+		assert.EqualValues(t, 1, cast.ToInt(data.Rows[0][0]))
+	}
+}
+
 // regression guard for issue #770: http_timeout must be raised on every DuckDB
 // session, not only when an S3/httpfs secret registers the extension.
 func TestDuckDbHttpTimeout(t *testing.T) {

tests/replications/r.103.duckdb_low_mem_buffer.yaml

@@ -0,0 +1,33 @@
+# DuckDB sizes its read_csv buffer as 16 x max_line_size and allocates it
+# eagerly. The 256MB max_line_size raise must not demand a 4 GiB block when
+# memory_limit is small; the bridge caps the buffer with buffer_size.
+# One row holds a 5MB payload, which also guards the #787 large-line raise.
+source: LOCAL
+target: DUCK_LOW_MEM
+
+defaults:
+  mode: full-refresh
+
+hooks:
+  end:
+    - type: check
+      check: execution.status.error == 0
+      on_failure: break
+
+    - type: query
+      connection: '{target.name}'
+      query: select count(*) cnt, max(length(payload)) max_len from main.big_rows
+      into: result
+
+    - type: check
+      check: int_parse(store.result[0].cnt) == 2
+
+    - type: check
+      check: int_parse(store.result[0].max_len) == 5242880
+
+    - type: log
+      message: 'SUCCESS: 5MB row loaded into duckdb with 1GB memory_limit'
+
+streams:
+  "file://temp/duckdb_low_mem/big.csv":
+    object: main.big_rows

tests/suite.cli.yaml

@@ -2676,3 +2676,19 @@
     sling run -d --src-stream 'http://127.0.0.1:18937/' --tgt-object 'file://temp/merge_latency/out.csv'
   output_contains:
     - 'execution succeeded'
+
+# DuckDB allocates its read_csv buffer as 16 x max_line_size, so the 256MB
+# raise (#787) demanded a 4 GiB block and OOMed hosts with memory_limit
+# below ~4 GiB. The bridge must cap the buffer via buffer_size.
+# Unit guard: TestDuckDbReadCsvExprLowMemory in core/dbio/iop/duckdb_test.go
+- id: 321
+  name: 'duckdb csv bridge loads large rows under low memory_limit'
+  env:
+    DUCK_LOW_MEM: '{type: duckdb, instance: "temp/duckdb_low_mem/test.duckdb", memory_limit: 1GB}'
+  run: |
+    mkdir -p temp/duckdb_low_mem
+    rm -f temp/duckdb_low_mem/test.duckdb temp/duckdb_low_mem/big.csv
+    python3 -c 'import csv; w=csv.writer(open("temp/duckdb_low_mem/big.csv","w")); w.writerow(["id","payload"]); w.writerow([1,"x"*5242880]); w.writerow([2,"small"])'
+    sling run -d -r tests/replications/r.103.duckdb_low_mem_buffer.yaml
+  output_contains:
+    - 'SUCCESS: 5MB row loaded into duckdb with 1GB memory_limit'