fix: allow repeated alter table in explicit transactions

pkg/frontend/databranchutils/branch_dag_test.go

@@ -272,6 +272,71 @@ func TestComputeAlterLineageCompactionPlan(t *testing.T) {
 	)
 }
 
+func TestComponentHasLiveLogicalBranch(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		start uint64
+		rows  []DataBranchMetadata
+		want  bool
+	}{
+		{
+			name:  "historical ALTER generations only",
+			start: 3,
+			rows: []DataBranchMetadata{
+				{TableID: 2, PTableID: 1, Level: AlterLineageLevel},
+				{TableID: 3, PTableID: 2, Level: AlterLineageLevel},
+			},
+		},
+		{
+			name:  "live logical branch",
+			start: 2,
+			rows: []DataBranchMetadata{
+				{TableID: 2, PTableID: 1, Level: "table"},
+			},
+			want: true,
+		},
+		{
+			name:  "logical ownership inherited through ALTER",
+			start: 3,
+			rows: []DataBranchMetadata{
+				{TableID: 2, PTableID: 1, Level: "table", TableDeleted: true},
+				{TableID: 3, PTableID: 2, Level: "alter:table"},
+			},
+			want: true,
+		},
+		{
+			name:  "live logical sibling across ancestor",
+			start: 3,
+			rows: []DataBranchMetadata{
+				{TableID: 2, PTableID: 1, Level: "table"},
+				{TableID: 3, PTableID: 1, Level: AlterLineageLevel},
+			},
+			want: true,
+		},
+		{
+			name:  "deleted logical sibling",
+			start: 3,
+			rows: []DataBranchMetadata{
+				{TableID: 2, PTableID: 1, Level: "table", TableDeleted: true},
+				{TableID: 3, PTableID: 1, Level: AlterLineageLevel},
+			},
+		},
+		{
+			name:  "historical ALTER cycle",
+			start: 1,
+			rows: []DataBranchMetadata{
+				{TableID: 1, PTableID: 2, Level: AlterLineageLevel},
+				{TableID: 2, PTableID: 1, Level: AlterLineageLevel},
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			dag := NewBranchReclaimDag(tc.rows)
+			require.Equal(t, tc.want, dag.ComponentHasLiveLogicalBranch(tc.start))
+		})
+	}
+}
+
 func TestComputeAlterLineageCompactionPlanReclaimsDeletedAlterGenerations(t *testing.T) {
 	rows := []DataBranchMetadata{
 		{TableID: 2, PTableID: 1, CloneTS: 100, Level: "alter", TableDeleted: true},

pkg/frontend/databranchutils/branch_protect_snapshot.go

@@ -232,32 +232,7 @@ func ComputeAlterLineageCompactionPlan(
 		if _, ok := visited[start]; ok {
 			continue
 		}
-		component := make([]uint64, 0, 4)
-		stack := []uint64{start}
-		logicalOwner := false
-		for len(stack) > 0 {
-			last := len(stack) - 1
-			tableID := stack[last]
-			stack = stack[:last]
-			if tableID == 0 {
-				continue
-			}
-			if _, ok := visited[tableID]; ok {
-				continue
-			}
-			visited[tableID] = struct{}{}
-			component = append(component, tableID)
-
-			if meta, ok := dag.Info[tableID]; ok {
-				if !meta.Deleted && isLogicalBranchOwnerLevel(meta.Level) {
-					logicalOwner = true
-				}
-				if meta.ParentTableID != 0 {
-					stack = append(stack, meta.ParentTableID)
-				}
-			}
-			stack = append(stack, dag.Children[tableID]...)
-		}
+		component, logicalOwner := dag.connectedComponent(start, visited)
 
 		if logicalOwner {
 			continue
@@ -315,6 +290,48 @@ func ComputeAlterLineageCompactionPlan(
 	return plan
 }
 
+func (d BranchReclaimDag) connectedComponent(
+	start uint64,
+	visited map[uint64]struct{},
+) (component []uint64, hasLiveLogicalBranch bool) {
+	stack := []uint64{start}
+	for len(stack) > 0 {
+		last := len(stack) - 1
+		tableID := stack[last]
+		stack = stack[:last]
+		if tableID == 0 {
+			continue
+		}
+		if _, ok := visited[tableID]; ok {
+			continue
+		}
+		visited[tableID] = struct{}{}
+		component = append(component, tableID)
+
+		if meta, ok := d.Info[tableID]; ok {
+			if !meta.Deleted && isLogicalBranchOwnerLevel(meta.Level) {
+				hasLiveLogicalBranch = true
+			}
+			if meta.ParentTableID != 0 {
+				stack = append(stack, meta.ParentTableID)
+			}
+		}
+		stack = append(stack, d.Children[tableID]...)
+	}
+	return component, hasLiveLogicalBranch
+}
+
+// ComponentHasLiveLogicalBranch reports whether the lineage component that
+// contains start has a live logical branch owner. Plain "alter" rows only pin
+// historical physical generations and do not represent logical branches.
+func (d BranchReclaimDag) ComponentHasLiveLogicalBranch(start uint64) bool {
+	_, hasLiveLogicalBranch := d.connectedComponent(
+		start,
+		make(map[uint64]struct{}, len(d.Info)),
+	)
+	return hasLiveLogicalBranch
+}
+
 // BuildAlterLineageSnapshotDeleteSQL deletes only branch-managed snapshots.
 func BuildAlterLineageSnapshotDeleteSQL(snames []string) string {
 	return BuildBranchSnapshotDeleteSQL(snames)

pkg/sql/compile/alter.go

@@ -95,6 +95,14 @@ func alterCopySQLAtLineageSnapshot(sql string, plan alterDataBranchLineagePlan)
 	return sql + fmt.Sprintf(" {MO_TS = %d}", plan.cloneTS)
 }
 
+func isExplicitAlterTxn(byBegin, autocommit bool) bool {
+	return byBegin || !autocommit
+}
+
+func shouldUseFixedAlterCopySnapshot(snapshotAdvanced, txnHasWorkspaceHistory bool) bool {
+	return snapshotAdvanced && !txnHasWorkspaceHistory
+}
+
 func alterDataBranchParticipationSQL(oldTableID uint64) string {
 	return fmt.Sprintf(
 		"select 1 from %s.%s where table_id = %d or p_table_id = %d limit 1",
@@ -281,12 +289,22 @@ func (c *Compile) prepareAlterDataBranchLineage(
 	}
 	hasLiveLineage := false
 	if participates {
-		op := c.proc.GetTxnOperator()
-		opts := op.TxnOptions()
-		if err = validateAlterDataBranchLineageTxn(
-			opts.GetByBegin(), opts.GetAutocommit(), op.Txn().IsPessimistic(),
-		); err != nil {
-			return alterDataBranchLineagePlan{}, err
+		// ALTER-only rows preserve physical history for a snapshot or PITR but
+		// are not logical data branches. Inspect the complete connected
+		// component so an ALTER generation neither triggers a false transaction
+		// restriction nor hides a live logical sibling behind an ancestor.
+		ownershipDAG, dagErr := c.loadAlterDataBranchDAG(false)
+		if dagErr != nil {
+			return alterDataBranchLineagePlan{}, dagErr
+		}
+		if ownershipDAG.ComponentHasLiveLogicalBranch(oldTableID) {
+			op := c.proc.GetTxnOperator()
+			opts := op.TxnOptions()
+			if err = validateAlterDataBranchLineageTxn(
+				opts.GetByBegin(), opts.GetAutocommit(), op.Txn().IsPessimistic(),
+			); err != nil {
+				return alterDataBranchLineagePlan{}, err
+			}
 		}
 		if err = c.compactExpiredAlterDataBranchLineage(time.Time{}); err != nil {
 			return alterDataBranchLineagePlan{}, err
@@ -316,7 +334,7 @@ func (c *Compile) prepareAlterDataBranchLineage(
 }
 
 func validateAlterDataBranchLineageTxn(byBegin, autocommit, _ bool) error {
-	if byBegin || !autocommit {
+	if isExplicitAlterTxn(byBegin, autocommit) {
 		return moerr.NewNotSupportedNoCtx(
 			"ALTER on a data-branch lineage is not supported inside an explicit transaction",
 		)
@@ -1165,7 +1183,19 @@ func (s *Scope) AlterTableCopy(c *Compile) (err error) {
 	if lineagePlan.enabled {
 		if lineageSnapshotAdvanced {
 			lineagePlan.cloneTS = lineageCloneTS
-			lineagePlan.fixedCopyTS = true
+			// A snapshot hint cannot see this transaction's workspace: it would
+			// lose earlier DML and cannot resolve a generation created by earlier
+			// DDL. The current operator already has the lock-held advanced snapshot
+			// and overlays that workspace, so explicit transactions copy from it.
+			lineageTxnOpts := lineageTxnOp.TxnOptions()
+			txnHasWorkspaceHistory := isExplicitAlterTxn(
+				lineageTxnOpts.GetByBegin(),
+				lineageTxnOpts.GetAutocommit(),
+			) || c.getHaveDDL()
+			lineagePlan.fixedCopyTS = shouldUseFixedAlterCopySnapshot(
+				lineageSnapshotAdvanced,
+				txnHasWorkspaceHistory,
+			)
 		} else {
 			// Optimistic mode has no row-lock snapshot barrier. Its statement
 			// snapshot is nevertheless the exact source view copied by ALTER, so

pkg/sql/compile/alter_test.go

@@ -47,6 +47,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/pb/lock"
 	plan2 "github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
+	"github.com/matrixorigin/matrixone/pkg/pb/txn"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
@@ -61,6 +62,31 @@ func TestShouldEnableAlterCopyPipelineFlush(t *testing.T) {
 	assert.True(t, shouldEnableAlterCopyPipelineFlush(&plan2.AlterCopyOpt{SkipPkDedup: true}))
 }
 
+func TestShouldUseFixedAlterCopySnapshot(t *testing.T) {
+	require.True(t, isExplicitAlterTxn(true, true))
+	require.True(t, isExplicitAlterTxn(false, false))
+	require.False(t, isExplicitAlterTxn(false, true))
+
+	require.True(t, shouldUseFixedAlterCopySnapshot(true, false))
+	require.False(t, shouldUseFixedAlterCopySnapshot(true, true))
+	require.False(t, shouldUseFixedAlterCopySnapshot(false, false))
+	require.False(t, shouldUseFixedAlterCopySnapshot(false, true))
+}
+
+func TestAlterCopySQLAtLineageSnapshot(t *testing.T) {
+	const sql = "insert into copy select * from source"
+	require.Equal(t, sql, alterCopySQLAtLineageSnapshot(sql, alterDataBranchLineagePlan{}))
+	require.Equal(t, sql, alterCopySQLAtLineageSnapshot(sql, alterDataBranchLineagePlan{
+		enabled: true,
+		cloneTS: 123,
+	}))
+	require.Equal(t, sql+" {MO_TS = 123}", alterCopySQLAtLineageSnapshot(sql, alterDataBranchLineagePlan{
+		enabled:     true,
+		fixedCopyTS: true,
+		cloneTS:     123,
+	}))
+}
+
 func TestAlterCopySameStatementColumnReplacement(t *testing.T) {
 	tableDef := &plan2.TableDef{Cols: []*plan2.ColDef{
 		{Name: "a", ColId: 1, Seqnum: 0},
@@ -285,6 +311,70 @@ func TestPrepareAlterDataBranchLineageAllowsHistoricalSourceTxn(t *testing.T) {
 	}
 }
 
+func TestPrepareAlterDataBranchLineageAllowsHistoricalOnlyGenerationInExplicitTxn(t *testing.T) {
+	const (
+		oldTableID    = uint64(42)
+		parentTableID = uint64(41)
+		database      = "test"
+		table         = "dept"
+		cloneTS       = int64(100)
+	)
+	ctrl := gomock.NewController(t)
+	spyExec := &alterCopyInsertSpyExecutor{
+		results:         make(map[string]executor.Result),
+		resultSequences: make(map[string][]executor.Result),
+	}
+	c := newAlterCopyPrecheckCompile(t, ctrl, spyExec)
+	txnOp := mock_frontend.NewMockTxnOperator(ctrl)
+	txnOp.EXPECT().TxnOptions().Return(txn.TxnOptions{ByBegin: true, Autocommit: true}).AnyTimes()
+	txnOp.EXPECT().Txn().Return(txn.TxnMeta{}).AnyTimes()
+	txnOp.EXPECT().SnapshotTS().Return(timestamp.Timestamp{PhysicalTime: cloneTS + 1}).AnyTimes()
+	c.proc.Base.TxnOperator = txnOp
+
+	participationSQL := alterDataBranchParticipationSQL(oldTableID)
+	metadataSQL := "select table_id, p_table_id, clone_ts, creator, level, table_deleted from mo_catalog.mo_branch_metadata"
+	lockedMetadataSQL := metadataSQL + " for update"
+	edgeSQL := alterDataBranchLineageEdgeSQL()
+	snapshotSourceSQL := alterDataBranchSnapshotSourceSQL()
+	pitrSourceSQL := alterDataBranchPitrSourceSQL()
+	spyExec.results[participationSQL] = newAlterCopyFixedResult(
+		t, c.proc.Mp(), types.T_int32.ToType(), []int32{1},
+	)
+	newMetadataResult := func() executor.Result {
+		return newAlterLineageMetadataResult(
+			t, c.proc.Mp(), []uint64{oldTableID}, []uint64{parentTableID}, []int64{cloneTS},
+			[]uint64{uint64(catalog.System_Account)}, []string{databranchutils.AlterLineageLevel}, []bool{false},
+		)
+	}
+	spyExec.resultSequences[metadataSQL] = []executor.Result{newMetadataResult(), newMetadataResult()}
+	spyExec.results[lockedMetadataSQL] = newMetadataResult()
+	spyExec.results[edgeSQL] = newAlterLineageEdgeResult(
+		t, c.proc.Mp(), []string{databranchutils.BranchSnapshotName(oldTableID)}, []int64{cloneTS},
+		[]string{""}, []string{database}, []string{table}, []uint64{parentTableID},
+	)
+	spyExec.results[snapshotSourceSQL] = newAlterLineageSnapshotSourceResult(
+		t, c.proc.Mp(), []int64{cloneTS - 1}, []string{"table"}, []string{""},
+		[]string{database}, []string{table}, []uint64{parentTableID},
+	)
+	spyExec.results[pitrSourceSQL] = newAlterLineagePitrSourceResult(
+		t, c.proc.Mp(), nil, nil, nil, nil, nil, nil, nil,
+	)
+
+	lineagePlan, err := c.prepareAlterDataBranchLineage(oldTableID, database, table)
+	require.NoError(t, err)
+	require.True(t, lineagePlan.enabled)
+	require.False(t, lineagePlan.preserveHistoricalSource)
+	require.Equal(t, []string{
+		participationSQL,
+		metadataSQL,
+		lockedMetadataSQL,
+		edgeSQL,
+		snapshotSourceSQL,
+		pitrSourceSQL,
+		metadataSQL,
+	}, spyExec.executedSQLs)
+}
+
 func TestShouldAdvanceAlterDataBranchLineageSnapshot(t *testing.T) {
 	require.True(t, shouldAdvanceAlterDataBranchLineageSnapshot(true, true))
 	require.False(t, shouldAdvanceAlterDataBranchLineageSnapshot(true, false))
@@ -467,13 +557,14 @@ func TestAlterCopyAutoIncrementCleanupDiscardsTrackedReset(t *testing.T) {
 }
 
 type alterCopyInsertSpyExecutor struct {
-	insertSQL    string
-	insertErr    error
-	insertCtx    context.Context
-	insertOption executor.StatementOption
-	results      map[string]executor.Result
-	errs         map[string]error
-	executedSQLs []string
+	insertSQL       string
+	insertErr       error
+	insertCtx       context.Context
+	insertOption    executor.StatementOption
+	results         map[string]executor.Result
+	resultSequences map[string][]executor.Result
+	errs            map[string]error
+	executedSQLs    []string
 }
 
 func TestReconcileAlterCopyAutoIncrementUsesStableIdentityAndSafeBounds(t *testing.T) {
@@ -643,6 +734,10 @@ func (e *alterCopyInsertSpyExecutor) Exec(
 			return executor.Result{}, err
 		}
 	}
+	if results := e.resultSequences[sql]; len(results) > 0 {
+		e.resultSequences[sql] = results[1:]
+		return results[0], nil
+	}
 	if e.results != nil {
 		if res, ok := e.results[sql]; ok {
 			return res, nil