Validate row field counts before serialization

cmd/internal/server/handlers/schema_aware_serializer.go

@@ -76,12 +76,20 @@ func (s *schemaAwareRecordSerializer) Serialize(before *sqltypes.Result, after *
 	)
 
 	if opType == lib.OpType_Update {
-		beforeMap = convertRowToMap(&before.Rows[0], columns)
+		var err error
+		beforeMap, err = convertRowToMap(&before.Rows[0], columns)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	for _, row := range after.Rows {
 		record := make(map[string]*fivetransdk.ValueType)
-		afterMap = convertRowToMap(&row, columns)
+		var err error
+		afterMap, err = convertRowToMap(&row, columns)
+		if err != nil {
+			return nil, err
+		}
 		for colName, val := range afterMap {
 			if selected := s.columnSelection[colName]; !selected {
 				continue
@@ -130,18 +138,18 @@ func (s *schemaAwareRecordSerializer) Serialize(before *sqltypes.Result, after *
 	return data, nil
 }
 
-func convertRowToMap(row *sqltypes.Row, columns []string) map[string]sqltypes.Value {
+func convertRowToMap(row *sqltypes.Row, columns []string) (map[string]sqltypes.Value, error) {
+	if len(*row) != len(columns) {
+		return nil, fmt.Errorf("row value count %d does not match column count %d", len(*row), len(columns))
+	}
+
 	record := map[string]sqltypes.Value{}
 	for idx, val := range *row {
-		if idx > len(columns) {
-			// if there's more values than columns, exit this loop
-			break
-		}
 		colName := columns[idx]
 		record[colName] = val
 	}
 
-	return record
+	return record, nil
 }
 
 func NewSchemaAwareSerializer(sender LogSender, prefix string, serializeTinyIntAsBool bool, schemaList *fivetransdk.SchemaList, enumsAndSets SchemaEnumsAndSets) Serializer {

cmd/internal/server/handlers/schema_aware_serializer_test.go

@@ -76,6 +76,36 @@ func TestCanSerializeInsert(t *testing.T) {
 	assert.Equal(t, "string:\"enum_value\"", data["enum_value"].String())
 }
 
+func TestConvertRowToMapRequiresMatchingColumnCount(t *testing.T) {
+	tests := []struct {
+		name    string
+		row     sqltypes.Row
+		columns []string
+		wantErr string
+	}{
+		{
+			name:    "extra value",
+			row:     sqltypes.Row{sqltypes.NewInt32(1), sqltypes.NewVarChar("extra")},
+			columns: []string{"id"},
+			wantErr: "row value count 2 does not match column count 1",
+		},
+		{
+			name:    "missing value",
+			row:     sqltypes.Row{sqltypes.NewInt32(1)},
+			columns: []string{"id", "name"},
+			wantErr: "row value count 1 does not match column count 2",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := convertRowToMap(&tt.row, tt.columns)
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), tt.wantErr)
+		})
+	}
+}
+
 func TestCanSerializeMappedEnumsAndSets(t *testing.T) {
 	fivetranSchema := fivetransdk.Schema{
 		Name: "Customers",