ddl: support storage class table attributes

pkg/ddl/BUILD.bazel

@@ -38,6 +38,7 @@ go_library(
         "delete_range_util.go",
         "dist_owner.go",
         "doc.go",
+        "engine_attribute.go",
         "executor.go",
         "foreign_key.go",
         "generated_column.go",
@@ -66,6 +67,7 @@ go_library(
         "schema_version.go",
         "sequence.go",
         "split_region.go",
+        "storage_class.go",
         "stat.go",
         "table.go",
         "table_lock.go",
@@ -302,6 +304,8 @@ go_test(
         "schema_version_test.go",
         "sequence_test.go",
         "stat_test.go",
+        "storage_class_partition_test.go",
+        "storage_class_test.go",
         "table_mode_test.go",
         "table_modify_test.go",
         "table_split_test.go",

pkg/ddl/create_table.go

@@ -523,6 +523,9 @@ func checkTableInfoValidWithStmt(ctx *metabuild.Context, tbInfo *model.TableInfo
 		if err := checkPartitionDefinitionConstraints(ctx.GetExprCtx(), tbInfo); err != nil {
 			return errors.Trace(err)
 		}
+		if err := rebuildStorageClassForPartitions(tbInfo, tbInfo.Partition.Definitions); err != nil {
+			return errors.Trace(err)
+		}
 		if s.Partition != nil {
 			if err := checkPartitionFuncType(ctx.GetExprCtx(), s.Partition.Expr, s.Table.Schema.O, tbInfo); err != nil {
 				return errors.Trace(err)
@@ -939,6 +942,11 @@ func extractAutoRandomBitsFromColDef(colDef *ast.ColumnDef) (shardBits, rangeBit
 func handleTableOptions(options []*ast.TableOption, tbInfo *model.TableInfo) error {
 	var ttlOptionsHandled bool
 
+	engineAttribute, hasEngineAttribute, engineAttributeErr := GetEngineAttributeFromStorageClassTableOptions(options)
+	if engineAttributeErr != nil {
+		return engineAttributeErr
+	}
+
 	for _, op := range options {
 		switch op.Tp {
 		case ast.TableOptionAutoIncrement:
@@ -1000,8 +1008,12 @@ func handleTableOptions(options []*ast.TableOption, tbInfo *model.TableInfo) err
 				return errors.Trace(dbterror.ErrInvalidTableAffinity.GenWithStackByArgs(fmt.Sprintf("'%s'", op.StrValue)))
 			}
 			tbInfo.Affinity = affinity
-		case ast.TableOptionEngineAttribute:
-			return errors.Trace(dbterror.ErrUnsupportedEngineAttribute)
+		case ast.TableOptionEngineAttribute, ast.TableOptionStorageClass:
+		}
+	}
+	if hasEngineAttribute {
+		if err := handleEngineAttributeForCreateTable(engineAttribute, tbInfo); err != nil {
+			return errors.Trace(err)
 		}
 	}
 	shardingBits := shardingBits(tbInfo)

pkg/ddl/engine_attribute.go

@@ -0,0 +1,179 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ddl
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/pkg/infoschema"
+	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/sessionctx"
+	"github.com/pingcap/tidb/pkg/util/dbterror"
+	"github.com/pingcap/tidb/pkg/util/logutil"
+	"go.uber.org/zap"
+)
+
+func handleEngineAttributeForCreateTable(input string, tbInfo *model.TableInfo) error {
+	attr, err := model.ParseEngineAttributeFromString(input)
+	if err != nil {
+		return dbterror.ErrEngineAttributeInvalidFormat.GenWithStackByArgs(fmt.Sprintf("'%v'", err))
+	}
+
+	// Keep the original string for SHOW CREATE TABLE.
+	tbInfo.EngineAttribute = input
+
+	if attr.StorageClass != nil {
+		settings, err := BuildStorageClassSettingsFromJSON(attr.StorageClass)
+		if err != nil {
+			return errors.Trace(err)
+		}
+
+		logutil.BgLogger().Info("storage class: create table with settings",
+			zap.Int64("tableID", tbInfo.ID), zap.Any("settings", settings))
+
+		if err = BuildStorageClassForTable(tbInfo, settings); err != nil {
+			return errors.Trace(err)
+		}
+	}
+
+	// Handle other fields in the future.
+
+	return nil
+}
+
+func getStorageClassSettingsFromTableInfo(tbInfo *model.TableInfo) (*model.StorageClassSettings, error) {
+	attr, err := model.ParseEngineAttributeFromString(tbInfo.EngineAttribute)
+	if err != nil {
+		return nil, dbterror.ErrEngineAttributeInvalidFormat.GenWithStackByArgs(fmt.Sprintf("'%v'", err))
+	}
+
+	if attr.StorageClass == nil {
+		return nil, nil
+	}
+
+	settings, err := BuildStorageClassSettingsFromJSON(attr.StorageClass)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	logutil.BgLogger().Info("storage class: get settings from table info",
+		zap.Int64("tableID", tbInfo.ID), zap.Any("settings", settings))
+	return settings, nil
+}
+
+func rebuildStorageClassForPartitions(tbInfo *model.TableInfo, partitions []model.PartitionDefinition) error {
+	settings, err := getStorageClassSettingsFromTableInfo(tbInfo)
+	if err != nil || settings == nil {
+		return errors.Trace(err)
+	}
+	return BuildStorageClassForPartitions(partitions, tbInfo, settings)
+}
+
+func onModifyTableEngineAttribute(jobCtx *jobContext, job *model.Job) (ver int64, _ error) {
+	args, err := model.GetModifyTableEngineAttributeArgs(job)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	attr, err := model.ParseEngineAttributeFromString(args.EngineAttribute)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	tblInfo, err := GetTableInfoAndCancelFaultJob(jobCtx.metaMut, job, job.SchemaID)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	if job.MultiSchemaInfo != nil && job.MultiSchemaInfo.Revertible {
+		job.MarkNonRevertible()
+		return ver, nil
+	}
+
+	// Keep the original string for SHOW CREATE TABLE.
+	tblInfo.EngineAttribute = args.EngineAttribute
+
+	if err := onAlterTableStorageClassSettings(attr.StorageClass, tblInfo); err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	ver, err = updateVersionAndTableInfo(jobCtx, job, tblInfo, true)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+	job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tblInfo)
+	return ver, nil
+}
+
+func onAlterTableStorageClassSettings(storageClass json.RawMessage, tblInfo *model.TableInfo) error {
+	settings, err := BuildStorageClassSettingsFromJSON(storageClass)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	logutil.BgLogger().Info("storage class: alter table settings",
+		zap.Int64("tableID", tblInfo.ID), zap.Any("settings", settings))
+
+	if err = BuildStorageClassForTable(tblInfo, settings); err != nil {
+		return errors.Trace(err)
+	}
+	if tblInfo.Partition != nil && len(tblInfo.Partition.Definitions) > 0 {
+		if err = BuildStorageClassForPartitions(tblInfo.Partition.Definitions, tblInfo, settings); err != nil {
+			return errors.Trace(err)
+		}
+	}
+
+	return nil
+}
+
+// AlterTableEngineAttribute updates the table engine attribute.
+func (e *executor) AlterTableEngineAttribute(ctx sessionctx.Context, ident ast.Ident, engineAttribute string) error {
+	is := e.infoCache.GetLatest()
+	schema, ok := is.SchemaByName(ident.Schema)
+	if !ok {
+		return infoschema.ErrDatabaseNotExists.GenWithStackByArgs(ident.Schema)
+	}
+
+	tb, err := is.TableByName(e.ctx, ident.Schema, ident.Name)
+	if err != nil {
+		return errors.Trace(infoschema.ErrTableNotExists.GenWithStackByArgs(ident.Schema, ident.Name))
+	}
+	if _, err = model.ParseEngineAttributeFromString(engineAttribute); err != nil {
+		return dbterror.ErrEngineAttributeInvalidFormat.GenWithStackByArgs(fmt.Sprintf("'%v'", err))
+	}
+
+	job := &model.Job{
+		Version:        model.GetJobVerInUse(),
+		SchemaID:       schema.ID,
+		TableID:        tb.Meta().ID,
+		SchemaName:     schema.Name.L,
+		TableName:      tb.Meta().Name.L,
+		Type:           model.ActionModifyEngineAttribute,
+		BinlogInfo:     &model.HistoryInfo{},
+		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
+		SQLMode:        ctx.GetSessionVars().SQLMode,
+	}
+	args := &model.ModifyTableEngineAttributeArgs{
+		EngineAttribute: engineAttribute,
+	}
+	err = e.doDDLJob2(ctx, job, args)
+	return errors.Trace(err)
+}

pkg/ddl/executor.go

@@ -1727,6 +1727,9 @@ func (e *executor) AlterTable(ctx context.Context, sctx sessionctx.Context, stmt
 			return dbterror.ErrOptOnCacheTable.GenWithStackByArgs("Alter Table")
 		}
 	}
+	if err := CheckStorageClassConflictInAlterTableSpecs(validSpecs); err != nil {
+		return err
+	}
 	if isMultiSchemaChanges(validSpecs) && (sctx.GetSessionVars().EnableRowLevelChecksum || vardef.EnableRowLevelChecksum.Load()) {
 		return dbterror.ErrRunMultiSchemaChanges.GenWithStack("Unsupported multi schema change when row level checksum is enabled")
 	}
@@ -1883,6 +1886,10 @@ func (e *executor) AlterTable(ctx context.Context, sctx sessionctx.Context, stmt
 		case ast.AlterTablePartition:
 			err = e.AlterTablePartitioning(sctx, ident, spec)
 		case ast.AlterTableOption:
+			engineAttribute, hasEngineAttribute, engineAttributeErr := GetEngineAttributeFromStorageClassTableOptions(spec.Options)
+			if engineAttributeErr != nil {
+				return engineAttributeErr
+			}
 			var placementPolicyRef *model.PolicyRefInfo
 			for i, opt := range spec.Options {
 				switch opt.Tp {
@@ -1923,8 +1930,7 @@ func (e *executor) AlterTable(ctx context.Context, sctx sessionctx.Context, stmt
 						Name: ast.NewCIStr(opt.StrValue),
 					}
 				case ast.TableOptionEngine:
-				case ast.TableOptionEngineAttribute:
-					err = dbterror.ErrUnsupportedEngineAttribute
+				case ast.TableOptionEngineAttribute, ast.TableOptionStorageClass:
 				case ast.TableOptionRowFormat:
 				case ast.TableOptionTTL, ast.TableOptionTTLEnable, ast.TableOptionTTLJobInterval:
 					var ttlInfo *model.TTLInfo
@@ -1952,6 +1958,13 @@ func (e *executor) AlterTable(ctx context.Context, sctx sessionctx.Context, stmt
 				}
 			}
 
+			if hasEngineAttribute {
+				err = e.AlterTableEngineAttribute(sctx, ident, engineAttribute)
+				if err != nil {
+					return errors.Trace(err)
+				}
+			}
+
 			if placementPolicyRef != nil {
 				err = e.AlterTablePlacement(sctx, ident, placementPolicyRef)
 			}
@@ -2304,13 +2317,7 @@ func (e *executor) AddTablePartitions(ctx sessionctx.Context, ident ast.Ident, s
 		}
 	}
 
-	// partInfo contains only the new added partition, we have to combine it with the
-	// old partitions to check all partitions is strictly increasing.
-	clonedMeta := meta.Clone()
-	tmp := *partInfo
-	tmp.Definitions = append(pi.Definitions, tmp.Definitions...)
-	clonedMeta.Partition = &tmp
-	if err := checkPartitionDefinitionConstraints(ctx.GetExprCtx(), clonedMeta); err != nil {
+	if err := CheckAndUpdateAddedPartitionDefinitions(ctx.GetExprCtx(), meta, partInfo, len(pi.Definitions)); err != nil {
 		if dbterror.ErrSameNamePartition.Equal(err) && spec.IfNotExists {
 			ctx.GetSessionVars().StmtCtx.AppendNote(err)
 			return nil
@@ -2684,6 +2691,13 @@ func checkReorgPartitionDefs(ctx sessionctx.Context, action model.ActionType, tb
 	if err := checkPartitionDefinitionConstraints(ctx.GetExprCtx(), clonedMeta); err != nil {
 		return errors.Trace(err)
 	}
+	definitionsOffset := 0
+	if action == model.ActionReorganizePartition {
+		definitionsOffset = firstPartIdx
+	}
+	if err := updatePartInfoDefinitionsFromFinalDefinitions(clonedMeta, partInfo, definitionsOffset); err != nil {
+		return errors.Trace(err)
+	}
 	if action == model.ActionReorganizePartition {
 		if pi.Type == ast.PartitionTypeRange {
 			if lastPartIdx == len(pi.Definitions)-1 {

pkg/ddl/job_worker.go

@@ -1092,6 +1092,8 @@ func (w *worker) runOneJobStep(
 		ver, err = onDropCheckConstraint(jobCtx, job)
 	case model.ActionAlterCheckConstraint:
 		ver, err = w.onAlterCheckConstraint(jobCtx, job)
+	case model.ActionModifyEngineAttribute:
+		ver, err = onModifyTableEngineAttribute(jobCtx, job)
 	case model.ActionRefreshMeta:
 		ver, err = onRefreshMeta(jobCtx, job)
 	case model.ActionAlterTableAffinity:

pkg/ddl/multi_schema_change.go

@@ -313,6 +313,7 @@ func fillMultiSchemaInfo(info *model.MultiSchemaInfo, job *JobWrapper) error {
 		idxName := job.JobArgs.(*model.AlterIndexVisibilityArgs).IndexName
 		info.AlterIndexes = append(info.AlterIndexes, idxName)
 	case model.ActionRebaseAutoID, model.ActionModifyTableComment, model.ActionModifyTableCharsetAndCollate:
+	case model.ActionModifyEngineAttribute:
 	case model.ActionAddForeignKey:
 		fkInfo := job.JobArgs.(*model.AddForeignKeyArgs).FkInfo
 		info.AddForeignKeys = append(info.AddForeignKeys, model.AddForeignKeyInfo{