embed time type check in column check

Author: elianddb

enginetest/queries/script_queries.go

@@ -10953,81 +10953,68 @@ where
 		},
 		Assertions: []ScriptTestAssertion{
 			{
-				Skip:  true,
 				Query: "create table child_timestamp0 (ts timestamp, foreign key (ts) references parent_timestamp6(ts));",
 				Expected: []sql.Row{
 					{types.NewOkResult(0)},
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child_timestamp0 values ('2001-02-03 12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
 			{
-				Skip:  true,
 				Query: "create table child_timestamp6 (ts timestamp(6), foreign key (ts) references parent_timestamp0(ts));",
 				Expected: []sql.Row{
 					{types.NewOkResult(0)},
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child_timestamp6 values ('2001-02-03 12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
 
 			{
-				Skip:  true,
 				Query: "create table child1_datetime0 (dt datetime, foreign key (dt) references parent_timestamp0(ts));",
 				Expected: []sql.Row{
 					{types.NewOkResult(0)},
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child1_datetime0 values ('2001-02-03 12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
 			{
-				Skip:  true,
 				Query: "create table child2_datetime0 (dt datetime, foreign key (dt) references parent_timestamp6(ts));",
 				Expected: []sql.Row{
 					{types.NewOkResult(0)},
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child2_datetime0 values ('2001-02-03 12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
 
 			{
-				Skip:  true,
 				Query: "create table child1_datetime6 (dt datetime(6), foreign key (dt) references parent_timestamp0(ts));",
 				Expected: []sql.Row{
 					{types.NewOkResult(0)},
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child1_datetime6 values ('2001-02-03 12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
 			{
-				Skip:  true,
 				Query: "create table child2_datetime6 (dt datetime(6), foreign key (dt) references parent_timestamp6(ts));",
 				Expected: []sql.Row{
 					{types.NewOkResult(0)},
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child2_datetime6 values ('2001-02-03 12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child2_datetime6 values ('2001-02-03 12:34:56.123456');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
@@ -11069,7 +11056,6 @@ where
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child_time0 values ('12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},
@@ -11080,7 +11066,6 @@ where
 				},
 			},
 			{
-				Skip:        true,
 				Query:       "insert into child_time6 values ('12:34:56');",
 				ExpectedErr: sql.ErrForeignKeyChildViolation,
 			},

sql/plan/foreign_key_editor.go

@@ -493,6 +493,7 @@ func (reference *ForeignKeyReferenceHandler) IsInitialized() bool {
 }
 
 // CheckReference checks that the given row has an index entry in the referenced table.
+// Performs MySQL-compatible foreign key constraint validation with type-specific checks.
 func (reference *ForeignKeyReferenceHandler) CheckReference(ctx *sql.Context, row sql.Row) error {
 	// If even one of the values are NULL then we don't check the parent
 	for _, pos := range reference.RowMapper.IndexPositions {
@@ -507,7 +508,7 @@ func (reference *ForeignKeyReferenceHandler) CheckReference(ctx *sql.Context, ro
 	}
 	defer rowIter.Close(ctx)
 
-	_, err = rowIter.Next(ctx)
+	parentRow, err := rowIter.Next(ctx)
 	if err != nil && err != io.EOF {
 		// For SET types, conversion failures during foreign key validation should be treated as foreign key violations
 		if sql.ErrConvertingToSet.Is(err) || sql.ErrInvalidSetValue.Is(err) {
@@ -518,12 +519,10 @@ func (reference *ForeignKeyReferenceHandler) CheckReference(ctx *sql.Context, ro
 	}
 	if err == nil {
 		// We have a parent row, but check for type-specific validation
-		if validationErr := reference.validateDecimalConstraints(row); validationErr != nil {
-			return validationErr
-		}
-		if validationErr := reference.validateTimeConstraints(row); validationErr != nil {
+		if validationErr := reference.validateColumnTypeConstraints(ctx, row, parentRow); validationErr != nil {
 			return validationErr
 		}
+		
 		// We have a parent row so throw no error
 		return nil
 	}
@@ -551,46 +550,14 @@ func (reference *ForeignKeyReferenceHandler) CheckReference(ctx *sql.Context, ro
 		reference.ForeignKey.ParentTable, reference.RowMapper.GetKeyString(row))
 }
 
-// validateDecimalConstraints checks that decimal foreign key columns have compatible scales.
-func (reference *ForeignKeyReferenceHandler) validateDecimalConstraints(row sql.Row) error {
-	if reference.RowMapper.Index == nil {
-		return nil
-	}
-	indexColumnTypes := reference.RowMapper.Index.ColumnExpressionTypes()
-	for parentIdx, parentCol := range indexColumnTypes {
-		if parentIdx >= len(reference.RowMapper.IndexPositions) {
-			break
-		}
-		parentType := parentCol.Type
-		childColIdx := reference.RowMapper.IndexPositions[parentIdx]
-		childType := reference.RowMapper.SourceSch[childColIdx].Type
-		childDecimal, ok := childType.(sql.DecimalType)
-		if !ok {
-			continue
-		}
-		parentDecimal, ok := parentType.(sql.DecimalType)
-		if !ok {
-			continue
-		}
-		if childDecimal.Scale() != parentDecimal.Scale() {
-			return sql.ErrForeignKeyChildViolation.New(
-				reference.ForeignKey.Name,
-				reference.ForeignKey.Table,
-				reference.ForeignKey.ParentTable,
-				reference.RowMapper.GetKeyString(row),
-			)
-		}
-	}
-	return nil
-}
 
-// validateTimeConstraints checks that time-related foreign key columns have exact type and precision matches.
-// MySQL requires strict matching for time types in foreign keys - even logically equivalent values
-// like '2001-02-03 12:34:56' vs '2001-02-03 12:34:56.000000' are rejected if precision differs.
-func (reference *ForeignKeyReferenceHandler) validateTimeConstraints(row sql.Row) error {
+// validateColumnTypeConstraints validates that column types meet MySQL foreign key requirements.
+// Centralizes type validation for decimal scale matching and exact time type precision matching.
+func (reference *ForeignKeyReferenceHandler) validateColumnTypeConstraints(ctx *sql.Context, childRow sql.Row, parentRow sql.Row) error {
 	if reference.RowMapper.Index == nil {
 		return nil
 	}
+	
 	indexColumnTypes := reference.RowMapper.Index.ColumnExpressionTypes()
 	for parentIdx, parentCol := range indexColumnTypes {
 		if parentIdx >= len(reference.RowMapper.IndexPositions) {
@@ -600,27 +567,40 @@ func (reference *ForeignKeyReferenceHandler) validateTimeConstraints(row sql.Row
 		childColIdx := reference.RowMapper.IndexPositions[parentIdx]
 		childType := reference.RowMapper.SourceSch[childColIdx].Type
 
-		// Check if both types are time-related
-		isChildTime := types.IsTime(childType) || types.IsTimespan(childType)
-		isParentTime := types.IsTime(parentType) || types.IsTimespan(parentType)
-		
-		if !isChildTime || !isParentTime {
-			continue
+		// Check decimal constraints
+		childDecimal, isChildDecimal := childType.(sql.DecimalType)
+		parentDecimal, isParentDecimal := parentType.(sql.DecimalType)
+		if isChildDecimal && isParentDecimal {
+			if childDecimal.Scale() != parentDecimal.Scale() {
+				return sql.ErrForeignKeyChildViolation.New(
+					reference.ForeignKey.Name,
+					reference.ForeignKey.Table,
+					reference.ForeignKey.ParentTable,
+					reference.RowMapper.GetKeyString(childRow),
+				)
+			}
 		}
 
-		// MySQL requires exact type matching for time types in foreign key validation
-		if !childType.Equals(parentType) {
-			return sql.ErrForeignKeyChildViolation.New(
-				reference.ForeignKey.Name,
-				reference.ForeignKey.Table,
-				reference.ForeignKey.ParentTable,
-				reference.RowMapper.GetKeyString(row),
-			)
+		// Check time type constraints - MySQL requires exact type matching
+		isChildTime := types.IsTime(childType) || types.IsTimespan(childType)
+		isParentTime := types.IsTime(parentType) || types.IsTimespan(parentType)
+		if isChildTime && isParentTime {
+			// Different precisions of the same base type (e.g., DATETIME vs DATETIME(6)) are rejected
+			// Cross-type references (e.g., DATETIME vs TIMESTAMP) are also rejected
+			if !childType.Equals(parentType) {
+				return sql.ErrForeignKeyChildViolation.New(
+					reference.ForeignKey.Name,
+					reference.ForeignKey.Table,
+					reference.ForeignKey.ParentTable,
+					reference.RowMapper.GetKeyString(childRow),
+				)
+			}
 		}
 	}
 	return nil
 }
 
+
 // CheckTable checks that every row in the table has an index entry in the referenced table.
 func (reference *ForeignKeyReferenceHandler) CheckTable(ctx *sql.Context, tbl sql.ForeignKeyTable) error {
 	partIter, err := tbl.Partitions(ctx)
@@ -678,6 +658,7 @@ func (mapper *ForeignKeyRowMapper) GetIter(ctx *sql.Context, row sql.Row, refChe
 		}
 
 		targetType := mapper.SourceSch[rowPos].Type
+		
 		// Transform the type of the value in this row to the one in the other table for the index lookup, if necessary
 		if mapper.TargetTypeConversions != nil && mapper.TargetTypeConversions[rowPos] != nil {
 			var err error