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/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) { 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.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) 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/database/database_duckdb.go b/core/dbio/database/database_duckdb.go index 751b0d00c..b4691f204 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") @@ -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/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/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 { 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/dbio/filesys/fs.go b/core/dbio/filesys/fs.go index 1c0b29810..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" @@ -1411,6 +1412,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 +1438,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, @@ -1708,51 +1719,91 @@ 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. + 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 } - ds.Context.Wg.Read.Add() - go func(node FileNode) { - defer ds.Context.Wg.Read.Done() + // 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()] - _, 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 + return nil, g.Error(err, "Error getting reader for %s", node.URI) } + return reader, nil + } - r := &iop.ReaderReady{Reader: reader, URI: node.URI} - readerChn <- r - }(node) + select { + case nodeChn <- r: + case <-ds.Context.Ctx.Done(): + return + } } + }() - ds.Context.Wg.Read.Wait() + // 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 g.In(fileType, dbio.FileTypeCsv, dbio.FileTypeJson, dbio.FileTypeJsonLines, dbio.FileTypeGeojson) { + if channelConsume { pipeW.Close() switch fileType { @@ -1768,13 +1819,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..19956f148 --- /dev/null +++ b/core/dbio/filesys/fs_merge_readers_test.go @@ -0,0 +1,361 @@ +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 + dieAt int // >0: die like a reset connection after this many bytes + 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.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 + } + 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 + 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 + 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) + } + + 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{ + data: data, + uri: uri, + openedAt: time.Now(), + idleWindow: fs.idleWindow, + dieAt: dieAt, + 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 +} + +// 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 + + 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) + + // 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") +} + +// 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 + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Hour, idleCSV) + fs.openPause = openPause + + started := time.Now() + count, err := readAll(t, fs, nodes, dbio.FileTypeCsv) + elapsed := time.Since(started) + + 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 + + fs, nodes := newIdleSensitiveFS(t, numFiles, rowsPerFile, time.Hour, idleCSV) + fs.dieAfterBytes = 1000 // dies mid-file; each file is ~8KB + + 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 +// 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") +} + +// 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 + 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, "a look-ahead body that sat idle for %v must be reopened", pause) + assert.Equal(t, numFiles*rowsPerFile, count) +} + +func TestMergeReadersJsonLinesResetRecovered(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, _, _, 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") +} + +// TestMergeReadersXmlOpenBound covers the pipe path: open bodies stay +// bounded by the prefetch window. +func TestMergeReadersXmlOpenBound(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*mergeConcurrency+4, "XML pipe path opened %d bodies at once", maxOpen) +} diff --git a/core/dbio/iop/arrow.go b/core/dbio/iop/arrow.go index dd24961ed..47bf41363 100644 --- a/core/dbio/iop/arrow.go +++ b/core/dbio/iop/arrow.go @@ -210,9 +210,26 @@ 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" } + // 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 @@ -575,19 +592,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 @@ -607,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) @@ -921,6 +950,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 +996,40 @@ 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 +// is unchanged for columns that carry no zone. +func arrowSchemaTimeZone(col Column) string { + if tz := col.Metadata["timeZone"]; tz != "" { + return 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 "" + } + 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 +1238,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..b21dc0c73 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,168 @@ 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") +} + +// 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. diff --git a/core/dbio/iop/datastream.go b/core/dbio/iop/datastream.go index 4a4ff8ccc..83ddf6d62 100644 --- a/core/dbio/iop/datastream.go +++ b/core/dbio/iop/datastream.go @@ -1193,17 +1193,192 @@ func (ds *Datastream) ConsumeXmlReader(reader io.Reader) (err error) { type ReaderReady struct { Reader io.Reader URI string + + // 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 + delivered int64 // raw bytes handed to the consumer, for resume on reopen + reopens int +} + +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") + } + 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 { + 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 { + 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 + return 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 +1399,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 +1435,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 +1454,9 @@ func (ds *Datastream) ConsumeJsonReaderChl(readerChn chan *ReaderReady, isXML bo goto processNext } + stopJSON() + } else if !hasNext { + stopJSON() } // set stream url @@ -1294,12 +1497,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 +1538,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 +1562,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 +1594,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 +1649,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 } @@ -2717,6 +2964,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 +3001,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/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/core/dbio/iop/duckdb.go b/core/dbio/iop/duckdb.go index 172f029ff..53d0da5ef 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,26 @@ 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. 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 + } + } + // 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 +812,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 +874,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 +888,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 +1307,8 @@ func (duck *DuckDb) initScanner() { return } + dq.touch() // process is responsive; reset the stall clock + mu.Lock() defer mu.Unlock() @@ -1676,6 +1722,24 @@ 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 +} + +// 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) { @@ -1704,12 +1768,23 @@ 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") } } // 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() { @@ -1723,9 +1798,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 +1850,20 @@ 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) + }() + // 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 @@ -1782,10 +1879,18 @@ 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()) + closeBatchReader(batchR) + return } // Stream data through pipe @@ -1797,8 +1902,26 @@ 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()) + closeBatchReader(batchR) + return + } + + select { + 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++ } @@ -1811,26 +1934,23 @@ 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. - maxLineSize := 2000000 - for _, c := range batchR.Columns { - if c.IsBinary() { - 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) - 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()) + closeBatchReader(batchR) + return } // Stream data through pipe @@ -1842,8 +1962,26 @@ 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()) + closeBatchReader(batchR) + return + } + + select { + 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++ } @@ -1865,6 +2003,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 f71e26741..054746e2b 100644 --- a/core/dbio/iop/duckdb_test.go +++ b/core/dbio/iop/duckdb_test.go @@ -525,6 +525,115 @@ 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) { + // 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 + maxLineSize string + }{ + {"binary column raises limit", BinaryType, "max_line_size=268435456"}, + {"text column raises limit", TextType, "max_line_size=268435456"}, + {"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 { + 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() + }) + } + }) +} + +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 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) + } + } + } +} 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/core/sling/task.go b/core/sling/task.go index 9efe0dce8..70a4946ea 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 { @@ -636,6 +645,8 @@ func ErrorHelper(err error) (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"): @@ -665,8 +676,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 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"): 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..064efb0b5 --- /dev/null +++ b/core/sling/task_test.go @@ -0,0 +1,40 @@ +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.Contains(t, helpString, "max_line_size` property") + 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") + }) + + 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") + }) +} 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/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/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/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 de1ca8e66..d5de5b54b 100644 --- a/tests/suite.cli.yaml +++ b/tests/suite.cli.yaml @@ -2634,3 +2634,45 @@ - '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' + +# 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' + +# 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' + +# 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'