fix(kernel): TIMESTAMP range + DDL affected-rows parity with Thrift

internal/arrowscan/arrowscan.go

@@ -173,7 +173,15 @@ func ScanCellCached(col arrow.Array, row int, loc *time.Location, keys *StructKe
 		// it. The cross-backend parity test pins both sides to µs (matching the wire
 		// reality), so it can't exercise a non-µs value; TestScanCellTimestampUnits
 		// covers this arm's unit-correctness directly instead.
-		return inLocation(c.Value(row).ToTime(dt.Unit), loc), nil
+		//
+		// Convert via the unit-specific time.Unix* constructors rather than arrow's
+		// ToTime: ToTime forms an int64-nanosecond intermediate (value*multiplier),
+		// which overflows for TIMESTAMPs outside ~1678–2262 and silently wraps a valid
+		// instant to a wrong one (e.g. TIMESTAMP '0001-01-01' → 1754). time.UnixMicro/
+		// UnixMilli/Unix split into (sec, nsec) internally, so the full 0001–9999 range
+		// round-trips. The Thrift default path never hits this — it receives TIMESTAMP
+		// as a preformatted server string — so this divergence is kernel-only.
+		return inLocation(timestampToTime(int64(c.Value(row)), dt.Unit), loc), nil
 	case *array.Decimal128:
 		dt := col.DataType().(*arrow.Decimal128Type)
 		return decimalfmt.ExactString(c.Value(row), dt.Scale), nil
@@ -276,6 +284,24 @@ func abs64(x int64) int64 {
 	return x
 }
 
+// timestampToTime converts an arrow timestamp value in the given unit to a UTC
+// time.Time without arrow's ToTime int64-nanosecond overflow (see the *array.Timestamp
+// scan arm). Each constructor takes its argument in native units and splits into
+// (seconds, nanoseconds) internally, so the whole 0001–9999 range is representable;
+// only the nanosecond arm passes ns directly (already the finest unit, cannot overflow).
+func timestampToTime(v int64, unit arrow.TimeUnit) time.Time {
+	switch unit {
+	case arrow.Second:
+		return time.Unix(v, 0).UTC()
+	case arrow.Millisecond:
+		return time.UnixMilli(v).UTC()
+	case arrow.Microsecond:
+		return time.UnixMicro(v).UTC()
+	default: // arrow.Nanosecond
+		return time.Unix(0, v).UTC()
+	}
+}
+
 // inLocation renders t in loc, matching the Thrift path's .In(location); a nil
 // loc leaves the value in UTC (arrow's ToTime default).
 func inLocation(t time.Time, loc *time.Location) time.Time {

internal/arrowscan/arrowscan_test.go

@@ -274,6 +274,40 @@ func TestScanCellTimestampUnits(t *testing.T) {
 	}
 }
 
+// TestScanCellTimestampOutOfNanoRange pins the fix for the microsecond→nanosecond
+// overflow: arrow's ToTime forms value*multiplier as an int64 ns count, which
+// overflows for instants outside ~1678–2262 and silently wraps to a wrong time
+// (TIMESTAMP '0001-01-01' scanned back as 1754, '9999-12-31' as 1816). The
+// unit-split constructors used by ScanCell must round-trip the full 0001–9999
+// range that Databricks TIMESTAMP allows.
+func TestScanCellTimestampOutOfNanoRange(t *testing.T) {
+	pool := memory.NewGoAllocator()
+	cases := []time.Time{
+		time.Date(1, time.January, 1, 0, 0, 0, 0, time.UTC),         // TIMESTAMP '0001-01-01 00:00:00'
+		time.Date(9999, time.December, 31, 23, 59, 59, 0, time.UTC), // TIMESTAMP '9999-12-31 23:59:59'
+		time.Date(1677, time.January, 1, 0, 0, 0, 0, time.UTC),      // just below the int64-ns floor
+		time.Date(2263, time.January, 1, 0, 0, 0, 0, time.UTC),      // just above the int64-ns ceiling
+	}
+	for _, want := range cases {
+		t.Run(want.Format("2006-01-02"), func(t *testing.T) {
+			// Databricks TIMESTAMP is always microseconds on the wire.
+			b := array.NewTimestampBuilder(pool, &arrow.TimestampType{Unit: arrow.Microsecond})
+			defer b.Release()
+			b.Append(arrow.Timestamp(want.UnixMicro()))
+			arr := b.NewArray()
+			defer arr.Release()
+
+			v, err := ScanCell(arr, 0, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got := v.(time.Time); !got.Equal(want) {
+				t.Errorf("got %v, want %v (arrow ToTime would have wrapped this)", got.UTC(), want)
+			}
+		})
+	}
+}
+
 // ScanCell renders nested types (list/struct/map) to a JSON string matching the
 // Thrift path.
 func TestScanCellNested(t *testing.T) {

internal/backend/kernel/affectedrows.go

@@ -0,0 +1,21 @@
+package kernel
+
+// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag.
+// It holds the pure normalization of the C ABI's affected-row count so the rule is
+// unit-testable in the default CGO_ENABLED=0 build (like paramBindArg in bindparams.go),
+// separate from the cgo call site in operation.go.
+
+// normalizeAffectedRows maps the kernel C ABI's modified-row count onto the value
+// database/sql's Result.RowsAffected() reports, matching the Thrift path.
+//
+// The C ABI returns -1 for "not applicable / unknown": DDL, SELECT, or a warehouse
+// that doesn't surface the counter (the kernel's num_modified_rows Option<i64> is
+// None). The Thrift path reports 0 in exactly those cases (TGetOperationStatusResp
+// defaults NumModifiedRows to 0), so the -1 sentinel is folded to 0 for parity.
+// A real DML count (>= 0) passes through unchanged.
+func normalizeAffectedRows(n int64) int64 {
+	if n < 0 {
+		return 0
+	}
+	return n
+}

internal/backend/kernel/affectedrows_test.go

@@ -0,0 +1,24 @@
+package kernel
+
+import "testing"
+
+func TestNormalizeAffectedRows(t *testing.T) {
+	cases := []struct {
+		name string
+		in   int64
+		want int64
+	}{
+		{"ddl or select sentinel (-1) folds to Thrift's 0", -1, 0},
+		{"any negative folds to 0", -42, 0},
+		{"zero real count is preserved", 0, 0},
+		{"positive DML count passes through", 7, 7},
+		{"large DML count passes through", 1_000_000, 1_000_000},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := normalizeAffectedRows(tc.in); got != tc.want {
+				t.Errorf("normalizeAffectedRows(%d) = %d, want %d", tc.in, got, tc.want)
+			}
+		})
+	}
+}

internal/backend/kernel/backend.go

@@ -132,6 +132,20 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		return err
 	}
 
+	// User-Agent so query history attributes the kernel path to this driver.
+	if k.cfg.UserAgent != "" {
+		name := newCStr("User-Agent")
+		val := newCStr(k.cfg.UserAgent)
+		errSet := call(func() C.KernelStatusCode {
+			return C.kernel_session_config_set_custom_header(cfg, name.c, val.c)
+		})
+		name.free()
+		val.free()
+		if errSet != nil {
+			return fmt.Errorf("kernel: set_custom_header[User-Agent]: %w", toConnError(errSet))
+		}
+	}
+
 	// TLS: crypto/tls's InsecureSkipVerify accepts any server cert, so relax both
 	// chain validation and the hostname check — mapping only one would leave the
 	// kernel path stricter than the Thrift path it mirrors (a self-signed cert

internal/backend/kernel/config.go

@@ -17,6 +17,10 @@ type Config struct {
 	WarehouseID string // bare warehouse id; preferred over HTTPPath when set
 	Auth        Auth   // PAT / OAuth M2M / OAuth U2M
 
+	// UserAgent is forwarded as the User-Agent header so the kernel path is
+	// attributed to this driver (not the kernel's built-in UA). Empty leaves it unset.
+	UserAgent string
+
 	// SessionConf carries server-bound session confs verbatim — the same map the
 	// Thrift backend forwards (STATEMENT_TIMEOUT, QUERY_TAGS, TIMEZONE, …).
 	SessionConf map[string]string

internal/backend/kernel/operation.go

@@ -194,7 +194,14 @@ func (k *KernelBackend) execute(ctx context.Context, req backend.ExecRequest) (b
 	// Capture the modified-row count and server query id now, while exec is live —
 	// the operation is closed (nulling exec) before these are read on the
 	// ExecContext path, and the query-id pointer is only valid while exec lives.
-	op.affectedRows = int64(C.kernel_executed_statement_num_modified_rows(exec))
+	//
+	// The C ABI returns -1 for "not applicable / unknown" (DDL, SELECT, or a
+	// warehouse that doesn't surface the counter — the kernel's Option<i64> None).
+	// The Thrift path reports 0 in that case (TGetOperationStatusResp defaults
+	// NumModifiedRows to 0), so normalize the sentinel to 0 to keep RowsAffected()
+	// identical across backends; real DML counts (>= 0) pass through unchanged.
+	// normalizeAffectedRows is the pure (CGO_ENABLED=0-testable) form of this rule.
+	op.affectedRows = normalizeAffectedRows(int64(C.kernel_executed_statement_num_modified_rows(exec)))
 	// A nil execErr means the statement reached a terminal state server-side (it
 	// committed). We deliberately do NOT re-check ctx.Err() here to convert a
 	// completed statement into a cancellation: for a non-idempotent DML that would

kernel_config.go

@@ -11,6 +11,7 @@ import (
 	"github.com/databricks/databricks-sql-go/auth/pat"
 	dbsqlerr "github.com/databricks/databricks-sql-go/errors"
 	"github.com/databricks/databricks-sql-go/internal/backend/kernel"
+	"github.com/databricks/databricks-sql-go/internal/client"
 	"github.com/databricks/databricks-sql-go/internal/config"
 )
 
@@ -116,6 +117,8 @@ func buildKernelConfig(cfg *config.Config, kauth kernel.Auth) kernel.Config {
 		WarehouseID: cfg.WarehouseID,
 		Auth:        kauth,
 		Location:    cfg.Location,
+		// Same UA the Thrift path sends, so query history attributes both alike.
+		UserAgent: client.BuildUserAgent(cfg),
 		// Initial namespace: no kernel config setter, so the kernel backend applies
 		// these post-connect via USE CATALOG / USE SCHEMA.
 		Catalog: cfg.Catalog,

kernel_config_test.go

@@ -329,9 +329,11 @@ var kernelConfigFieldDisposition = map[string]string{
 	"MaxRows":           "inert",
 	"UseLz4Compression": "inert", // kernel negotiates compression internally
 
+	// Rides in the forwarded User-Agent header (set_custom_header).
+	"UserAgentEntry": "forwarded",
+
 	// Not applicable to the kernel path (Thrift/HTTP-transport or telemetry knobs
 	// that don't reach the kernel binding).
-	"UserAgentEntry":           "inert", // TODO: forward once the kernel exposes a UA setter
 	"EnableTelemetry":          "inert",
 	"TelemetryBatchSize":       "inert",
 	"TelemetryFlushInterval":   "inert",