From e8a537a616ead9a2a1d809b6b622e053fe8b7688 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 12:10:21 +0300
Subject: [PATCH 01/16] SubsequentDiffStrategy: allow/reject multiple changes
 on same entity (applies to same table)

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/errors.go     | 15 +++++++++++++
 go/vt/schemadiff/table.go      |  6 ++++++
 go/vt/schemadiff/table_test.go | 39 ++++++++++++++++++++++++++++++++++
 go/vt/schemadiff/types.go      |  6 ++++++
 4 files changed, 66 insertions(+)

diff --git a/go/vt/schemadiff/errors.go b/go/vt/schemadiff/errors.go
index dac1e6ca31f..02a192a925d 100644
--- a/go/vt/schemadiff/errors.go
+++ b/go/vt/schemadiff/errors.go
@@ -453,3 +453,18 @@ type UnknownColumnCollationCharsetError struct {
 func (e *UnknownColumnCollationCharsetError) Error() string {
 	return fmt.Sprintf("unable to determine charset for column %s with collation %q", sqlescape.EscapeID(e.Column), e.Collation)
 }
+
+type SubsequentDiffRejectedError struct {
+	Table string
+	Diffs []EntityDiff
+}
+
+func (e *SubsequentDiffRejectedError) Error() string {
+	var b strings.Builder
+	b.WriteString(fmt.Sprintf("multiple changes not allowed on table %s. Found:", sqlescape.EscapeID(e.Table)))
+	for _, d := range e.Diffs {
+		b.WriteString("\n")
+		b.WriteString(d.CanonicalStatementString())
+	}
+	return b.String()
+}
diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index a5f04e75ec7..6da003bb8d7 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -946,6 +946,12 @@ func (c *CreateTableEntity) TableDiff(other *CreateTableEntity, hints *DiffHints
 	}
 	sortAlterOptions(parentAlterTableEntityDiff)
 
+	if hints.SubsequentDiffStrategy == SubsequentDiffStrategyReject {
+		if allSubsequent := AllSubsequent(parentAlterTableEntityDiff); len(allSubsequent) > 1 {
+			return nil, &SubsequentDiffRejectedError{Table: c.Name(), Diffs: allSubsequent}
+		}
+	}
+
 	return parentAlterTableEntityDiff, nil
 }
 
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index 0e6ef9c12b2..ea209ce4eea 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -46,6 +46,7 @@ func TestCreateTableDiff(t *testing.T) {
 		charset     int
 		algorithm   int
 		enumreorder int
+		subsequent  int
 		textdiffs   []string
 	}{
 		{
@@ -803,6 +804,13 @@ func TestCreateTableDiff(t *testing.T) {
 				"+	FULLTEXT KEY `name2_ft` (`name2`)",
 			},
 		},
+		{
+			name:       "add two fulltext keys, distinct statements, reject",
+			from:       "create table t1 (id int primary key, name1 tinytext not null, name2 tinytext not null)",
+			to:         "create table t1 (id int primary key, name1 tinytext not null, name2 tinytext not null, fulltext key name1_ft(name1), fulltext key name2_ft(name2))",
+			subsequent: SubsequentDiffStrategyReject,
+			errorMsg:   (&SubsequentDiffRejectedError{Table: "t1"}).Error(),
+		},
 		{
 			name:     "add two fulltext keys, unify statements",
 			from:     "create table t1 (id int primary key, name1 tinytext not null, name2 tinytext not null)",
@@ -815,6 +823,19 @@ func TestCreateTableDiff(t *testing.T) {
 				"+	FULLTEXT KEY `name2_ft` (`name2`)",
 			},
 		},
+		{
+			name:       "add two fulltext keys, unify statements, no reject",
+			from:       "create table t1 (id int primary key, name1 tinytext not null, name2 tinytext not null)",
+			to:         "create table t1 (id int primary key, name1 tinytext not null, name2 tinytext not null, fulltext key name1_ft(name1), fulltext key name2_ft(name2))",
+			fulltext:   FullTextKeyUnifyStatements,
+			subsequent: SubsequentDiffStrategyReject,
+			diff:       "alter table t1 add fulltext key name1_ft (name1), add fulltext key name2_ft (name2)",
+			cdiff:      "ALTER TABLE `t1` ADD FULLTEXT KEY `name1_ft` (`name1`), ADD FULLTEXT KEY `name2_ft` (`name2`)",
+			textdiffs: []string{
+				"+	FULLTEXT KEY `name1_ft` (`name1`)",
+				"+	FULLTEXT KEY `name2_ft` (`name2`)",
+			},
+		},
 		{
 			name: "no fulltext diff",
 			from: "create table t1 (id int primary key, name tinytext not null, fulltext key name_ft(name) with parser ngram)",
@@ -1358,6 +1379,14 @@ func TestCreateTableDiff(t *testing.T) {
 				"+ PARTITION `p4` VALUES LESS THAN (40)",
 			},
 		},
+		{
+			name:       "change partitioning range: statements, multiple, reject",
+			from:       "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:         "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (30), partition p4 values less than (40))",
+			rotation:   RangeRotationDistinctStatements,
+			subsequent: SubsequentDiffStrategyReject,
+			errorMsg:   (&SubsequentDiffRejectedError{Table: "t1"}).Error(),
+		},
 		{
 			name:     "change partitioning range: mixed with nonpartition changes",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
@@ -1371,6 +1400,14 @@ func TestCreateTableDiff(t *testing.T) {
 				"- PARTITION `p2` VALUES LESS THAN (20),",
 			},
 		},
+		{
+			name:       "change partitioning range: mixed with nonpartition changes, reject",
+			from:       "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:         "create table t1 (id int primary key, i int) partition by range (id) (partition p3 values less than (30))",
+			rotation:   RangeRotationDistinctStatements,
+			subsequent: SubsequentDiffStrategyReject,
+			errorMsg:   (&SubsequentDiffRejectedError{Table: "t1"}).Error(),
+		},
 		{
 			name:     "change partitioning range: single partition change, mixed with nonpartition changes",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20))",
@@ -1953,6 +1990,7 @@ func TestCreateTableDiff(t *testing.T) {
 			hints.TableCharsetCollateStrategy = ts.charset
 			hints.AlterTableAlgorithmStrategy = ts.algorithm
 			hints.EnumReorderStrategy = ts.enumreorder
+			hints.SubsequentDiffStrategy = ts.subsequent
 			alter, err := c.Diff(other, &hints)
 
 			require.Equal(t, len(ts.diffs), len(ts.cdiffs))
@@ -2367,6 +2405,7 @@ func TestValidate(t *testing.T) {
 			alter:     "alter table t add column i int",
 			expectErr: &ApplyDuplicatePartitionError{Table: "t1", Partition: "p2"},
 		},
+		// More columns and indexes
 		{
 			name:  "change to visible with alter column",
 			from:  "create table t (id int, i int invisible, primary key (id))",
diff --git a/go/vt/schemadiff/types.go b/go/vt/schemadiff/types.go
index e32882cc6b7..16cc15eb746 100644
--- a/go/vt/schemadiff/types.go
+++ b/go/vt/schemadiff/types.go
@@ -134,6 +134,11 @@ const (
 	ForeignKeyCheckStrategyIgnore
 )
 
+const (
+	SubsequentDiffStrategyAllow int = iota
+	SubsequentDiffStrategyReject
+)
+
 // DiffHints is an assortment of rules for diffing entities
 type DiffHints struct {
 	StrictIndexOrdering         bool
@@ -148,6 +153,7 @@ type DiffHints struct {
 	AlterTableAlgorithmStrategy int
 	EnumReorderStrategy         int
 	ForeignKeyCheckStrategy     int
+	SubsequentDiffStrategy      int
 }
 
 func EmptyDiffHints() *DiffHints {

From 919fcd957a901c504b913fd8492086c8df68dd18 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 13:07:59 +0300
Subject: [PATCH 02/16] DDL strategy: deprecating --fast-range-rotation part 1:
 assuming it is always 'true'

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schema/ddl_strategy.go      | 2 +-
 go/vt/schema/ddl_strategy_test.go | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/go/vt/schema/ddl_strategy.go b/go/vt/schema/ddl_strategy.go
index 71d434b5e09..5eb17acb7a8 100644
--- a/go/vt/schema/ddl_strategy.go
+++ b/go/vt/schema/ddl_strategy.go
@@ -209,7 +209,7 @@ func (setting *DDLStrategySetting) IsPreferInstantDDL() bool {
 
 // IsFastRangeRotationFlag checks if strategy options include --fast-range-rotation
 func (setting *DDLStrategySetting) IsFastRangeRotationFlag() bool {
-	return setting.hasFlag(fastRangeRotationFlag)
+	return true
 }
 
 // isCutOverThresholdFlag returns true when given option denotes a `--cut-over-threshold=[...]` flag
diff --git a/go/vt/schema/ddl_strategy_test.go b/go/vt/schema/ddl_strategy_test.go
index ae6c65815cc..c9c947c0495 100644
--- a/go/vt/schema/ddl_strategy_test.go
+++ b/go/vt/schema/ddl_strategy_test.go
@@ -384,7 +384,7 @@ func TestParseDDLStrategy(t *testing.T) {
 			assert.Equal(t, ts.isPostponeLaunch, setting.IsPostponeLaunch())
 			assert.Equal(t, ts.isAllowConcurrent, setting.IsAllowConcurrent())
 			assert.Equal(t, ts.fastOverRevertible, setting.IsPreferInstantDDL())
-			assert.Equal(t, ts.fastRangeRotation, setting.IsFastRangeRotationFlag())
+			assert.True(t, setting.IsFastRangeRotationFlag())
 			assert.Equal(t, ts.allowForeignKeys, setting.IsAllowForeignKeysFlag())
 			assert.Equal(t, ts.analyzeTable, setting.IsAnalyzeTableFlag())
 			cutOverThreshold, err := setting.CutOverThreshold()

From 5dca6b4dd858ad6e64951fc53fbb238e86b7d9e7 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 13:09:58 +0300
Subject: [PATCH 03/16] removing '--fast-range-rotation' flag from endtoend
 tests

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go b/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
index 9a63ac29960..0efed92f440 100644
--- a/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
+++ b/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
@@ -1142,7 +1142,7 @@ func testRevert(t *testing.T) {
 		checkPartitionedTableCountRows(t, 6)
 	})
 	t.Run("partitions: drop first partition", func(t *testing.T) {
-		uuid := testOnlineDDLStatementForTable(t, "alter table part_test drop partition `p1`", ddlStrategy+" --fast-range-rotation", "vtgate", "")
+		uuid := testOnlineDDLStatementForTable(t, "alter table part_test drop partition `p1`", ddlStrategy, "vtgate", "")
 		uuids = append(uuids, uuid)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
 		checkTable(t, partitionedTableName, true)
@@ -1157,7 +1157,7 @@ func testRevert(t *testing.T) {
 		checkPartitionedTableCountRows(t, 5)
 	})
 	t.Run("partitions: add new partition", func(t *testing.T) {
-		uuid := testOnlineDDLStatementForTable(t, "alter table part_test add partition (PARTITION p7 VALUES LESS THAN (70))", ddlStrategy+" --fast-range-rotation", "vtgate", "")
+		uuid := testOnlineDDLStatementForTable(t, "alter table part_test add partition (PARTITION p7 VALUES LESS THAN (70))", ddlStrategy, "vtgate", "")
 		uuids = append(uuids, uuid)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
 		checkTable(t, partitionedTableName, true)

From 49b31dc36b393924b0a443e5637b9ca048925136 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 13:43:17 +0300
Subject: [PATCH 04/16] Remvoe check of IsFastRangeRotationFlag(), which is now
 assumed to always be true

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/vttablet/onlineddl/analysis.go | 17 +++++++++++++++--
 1 file changed, 15 insertions(+), 2 deletions(-)

diff --git a/go/vt/vttablet/onlineddl/analysis.go b/go/vt/vttablet/onlineddl/analysis.go
index 1dc073bb7d0..31ec6f6de5a 100644
--- a/go/vt/vttablet/onlineddl/analysis.go
+++ b/go/vt/vttablet/onlineddl/analysis.go
@@ -208,8 +208,15 @@ func (e *Executor) analyzeSpecialAlterPlan(ctx context.Context, onlineDDL *schem
 	}
 
 	// special plans which support reverts are trivially desired:
-	// special plans which do not support reverts are flag protected:
-	if onlineDDL.StrategySetting().IsFastRangeRotationFlag() {
+	//
+	// - nothing here thus far
+	//
+	// special plans that do not support revert, but are always desired over Online DDL,
+	// hence not flag protected:
+	{
+		// Dropping a range partition has to run directly. It is incorrect to run with Online DDL
+		// because the table copy will make the second-oldest partition "adopt" the rows which
+		// we really want purged from the oldest partition.
 		op, err := analyzeDropRangePartition(alterTable, createTable)
 		if err != nil {
 			return nil, err
@@ -217,10 +224,16 @@ func (e *Executor) analyzeSpecialAlterPlan(ctx context.Context, onlineDDL *schem
 		if op != nil {
 			return op, nil
 		}
+	}
+	{
+		// Adding a range partition _can_ technically run with Online DDL, but it is wasteful
+		// and pointless. The user fully expects the operation to run immediately and without
+		// any copy of data.
 		if op := analyzeAddRangePartition(alterTable, createTable); op != nil {
 			return op, nil
 		}
 	}
+	// special plans which do not support reverts are flag protected:
 	if onlineDDL.StrategySetting().IsPreferInstantDDL() {
 		op, err := analyzeInstantDDL(alterTable, createTable, capableOf)
 		if err != nil {

From 89a1132f3c0dceba43d5de8b09cf116b821b334b Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 14:13:43 +0300
Subject: [PATCH 05/16] RangeRotationStrategy diff hint supports
 RangeRotationCombinedStatements, to allow multi-table DROP PARTITION
 statements

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/table.go      | 29 ++++++++++++++++++++++++-----
 go/vt/schemadiff/table_test.go | 25 ++++++++++++++++++++++++-
 go/vt/schemadiff/types.go      |  1 +
 3 files changed, 49 insertions(+), 6 deletions(-)

diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index 6da003bb8d7..99930fc95c3 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -1195,6 +1195,7 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 	annotations *TextualAnnotations,
 	t1Partitions *sqlparser.PartitionOption,
 	t2Partitions *sqlparser.PartitionOption,
+	hints *DiffHints,
 ) (bool, []*sqlparser.PartitionSpec, error) {
 	// Validate that both tables have range partitioning
 	if t1Partitions.Type != t2Partitions.Type {
@@ -1240,15 +1241,32 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 		definitions2 = definitions2[1:]
 	}
 	addedPartitions2 := definitions2
-	partitionSpecs := make([]*sqlparser.PartitionSpec, 0, len(droppedPartitions1)+len(addedPartitions2))
-	for _, p := range droppedPartitions1 {
+	var partitionSpecs []*sqlparser.PartitionSpec
+	// Dropped partitions:
+	switch hints.RangeRotationStrategy {
+	case RangeRotationCombinedStatements:
+		// A single DROP PARTITION with multiple partition names
 		partitionSpec := &sqlparser.PartitionSpec{
 			Action: sqlparser.DropAction,
-			Names:  []sqlparser.IdentifierCI{p.Name},
+		}
+		for _, p := range droppedPartitions1 {
+			partitionSpec.Names = append(partitionSpec.Names, p.Name)
 		}
 		partitionSpecs = append(partitionSpecs, partitionSpec)
+	default:
+		// Multiple DROP PARTITION each with a single partition name
+		for _, p := range droppedPartitions1 {
+			partitionSpec := &sqlparser.PartitionSpec{
+				Action: sqlparser.DropAction,
+				Names:  []sqlparser.IdentifierCI{p.Name},
+			}
+			partitionSpecs = append(partitionSpecs, partitionSpec)
+		}
+	}
+	for _, p := range droppedPartitions1 {
 		annotations.MarkRemoved(sqlparser.CanonicalString(p))
 	}
+	// Added partitions:
 	for _, p := range addedPartitions2 {
 		partitionSpec := &sqlparser.PartitionSpec{
 			Action:      sqlparser.AddAction,
@@ -1296,7 +1314,7 @@ func (c *CreateTableEntity) diffPartitions(alterTable *sqlparser.AlterTable,
 		// Having said that, we _do_ analyze the scenario of a RANGE partitioning rotation of partitions:
 		// where zero or more partitions may have been dropped from the earlier range, and zero or more
 		// partitions have been added with a later range:
-		isRotation, partitionSpecs, err := c.isRangePartitionsRotation(annotations, t1Partitions, t2Partitions)
+		isRotation, partitionSpecs, err := c.isRangePartitionsRotation(annotations, t1Partitions, t2Partitions, hints)
 		if err != nil {
 			return nil, err
 		}
@@ -1304,7 +1322,8 @@ func (c *CreateTableEntity) diffPartitions(alterTable *sqlparser.AlterTable,
 			switch hints.RangeRotationStrategy {
 			case RangeRotationIgnore:
 				return nil, nil
-			case RangeRotationDistinctStatements:
+			case RangeRotationCombinedStatements,
+				RangeRotationDistinctStatements:
 				return partitionSpecs, nil
 			case RangeRotationFullSpec:
 				// proceed to return a full rebuild
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index ea209ce4eea..aa17ab22080 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -1332,6 +1332,17 @@ func TestCreateTableDiff(t *testing.T) {
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
 			},
 		},
+		{
+			name:     "change partitioning range: statements, drop, combined",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (30))",
+			rotation: RangeRotationCombinedStatements, // combining just 1 statement is trivially identical to distinct statements output
+			diff:     "alter table t1 drop partition p1",
+			cdiff:    "ALTER TABLE `t1` DROP PARTITION `p1`",
+			textdiffs: []string{
+				"-(PARTITION `p1` VALUES LESS THAN (10),",
+			},
+		},
 		{
 			name:     "change partitioning range: statements, add",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20))",
@@ -1344,7 +1355,19 @@ func TestCreateTableDiff(t *testing.T) {
 			},
 		},
 		{
-			name:     "change partitioning range: statements, multiple drops",
+			name:     "change partitioning range: statements, multiple drops, combined",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p3 values less than (30))",
+			rotation: RangeRotationCombinedStatements,
+			diffs:    []string{"alter table t1 drop partition p1, p2"},
+			cdiffs:   []string{"ALTER TABLE `t1` DROP PARTITION `p1`, `p2`"},
+			textdiffs: []string{
+				"-(PARTITION `p1` VALUES LESS THAN (10),",
+				"- PARTITION `p2` VALUES LESS THAN (20),",
+			},
+		},
+		{
+			name:     "change partitioning range: statements, multiple drops, distinct",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p3 values less than (30))",
 			rotation: RangeRotationDistinctStatements,
diff --git a/go/vt/schemadiff/types.go b/go/vt/schemadiff/types.go
index 16cc15eb746..4c89ea1ed3b 100644
--- a/go/vt/schemadiff/types.go
+++ b/go/vt/schemadiff/types.go
@@ -81,6 +81,7 @@ const (
 
 const (
 	RangeRotationFullSpec = iota
+	RangeRotationCombinedStatements
 	RangeRotationDistinctStatements
 	RangeRotationIgnore
 )

From 195c1820170b4395d4cddaa7c6779e95febd1a6e Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 14:13:57 +0300
Subject: [PATCH 06/16] adding code comments

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schema/ddl_strategy.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/go/vt/schema/ddl_strategy.go b/go/vt/schema/ddl_strategy.go
index 5eb17acb7a8..1121881fa99 100644
--- a/go/vt/schema/ddl_strategy.go
+++ b/go/vt/schema/ddl_strategy.go
@@ -324,7 +324,7 @@ func (setting *DDLStrategySetting) RuntimeOptions() []string {
 		}
 		switch {
 		case isFlag(opt, declarativeFlag):
-		case isFlag(opt, skipTopoFlag):
+		case isFlag(opt, skipTopoFlag): // deprecated flag, parsed for backwards compatibility
 		case isFlag(opt, singletonFlag):
 		case isFlag(opt, singletonContextFlag):
 		case isFlag(opt, allowZeroInDateFlag):
@@ -333,7 +333,7 @@ func (setting *DDLStrategySetting) RuntimeOptions() []string {
 		case isFlag(opt, inOrderCompletionFlag):
 		case isFlag(opt, allowConcurrentFlag):
 		case isFlag(opt, preferInstantDDL):
-		case isFlag(opt, fastRangeRotationFlag):
+		case isFlag(opt, fastRangeRotationFlag): // deprecated flag, parsed for backwards compatibility
 		case isFlag(opt, vreplicationTestSuite):
 		case isFlag(opt, allowForeignKeysFlag):
 		case isFlag(opt, analyzeTableFlag):

From 0f4f1002e56f03a2420d6466827722d831a73bcb Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 14:17:35 +0300
Subject: [PATCH 07/16] validate correct behavior when no partitions are
 dropped

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/table.go      | 36 ++++++++++++++++++----------------
 go/vt/schemadiff/table_test.go | 12 ++++++++++++
 2 files changed, 31 insertions(+), 17 deletions(-)

diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index 99930fc95c3..a1089732376 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -1243,28 +1243,30 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 	addedPartitions2 := definitions2
 	var partitionSpecs []*sqlparser.PartitionSpec
 	// Dropped partitions:
-	switch hints.RangeRotationStrategy {
-	case RangeRotationCombinedStatements:
-		// A single DROP PARTITION with multiple partition names
-		partitionSpec := &sqlparser.PartitionSpec{
-			Action: sqlparser.DropAction,
-		}
-		for _, p := range droppedPartitions1 {
-			partitionSpec.Names = append(partitionSpec.Names, p.Name)
-		}
-		partitionSpecs = append(partitionSpecs, partitionSpec)
-	default:
-		// Multiple DROP PARTITION each with a single partition name
-		for _, p := range droppedPartitions1 {
+	if len(droppedPartitions1) > 0 {
+		switch hints.RangeRotationStrategy {
+		case RangeRotationCombinedStatements:
+			// A single DROP PARTITION with multiple partition names
 			partitionSpec := &sqlparser.PartitionSpec{
 				Action: sqlparser.DropAction,
-				Names:  []sqlparser.IdentifierCI{p.Name},
+			}
+			for _, p := range droppedPartitions1 {
+				partitionSpec.Names = append(partitionSpec.Names, p.Name)
 			}
 			partitionSpecs = append(partitionSpecs, partitionSpec)
+		default:
+			// Multiple DROP PARTITION each with a single partition name
+			for _, p := range droppedPartitions1 {
+				partitionSpec := &sqlparser.PartitionSpec{
+					Action: sqlparser.DropAction,
+					Names:  []sqlparser.IdentifierCI{p.Name},
+				}
+				partitionSpecs = append(partitionSpecs, partitionSpec)
+			}
+		}
+		for _, p := range droppedPartitions1 {
+			annotations.MarkRemoved(sqlparser.CanonicalString(p))
 		}
-	}
-	for _, p := range droppedPartitions1 {
-		annotations.MarkRemoved(sqlparser.CanonicalString(p))
 	}
 	// Added partitions:
 	for _, p := range addedPartitions2 {
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index aa17ab22080..fc052d10b03 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -1390,6 +1390,18 @@ func TestCreateTableDiff(t *testing.T) {
 				"+ PARTITION `p3` VALUES LESS THAN (30)",
 			},
 		},
+		{
+			name:     "change partitioning range: statements, multiple adds, combined",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			rotation: RangeRotationCombinedStatements, // ADD PARTITION syntax only supports a single partition so _combining_ is not different than RangeRotationDistinctStatements
+			diffs:    []string{"alter table t1 add partition (partition p2 values less than (20))", "alter table t1 add partition (partition p3 values less than (30))"},
+			cdiffs:   []string{"ALTER TABLE `t1` ADD PARTITION (PARTITION `p2` VALUES LESS THAN (20))", "ALTER TABLE `t1` ADD PARTITION (PARTITION `p3` VALUES LESS THAN (30))"},
+			textdiffs: []string{
+				"+ PARTITION `p2` VALUES LESS THAN (20),",
+				"+ PARTITION `p3` VALUES LESS THAN (30)",
+			},
+		},
 		{
 			name:     "change partitioning range: statements, multiple, assorted",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",

From 094379064ceb6933a57118c12df9e130d7d14be9 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Mon, 8 Apr 2024 15:26:48 +0300
Subject: [PATCH 08/16] remove RangeRotationCombinedStatements. Make
 RangeRotationDistinctStatements combine DROP PARTITION statements

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/table.go      | 27 +++++---------------
 go/vt/schemadiff/table_test.go | 45 ++++------------------------------
 go/vt/schemadiff/types.go      |  1 -
 3 files changed, 11 insertions(+), 62 deletions(-)

diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index a1089732376..677703d146f 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -1244,29 +1244,15 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 	var partitionSpecs []*sqlparser.PartitionSpec
 	// Dropped partitions:
 	if len(droppedPartitions1) > 0 {
-		switch hints.RangeRotationStrategy {
-		case RangeRotationCombinedStatements:
-			// A single DROP PARTITION with multiple partition names
-			partitionSpec := &sqlparser.PartitionSpec{
-				Action: sqlparser.DropAction,
-			}
-			for _, p := range droppedPartitions1 {
-				partitionSpec.Names = append(partitionSpec.Names, p.Name)
-			}
-			partitionSpecs = append(partitionSpecs, partitionSpec)
-		default:
-			// Multiple DROP PARTITION each with a single partition name
-			for _, p := range droppedPartitions1 {
-				partitionSpec := &sqlparser.PartitionSpec{
-					Action: sqlparser.DropAction,
-					Names:  []sqlparser.IdentifierCI{p.Name},
-				}
-				partitionSpecs = append(partitionSpecs, partitionSpec)
-			}
+		// A single DROP PARTITION clause can specify multiple partition names
+		partitionSpec := &sqlparser.PartitionSpec{
+			Action: sqlparser.DropAction,
 		}
 		for _, p := range droppedPartitions1 {
+			partitionSpec.Names = append(partitionSpec.Names, p.Name)
 			annotations.MarkRemoved(sqlparser.CanonicalString(p))
 		}
+		partitionSpecs = append(partitionSpecs, partitionSpec)
 	}
 	// Added partitions:
 	for _, p := range addedPartitions2 {
@@ -1324,8 +1310,7 @@ func (c *CreateTableEntity) diffPartitions(alterTable *sqlparser.AlterTable,
 			switch hints.RangeRotationStrategy {
 			case RangeRotationIgnore:
 				return nil, nil
-			case RangeRotationCombinedStatements,
-				RangeRotationDistinctStatements:
+			case RangeRotationDistinctStatements:
 				return partitionSpecs, nil
 			case RangeRotationFullSpec:
 				// proceed to return a full rebuild
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index fc052d10b03..939d386296a 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -1332,17 +1332,6 @@ func TestCreateTableDiff(t *testing.T) {
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
 			},
 		},
-		{
-			name:     "change partitioning range: statements, drop, combined",
-			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
-			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (30))",
-			rotation: RangeRotationCombinedStatements, // combining just 1 statement is trivially identical to distinct statements output
-			diff:     "alter table t1 drop partition p1",
-			cdiff:    "ALTER TABLE `t1` DROP PARTITION `p1`",
-			textdiffs: []string{
-				"-(PARTITION `p1` VALUES LESS THAN (10),",
-			},
-		},
 		{
 			name:     "change partitioning range: statements, add",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20))",
@@ -1354,25 +1343,13 @@ func TestCreateTableDiff(t *testing.T) {
 				"+ PARTITION `p3` VALUES LESS THAN (30)",
 			},
 		},
-		{
-			name:     "change partitioning range: statements, multiple drops, combined",
-			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
-			to:       "create table t1 (id int primary key) partition by range (id) (partition p3 values less than (30))",
-			rotation: RangeRotationCombinedStatements,
-			diffs:    []string{"alter table t1 drop partition p1, p2"},
-			cdiffs:   []string{"ALTER TABLE `t1` DROP PARTITION `p1`, `p2`"},
-			textdiffs: []string{
-				"-(PARTITION `p1` VALUES LESS THAN (10),",
-				"- PARTITION `p2` VALUES LESS THAN (20),",
-			},
-		},
 		{
 			name:     "change partitioning range: statements, multiple drops, distinct",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p3 values less than (30))",
 			rotation: RangeRotationDistinctStatements,
-			diffs:    []string{"alter table t1 drop partition p1", "alter table t1 drop partition p2"},
-			cdiffs:   []string{"ALTER TABLE `t1` DROP PARTITION `p1`", "ALTER TABLE `t1` DROP PARTITION `p2`"},
+			diffs:    []string{"alter table t1 drop partition p1, p2"},
+			cdiffs:   []string{"ALTER TABLE `t1` DROP PARTITION `p1`, `p2`"},
 			textdiffs: []string{
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
 				"- PARTITION `p2` VALUES LESS THAN (20),",
@@ -1390,18 +1367,6 @@ func TestCreateTableDiff(t *testing.T) {
 				"+ PARTITION `p3` VALUES LESS THAN (30)",
 			},
 		},
-		{
-			name:     "change partitioning range: statements, multiple adds, combined",
-			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10))",
-			to:       "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
-			rotation: RangeRotationCombinedStatements, // ADD PARTITION syntax only supports a single partition so _combining_ is not different than RangeRotationDistinctStatements
-			diffs:    []string{"alter table t1 add partition (partition p2 values less than (20))", "alter table t1 add partition (partition p3 values less than (30))"},
-			cdiffs:   []string{"ALTER TABLE `t1` ADD PARTITION (PARTITION `p2` VALUES LESS THAN (20))", "ALTER TABLE `t1` ADD PARTITION (PARTITION `p3` VALUES LESS THAN (30))"},
-			textdiffs: []string{
-				"+ PARTITION `p2` VALUES LESS THAN (20),",
-				"+ PARTITION `p3` VALUES LESS THAN (30)",
-			},
-		},
 		{
 			name:     "change partitioning range: statements, multiple, assorted",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
@@ -1427,8 +1392,8 @@ func TestCreateTableDiff(t *testing.T) {
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key, i int) partition by range (id) (partition p3 values less than (30))",
 			rotation: RangeRotationDistinctStatements,
-			diffs:    []string{"alter table t1 add column i int", "alter table t1 drop partition p1", "alter table t1 drop partition p2"},
-			cdiffs:   []string{"ALTER TABLE `t1` ADD COLUMN `i` int", "ALTER TABLE `t1` DROP PARTITION `p1`", "ALTER TABLE `t1` DROP PARTITION `p2`"},
+			diffs:    []string{"alter table t1 add column i int", "alter table t1 drop partition p1, p2"},
+			cdiffs:   []string{"ALTER TABLE `t1` ADD COLUMN `i` int", "ALTER TABLE `t1` DROP PARTITION `p1`, `p2`"},
 			textdiffs: []string{
 				"+	`i` int",
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
@@ -2071,7 +2036,7 @@ func TestCreateTableDiff(t *testing.T) {
 				if len(ts.diffs) > 0 {
 
 					allSubsequentDiffs := AllSubsequent(alter)
-					require.Equal(t, len(ts.diffs), len(allSubsequentDiffs))
+					require.Equal(t, len(ts.diffs), len(allSubsequentDiffs), alter.CanonicalStatementString())
 					require.Equal(t, len(ts.cdiffs), len(allSubsequentDiffs))
 					for i := range ts.diffs {
 						assert.Equal(t, ts.diffs[i], allSubsequentDiffs[i].StatementString())
diff --git a/go/vt/schemadiff/types.go b/go/vt/schemadiff/types.go
index 4c89ea1ed3b..16cc15eb746 100644
--- a/go/vt/schemadiff/types.go
+++ b/go/vt/schemadiff/types.go
@@ -81,7 +81,6 @@ const (
 
 const (
 	RangeRotationFullSpec = iota
-	RangeRotationCombinedStatements
 	RangeRotationDistinctStatements
 	RangeRotationIgnore
 )

From 27207612a1e2dab75e6ddbc3858d52ee0c00d100 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 9 Apr 2024 13:44:11 +0300
Subject: [PATCH 09/16] validate partition-related sybrtax errors

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/sqlparser/parse_test.go | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/go/vt/sqlparser/parse_test.go b/go/vt/sqlparser/parse_test.go
index 44c62be35fe..0636071d2f2 100644
--- a/go/vt/sqlparser/parse_test.go
+++ b/go/vt/sqlparser/parse_test.go
@@ -6078,6 +6078,18 @@ var (
 		input:        "create table 2t.3t2 (c1 bigint not null, c2 text, primary key(c1))",
 		output:       "syntax error at position 18 near '.3'",
 		excludeMulti: true,
+	}, {
+		input:  "ALTER TABLE t ADD PARTITION (PARTITION p10 VALUES LESS THAN (10)), ADD PARTITION (PARTITION p20 VALUES LESS THAN (20))",
+		output: "syntax error at position 67",
+	}, {
+		input:  "ALTER TABLE t DROP PARTITION p1, DROP PARTITION p2",
+		output: "syntax error at position 38 near 'DROP'",
+	}, {
+		input:  "ALTER TABLE t DROP PARTITION p1, ADD COLUMN c INT",
+		output: "syntax error at position 37 near 'ADD'",
+	}, {
+		input:  "ALTER TABLE t ADD COLUMN c INT, DROP PARTITION p1",
+		output: "syntax error at position 47 near 'PARTITION'",
 	}, {
 		input:  "execute stmt1 using a, @b",
 		output: "syntax error at position 22 near 'a'",

From 27afaf7ae1e6bfe0458c813a711ce6965391d2ba Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 9 Apr 2024 14:16:49 +0300
Subject: [PATCH 10/16] support DROP PARTITION from the middle of a RANGE
 PARTITION scheme

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/table.go      | 49 +++++++++++---------
 go/vt/schemadiff/table_test.go | 83 ++++++++++++++++++++++++++++------
 2 files changed, 97 insertions(+), 35 deletions(-)

diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index 677703d146f..f8ecfd6081e 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -19,6 +19,7 @@ package schemadiff
 import (
 	"fmt"
 	"math"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -1195,7 +1196,6 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 	annotations *TextualAnnotations,
 	t1Partitions *sqlparser.PartitionOption,
 	t2Partitions *sqlparser.PartitionOption,
-	hints *DiffHints,
 ) (bool, []*sqlparser.PartitionSpec, error) {
 	// Validate that both tables have range partitioning
 	if t1Partitions.Type != t2Partitions.Type {
@@ -1204,43 +1204,50 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 	if t1Partitions.Type != sqlparser.RangeType {
 		return false, nil, nil
 	}
-	definitions1 := t1Partitions.Definitions
+	definitions1 := slices.Clone(t1Partitions.Definitions)
 	definitions2 := t2Partitions.Definitions
-	// there has to be a non-empty shared list, therefore both definitions must be non-empty:
 	if len(definitions1) == 0 {
 		return false, nil, nil
 	}
 	if len(definitions2) == 0 {
 		return false, nil, nil
 	}
+	definitions2map := make(map[string]*sqlparser.PartitionDefinition, len(definitions2))
+	for _, definition := range definitions2 {
+		definitions2map[sqlparser.CanonicalString(definition)] = definition
+	}
+	// Find dropped partitions:
 	var droppedPartitions1 []*sqlparser.PartitionDefinition
-	// It's OK for prefix of t1 partitions to be nonexistent in t2 (as they may have been rotated away in t2)
-	for len(definitions1) > 0 && !sqlparser.Equals.RefOfPartitionDefinition(definitions1[0], definitions2[0]) {
-		droppedPartitions1 = append(droppedPartitions1, definitions1[0])
-		definitions1 = definitions1[1:]
+	for i := len(definitions1) - 1; i >= 0; i-- {
+		definition := definitions1[i]
+		if _, ok := definitions2map[sqlparser.CanonicalString(definition)]; !ok {
+			// In range partitioning, it's allowed to drop any partition, whether it's the first, somewhere in the middle, or last.
+			droppedPartitions1 = append(droppedPartitions1, definition)
+			// We remove the definition from the list, so that we can then compare the remaining definitions
+			definitions1 = append(definitions1[:i], definitions1[i+1:]...)
+		}
 	}
+	slices.Reverse(droppedPartitions1)
 	if len(definitions1) == 0 {
-		// We've exhausted definition1 trying to find a shared partition with definitions2. Nothing found.
-		// so there is no shared sequence between the two tables.
+		// Nothing shared between the two partition lists.
 		return false, nil, nil
 	}
+	// In range partitioning, it's only allowed to ADD one partition at the end of the range.
+	// We allow multiple here, and the diff mechanism will later split them to subsequent diffs.
+
+	// Let's now validate that any added partitions in t2Partitions are strictly a suffix of t1Partitions
 	if len(definitions1) > len(definitions2) {
 		return false, nil, nil
 	}
-	// To save computation, and because we've already shown that sqlparser.EqualsRefOfPartitionDefinition(definitions1[0], definitions2[0]), nil,
-	// we can skip one element
-	definitions1 = definitions1[1:]
-	definitions2 = definitions2[1:]
-	// Now let's ensure that whatever is remaining in definitions1 is an exact match for a prefix of definitions2
-	// It's ok if we end up with leftover elements in definition2
-	for len(definitions1) > 0 {
-		if !sqlparser.Equals.RefOfPartitionDefinition(definitions1[0], definitions2[0]) {
+	for i := range definitions1 {
+		if !sqlparser.Equals.RefOfPartitionDefinition(definitions1[i], definitions2[i]) {
+			// Not a suffix
 			return false, nil, nil
 		}
-		definitions1 = definitions1[1:]
-		definitions2 = definitions2[1:]
 	}
-	addedPartitions2 := definitions2
+	// And the suffix is any remaining definitions
+	addedPartitions2 := definitions2[len(definitions1):]
+
 	var partitionSpecs []*sqlparser.PartitionSpec
 	// Dropped partitions:
 	if len(droppedPartitions1) > 0 {
@@ -1302,7 +1309,7 @@ func (c *CreateTableEntity) diffPartitions(alterTable *sqlparser.AlterTable,
 		// Having said that, we _do_ analyze the scenario of a RANGE partitioning rotation of partitions:
 		// where zero or more partitions may have been dropped from the earlier range, and zero or more
 		// partitions have been added with a later range:
-		isRotation, partitionSpecs, err := c.isRangePartitionsRotation(annotations, t1Partitions, t2Partitions, hints)
+		isRotation, partitionSpecs, err := c.isRangePartitionsRotation(annotations, t1Partitions, t2Partitions)
 		if err != nil {
 			return nil, err
 		}
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index 939d386296a..2177f1570e9 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -1332,6 +1332,17 @@ func TestCreateTableDiff(t *testing.T) {
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
 			},
 		},
+		{
+			name:     "change partitioning range: statements, drop middle",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p3 values less than (30))",
+			rotation: RangeRotationDistinctStatements,
+			diff:     "alter table t1 drop partition p2",
+			cdiff:    "ALTER TABLE `t1` DROP PARTITION `p2`",
+			textdiffs: []string{
+				"- PARTITION `p2` VALUES LESS THAN (20),",
+			},
+		},
 		{
 			name:     "change partitioning range: statements, add",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20))",
@@ -1438,7 +1449,7 @@ func TestCreateTableDiff(t *testing.T) {
 			},
 		},
 		{
-			name:     "change partitioning range: ignore rotate, not a rotation",
+			name:     "change partitioning range: not a rotation, ignore",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (25), partition p3 values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
@@ -1456,43 +1467,83 @@ func TestCreateTableDiff(t *testing.T) {
 			},
 		},
 		{
-			name:     "change partitioning range: ignore rotate, not a rotation 2",
+			name:     "change partitioning range: not a rotation, ignore 2",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
-			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (35), partition p4 values less than (40))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (25), partition p3 values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (20),\n partition p3 values less than (35),\n partition p4 values less than (40))",
-			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (20),\n PARTITION `p3` VALUES LESS THAN (35),\n PARTITION `p4` VALUES LESS THAN (40))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (25),\n partition p3 values less than (30),\n partition p4 values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (25),\n PARTITION `p3` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
 			textdiffs: []string{
 				"-PARTITION BY RANGE (`id`)",
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
 				"- PARTITION `p2` VALUES LESS THAN (20),",
 				"- PARTITION `p3` VALUES LESS THAN (30))",
 				"+PARTITION BY RANGE (`id`)",
-				"+(PARTITION `p2` VALUES LESS THAN (20)",
+				"+(PARTITION `p2` VALUES LESS THAN (25)",
+				"+ PARTITION `p3` VALUES LESS THAN (30),",
+				"+ PARTITION `p4` VALUES LESS THAN (40))",
+			},
+		},
+		{
+			name:     "change partitioning range: complex rotate, ignore",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (35), partition p4 values less than (40))",
+			rotation: RangeRotationIgnore,
+		},
+		{
+			name:     "change partitioning range: complex rotate, distinct",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (35), partition p4 values less than (40))",
+			rotation: RangeRotationDistinctStatements,
+			diffs:    []string{"alter table t1 drop partition p1, p3", "alter table t1 add partition (partition p3 values less than (35))", "alter table t1 add partition (partition p4 values less than (40))"},
+			cdiffs:   []string{"ALTER TABLE `t1` DROP PARTITION `p1`, `p3`", "ALTER TABLE `t1` ADD PARTITION (PARTITION `p3` VALUES LESS THAN (35))", "ALTER TABLE `t1` ADD PARTITION (PARTITION `p4` VALUES LESS THAN (40))"},
+			textdiffs: []string{
+				"-(PARTITION `p1` VALUES LESS THAN (10),",
+				"- PARTITION `p3` VALUES LESS THAN (30))",
 				"+ PARTITION `p3` VALUES LESS THAN (35),",
 				"+ PARTITION `p4` VALUES LESS THAN (40))",
 			},
 		},
 		{
-			name:     "change partitioning range: ignore rotate, not a rotation 3",
+			name:     "change partitioning range: complex rotate 2, ignore",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition pX values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (20),\n partition pX values less than (30),\n partition p4 values less than (40))",
-			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (20),\n PARTITION `pX` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
+		},
+		{
+			name:     "change partitioning range: complex rotate 2, distinct",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition pX values less than (30), partition p4 values less than (40))",
+			rotation: RangeRotationDistinctStatements,
+			diffs:    []string{"alter table t1 drop partition p1, p3", "alter table t1 add partition (partition pX values less than (30))", "alter table t1 add partition (partition p4 values less than (40))"},
+			cdiffs:   []string{"ALTER TABLE `t1` DROP PARTITION `p1`, `p3`", "ALTER TABLE `t1` ADD PARTITION (PARTITION `pX` VALUES LESS THAN (30))", "ALTER TABLE `t1` ADD PARTITION (PARTITION `p4` VALUES LESS THAN (40))"},
+			textdiffs: []string{
+				"-(PARTITION `p1` VALUES LESS THAN (10),",
+				"- PARTITION `p3` VALUES LESS THAN (30))",
+				"+ PARTITION `pX` VALUES LESS THAN (30),",
+				"+ PARTITION `p4` VALUES LESS THAN (40))",
+			},
+		},
+		{
+			name:     "change partitioning range: not a rotation",
+			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (25), partition p3 values less than (30), partition p4 values less than (40))",
+			rotation: RangeRotationDistinctStatements,
+			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (25),\n partition p3 values less than (30),\n partition p4 values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (25),\n PARTITION `p3` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
 			textdiffs: []string{
 				"-PARTITION BY RANGE (`id`)",
 				"-(PARTITION `p1` VALUES LESS THAN (10),",
 				"- PARTITION `p2` VALUES LESS THAN (20),",
 				"- PARTITION `p3` VALUES LESS THAN (30))",
 				"+PARTITION BY RANGE (`id`)",
-				"+(PARTITION `p2` VALUES LESS THAN (20)",
-				"+ PARTITION `pX` VALUES LESS THAN (30),",
+				"+(PARTITION `p2` VALUES LESS THAN (25)",
+				"+ PARTITION `p3` VALUES LESS THAN (30),",
 				"+ PARTITION `p4` VALUES LESS THAN (40))",
 			},
 		},
 		{
-			name:     "change partitioning range: ignore rotate, not a rotation 4",
+			name:     "change partitioning range: ignore rotate, not a rotation 2",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition pX values less than (20), partition p3 values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
@@ -2036,8 +2087,12 @@ func TestCreateTableDiff(t *testing.T) {
 				if len(ts.diffs) > 0 {
 
 					allSubsequentDiffs := AllSubsequent(alter)
-					require.Equal(t, len(ts.diffs), len(allSubsequentDiffs), alter.CanonicalStatementString())
-					require.Equal(t, len(ts.cdiffs), len(allSubsequentDiffs))
+					allSubsequentDiffsStatements := []string{}
+					for _, d := range allSubsequentDiffs {
+						allSubsequentDiffsStatements = append(allSubsequentDiffsStatements, d.CanonicalStatementString())
+					}
+					require.Equal(t, len(ts.diffs), len(allSubsequentDiffs), allSubsequentDiffsStatements)
+					require.Equal(t, len(ts.cdiffs), len(allSubsequentDiffs), allSubsequentDiffsStatements)
 					for i := range ts.diffs {
 						assert.Equal(t, ts.diffs[i], allSubsequentDiffs[i].StatementString())
 						assert.Equal(t, ts.cdiffs[i], allSubsequentDiffs[i].CanonicalStatementString())

From 2b0cebfc0b321c7cf22a8a396a9993087f9b2335 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 9 Apr 2024 14:17:27 +0300
Subject: [PATCH 11/16] schemadiff: analyze range partition rotation

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/errors.go | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/go/vt/schemadiff/errors.go b/go/vt/schemadiff/errors.go
index 02a192a925d..a941c406be0 100644
--- a/go/vt/schemadiff/errors.go
+++ b/go/vt/schemadiff/errors.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"vitess.io/vitess/go/sqlescape"
+	"vitess.io/vitess/go/vt/sqlparser"
 )
 
 var (
@@ -468,3 +469,21 @@ func (e *SubsequentDiffRejectedError) Error() string {
 	}
 	return b.String()
 }
+
+// PartitionSpecNonExclusiveError is returned when a partition spec change is found alongside other changes.
+// for example, in MySQL it is invalid to both DROP PARTITION (a partition spec change) and ADD COLUMN
+// in the same ALTER TABLE statement. In fact, even two partition spec changes in the same ALTER TABLE
+// statement are not allowed.
+// This error should never be encountered in normal circumstances, because:
+// - `sqlparser` should not allow such statements to be parsed.
+// - schemadiff's `Diff()` function will never generate a single `ALTER TABLE` statement with such multiple changes.
+// The error is used for integrity checks only, and should be considered a bug if encountered.
+type PartitionSpecNonExclusiveError struct {
+	Table                string
+	PartitionSpec        *sqlparser.PartitionSpec
+	ConflictingStatement string
+}
+
+func (e *PartitionSpecNonExclusiveError) Error() string {
+	return fmt.Sprintf("ALTER TABLE on %s, may only have a single partition spec change, and other changes are not allowed. Found spec: %s; and change: %s", sqlescape.EscapeID(e.Table), sqlparser.CanonicalString(e.PartitionSpec), e.ConflictingStatement)
+}

From ac0d6011dbb35240b24528dbf198fb99aedad0cb Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 9 Apr 2024 14:23:02 +0300
Subject: [PATCH 12/16] schemadiff: AlterTableRotatesRangePartition

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/analysis.go      | 58 ++++++++++++++++++++++++
 go/vt/schemadiff/analysis_test.go | 74 +++++++++++++++++++++++++++++++
 2 files changed, 132 insertions(+)
 create mode 100644 go/vt/schemadiff/analysis.go
 create mode 100644 go/vt/schemadiff/analysis_test.go

diff --git a/go/vt/schemadiff/analysis.go b/go/vt/schemadiff/analysis.go
new file mode 100644
index 00000000000..311e2bbda96
--- /dev/null
+++ b/go/vt/schemadiff/analysis.go
@@ -0,0 +1,58 @@
+/*
+Copyright 2024 The Vitess Authors.
+
+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 schemadiff
+
+import "vitess.io/vitess/go/vt/sqlparser"
+
+// AlterTableRotatesRangePartition answers `true` when the given ALTER TABLE statemnts performas any sort
+// of range partition rotation, that is applicable immediately and without moving data.
+// Such would be:
+// - Dropping any partition(s)
+// - Adding a new partition (empty, at the end of the list)
+func AlterTableRotatesRangePartition(alterTable *sqlparser.AlterTable) (bool, error) {
+	spec := alterTable.PartitionSpec
+	if spec == nil {
+		return false, nil
+	}
+	errorResult := func(conflictingNode sqlparser.SQLNode) error {
+		return &PartitionSpecNonExclusiveError{
+			Table:                alterTable.Table.Name.String(),
+			PartitionSpec:        spec,
+			ConflictingStatement: sqlparser.CanonicalString(conflictingNode),
+		}
+	}
+	if len(alterTable.AlterOptions) > 0 {
+		// This should never happen, unless someone programmatically tampered with the AlterTable AST.
+		return false, errorResult(alterTable.AlterOptions[0])
+	}
+	if alterTable.PartitionOption != nil {
+		// This should never happen, unless someone programmatically tampered with the AlterTable AST.
+		return false, errorResult(alterTable.PartitionOption)
+	}
+	switch spec.Action {
+	case sqlparser.AddAction:
+		if len(spec.Definitions) > 1 {
+			// This should never happen, unless someone programmatically tampered with the AlterTable AST.
+			return false, errorResult(spec.Definitions[1])
+		}
+		return true, nil
+	case sqlparser.DropAction:
+		return true, nil
+	default:
+		return false, nil
+	}
+}
diff --git a/go/vt/schemadiff/analysis_test.go b/go/vt/schemadiff/analysis_test.go
new file mode 100644
index 00000000000..514babfee81
--- /dev/null
+++ b/go/vt/schemadiff/analysis_test.go
@@ -0,0 +1,74 @@
+/*
+Copyright 2024 The Vitess Authors.
+
+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 schemadiff
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/vt/sqlparser"
+)
+
+// AnalyzePartitionRotation analyzes a given AlterTable statement to see whether it has partition rotation
+// commands, and if so, is the ALTER TABLE statement valid in MySQL. In MySQL, a single ALTER TABLE statement
+// cannot apply multiple rotation commands, nor can it mix rotation commands with other types of changes.
+func TestAlterTableRotatesRangePartition(t *testing.T) {
+	tcases := []struct {
+		alter  string
+		expect bool
+	}{
+		{
+			alter:  "ALTER TABLE t ADD PARTITION (PARTITION p1 VALUES LESS THAN (10))",
+			expect: true,
+		},
+		{
+			alter:  "ALTER TABLE t DROP PARTITION p1",
+			expect: true,
+		},
+		{
+			alter:  "ALTER TABLE t DROP PARTITION p1, p2",
+			expect: true,
+		},
+		{
+			alter: "ALTER TABLE t TRUNCATE PARTITION p3",
+		},
+		{
+			alter: "ALTER TABLE t COALESCE PARTITION 3",
+		},
+		{
+			alter: "ALTER TABLE t partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
+		},
+		{
+			alter: "ALTER TABLE t ADD COLUMN c1 INT, DROP COLUMN c2",
+		},
+	}
+
+	for _, tcase := range tcases {
+		t.Run(tcase.alter, func(t *testing.T) {
+			stmt, err := sqlparser.NewTestParser().ParseStrictDDL(tcase.alter)
+			require.NoError(t, err)
+			alterTable, ok := stmt.(*sqlparser.AlterTable)
+			require.True(t, ok)
+
+			result, err := AlterTableRotatesRangePartition(alterTable)
+			require.NoError(t, err)
+			assert.Equal(t, tcase.expect, result)
+		})
+	}
+}

From 21cf7ef20c47d415a7c2f1d83ed75d99241c26d1 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 9 Apr 2024 14:36:35 +0300
Subject: [PATCH 13/16] Online DDL uses schemadiff's
 'AlterTableRotatesRangePartition()'. Also, it does not mark any particular
 details about a partition rotation special operation, as we do not expect
 such an operation to be revertible

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/vttablet/onlineddl/analysis.go | 110 ++-------------------------
 go/vt/vttablet/onlineddl/executor.go |  37 +--------
 2 files changed, 9 insertions(+), 138 deletions(-)

diff --git a/go/vt/vttablet/onlineddl/analysis.go b/go/vt/vttablet/onlineddl/analysis.go
index 31ec6f6de5a..28c48fd5113 100644
--- a/go/vt/vttablet/onlineddl/analysis.go
+++ b/go/vt/vttablet/onlineddl/analysis.go
@@ -31,9 +31,8 @@ import (
 type specialAlterOperation string
 
 const (
-	instantDDLSpecialOperation         specialAlterOperation = "instant-ddl"
-	dropRangePartitionSpecialOperation specialAlterOperation = "drop-range-partition"
-	addRangePartitionSpecialOperation  specialAlterOperation = "add-range-partition"
+	instantDDLSpecialOperation     specialAlterOperation = "instant-ddl"
+	rangePartitionSpecialOperation specialAlterOperation = "range-partition"
 )
 
 type SpecialAlterPlan struct {
@@ -86,95 +85,6 @@ func (e *Executor) getCreateTableStatement(ctx context.Context, tableName string
 	return createTable, nil
 }
 
-// analyzeDropRangePartition sees if the online DDL drops a single partition in a range partitioned table
-func analyzeDropRangePartition(alterTable *sqlparser.AlterTable, createTable *sqlparser.CreateTable) (*SpecialAlterPlan, error) {
-	// we are looking for a `ALTER TABLE <table> DROP PARTITION <name>` statement with nothing else
-	if len(alterTable.AlterOptions) > 0 {
-		return nil, nil
-	}
-	if alterTable.PartitionOption != nil {
-		return nil, nil
-	}
-	spec := alterTable.PartitionSpec
-	if spec == nil {
-		return nil, nil
-	}
-	if spec.Action != sqlparser.DropAction {
-		return nil, nil
-	}
-	if len(spec.Names) != 1 {
-		return nil, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "vitess only supports dropping a single partition per query: %v", sqlparser.CanonicalString(alterTable))
-	}
-	partitionName := spec.Names[0].String()
-	// OK then!
-
-	// Now, is this query dropping the first partition in a RANGE partitioned table?
-	part := createTable.TableSpec.PartitionOption
-	if part.Type != sqlparser.RangeType {
-		return nil, nil
-	}
-	if len(part.Definitions) == 0 {
-		return nil, nil
-	}
-	var partitionDefinition *sqlparser.PartitionDefinition
-	var nextPartitionName string
-	for i, p := range part.Definitions {
-		if p.Name.String() == partitionName {
-			partitionDefinition = p
-			if i+1 < len(part.Definitions) {
-				nextPartitionName = part.Definitions[i+1].Name.String()
-			}
-			break
-		}
-	}
-	if partitionDefinition == nil {
-		// dropping a nonexistent partition. We'll let the "standard" migration execution flow deal with that.
-		return nil, nil
-	}
-	op := NewSpecialAlterOperation(dropRangePartitionSpecialOperation, alterTable, createTable)
-	op.SetDetail("partition_name", partitionName)
-	op.SetDetail("partition_definition", sqlparser.CanonicalString(partitionDefinition))
-	op.SetDetail("next_partition_name", nextPartitionName)
-	return op, nil
-}
-
-// analyzeAddRangePartition sees if the online DDL adds a partition in a range partitioned table
-func analyzeAddRangePartition(alterTable *sqlparser.AlterTable, createTable *sqlparser.CreateTable) *SpecialAlterPlan {
-	// we are looking for a `ALTER TABLE <table> ADD PARTITION (PARTITION ...)` statement with nothing else
-	if len(alterTable.AlterOptions) > 0 {
-		return nil
-	}
-	if alterTable.PartitionOption != nil {
-		return nil
-	}
-	spec := alterTable.PartitionSpec
-	if spec == nil {
-		return nil
-	}
-	if spec.Action != sqlparser.AddAction {
-		return nil
-	}
-	if len(spec.Definitions) != 1 {
-		return nil
-	}
-	partitionDefinition := spec.Definitions[0]
-	partitionName := partitionDefinition.Name.String()
-	// OK then!
-
-	// Now, is this query adding a partition in a RANGE partitioned table?
-	part := createTable.TableSpec.PartitionOption
-	if part.Type != sqlparser.RangeType {
-		return nil
-	}
-	if len(part.Definitions) == 0 {
-		return nil
-	}
-	op := NewSpecialAlterOperation(addRangePartitionSpecialOperation, alterTable, createTable)
-	op.SetDetail("partition_name", partitionName)
-	op.SetDetail("partition_definition", sqlparser.CanonicalString(partitionDefinition))
-	return op
-}
-
 // analyzeInstantDDL takes declarative CreateTable and AlterTable, as well as a server version, and checks whether it is possible to run the ALTER
 // using ALGORITHM=INSTANT for that version.
 func analyzeInstantDDL(alterTable *sqlparser.AlterTable, createTable *sqlparser.CreateTable, capableOf capabilities.CapableOf) (*SpecialAlterPlan, error) {
@@ -217,19 +127,15 @@ func (e *Executor) analyzeSpecialAlterPlan(ctx context.Context, onlineDDL *schem
 		// Dropping a range partition has to run directly. It is incorrect to run with Online DDL
 		// because the table copy will make the second-oldest partition "adopt" the rows which
 		// we really want purged from the oldest partition.
-		op, err := analyzeDropRangePartition(alterTable, createTable)
-		if err != nil {
-			return nil, err
-		}
-		if op != nil {
-			return op, nil
-		}
-	}
-	{
 		// Adding a range partition _can_ technically run with Online DDL, but it is wasteful
 		// and pointless. The user fully expects the operation to run immediately and without
 		// any copy of data.
-		if op := analyzeAddRangePartition(alterTable, createTable); op != nil {
+		isRangeRotation, err := schemadiff.AlterTableRotatesRangePartition(alterTable)
+		if err != nil {
+			return nil, err
+		}
+		if isRangeRotation {
+			op := NewSpecialAlterOperation(rangePartitionSpecialOperation, alterTable, createTable)
 			return op, nil
 		}
 	}
diff --git a/go/vt/vttablet/onlineddl/executor.go b/go/vt/vttablet/onlineddl/executor.go
index 48d09db0f42..57033f43b30 100644
--- a/go/vt/vttablet/onlineddl/executor.go
+++ b/go/vt/vttablet/onlineddl/executor.go
@@ -3177,42 +3177,7 @@ func (e *Executor) executeSpecialAlterDDLActionMigrationIfApplicable(ctx context
 		if _, err := e.executeDirectly(ctx, onlineDDL); err != nil {
 			return false, err
 		}
-	case dropRangePartitionSpecialOperation:
-		dropPartition := func() error {
-			artifactTableName, err := schema.GenerateGCTableName(schema.HoldTableGCState, newGCTableRetainTime())
-			if err != nil {
-				return err
-			}
-			if err := e.updateArtifacts(ctx, onlineDDL.UUID, artifactTableName); err != nil {
-				return err
-			}
-
-			// Apply CREATE TABLE for artifact table
-			if _, _, err := e.createDuplicateTableLike(ctx, artifactTableName, onlineDDL, conn); err != nil {
-				return err
-			}
-			// Remove partitioning
-			parsed := sqlparser.BuildParsedQuery(sqlAlterTableRemovePartitioning, artifactTableName)
-			if _, err := conn.ExecuteFetch(parsed.Query, 0, false); err != nil {
-				return err
-			}
-			// Exchange with partition
-			partitionName := specialPlan.Detail("partition_name")
-			parsed = sqlparser.BuildParsedQuery(sqlAlterTableExchangePartition, onlineDDL.Table, partitionName, artifactTableName)
-			if _, err := conn.ExecuteFetch(parsed.Query, 0, false); err != nil {
-				return err
-			}
-			// Drop table's partition
-			parsed = sqlparser.BuildParsedQuery(sqlAlterTableDropPartition, onlineDDL.Table, partitionName)
-			if _, err := conn.ExecuteFetch(parsed.Query, 0, false); err != nil {
-				return err
-			}
-			return nil
-		}
-		if err := dropPartition(); err != nil {
-			return false, err
-		}
-	case addRangePartitionSpecialOperation:
+	case rangePartitionSpecialOperation:
 		if _, err := e.executeDirectly(ctx, onlineDDL); err != nil {
 			return false, err
 		}

From b888e2b3b652b95286ffe4f6d6f5272f895039ed Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Thu, 11 Apr 2024 15:57:28 +0300
Subject: [PATCH 14/16] remove IsFastRangeRotationFlag() function

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schema/ddl_strategy.go      | 5 -----
 go/vt/schema/ddl_strategy_test.go | 1 -
 2 files changed, 6 deletions(-)

diff --git a/go/vt/schema/ddl_strategy.go b/go/vt/schema/ddl_strategy.go
index 1121881fa99..e3b03c3f330 100644
--- a/go/vt/schema/ddl_strategy.go
+++ b/go/vt/schema/ddl_strategy.go
@@ -207,11 +207,6 @@ func (setting *DDLStrategySetting) IsPreferInstantDDL() bool {
 	return setting.hasFlag(preferInstantDDL)
 }
 
-// IsFastRangeRotationFlag checks if strategy options include --fast-range-rotation
-func (setting *DDLStrategySetting) IsFastRangeRotationFlag() bool {
-	return true
-}
-
 // isCutOverThresholdFlag returns true when given option denotes a `--cut-over-threshold=[...]` flag
 func isCutOverThresholdFlag(opt string) (string, bool) {
 	submatch := cutOverThresholdFlagRegexp.FindStringSubmatch(opt)
diff --git a/go/vt/schema/ddl_strategy_test.go b/go/vt/schema/ddl_strategy_test.go
index c9c947c0495..f27f0963e80 100644
--- a/go/vt/schema/ddl_strategy_test.go
+++ b/go/vt/schema/ddl_strategy_test.go
@@ -384,7 +384,6 @@ func TestParseDDLStrategy(t *testing.T) {
 			assert.Equal(t, ts.isPostponeLaunch, setting.IsPostponeLaunch())
 			assert.Equal(t, ts.isAllowConcurrent, setting.IsAllowConcurrent())
 			assert.Equal(t, ts.fastOverRevertible, setting.IsPreferInstantDDL())
-			assert.True(t, setting.IsFastRangeRotationFlag())
 			assert.Equal(t, ts.allowForeignKeys, setting.IsAllowForeignKeysFlag())
 			assert.Equal(t, ts.analyzeTable, setting.IsAnalyzeTableFlag())
 			cutOverThreshold, err := setting.CutOverThreshold()

From 53862179257b297d85f8ba60294816fba66e1a76 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 30 Apr 2024 17:11:59 +0300
Subject: [PATCH 15/16] AlterTableRotatesRangePartition takes CREATE TABLE
 statement into account and validates the table has RANGE partitioning

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/analysis.go         | 14 ++++++++++++--
 go/vt/schemadiff/analysis_test.go    | 13 +++++++++++--
 go/vt/vttablet/onlineddl/analysis.go |  2 +-
 3 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/go/vt/schemadiff/analysis.go b/go/vt/schemadiff/analysis.go
index 311e2bbda96..2fb9d825b65 100644
--- a/go/vt/schemadiff/analysis.go
+++ b/go/vt/schemadiff/analysis.go
@@ -16,14 +16,24 @@ limitations under the License.
 
 package schemadiff
 
-import "vitess.io/vitess/go/vt/sqlparser"
+import (
+	"vitess.io/vitess/go/vt/sqlparser"
+)
 
 // AlterTableRotatesRangePartition answers `true` when the given ALTER TABLE statemnts performas any sort
 // of range partition rotation, that is applicable immediately and without moving data.
 // Such would be:
 // - Dropping any partition(s)
 // - Adding a new partition (empty, at the end of the list)
-func AlterTableRotatesRangePartition(alterTable *sqlparser.AlterTable) (bool, error) {
+func AlterTableRotatesRangePartition(createTable *sqlparser.CreateTable, alterTable *sqlparser.AlterTable) (bool, error) {
+	// Validate original table is partitioned by RANGE
+	if createTable.TableSpec.PartitionOption == nil {
+		return false, nil
+	}
+	if createTable.TableSpec.PartitionOption.Type != sqlparser.RangeType {
+		return false, nil
+	}
+
 	spec := alterTable.PartitionSpec
 	if spec == nil {
 		return false, nil
diff --git a/go/vt/schemadiff/analysis_test.go b/go/vt/schemadiff/analysis_test.go
index 514babfee81..b0092fb7aac 100644
--- a/go/vt/schemadiff/analysis_test.go
+++ b/go/vt/schemadiff/analysis_test.go
@@ -30,6 +30,7 @@ import (
 // cannot apply multiple rotation commands, nor can it mix rotation commands with other types of changes.
 func TestAlterTableRotatesRangePartition(t *testing.T) {
 	tcases := []struct {
+		create string
 		alter  string
 		expect bool
 	}{
@@ -61,12 +62,20 @@ func TestAlterTableRotatesRangePartition(t *testing.T) {
 
 	for _, tcase := range tcases {
 		t.Run(tcase.alter, func(t *testing.T) {
-			stmt, err := sqlparser.NewTestParser().ParseStrictDDL(tcase.alter)
+			if tcase.create == "" {
+				tcase.create = "CREATE TABLE t (id int PRIMARY KEY) PARTITION BY RANGE (id) (PARTITION p0 VALUES LESS THAN (10))"
+			}
+			stmt, err := sqlparser.NewTestParser().ParseStrictDDL(tcase.create)
+			require.NoError(t, err)
+			createTable, ok := stmt.(*sqlparser.CreateTable)
+			require.True(t, ok)
+
+			stmt, err = sqlparser.NewTestParser().ParseStrictDDL(tcase.alter)
 			require.NoError(t, err)
 			alterTable, ok := stmt.(*sqlparser.AlterTable)
 			require.True(t, ok)
 
-			result, err := AlterTableRotatesRangePartition(alterTable)
+			result, err := AlterTableRotatesRangePartition(createTable, alterTable)
 			require.NoError(t, err)
 			assert.Equal(t, tcase.expect, result)
 		})
diff --git a/go/vt/vttablet/onlineddl/analysis.go b/go/vt/vttablet/onlineddl/analysis.go
index 28c48fd5113..970104877f2 100644
--- a/go/vt/vttablet/onlineddl/analysis.go
+++ b/go/vt/vttablet/onlineddl/analysis.go
@@ -130,7 +130,7 @@ func (e *Executor) analyzeSpecialAlterPlan(ctx context.Context, onlineDDL *schem
 		// Adding a range partition _can_ technically run with Online DDL, but it is wasteful
 		// and pointless. The user fully expects the operation to run immediately and without
 		// any copy of data.
-		isRangeRotation, err := schemadiff.AlterTableRotatesRangePartition(alterTable)
+		isRangeRotation, err := schemadiff.AlterTableRotatesRangePartition(createTable, alterTable)
 		if err != nil {
 			return nil, err
 		}

From c0b7a45cc8d0aff325291f632fd958ddfed22b36 Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Tue, 30 Apr 2024 17:18:52 +0300
Subject: [PATCH 16/16] typo

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 go/vt/schemadiff/analysis.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/go/vt/schemadiff/analysis.go b/go/vt/schemadiff/analysis.go
index 2fb9d825b65..ae0f22559f2 100644
--- a/go/vt/schemadiff/analysis.go
+++ b/go/vt/schemadiff/analysis.go
@@ -20,7 +20,7 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 )
 
-// AlterTableRotatesRangePartition answers `true` when the given ALTER TABLE statemnts performas any sort
+// AlterTableRotatesRangePartition answers `true` when the given ALTER TABLE statement performs any sort
 // of range partition rotation, that is applicable immediately and without moving data.
 // Such would be:
 // - Dropping any partition(s)