microsoft_sql_server_cdc: capture schema metadata

Add common schema metadata to messages produced by the
microsoft_sql_server_cdc input, matching the behavior of the
postgres_cdc and mysql_cdc inputs. The metadata key is
"common_schema" (rather than "schema") to avoid colliding with
the existing "schema" key that carries the database schema name.

Also adds explicit type mappings for MONEY/SMALLMONEY (string)
and TIMESTAMP/ROWVERSION (byte array), and json:"-" tags on
internal-only MessageEvent fields.

Author: Jeffail

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 - (Benthos) The `command` processor now emits the `exit_code` metadata field. (@mihaitodor)
 - The `postgres_cdc` input now adds schema metadata to consumed messages, this can be used for automatic schema conversion in processors such as `schema_registry_encode`. (@Jeffail)
 - New `iceberg` output, allows writing Iceberg data to REST catalogs in s3, gcs and adls. (@rockwotj)
+- The `microsoft_sql_server_cdc` input now adds schema metadata to consumed messages, this can be used for automatic schema conversion in processors such as `schema_registry_encode`. (@Jeffail)
 
 ### Changed
 

docs/modules/components/pages/inputs/microsoft_sql_server_cdc.adoc

@@ -98,6 +98,7 @@ Additionally, if `stream_snapshot` is set to true, then the existing data in the
 
 This input adds the following metadata fields to each message:
 - schema (Schema of the table that the message originated from)
+- common_schema (The table schema in benthos common schema format, compatible with processors like parquet_encode)
 - table (Name of the table that the message originated from)
 - operation (Type of operation that generated the message: "read", "delete", "insert", or "update_before" and "update_after". "read" is from messages that are read in the initial snapshot phase.)
 - lsn (the Log Sequence Number in Microsoft SQL Server)

internal/impl/mssqlserver/batcher.go

@@ -10,6 +10,7 @@ package mssqlserver
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"fmt"
 	"sync"
@@ -28,6 +29,14 @@ type batchPublisher struct {
 	batcher   *service.Batcher
 	batcherMu sync.Mutex
 
+	// tableSchemas caches the computed common schema for each table. No
+	// invalidation is needed because MSSQL CDC capture instances are immutable:
+	// an ALTER TABLE requires creating a new capture instance, which the input
+	// won't discover until it restarts (at which point a fresh batchPublisher
+	// with an empty cache is created).
+	tableSchemas   map[string]any
+	tableSchemasMu sync.RWMutex
+
 	checkpoint *checkpoint.Capped[replication.LSN]
 	msgChan    chan asyncMessage
 	log        *service.Logger
@@ -38,11 +47,12 @@ type batchPublisher struct {
 // newBatchPublisher creates an instance of batchPublisher.
 func newBatchPublisher(batcher *service.Batcher, checkpoint *checkpoint.Capped[replication.LSN], logger *service.Logger) *batchPublisher {
 	b := &batchPublisher{
-		batcher:    batcher,
-		checkpoint: checkpoint,
-		log:        logger,
-		msgChan:    make(chan asyncMessage),
-		shutSig:    shutdown.NewSignaller(),
+		batcher:      batcher,
+		checkpoint:   checkpoint,
+		log:          logger,
+		msgChan:      make(chan asyncMessage),
+		shutSig:      shutdown.NewSignaller(),
+		tableSchemas: make(map[string]any),
 	}
 	go b.loop()
 	return b
@@ -125,6 +135,28 @@ func (p *batchPublisher) loop() {
 	}
 }
 
+// getOrComputeTableSchema returns the cached schema for tableName. If not yet
+// cached and colTypes is non-empty, it computes and caches the schema from the
+// provided column metadata.
+func (b *batchPublisher) getOrComputeTableSchema(tableName string, colNames []string, colTypes []*sql.ColumnType) any {
+	b.tableSchemasMu.RLock()
+	if s, ok := b.tableSchemas[tableName]; ok {
+		b.tableSchemasMu.RUnlock()
+		return s
+	}
+	b.tableSchemasMu.RUnlock()
+
+	if len(colTypes) == 0 {
+		return nil
+	}
+
+	s := columnTypesToSchema(tableName, colNames, colTypes)
+	b.tableSchemasMu.Lock()
+	b.tableSchemas[tableName] = s
+	b.tableSchemasMu.Unlock()
+	return s
+}
+
 // Publish turns the provided message into a service.Message before batching and
 // flushing them based on batch size or time elapsed.
 func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent) error {
@@ -140,6 +172,9 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	if len(m.LSN) != 0 {
 		msg.MetaSet("lsn", string(m.LSN))
 	}
+	if s := b.getOrComputeTableSchema(m.Table, m.ColumnNames, m.ColumnTypes); s != nil {
+		msg.MetaSetImmut("common_schema", service.ImmutableAny{V: s})
+	}
 
 	var flushedBatch []*service.Message
 	b.batcherMu.Lock()

internal/impl/mssqlserver/input_mssqlserver_cdc.go

@@ -59,6 +59,7 @@ Additionally, if ` + "`" + fieldStreamSnapshot + "`" + ` is set to true, then th
 
 This input adds the following metadata fields to each message:
 - schema (Schema of the table that the message originated from)
+- common_schema (The table schema in benthos common schema format, compatible with processors like parquet_encode)
 - table (Name of the table that the message originated from)
 - operation (Type of operation that generated the message: "read", "delete", "insert", or "update_before" and "update_after". "read" is from messages that are read in the initial snapshot phase.)
 - lsn (the Log Sequence Number in Microsoft SQL Server)

internal/impl/mssqlserver/integration_test.go

@@ -709,6 +709,129 @@ microsoft_sql_server_cdc:
 	}
 }
 
+func TestIntegration_MicrosoftSQLServerCDC_SchemaMetadata(t *testing.T) {
+	integration.CheckSkip(t)
+	t.Parallel()
+
+	connStr, db := mssqlservertest.SetupTestWithMicrosoftSQLServerVersion(t, "2022-latest")
+	require.NoError(t, db.CreateTableWithCDCEnabledIfNotExists(t.Context(), "dbo.schema_meta_test", `
+		CREATE TABLE dbo.schema_meta_test (
+			id      INT          PRIMARY KEY,
+			label   NVARCHAR(50) NOT NULL,
+			active  BIT          NOT NULL,
+			score   FLOAT        NOT NULL,
+			created DATETIME2    NOT NULL
+		);`))
+
+	// Disable CDC so the first row becomes a snapshot row, then re-enable CDC.
+	db.MustDisableCDC(t.Context(), "dbo.schema_meta_test")
+	db.MustExecContext(t.Context(), `INSERT INTO dbo.schema_meta_test VALUES (1, N'snapshot', 1, 3.14, SYSDATETIME())`)
+	db.MustEnableCDC(t.Context(), "dbo.schema_meta_test")
+
+	type msgMeta struct {
+		schema any
+		op     string
+	}
+	var received []msgMeta
+	var receivedMu sync.Mutex
+
+	cfg := fmt.Sprintf(`
+microsoft_sql_server_cdc:
+  connection_string: %s
+  stream_snapshot: true
+  include: ["schema_meta_test"]`, connStr)
+
+	streamBuilder := service.NewStreamBuilder()
+	require.NoError(t, streamBuilder.AddInputYAML(cfg))
+	require.NoError(t, streamBuilder.AddBatchConsumerFunc(func(_ context.Context, mb service.MessageBatch) error {
+		for _, msg := range mb {
+			s, _ := msg.MetaGetMut("common_schema")
+			op, _ := msg.MetaGet("operation")
+			receivedMu.Lock()
+			received = append(received, msgMeta{schema: s, op: op})
+			receivedMu.Unlock()
+		}
+		return nil
+	}))
+
+	stream, err := streamBuilder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(stream.Resources())
+
+	go func() {
+		require.NoError(t, stream.Run(t.Context()))
+	}()
+
+	// Wait for the snapshot row to arrive.
+	assert.Eventually(t, func() bool {
+		receivedMu.Lock()
+		defer receivedMu.Unlock()
+		for _, m := range received {
+			if m.op == "read" {
+				return true
+			}
+		}
+		return false
+	}, time.Second*30, time.Millisecond*100)
+
+	// Insert a CDC row and wait for it to arrive.
+	db.MustExecContext(t.Context(), `INSERT INTO dbo.schema_meta_test VALUES (2, N'cdc', 0, 2.71, SYSDATETIME())`)
+	assert.Eventually(t, func() bool {
+		receivedMu.Lock()
+		defer receivedMu.Unlock()
+		for _, m := range received {
+			if m.op == "insert" {
+				return true
+			}
+		}
+		return false
+	}, time.Second*30, time.Millisecond*100)
+
+	require.NoError(t, stream.StopWithin(time.Second*10))
+
+	receivedMu.Lock()
+	defer receivedMu.Unlock()
+
+	require.Len(t, received, 2, "expected 1 snapshot message and 1 CDC message")
+
+	// Expected column name → benthos common type string for dbo.schema_meta_test.
+	expectedCols := map[string]string{
+		"id":      "INT64",
+		"label":   "STRING",
+		"active":  "BOOLEAN",
+		"score":   "FLOAT64",
+		"created": "TIMESTAMP",
+	}
+
+	for i, m := range received {
+		require.NotNilf(t, m.schema, "message %d (op=%q) is missing schema metadata", i, m.op)
+
+		schemaMap, ok := m.schema.(map[string]any)
+		require.Truef(t, ok, "message %d schema is not map[string]any, got %T", i, m.schema)
+
+		assert.Equalf(t, "OBJECT", schemaMap["type"], "message %d schema type", i)
+		assert.Equalf(t, "schema_meta_test", schemaMap["name"], "message %d schema name", i)
+
+		children, ok := schemaMap["children"].([]any)
+		require.Truef(t, ok, "message %d schema children is not []any", i)
+		assert.Lenf(t, children, len(expectedCols), "message %d schema children count", i)
+
+		for _, child := range children {
+			childMap, ok := child.(map[string]any)
+			require.Truef(t, ok, "message %d child schema is not map[string]any", i)
+
+			name, _ := childMap["name"].(string)
+			typ, _ := childMap["type"].(string)
+			optional, _ := childMap["optional"].(bool)
+
+			expectedType, exists := expectedCols[name]
+			assert.Truef(t, exists, "message %d: unexpected column %q in schema", i, name)
+			assert.Equalf(t, expectedType, typ, "message %d column %q type mismatch", i, name)
+			assert.Truef(t, optional, "message %d column %q should be optional", i, name)
+		}
+	}
+}
+
 // Test_ManualTesting_AddTestDataWithUniqueLSN adds data to an existing table and ensures each change has its own LSN
 func Test_ManualTesting_AddTestDataWithUniqueLSN(t *testing.T) {
 	t.Skip("This test requires a remote database to run. Aimed to seed initial data in a remote test databases")

internal/impl/mssqlserver/replication/snapshot.go

@@ -163,11 +163,13 @@ func (s *Snapshot) snapshotTable(ctx context.Context, table UserDefinedTable, ma
 				}
 
 				m := MessageEvent{
-					Table:     table.Name,
-					Schema:    table.Schema,
-					Data:      row,
-					Operation: MessageOperationRead.String(),
-					LSN:       nil,
+					Table:       table.Name,
+					Schema:      table.Schema,
+					Data:        row,
+					Operation:   MessageOperationRead.String(),
+					LSN:         nil,
+					ColumnNames: columns,
+					ColumnTypes: types,
 				}
 				if err = s.publisher.Publish(ctx, m); err != nil {
 					return fmt.Errorf("handling snapshot table row: %w", err)

internal/impl/mssqlserver/replication/stream.go

@@ -16,6 +16,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/redpanda-data/benthos/v4/public/service"
@@ -92,6 +93,11 @@ type changeTableRowIter struct {
 	log      *service.Logger
 
 	vals []any
+
+	// userColNames and userColTypes are the user-defined columns only,
+	// excluding MSSQL system columns (those with __$ prefix).
+	userColNames []string
+	userColTypes []*sql.ColumnType
 }
 
 // newChangeTableRowIter returns an custom row iterator for the given changeTable.
@@ -125,6 +131,17 @@ func newChangeTableRowIter(
 		return nil, err
 	}
 
+	// Compute user-defined column lists by filtering out MSSQL system columns
+	// (those with the __$ prefix, e.g. __$start_lsn, __$operation, etc.).
+	userColNames := make([]string, 0, len(cols))
+	userColTypes := make([]*sql.ColumnType, 0, len(cols))
+	for i, c := range cols {
+		if !strings.HasPrefix(c, "__$") {
+			userColNames = append(userColNames, c)
+			userColTypes = append(userColTypes, colTypes[i])
+		}
+	}
+
 	// pre-allocate slice of pointers for sql.Scan operations
 	vals := make([]any, len(cols))
 	for i := range vals {
@@ -133,12 +150,14 @@ func newChangeTableRowIter(
 	}
 
 	iter := &changeTableRowIter{
-		table:    changeTable,
-		rows:     rows,
-		cols:     cols,
-		colTypes: colTypes,
-		vals:     vals,
-		log:      logger,
+		table:        changeTable,
+		rows:         rows,
+		cols:         cols,
+		colTypes:     colTypes,
+		vals:         vals,
+		log:          logger,
+		userColNames: userColNames,
+		userColTypes: userColTypes,
 	}
 	// Prime the iterator by loading the first row
 	if err := iter.next(); err != nil {
@@ -344,11 +363,13 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 			cur := item.iter.current
 
 			msg := MessageEvent{
-				Table:     item.iter.table.Name,
-				Schema:    item.iter.table.Schema,
-				Data:      cur.columns,
-				LSN:       cur.startLSN,
-				Operation: cur.operation.String(),
+				Table:       item.iter.table.Name,
+				Schema:      item.iter.table.Schema,
+				Data:        cur.columns,
+				LSN:         cur.startLSN,
+				Operation:   cur.operation.String(),
+				ColumnNames: item.iter.userColNames,
+				ColumnTypes: item.iter.userColTypes,
 			}
 
 			if err := r.publisher.Publish(ctx, msg); err != nil {

internal/impl/mssqlserver/replication/stream_message.go

@@ -9,6 +9,7 @@
 package replication
 
 import (
+	"database/sql"
 	"encoding/hex"
 	"fmt"
 )
@@ -87,4 +88,10 @@ type MessageEvent struct {
 	Schema    string `json:"schema"`
 	Table     string `json:"table"`
 	Data      any    `json:"data"`
+
+	// ColumnNames and ColumnTypes carry user-defined column metadata (excluding
+	// MSSQL system columns with __$ prefix). They are used to build schema
+	// metadata on the outgoing message and are not serialised to JSON.
+	ColumnNames []string          `json:"-"`
+	ColumnTypes []*sql.ColumnType `json:"-"`
 }