Repository: pingcap/tidb
File: pkg/ddl/modify_column.go (Go)

Code before:
	sctx sessionctx.Context,
	is infoschema.InfoSchema, // WARN: is maybe nil here.
	ident ast.Ident,
	originalColName ast.CIStr,
	schema *model.DBInfo,
	t table.Table,
	spec *ast.AlterTableSpec,
) (*JobWrapper, error) {
	var err error
	specNewColumn := spec.NewColumns[0]

	col := table.FindCol(t.Cols(), originalColName.L)
	if col == nil {
		return nil, infoschema.ErrColumnNotExists.GenWithStackByArgs(originalColName, ident.Name)
	}
	err = checkColumnReferencedByPartialCondition(t.Meta(), col.ColumnInfo.Name)
	if err != nil {
		return nil, errors.Trace(err)
	}
	newColName := specNewColumn.Name.Name
	if newColName.L == model.ExtraHandleName.L {
		return nil, dbterror.ErrWrongColumnName.GenWithStackByArgs(newColName.L)
	}
	errG := checkModifyColumnWithGeneratedColumnsConstraint(t.Cols(), originalColName)

	// If we want to rename the column name, we need to check whether it already exists.
	if newColName.L != originalColName.L {
		c := table.FindCol(t.Cols(), newColName.L)
		if c != nil {
			return nil, infoschema.ErrColumnExists.GenWithStackByArgs(newColName)
		}

		// And also check the generated columns dependency, if some generated columns
		// depend on this column, we can't rename the column name.
		if errG != nil {
			return nil, errors.Trace(errG)
		}
	}

	// Constraints in the new column means adding new constraints. Errors should thrown,
	// which will be done by `processColumnOptions` later.
	if specNewColumn.Tp == nil {
		// Make sure the column definition is simple field type.
		return nil, errors.Trace(dbterror.ErrUnsupportedModifyColumn)
	}

	if err = checkColumnAttributes(specNewColumn.Name.OrigColName(), specNewColumn.Tp); err != nil {
		return nil, errors.Trace(err)
	}

	newCol := table.ToColumn(&model.ColumnInfo{

Review comment (suggestion):
For non-partitioned table, if we are changing to a more narrow integer type in strict mode, like bigint-> int, we will check if all the current data are in the new range. If not, the ddl won't be executed.

Not related to this PR, I think I can adapt this logic to partition-table too later.

Code after:
				partCol = pc
				break
			}
		}
		if partCol != nil {
			return checkPartitionColumnModifiable(sctx, tblInfo, col, newCol)
		}
	}
	return nil
}

func checkPartitionColumnModifiable(sctx sessionctx.Context, tblInfo *model.TableInfo, col, newCol *model.ColumnInfo) error {
	if col.Name.L != newCol.Name.L {
		return dbterror.ErrDependentByPartitionFunctional.GenWithStackByArgs(col.Name.L)
	}
	if !isColTypeAllowedAsPartitioningCol(tblInfo.Partition.Type, newCol.FieldType) {
		return dbterror.ErrNotAllowedTypeInPartition.GenWithStackByArgs(newCol.Name.O)
	}
	pi := tblInfo.GetPartitionInfo()
	if len(pi.Columns) == 0 {
		// non COLUMNS partitioning, only checks INTs, not their actual range
		// There are many edge cases, like when truncating SQL Mode is allowed
		// which will change the partitioning expression value resulting in a
		// different partition. Better be safe and not allow decreasing of length.
		// TODO: Should we allow it in strict mode? Wait for a use case / request.
		if newCol.FieldType.GetFlen() < col.FieldType.GetFlen() {
			return dbterror.ErrUnsupportedModifyColumn.GenWithStack("Unsupported modify column, decreasing length of int may result in truncation and change of partition")
		}
	}
	// Basically only allow changes of the length/decimals for the column
	// Note that enum is not allowed, so elems are not checked
	// TODO: support partition by ENUM
	if newCol.FieldType.EvalType() != col.FieldType.EvalType() ||
		newCol.FieldType.GetFlag() != col.FieldType.GetFlag() ||
		newCol.FieldType.GetCollate() != col.FieldType.GetCollate() ||
		newCol.FieldType.GetCharset() != col.FieldType.GetCharset() {
		return dbterror.ErrUnsupportedModifyColumn.GenWithStackByArgs("can't change the partitioning column, since it would require reorganize all partitions")
	}
	// Generate a new PartitionInfo and validate it together with the new column definition
	// Checks if all partition definition values are compatible.
	// Similar to what buildRangePartitionDefinitions would do in terms of checks.

	newTblInfo := *tblInfo
	// Replace col with newCol and see if we can generate a new SHOW CREATE TABLE
	// and reparse it and build new partition definitions (which will do additional
	// checks columns vs partition definition values
	newCols := make([]*model.ColumnInfo, 0, len(newTblInfo.Columns))
	for _, c := range newTblInfo.Columns {
		if c.ID == col.ID {
			newCols = append(newCols, newCol)
			continue