V1.5.25

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("")

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) {

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,15 +965,29 @@ 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}"
 		}
 
 		_, 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:

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)
+		})
+	}
+}

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)
 

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,

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 + `"`
 		})

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")