Merge pull request #58 from PostHog/fix/copy-multiline-csv

Fix COPY FROM STDIN to handle multi-line CSV fields

Author: EDsCODE

server/conn.go

@@ -836,11 +836,28 @@ func (c *clientConn) handleCopyIn(query, upperQuery string) error {
 			allData.Write(body)
 
 		case msgCopyDone:
-			// Process all data
-			lines := strings.Split(allData.String(), "\n")
-			for _, line := range lines {
-				line = strings.TrimSpace(line)
-				if line == "" {
+			// Process all data using proper CSV reader
+			// This correctly handles multi-line quoted fields (e.g., JSON with embedded newlines)
+			csvReader := csv.NewReader(&allData)
+			csvReader.Comma = rune(delimiter[0])
+			csvReader.LazyQuotes = true
+			csvReader.FieldsPerRecord = -1 // Allow variable number of fields
+
+			for {
+				values, err := csvReader.Read()
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					c.sendError("ERROR", "22P02", fmt.Sprintf("invalid CSV input: %v", err))
+					c.setTxError()
+					writeReadyForQuery(c.writer, c.txStatus)
+					c.writer.Flush()
+					return nil
+				}
+
+				// Skip empty rows
+				if len(values) == 0 || (len(values) == 1 && values[0] == "") {
 					continue
 				}
 
@@ -850,12 +867,6 @@ func (c *clientConn) handleCopyIn(query, upperQuery string) error {
 					continue
 				}
 
-				// Parse values and insert
-				values := c.parseCopyLine(line, delimiter)
-				if len(values) == 0 {
-					continue
-				}
-
 				// Build INSERT statement
 				placeholders := make([]string, len(values))
 				args := make([]interface{}, len(values))

server/conn_test.go

@@ -1,6 +1,9 @@
 package server
 
 import (
+	"encoding/csv"
+	"io"
+	"strings"
 	"testing"
 )
 
@@ -642,3 +645,90 @@ func TestParseCopyLine(t *testing.T) {
 		})
 	}
 }
+
+func TestParseMultiLineCSV(t *testing.T) {
+	// This tests the fix for COPY FROM STDIN with multi-line quoted fields.
+	// Previously, we split by newlines first then parsed each line, which broke
+	// when quoted fields contained embedded newlines (e.g., JSON with formatting).
+	// Now we use csv.Reader on the entire buffer which handles this correctly.
+
+	tests := []struct {
+		name      string
+		input     string
+		delimiter string
+		expected  [][]string
+	}{
+		{
+			name:      "simple rows no newlines",
+			input:     "a,b,c\n1,2,3\n",
+			delimiter: ",",
+			expected:  [][]string{{"a", "b", "c"}, {"1", "2", "3"}},
+		},
+		{
+			name:      "quoted field with embedded newline",
+			input:     "id,json,status\n1,\"{\"\"key\"\":\n\"\"value\"\"}\",active\n",
+			delimiter: ",",
+			expected:  [][]string{{"id", "json", "status"}, {"1", "{\"key\":\n\"value\"}", "active"}},
+		},
+		{
+			name:      "multiple fields with newlines",
+			input:     "a,\"line1\nline2\",b\nc,\"x\ny\nz\",d\n",
+			delimiter: ",",
+			expected:  [][]string{{"a", "line1\nline2", "b"}, {"c", "x\ny\nz", "d"}},
+		},
+		{
+			name:      "JSON metadata like Fivetran sends",
+			input:     "customer_id,metadata,type\ncust_123,\"{\"\"subscription\"\":\n  \"\"active\"\",\n  \"\"plan\"\": \"\"pro\"\"}\",invoice\n",
+			delimiter: ",",
+			expected:  [][]string{{"customer_id", "metadata", "type"}, {"cust_123", "{\"subscription\":\n  \"active\",\n  \"plan\": \"pro\"}", "invoice"}},
+		},
+		{
+			name:      "13 columns with multiline JSON in middle",
+			input:     "c1,c2,c3,c4,c5,c6,c7,c8,c9,c10,c11,c12,c13\nv1,v2,v3,v4,v5,v6,\"{\"\"a\"\":\n\"\"b\"\"}\",v8,v9,v10,v11,v12,v13\n",
+			delimiter: ",",
+			expected:  [][]string{{"c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9", "c10", "c11", "c12", "c13"}, {"v1", "v2", "v3", "v4", "v5", "v6", "{\"a\":\n\"b\"}", "v8", "v9", "v10", "v11", "v12", "v13"}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Simulate what handleCopyIn does: use csv.Reader on entire buffer
+			reader := csv.NewReader(strings.NewReader(tt.input))
+			reader.Comma = rune(tt.delimiter[0])
+			reader.LazyQuotes = true
+			reader.FieldsPerRecord = -1
+
+			var rows [][]string
+			for {
+				record, err := reader.Read()
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					t.Fatalf("csv.Read() error: %v", err)
+				}
+				rows = append(rows, record)
+			}
+
+			if len(rows) != len(tt.expected) {
+				t.Errorf("got %d rows, want %d\nGot: %v\nWant: %v",
+					len(rows), len(tt.expected), rows, tt.expected)
+				return
+			}
+
+			for i, row := range rows {
+				if len(row) != len(tt.expected[i]) {
+					t.Errorf("row %d: got %d columns, want %d\nGot: %v\nWant: %v",
+						i, len(row), len(tt.expected[i]), row, tt.expected[i])
+					continue
+				}
+				for j, val := range row {
+					if val != tt.expected[i][j] {
+						t.Errorf("row %d col %d: got %q, want %q",
+							i, j, val, tt.expected[i][j])
+					}
+				}
+			}
+		})
+	}
+}