gcp/changestreams: refactor dialect to use databasepb.DatabaseDialect and add test

Fixes CON-65

Author: mmatczuk

internal/impl/gcp/changestreams/dialect.go

@@ -31,45 +31,33 @@ package changestreams
 
 import (
 	"context"
-	"fmt"
 
 	"cloud.google.com/go/spanner"
+	"cloud.google.com/go/spanner/admin/database/apiv1/databasepb"
 )
 
-type dialect int
+type dialect = databasepb.DatabaseDialect
 
-const (
-	dialectUnknown dialect = iota
-	dialectGoogleSQL
-	dialectPostgreSQL
+var (
+	dialectGoogleSQL  = databasepb.DatabaseDialect_GOOGLE_STANDARD_SQL
+	dialectPostgreSQL = databasepb.DatabaseDialect_POSTGRESQL
 )
 
-func (d dialect) String() string {
-	switch d {
-	case dialectGoogleSQL:
-		return "GoogleSQL"
-	case dialectPostgreSQL:
-		return "PostgreSQL"
-	default:
-		return ""
-	}
-}
-
 func detectDialect(ctx context.Context, client *spanner.Client) (dialect, error) {
-	var value string
-	stmt := spanner.NewStatement("SELECT option_value FROM information_schema.database_options WHERE option_name = 'database_dialect'")
-	if err := client.Single().Query(ctx, stmt).Do(func(r *spanner.Row) error {
-		return r.ColumnByName("option_value", &value)
+	const stmt = "SELECT option_value FROM information_schema.database_options WHERE option_name = 'database_dialect'"
+	var v string
+	if err := client.Single().Query(ctx, spanner.NewStatement(stmt)).Do(func(r *spanner.Row) error {
+		return r.ColumnByName("option_value", &v)
 	}); err != nil {
-		return dialectUnknown, err
+		return databasepb.DatabaseDialect_DATABASE_DIALECT_UNSPECIFIED, err
 	}
 
-	switch value {
+	switch v {
 	case "GOOGLE_STANDARD_SQL", "":
 		return dialectGoogleSQL, nil
 	case "POSTGRESQL":
 		return dialectPostgreSQL, nil
 	default:
-		return dialectUnknown, fmt.Errorf("invalid dialect: %q", value)
+		return databasepb.DatabaseDialect_DATABASE_DIALECT_UNSPECIFIED, nil
 	}
 }

internal/impl/gcp/changestreams/dialect_test.go

@@ -0,0 +1,63 @@
+// Copyright 2025 Redpanda Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package changestreams
+
+import (
+	"fmt"
+	"testing"
+
+	adminpb "cloud.google.com/go/spanner/admin/database/apiv1/databasepb"
+)
+
+func TestIntegrationDetectDialect(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping test in short mode")
+	}
+
+	adm := newEmulatorHelper(t)
+
+	testCases := []struct {
+		dialect dialect
+		fn      func(*adminpb.CreateDatabaseRequest)
+	}{
+		{
+			dialect: dialectGoogleSQL,
+		},
+		{
+			dialect: dialectPostgreSQL,
+			fn: func(req *adminpb.CreateDatabaseRequest) {
+				req.DatabaseDialect = dialectPostgreSQL
+			},
+		},
+	}
+
+	for i, tc := range testCases {
+		t.Run(tc.dialect.String(), func(t *testing.T) {
+			dbName := fmt.Sprintf("dialect%d", i)
+
+			var opts []func(*adminpb.CreateDatabaseRequest)
+			if tc.fn != nil {
+				opts = append(opts, tc.fn)
+			}
+			dd, err := detectDialect(t.Context(), adm.CreateTestDatabase(dbName, opts...))
+			if err != nil {
+				t.Fatalf("failed to detect dialect: %v", err)
+			}
+			if dd != tc.dialect {
+				t.Fatalf("expected dialect %s, got %s", tc.dialect, dd)
+			}
+		})
+	}
+}