Check constraint backend (#9)

Backend Support for Check Constraint

internal/convert.go

@@ -132,6 +132,7 @@ const (
 	ForeignKeyActionNotSupported
 	NumericPKNotSupported
 	DefaultValueError
+	TypeMismatch
 )
 
 const (

internal/helpers.go

@@ -65,6 +65,11 @@ func GenerateForeignkeyId() string {
 func GenerateIndexesId() string {
 	return GenerateId("i")
 }
+
+func GenerateCheckConstrainstId() string {
+	return GenerateId("ck")
+}
+
 func GenerateRuleId() string {
 	return GenerateId("r")
 }

internal/mapping.go

@@ -243,6 +243,19 @@ func ToSpannerIndexName(conv *Conv, srcIndexName string) string {
 	return getSpannerValidName(conv, srcIndexName)
 }
 
+// Note that the check constraints names in spanner have to be globally unique
+// (across the database). But in some source databases, such as MySQL,
+// they only have to be unique for a table. Hence we must map each source
+// constraint name to a unique spanner constraint name.
+func ToSpannerCheckConstraintName(conv *Conv, srcCheckConstraintName string) string {
+	return getSpannerValidName(conv, srcCheckConstraintName)
+}
+
+func GetSpannerValidExpression(cks []ddl.CheckConstraint) []ddl.CheckConstraint {
+	// TODO validate the check constraints data with batch verification then send back
+	return cks
+}
+
 // conv.UsedNames tracks Spanner names that have been used for table names, foreign key constraints
 // and indexes. We use this to ensure we generate unique names when
 // we map from source dbs to Spanner since Spanner requires all these names to be

internal/reports/report_helpers.go

@@ -409,6 +409,13 @@ func buildTableReportBody(conv *internal.Conv, tableId string, issues map[string
 						Description: fmt.Sprintf("%s for table '%s' column '%s'", IssueDB[i].Brief, conv.SpSchema[tableId].Name, spColName),
 					}
 					l = append(l, toAppend)
+				case internal.TypeMismatch:
+					toAppend := Issue{
+						Category:    IssueDB[i].Category,
+						Description: fmt.Sprintf("Table '%s': Type mismatch in '%s'column affecting check constraints. Verify data type compatibility with constraint logic", conv.SpSchema[tableId].Name, conv.SpSchema[tableId].ColDefs[colId].Name),
+					}
+					l = append(l, toAppend)
+
 				default:
 					toAppend := Issue{
 						Category:    IssueDB[i].Category,

schema/schema.go

@@ -35,15 +35,16 @@ import (
 
 // Table represents a database table.
 type Table struct {
-	Name         string
-	Schema       string
-	ColIds       []string          // List of column Ids (for predictable iteration order e.g. printing).
-	ColDefs      map[string]Column // Details of columns.
-	ColNameIdMap map[string]string `json:"-"` // Computed every time just after conv is generated or after any column renaming
-	PrimaryKeys  []Key
-	ForeignKeys  []ForeignKey
-	Indexes      []Index
-	Id           string
+	Name             string
+	Schema           string
+	ColIds           []string          // List of column Ids (for predictable iteration order e.g. printing).
+	ColDefs          map[string]Column // Details of columns.
+	ColNameIdMap     map[string]string `json:"-"` // Computed every time just after conv is generated or after any column renaming
+	PrimaryKeys      []Key
+	ForeignKeys      []ForeignKey
+	CheckConstraints []CheckConstraint
+	Indexes          []Index
+	Id               string
 }
 
 // Column represents a database column.
@@ -77,6 +78,13 @@ type ForeignKey struct {
 	Id               string
 }
 
+// CheckConstraints represents a check constraint defined in the schema.
+type CheckConstraint struct {
+	Name string
+	Expr string
+	Id   string
+}
+
 // Key respresents a primary key or index key.
 type Key struct {
 	ColId string

sources/common/infoschema.go

@@ -38,7 +38,7 @@ type InfoSchema interface {
 	GetColumns(conv *internal.Conv, table SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error)
 	GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error)
 	GetRowCount(table SchemaAndName) (int64, error)
-	GetConstraints(conv *internal.Conv, table SchemaAndName) ([]string, map[string][]string, error)
+	GetConstraints(conv *internal.Conv, table SchemaAndName) ([]string, []schema.CheckConstraint, map[string][]string, error)
 	GetForeignKeys(conv *internal.Conv, table SchemaAndName) (foreignKeys []schema.ForeignKey, err error)
 	GetIndexes(conv *internal.Conv, table SchemaAndName, colNameIdMp map[string]string) ([]schema.Index, error)
 	ProcessData(conv *internal.Conv, tableId string, srcSchema schema.Table, spCols []string, spSchema ddl.CreateTable, additionalAttributes internal.AdditionalDataAttributes) error
@@ -187,7 +187,7 @@ func (is *InfoSchemaImpl) processTable(conv *internal.Conv, table SchemaAndName,
 	var t schema.Table
 	fmt.Println("processing schema for table", table)
 	tblId := internal.GenerateTableId()
-	primaryKeys, constraints, err := infoSchema.GetConstraints(conv, table)
+	primaryKeys, checkConstraints, constraints, err := infoSchema.GetConstraints(conv, table)
 	if err != nil {
 		return t, fmt.Errorf("couldn't get constraints for table %s.%s: %s", table.Schema, table.Name, err)
 	}
@@ -217,15 +217,16 @@ func (is *InfoSchemaImpl) processTable(conv *internal.Conv, table SchemaAndName,
 		schemaPKeys = append(schemaPKeys, schema.Key{ColId: colNameIdMap[k]})
 	}
 	t = schema.Table{
-		Id:           tblId,
-		Name:         name,
-		Schema:       table.Schema,
-		ColIds:       colIds,
-		ColNameIdMap: colNameIdMap,
-		ColDefs:      colDefs,
-		PrimaryKeys:  schemaPKeys,
-		Indexes:      indexes,
-		ForeignKeys:  foreignKeys}
+		Id:               tblId,
+		Name:             name,
+		Schema:           table.Schema,
+		ColIds:           colIds,
+		ColNameIdMap:     colNameIdMap,
+		ColDefs:          colDefs,
+		PrimaryKeys:      schemaPKeys,
+		CheckConstraints: checkConstraints,
+		Indexes:          indexes,
+		ForeignKeys:      foreignKeys}
 	return t, nil
 }
 

sources/common/toddl.go

@@ -167,14 +167,15 @@ func (ss *SchemaToSpannerImpl) SchemaToSpannerDDLHelper(conv *internal.Conv, tod
 	}
 	comment := "Spanner schema for source table " + quoteIfNeeded(srcTable.Name)
 	conv.SpSchema[srcTable.Id] = ddl.CreateTable{
-		Name:        spTableName,
-		ColIds:      spColIds,
-		ColDefs:     spColDef,
-		PrimaryKeys: cvtPrimaryKeys(srcTable.PrimaryKeys),
-		ForeignKeys: cvtForeignKeys(conv, spTableName, srcTable.Id, srcTable.ForeignKeys, isRestore),
-		Indexes:     cvtIndexes(conv, srcTable.Id, srcTable.Indexes, spColIds, spColDef),
-		Comment:     comment,
-		Id:          srcTable.Id}
+		Name:             spTableName,
+		ColIds:           spColIds,
+		ColDefs:          spColDef,
+		PrimaryKeys:      cvtPrimaryKeys(srcTable.PrimaryKeys),
+		ForeignKeys:      cvtForeignKeys(conv, spTableName, srcTable.Id, srcTable.ForeignKeys, isRestore),
+		CheckConstraints: cvtCheckConstraint(conv, srcTable.CheckConstraints),
+		Indexes:          cvtIndexes(conv, srcTable.Id, srcTable.Indexes, spColIds, spColDef),
+		Comment:          comment,
+		Id:               srcTable.Id}
 	return nil
 }
 
@@ -234,6 +235,21 @@ func cvtForeignKeys(conv *internal.Conv, spTableName string, srcTableId string,
 	return spKeys
 }
 
+func cvtCheckConstraint(conv *internal.Conv, srcKeys []schema.CheckConstraint) []ddl.CheckConstraint {
+	var spcks []ddl.CheckConstraint
+
+	for _, cks := range srcKeys {
+		spcks = append(spcks, ddl.CheckConstraint{
+			Id:   cks.Id,
+			Name: internal.ToSpannerCheckConstraintName(conv, cks.Name),
+			Expr: cks.Expr,
+		})
+
+	}
+
+	return internal.GetSpannerValidExpression(spcks)
+}
+
 func CvtForeignKeysHelper(conv *internal.Conv, spTableName string, srcTableId string, srcKey schema.ForeignKey, isRestore bool) (ddl.Foreignkey, error) {
 	if len(srcKey.ColIds) != len(srcKey.ReferColumnIds) {
 		conv.Unexpected(fmt.Sprintf("ConvertForeignKeys: ColIds and referColumns don't have the same lengths: len(columns)=%d, len(referColumns)=%d for source tableId: %s, referenced table: %s", len(srcKey.ColIds), len(srcKey.ReferColumnIds), srcTableId, srcKey.ReferTableId))

sources/common/toddl_test.go

@@ -536,4 +536,33 @@ func TestSpannerSchemaApplyExpressions(t *testing.T) {
 			assert.Equal(t, tc.expectedConv, tc.conv)
 		})
 	}
+func Test_cvtCheckContraint(t *testing.T) {
+
+	conv := internal.MakeConv()
+	srcSchema := []schema.CheckConstraint{
+		{
+			Id:   "ck1",
+			Name: "check_1",
+			Expr: "age > 0",
+		},
+		{
+			Id:   "ck1",
+			Name: "check_2",
+			Expr: "age < 99",
+		},
+	}
+	spSchema := []ddl.CheckConstraint{
+		{
+			Id:   "ck1",
+			Name: "check_1",
+			Expr: "age > 0",
+		},
+		{
+			Id:   "ck1",
+			Name: "check_2",
+			Expr: "age < 99",
+		},
+	}
+	result := cvtCheckConstraint(conv, srcSchema)
+	assert.Equal(t, spSchema, result)
 }

sources/dynamodb/schema.go

@@ -129,20 +129,20 @@ func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error)
 	return *result.Table.ItemCount, err
 }
 
-func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) (primaryKeys []string, constraints map[string][]string, err error) {
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) (primaryKeys []string, checkConstraints []schema.CheckConstraint, constraints map[string][]string, err error) {
 	input := &dynamodb.DescribeTableInput{
 		TableName: aws.String(table.Name),
 	}
 	result, err := isi.DynamoClient.DescribeTable(input)
 	if err != nil {
-		return primaryKeys, constraints, fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", table.Name, err)
+		return primaryKeys, checkConstraints, constraints, fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", table.Name, err)
 	}
 
 	// Primary keys.
 	for _, i := range result.Table.KeySchema {
 		primaryKeys = append(primaryKeys, *i.AttributeName)
 	}
-	return primaryKeys, constraints, nil
+	return primaryKeys, checkConstraints, constraints, nil
 }
 
 func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {

sources/dynamodb/schema_test.go

@@ -633,7 +633,7 @@ func TestInfoSchemaImpl_GetConstraints(t *testing.T) {
 	dySchema := common.SchemaAndName{Name: "test"}
 	conv := internal.MakeConv()
 	isi := InfoSchemaImpl{client, nil, 10}
-	primaryKeys, constraints, err := isi.GetConstraints(conv, dySchema)
+	primaryKeys, _, constraints, err := isi.GetConstraints(conv, dySchema)
 	assert.Nil(t, err)
 
 	pKeys := []string{"a", "b"}
@@ -705,7 +705,7 @@ func TestInfoSchemaImpl_GetColumns(t *testing.T) {
 	client := &mockDynamoClient{
 		scanOutputs: scanOutputs,
 	}
-	dySchema := common.SchemaAndName{Name: "test", Id:  "t1"}
+	dySchema := common.SchemaAndName{Name: "test", Id: "t1"}
 
 	isi := InfoSchemaImpl{client, nil, 10}
 

sources/mysql/infoschema.go

@@ -194,16 +194,6 @@ func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAnd
 			continue
 		}
 		ignored := schema.Ignored{}
-		for _, c := range constraints[colName] {
-			// c can be UNIQUE, PRIMARY KEY, FOREIGN KEY or CHECK
-			// We've already filtered out PRIMARY KEY.
-			switch c {
-			case "CHECK":
-				ignored.Check = true
-			case "FOREIGN KEY", "PRIMARY KEY", "UNIQUE":
-				// Nothing to do here -- these are all handled elsewhere.
-			}
-		}
 		ignored.Default = colDefault.Valid
 		colId := internal.GenerateColumnId()
 		if colExtra.String == "auto_increment" {
@@ -250,38 +240,98 @@ func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAnd
 // other constraints.  Note: we need to preserve ordinal order of
 // columns in primary key constraints.
 // Note that foreign key constraints are handled in getForeignKeys.
-func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, map[string][]string, error) {
-	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
-              FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
-                INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
-                  ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA AND t.TABLE_NAME=k.TABLE_NAME
-              WHERE k.TABLE_SCHEMA = ? AND k.TABLE_NAME = ? ORDER BY k.ordinal_position;`
-	rows, err := isi.Db.Query(q, table.Schema, table.Name)
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, []schema.CheckConstraint, map[string][]string, error) {
+	tableExists, err := isi.isCheckConstraintsTablePresent()
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
+	}
+
+	finalQuery := isi.getQuery(tableExists)
+	rows, err := isi.Db.Query(finalQuery, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, nil, err
 	}
 	defer rows.Close()
+
 	var primaryKeys []string
-	var col, constraint string
+	var checkKeys []schema.CheckConstraint
 	m := make(map[string][]string)
+
 	for rows.Next() {
-		err := rows.Scan(&col, &constraint)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
-			continue
-		}
-		if col == "" || constraint == "" {
-			conv.Unexpected(fmt.Sprintf("Got empty col or constraint"))
+		if err := isi.processRow(rows, tableExists, conv, &primaryKeys, &checkKeys, m); err != nil {
 			continue
 		}
-		switch constraint {
-		case "PRIMARY KEY":
-			primaryKeys = append(primaryKeys, col)
-		default:
-			m[col] = append(m[col], constraint)
-		}
 	}
-	return primaryKeys, m, nil
+
+	return primaryKeys, checkKeys, m, nil
+}
+
+// checkCheckConstraintsTableExists checks if the CHECK_CONSTRAINTS table exists.
+func (isi InfoSchemaImpl) isCheckConstraintsTablePresent() (bool, error) {
+	var tableExistsCount int
+	checkQuery := `SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'INFORMATION_SCHEMA' AND TABLE_NAME = 'CHECK_CONSTRAINTS';`
+	err := isi.Db.QueryRow(checkQuery).Scan(&tableExistsCount)
+	if err != nil {
+		return false, err
+	}
+	return tableExistsCount > 0, nil
+}
+
+// getQuery returns the appropriate SQL query based on the existence of CHECK_CONSTRAINTS.
+func (isi InfoSchemaImpl) getQuery(tableExists bool) string {
+	if tableExists {
+		return `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE, COALESCE(c.CHECK_CLAUSE, '') AS CHECK_CLAUSE
+            FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
+            LEFT JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
+            ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME 
+            AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA 
+            AND t.TABLE_NAME = k.TABLE_NAME
+            LEFT JOIN INFORMATION_SCHEMA.CHECK_CONSTRAINTS AS c
+            ON t.CONSTRAINT_NAME = c.CONSTRAINT_NAME
+            WHERE t.TABLE_SCHEMA = ? 
+            AND t.TABLE_NAME = ?
+            ORDER BY k.ORDINAL_POSITION;`
+	}
+	return `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
+            FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
+            INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
+            ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME 
+            AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA 
+            AND t.TABLE_NAME = k.TABLE_NAME
+            WHERE t.TABLE_SCHEMA = ?
+            AND t.TABLE_NAME = ?
+            ORDER BY k.ORDINAL_POSITION;`
+}
+
+// processRow handles scanning and processing of a database row for GetConstraints.
+func (isi InfoSchemaImpl) processRow(
+	rows *sql.Rows, tableExists bool, conv *internal.Conv, primaryKeys *[]string,
+	checkKeys *[]schema.CheckConstraint, m map[string][]string) error {
+
+	var col, constraintType, checkClause string
+	err := rows.Scan(&col, &constraintType, &checkClause)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+		return err
+	}
+
+	if col == "" || constraintType == "" {
+		conv.Unexpected("Got empty column or constraint type")
+		return nil
+	}
+
+	switch constraintType {
+	case "PRIMARY KEY":
+		*primaryKeys = append(*primaryKeys, col)
+	case "CHECK":
+		checkClause = strings.ReplaceAll(checkClause, "_utf8mb4\\", "")
+		checkClause = strings.ReplaceAll(checkClause, "\\", "")
+		constraintName := fmt.Sprintf("%s_check", col)
+		*checkKeys = append(*checkKeys, schema.CheckConstraint{Name: constraintName, Expr: checkClause, Id: internal.GenerateCheckConstrainstId()})
+	default:
+		m[col] = append(m[col], constraintType)
+	}
+	return nil
 }
 
 // GetForeignKeys return list all the foreign keys constraints.