From b883824cc06d39a8bdb0956a0e2a51ce323bd573 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Mon, 10 Aug 2026 13:23:20 -0300 Subject: [PATCH 01/16] fix: preserve timestamp timezone through arrow conversion Store the source timezone in column metadata and reconstruct the arrow `TimestampType` with that zone instead of relying on the fixed-width globals (which are hardcoded to UTC). On read, values are returned in the schema's zone rather than forced to UTC, so the original offset survives the round trip into targets such as Snowflake `TIMESTAMP_TZ`. - Default timestamp unit is now microsecond (iceberg-compatible) - Zone-less `datetime` columns stay zone-less instead of being labelled UTC - Append path shifts zone-less wall-clock values onto UTC so digits are preserved without re-expressing the instant --- core/dbio/iop/arrow.go | 69 ++++++++++++++++---- core/dbio/iop/arrow_test.go | 124 ++++++++++++++++++++++++++++++++++++ 2 files changed, 182 insertions(+), 11 deletions(-) diff --git a/core/dbio/iop/arrow.go b/core/dbio/iop/arrow.go index dd24961ed..57660f3a4 100644 --- a/core/dbio/iop/arrow.go +++ b/core/dbio/iop/arrow.go @@ -210,6 +210,7 @@ func ArrowSchemaToColumns(schema *arrow.Schema) Columns { if tsType, ok := field.Type.(*arrow.TimestampType); ok { col.Metadata["timeUnit"] = tsType.Unit.String() if tsType.TimeZone != "" { + col.Metadata["timeZone"] = tsType.TimeZone col.Type = TimestampzType col.DbType = "TIMESTAMPTZ" } @@ -575,19 +576,22 @@ func ColumnsToArrowSchema(columns Columns) *arrow.Schema { case DateType: arrowType = arrow.FixedWidthTypes.Date32 case DatetimeType, TimestampType, TimestampzType: - arrowType = arrow.FixedWidthTypes.Timestamp_ns + unit := arrow.Microsecond // iceberg does not support nano, micro as default switch col.Metadata["timeUnit"] { case "s": - arrowType = arrow.FixedWidthTypes.Timestamp_s + unit = arrow.Second case "ms": - arrowType = arrow.FixedWidthTypes.Timestamp_ms + unit = arrow.Millisecond case "us": - arrowType = arrow.FixedWidthTypes.Timestamp_us + unit = arrow.Microsecond case "ns": - arrowType = arrow.FixedWidthTypes.Timestamp_ns - default: - arrowType = arrow.FixedWidthTypes.Timestamp_us // iceberg does not support nano, micro as default + unit = arrow.Nanosecond } + // Carry the column's zone in the schema so the read path can + // restore the original offset. The FixedWidthTypes.Timestamp_* + // globals are all hardcoded to "UTC", so building the type here is + // what lets a non-UTC connection `loc` survive the round trip. + arrowType = &arrow.TimestampType{Unit: unit, TimeZone: arrowSchemaTimeZone(col)} case TimeType, TimezType: // Iceberg Time (microsecond precision), matching iopTypeToIcebergPrimitiveType arrowType = arrow.FixedWidthTypes.Time64us @@ -921,6 +925,13 @@ func AppendToBuilder(builder array.Builder, col *Column, val interface{}) { } case *array.TimestampBuilder: tVal, _ := cast.ToTimeE(val) + // A zone-less arrow timestamp stores a wall clock, not an instant. Shift + // the label onto UTC so the digits survive; taking the epoch instead would + // re-express the value in UTC and move the wall clock by the offset. + if tsType, ok := b.Type().(*arrow.TimestampType); ok && tsType.TimeZone == "" { + tVal = time.Date(tVal.Year(), tVal.Month(), tVal.Day(), tVal.Hour(), + tVal.Minute(), tVal.Second(), tVal.Nanosecond(), time.UTC) + } switch col.Metadata["timeUnit"] { case "s": b.Append(arrow.Timestamp(tVal.Unix())) @@ -960,6 +971,36 @@ func AppendToBuilder(builder array.Builder, col *Column, val interface{}) { } } +// arrowSchemaTimeZone returns the zone to record in a timestamp field's schema. +// Uses the column's "timeZone" metadata when present (set on read, or by +// producers that know the connection's `loc`), defaulting to UTC so behaviour +// is unchanged for columns that carry no zone. +func arrowSchemaTimeZone(col Column) string { + if tz := col.Metadata["timeZone"]; tz != "" { + return tz + } + // A zone-less type must stay zone-less. Arrow treats any non-empty TimeZone + // as "this is an instant", so defaulting to UTC here would round-trip a + // datetime back as timestampz and land it in targets as TIMESTAMP_TZ. + if col.Type == DatetimeType { + return "" + } + return "UTC" +} + +// arrowTimestampLocation resolves the zone recorded in an arrow timestamp +// schema. Falls back to UTC when the zone is absent or not loadable, which +// preserves the previous behaviour for schemas that carry no zone. +func arrowTimestampLocation(tsType *arrow.TimestampType) *time.Location { + if tsType == nil || tsType.TimeZone == "" { + return time.UTC + } + if loc, err := time.LoadLocation(tsType.TimeZone); err == nil { + return loc + } + return time.UTC +} + // GetValueFromArrowArray extracts a value from an arrow array at the given index func GetValueFromArrowArray(arr arrow.Array, idx int) any { if arr.IsNull(idx) { @@ -1168,15 +1209,21 @@ func GetValueFromArrowArray(arr arrow.Array, idx int) any { case *array.Timestamp: val := a.Value(idx) tsType := a.DataType().(*arrow.TimestampType) + // Restore the schema's zone, not UTC. The instant is the same either + // way, but the label is written out with RFC3339Nano and becomes the + // stored offset in targets like Snowflake TIMESTAMP_TZ. Forcing UTC + // here made CDC rows land as +00:00 while snapshot rows of the same + // table kept the connection's offset. + loc := arrowTimestampLocation(tsType) switch tsType.Unit { case arrow.Second: - return time.Unix(int64(val), 0).UTC() + return time.Unix(int64(val), 0).In(loc) case arrow.Millisecond: - return time.UnixMilli(int64(val)).UTC() + return time.UnixMilli(int64(val)).In(loc) case arrow.Microsecond: - return time.UnixMicro(int64(val)).UTC() + return time.UnixMicro(int64(val)).In(loc) case arrow.Nanosecond: - return time.Unix(0, int64(val)).UTC() + return time.Unix(0, int64(val)).In(loc) } case *array.Uint16: return int64(a.Value(idx)) diff --git a/core/dbio/iop/arrow_test.go b/core/dbio/iop/arrow_test.go index d9f3bfac7..10b4864a2 100644 --- a/core/dbio/iop/arrow_test.go +++ b/core/dbio/iop/arrow_test.go @@ -5,6 +5,7 @@ import ( "context" "os" "testing" + "time" "github.com/apache/arrow-go/v18/arrow" "github.com/apache/arrow-go/v18/arrow/array" @@ -12,6 +13,7 @@ import ( "github.com/apache/arrow-go/v18/arrow/memory" "github.com/shopspring/decimal" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestArrowReadWrite(t *testing.T) { @@ -209,6 +211,128 @@ func TestArrowColumnsToArrowSchemaTimeUUID(t *testing.T) { assert.True(t, isUUID, "uuid column should map to the arrow.uuid extension type, got %T", schema.Field(2).Type) } +// TestArrowTimestampZonePreserved asserts a timestamp survives the arrow round +// trip with its zone *label* intact, not just its instant. +// +// The label matters because values are written out with RFC3339Nano, so it +// becomes the stored offset in targets like Snowflake TIMESTAMP_TZ. The read +// path used to force .UTC() on every timestamp, and ColumnsToArrowSchema built +// the field from the arrow.FixedWidthTypes.Timestamp_* globals, which are +// hardcoded to "UTC". Together those relabeled every CDC value: rows landed as +// +00:00 while snapshot rows of the same table kept -07:00/-08:00. +func TestArrowTimestampZonePreserved(t *testing.T) { + la, err := time.LoadLocation("America/Los_Angeles") + require.NoError(t, err) + + col := Column{ + Name: "create_time", + Type: TimestampzType, + Position: 1, + Metadata: map[string]string{"timeUnit": "ns", "timeZone": la.String()}, + } + + // The schema must carry the zone; otherwise the read path has nothing to + // restore from. + schema := ColumnsToArrowSchema(Columns{col}) + tsType, ok := schema.Field(0).Type.(*arrow.TimestampType) + require.True(t, ok, "expected arrow.TimestampType, got %T", schema.Field(0).Type) + assert.Equal(t, la.String(), tsType.TimeZone, "schema must carry the column's zone") + + // 18:49:20 PDT — the wall clock a MySQL DATETIME would hold. + orig := time.Date(2026, 8, 9, 18, 49, 20, 0, la) + + builder := array.NewTimestampBuilder(memory.NewGoAllocator(), tsType) + defer builder.Release() + AppendToBuilder(builder, &col, orig) + + arr := builder.NewArray() + defer arr.Release() + + got, ok := GetValueFromArrowArray(arr, 0).(time.Time) + require.True(t, ok, "expected time.Time from arrow array") + + assert.True(t, got.Equal(orig), "instant must be unchanged: got %s, want %s", got, orig) + assert.Equal(t, orig.Format(time.RFC3339Nano), got.Format(time.RFC3339Nano), + "zone label must survive the round trip (this is what reaches Snowflake TIMESTAMP_TZ)") +} + +// A column with no zone metadata must still round-trip as UTC, so existing +// behaviour is unchanged for producers that never set a zone. +func TestArrowTimestampDefaultsToUTC(t *testing.T) { + col := Column{ + Name: "ts", + Type: TimestampType, + Position: 1, + Metadata: map[string]string{"timeUnit": "ns"}, + } + + schema := ColumnsToArrowSchema(Columns{col}) + tsType, ok := schema.Field(0).Type.(*arrow.TimestampType) + require.True(t, ok) + assert.Equal(t, "UTC", tsType.TimeZone) + + orig := time.Date(2026, 8, 9, 18, 49, 20, 0, time.UTC) + + builder := array.NewTimestampBuilder(memory.NewGoAllocator(), tsType) + defer builder.Release() + AppendToBuilder(builder, &col, orig) + + arr := builder.NewArray() + defer arr.Release() + + got, ok := GetValueFromArrowArray(arr, 0).(time.Time) + require.True(t, ok) + assert.Equal(t, "2026-08-09T18:49:20Z", got.Format(time.RFC3339Nano)) +} + +// A datetime is a zone-less wall clock and must stay one across the arrow round +// trip. Arrow reads any non-empty TimeZone as "this is an instant", so giving +// the field a UTC zone by default promoted datetime to timestampz and landed +// MySQL DATETIME columns in Snowflake as TIMESTAMP_TZ instead of TIMESTAMP_NTZ. +func TestArrowDatetimeStaysZoneless(t *testing.T) { + dt := Column{Name: "new_dt", Type: DatetimeType, Position: 1} + tz := Column{ + Name: "new_ts", + Type: TimestampzType, + Position: 2, + Metadata: map[string]string{"timeZone": "America/Los_Angeles"}, + } + + schema := ColumnsToArrowSchema(Columns{dt, tz}) + + dtType, ok := schema.Field(0).Type.(*arrow.TimestampType) + require.True(t, ok) + assert.Empty(t, dtType.TimeZone, "datetime must not carry a zone") + + tzType, ok := schema.Field(1).Type.(*arrow.TimestampType) + require.True(t, ok) + assert.Equal(t, "America/Los_Angeles", tzType.TimeZone, "timestampz must keep its zone") + + // The inferred columns are what drive target DDL. + cols := ArrowSchemaToColumns(schema) + assert.Equal(t, DatetimeType, cols[0].Type, "datetime must not be promoted to timestampz") + assert.Equal(t, TimestampzType, cols[1].Type) + + // A zone-less column stores a wall clock, so the digits must survive even + // when the incoming value carries an offset. Taking the epoch here shifted + // 23:45 -07:00 to 06:45 in Snowflake TIMESTAMP_NTZ. + la, err := time.LoadLocation("America/Los_Angeles") + require.NoError(t, err) + orig := time.Date(2026, 8, 9, 23, 45, 0, 0, la) + + builder := array.NewTimestampBuilder(memory.NewGoAllocator(), dtType) + defer builder.Release() + AppendToBuilder(builder, &dt, orig) + + arr := builder.NewArray() + defer arr.Release() + + got, ok := GetValueFromArrowArray(arr, 0).(time.Time) + require.True(t, ok) + assert.Equal(t, "2026-08-09T23:45:00Z", got.Format(time.RFC3339Nano), + "wall clock must be preserved for a zone-less column") +} + // AppendToBuilder must fill a Time64 builder from a bare time-of-day string (as // emitted by SQL `time` columns). Previously cast.ToTimeE rejected these and the // value silently zeroed to 00:00:00. From 1cca6e39d829857451588e96bf725200305fbdb7 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Mon, 10 Aug 2026 22:16:41 -0300 Subject: [PATCH 02/16] fix: prevent DuckDB producer/consumer hangs with cancel and stall detection Call streamPart.Cancel when exiting loops early in importViaHTTP and WriteDataflowReadyViaDuckDB, so the producer doesn't stay parked on an unbuffered send and hang the process. Add stall detection to the DuckDB query watcher: if a query produces no output for 10 minutes (configurable via SLING_DUCKDB_STALL_TIMEOUT, 0 disables), the process is killed and the reader is unblocked. This handles cases where the DuckDB process is alive but wedged and the scanner can't acquire the process mutex to make progress. --- core/dbio/database/database_duckdb.go | 10 ++ core/dbio/filesys/fs.go | 10 ++ core/dbio/iop/duckdb.go | 131 +++++++++++++++++++++++--- 3 files changed, 136 insertions(+), 15 deletions(-) diff --git a/core/dbio/database/database_duckdb.go b/core/dbio/database/database_duckdb.go index 751b0d00c..0a0e2433d 100644 --- a/core/dbio/database/database_duckdb.go +++ b/core/dbio/database/database_duckdb.go @@ -287,8 +287,18 @@ func (conn *DuckDbConn) importViaHTTP(tableFName string, df *iop.Dataflow, forma return 0, g.Error(err, "could not setup http stream") } + // Unblock the producer if we leave the loop early (insert error). Without + // this it stays parked on an unbuffered send and the process hangs. + var cancelStream context.CancelFunc + defer func() { + if cancelStream != nil { + cancelStream() + } + }() + // Process each stream part for streamPart := range streamPartChn { + cancelStream = streamPart.Cancel columnNames := lo.Map(streamPart.Columns.Names(), func(col string, i int) string { return `"` + col + `"` }) diff --git a/core/dbio/filesys/fs.go b/core/dbio/filesys/fs.go index 1c0b29810..0c2c15d15 100755 --- a/core/dbio/filesys/fs.go +++ b/core/dbio/filesys/fs.go @@ -1411,6 +1411,15 @@ func WriteDataflowReadyViaDuckDB(fs FileSysClient, df *iop.Dataflow, uri string, return bw, g.Error(err) } + // Unblock the producer on any early return below (delete failure, copy + // error). Otherwise it stays parked on an unbuffered send and hangs. + var cancelStream context.CancelFunc + defer func() { + if cancelStream != nil { + cancelStream() + } + }() + fileFormat := dbio.FileType(strings.ToLower(cast.ToString(fs.GetProp("FORMAT")))) if fileFormat == dbio.FileTypeNone { fileFormat = InferFileFormat(uri) @@ -1428,6 +1437,7 @@ func WriteDataflowReadyViaDuckDB(fs FileSysClient, df *iop.Dataflow, uri string, } for streamPart := range streamPartChn { + cancelStream = streamPart.Cancel copyOptions := iop.DuckDbCopyOptions{ Format: fileFormat, Compression: sc.Compression, diff --git a/core/dbio/iop/duckdb.go b/core/dbio/iop/duckdb.go index 172f029ff..2b4d7b7c0 100644 --- a/core/dbio/iop/duckdb.go +++ b/core/dbio/iop/duckdb.go @@ -71,10 +71,25 @@ type duckDbQuery struct { err error started bool done bool + activity time.Time // last time the scanner saw output; for stall detection closed chan struct{} // closed when the query finishes, to stop the watcher closeOnce sync.Once } +// lastActivity is the last time the scanner saw output for this query. +func (dq *duckDbQuery) lastActivity() time.Time { + dq.mu.RLock() + defer dq.mu.RUnlock() + return dq.activity +} + +// touch records that the process is still responsive. +func (dq *duckDbQuery) touch() { + dq.mu.Lock() + defer dq.mu.Unlock() + dq.activity = time.Now() +} + func (dq *duckDbQuery) getErr() error { dq.mu.RLock() defer dq.mu.RUnlock() @@ -759,14 +774,25 @@ func (duck *DuckDb) ExecContext(ctx context.Context, sql string, args ...any) (r func (duck *DuckDb) newQuery(ctx context.Context, sql string) (query *duckDbQuery) { stdOutReader, stdOutWriter := io.Pipe() // new pipe dq := &duckDbQuery{ - SQL: sql, - Context: g.NewContext(ctx), - reader: stdOutReader, - writer: stdOutWriter, - closed: make(chan struct{}), + SQL: sql, + Context: g.NewContext(ctx), + reader: stdOutReader, + writer: stdOutWriter, + activity: time.Now(), + closed: make(chan struct{}), } duck.setQuery(dq) + // Abort a query whose process stops producing output while still alive. + // 0 disables. Generous by default: legitimate long queries emit nothing + // while computing, so this is a last-resort backstop, not a query timeout. + stallTimeout := 10 * time.Minute + if val := os.Getenv("SLING_DUCKDB_STALL_TIMEOUT"); val != "" { + if d, err := time.ParseDuration(val); err == nil { + stallTimeout = d + } + } + // watcher: unblock a stuck reader (and release duck.Context's lock) when the // query context is cancelled, or the process dies / its stdout scanner stops // before the EOF marker arrives. Otherwise ConsumeCsvReader blocks forever. @@ -785,6 +811,17 @@ func (duck *DuckDb) newQuery(ctx context.Context, sql string) (query *duckDbQuer duck.kill() // kill the proc so it won't block subsequent queries return case <-ticker.C: + // Check the stall first: it needs no Proc lock, so it still fires + // if the scanner is wedged holding the process mutex. + if !dq.isDone() && stallTimeout > 0 && time.Since(dq.lastActivity()) > stallTimeout { + err := g.Error("duckdb query stalled: no output for %s", stallTimeout) + dq.setErr(err) + dq.writer.CloseWithError(err) + dq.reader.CloseWithError(err) + duck.kill() // unresponsive; kill so it won't block subsequent queries + return + } + if duck.Proc != nil && !dq.isDone() && (duck.Proc.Exited() || duck.Proc.GetScanErr() != nil) { reason := "duckdb process exited before query completed" if duck.Proc.GetScanErr() != nil { @@ -836,6 +873,9 @@ func (duck *DuckDb) waitForResult(dq *duckDbQuery) (result sql.Result, err error // No more data available, but EOF marker not found yet goto next } + if qErr := dq.getErr(); qErr != nil { + return result, qErr // stall/process-death cause, not "closed pipe" + } return result, g.Error(err, "could not read output from stdout") } @@ -847,6 +887,9 @@ func (duck *DuckDb) waitForResult(dq *duckDbQuery) (result sql.Result, err error // No more data available, but EOF marker not found yet goto next } + if qErr := dq.getErr(); qErr != nil { + return result, qErr // stall/process-death cause, not "closed pipe" + } return result, g.Error(err, "could not read output from stdout") } @@ -1263,6 +1306,8 @@ func (duck *DuckDb) initScanner() { return } + dq.touch() // process is responsive; reset the stall clock + mu.Lock() defer mu.Unlock() @@ -1676,6 +1721,12 @@ type HttpStreamPart struct { Index int FromExpr string Columns Columns + + // Cancel unblocks the producer goroutine in DataflowToHttpStream. Consumers + // MUST defer it when they range over streamPartChn: leaving the loop early + // (on an insert error) otherwise strands the producer on an unbuffered send + // with no reader, hanging the process. + Cancel context.CancelFunc } func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamPartChn chan HttpStreamPart, err error) { @@ -1723,9 +1774,19 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP // server.Use(middleware.Logger()) server.Add(http.MethodGet, "/data", func(c echo.Context) (err error) { - reader := <-readerCh + var reader io.Reader + select { + case reader = <-readerCh: + case <-importContext.Ctx.Done(): + return c.NoContent(http.StatusOK) + } if reader != nil { - defer func() { doneCh <- true }() + defer func() { + select { + case doneCh <- true: + case <-importContext.Ctx.Done(): + } + }() return c.Stream(200, contentType, reader) } return c.NoContent(http.StatusOK) @@ -1765,8 +1826,12 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP defer close(streamPartChn) defer func() { // Shut down HTTP server immediately after all batches are processed - // to prevent interference with subsequent DuckDB queries - server.Shutdown(importContext.Ctx) + // to prevent interference with subsequent DuckDB queries. Use a fresh + // context: importContext may already be cancelled, which would skip + // the graceful shutdown entirely. + shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + server.Shutdown(shutCtx) }() var partIndex int @@ -1782,10 +1847,17 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP // Use read_arrow for Arrow format fromExpr := g.F(`read_arrow('%s')`, httpURL) - streamPartChn <- HttpStreamPart{ + select { + case streamPartChn <- HttpStreamPart{ Index: partIndex, FromExpr: fromExpr, Columns: batchR.Columns, + Cancel: importContext.Cancel, + }: + case <-importContext.Ctx.Done(): + pipeR.CloseWithError(importContext.Ctx.Err()) + pipeW.CloseWithError(importContext.Ctx.Err()) + return } // Stream data through pipe @@ -1797,8 +1869,19 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP } }() - readerCh <- pipeR - <-doneCh + select { + case readerCh <- pipeR: + case <-importContext.Ctx.Done(): + pipeR.CloseWithError(importContext.Ctx.Err()) + return + } + + select { + case <-doneCh: + case <-importContext.Ctx.Done(): + pipeR.CloseWithError(importContext.Ctx.Err()) + return + } partIndex++ } @@ -1827,10 +1910,17 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP // 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) - streamPartChn <- HttpStreamPart{ + select { + case streamPartChn <- HttpStreamPart{ Index: partIndex, FromExpr: fromExpr, Columns: batchR.Columns, + Cancel: importContext.Cancel, + }: + case <-importContext.Ctx.Done(): + pipeR.CloseWithError(importContext.Ctx.Err()) + pipeW.CloseWithError(importContext.Ctx.Err()) + return } // Stream data through pipe @@ -1842,8 +1932,19 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP } }() - readerCh <- pipeR - <-doneCh + select { + case readerCh <- pipeR: + case <-importContext.Ctx.Done(): + pipeR.CloseWithError(importContext.Ctx.Err()) + return + } + + select { + case <-doneCh: + case <-importContext.Ctx.Done(): + pipeR.CloseWithError(importContext.Ctx.Err()) + return + } partIndex++ } From 3a89e354be5c706f7256a5f01f44804a48247cfe Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Tue, 11 Aug 2026 23:26:38 -0300 Subject: [PATCH 03/16] fix: prevent deadlocks when consumers stop reading mid-stream When duckdb's read_arrow stops reading after it has enough data, the arrow/csv writer's flush blocks forever on the pipe with no reader. Add closeArrowWriter to abort the flush when the stream context is cancelled, and closeBatchReader to fail fast on abandoned batch pipes. Bridge the dataflow and import contexts so failures propagate across both, and bump the stall timeout from 10 to 60 minutes to avoid killing legitimate long-running direct_insert queries. --- core/dbio/iop/datastream.go | 26 ++++++++++++++++-- core/dbio/iop/duckdb.go | 54 ++++++++++++++++++++++++++++++++++--- 2 files changed, 75 insertions(+), 5 deletions(-) diff --git a/core/dbio/iop/datastream.go b/core/dbio/iop/datastream.go index 4a4ff8ccc..d532bec88 100644 --- a/core/dbio/iop/datastream.go +++ b/core/dbio/iop/datastream.go @@ -2717,6 +2717,25 @@ func (ds *Datastream) NewJsonLinesReaderChnl(sc StreamConfig) (readerChn chan *i return readerChn } +// closeArrowWriter closes aw, aborting the flush if the stream context is +// cancelled. Without this, a consumer that stops reading mid-batch (duckdb's +// read_arrow) leaves the flush blocked on the pipe with no reader forever. +func (ds *Datastream) closeArrowWriter(aw *ArrowWriter, pipeW *io.PipeWriter) error { + done := make(chan error, 1) + go func() { done <- aw.Close() }() + + select { + case err := <-done: + return err + case <-ds.Context.Ctx.Done(): + if pipeW != nil { + pipeW.CloseWithError(ds.Context.Ctx.Err()) // unblock the flush + } + <-done // it returns once the pipe is broken + return ds.Context.Ctx.Err() + } +} + // NewArrowReaderChnl provides a channel of readers as the limit is reached // each channel flows as fast as the consumer consumes func (ds *Datastream) NewArrowReaderChnl(sc StreamConfig) (readerChn chan *BatchReader) { @@ -2735,8 +2754,11 @@ func (ds *Datastream) NewArrowReaderChnl(sc StreamConfig) (readerChn chan *Batch nextPipe := func(batch *Batch) error { if aw != nil { - err := aw.Close() - if err != nil { + // Closing flushes the tail of the batch into the prior pipe. If the + // consumer already stopped reading (duckdb's read_arrow stops once + // it has what it needs), that write blocks forever, so unblock it + // when the stream context is cancelled. + if err := ds.closeArrowWriter(aw, pipeW); err != nil { return g.Error(err, "could not close arrow writer") } } diff --git a/core/dbio/iop/duckdb.go b/core/dbio/iop/duckdb.go index 2b4d7b7c0..54fc148ae 100644 --- a/core/dbio/iop/duckdb.go +++ b/core/dbio/iop/duckdb.go @@ -784,9 +784,10 @@ func (duck *DuckDb) newQuery(ctx context.Context, sql string) (query *duckDbQuer duck.setQuery(dq) // Abort a query whose process stops producing output while still alive. - // 0 disables. Generous by default: legitimate long queries emit nothing - // while computing, so this is a last-resort backstop, not a query timeout. - stallTimeout := 10 * time.Minute + // 0 disables. This is a last-resort backstop, not a query timeout: a single + // large direct_insert emits nothing between issue and completion, and runs + // past 15 min are normal, so keep it well clear of legitimate work. + stallTimeout := 60 * time.Minute if val := os.Getenv("SLING_DUCKDB_STALL_TIMEOUT"); val != "" { if d, err := time.ParseDuration(val); err == nil { stallTimeout = d @@ -1729,6 +1730,18 @@ type HttpStreamPart struct { Cancel context.CancelFunc } +// closeBatchReader stops reading a batch's pipe. The arrow/csv writer flushes +// the tail of a batch into this pipe; if the consumer abandoned it (duckdb +// dropped the response), that flush blocks forever. Closing makes it fail fast. +func closeBatchReader(br *BatchReader) { + if br == nil { + return + } + if pr, ok := br.Reader.(*io.PipeReader); ok { + pr.CloseWithError(g.Error("batch reader closed by consumer")) + } +} + func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamPartChn chan HttpStreamPart, err error) { // create fromExprChn channel streamPartChn = make(chan HttpStreamPart) @@ -1761,6 +1774,17 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP // create http server to serve data importContext := g.NewContext(duck.Context.Ctx) + + // The dataflow and the duckdb connection live on separate context trees, so + // a failure on the source side would otherwise never reach this producer + // (and vice versa), deadlocking the readerCh/doneCh handshake. Bridge them. + go func() { + select { + case <-df.Context.Ctx.Done(): + importContext.Cancel() + case <-importContext.Ctx.Done(): + } + }() httpURL := g.F("http://localhost:%d/data", port) server := echo.New() { @@ -1833,6 +1857,14 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP defer cancel() server.Shutdown(shutCtx) }() + // On an early exit the arrow/csv writer may be blocked flushing into a + // pipe this loop will no longer read. Cancelling the merged datastream + // breaks that write so its goroutine can finish. + defer func() { + if importContext.Ctx.Err() != nil { + ds.Context.Cancel() + } + }() var partIndex int if format == dbio.FileTypeArrow { @@ -1857,6 +1889,7 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP case <-importContext.Ctx.Done(): pipeR.CloseWithError(importContext.Ctx.Err()) pipeW.CloseWithError(importContext.Ctx.Err()) + closeBatchReader(batchR) return } @@ -1873,6 +1906,7 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP case readerCh <- pipeR: case <-importContext.Ctx.Done(): pipeR.CloseWithError(importContext.Ctx.Err()) + closeBatchReader(batchR) return } @@ -1880,9 +1914,15 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP case <-doneCh: case <-importContext.Ctx.Done(): pipeR.CloseWithError(importContext.Ctx.Err()) + closeBatchReader(batchR) return } + // Done with this batch. If duckdb dropped the response early, + // io.Copy stopped mid-batch and left data buffered in the batch + // pipe; the writer's next flush would block on it forever. + closeBatchReader(batchR) + partIndex++ } } else { @@ -1920,6 +1960,7 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP case <-importContext.Ctx.Done(): pipeR.CloseWithError(importContext.Ctx.Err()) pipeW.CloseWithError(importContext.Ctx.Err()) + closeBatchReader(batchR) return } @@ -1936,6 +1977,7 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP case readerCh <- pipeR: case <-importContext.Ctx.Done(): pipeR.CloseWithError(importContext.Ctx.Err()) + closeBatchReader(batchR) return } @@ -1943,9 +1985,15 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP case <-doneCh: case <-importContext.Ctx.Done(): pipeR.CloseWithError(importContext.Ctx.Err()) + closeBatchReader(batchR) return } + // Done with this batch. If duckdb dropped the response early, + // io.Copy stopped mid-batch and left data buffered in the batch + // pipe; the writer's next flush would block on it forever. + closeBatchReader(batchR) + partIndex++ } } From 2790c3957388f27e3157f661f3b3d55066de9f0e Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Wed, 12 Aug 2026 07:57:56 -0300 Subject: [PATCH 04/16] fix: retain datetime type for arrow columns with timezones - Add metadata `sling:declaredType` to Arrow fields to retain the original iop type. - Prevents a zone-labeled `datetime` from being promoted to `timestamptz` when converting back from Arrow. - This ensures CDC rows match snapshot rows and target DDLs are generated correctly as `TIMESTAMP` rather than `TIMESTAMPTZ`. - Add test to verify the zone label is preserved without altering the column type. --- core/dbio/iop/arrow.go | 35 +++++++++++++++++++++++++++++--- core/dbio/iop/arrow_test.go | 40 +++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 3 deletions(-) diff --git a/core/dbio/iop/arrow.go b/core/dbio/iop/arrow.go index 57660f3a4..47bf41363 100644 --- a/core/dbio/iop/arrow.go +++ b/core/dbio/iop/arrow.go @@ -214,6 +214,22 @@ func ArrowSchemaToColumns(schema *arrow.Schema) Columns { col.Type = TimestampzType col.DbType = "TIMESTAMPTZ" } + // The declared type wins when present. A datetime may carry a + // zone purely to label its values, and that must not turn it + // into a timestampz in the target DDL. + if declared, ok := field.Metadata.GetValue(arrowDeclaredTypeKey); ok { + switch ColumnType(declared) { + case DatetimeType: + col.Type = DatetimeType + col.DbType = "TIMESTAMP" + case TimestampType: + col.Type = TimestampType + col.DbType = "TIMESTAMP" + case TimestampzType: + col.Type = TimestampzType + col.DbType = "TIMESTAMPTZ" + } + } } case arrow.STRING, arrow.LARGE_STRING: col.Type = StringType @@ -611,6 +627,15 @@ func ColumnsToArrowSchema(columns Columns) *arrow.Schema { Type: arrowType, Nullable: true, } + + // Record the declared type for timestamps. Arrow's TimeZone is the only + // zone signal it has, so a zone-carrying datetime is otherwise + // indistinguishable from a timestampz on the way back. Keeping the + // declared type here lets the zone act purely as a label. + if col.Type == DatetimeType || col.Type == TimestampType || col.Type == TimestampzType { + fields[i].Metadata = arrow.NewMetadata( + []string{arrowDeclaredTypeKey}, []string{string(col.Type)}) + } } return arrow.NewSchema(fields, nil) @@ -971,6 +996,10 @@ func AppendToBuilder(builder array.Builder, col *Column, val interface{}) { } } +// arrowDeclaredTypeKey names the arrow field metadata that carries a timestamp +// column's declared iop type across the cache round trip. +const arrowDeclaredTypeKey = "sling:declaredType" + // arrowSchemaTimeZone returns the zone to record in a timestamp field's schema. // Uses the column's "timeZone" metadata when present (set on read, or by // producers that know the connection's `loc`), defaulting to UTC so behaviour @@ -979,9 +1008,9 @@ func arrowSchemaTimeZone(col Column) string { if tz := col.Metadata["timeZone"]; tz != "" { return tz } - // A zone-less type must stay zone-less. Arrow treats any non-empty TimeZone - // as "this is an instant", so defaulting to UTC here would round-trip a - // datetime back as timestampz and land it in targets as TIMESTAMP_TZ. + // A datetime with no zone of its own stays zone-less. Defaulting it to UTC + // would relabel its wall clock. When a zone is supplied the declared type + // keeps it from being promoted to timestampz. if col.Type == DatetimeType { return "" } diff --git a/core/dbio/iop/arrow_test.go b/core/dbio/iop/arrow_test.go index 10b4864a2..b21dc0c73 100644 --- a/core/dbio/iop/arrow_test.go +++ b/core/dbio/iop/arrow_test.go @@ -333,6 +333,46 @@ func TestArrowDatetimeStaysZoneless(t *testing.T) { "wall clock must be preserved for a zone-less column") } +// A datetime given a zone keeps that label through the round trip without being +// promoted to timestampz. The CDC readers set the zone from the connection's +// `loc` so their rows carry the same offset the snapshot path gets from the +// driver, while the column still lands as TIMESTAMP_NTZ. +func TestArrowDatetimeKeepsZoneLabel(t *testing.T) { + la, err := time.LoadLocation("America/Los_Angeles") + require.NoError(t, err) + + col := Column{ + Name: "create_time", + Type: DatetimeType, + Position: 1, + Metadata: map[string]string{"timeZone": la.String()}, + } + + schema := ColumnsToArrowSchema(Columns{col}) + tsType, ok := schema.Field(0).Type.(*arrow.TimestampType) + require.True(t, ok) + assert.Equal(t, la.String(), tsType.TimeZone, "zone must reach the schema") + + cols := ArrowSchemaToColumns(schema) + assert.Equal(t, DatetimeType, cols[0].Type, + "a zone must not promote a datetime to timestampz") + + orig := time.Date(2026, 8, 11, 14, 30, 0, 0, la) + + builder := array.NewTimestampBuilder(memory.NewGoAllocator(), tsType) + defer builder.Release() + AppendToBuilder(builder, &col, orig) + + arr := builder.NewArray() + defer arr.Release() + + got, ok := GetValueFromArrowArray(arr, 0).(time.Time) + require.True(t, ok) + assert.True(t, got.Equal(orig), "instant must be unchanged") + assert.Equal(t, orig.Format(time.RFC3339Nano), got.Format(time.RFC3339Nano), + "zone label must survive so CDC rows match snapshot rows") +} + // AppendToBuilder must fill a Time64 builder from a bare time-of-day string (as // emitted by SQL `time` columns). Previously cast.ToTimeE rejected these and the // value silently zeroed to 00:00:00. From 3fb5b97944723df261edeb3ac8069ebeac84214b Mon Sep 17 00:00:00 2001 From: Ethan Setnik Date: Thu, 13 Aug 2026 02:03:30 -0400 Subject: [PATCH 05/16] fix: raise duckdb read_csv max_line_size for large text-class columns The max_line_size bump added in v1.5.19 for hex-encoded binary columns only triggered when a binary column was present. Sources with large text-class columns (sqlserver text/ntext/xml/varchar(max), clob) still generated read_csv with the default max_line_size=2000000, so any row whose serialized CSV line exceeded 2 MB failed the whole stream with: Invalid Input Error: CSV Error ... Maximum line size of 2000000 bytes exceeded. Actual Size:5022477 bytes. Extend the existing condition to also bump max_line_size to 256 MB when any column is TextType, mirroring the binary/hex raise. Fixes #787 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01H1JYFWjSiEVsGKWjGVpqr1 --- core/dbio/iop/duckdb.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/core/dbio/iop/duckdb.go b/core/dbio/iop/duckdb.go index 172f029ff..a8328edcb 100644 --- a/core/dbio/iop/duckdb.go +++ b/core/dbio/iop/duckdb.go @@ -1816,9 +1816,11 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP // when any binary column is present so large LOBs don't blow up // DuckDB's default 2 MB line limit. 256 MB covers Snowflake's // 64 MB BINARY ceiling with comfortable headroom for hex + quoting. + // Large text-class columns (text, ntext, xml, varchar(max), clob) + // can likewise exceed the 2 MB default, so they get the same bump. maxLineSize := 2000000 for _, c := range batchR.Columns { - if c.IsBinary() { + if c.IsBinary() || c.Type == TextType { maxLineSize = 256 * 1024 * 1024 break } From 518c19e9579e5dfcf2450806081bba37c7815298 Mon Sep 17 00:00:00 2001 From: Ethan Setnik Date: Thu, 13 Aug 2026 02:06:14 -0400 Subject: [PATCH 06/16] test: cover max_line_size raise; fix misleading arrow debug message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add unit test cases for the read_csv max_line_size computation in DataflowToHttpStream, covering all three branches: binary column raise (v1.5.19, previously untested), text-class column raise (#787 fix), and the 2 MB default for plain string columns. Also correct the debug message emitted when arrow streaming is disabled via SLING_DUCKDB_ARROW — it referred to "duckdb extension arrow" but the toggle disables sling's arrow streaming format, not a DuckDB extension. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01H1JYFWjSiEVsGKWjGVpqr1 --- core/dbio/iop/duckdb.go | 2 +- core/dbio/iop/duckdb_test.go | 70 ++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/core/dbio/iop/duckdb.go b/core/dbio/iop/duckdb.go index a8328edcb..61c2d80d4 100644 --- a/core/dbio/iop/duckdb.go +++ b/core/dbio/iop/duckdb.go @@ -1704,7 +1704,7 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP contentType = "application/vnd.apache.arrow.stream" format = dbio.FileTypeArrow } else { - g.Debug("duckdb extension arrow is disabled, using csv") + g.Debug("duckdb arrow streaming is disabled via SLING_DUCKDB_ARROW, using csv") } } diff --git a/core/dbio/iop/duckdb_test.go b/core/dbio/iop/duckdb_test.go index f71e26741..6281fd075 100644 --- a/core/dbio/iop/duckdb_test.go +++ b/core/dbio/iop/duckdb_test.go @@ -525,6 +525,76 @@ func TestDuckDbDataflowToHttpStream(t *testing.T) { t.Logf("Test completed - received %d Arrow parts. Implementation uses io.Pipe for streaming.", len(parts)) }) + + t.Run("CSV streaming - max_line_size raised for binary and text columns", func(t *testing.T) { + // Binary columns are hex-encoded (2x byte size) and text-class columns + // (text, ntext, xml, varchar(max), clob) can exceed DuckDB's default + // 2 MB line limit, so both must bump max_line_size to 256 MB (issue #787). + testCases := []struct { + name string + columnType ColumnType + maxLineSize string + }{ + {"binary column raises limit", BinaryType, "max_line_size=268435456"}, + {"text column raises limit", TextType, "max_line_size=268435456"}, + {"string column keeps default", StringType, "max_line_size=2000000"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + df := NewDataflow() + columns := NewColumnsFromFields("id", "payload") + columns[0].Type = IntegerType + columns[1].Type = tc.columnType + df.Columns = columns + df.Ready = true + + ds := NewDatastreamContext(ctx, columns) + ds.SetConfig(map[string]string{}) + ds.Buffer = append(ds.Buffer, []any{int64(1), "some payload"}) + ds.Count = 1 + ds.Ready = true + + df.Streams = append(df.Streams, ds) + + go func() { + defer close(df.StreamCh) + df.StreamCh <- ds + time.Sleep(50 * time.Millisecond) + ds.Close() + }() + + duck := NewDuckDb(ctx) + defer duck.Close() + + sc := StreamConfig{ + Format: dbio.FileTypeCsv, + BatchLimit: 10, + FileMaxBytes: 1024 * 1024, + } + + streamPartChn, err := duck.DataflowToHttpStream(df, sc) + assert.NoError(t, err) + assert.NotNil(t, streamPartChn) + + select { + case part, ok := <-streamPartChn: + if assert.True(t, ok, "should have received a stream part") { + assert.Contains(t, part.FromExpr, "read_csv") + assert.Contains(t, part.FromExpr, tc.maxLineSize) + t.Logf("Received part: %s", part.FromExpr) + } + case <-time.After(3 * time.Second): + t.Error("timed out waiting for stream part") + } + + cancel() + }) + } + }) } // regression guard for issue #770: http_timeout must be raised on every DuckDB From 2f741107b24a376e898bdbb52a0e91f070f7e0d8 Mon Sep 17 00:00:00 2001 From: Ethan Setnik Date: Thu, 13 Aug 2026 02:17:28 -0400 Subject: [PATCH 07/16] fix: tailor CSV bridge error help by connection type MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ErrorHelper hint for "Invalid Input Error: CSV Error on Line:" unconditionally suggested `copy_method: arrow_http`, but that property only applies to DuckDB / MotherDuck / DuckLake connections — for other sources (e.g. sqlserver) the hint is a dead end (noted in #787). - ErrorHelper now accepts the task's connection types (variadic, so the existing signature remains compatible) and only suggests arrow_http when a DuckDB-class connection is involved. - Add a specific help message for the max_line_size-exceeded failure signature, which previously fell through to the misleading arrow_http hint. - Add unit tests covering the tailored help strings. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01H1JYFWjSiEVsGKWjGVpqr1 --- cmd/sling/sling_run.go | 2 +- core/sling/task.go | 17 ++++++++++++++--- core/sling/task_test.go | 32 ++++++++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 4 deletions(-) create mode 100644 core/sling/task_test.go diff --git a/cmd/sling/sling_run.go b/cmd/sling/sling_run.go index 2c0f76b76..bc48eb757 100755 --- a/cmd/sling/sling_run.go +++ b/cmd/sling/sling_run.go @@ -508,7 +508,7 @@ func runTask(cfg *sling.Config, replication *sling.ReplicationConfig) (err error } // show help text - if eh := sling.ErrorHelper(err); eh != "" { + if eh := sling.ErrorHelper(err, task.Config.SrcConn.GetType(), task.Config.TgtConn.GetType()); eh != "" { env.Println("") env.Println(env.MagentaString(eh)) env.Println("") diff --git a/core/sling/task.go b/core/sling/task.go index 9efe0dce8..120abef46 100644 --- a/core/sling/task.go +++ b/core/sling/task.go @@ -615,7 +615,7 @@ const ( raiseIssueNotice = "Feel free to open an issue @ https://github.com/slingdata-io/sling-cli" ) -func ErrorHelper(err error) (helpString string) { +func ErrorHelper(err error, connTypes ...dbio.Type) (helpString string) { if err != nil { errString := strings.ToLower(err.Error()) E, ok := err.(*g.ErrType) @@ -623,6 +623,15 @@ func ErrorHelper(err error) (helpString string) { errString = strings.ToLower(E.Full()) } + // whether one of the task's connections is a DuckDB-class connection, + // which accepts the `copy_method` property + usesDuckDb := false + for _, connType := range connTypes { + if g.In(connType, dbio.TypeDbDuckDb, dbio.TypeDbMotherDuck, dbio.TypeDbDuckLake) { + usesDuckDb = true + } + } + contains := func(text ...string) bool { met := true for _, t := range text { @@ -665,8 +674,10 @@ func ErrorHelper(err error) (helpString string) { helpString = "See https://docs.slingdata.io/ for creating a custom connection template." case contains("CSV") && contains("encountered too many errors"): helpString = "Perhaps trying to load with `target_options.format=parquet` could help? This will use Parquet files instead of CSV files." - case contains("Invalid Input Error: CSV Error on Line:"): - helpString = "By default, Sling uses CSV serialization to pipe data into DuckDB. Try setting the `copy_method: arrow_http` property in your connection to avoid serialization errors. See https://docs.slingdata.io/connections/database-connections for more details." + case contains("Maximum line size of", "bytes exceeded"): + helpString = "A row's serialized size exceeded the max_line_size limit of Sling's internal DuckDB CSV bridge. Sling raises this limit automatically when binary or large text-class columns are present in the source schema. If you are still seeing this error, please open an issue @ https://github.com/slingdata-io/sling-cli" + case contains("Invalid Input Error: CSV Error on Line:") && usesDuckDb: + helpString = "By default, Sling uses CSV serialization to pipe data into DuckDB. Try setting the `copy_method: arrow_http` property in your DuckDB / MotherDuck connection to avoid serialization errors. See https://docs.slingdata.io/connections/database-connections for more details." case contains("it does not have a replica identity and publishes updates"): helpString = `Since PG replication is turned on, you'll need to create a replica identity on the respective table for executing UPDATE/DELETE operations. You can use target_options.table_ddl to specify an extra statement to define the replication identity upon creation, such as: diff --git a/core/sling/task_test.go b/core/sling/task_test.go new file mode 100644 index 000000000..2a7e62b97 --- /dev/null +++ b/core/sling/task_test.go @@ -0,0 +1,32 @@ +package sling + +import ( + "testing" + + "github.com/flarco/g" + "github.com/slingdata-io/sling-cli/core/dbio" + "github.com/stretchr/testify/assert" +) + +func TestErrorHelper(t *testing.T) { + maxLineSizeErr := g.Error("Invalid Input Error: CSV Error on Line: 2\nMaximum line size of 2000000 bytes exceeded. Actual Size:5022477 bytes.") + csvErr := g.Error("Invalid Input Error: CSV Error on Line: 2\nsome other serialization error") + + t.Run("max_line_size exceeded gets specific help, not arrow_http", func(t *testing.T) { + helpString := ErrorHelper(maxLineSizeErr, dbio.TypeDbSQLServer, dbio.TypeFileS3) + assert.Contains(t, helpString, "max_line_size") + assert.NotContains(t, helpString, "arrow_http") + }) + + t.Run("csv error suggests arrow_http for duckdb-class connections", func(t *testing.T) { + for _, connType := range []dbio.Type{dbio.TypeDbDuckDb, dbio.TypeDbMotherDuck, dbio.TypeDbDuckLake} { + helpString := ErrorHelper(csvErr, dbio.TypeDbPostgres, connType) + assert.Contains(t, helpString, "arrow_http", "connType=%s", connType) + } + }) + + t.Run("csv error does not suggest arrow_http for non-duckdb connections", func(t *testing.T) { + helpString := ErrorHelper(csvErr, dbio.TypeDbSQLServer, dbio.TypeFileS3) + assert.NotContains(t, helpString, "arrow_http") + }) +} From a4130515bdb6f1254fc6dad5fe8e273282d8613d Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Thu, 13 Aug 2026 08:35:11 -0300 Subject: [PATCH 08/16] fix: handle missing PK column in merge config Add nil checks for source and target primary key columns in GenerateMergeConfigWithStrategy to prevent nil pointer dereferences and provide a clearer error message when a specified primary key is missing. --- core/dbio/database/database.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/core/dbio/database/database.go b/core/dbio/database/database.go index ed276dd91..f3a00f9a1 100755 --- a/core/dbio/database/database.go +++ b/core/dbio/database/database.go @@ -3081,6 +3081,12 @@ func (conn *BaseConn) GenerateMergeConfigWithStrategy(srcTable string, tgtTable // don't normalize, use raw name srcCol := srcColumns.GetColumn(pkField) tgtCol := tgtColumns.GetColumn(pkField) + if srcCol == nil { + return mc, g.Error("did not find source PK column: %s (has %s)", pkField, g.Marshal(srcColumns.Names())) + } + if tgtCol == nil { + return mc, g.Error("did not find target PK column: %s (has %s)", pkField, g.Marshal(tgtColumns.Names())) + } srcField := conn.Quote(srcCol.Name) tgtField := conn.Quote(tgtCol.Name) From 7479b9464a8d4fee8582a668642c6f75e9aa3bc4 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Thu, 13 Aug 2026 10:10:40 -0300 Subject: [PATCH 09/16] fix(duckdb): centralize max_line_size logic across all import paths Extract MaxLineSize into a reusable method on DuckDb and apply it consistently to CSV imports via temp files, named pipes, and HTTP streaming. The method bumps the line limit for string columns (text, binary, etc.) to 256 MB instead of DuckDB's 2 MB default, and honors a new `max_line_size` property override. This fixes the case where large strings or binaries caused failures during temp-CSV and named-pipe ingestion, which previously used the hardcoded default. --- core/dbio/database/database_duckdb.go | 2 +- core/dbio/database/database_duckdb_unix.go | 2 +- core/dbio/iop/duckdb.go | 41 +++++++----- core/dbio/iop/duckdb_test.go | 47 ++++++++++++-- core/sling/task.go | 2 +- core/sling/task_test.go | 1 + .../r.101.duckdb_max_line_size.yaml | 62 +++++++++++++++++++ tests/suite.cli.yaml | 7 +++ 8 files changed, 143 insertions(+), 21 deletions(-) create mode 100644 tests/replications/r.101.duckdb_max_line_size.yaml diff --git a/core/dbio/database/database_duckdb.go b/core/dbio/database/database_duckdb.go index 751b0d00c..4e8e3cc94 100644 --- a/core/dbio/database/database_duckdb.go +++ b/core/dbio/database/database_duckdb.go @@ -236,7 +236,7 @@ func (conn *DuckDbConn) importViaTempCSVs(tableFName string, df *iop.Dataflow) ( }) sqlLines := []string{ - g.F(`insert into %s (%s) select * from read_csv('%s', delim=',', header=True, columns=%s, max_line_size=2000000, parallel=false, quote='"', escape='"', nullstr='\N', auto_detect=false);`, table.FDQN(), strings.Join(columnNames, ", "), file.Node.Path(), conn.generateCsvColumns(file.Columns)), + g.F(`insert into %s (%s) select * from read_csv('%s', delim=',', header=True, columns=%s, max_line_size=%d, parallel=false, quote='"', escape='"', nullstr='\N', auto_detect=false);`, table.FDQN(), strings.Join(columnNames, ", "), file.Node.Path(), conn.generateCsvColumns(file.Columns), conn.duck.MaxLineSize(file.Columns)), } sql := strings.Join(sqlLines, ";\n") diff --git a/core/dbio/database/database_duckdb_unix.go b/core/dbio/database/database_duckdb_unix.go index 56f5c60a4..d473ea57b 100644 --- a/core/dbio/database/database_duckdb_unix.go +++ b/core/dbio/database/database_duckdb_unix.go @@ -98,7 +98,7 @@ func (conn *DuckDbConn) importViaNamedPipe(tableFName string, df *iop.Dataflow) }) sqlLines := []string{ - g.F(`insert into %s (%s) select * from read_csv('%s', delim=',', auto_detect=False, header=True, columns=%s, max_line_size=2000000, parallel=false, quote='"', escape='"', nullstr='\N', auto_detect=false);`, table.FDQN(), strings.Join(columnNames, ", "), pipePath, conn.generateCsvColumns(df.Columns)), + g.F(`insert into %s (%s) select * from read_csv('%s', delim=',', auto_detect=False, header=True, columns=%s, max_line_size=%d, parallel=false, quote='"', escape='"', nullstr='\N', auto_detect=false);`, table.FDQN(), strings.Join(columnNames, ", "), pipePath, conn.generateCsvColumns(df.Columns), conn.duck.MaxLineSize(df.Columns)), } sql := strings.Join(sqlLines, ";\n") diff --git a/core/dbio/iop/duckdb.go b/core/dbio/iop/duckdb.go index 61c2d80d4..4bba2dd1f 100644 --- a/core/dbio/iop/duckdb.go +++ b/core/dbio/iop/duckdb.go @@ -1811,20 +1811,7 @@ func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamP // Create a pipe to stream data through pipeR, pipeW := io.Pipe() - // Binary columns are streamed as hex text (2x the byte size) — a - // single 64 MB BLOB becomes ~128 MB on the wire. Bump max_line_size - // when any binary column is present so large LOBs don't blow up - // DuckDB's default 2 MB line limit. 256 MB covers Snowflake's - // 64 MB BINARY ceiling with comfortable headroom for hex + quoting. - // Large text-class columns (text, ntext, xml, varchar(max), clob) - // can likewise exceed the 2 MB default, so they get the same bump. - maxLineSize := 2000000 - for _, c := range batchR.Columns { - if c.IsBinary() || c.Type == TextType { - maxLineSize = 256 * 1024 * 1024 - break - } - } + 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) @@ -1867,6 +1854,32 @@ func (duck *DuckDb) DefaultCsvConfig() (config StreamConfig) { return config } +const ( + DuckDbDefaultMaxLineSize = 2000000 + DuckDbLargeMaxLineSize = 256 * 1024 * 1024 +) + +// MaxLineSize returns the read_csv max_line_size for the given columns. +// Any column that can hold an unbounded value raises the limit, since one row +// would otherwise exceed DuckDB's 2 MB default. It is a limit, not an +// allocation. The `max_line_size` prop overrides it. +func (duck *DuckDb) MaxLineSize(columns Columns) int { + if override := duck.GetProp("max_line_size"); override != "" { + if size := cast.ToInt(override); size > 0 { + return size + } + g.Warn("invalid max_line_size value '%s', ignoring", override) + } + + for _, c := range columns { + if c.IsString() { + return DuckDbLargeMaxLineSize + } + } + + return DuckDbDefaultMaxLineSize +} + func (duck *DuckDb) GenerateCsvColumns(columns Columns) (colStr string) { // {'FlightDate': 'DATE', 'UniqueCarrier': 'VARCHAR', 'OriginCityName': 'VARCHAR', 'DestCityName': 'VARCHAR'} diff --git a/core/dbio/iop/duckdb_test.go b/core/dbio/iop/duckdb_test.go index 6281fd075..054746e2b 100644 --- a/core/dbio/iop/duckdb_test.go +++ b/core/dbio/iop/duckdb_test.go @@ -527,9 +527,8 @@ func TestDuckDbDataflowToHttpStream(t *testing.T) { }) t.Run("CSV streaming - max_line_size raised for binary and text columns", func(t *testing.T) { - // Binary columns are hex-encoded (2x byte size) and text-class columns - // (text, ntext, xml, varchar(max), clob) can exceed DuckDB's default - // 2 MB line limit, so both must bump max_line_size to 256 MB (issue #787). + // columns that can hold unbounded values must raise max_line_size, + // else a single large row fails the stream (issue #787) testCases := []struct { name string columnType ColumnType @@ -537,7 +536,9 @@ func TestDuckDbDataflowToHttpStream(t *testing.T) { }{ {"binary column raises limit", BinaryType, "max_line_size=268435456"}, {"text column raises limit", TextType, "max_line_size=268435456"}, - {"string column keeps default", StringType, "max_line_size=2000000"}, + {"string column raises limit", StringType, "max_line_size=268435456"}, + {"json column raises limit", JsonType, "max_line_size=268435456"}, + {"integer column keeps default", IntegerType, "max_line_size=2000000"}, } for _, tc := range testCases { @@ -597,6 +598,44 @@ func TestDuckDbDataflowToHttpStream(t *testing.T) { }) } +func TestDuckDbMaxLineSize(t *testing.T) { + colOf := func(t ColumnType) Columns { + cols := NewColumnsFromFields("id", "payload") + cols[0].Type = IntegerType + cols[1].Type = t + return cols + } + + duckOf := func(props ...string) *DuckDb { + return NewDuckDb(context.Background(), props...) + } + + t.Run("unbounded types raise the limit", func(t *testing.T) { + duck := duckOf() + for _, ct := range []ColumnType{StringType, TextType, JsonType, BinaryType, UUIDType, GeometryType} { + assert.Equal(t, DuckDbLargeMaxLineSize, duck.MaxLineSize(colOf(ct)), "colType=%s", ct) + } + }) + + t.Run("bounded types keep the default", func(t *testing.T) { + duck := duckOf() + for _, ct := range []ColumnType{IntegerType, BigIntType, DecimalType, BoolType, DateType, DatetimeType} { + assert.Equal(t, DuckDbDefaultMaxLineSize, duck.MaxLineSize(colOf(ct)), "colType=%s", ct) + } + }) + + t.Run("max_line_size prop overrides", func(t *testing.T) { + duck := duckOf("max_line_size=999") + assert.Equal(t, 999, duck.MaxLineSize(colOf(TextType))) + assert.Equal(t, 999, duck.MaxLineSize(colOf(IntegerType))) + }) + + t.Run("invalid prop is ignored", func(t *testing.T) { + duck := duckOf("max_line_size=abc") + assert.Equal(t, DuckDbLargeMaxLineSize, duck.MaxLineSize(colOf(TextType))) + }) +} + // 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) { diff --git a/core/sling/task.go b/core/sling/task.go index 120abef46..58f461c9d 100644 --- a/core/sling/task.go +++ b/core/sling/task.go @@ -675,7 +675,7 @@ func ErrorHelper(err error, connTypes ...dbio.Type) (helpString string) { case contains("CSV") && contains("encountered too many errors"): helpString = "Perhaps trying to load with `target_options.format=parquet` could help? This will use Parquet files instead of CSV files." case contains("Maximum line size of", "bytes exceeded"): - helpString = "A row's serialized size exceeded the max_line_size limit of Sling's internal DuckDB CSV bridge. Sling raises this limit automatically when binary or large text-class columns are present in the source schema. If you are still seeing this error, please open an issue @ https://github.com/slingdata-io/sling-cli" + helpString = "A row exceeded the max_line_size limit of Sling's internal DuckDB CSV bridge. Sling raises this limit to 256MB when the source schema has string, text, json or binary columns. For larger values, set the `max_line_size` property in your connection to a higher byte value." case contains("Invalid Input Error: CSV Error on Line:") && usesDuckDb: helpString = "By default, Sling uses CSV serialization to pipe data into DuckDB. Try setting the `copy_method: arrow_http` property in your DuckDB / MotherDuck connection to avoid serialization errors. See https://docs.slingdata.io/connections/database-connections for more details." case contains("it does not have a replica identity and publishes updates"): diff --git a/core/sling/task_test.go b/core/sling/task_test.go index 2a7e62b97..18ef21e2c 100644 --- a/core/sling/task_test.go +++ b/core/sling/task_test.go @@ -15,6 +15,7 @@ func TestErrorHelper(t *testing.T) { t.Run("max_line_size exceeded gets specific help, not arrow_http", func(t *testing.T) { helpString := ErrorHelper(maxLineSizeErr, dbio.TypeDbSQLServer, dbio.TypeFileS3) assert.Contains(t, helpString, "max_line_size") + assert.Contains(t, helpString, "max_line_size` property") assert.NotContains(t, helpString, "arrow_http") }) diff --git a/tests/replications/r.101.duckdb_max_line_size.yaml b/tests/replications/r.101.duckdb_max_line_size.yaml new file mode 100644 index 000000000..79b83f370 --- /dev/null +++ b/tests/replications/r.101.duckdb_max_line_size.yaml @@ -0,0 +1,62 @@ +# issue #787: a row larger than duckdb's 2MB read_csv max_line_size failed the +# stream. Values land in text/json columns, so both must raise the limit. +source: postgres +target: local + +defaults: + mode: full-refresh + +hooks: + start: + - type: query + connection: '{source.name}' + query: | + drop table if exists public.test_max_line_size; + create table public.test_max_line_size ( + id bigint, + big_text text, + big_json jsonb + ); + insert into public.test_max_line_size (id, big_text, big_json) + values ( + 1, + repeat('x', 3000000), + jsonb_build_object('payload', repeat('y', 3000000)) + ); + + end: + - type: check + check: execution.status.error == 0 + on_failure: break + + - type: query + connection: duckdb + query: | + select + length(big_text) as text_len, + length(big_json) as json_len + from read_parquet('/tmp/test_max_line_size.parquet') + into: result + + - type: log + message: 'text_len => {store.result[0].text_len}, json_len => {store.result[0].json_len}' + + # the 3MB values must survive intact, not truncate or fail the stream + - type: check + check: int_parse(store.result[0].text_len) == 3000000 + + - type: check + check: int_parse(store.result[0].json_len) > 3000000 + + - type: log + message: 'SUCCESS: rows larger than the 2MB max_line_size replicated intact' + + - type: query + connection: '{source.name}' + query: drop table if exists public.test_max_line_size + +streams: + public.test_max_line_size: + object: '/tmp/test_max_line_size.parquet' + target_options: + format: parquet diff --git a/tests/suite.cli.yaml b/tests/suite.cli.yaml index de1ca8e66..8d6a179c7 100644 --- a/tests/suite.cli.yaml +++ b/tests/suite.cli.yaml @@ -2634,3 +2634,10 @@ - 'snowflake-adbc' - 'bigquery-adbc' - 'SUCCESS: SLING_USE_ADBC routed postgres, mysql, sqlserver, duckdb, snowflake and bigquery through ADBC' + +# issue #787: rows larger than duckdb's 2MB read_csv max_line_size failed the stream +- id: 317 + name: 'rows larger than the duckdb 2MB max_line_size replicate intact (text + json)' + run: 'sling run -d -r tests/replications/r.101.duckdb_max_line_size.yaml' + output_contains: + - 'SUCCESS: rows larger than the 2MB max_line_size replicated intact' From 668391e09ab319f64ff8a53c8c30193f1fe7f8ed Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Thu, 13 Aug 2026 20:33:19 -0300 Subject: [PATCH 10/16] fix(adbc): target correct catalog for 3-part table names Previously, ADBC ingestion ignored the database/catalog part of a 3-part table name (database.schema.table) and ingested data into the connection's default database. This sets the `Catalog` property in `IngestStreamOptions` and updates the SQL Server metadata template to query the correct database context. Adds a test case to verify ingestion into a non-default database (issue #785). --- core/dbio/database/database_adbc.go | 19 +++--- core/dbio/templates/sqlserver.yaml | 2 +- .../r.102.adbc_three_part_name.yaml | 59 +++++++++++++++++++ tests/suite.cli.yaml | 9 +++ 4 files changed, 80 insertions(+), 9 deletions(-) create mode 100644 tests/replications/r.102.adbc_three_part_name.yaml diff --git a/core/dbio/database/database_adbc.go b/core/dbio/database/database_adbc.go index fa794da09..3e0254b36 100644 --- a/core/dbio/database/database_adbc.go +++ b/core/dbio/database/database_adbc.go @@ -1351,12 +1351,21 @@ func (conn *ArrowDBConn) BulkImportStream(tableFName string, ds *iop.Datastream) return 0, g.Error("ADBC connection is not open") } - // Parse table name to get schema - table, _ := ParseTableName(tableFName, conn.Type) + // Parse table name to get catalog and schema + table, err := ParseTableName(tableFName, conn.Type) + if err != nil { + return count, g.Error(err, "could not parse table name: %s", tableFName) + } // Get ingest mode from property, default to append ingestMode := conn.getIngestMode() + // Target the catalog/schema of the table, not the connection defaults + opts := adbc.IngestStreamOptions{ + Catalog: table.Database, + DBSchema: table.Schema, + } + g.Trace("arrow schema => %s", iop.ColumnsToArrowSchema(ds.Columns)) for batch := range ds.BatchChan { @@ -1366,12 +1375,6 @@ func (conn *ArrowDBConn) BulkImportStream(tableFName string, ds *iop.Datastream) return count, g.Error(err, "error converting batch to Arrow") } - // Ingest using ADBC - opts := adbc.IngestStreamOptions{} - if table.Schema != "" { - opts.DBSchema = table.Schema - } - ingested, err := adbc.IngestStream( conn.Context().Ctx, conn.Conn, diff --git a/core/dbio/templates/sqlserver.yaml b/core/dbio/templates/sqlserver.yaml index 2a0c8fcc0..a2197a492 100755 --- a/core/dbio/templates/sqlserver.yaml +++ b/core/dbio/templates/sqlserver.yaml @@ -175,7 +175,7 @@ metadata: numeric_scale as scale, collation_name, cast(collationProperty(collation_name, 'CodePage') as varchar) as collation_code - from INFORMATION_SCHEMA.COLUMNS + from {{if .database -}} "{database}". {{- end}}INFORMATION_SCHEMA.COLUMNS where table_schema = '{schema}' and table_name = '{table}' order by ordinal_position diff --git a/tests/replications/r.102.adbc_three_part_name.yaml b/tests/replications/r.102.adbc_three_part_name.yaml new file mode 100644 index 000000000..30ea9c8d4 --- /dev/null +++ b/tests/replications/r.102.adbc_three_part_name.yaml @@ -0,0 +1,59 @@ +# issue #785: ADBC ingestion drops the catalog (database) part of the target name. +# The temp table is created in the target database, but the ADBC ingest looks for +# it in the connection's default database, so it fails with "Invalid object name". + +source: LOCAL +target: MSSQL_ADBC + +defaults: + mode: full-refresh + +hooks: + start: + - type: query + connection: '{target.name}' + query: IF DB_ID('slingdb785') IS NULL CREATE DATABASE slingdb785 + + - type: write + content: | + id,name,value + 1,alice,100.5 + 2,bob,200.75 + 3,charlie,300.25 + to: file:///tmp/r.102.adbc_three_part_name.csv + + end: + - type: check + check: execution.status.error == 0 + on_failure: break + + - type: query + connection: '{target.name}' + query: select * from slingdb785.dbo.adbc_three_part order by id + into: result + + - type: log + message: "Query result: {store.result}" + + - type: check + check: int_parse(store.result[0].id) == 1 + message: "Expected first row id to be 1" + + - type: check + check: store.result[0].name == "alice" + message: "Expected first row name to be alice" + + - type: check + check: int_parse(store.result[2].id) == 3 + message: "Expected 3 rows loaded into the non-default database" + + - type: query + connection: '{target.name}' + query: drop table slingdb785.dbo.adbc_three_part + + - type: log + message: "SUCCESS: ADBC loaded into a 3-part (database.schema.table) target name" + +streams: + file:///tmp/r.102.adbc_three_part_name.csv: + object: 'slingdb785.dbo.adbc_three_part' diff --git a/tests/suite.cli.yaml b/tests/suite.cli.yaml index 8d6a179c7..b878795a1 100644 --- a/tests/suite.cli.yaml +++ b/tests/suite.cli.yaml @@ -2641,3 +2641,12 @@ run: 'sling run -d -r tests/replications/r.101.duckdb_max_line_size.yaml' output_contains: - 'SUCCESS: rows larger than the 2MB max_line_size replicated intact' + +# issue #785: ADBC ingestion never sets the catalog, so a 3-part target name +# writes the temp table to one database and ingests into the default one. +# Requires the mssql ADBC driver (dbc install mssql) +- id: 318 + name: 'ADBC ingests into a 3-part (database.schema.table) target name' + run: 'sling run -d -r tests/replications/r.102.adbc_three_part_name.yaml' + output_contains: + - 'SUCCESS: ADBC loaded into a 3-part (database.schema.table) target name' From ceeea25cecca793d6b0300e559a3f23fa420bd11 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Thu, 13 Aug 2026 23:21:10 -0300 Subject: [PATCH 11/16] fix(parquet): support time and uuid column types in writer createBuilder had no case for TIME32, TIME64, or EXTENSION (UUID) arrow types, so they fell through to a string builder while the schema declared them as time/uuid. This caused a panic when building the record due to the type mismatch, making any parquet write with a time or uuid column fail. Add explicit cases for TIME32/TIME64 builders and use the generic array.NewBuilder for EXTENSION types so arrow selects the proper builder (e.g. UUIDBuilder). Added a test that writes and reads back time, timez, and uuid columns, including null values. --- core/dbio/iop/parquet_arrow.go | 8 +++ core/dbio/iop/parquet_arrow_test.go | 78 +++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/core/dbio/iop/parquet_arrow.go b/core/dbio/iop/parquet_arrow.go index 1eed2506a..b451b04f0 100644 --- a/core/dbio/iop/parquet_arrow.go +++ b/core/dbio/iop/parquet_arrow.go @@ -283,10 +283,18 @@ func (p *ParquetArrowWriter) createBuilder(dtype arrow.DataType) array.Builder { return array.NewDate32Builder(p.mem) case arrow.TIMESTAMP: return array.NewTimestampBuilder(p.mem, dtype.(*arrow.TimestampType)) + case arrow.TIME32: + return array.NewTime32Builder(p.mem, dtype.(*arrow.Time32Type)) + case arrow.TIME64: + return array.NewTime64Builder(p.mem, dtype.(*arrow.Time64Type)) case arrow.STRING: return array.NewStringBuilder(p.mem) case arrow.BINARY: return array.NewBinaryBuilder(p.mem, dtype.(*arrow.BinaryType)) + case arrow.EXTENSION: + // arrow picks the extension's own builder (e.g. UUIDBuilder), which the + // generic ExtensionBuilder would not give us + return array.NewBuilder(p.mem, dtype) default: return array.NewStringBuilder(p.mem) } diff --git a/core/dbio/iop/parquet_arrow_test.go b/core/dbio/iop/parquet_arrow_test.go index af109386f..5fcd2d7af 100644 --- a/core/dbio/iop/parquet_arrow_test.go +++ b/core/dbio/iop/parquet_arrow_test.go @@ -416,3 +416,81 @@ func TestDecimal128ToString(t *testing.T) { }) } } + +// createBuilder had no case for time or uuid columns, so it fell through to a +// string builder while the schema declared time64/uuid. Building the record +// then panicked on the type mismatch, making any parquet write with a time or +// uuid column fail. +func TestParquetArrowWriterTimeAndUUID(t *testing.T) { + columns := NewColumns( + Columns{ + {Name: "col_id", Type: BigIntType}, + {Name: "col_time", Type: TimeType}, + {Name: "col_timez", Type: TimezType}, + {Name: "col_uuid", Type: UUIDType}, + }..., + ) + + rows := [][]any{ + {int64(1), "10:00:00", "10:00:00", "6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + {int64(2), "23:59:59", "00:00:01", "6ba7b811-9dad-11d1-80b4-00c04fd430c8"}, + {int64(3), nil, nil, nil}, + } + + testFile := "/tmp/test_time_uuid.parquet" + f, err := os.Create(testFile) + assert.NoError(t, err) + defer f.Close() + defer os.Remove(testFile) + + pw, err := NewParquetArrowWriter(f, columns, compress.Codecs.Snappy) + assert.NoError(t, err) + + for _, row := range rows { + assert.NoError(t, pw.WriteRow(row)) + } + assert.NoError(t, pw.Close()) + + stat, err := os.Stat(testFile) + assert.NoError(t, err) + assert.Greater(t, stat.Size(), int64(0)) + + f2, err := os.Open(testFile) + assert.NoError(t, err) + defer f2.Close() + + reader, err := NewParquetArrowReader(f2, nil) + assert.NoError(t, err) + + readCols := reader.Columns() + assert.Equal(t, len(columns), len(readCols)) + + table, err := reader.Reader.ReadTable(context.Background()) + assert.NoError(t, err) + defer table.Release() + assert.Equal(t, len(rows), int(table.NumRows())) + + for rowIdx, originalRow := range rows { + for colIdx, col := range columns { + chunk := table.Column(colIdx).Data().Chunk(0) + read := GetValueFromArrowArray(chunk, rowIdx) + + if originalRow[colIdx] == nil { + assert.Nil(t, read, "row %d, %s: expected nil", rowIdx, col.Name) + continue + } + assert.NotNil(t, read, "row %d, col %s (%s): expected a value", rowIdx, col.Name, col.Type) + t.Logf("row %d %s = %v", rowIdx, col.Name, read) + + // times land as time.Time, uuid as its canonical string + switch col.Type { + case TimeType, TimezType: + assert.Contains(t, cast.ToString(read), cast.ToString(originalRow[colIdx]), + "row %d, %s value mismatch", rowIdx, col.Name) + case UUIDType: + assert.Equal(t, originalRow[colIdx], cast.ToString(read), + "row %d, %s value mismatch", rowIdx, col.Name) + } + } + } +} From eb13a603b873110e485d7bfde5a3cfb7fa39f016 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Fri, 14 Aug 2026 08:04:29 -0300 Subject: [PATCH 12/16] fix: defer reader opening until consumption in MergeReaders Open remote reader bodies lazily via ReaderReady.Open to prevent idle connections from being reset before they are consumed. Previously, readers were opened eagerly and queued in a channel, leaving remote HTTP bodies idle until the pipe consumer reached them. --- core/dbio/filesys/fs.go | 73 ++-- core/dbio/filesys/fs_merge_readers_test.go | 332 ++++++++++++++++++ core/dbio/iop/datastream.go | 159 ++++++++- core/dbio/iop/datastream_test.go | 41 +++ .../p.49.merge_readers_many_files.yaml | 62 ++++ tests/suite.cli.yaml | 8 + 6 files changed, 646 insertions(+), 29 deletions(-) create mode 100644 core/dbio/filesys/fs_merge_readers_test.go create mode 100644 tests/pipelines/p.49.merge_readers_many_files.yaml diff --git a/core/dbio/filesys/fs.go b/core/dbio/filesys/fs.go index 0c2c15d15..068a33b51 100755 --- a/core/dbio/filesys/fs.go +++ b/core/dbio/filesys/fs.go @@ -1718,6 +1718,10 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, cfg concurrency = 3 } + // CSV/JSON are read one file at a time. XML is copied one file at a time. + // Open the body at first read so unread remote bodies do not sit idle. + channelConsume := g.In(fileType, dbio.FileTypeCsv, dbio.FileTypeJson, dbio.FileTypeJsonLines, dbio.FileTypeGeojson) + g.Debug("merging %s readers of %d files [concurrency=%d] from %s", fileType, len(nodes), concurrency, url) readerChn := make(chan *iop.ReaderReady, concurrency) go func() { @@ -1729,40 +1733,40 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, cfg continue } - ds.Context.Wg.Read.Add() - go func(node FileNode) { - defer ds.Context.Wg.Read.Done() - - if !includeAll { - _, uriExclude := excludeMap[node.URI] - _, pathExclude := excludeMap[node.Path()] - _, uriInclude := includeMap[node.URI] - _, pathInclude := includeMap[node.Path()] + if !includeAll { + _, uriExclude := excludeMap[node.URI] + _, pathExclude := excludeMap[node.Path()] + _, uriInclude := includeMap[node.URI] + _, pathInclude := includeMap[node.Path()] - if (uriExclude || pathExclude) || (!uriInclude && !pathInclude) { - g.Debug("skipping %s", node.URI) - return - } + if (uriExclude || pathExclude) || (!uriInclude && !pathInclude) { + g.Debug("skipping %s", node.URI) + continue } + } + node := node + r := &iop.ReaderReady{URI: node.URI} + r.Open = func() (io.Reader, error) { g.Debug("processing reader from %s", node.URI) - reader, err := fs.Self().GetReader(node.URI) if err != nil { - setError(g.Error(err, "Error getting reader")) - return + err = g.Error(err, "Error getting reader") + setError(err) + return nil, err } + return reader, nil + } - r := &iop.ReaderReady{Reader: reader, URI: node.URI} - readerChn <- r - }(node) + select { + case readerChn <- r: + case <-ds.Context.Ctx.Done(): + return + } } - - ds.Context.Wg.Read.Wait() - }() - if g.In(fileType, dbio.FileTypeCsv, dbio.FileTypeJson, dbio.FileTypeJsonLines, dbio.FileTypeGeojson) { + if channelConsume { pipeW.Close() switch fileType { @@ -1778,13 +1782,30 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, cfg defer pipeW.Close() for reader := range readerChn { - _, err = io.Copy(pipeW, reader.Reader) - if err != nil { - setError(g.Error(err, "Error copying reader to pipe writer")) + src, rerr := reader.GetReader() + if rerr != nil { + reader.Close() + setError(g.Error(rerr, "Error getting reader")) + for leftover := range readerChn { + leftover.Close() + } + return + } + + _, rerr = io.Copy(pipeW, src) + reader.Close() + if rerr != nil { + setError(g.Error(rerr, "Error copying reader to pipe writer")) + for leftover := range readerChn { + leftover.Close() + } return } if cfg.Limit > 0 && (ds.Limited(cfg.Limit) || len(ds.Buffer) >= cfg.Limit) { + for leftover := range readerChn { + leftover.Close() + } return } } diff --git a/core/dbio/filesys/fs_merge_readers_test.go b/core/dbio/filesys/fs_merge_readers_test.go new file mode 100644 index 000000000..c9e3ede54 --- /dev/null +++ b/core/dbio/filesys/fs_merge_readers_test.go @@ -0,0 +1,332 @@ +package filesys + +import ( + "bytes" + "context" + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/flarco/g" + "github.com/slingdata-io/sling-cli/core/dbio" + "github.com/slingdata-io/sling-cli/core/dbio/iop" + "github.com/stretchr/testify/assert" +) + +// idleTrackingReader emulates an S3 body: it records the idle time before the +// first read, and fails like a reset connection once that exceeds the window. +type idleTrackingReader struct { + data []byte + pos int + uri string + openedAt time.Time + idleWindow time.Duration + firstRead bool + dead bool + closed bool + forceClosed bool + fs *idleSensitiveFS +} + +func (r *idleTrackingReader) close() { + if r.closed { + return + } + r.closed = true + + r.fs.mu.Lock() + r.fs.curOpen-- + r.fs.mu.Unlock() +} + +func (r *idleTrackingReader) Close() error { + r.forceClosed = true + r.close() + return nil +} + +func (r *idleTrackingReader) Read(p []byte) (int, error) { + // Close() on an SFTP pipe makes a later Read fail this way. + // EOF cleanup must not; Peek reads to EOF then reads again. + if r.forceClosed { + return 0, io.ErrClosedPipe + } + + if !r.firstRead { + r.firstRead = true + idle := time.Since(r.openedAt) + + r.fs.mu.Lock() + if idle > r.fs.maxIdle { + r.fs.maxIdle = idle + r.fs.maxIdleURI = r.uri + } + r.fs.mu.Unlock() + + if idle > r.idleWindow { + r.dead = true + } + + // pause after the first read so a body that was opened too early + // sits idle while this file is consumed (Start() also hits this). + if r.fs.readPause > 0 { + time.Sleep(r.fs.readPause) + } + } + + if r.dead { + return 0, fmt.Errorf("read tcp 10.0.0.1:54567->52.216.0.1:443: read: connection reset by peer") + } + if r.pos >= len(r.data) { + r.close() + return 0, io.EOF + } + n := copy(p, r.data[r.pos:]) + r.pos += n + return n, nil +} + +// idleSensitiveFS hands back bodies that go stale if not consumed promptly. +type idleSensitiveFS struct { + LocalFileSysClient + contents map[string][]byte + idleWindow time.Duration + readPause time.Duration + + mu sync.Mutex + openCount int + curOpen int + maxOpen int + maxIdle time.Duration + maxIdleURI string +} + +func (fs *idleSensitiveFS) Init(ctx context.Context) (err error) { + var instance FileSysClient = fs + fs.BaseFileSysClient.instance = &instance + fs.BaseFileSysClient.context = g.NewContext(ctx) + fs.BaseFileSysClient.fsType = dbio.TypeFileS3 + fs.BaseFileSysClient.properties = map[string]string{} + return +} + +func (fs *idleSensitiveFS) GetReader(uri string) (io.Reader, error) { + data, ok := fs.contents[uri] + if !ok { + return nil, fmt.Errorf("no such object: %s", uri) + } + + fs.mu.Lock() + fs.openCount++ + fs.curOpen++ + if fs.curOpen > fs.maxOpen { + fs.maxOpen = fs.curOpen + } + fs.mu.Unlock() + + return &idleTrackingReader{ + data: data, + uri: uri, + openedAt: time.Now(), + idleWindow: fs.idleWindow, + fs: fs, + }, nil +} + +type idleFileKind int + +const ( + idleCSV idleFileKind = iota + idleJSONL + idleXML +) + +func newIdleSensitiveFS(t *testing.T, numFiles, rowsPerFile int, idleWindow time.Duration, kind idleFileKind) (*idleSensitiveFS, FileNodes) { + t.Helper() + + contents := map[string][]byte{} + nodes := FileNodes{} + for i := 0; i < numFiles; i++ { + var uri string + var buf bytes.Buffer + switch kind { + case idleJSONL: + uri = fmt.Sprintf("s3://test-bucket/unload/u01-%04d_part_00.jsonl", i) + for r := 0; r < rowsPerFile; r++ { + fmt.Fprintf(&buf, "{\"id\":%d,\"name\":\"name_%d_%d\"}\n", i*rowsPerFile+r, i, r) + } + case idleXML: + uri = fmt.Sprintf("s3://test-bucket/unload/u01-%04d_part_00.xml", i) + buf.WriteString("\n") + for r := 0; r < rowsPerFile; r++ { + fmt.Fprintf(&buf, "%dname_%d_%d\n", i*rowsPerFile+r, i, r) + } + buf.WriteString("\n") + default: + uri = fmt.Sprintf("s3://test-bucket/unload/u01-%04d_part_00.csv", i) + buf.WriteString("id,name\n") + for r := 0; r < rowsPerFile; r++ { + fmt.Fprintf(&buf, "%d,name_%d_%d\n", i*rowsPerFile+r, i, r) + } + } + + contents[uri] = buf.Bytes() + nodes = append(nodes, FileNode{URI: uri, Size: uint64(buf.Len())}) + } + + fs := &idleSensitiveFS{contents: contents, idleWindow: idleWindow} + if err := fs.Init(context.Background()); err != nil { + t.Fatalf("init: %v", err) + } + fs.SetProp("url", "s3://test-bucket/unload") + + return fs, nodes +} + +func readAll(t *testing.T, fs *idleSensitiveFS, nodes FileNodes, fileType dbio.FileType) (count int, err error) { + t.Helper() + + ds, err := MergeReaders(fs, fileType, nodes, iop.FileStreamConfig{}) + if err != nil { + t.Fatalf("MergeReaders: %v", err) + } + + for range ds.Rows() { + count++ + } + + return count, ds.Err() +} + +func (fs *idleSensitiveFS) snapshot() (maxOpen int, maxIdle time.Duration, maxIdleURI string, curOpen int) { + fs.mu.Lock() + defer fs.mu.Unlock() + return fs.maxOpen, fs.maxIdle, fs.maxIdleURI, fs.curOpen +} + +// TestMergeReadersIdleBoundedIssue789 guards issue #789. MergeReaders used to +// open every file's body up front while the consumer read one at a time. +func TestMergeReadersIdleBoundedIssue789(t *testing.T) { + const numFiles = 40 + const rowsPerFile = 200 + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Hour, idleCSV) + + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + assert.NoError(t, err) + assert.Equal(t, numFiles*rowsPerFile, count) + + maxOpen, maxIdle, maxIdleURI, curOpen := fs.snapshot() + t.Logf("max open at once: %d, longest idle: %v (%s)", maxOpen, maxIdle.Round(time.Millisecond), maxIdleURI) + + // current file plus at most one transition. 20 idle bodies is the old bug. + assert.LessOrEqual(t, maxOpen, 2, "too many bodies open at once: %d", maxOpen) + assert.Less(t, maxIdle, time.Second, + "a body idled %v before its first read; open the body at first read", + maxIdle.Round(time.Millisecond)) + assert.Equal(t, 0, curOpen, "bodies must be closed after use") +} + +// TestMergeReadersIdleDoesNotGrowWithFileCount is the core of the #789 fix: +// a larger unload must not mean a longer idle wait. +func TestMergeReadersIdleDoesNotGrowWithFileCount(t *testing.T) { + measure := func(numFiles int) time.Duration { + fs, nodes := newIdleSensitiveFS(t, numFiles, 200, time.Hour, idleCSV) + + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + assert.NoError(t, err) + assert.Equal(t, numFiles*200, count) + + _, maxIdle, _, _ := fs.snapshot() + return maxIdle + } + + small := measure(10) + large := measure(40) + + t.Logf("longest idle: 10 files=%v, 40 files=%v", + small.Round(time.Millisecond), large.Round(time.Millisecond)) + + // both must stay far below a reset window; do not ratio two near-zero times + assert.Less(t, small, 200*time.Millisecond, "10-file idle %v is not bounded", small) + assert.Less(t, large, 200*time.Millisecond, "40-file idle %v is not bounded", large) +} + +// TestMergeReadersSurvivesIdleReset is the end-to-end guard: with a window +// that a reset would trip, the stream must still complete. +func TestMergeReadersSurvivesIdleReset(t *testing.T) { + const numFiles = 40 + const rowsPerFile = 200 + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Second, idleCSV) + fs.readPause = 100 * time.Millisecond + + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + + assert.NoError(t, err, "stream must not die from a reset idle body") + assert.Equal(t, numFiles*rowsPerFile, count, "all rows should be read") +} + +// TestMergeReadersSlowConsumerDoesNotResetLookahead fails if the next body +// is opened while the current file is still being read. Prefetch of live +// bodies sits idle for the pause (2s) and trips the 1s window. +func TestMergeReadersSlowConsumerDoesNotResetLookahead(t *testing.T) { + const numFiles = 4 + const rowsPerFile = 200 + const pause = 2 * time.Second + const idleWindow = time.Second + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, idleWindow, idleCSV) + fs.readPause = pause + + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + + assert.NoError(t, err, "look-ahead opened a body that sat idle for %v", pause) + assert.Equal(t, numFiles*rowsPerFile, count) + + maxOpen, maxIdle, maxIdleURI, _ := fs.snapshot() + t.Logf("max open at once: %d, longest idle: %v (%s)", maxOpen, maxIdle.Round(time.Millisecond), maxIdleURI) + + assert.LessOrEqual(t, maxOpen, 2, "look-ahead opened extra bodies: maxOpen=%d", maxOpen) + assert.Less(t, maxIdle, idleWindow, + "a body idled %v; open the body at first read", maxIdle.Round(time.Millisecond)) +} + +func TestMergeReadersJsonLinesIdleBounded(t *testing.T) { + const numFiles = 10 + const rowsPerFile = 50 + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Second, idleJSONL) + + count, err := readAll(t, fs, nodes, dbio.FileTypeJsonLines) + assert.NoError(t, err) + assert.Equal(t, numFiles*rowsPerFile, count) + + maxOpen, maxIdle, _, curOpen := fs.snapshot() + assert.LessOrEqual(t, maxOpen, 2, "too many JSONL bodies open at once: %d", maxOpen) + assert.Less(t, maxIdle, time.Second, "JSONL body idled %v", maxIdle.Round(time.Millisecond)) + assert.Equal(t, 0, curOpen, "JSONL bodies must be closed after use") +} + +// TestMergeReadersXmlOpensNearPointOfUse covers the pipe path. Producers used +// to call GetReader for every file, then copy one by one. +func TestMergeReadersXmlOpensNearPointOfUse(t *testing.T) { + const numFiles = 20 + const rowsPerFile = 10 + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Hour, idleXML) + + _, err := readAll(t, fs, nodes, dbio.FileTypeXml) + assert.NoError(t, err) + + maxOpen, _, _, _ := fs.snapshot() + t.Logf("xml max open at once: %d (openCount=%d)", maxOpen, func() int { + fs.mu.Lock() + defer fs.mu.Unlock() + return fs.openCount + }()) + + assert.LessOrEqual(t, maxOpen, 2, "XML pipe path opened %d bodies at once", maxOpen) +} diff --git a/core/dbio/iop/datastream.go b/core/dbio/iop/datastream.go index d532bec88..d317fe6be 100644 --- a/core/dbio/iop/datastream.go +++ b/core/dbio/iop/datastream.go @@ -1193,17 +1193,98 @@ func (ds *Datastream) ConsumeXmlReader(reader io.Reader) (err error) { type ReaderReady struct { Reader io.Reader URI string + + // Open obtains the reader at first use. Remote stores reset an unread body. + Open func() (io.Reader, error) + + mu sync.Mutex + opened bool + closed bool + err error +} + +// GetReader returns the reader. It opens on first use. A failed Open is retried. +func (rr *ReaderReady) GetReader() (io.Reader, error) { + if rr == nil { + return nil, g.Error("nil reader") + } + if rr.Open == nil { + return rr.Reader, nil + } + + rr.mu.Lock() + defer rr.mu.Unlock() + + if rr.closed { + return nil, g.Error("reader is closed") + } + if rr.opened { + return rr.Reader, rr.err + } + + reader, err := rr.Open() + if err != nil { + // do not cache; the consumer retries at the point of use + rr.err = err + return nil, err + } + + rr.Reader = reader + rr.err = nil + rr.opened = true + return rr.Reader, nil +} + +// Close closes an unused or finished body. Safe to call more than once. +func (rr *ReaderReady) Close() error { + if rr == nil { + return nil + } + + rr.mu.Lock() + defer rr.mu.Unlock() + + if rr.closed { + return nil + } + rr.closed = true + + if c, ok := rr.Reader.(io.Closer); ok && rr.Reader != nil { + return c.Close() + } + return nil +} + +func drainReaderReadyCh(ch <-chan *ReaderReady) { + if ch == nil { + return + } + for rr := range ch { + rr.Close() + } } func (ds *Datastream) ConsumeJsonReaderChl(readerChn chan *ReaderReady, isXML bool) (err error) { + var current *ReaderReady + nextJSON := func(reader *ReaderReady) (*jsonStream, error) { + if current != nil { + current.Close() + } + current = reader // set URI ds.Metadata.StreamURL.Value = reader.URI + // open now that we are ready to consume + reader1, err := reader.GetReader() + if err != nil { + return nil, g.Error(err, "could not get reader") + } + // decompress if needed - reader2, err := AutoDecompress(reader.Reader) + reader2, err := AutoDecompress(reader1) if err != nil { return nil, g.Error(err, "could not auto-decompress") } @@ -1224,12 +1305,34 @@ func (ds *Datastream) ConsumeJsonReaderChl(readerChn chan *ReaderReady, isXML bo return jsNew, nil } - js, err := nextJSON(<-readerChn) + first, ok := <-readerChn + if !ok || first == nil { + ds.SetReady() + ds.Close() + return nil + } + + js, err := nextJSON(first) if err != nil { + if current != nil { + current.Close() + } + go drainReaderReadyCh(readerChn) return } + stopJSON := func() { + if current != nil { + current.Close() + current = nil + } + js = nil + } + nextFunc := func(it *Iterator) bool { + if js == nil { + return false + } processNext: hasNext := js.NextFunc(it) @@ -1238,12 +1341,15 @@ func (ds *Datastream) ConsumeJsonReaderChl(readerChn chan *ReaderReady, isXML bo // next reader for reader := range readerChn { if reader == nil { + stopJSON() return false } jsNew, err := nextJSON(reader) if err != nil { it.ds.Context.CaptureErr(g.Error(err, "Error getting next reader")) + stopJSON() + go drainReaderReadyCh(readerChn) return false } @@ -1254,6 +1360,9 @@ func (ds *Datastream) ConsumeJsonReaderChl(readerChn chan *ReaderReady, isXML bo goto processNext } + stopJSON() + } else if !hasNext { + stopJSON() } // set stream url @@ -1294,12 +1403,22 @@ func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan *ReaderReady) (err erro c.Delimiter = ds.config.Delimiter } + var current *ReaderReady + nextCSV := func(reader *ReaderReady) (r csv.CsvReaderLike, err error) { - c.Reader = reader.Reader + if current != nil { + current.Close() + } + current = reader // set URI ds.Metadata.StreamURL.Value = reader.URI + // open now that we are ready to consume + if c.Reader, err = reader.GetReader(); err != nil { + return r, g.Error(err, "could not get reader") + } + // decompress if needed readerDecompr, err := AutoDecompress(c.Reader) if err != nil { @@ -1325,8 +1444,16 @@ func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan *ReaderReady) (err erro for reader := range readerChn { r, err = nextCSV(reader) if err != nil { + if current != nil { + current.Close() + } + go drainReaderReadyCh(readerChn) return } else if r == nil { + if current != nil { + current.Close() + } + go drainReaderReadyCh(readerChn) return g.Error("no reader was returned for: %s", ds.Metadata.StreamURL.Value) } @@ -1341,6 +1468,10 @@ func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan *ReaderReady) (err erro } else if err != nil { err = g.Error(err, "could not parse header line") ds.Context.CaptureErr(err) + if current != nil { + current.Close() + } + go drainReaderReadyCh(readerChn) return err } @@ -1369,12 +1500,23 @@ func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan *ReaderReady) (err erro for reader := range readerChn { if reader == nil { + if current != nil { + current.Close() + current = nil + } + r = nil return false } r, err = nextCSV(reader) if err != nil { it.ds.Context.CaptureErr(g.Error(err, "Error getting next reader")) + if current != nil { + current.Close() + current = nil + } + r = nil + go drainReaderReadyCh(readerChn) return false } else if r == nil { continue @@ -1413,9 +1555,20 @@ func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan *ReaderReady) (err erro goto processNext } + if current != nil { + current.Close() + current = nil + } + r = nil return false } else if err != nil { it.ds.Context.CaptureErr(g.Error(err, "Error reading file")) + if current != nil { + current.Close() + current = nil + } + r = nil + go drainReaderReadyCh(readerChn) return false } diff --git a/core/dbio/iop/datastream_test.go b/core/dbio/iop/datastream_test.go index acea501e4..72e2166ff 100644 --- a/core/dbio/iop/datastream_test.go +++ b/core/dbio/iop/datastream_test.go @@ -2,7 +2,9 @@ package iop import ( "encoding/json" + "errors" "io" + "strings" "testing" "github.com/flarco/g/csv" @@ -190,3 +192,42 @@ func TestEncodeRowAsJSONObject(t *testing.T) { }) } } + +func TestReaderReadyRetriesFailedOpenAndClose(t *testing.T) { + opens := 0 + rr := &ReaderReady{ + URI: "s3://bucket/file.csv", + Open: func() (io.Reader, error) { + opens++ + if opens == 1 { + return nil, errors.New("temporary") + } + return io.NopCloser(strings.NewReader("ok")), nil + }, + } + + if _, err := rr.GetReader(); err == nil { + t.Fatal("expected first Open to fail") + } + r, err := rr.GetReader() + if err != nil { + t.Fatalf("retry should succeed: %v", err) + } + if opens != 2 { + t.Fatalf("opens=%d, want 2", opens) + } + buf := make([]byte, 2) + n, _ := r.Read(buf) + if string(buf[:n]) != "ok" { + t.Fatalf("got %q", buf[:n]) + } + if err := rr.Close(); err != nil { + t.Fatal(err) + } + if err := rr.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + if _, err := rr.GetReader(); err == nil { + t.Fatal("GetReader after Close should fail") + } +} diff --git a/tests/pipelines/p.49.merge_readers_many_files.yaml b/tests/pipelines/p.49.merge_readers_many_files.yaml new file mode 100644 index 000000000..b89fd6a0a --- /dev/null +++ b/tests/pipelines/p.49.merge_readers_many_files.yaml @@ -0,0 +1,62 @@ +# issue #789: row integrity across many merged CSV files. +# Idle-reset coverage is in core/dbio/filesys/fs_merge_readers_test.go. +steps: + - type: query + connection: POSTGRES + query: drop table if exists public.issue789_many_files + + # write many small files, like a redshift unload + - type: replication + replication: + source: POSTGRES + target: LOCAL + defaults: + mode: full-refresh + target_options: + file_max_rows: 500 + format: csv + streams: + gen_data: + sql: | + select g as id, md5(g::text) as name + from generate_series(1, 8000) g + object: file://temp/issue789_suite/data.csv + + # read them back through MergeReaders + - type: replication + replication: + source: LOCAL + target: POSTGRES + defaults: + mode: full-refresh + streams: + file://temp/issue789_suite/data.csv: + object: public.issue789_many_files + + - type: query + connection: POSTGRES + query: | + select count(*) as cnt, count(distinct id) as ids, sum(id) as total + from public.issue789_many_files + into: result + + - type: log + message: 'rows={store.result[0].cnt} distinct={store.result[0].ids} sum={store.result[0].total}' + + - type: check + check: int_parse(store.result[0].cnt) == 8000 + failure_message: 'expected 8000 rows, got {store.result[0].cnt}' + + - type: check + check: int_parse(store.result[0].ids) == 8000 + failure_message: 'expected 8000 distinct ids, got {store.result[0].ids}' + + # 1..8000 sums to 32004000, so nothing is dropped or duplicated + - type: check + check: int_parse(store.result[0].total) == 32004000 + failure_message: 'id sum was {store.result[0].total}, expected 32004000' + success_message: 'SUCCESS: all 8000 rows read intact from many merged files' + + - type: query + connection: POSTGRES + query: drop table if exists public.issue789_many_files diff --git a/tests/suite.cli.yaml b/tests/suite.cli.yaml index b878795a1..285beea08 100644 --- a/tests/suite.cli.yaml +++ b/tests/suite.cli.yaml @@ -2650,3 +2650,11 @@ run: 'sling run -d -r tests/replications/r.102.adbc_three_part_name.yaml' output_contains: - 'SUCCESS: ADBC loaded into a 3-part (database.schema.table) target name' + +# issue #789: row-integrity check across many merged CSV files. The idle-reset +# guard lives in core/dbio/filesys/fs_merge_readers_test.go (local disk does not reset). +- id: 319 + name: 'many merged files read intact (issue #789)' + run: 'sling run -d -p tests/pipelines/p.49.merge_readers_many_files.yaml' + output_contains: + - 'SUCCESS: all 8000 rows read intact from many merged files' From 7c7b30ff7f4bc816b6a1661df4a8b37330fc71f6 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Fri, 14 Aug 2026 08:30:25 -0300 Subject: [PATCH 13/16] feat(sqlserver): add support for named instance handling and related error messages --- core/dbio/connection/connection.go | 28 +++ core/dbio/connection/connection_test.go | 215 +++++++++++++++++++++++ core/dbio/database/database_sqlserver.go | 11 ++ core/sling/task.go | 2 + core/sling/task_test.go | 7 + 5 files changed, 263 insertions(+) diff --git a/core/dbio/connection/connection.go b/core/dbio/connection/connection.go index be1de3847..d1fbb5929 100644 --- a/core/dbio/connection/connection.go +++ b/core/dbio/connection/connection.go @@ -552,6 +552,20 @@ func (c *Connection) ConnSetDatabase(dbName string) *Connection { return &c2 } +// splitSQLServerHostInstance splits host\instance or host/instance. +func splitSQLServerHostInstance(host string) (hostNew, instance string) { + host = strings.TrimSpace(host) + if host == "" || strings.HasPrefix(host, "[") { + return host, "" + } + normalized := strings.ReplaceAll(host, `\`, "/") + i := strings.LastIndex(normalized, "/") + if i <= 0 || i == len(normalized)-1 { + return host, "" + } + return normalized[:i], normalized[i+1:] +} + func (c *Connection) setURL() (err error) { c.setFromEnv() c.setUseADBC() @@ -951,6 +965,13 @@ func (c *Connection) setURL() (err error) { setIfMissing("password", "") setIfMissing("app_name", "sling") + if host, inst := splitSQLServerHostInstance(cast.ToString(c.Data["host"])); host != "" { + c.Data["host"] = host + if inst != "" { + setIfMissing("instance", inst) + } + } + template = "sqlserver://{username}:{password}@{host}" if c.Type == dbio.TypeDbFabric { template = "fabric://{username}:{password}@{host}" @@ -958,8 +979,15 @@ func (c *Connection) setURL() (err error) { _, port_ok := c.Data["port"] _, instance_ok := c.Data["instance"] + buildingURL := cast.ToString(c.Data["url"]) == "" switch { + case port_ok && instance_ok: + template += ":{port}/{instance}" + g.Debug("SQL Server: port %s and instance %s are both set. The driver uses the port and ignores the instance name.", c.Data["port"], c.Data["instance"]) + if buildingURL && cast.ToInt(c.Data["port"]) == 1433 { + g.Warn("SQL Server: port 1433 and instance %s are both set. The driver uses port 1433 and ignores the instance name. For a named instance, set `port` to the instance TCP port or omit `port`.", c.Data["instance"]) + } case port_ok: template += ":{port}" case instance_ok: diff --git a/core/dbio/connection/connection_test.go b/core/dbio/connection/connection_test.go index 033f33dc2..f5aa43267 100644 --- a/core/dbio/connection/connection_test.go +++ b/core/dbio/connection/connection_test.go @@ -1,10 +1,14 @@ package connection import ( + "strings" "testing" "github.com/flarco/g" + "github.com/microsoft/go-mssqldb/msdsn" + "github.com/slingdata-io/sling-cli/core/dbio" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestConnectionDiscover(t *testing.T) { @@ -272,3 +276,214 @@ func TestQueryURL(t *testing.T) { // println(url.QueryEscape(password)) _ = password } + +// GitHub #780: named-instance URL building. +// The driver calls SQL Browser (UDP 1434) only when instance is set and port is 0. +// If a port is set, the driver uses that port and ignores the instance name. +func TestSQLServerNamedInstance(t *testing.T) { + type want struct { + host string + port uint64 + instance string + database string + callBrowser bool + dataHost string + dataInstance string + urlHas []string + urlNotHas []string + } + + base := map[string]any{ + "type": "sqlserver", + "user": "the_user", + "password": "secret", + "database": "dbname", + } + + with := func(extra map[string]any) map[string]any { + data := map[string]any{} + for k, v := range base { + data[k] = v + } + for k, v := range extra { + data[k] = v + } + return data + } + + cases := []struct { + name string + data map[string]any + want want + }{ + { + name: "instance only does not inject default port", + data: with(map[string]any{ + "host": "THEHOST", + "instance": "Instance", + }), + want: want{ + host: "THEHOST", + port: 0, + instance: "Instance", + database: "dbname", + callBrowser: true, + dataHost: "THEHOST", + dataInstance: "Instance", + urlHas: []string{"@THEHOST/Instance"}, + urlNotHas: []string{":1433"}, + }, + }, + { + name: "port only connects to that port", + data: with(map[string]any{ + "host": "THEHOST", + "port": 1433, + }), + want: want{ + host: "THEHOST", + port: 1433, + instance: "", + database: "dbname", + callBrowser: false, + dataHost: "THEHOST", + urlHas: []string{"@THEHOST:1433"}, + }, + }, + { + name: "port and instance keep both in the url", + data: with(map[string]any{ + "host": "THEHOST", + "port": 1433, + "instance": "Instance", + }), + want: want{ + host: "THEHOST", + port: 1433, + instance: "Instance", + database: "dbname", + callBrowser: false, + dataHost: "THEHOST", + dataInstance: "Instance", + urlHas: []string{"@THEHOST:1433/Instance"}, + }, + }, + { + name: "host slash instance without port", + data: with(map[string]any{ + "host": "THEHOST/Instance", + }), + want: want{ + host: "THEHOST", + port: 0, + instance: "Instance", + database: "dbname", + callBrowser: true, + dataHost: "THEHOST", + dataInstance: "Instance", + urlHas: []string{"@THEHOST/Instance"}, + urlNotHas: []string{"Instance:1433", "@THEHOST/Instance:1433"}, + }, + }, + { + name: "host backslash instance without port", + data: with(map[string]any{ + "host": `THEHOST\Instance`, + }), + want: want{ + host: "THEHOST", + port: 0, + instance: "Instance", + database: "dbname", + callBrowser: true, + dataHost: "THEHOST", + dataInstance: "Instance", + urlHas: []string{"@THEHOST/Instance"}, + urlNotHas: []string{`THEHOST\Instance`, "Instance:1433"}, + }, + }, + { + name: "host slash instance with explicit port", + data: with(map[string]any{ + "host": "THEHOST/Instance", + "port": 51433, + }), + want: want{ + host: "THEHOST", + port: 51433, + instance: "Instance", + database: "dbname", + callBrowser: false, + dataHost: "THEHOST", + dataInstance: "Instance", + urlHas: []string{"@THEHOST:51433/Instance"}, + urlNotHas: []string{"Instance:51433"}, + }, + }, + { + name: "url with port and instance keeps both", + data: map[string]any{ + "url": "sqlserver://myuser:mypass@host.ip:51433/my_instance?database=dbname", + }, + want: want{ + host: "host.ip", + port: 51433, + instance: "my_instance", + database: "dbname", + callBrowser: false, + dataHost: "host.ip", + dataInstance: "my_instance", + urlHas: []string{"@host.ip:51433/my_instance"}, + }, + }, + { + name: "url with instance only does not inject default port", + data: map[string]any{ + "url": "sqlserver://myuser:mypass@host.ip/my_instance?database=dbname", + }, + want: want{ + host: "host.ip", + port: 0, + instance: "my_instance", + database: "dbname", + callBrowser: true, + dataHost: "host.ip", + dataInstance: "my_instance", + urlHas: []string{"@host.ip/my_instance"}, + urlNotHas: []string{":1433"}, + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c, err := NewConnection("TEST", dbio.TypeDbSQLServer, tc.data) + require.NoError(t, err) + + gotURL := c.URL() + for _, s := range tc.want.urlHas { + assert.Contains(t, gotURL, s, "url=%s", gotURL) + } + for _, s := range tc.want.urlNotHas { + assert.NotContains(t, gotURL, s, "url=%s", gotURL) + } + + if tc.want.dataHost != "" { + assert.Equal(t, tc.want.dataHost, c.Data["host"]) + } + if tc.want.dataInstance != "" { + assert.Equal(t, tc.want.dataInstance, c.Data["instance"]) + } + + cfg, err := msdsn.Parse(gotURL) + require.NoError(t, err) + assert.Equal(t, tc.want.host, cfg.Host) + assert.Equal(t, tc.want.port, cfg.Port) + assert.Equal(t, tc.want.instance, cfg.Instance) + assert.Equal(t, tc.want.database, cfg.Database) + assert.Equal(t, tc.want.callBrowser, len(cfg.Instance) > 0 && cfg.Port == 0) + + assert.False(t, strings.Contains(cfg.Instance, ":"), "instance must not include a port: %q", cfg.Instance) + }) + } +} diff --git a/core/dbio/database/database_sqlserver.go b/core/dbio/database/database_sqlserver.go index ee60a6346..9d1abf546 100755 --- a/core/dbio/database/database_sqlserver.go +++ b/core/dbio/database/database_sqlserver.go @@ -375,6 +375,14 @@ func (conn *MsSQLServerConn) connectAccessToken(timeOut ...int) error { return nil } +func isSQLServerBrowserError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "sql server browser") || strings.Contains(msg, "no instance matching") +} + func (conn *MsSQLServerConn) Connect(timeOut ...int) (err error) { // Check if this is a Cloud SQL connection @@ -387,6 +395,9 @@ func (conn *MsSQLServerConn) Connect(timeOut ...int) (err error) { err = conn.BaseConn.Connect(timeOut...) } if err != nil { + if isSQLServerBrowserError(err) { + return g.Error(err, "SQL Server Browser (UDP 1434) did not return the named instance. Set `port` to the instance TCP port and omit `instance`, or allow UDP 1434. See https://docs.slingdata.io/connections/database-connections/sqlserver") + } return err } diff --git a/core/sling/task.go b/core/sling/task.go index 58f461c9d..70a4946ea 100644 --- a/core/sling/task.go +++ b/core/sling/task.go @@ -645,6 +645,8 @@ func ErrorHelper(err error, connTypes ...dbio.Type) (helpString string) { switch { case contains("utf8") || contains("ascii"): helpString = "Perhaps the 'encodings' source option could help? See https://docs.slingdata.io/concepts/replication/source-options#supported-encodings. Also try the `replace_non_printable` transform. See https://docs.slingdata.io/concepts/replication/transforms" + case contains("sql server browser") || contains("no instance matching"): + helpString = "A named instance needs SQL Server Browser (UDP 1434), or set `port` to the instance TCP port and omit `instance`. See https://docs.slingdata.io/connections/database-connections/sqlserver" case contains("failed to verify certificate"): helpString = "Perhaps specifying `encrypt=true` and `TrustServerCertificate=true` properties could help? See https://docs.slingdata.io/connections/database-connections/sqlserver" case contains("ssl is not enabled on the server"): diff --git a/core/sling/task_test.go b/core/sling/task_test.go index 18ef21e2c..064efb0b5 100644 --- a/core/sling/task_test.go +++ b/core/sling/task_test.go @@ -30,4 +30,11 @@ func TestErrorHelper(t *testing.T) { helpString := ErrorHelper(csvErr, dbio.TypeDbSQLServer, dbio.TypeFileS3) assert.NotContains(t, helpString, "arrow_http") }) + + t.Run("sql browser timeout explains named instance port", func(t *testing.T) { + err := g.Error("unable to get instances from Sql Server Browser on host THEHOST: read udp 127.0.0.1:50533->192.168.0.1:1434: i/o timeout") + helpString := ErrorHelper(err, dbio.TypeDbSQLServer) + assert.Contains(t, helpString, "UDP 1434") + assert.Contains(t, helpString, "instance TCP port") + }) } From 4548379d0c0d33cd3eb5571b3f053585224c2364 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Fri, 14 Aug 2026 08:30:33 -0300 Subject: [PATCH 14/16] fix(tests): comment out azure_sql test in SQL Server suite --- cmd/sling/sling_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/sling/sling_test.go b/cmd/sling/sling_test.go index 75792393d..560dcca27 100755 --- a/cmd/sling/sling_test.go +++ b/cmd/sling/sling_test.go @@ -1222,7 +1222,7 @@ func TestSuiteDatabaseSQLServer(t *testing.T) { } testSuite(t, dbio.Type("sqlserver_adbc")) testSuite(t, dbio.Type("sqlserver_odbc")) - testSuite(t, dbio.Type("azure_sql")) + // testSuite(t, dbio.Type("azure_sql")) } func TestSuiteDatabaseFabric(t *testing.T) { From 401f23962031d14a78a715c6f63b17f8f810bf4a Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Fri, 14 Aug 2026 09:52:10 -0300 Subject: [PATCH 15/16] fix(filesys): bound reader prefetch and dedupe file listings MergeReaders opened each body lazily at first read, serializing remote opens behind a single-file consumer. Add a prefetch worker pool that opens bodies ahead of the consumer while keeping concurrently open bodies bounded by the worker count plus channel buffer, rather than growing with the file count (issue #789). Bodies that a remote store resets while waiting are reopened at read time. Also skip duplicate paths in listings, since stores like Google Drive can list the same file more than once. Update tests to assert the new bounded prefetch window and cover bodies reset mid-stream. --- core/dbio/filesys/fs.go | 49 +++++++- core/dbio/filesys/fs_merge_readers_test.go | 127 +++++++++++++-------- core/dbio/iop/datastream.go | 122 +++++++++++++++++--- tests/files/slow_server.py | 39 +++++++ tests/suite.cli.yaml | 18 +++ 5 files changed, 286 insertions(+), 69 deletions(-) create mode 100644 tests/files/slow_server.py diff --git a/core/dbio/filesys/fs.go b/core/dbio/filesys/fs.go index 068a33b51..ee770c59a 100755 --- a/core/dbio/filesys/fs.go +++ b/core/dbio/filesys/fs.go @@ -12,6 +12,7 @@ import ( "runtime" "runtime/debug" "strings" + "sync" "sync/atomic" "time" @@ -1719,20 +1720,28 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, cfg } // CSV/JSON are read one file at a time. XML is copied one file at a time. - // Open the body at first read so unread remote bodies do not sit idle. channelConsume := g.In(fileType, dbio.FileTypeCsv, dbio.FileTypeJson, dbio.FileTypeJsonLines, dbio.FileTypeGeojson) g.Debug("merging %s readers of %d files [concurrency=%d] from %s", fileType, len(nodes), concurrency, url) + nodeChn := make(chan *iop.ReaderReady, concurrency) readerChn := make(chan *iop.ReaderReady, concurrency) go func() { - defer close(readerChn) + defer close(nodeChn) + seen := map[string]struct{}{} for _, node := range nodes { if strings.HasSuffix(node.URI, "/") { g.Debug("skipping %s because is not file", node.URI) continue } + // stores like Google Drive can list the same path more than once + if _, ok := seen[node.URI]; ok { + g.Debug("skipping duplicate listing of %s", node.URI) + continue + } + seen[node.URI] = struct{}{} + if !includeAll { _, uriExclude := excludeMap[node.URI] _, pathExclude := excludeMap[node.Path()] @@ -1751,21 +1760,49 @@ func MergeReaders(fs FileSysClient, fileType dbio.FileType, nodes FileNodes, cfg g.Debug("processing reader from %s", node.URI) reader, err := fs.Self().GetReader(node.URI) if err != nil { - err = g.Error(err, "Error getting reader") - setError(err) - return nil, err + return nil, g.Error(err, "Error getting reader for %s", node.URI) } return reader, nil } select { - case readerChn <- r: + case nodeChn <- r: case <-ds.Context.Ctx.Done(): return } } }() + // Prefetch pool: open bodies ahead of the consumer, bounded by the worker + // count plus the channel buffer. A body that a remote store resets while + // it waits is reopened at read time (see iop.ReaderReady.Read). + go func() { + defer close(readerChn) + + wg := sync.WaitGroup{} + for i := 0; i < concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for r := range nodeChn { + select { + case <-ds.Context.Ctx.Done(): + // stream ended; pass unopened for the drain to close + default: + r.Prefetch() + } + select { + case readerChn <- r: + case <-ds.Context.Ctx.Done(): + r.Close() + return + } + } + }() + } + wg.Wait() + }() + if channelConsume { pipeW.Close() diff --git a/core/dbio/filesys/fs_merge_readers_test.go b/core/dbio/filesys/fs_merge_readers_test.go index c9e3ede54..19956f148 100644 --- a/core/dbio/filesys/fs_merge_readers_test.go +++ b/core/dbio/filesys/fs_merge_readers_test.go @@ -27,6 +27,7 @@ type idleTrackingReader struct { dead bool closed bool forceClosed bool + dieAt int // >0: die like a reset connection after this many bytes fs *idleSensitiveFS } @@ -79,6 +80,10 @@ func (r *idleTrackingReader) Read(p []byte) (int, error) { if r.dead { return 0, fmt.Errorf("read tcp 10.0.0.1:54567->52.216.0.1:443: read: connection reset by peer") } + if r.dieAt > 0 && r.pos >= r.dieAt { + r.close() + return 0, fmt.Errorf("read tcp 10.0.0.1:54567->52.216.0.1:443: read: connection reset by peer") + } if r.pos >= len(r.data) { r.close() return 0, io.EOF @@ -91,9 +96,12 @@ func (r *idleTrackingReader) Read(p []byte) (int, error) { // idleSensitiveFS hands back bodies that go stale if not consumed promptly. type idleSensitiveFS struct { LocalFileSysClient - contents map[string][]byte - idleWindow time.Duration - readPause time.Duration + contents map[string][]byte + idleWindow time.Duration + readPause time.Duration + openPause time.Duration // latency per GetReader, like a remote store + dieAfterBytes int // >0: the first body per URI dies mid-stream + killed map[string]bool // URIs whose first body already died mu sync.Mutex openCount int @@ -118,12 +126,24 @@ func (fs *idleSensitiveFS) GetReader(uri string) (io.Reader, error) { return nil, fmt.Errorf("no such object: %s", uri) } + if fs.openPause > 0 { + time.Sleep(fs.openPause) + } + fs.mu.Lock() fs.openCount++ fs.curOpen++ if fs.curOpen > fs.maxOpen { fs.maxOpen = fs.curOpen } + dieAt := 0 + if fs.dieAfterBytes > 0 && !fs.killed[uri] { + if fs.killed == nil { + fs.killed = map[string]bool{} + } + fs.killed[uri] = true + dieAt = fs.dieAfterBytes + } fs.mu.Unlock() return &idleTrackingReader{ @@ -131,6 +151,7 @@ func (fs *idleSensitiveFS) GetReader(uri string) (io.Reader, error) { uri: uri, openedAt: time.Now(), idleWindow: fs.idleWindow, + dieAt: dieAt, fs: fs, }, nil } @@ -206,9 +227,14 @@ func (fs *idleSensitiveFS) snapshot() (maxOpen int, maxIdle time.Duration, maxId return fs.maxOpen, fs.maxIdle, fs.maxIdleURI, fs.curOpen } -// TestMergeReadersIdleBoundedIssue789 guards issue #789. MergeReaders used to -// open every file's body up front while the consumer read one at a time. -func TestMergeReadersIdleBoundedIssue789(t *testing.T) { +// mergeConcurrency mirrors the concurrency pick in MergeReaders for the +// mock FS (S3 type, small node counts). +const mergeConcurrency = 10 + +// TestMergeReadersOpenBoundIssue789 guards issue #789's resource side. +// Look-ahead is allowed, but the open-body count must stay bounded by the +// prefetch window, not grow with the file count. +func TestMergeReadersOpenBoundIssue789(t *testing.T) { const numFiles = 40 const rowsPerFile = 200 @@ -221,37 +247,48 @@ func TestMergeReadersIdleBoundedIssue789(t *testing.T) { maxOpen, maxIdle, maxIdleURI, curOpen := fs.snapshot() t.Logf("max open at once: %d, longest idle: %v (%s)", maxOpen, maxIdle.Round(time.Millisecond), maxIdleURI) - // current file plus at most one transition. 20 idle bodies is the old bug. - assert.LessOrEqual(t, maxOpen, 2, "too many bodies open at once: %d", maxOpen) - assert.Less(t, maxIdle, time.Second, - "a body idled %v before its first read; open the body at first read", - maxIdle.Round(time.Millisecond)) + // workers + channel buffer + consumer transition; 40 open bodies is the old bug + assert.LessOrEqual(t, maxOpen, 2*mergeConcurrency+4, "too many bodies open at once: %d", maxOpen) assert.Equal(t, 0, curOpen, "bodies must be closed after use") } -// TestMergeReadersIdleDoesNotGrowWithFileCount is the core of the #789 fix: -// a larger unload must not mean a longer idle wait. -func TestMergeReadersIdleDoesNotGrowWithFileCount(t *testing.T) { - measure := func(numFiles int) time.Duration { - fs, nodes := newIdleSensitiveFS(t, numFiles, 200, time.Hour, idleCSV) +// TestMergeReadersOpensArePipelined guards the staging timeout regression +// (exec 3HuEiSRkALXQERuORc7KDWC5XAn): with per-open latency, opens must +// overlap. A serial open of 30 files at 50ms each takes 1.5s+. +func TestMergeReadersOpensArePipelined(t *testing.T) { + const numFiles = 30 + const rowsPerFile = 50 + const openPause = 50 * time.Millisecond - count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) - assert.NoError(t, err) - assert.Equal(t, numFiles*200, count) + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Hour, idleCSV) + fs.openPause = openPause - _, maxIdle, _, _ := fs.snapshot() - return maxIdle - } + started := time.Now() + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + elapsed := time.Since(started) - small := measure(10) - large := measure(40) + assert.NoError(t, err) + assert.Equal(t, numFiles*rowsPerFile, count) + + serial := time.Duration(numFiles) * openPause + t.Logf("elapsed: %v (serial would be %v+)", elapsed.Round(time.Millisecond), serial) + assert.Less(t, elapsed, serial/2, "opens are not pipelined: %v", elapsed.Round(time.Millisecond)) +} + +// TestMergeReadersResumesAfterMidStreamReset covers the reopen path with a +// non-zero offset: a body that dies mid-file is reopened and skipped to the +// bytes already delivered, with no loss and no duplicates. +func TestMergeReadersResumesAfterMidStreamReset(t *testing.T) { + const numFiles = 3 + const rowsPerFile = 500 - t.Logf("longest idle: 10 files=%v, 40 files=%v", - small.Round(time.Millisecond), large.Round(time.Millisecond)) + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Hour, idleCSV) + fs.dieAfterBytes = 1000 // dies mid-file; each file is ~8KB - // both must stay far below a reset window; do not ratio two near-zero times - assert.Less(t, small, 200*time.Millisecond, "10-file idle %v is not bounded", small) - assert.Less(t, large, 200*time.Millisecond, "40-file idle %v is not bounded", large) + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + + assert.NoError(t, err, "stream must resume after a mid-stream reset") + assert.Equal(t, numFiles*rowsPerFile, count, "rows must not be lost or duplicated") } // TestMergeReadersSurvivesIdleReset is the end-to-end guard: with a window @@ -269,10 +306,10 @@ func TestMergeReadersSurvivesIdleReset(t *testing.T) { assert.Equal(t, numFiles*rowsPerFile, count, "all rows should be read") } -// TestMergeReadersSlowConsumerDoesNotResetLookahead fails if the next body -// is opened while the current file is still being read. Prefetch of live -// bodies sits idle for the pause (2s) and trips the 1s window. -func TestMergeReadersSlowConsumerDoesNotResetLookahead(t *testing.T) { +// TestMergeReadersSlowConsumerRecoversResetLookahead: a slow consumer (2s +// per file) lets prefetched bodies idle past the 1s reset window. The +// stream must recover each reset body by a reopen at first read. +func TestMergeReadersSlowConsumerRecoversResetLookahead(t *testing.T) { const numFiles = 4 const rowsPerFile = 200 const pause = 2 * time.Second @@ -283,18 +320,11 @@ func TestMergeReadersSlowConsumerDoesNotResetLookahead(t *testing.T) { count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) - assert.NoError(t, err, "look-ahead opened a body that sat idle for %v", pause) + assert.NoError(t, err, "a look-ahead body that sat idle for %v must be reopened", pause) assert.Equal(t, numFiles*rowsPerFile, count) - - maxOpen, maxIdle, maxIdleURI, _ := fs.snapshot() - t.Logf("max open at once: %d, longest idle: %v (%s)", maxOpen, maxIdle.Round(time.Millisecond), maxIdleURI) - - assert.LessOrEqual(t, maxOpen, 2, "look-ahead opened extra bodies: maxOpen=%d", maxOpen) - assert.Less(t, maxIdle, idleWindow, - "a body idled %v; open the body at first read", maxIdle.Round(time.Millisecond)) } -func TestMergeReadersJsonLinesIdleBounded(t *testing.T) { +func TestMergeReadersJsonLinesResetRecovered(t *testing.T) { const numFiles = 10 const rowsPerFile = 50 @@ -304,15 +334,14 @@ func TestMergeReadersJsonLinesIdleBounded(t *testing.T) { assert.NoError(t, err) assert.Equal(t, numFiles*rowsPerFile, count) - maxOpen, maxIdle, _, curOpen := fs.snapshot() - assert.LessOrEqual(t, maxOpen, 2, "too many JSONL bodies open at once: %d", maxOpen) - assert.Less(t, maxIdle, time.Second, "JSONL body idled %v", maxIdle.Round(time.Millisecond)) + maxOpen, _, _, curOpen := fs.snapshot() + assert.LessOrEqual(t, maxOpen, 2*mergeConcurrency+4, "too many JSONL bodies open at once: %d", maxOpen) assert.Equal(t, 0, curOpen, "JSONL bodies must be closed after use") } -// TestMergeReadersXmlOpensNearPointOfUse covers the pipe path. Producers used -// to call GetReader for every file, then copy one by one. -func TestMergeReadersXmlOpensNearPointOfUse(t *testing.T) { +// TestMergeReadersXmlOpenBound covers the pipe path: open bodies stay +// bounded by the prefetch window. +func TestMergeReadersXmlOpenBound(t *testing.T) { const numFiles = 20 const rowsPerFile = 10 @@ -328,5 +357,5 @@ func TestMergeReadersXmlOpensNearPointOfUse(t *testing.T) { return fs.openCount }()) - assert.LessOrEqual(t, maxOpen, 2, "XML pipe path opened %d bodies at once", maxOpen) + assert.LessOrEqual(t, maxOpen, 2*mergeConcurrency+4, "XML pipe path opened %d bodies at once", maxOpen) } diff --git a/core/dbio/iop/datastream.go b/core/dbio/iop/datastream.go index d317fe6be..83ddf6d62 100644 --- a/core/dbio/iop/datastream.go +++ b/core/dbio/iop/datastream.go @@ -1194,16 +1194,23 @@ type ReaderReady struct { Reader io.Reader URI string - // Open obtains the reader at first use. Remote stores reset an unread body. + // Open obtains the body. It can run again to replace a body that a + // remote store reset while the body sat idle (e.g. in a prefetch queue). Open func() (io.Reader, error) - mu sync.Mutex - opened bool - closed bool - err error + mu sync.Mutex + opened bool + closed bool + delivered int64 // raw bytes handed to the consumer, for resume on reopen + reopens int } -// GetReader returns the reader. It opens on first use. A failed Open is retried. +const readerReadyMaxReopens = 2 + +// GetReader returns the reader. It opens on first use. A failed Open is +// retried on the next call. When Open is set, the returned reader recovers +// from a reset body: it reopens, skips the bytes already delivered, and +// resumes (see Read). func (rr *ReaderReady) GetReader() (io.Reader, error) { if rr == nil { return nil, g.Error("nil reader") @@ -1218,21 +1225,108 @@ func (rr *ReaderReady) GetReader() (io.Reader, error) { if rr.closed { return nil, g.Error("reader is closed") } - if rr.opened { - return rr.Reader, rr.err + + if !rr.opened { + reader, err := rr.Open() + if err != nil { + // do not cache; the consumer retries at the point of use + return nil, err + } + rr.Reader = reader + rr.opened = true + } + + return rr, nil +} + +// Prefetch opens the body ahead of consumption. An error is not returned: +// GetReader retries the open at the point of use. +func (rr *ReaderReady) Prefetch() { + if rr == nil || rr.Open == nil { + return + } + + rr.mu.Lock() + defer rr.mu.Unlock() + + if rr.closed || rr.opened { + return + } + + if reader, err := rr.Open(); err == nil { + rr.Reader = reader + rr.opened = true + } +} + +// Read delegates to the body. On a read error before any byte of this call, +// it reopens the body, skips the bytes already delivered, and retries. This +// recovers a body that a remote store reset while it sat idle. +func (rr *ReaderReady) Read(p []byte) (n int, err error) { + rr.mu.Lock() + defer rr.mu.Unlock() + + if rr.closed { + return 0, g.Error("reader is closed") + } + + if !rr.opened { + reader, oerr := rr.Open() + if oerr != nil { + return 0, oerr + } + rr.Reader = reader + rr.opened = true + } + + for { + n, err = rr.Reader.Read(p) + if n > 0 { + rr.delivered += int64(n) + } + if err == nil || err == io.EOF { + return n, err + } + if n > 0 { + // deliver the bytes; the error resurfaces on the next call + return n, nil + } + if rerr := rr.reopenLocked(err); rerr != nil { + return 0, rerr + } } +} + +// reopenLocked replaces a dead body and skips to the resume offset. +// The caller must hold rr.mu. +func (rr *ReaderReady) reopenLocked(cause error) error { + if rr.Open == nil || rr.reopens >= readerReadyMaxReopens { + return cause + } + rr.reopens++ + + if c, ok := rr.Reader.(io.Closer); ok && rr.Reader != nil { + c.Close() + } + + g.Debug("reopening reader (attempt %d) from %s after read error: %s", rr.reopens+1, rr.URI, cause.Error()) reader, err := rr.Open() if err != nil { - // do not cache; the consumer retries at the point of use - rr.err = err - return nil, err + return g.Error(err, "could not reopen reader for %s (after read error: %s)", rr.URI, cause.Error()) + } + + if rr.delivered > 0 { + if _, err = io.CopyN(io.Discard, reader, rr.delivered); err != nil { + if c, ok := reader.(io.Closer); ok { + c.Close() + } + return g.Error(err, "could not skip to offset %d on reopened reader for %s", rr.delivered, rr.URI) + } } rr.Reader = reader - rr.err = nil - rr.opened = true - return rr.Reader, nil + return nil } // Close closes an unused or finished body. Safe to call more than once. diff --git a/tests/files/slow_server.py b/tests/files/slow_server.py new file mode 100644 index 000000000..40908b8c7 --- /dev/null +++ b/tests/files/slow_server.py @@ -0,0 +1,39 @@ +"""HTTP server that adds latency to each file download. + +Emulates a high-latency file store (Google Drive) for MergeReaders tests. +Usage: python3 slow_server.py [] +""" +import http.server +import socketserver +import sys +import threading +import time + +PORT = int(sys.argv[1]) +DIR = sys.argv[2] +DELAY = float(sys.argv[3]) +LIFE = float(sys.argv[4]) if len(sys.argv) > 4 else 300 + + +class Handler(http.server.SimpleHTTPRequestHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, directory=DIR, **kwargs) + + def do_GET(self): + # delay file downloads only, not the index page + if self.path.endswith(".csv"): + time.sleep(DELAY) + super().do_GET() + + def log_message(self, format, *args): + pass + + +class Server(socketserver.ThreadingTCPServer): + allow_reuse_address = True + + +srv = Server(("127.0.0.1", PORT), Handler) +threading.Timer(LIFE, srv.shutdown).start() +print("ready", flush=True) +srv.serve_forever() diff --git a/tests/suite.cli.yaml b/tests/suite.cli.yaml index 285beea08..d5de5b54b 100644 --- a/tests/suite.cli.yaml +++ b/tests/suite.cli.yaml @@ -2658,3 +2658,21 @@ run: 'sling run -d -p tests/pipelines/p.49.merge_readers_many_files.yaml' output_contains: - 'SUCCESS: all 8000 rows read intact from many merged files' + +# MergeReaders must pipeline remote reader opens. A serial open of 40 files at 0.4s latency each +# takes 16s+ and breaks the 12s SLING_TIMEOUT. A concurrent open takes ~4s. +- id: 320 + name: 'many high-latency remote files open concurrently' + env: + SLING_TIMEOUT: '0.2' + run: | + mkdir -p temp/merge_latency/src + rm -f temp/merge_latency/src/*.csv temp/merge_latency/out.csv + for i in $(seq -w 1 40); do printf 'id,name\n%s,row_%s\n' "$i" "$i" > temp/merge_latency/src/file_$i.csv; done + python3 tests/files/slow_server.py 18937 temp/merge_latency/src 0.4 120 & + SRV_PID=$! + trap 'kill $SRV_PID 2>/dev/null || true' EXIT + sleep 1 + sling run -d --src-stream 'http://127.0.0.1:18937/' --tgt-object 'file://temp/merge_latency/out.csv' + output_contains: + - 'execution succeeded' From 4f3cbe6eba32b7a6d4ae080cd1bf37765ec4f011 Mon Sep 17 00:00:00 2001 From: Fritz Larco Date: Sun, 16 Aug 2026 08:13:42 -0300 Subject: [PATCH 16/16] feat(iceberg): implement icebergArrowSchema for timestamp fields with UTC zone --- core/dbio/database/database_iceberg.go | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/core/dbio/database/database_iceberg.go b/core/dbio/database/database_iceberg.go index c094b8c2b..87705a3a6 100644 --- a/core/dbio/database/database_iceberg.go +++ b/core/dbio/database/database_iceberg.go @@ -1282,7 +1282,7 @@ func (conn *IcebergConn) BulkImportStream(tableFName string, ds *iop.Datastream) // Process batches from the datastream for batch := range ds.BatchChan { // Create Arrow schema from batch columns (in case they changed) - arrowSchema := iop.ColumnsToArrowSchema(batch.Columns) + arrowSchema := conn.icebergArrowSchema(batch.Columns) // Create memory allocator alloc := memory.NewGoAllocator() @@ -1597,6 +1597,30 @@ func (conn *IcebergConn) generateIcebergSchema(columns iop.Columns) (*iceberg.Sc return schema, nil } +// icebergArrowSchema builds the arrow schema used to append into an iceberg +// table. All timestamp fields get a zone because iopTypeToIcebergPrimitiveType +// declares every timestamp column as iceberg `timestamptz`. A zone-less arrow +// timestamp reads back as iceberg `timestamp`, which iceberg refuses to promote. +func (conn *IcebergConn) icebergArrowSchema(columns iop.Columns) *arrow.Schema { + schema := iop.ColumnsToArrowSchema(columns) + + fields := schema.Fields() + changed := false + for i, field := range fields { + tsType, ok := field.Type.(*arrow.TimestampType) + if !ok || tsType.TimeZone != "" { + continue + } + fields[i].Type = &arrow.TimestampType{Unit: tsType.Unit, TimeZone: "UTC"} + changed = true + } + + if !changed { + return schema + } + return arrow.NewSchema(fields, nil) +} + // iopTypeToIcebergPrimitiveType converts iop column type to Iceberg primitive type func (conn *IcebergConn) iopTypeToIcebergPrimitiveType(col iop.Column) iceberg.Type { switch col.Type {