Support check constraint frontend v2

.github/workflows/integration-tests-against-emulator.yaml

@@ -52,7 +52,7 @@ jobs:
 
     services:
       spanner_emulator:
-        image: gcr.io/cloud-spanner-emulator/emulator:1.5.27
+        image: gcr.io/cloud-spanner-emulator/emulator:1.5.28
         ports:
           - 9010:9010
           - 9020:9020
@@ -114,21 +114,28 @@ jobs:
       - run: mysql -v -P 3306 --protocol=tcp -u root -proot < test_data/mysql_foreignkeyaction_dump.test.out
 
       # init sql server with test_data
+      # since we use ubuntu-latest container, we should ensure that the path matches the latest from  https://packages.microsoft.com/config/ubuntu/
+      # while its possible to infer the latest from the path in the run script, it will make the run section more complex and hard to maintian.
       - name: Install sqlcmd required for loading .sql files
         run: |
           curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
-          curl https://packages.microsoft.com/config/ubuntu/16.04/prod.list | sudo tee /etc/apt/sources.list.d/msprod.list
-          sudo apt-get update 
-          sudo apt-get install mssql-tools unixodbc-dev
-          echo 'export PATH="$PATH:/opt/mssql-tools/bin"' >> ~/.bash_profile
-      - run: sqlcmd -?
-      - run: sqlcmd -U sa -P ${MSSQL_SA_PASSWORD} -i test_data/sqlserver.test.out
+          curl https://packages.microsoft.com/config/ubuntu/24.04/prod.list | sudo tee /etc/apt/sources.list.d/msprod.list
+          sudo apt-get update
+          sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18
+          sudo apt-get install mssql-tools18 unixodbc-dev
+          set -x
+          ls /opt/mssql-tools18/bin/
+          set +x
+          echo 'export PATH="$PATH:/opt/mssql-tools18/bin"' >> ~/.bashrc
+      - run: /opt/mssql-tools18/bin/sqlcmd -C -?
+      - run: /opt/mssql-tools18/bin/sqlcmd -U sa -P ${MSSQL_SA_PASSWORD} -i test_data/sqlserver.test.out -C
 
       # sqlplus set up init oracle db.
       - name: Install sqlplus required for loading .sql files
         run: |
           sudo apt-get update
-          sudo apt-get install -y libaio1 rpm2cpio cpio
+          sudo apt-get install -y libaio1t64 rpm2cpio cpio
+          sudo ln -s /usr/lib/x86_64-linux-gnu/libaio.so.1t64 /usr/lib/libaio.so.1
           curl -O https://download.oracle.com/otn_software/linux/instantclient/2340000/oracle-instantclient-basic-23.4.0.24.05-1.el9.x86_64.rpm
           curl -O https://download.oracle.com/otn_software/linux/instantclient/2340000/oracle-instantclient-sqlplus-23.4.0.24.05-1.el9.x86_64.rpm
           rpm2cpio oracle-instantclient-basic-23.4.0.24.05-1.el9.x86_64.rpm | sudo cpio -idmv

common/constants/constants.go

@@ -74,22 +74,22 @@ const (
 	AddIndex             = "add_index"
 	EditColumnMaxLength  = "edit_column_max_length"
 	AddShardIdPrimaryKey = "add_shard_id_primary_key"
-	//bulk migration type
+	// bulk migration type
 	BULK_MIGRATION = "bulk"
-	//dataflow migration type
+	// dataflow migration type
 	DATAFLOW_MIGRATION = "dataflow"
-	//DMS migration type
+	// DMS migration type
 	DMS_MIGRATION = "dms"
 
 	SESSION_FILE = "sessionFile"
 
-	//Default shardId
+	// Default shardId
 	DEFAULT_SHARD_ID string = "smt-default"
-	//Metadata database name
+	// Metadata database name
 	METADATA_DB string = "spannermigrationtool_metadata"
-	//Migration types
+	// Migration types
 	MINIMAL_DOWNTIME_MIGRATION = "minimal_downtime"
-	//Job Resource Types
+	// Job Resource Types
 	DATAFLOW_RESOURCE         string = "dataflow"
 	PUBSUB_RESOURCE           string = "pubsub"
 	DLQ_PUBSUB_RESOURCE       string = "dlq_pubsub"
@@ -111,7 +111,7 @@ const (
 	// Default gcs path of the Dataflow template.
 	DEFAULT_TEMPLATE_PATH string = "gs://dataflow-templates/latest/flex/Cloud_Datastream_to_Spanner"
 
-	//FK Actions
+	// FK Actions
 	FK_NO_ACTION   string = "NO ACTION"
 	FK_CASCADE     string = "CASCADE"
 	FK_SET_DEFAULT string = "SET DEFAULT"
@@ -122,8 +122,12 @@ const (
 	REGULAR_GCS string = "data"
 	DLQ_GCS     string = "dlq"
 
-	//VerifyExpresions API
-	CHECK_EXPRESSION = "CHECK"
+	// VerifyExpresions API
+	CHECK_EXPRESSION  = "CHECK"
 	DEFAUT_EXPRESSION = "DEFAULT"
-
+	DEFAULT_GENERATED = "DEFAULT_GENERATED"
+	TEMP_DB           = "smt-staging-db"
+
+	// Regex for matching database collation
+	DB_COLLATION_REGEX = `(_[a-zA-Z0-9]+\\|\\)`
 )

common/utils/utils.go

@@ -41,6 +41,7 @@ import (
 	"cloud.google.com/go/storage"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/parse"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/common"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/spanner"
@@ -445,7 +446,8 @@ func GetLegacyModeSupportedDrivers() []string {
 func ReadSpannerSchema(ctx context.Context, conv *internal.Conv, client *sp.Client) error {
 	infoSchema := spanner.InfoSchemaImpl{Client: client, Ctx: ctx, SpDialect: conv.SpDialect}
 	processSchema := common.ProcessSchemaImpl{}
-	err := processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, internal.AdditionalSchemaAttributes{IsSharded: false}, &common.SchemaToSpannerImpl{}, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
+	expressionVerificationAccessor, _ := expressions_api.NewExpressionVerificationAccessorImpl(ctx, conv.SpProjectId, conv.SpInstanceId)
+	err := processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, internal.AdditionalSchemaAttributes{IsSharded: false}, &common.SchemaToSpannerImpl{ExpressionVerificationAccessor: expressionVerificationAccessor}, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
 	if err != nil {
 		return fmt.Errorf("error trying to read and convert spanner schema: %v", err)
 	}

conversion/conversion.go

@@ -37,6 +37,7 @@ import (
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/task"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/internal/reports"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -79,7 +80,8 @@ func (ci *ConvImpl) SchemaConv(migrationProjectId string, sourceProfile profiles
 	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER, constants.ORACLE:
 		return schemaFromSource.schemaFromDatabase(migrationProjectId, sourceProfile, targetProfile, &GetInfoImpl{}, &common.ProcessSchemaImpl{})
 	case constants.PGDUMP, constants.MYSQLDUMP:
-		return schemaFromSource.SchemaFromDump(sourceProfile.Driver, targetProfile.Conn.Sp.Dialect, ioHelper, &ProcessDumpByDialectImpl{})
+		expressionVerificationAccessor, _ := expressions_api.NewExpressionVerificationAccessorImpl(context.Background(), targetProfile.Conn.Sp.Project, targetProfile.Conn.Sp.Instance)
+		return schemaFromSource.SchemaFromDump(sourceProfile.Driver, targetProfile.Conn.Sp.Dialect, ioHelper, &ProcessDumpByDialectImpl{expressionVerificationAccessor})
 	default:
 		return nil, fmt.Errorf("schema conversion for driver %s not supported", sourceProfile.Driver)
 	}

conversion/conversion_from_source.go

@@ -26,6 +26,7 @@ import (
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/metrics"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -54,6 +55,9 @@ type DataFromSourceImpl struct{}
 func (sads *SchemaFromSourceImpl) schemaFromDatabase(migrationProjectId string, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, getInfo GetInfoInterface, processSchema common.ProcessSchemaInterface) (*internal.Conv, error) {
 	conv := internal.MakeConv()
 	conv.SpDialect = targetProfile.Conn.Sp.Dialect
+	conv.SpProjectId = targetProfile.Conn.Sp.Project
+	conv.SpInstanceId = targetProfile.Conn.Sp.Instance
+	conv.Source = sourceProfile.Driver
 	//handle fetching schema differently for sharded migrations, we only connect to the primary shard to
 	//fetch the schema. We reuse the SourceProfileConnection object for this purpose.
 	var infoSchema common.InfoSchema
@@ -96,7 +100,11 @@ func (sads *SchemaFromSourceImpl) schemaFromDatabase(migrationProjectId string,
 	additionalSchemaAttributes := internal.AdditionalSchemaAttributes{
 		IsSharded: isSharded,
 	}
-	return conv, processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, additionalSchemaAttributes, &common.SchemaToSpannerImpl{}, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
+
+	ctx := context.Background()
+	expressionVerificationAccessor, _ := expressions_api.NewExpressionVerificationAccessorImpl(ctx, conv.SpProjectId, conv.SpInstanceId)
+
+	return conv, processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, additionalSchemaAttributes, &common.SchemaToSpannerImpl{expressionVerificationAccessor}, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
 }
 
 func (sads *SchemaFromSourceImpl) SchemaFromDump(driver string, spDialect string, ioHelper *utils.IOStreams, processDump ProcessDumpByDialectInterface) (*internal.Conv, error) {
@@ -109,6 +117,7 @@ func (sads *SchemaFromSourceImpl) SchemaFromDump(driver string, spDialect string
 	ioHelper.BytesRead = n
 	conv := internal.MakeConv()
 	conv.SpDialect = spDialect
+	conv.Source = driver
 	p := internal.NewProgress(n, "Generating schema", internal.Verbose(), false, int(internal.SchemaCreationInProgress))
 	r := internal.NewReader(bufio.NewReader(f), p)
 	conv.SetSchemaMode() // Build schema and ignore data in dump.
@@ -159,6 +168,9 @@ func (sads *DataFromSourceImpl) dataFromCSV(ctx context.Context, sourceProfile p
 		return nil, fmt.Errorf("dbName is mandatory in target-profile for csv source")
 	}
 	conv.SpDialect = targetProfile.Conn.Sp.Dialect
+	conv.SpProjectId = targetProfile.Conn.Sp.Project
+	conv.SpInstanceId = targetProfile.Conn.Sp.Instance
+	conv.Source = sourceProfile.Driver
 	dialect, err := targetProfile.FetchTargetDialect(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("could not fetch dialect: %v", err)

conversion/conversion_helper.go

@@ -29,6 +29,7 @@ import (
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/metrics"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -45,7 +46,9 @@ type ProcessDumpByDialectInterface interface{
 	ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error
 }
 
-type ProcessDumpByDialectImpl struct{}
+type ProcessDumpByDialectImpl struct{
+	ExpressionVerificationAccessor expressions_api.ExpressionVerificationAccessor
+}
 
 type PopulateDataConvInterface interface{
 	populateDataConv(conv *internal.Conv, config writer.BatchWriterConfig, client *sp.Client) *writer.BatchWriter
@@ -88,9 +91,9 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
 func (pdd *ProcessDumpByDialectImpl) ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 	switch driver {
 	case constants.MYSQLDUMP:
-		return common.ProcessDbDump(conv, r, mysql.DbDumpImpl{})
+		return common.ProcessDbDump(conv, r, mysql.DbDumpImpl{pdd.ExpressionVerificationAccessor})
 	case constants.PGDUMP:
-		return common.ProcessDbDump(conv, r, postgres.DbDumpImpl{})
+		return common.ProcessDbDump(conv, r, postgres.DbDumpImpl{pdd.ExpressionVerificationAccessor})
 	default:
 		return fmt.Errorf("process dump for driver %s not supported", driver)
 	}

expressions_api/expression_verify.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"sync"
 
+	spannerclient "github.com/GoogleCloudPlatform/spanner-migration-tool/accessors/clients/spanner/client"
 	spanneraccessor "github.com/GoogleCloudPlatform/spanner-migration-tool/accessors/spanner"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/task"
@@ -18,22 +19,50 @@ const THREAD_POOL = 500
 type ExpressionVerificationAccessor interface {
 	//Batch API which parallelizes expression verification calls
 	VerifyExpressions(ctx context.Context, verifyExpressionsInput internal.VerifyExpressionsInput) internal.VerifyExpressionsOutput
+	RefreshSpannerClient(ctx context.Context, project string, instance string) error
 }
 
 type ExpressionVerificationAccessorImpl struct {
 	SpannerAccessor *spanneraccessor.SpannerAccessorImpl
 }
 
 func NewExpressionVerificationAccessorImpl(ctx context.Context, project string, instance string) (*ExpressionVerificationAccessorImpl, error) {
-	spannerAccessor, err := spanneraccessor.NewSpannerAccessorClientImplWithSpannerClient(ctx, fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, "smt-staging-db"))
-	if err != nil {
-		return nil, err
+	var spannerAccessor *spanneraccessor.SpannerAccessorImpl
+	var err error
+	if project != "" && instance != "" {
+		spannerAccessor, err = spanneraccessor.NewSpannerAccessorClientImplWithSpannerClient(ctx, fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, constants.TEMP_DB))
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		spannerAccessor, err = spanneraccessor.NewSpannerAccessorClientImpl(ctx)
+		if err != nil {
+			return nil, err
+		}
 	}
 	return &ExpressionVerificationAccessorImpl{
 		SpannerAccessor: spannerAccessor,
 	}, nil
 }
 
+// APIs to verify and process Spanner DLL features such as Default Values, Check Constraints
+type DDLVerifier interface {
+	VerifySpannerDDL(conv *internal.Conv, expressionDetails []internal.ExpressionDetail) (internal.VerifyExpressionsOutput, error)
+	GetSourceExpressionDetails(conv *internal.Conv, tableIds []string) []internal.ExpressionDetail
+	GetSpannerExpressionDetails(conv *internal.Conv, tableIds []string) []internal.ExpressionDetail
+	RefreshSpannerClient(ctx context.Context, project string, instance string) error
+}
+type DDLVerifierImpl struct {
+	Expressions ExpressionVerificationAccessor
+}
+
+func NewDDLVerifierImpl(ctx context.Context, project string, instance string) (*DDLVerifierImpl, error) {
+	expVerifier, err := NewExpressionVerificationAccessorImpl(ctx, project, instance)
+	return &DDLVerifierImpl{
+		Expressions: expVerifier,
+	}, err
+}
+
 func (ev *ExpressionVerificationAccessorImpl) VerifyExpressions(ctx context.Context, verifyExpressionsInput internal.VerifyExpressionsInput) internal.VerifyExpressionsOutput {
 	err := ev.validateRequest(verifyExpressionsInput)
 	if err != nil {
@@ -79,6 +108,15 @@ func (ev *ExpressionVerificationAccessorImpl) VerifyExpressions(ctx context.Cont
 	return verifyExpressionsOutput
 }
 
+func (ev *ExpressionVerificationAccessorImpl) RefreshSpannerClient(ctx context.Context, project string, instance string) error {
+	spannerClient, err := spannerclient.NewSpannerClientImpl(ctx, fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, constants.TEMP_DB))
+	if err != nil {
+		return err
+	}
+	ev.SpannerAccessor.SpannerClient = spannerClient
+	return nil
+}
+
 func (ev *ExpressionVerificationAccessorImpl) verifyExpressionInternal(expressionDetail internal.ExpressionDetail, mutex *sync.Mutex) task.TaskResult[internal.ExpressionVerificationOutput] {
 	var sqlStatement string
 	switch expressionDetail.Type {
@@ -122,10 +160,77 @@ func (ev *ExpressionVerificationAccessorImpl) removeExpressions(inputConv *inter
 	//TODO: Implement similar checks for DEFAULT and CHECK constraints as well
 	convCopy.SpSequences = nil
 	for _, table := range convCopy.SpSchema {
+		table.CheckConstraints = []ddl.CheckConstraint{}
+		convCopy.SpSchema[table.Id] = table
+
 		for colName, colDef := range table.ColDefs {
 			colDef.AutoGen = ddl.AutoGenCol{}
 			table.ColDefs[colName] = colDef
 		}
 	}
 	return convCopy, nil
 }
+
+func (ddlv *DDLVerifierImpl) VerifySpannerDDL(conv *internal.Conv, expressionDetails []internal.ExpressionDetail) (internal.VerifyExpressionsOutput, error) {
+	ctx := context.Background()
+	verifyExpressionsInput := internal.VerifyExpressionsInput{
+		Conv:                 conv,
+		Source:               conv.Source,
+		ExpressionDetailList: expressionDetails,
+	}
+	verificationResults := ddlv.Expressions.VerifyExpressions(ctx, verifyExpressionsInput)
+
+	return verificationResults, verificationResults.Err
+}
+
+func (ddlv *DDLVerifierImpl) GetSourceExpressionDetails(conv *internal.Conv, tableIds []string) []internal.ExpressionDetail {
+	expressionDetails := []internal.ExpressionDetail{}
+	// Collect default values for verification
+	for _, tableId := range tableIds {
+		srcTable := conv.SrcSchema[tableId]
+		for _, srcColId := range srcTable.ColIds {
+			srcCol := srcTable.ColDefs[srcColId]
+			if srcCol.DefaultValue.IsPresent {
+				defaultValueExp := internal.ExpressionDetail{
+					ReferenceElement: internal.ReferenceElement{
+						Name: conv.SpSchema[tableId].ColDefs[srcColId].T.Name,
+					},
+					ExpressionId: srcCol.DefaultValue.Value.ExpressionId,
+					Expression:   srcCol.DefaultValue.Value.Statement,
+					Type:         "DEFAULT",
+					Metadata:     map[string]string{"TableId": tableId, "ColId": srcColId},
+				}
+				expressionDetails = append(expressionDetails, defaultValueExp)
+			}
+		}
+	}
+	return expressionDetails
+}
+
+func (ddlv *DDLVerifierImpl) GetSpannerExpressionDetails(conv *internal.Conv, tableIds []string) []internal.ExpressionDetail {
+	expressionDetails := []internal.ExpressionDetail{}
+	// Collect default values for verification
+	for _, tableId := range tableIds {
+		spTable := conv.SpSchema[tableId]
+		for _, spColId := range spTable.ColIds {
+			spCol := spTable.ColDefs[spColId]
+			if spCol.DefaultValue.IsPresent {
+				defaultValueExp := internal.ExpressionDetail{
+					ReferenceElement: internal.ReferenceElement{
+						Name: conv.SpSchema[tableId].ColDefs[spColId].T.Name,
+					},
+					ExpressionId: spCol.DefaultValue.Value.ExpressionId,
+					Expression:   spCol.DefaultValue.Value.Statement,
+					Type:         "DEFAULT",
+					Metadata:     map[string]string{"TableId": tableId, "ColId": spColId},
+				}
+				expressionDetails = append(expressionDetails, defaultValueExp)
+			}
+		}
+	}
+	return expressionDetails
+}
+
+func (ddlv *DDLVerifierImpl) RefreshSpannerClient(ctx context.Context, project string, instance string) error {
+	return ddlv.Expressions.RefreshSpannerClient(ctx, project, instance)
+}