ddl,parser: modify schema to store the partial condition

pkg/ddl/create_table.go

@@ -1330,6 +1330,12 @@ func BuildTableInfo(
 			if err != nil {
 				return nil, err
 			}
+			if constr.Option != nil && constr.Option.Condition != nil {
+				// Theoretically, if the index is not a clustered index and also not PKIsHandle, it can be a partial index
+				// because it'll have no difference compared to a normal index. However, for simplicity, this branch blocks
+				// all partial primary key.
+				return nil, dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs("create an primary key with partial index is not supported")
+			}
 			isSingleIntPK := isSingleIntPK(constr, lastCol)
 			if ShouldBuildClusteredIndex(ctx.GetClusteredIndexDefMode(), constr.Option, isSingleIntPK) {
 				if isSingleIntPK {

pkg/ddl/executor.go

@@ -4940,6 +4940,13 @@ func (e *executor) createIndex(ctx sessionctx.Context, ti ast.Ident, keyType ast
 		return errors.Trace(err)
 	}
 
+	var conditionString string
+	if indexOption != nil {
+		conditionString, err = CheckAndBuildIndexConditionString(tblInfo, indexOption.Condition)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
 	args := &model.ModifyIndexArgs{
 		IndexArgs: []*model.IndexArg{{
 			Unique:                  unique,
@@ -4949,6 +4956,7 @@ func (e *executor) createIndex(ctx sessionctx.Context, ti ast.Ident, keyType ast
 			HiddenCols:              hiddenCols,
 			Global:                  global,
 			SplitOpt:                splitOpt,
+			ConditionString:         conditionString,
 		}},
 		OpType: model.OpAddIndex,
 	}

pkg/ddl/index.go

@@ -61,7 +61,9 @@ import (
 	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/charset"
+	"github.com/pingcap/tidb/pkg/parser/format"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/parser/opcode"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
@@ -398,6 +400,12 @@ func BuildIndexInfo(
 			idxInfo.Tp = indexOption.Tp
 		}
 		idxInfo.Global = indexOption.Global
+
+		conditionString, err := CheckAndBuildIndexConditionString(tblInfo, indexOption.Condition)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		idxInfo.ConditionExprString = conditionString
 	} else {
 		// Use btree as default index type.
 		idxInfo.Tp = ast.IndexTypeBtree
@@ -1084,6 +1092,10 @@ func (w *worker) onCreateIndex(jobCtx *jobContext, job *model.Job, isPK bool) (v
 			return ver, errors.Trace(err)
 		}
 		allIndexInfos = append(allIndexInfos, indexInfo)
+		// The condition in the index option is not marshaled, so we need to set it here.
+		if len(arg.ConditionString) > 0 {
+			indexInfo.ConditionExprString = arg.ConditionString
+		}
 	}
 
 	originalState := allIndexInfos[0].State
@@ -3595,3 +3607,192 @@ func renameHiddenColumns(tblInfo *model.TableInfo, from, to ast.CIStr) {
 		}
 	}
 }
+
+// CheckAndBuildIndexConditionString validates whether the given expression is compatible with
+// the table schema and returns a string representation of the expression.
+func CheckAndBuildIndexConditionString(tblInfo *model.TableInfo, indexConditionExpr ast.ExprNode) (string, error) {
+	if indexConditionExpr == nil {
+		return "", nil
+	}
+
+	// check partial index condition expression
+	err := checkIndexCondition(tblInfo, indexConditionExpr)
+	if err != nil {
+		return "", errors.Trace(err)
+	}
+
+	var sb strings.Builder
+	restoreFlags := format.RestoreStringSingleQuotes | format.RestoreKeyWordLowercase | format.RestoreNameBackQuotes |
+		format.RestoreSpacesAroundBinaryOperation | format.RestoreWithoutSchemaName | format.RestoreWithoutTableName
+	restoreCtx := format.NewRestoreCtx(restoreFlags, &sb)
+	sb.Reset()
+	err = indexConditionExpr.Restore(restoreCtx)
+	if err != nil {
+		return "", errors.Trace(err)
+	}
+
+	return sb.String(), nil
+}
+
+func checkIndexCondition(tblInfo *model.TableInfo, indexCondition ast.ExprNode) error {
+	// Only the following expressions are supported:
+	// 1. column IS NULL
+	// 2. column IS NOT NULL
+	// 3. column = / != / > / < / >= / <= const
+	// The column must be a visible column in the table, and the const must be a literal value with
+	// the same type as the column.
+	// The column must **NOT** be a generated column. We can loosen this restriction in the future.
+	//
+	// TODO: support more expressions in the future.
+	if indexCondition == nil {
+		return nil
+	}
+
+	switch cond := indexCondition.(type) {
+	case *ast.IsNullExpr:
+		// `IS NULL` and `IS NOT NULL` are both in this branch.
+		columnName, ok := cond.Expr.(*ast.ColumnNameExpr)
+		if !ok {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				"partial index condition must include a column name in the IS NULL expression")
+		}
+		columnInfo := model.FindColumnInfo(tblInfo.Columns, columnName.Name.Name.L)
+		if columnInfo == nil {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("column name %s referenced in partial index condition is not found in table",
+					columnName.Name.Name.L))
+		}
+		if columnInfo.IsGenerated() {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("generated column %s cannot be used in partial index condition", columnName.Name.Name.L))
+		}
+
+		return nil
+	case *ast.BinaryOperationExpr:
+		if cond.Op != opcode.EQ && cond.Op != opcode.NE && cond.Op != opcode.GT &&
+			cond.Op != opcode.LT && cond.Op != opcode.GE && cond.Op != opcode.LE {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("binary operation %s is not supported", cond.Op.String()))
+		}
+
+		var columnName *ast.ColumnNameExpr
+		var anotherSide ast.ExprNode
+		columnName, ok := cond.L.(*ast.ColumnNameExpr)
+		if !ok {
+			// maybe the right side is a column name
+			columnName, ok = cond.R.(*ast.ColumnNameExpr)
+			if !ok {
+				return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+					"partial index condition must include a column name in the binary operation")
+			}
+
+			anotherSide = cond.L
+		} else {
+			anotherSide = cond.R
+		}
+		columnInfo := model.FindColumnInfo(tblInfo.Columns, columnName.Name.Name.L)
+		if columnInfo == nil {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("column name `%s` referenced in partial index condition is not found in table",
+					columnName.Name.Name.L))
+		}
+		if columnInfo.IsGenerated() {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("generated column %s cannot be used in partial index condition", columnName.Name.Name.L))
+		}
+
+		// The another side must be a literal value, and it must have the same type as the column.
+		constantExpr, ok := anotherSide.(ast.ValueExpr)
+		if !ok {
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				"partial index condition must include a literal value on the other side of the binary operation")
+		}
+		// Reference `types.DefaultTypeForValue`, they are all possible types for literal values.
+		// However, this switch-case still includes more types than the ones we have in that function
+		// to avoid breaking in the future.
+		//
+		// Accept tiny type conversion as the type of the literal value is too limited. We shouldn't
+		// force the user to use such a limited range of types.
+		//
+		// It'll allow precision / length difference in most of the cases.
+		switch constantExpr.GetType().GetType() {
+		case mysql.TypeTiny, mysql.TypeShort, mysql.TypeLong, mysql.TypeLonglong,
+			mysql.TypeInt24, mysql.TypeBit, mysql.TypeYear:
+			// the target column must be an integer type or enum or set
+			if columnInfo.GetType() != mysql.TypeTiny &&
+				columnInfo.GetType() != mysql.TypeShort &&
+				columnInfo.GetType() != mysql.TypeLong &&
+				columnInfo.GetType() != mysql.TypeLonglong &&
+				columnInfo.GetType() != mysql.TypeInt24 &&
+				columnInfo.GetType() != mysql.TypeBit &&
+				columnInfo.GetType() != mysql.TypeYear &&
+				columnInfo.GetType() != mysql.TypeEnum &&
+				columnInfo.GetType() != mysql.TypeSet {
+				return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+					fmt.Sprintf("the type %s of the column `%s` in partial index condition is not compatible with the literal value type %s",
+						columnInfo.FieldType.String(), columnName.Name.Name.L, constantExpr.GetType().String()))
+			}
+			return nil
+		case mysql.TypeFloat, mysql.TypeDouble, mysql.TypeNewDecimal:
+			// the target column must be either a float or double type
+			// TODO: consider whether need to support decimal type in this branch
+			if columnInfo.GetType() != mysql.TypeFloat &&
+				columnInfo.GetType() != mysql.TypeDouble &&
+				columnInfo.GetType() != mysql.TypeNewDecimal {
+				return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+					fmt.Sprintf("the type %s of the column `%s` in partial index condition is not compatible with the literal value type %s",
+						columnInfo.FieldType.String(), columnName.Name.Name.L, constantExpr.GetType().String()))
+			}
+			return nil
+		case mysql.TypeVarchar, mysql.TypeVarString, mysql.TypeString,
+			mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
+			if types.IsString(columnInfo.GetType()) {
+				// check the collation of the column and the literal value
+				if columnInfo.FieldType.GetCharset() != constantExpr.GetType().GetCharset() {
+					return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+						fmt.Sprintf("the charset %s of the column `%s` in partial index condition is not compatible with the literal value charset %s",
+							columnInfo.FieldType.GetCharset(), columnName.Name.Name.L, constantExpr.GetType().GetCharset()))
+				}
+
+				return nil
+			}
+
+			// Allow to compare a datetime type column with a string literal, because we don't have a datetime literal.
+			// This branch will allow users to use datetime columns in index condition.
+			if columnInfo.GetType() == mysql.TypeTimestamp ||
+				columnInfo.GetType() == mysql.TypeDate ||
+				columnInfo.GetType() == mysql.TypeDuration ||
+				columnInfo.GetType() == mysql.TypeNewDate ||
+				columnInfo.GetType() == mysql.TypeDatetime {
+				return nil
+			}
+
+			// ENUM and SET are also allowed for string literal.
+			if columnInfo.GetType() == mysql.TypeEnum || columnInfo.GetType() == mysql.TypeSet {
+				return nil
+			}
+
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("the type %s of the column `%s` in partial index condition is not compatible with the literal value type %s",
+					columnInfo.FieldType.String(), columnName.Name.Name.L, constantExpr.GetType().String()))
+		case mysql.TypeNull:
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				"= NULL is not supported in partial index condition because it is always false")
+		case mysql.TypeTimestamp, mysql.TypeDate, mysql.TypeDuration, mysql.TypeNewDate,
+			mysql.TypeDatetime, mysql.TypeJSON, mysql.TypeEnum, mysql.TypeSet:
+			// The `DATE '2025-07-28'` is actually a `cast` function, so they are also not supported yet.
+			intest.Assert(false, "should never generate literal values of these types")
+
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("the type %s of the literal value in partial index condition is not supported",
+					constantExpr.GetType().String()))
+		default:
+			return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+				fmt.Sprintf("the type %s of the literal value in partial index condition is not supported",
+					constantExpr.GetType().String()))
+		}
+	default:
+		return dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs(
+			"the kind of partial index condition is not supported")
+	}
+}

pkg/ddl/integration_test.go

@@ -15,11 +15,13 @@
 package ddl_test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
+	"github.com/pingcap/tidb/pkg/util/dbterror"
 	"github.com/stretchr/testify/require"
 )
 
@@ -53,3 +55,98 @@ func TestDDLStatementsBackFill(t *testing.T) {
 		require.Equal(t, tc.expectedNeedReorg, needReorg, tc)
 	}
 }
+
+func TestPartialIndex(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test;")
+
+	// test validate column exists in create table
+	tk.MustExec("create table t (a int, b int, key(b) where a = 1);")
+	tk.MustGetDBError("create table t1 (a int, b int, key(b) where c = 1);",
+		dbterror.ErrUnsupportedAddPartialIndex)
+	tk.MustExec("drop table t;")
+
+	// test primary key is not allowed in partial index
+	tk.MustExec("create table t (a int, b int, key(b) where a = 1);")
+	tk.MustGetDBError("create table t2 (a int, b int, primary key(b) where a = 1);",
+		dbterror.ErrUnsupportedAddPartialIndex)
+	tk.MustExec("drop table t;")
+
+	checkColumnTypes := func(columnTypes []string, literals []string, shouldAllowed bool) {
+		for _, columnType := range columnTypes {
+			for _, literal := range literals {
+				tk.MustExec("drop table if exists t;")
+				sql := fmt.Sprintf("create table t (a %s, b int, key(b) where a = %s);", columnType, literal)
+				if shouldAllowed {
+					tk.MustExec(sql)
+					tk.MustExec("drop table t;")
+				} else {
+					tk.MustGetDBError(sql, dbterror.ErrUnsupportedAddPartialIndex)
+				}
+			}
+		}
+	}
+
+	// test create table type validation
+	differentTypeLiterals := [][]string{
+		{"1", "true", "1998"}, // int
+		{"'1'"},               // string with default collate
+		{"1.0"},               // float
+		{"b'101010'", "0x1234567890abcdef", "0b10"}, // binary literal
+		{"null"}, // null
+	}
+	differentColumnTypes := [][]string{
+		{"int", "bigint", "tinyint", "smallint", "year"},
+		{"char(25)", "varchar(123)", "text", "char(25) collate utf8mb4_general_ci", "char(25) collate utf8mb4_bin"},
+		{"float", "double"},
+		{"binary(25) collate binary", "varbinary(123)", "blob", "char(25) collate binary"},
+		{},
+	}
+	for i, columnTypes := range differentColumnTypes {
+		for j, literals := range differentTypeLiterals {
+			checkColumnTypes(columnTypes, literals, i == j)
+		}
+	}
+
+	// test comparing between time column and string constant is allowed.
+	timeColumnTypes := []string{"timestamp", "datetime", "date", "time"}
+	allowedLiterals := []string{"'2025-07-28 12:34:56'", "'2025-07-28'", "'12:34:56'"}
+	notAllowedLiterals := []string{"1", "1.0", "true", "null"}
+	checkColumnTypes(timeColumnTypes, allowedLiterals, true)
+	checkColumnTypes(timeColumnTypes, notAllowedLiterals, false)
+
+	// test comparing between enum/set column and int/string constant is allowed.
+	enumSetColumnTypes := []string{"enum('a', 'b', 'c')", "set('a', 'b', 'c')"}
+	allowedLiterals = []string{"1", "'1'", "'a'"}
+	notAllowedLiterals = []string{"1.0", "null"}
+	checkColumnTypes(enumSetColumnTypes, allowedLiterals, true)
+	checkColumnTypes(enumSetColumnTypes, notAllowedLiterals, false)
+
+	// test validate column exists in alter table
+	tk.MustExec("create table t (a int, b int);")
+	tk.MustExec("alter table t add index idx_b(b) where a = 1;")
+	tk.MustGetDBError("alter table t add index idx_b_2(b) where c = 1;",
+		dbterror.ErrUnsupportedAddPartialIndex)
+	tk.MustExec("drop table t;")
+
+	// test alter table type validation
+	for i, literals := range differentTypeLiterals {
+		for _, literal := range literals {
+			for j, columnTypes := range differentColumnTypes {
+				tk.MustExec("drop table if exists t;")
+				for _, columnType := range columnTypes {
+					sql := fmt.Sprintf("create table t (a %s, b int);", columnType)
+					tk.MustExec(sql)
+					sql = fmt.Sprintf("alter table t add index idx_b(b) where a = %s;", literal)
+					if i == j {
+						tk.MustExec(sql)
+					} else {
+						tk.MustGetDBError(sql, dbterror.ErrUnsupportedAddPartialIndex)
+					}
+					tk.MustExec("drop table t;")
+				}
+			}
+		}
+	}
+}