FK Actions: Handling interleaving & dropping issues correctly when dropping FKs

common/metrics/metrics.go

@@ -76,10 +76,10 @@ func getMigrationDataSchemaPatterns(conv *internal.Conv, migrationData *migratio
 		numIndexes += int32(len(table.Indexes))
 		depth := 0
 		tableId := table.Id
-		for conv.SpSchema[tableId].ParentId != "" {
+		for conv.SpSchema[tableId].ParentTable.Id != "" {
 			numInterleaves++
 			depth++
-			tableId = conv.SpSchema[tableId].ParentId
+			tableId = conv.SpSchema[tableId].ParentTable.Id
 		}
 		maxInterleaveDepth = int32(math.Max(float64(maxInterleaveDepth), float64(depth)))
 	}

common/utils/utils.go

@@ -513,19 +513,19 @@ func ReadSpannerSchema(ctx context.Context, conv *internal.Conv, client *sp.Clie
 	if err != nil {
 		return fmt.Errorf("error trying to read and convert spanner schema: %v", err)
 	}
-	parentTables, err := infoSchema.GetInterleaveTables()
+	parentTables, err := infoSchema.GetInterleaveTables(conv.SpSchema)
 	if err != nil {
 		// We should ideally throw an error here as it could potentially cause a lot of failed writes.
 		// We raise an unexpected error for now to make it compatible with the integration tests.
 		// In the emulator, the interleave_type column in not supported hence the query fails.
 		conv.Unexpected(fmt.Sprintf("error trying to fetch interleave table info from schema: %v", err))
 	}
 	// Assign parents if any.
-	for tableName, parentName := range parentTables {
+	for tableName, parentTable := range parentTables {
 		tableId, _ := internal.GetTableIdFromSpName(conv.SpSchema, tableName)
-		parentTableId, _ := internal.GetTableIdFromSpName(conv.SpSchema, parentName)
 		spTable := conv.SpSchema[tableId]
-		spTable.ParentId = parentTableId
+		spTable.ParentTable.Id = parentTable.Id
+		spTable.ParentTable.OnDelete = parentTable.OnDelete
 		conv.SpSchema[tableId] = spTable
 	}
 	return nil
@@ -542,8 +542,8 @@ func CompareSchema(sessionFileConv, actualSpannerConv *internal.Conv) error {
 			return fmt.Errorf("table %v not found in the spanner database schema but found in the session file. If this table does not need to be migrated, please exclude it during the schema conversion and migration process", sessionTable.Name)
 		}
 		spannerTable := actualSpannerConv.SpSchema[spannerTableId]
-		sessionTableParentName := sessionFileConv.SpSchema[sessionTable.ParentId].Name
-		spannerTableParentName := actualSpannerConv.SpSchema[spannerTable.ParentId].Name
+		sessionTableParentName := sessionFileConv.SpSchema[sessionTable.ParentTable.Id].Name
+		spannerTableParentName := actualSpannerConv.SpSchema[spannerTable.ParentTable.Id].Name
 
 		//table names should match
 		if sessionTable.Name != spannerTable.Name {
@@ -555,6 +555,11 @@ func CompareSchema(sessionFileConv, actualSpannerConv *internal.Conv) error {
 			return fmt.Errorf("parent table name don't match: session table %v, parent session table name: %v, spanner table %v, parent spanner table name: %v", sessionTable.Name, sessionTableParentName, spannerTable.Name, spannerTableParentName)
 		}
 
+		//parent table on delete actions should match
+		if sessionTable.ParentTable.OnDelete != spannerTable.ParentTable.OnDelete {
+			return fmt.Errorf("parent table on delete actions don't match: session table %v, parent session table name: %v, spanner table %v, parent spanner table name: %v", sessionTable.Name, sessionTable.ParentTable.OnDelete, spannerTable.Name, spannerTable.ParentTable.OnDelete)
+		}
+
 		//number of columns should match
 		if len(sessionTable.ColDefs) != len(spannerTable.ColDefs) {
 			return fmt.Errorf("number of columns don't match: session table %v, spanner table %v", sessionTable.Name, spannerTable.Name)
@@ -659,4 +664,4 @@ func SetDataflowTemplatePath(path string) {
 
 func GetDataflowTemplatePath() string {
 	return dataflowTemplatePath
-}
+}

sources/spanner/infoschema.go

@@ -387,19 +387,19 @@ func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAnd
 	return indexes, nil
 }
 
-func (isi InfoSchemaImpl) GetInterleaveTables() (map[string]string, error) {
-	q := `SELECT table_name, parent_table_name FROM information_schema.tables 
+func (isi InfoSchemaImpl) GetInterleaveTables(spSchema ddl.Schema) (map[string]ddl.InterleavedParent, error) {
+	q := `SELECT table_name, parent_table_name, on_delete_action FROM information_schema.tables 
 	WHERE interleave_type = 'IN PARENT' AND table_type = 'BASE TABLE' AND table_schema = ''`
 	if isi.SpDialect == constants.DIALECT_POSTGRESQL {
-		q = `SELECT table_name, parent_table_name FROM information_schema.tables 
+		q = `SELECT table_name, parent_table_name, on_delete_action FROM information_schema.tables 
 		WHERE interleave_type = 'IN PARENT' AND table_type = 'BASE TABLE' AND table_schema = 'public'`
 	}
 	stmt := spanner.Statement{SQL: q}
 	iter := isi.Client.Single().Query(isi.Ctx, stmt)
 	defer iter.Stop()
 
-	var tableName, parentTable string
-	parentTables := map[string]string{}
+	var tableName, parentTableName, onDelete string
+	parentTables := map[string]ddl.InterleavedParent{}
 	for {
 		row, err := iter.Next()
 		if err == iterator.Done {
@@ -408,11 +408,12 @@ func (isi InfoSchemaImpl) GetInterleaveTables() (map[string]string, error) {
 		if err != nil {
 			return nil, fmt.Errorf("couldn't read row while fetching interleaved tables: %w", err)
 		}
-		err = row.Columns(&tableName, &parentTable)
+		err = row.Columns(&tableName, &parentTableName, &onDelete)
 		if err != nil {
 			return nil, err
 		}
-		parentTables[tableName] = parentTable
+		parentTableId, _ := internal.GetTableIdFromSpName(spSchema, parentTableName)
+		parentTables[tableName] = ddl.InterleavedParent{Id: parentTableId, OnDelete: onDelete}
 	}
 	return parentTables, nil
 }

spanner/ddl/ast.go

@@ -282,6 +282,14 @@ type Foreignkey struct {
 	OnUpdate       string
 }
 
+// InterleavedParent encodes the following DDL definition:
+//
+//	INTERLEAVE IN PARENT parent_name ON DELETE delete_rule
+type InterleavedParent struct {
+	Id       string
+	OnDelete string
+}
+
 // PrintForeignKey unparses the foreign keys.
 func (k Foreignkey) PrintForeignKey(c Config) string {
 	var cols, referCols []string
@@ -311,7 +319,7 @@ type CreateTable struct {
 	PrimaryKeys   []IndexKey
 	ForeignKeys   []Foreignkey
 	Indexes       []CreateIndex
-	ParentId      string //if not empty, this table will be interleaved
+	ParentTable   InterleavedParent //if not empty, this table will be interleaved
 	Comment       string
 	Id            string
 }
@@ -353,15 +361,18 @@ func (ct CreateTable) PrintCreateTable(spSchema Schema, config Config) string {
 	}
 
 	var interleave string
-	if ct.ParentId != "" {
-		parent := spSchema[ct.ParentId].Name
+	if ct.ParentTable.Id != "" {
+		parent := spSchema[ct.ParentTable.Id].Name
 		if config.SpDialect == constants.DIALECT_POSTGRESQL {
 			// PG spanner only supports PRIMARY KEY() inside the CREATE TABLE()
 			// and thus INTERLEAVE follows immediately after closing brace.
 			interleave = " INTERLEAVE IN PARENT " + config.quote(parent)
 		} else {
 			interleave = ",\nINTERLEAVE IN PARENT " + config.quote(parent)
 		}
+		if ct.ParentTable.OnDelete != "" {
+			interleave = interleave + " ON DELETE " + ct.ParentTable.OnDelete
+		}
 	}
 
 	if len(keys) == 0 {
@@ -507,14 +518,14 @@ func GetSortedTableIdsBySpName(s Schema) []string {
 		table := s[tableNameIdMap[tableName]]
 		tableQueue = tableQueue[1:]
 		parentTableExists := false
-		if table.ParentId != "" {
-			_, parentTableExists = s[table.ParentId]
+		if table.ParentTable.Id != "" {
+			_, parentTableExists = s[table.ParentTable.Id]
 		}
 
 		// Add table t if either:
 		// a) t is not interleaved in another table, or
 		// b) t is interleaved in another table and that table has already been added to the list.
-		if table.ParentId == "" || tableAdded[s[table.ParentId].Name] || !parentTableExists {
+		if table.ParentTable.Id == "" || tableAdded[s[table.ParentTable.Id].Name] || !parentTableExists {
 			sortedTableNames = append(sortedTableNames, tableName)
 			tableAdded[tableName] = true
 		} else {
@@ -578,7 +589,7 @@ func GetDDL(c Config, tableSchema Schema, sequenceSchema map[string]Sequence) []
 // CheckInterleaved checks if schema contains interleaved tables.
 func (s Schema) CheckInterleaved() bool {
 	for _, table := range s {
-		if table.ParentId != "" {
+		if table.ParentTable.Id != "" {
 			return true
 		}
 	}