diff --git a/pkg/incrservice/types.go b/pkg/incrservice/types.go
index 12883ec12bf45..a5bf763908ee0 100644
--- a/pkg/incrservice/types.go
+++ b/pkg/incrservice/types.go
@@ -16,8 +16,11 @@ package incrservice
 
 import (
 	"context"
+	"math"
 
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
@@ -165,11 +168,62 @@ type AutoColumn struct {
 	Step     uint64
 }
 
-// GetAutoColumnFromDef get auto columns from table def
+// ValidateAutoColumnOffset rejects allocator offsets that cannot be represented
+// by the destination AUTO_INCREMENT column type.
+func ValidateAutoColumnOffset(ctx context.Context, typ types.T, offset uint64) error {
+	var limit uint64
+	switch typ {
+	case types.T_uint8:
+		limit = math.MaxUint8
+	case types.T_uint16:
+		limit = math.MaxUint16
+	case types.T_uint32:
+		limit = math.MaxUint32
+	case types.T_uint64:
+		return nil
+	case types.T_int8:
+		limit = math.MaxInt8
+	case types.T_int16:
+		limit = math.MaxInt16
+	case types.T_int32:
+		limit = math.MaxInt32
+	case types.T_int64:
+		limit = math.MaxInt64
+	default:
+		return nil
+	}
+	if offset <= limit {
+		return nil
+	}
+	return moerr.NewOutOfRangef(
+		ctx,
+		typ.ToType().String(),
+		"AUTO_INCREMENT value %d",
+		offset,
+	)
+}
+
+// GetAutoColumnFromDef gets all allocator-owned columns from a table definition,
+// including internal hidden columns such as __mo_fake_pk_col.
 func GetAutoColumnFromDef(def *plan.TableDef) []AutoColumn {
+	return getAutoColumnsFromDef(def, func(*plan.ColDef) bool { return true })
+}
+
+// GetUserAutoColumnFromDef gets only SQL-visible AUTO_INCREMENT columns.
+func GetUserAutoColumnFromDef(def *plan.TableDef) []AutoColumn {
+	return getAutoColumnsFromDef(def, func(col *plan.ColDef) bool { return !col.Hidden })
+}
+
+// GetInternalAutoColumnFromDef gets allocator-owned hidden columns. User offset
+// requests must not change these columns.
+func GetInternalAutoColumnFromDef(def *plan.TableDef) []AutoColumn {
+	return getAutoColumnsFromDef(def, func(col *plan.ColDef) bool { return col.Hidden })
+}
+
+func getAutoColumnsFromDef(def *plan.TableDef, include func(*plan.ColDef) bool) []AutoColumn {
 	var cols []AutoColumn
 	for i, col := range def.Cols {
-		if col.Typ.AutoIncr {
+		if col.Typ.AutoIncr && include(col) {
 			cols = append(cols, AutoColumn{
 				ColName:  col.Name,
 				TableID:  def.TblId,
diff --git a/pkg/incrservice/types_test.go b/pkg/incrservice/types_test.go
new file mode 100644
index 0000000000000..87dca9cf5c516
--- /dev/null
+++ b/pkg/incrservice/types_test.go
@@ -0,0 +1,51 @@
+// Copyright 2026 Matrix Origin
+//
+// 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 incrservice
+
+import (
+	"context"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+)
+
+func TestAutoColumnVisibilityAndOffsetValidation(t *testing.T) {
+	def := &plan.TableDef{TblId: 7, Cols: []*plan.ColDef{
+		{Name: "id", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+		{Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+		{Name: "__mo_fake_pk_col", Hidden: true, Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+	}}
+
+	require.Equal(t, []string{"id", "__mo_fake_pk_col"}, autoColumnNames(GetAutoColumnFromDef(def)))
+	require.Equal(t, []string{"id"}, autoColumnNames(GetUserAutoColumnFromDef(def)))
+	require.Equal(t, []string{"__mo_fake_pk_col"}, autoColumnNames(GetInternalAutoColumnFromDef(def)))
+
+	require.NoError(t, ValidateAutoColumnOffset(context.Background(), types.T_uint8, math.MaxUint8))
+	err := ValidateAutoColumnOffset(context.Background(), types.T_uint8, math.MaxUint8+1)
+	require.True(t, moerr.IsMoErrCode(err, moerr.ErrOutOfRange), err)
+}
+
+func autoColumnNames(cols []AutoColumn) []string {
+	names := make([]string, len(cols))
+	for i := range cols {
+		names[i] = cols[i].ColName
+	}
+	return names
+}
diff --git a/pkg/sql/colexec/table_clone/table_clone.go b/pkg/sql/colexec/table_clone/table_clone.go
index e5ada6c710ac6..b6d4c7fa8b8c9 100644
--- a/pkg/sql/colexec/table_clone/table_clone.go
+++ b/pkg/sql/colexec/table_clone/table_clone.go
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
@@ -30,7 +31,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/pb/partition"
-	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	"github.com/matrixorigin/matrixone/pkg/sql/features"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
@@ -101,9 +101,15 @@ func (tc *TableClone) Reset(proc *process.Process, pipelineFailed bool, err erro
 	tc.dstRel = nil
 	tc.dstIdxRel = nil
 
+	if tc.Ctx.SrcAutoIncrMaxValues != nil {
+		clear(tc.Ctx.SrcAutoIncrMaxValues)
+	}
 	if tc.Ctx.SrcAutoIncrOffsets != nil {
 		clear(tc.Ctx.SrcAutoIncrOffsets)
 	}
+	if tc.Ctx.IndexAutoIncrStates != nil {
+		clear(tc.Ctx.IndexAutoIncrStates)
+	}
 }
 
 func (tc *TableClone) String(buf *bytes.Buffer) {
@@ -468,85 +474,92 @@ func (tc *TableClone) updateDstAutoIncrColumns(
 	proc *process.Process,
 ) error {
 
-	if tc.Ctx.SrcAutoIncrOffsets == nil {
+	if tc.Ctx.SrcAutoIncrMaxValues == nil &&
+		tc.Ctx.SrcAutoIncrOffsets == nil &&
+		len(tc.Ctx.IndexAutoIncrStates) == 0 {
 		return nil
 	}
 
-	var (
-		err      error
-		typs     []types.Type
-		incrCols []incrservice.AutoColumn
-
-		maxVal int64
-
-		dstTblDef *plan.TableDef
-	)
-
-	dstTblDef = tc.dstMasterRel.GetTableDef(dstCtx)
-	_, typs, _, _, _ = colexec.GetSequmsAttrsSortKeyIdxFromTableDef(dstTblDef)
-	incrCols = incrservice.GetAutoColumnFromDef(dstTblDef)
-
-	vecs := make([]*vector.Vector, len(typs))
-	for i, typ := range typs {
-		vecs[i] = vector.NewVec(typ)
+	if tc.Ctx.SrcAutoIncrMaxValues != nil || tc.Ctx.SrcAutoIncrOffsets != nil {
+		if err := updateRelationAutoIncrement(
+			dstCtx,
+			proc,
+			tc.dstMasterRel,
+			AutoIncrementState{
+				MaxValues: tc.Ctx.SrcAutoIncrMaxValues,
+				Offsets:   tc.Ctx.SrcAutoIncrOffsets,
+			},
+			tc.Ctx.RequestedAutoIncrOffset,
+			true,
+		); err != nil {
+			return err
+		}
 	}
 
-	defer func() {
-		for i := range vecs {
-			vecs[i].Free(proc.Mp())
+	for key, rel := range tc.dstIdxRel {
+		state, ok := tc.Ctx.IndexAutoIncrStates[strings.ToLower(key)]
+		if !ok {
+			continue
 		}
-	}()
-
-	rows := 0
-	for _, col := range incrCols {
-		maxVal = int64(tc.Ctx.SrcAutoIncrOffsets[int32(col.ColIndex)])
-
-		var val any
-		switch typs[col.ColIndex].Oid {
-		case types.T_uint8:
-			val = uint8(maxVal)
-		case types.T_uint16:
-			val = uint16(maxVal)
-		case types.T_uint32:
-			val = uint32(maxVal)
-		case types.T_uint64:
-			val = uint64(maxVal)
-		case types.T_int8:
-			val = int8(maxVal)
-		case types.T_int16:
-			val = int16(maxVal)
-		case types.T_int32:
-			val = int32(maxVal)
-		case types.T_int64:
-			val = int64(maxVal)
+		if err := updateRelationAutoIncrement(dstCtx, proc, rel, state, 0, false); err != nil {
+			return err
 		}
+	}
 
-		if err = vector.AppendAny(
-			vecs[col.ColIndex], val, false, proc.Mp(),
-		); err != nil {
+	return nil
+}
+
+func updateRelationAutoIncrement(
+	ctx context.Context,
+	proc *process.Process,
+	rel engine.Relation,
+	state AutoIncrementState,
+	requestedOffset uint64,
+	separateUserColumns bool,
+) error {
+	def := rel.GetTableDef(ctx)
+	var typs []types.Type
+	_, typs, _, _, _ = colexec.GetSequmsAttrsSortKeyIdxFromTableDef(def)
+	tableID := rel.GetTableID(ctx)
+	setOffset := func(col incrservice.AutoColumn, offset uint64) error {
+		if err := incrservice.ValidateAutoColumnOffset(ctx, typs[col.ColIndex].Oid, offset); err != nil {
 			return err
 		}
-		if rows == 0 {
-			rows = vecs[col.ColIndex].Length()
-		}
+		return proc.GetIncrService().SetOffset(
+			ctx,
+			tableID,
+			col.ColName,
+			offset,
+			proc.GetTxnOperator(),
+		)
 	}
 
-	if rows == 0 {
+	if !separateUserColumns {
+		for _, col := range incrservice.GetAutoColumnFromDef(def) {
+			name := strings.ToLower(col.ColName)
+			if err := setOffset(col, max(state.MaxValues[name], state.Offsets[name])); err != nil {
+				return err
+			}
+		}
 		return nil
 	}
 
-	if _, err = proc.GetIncrService().InsertValues(
-		dstCtx,
-		tc.dstMasterRel.GetTableID(dstCtx),
-		dstTblDef.AutoIncrEpoch,
-		proc.GetTxnOperator(),
-		vecs,
-		rows,
-		int64(rows),
-	); err != nil {
-		return err
+	for _, col := range incrservice.GetUserAutoColumnFromDef(def) {
+		name := strings.ToLower(col.ColName)
+		if err := setOffset(col, max(requestedOffset, state.MaxValues[name], state.Offsets[name])); err != nil {
+			return err
+		}
+	}
+	for _, col := range incrservice.GetInternalAutoColumnFromDef(def) {
+		name := strings.ToLower(col.ColName)
+		offset, ok := state.Offsets[name]
+		if !ok {
+			continue
+		}
+		if err := setOffset(col, offset); err != nil {
+			return err
+		}
 	}
-
 	return nil
 }
 
diff --git a/pkg/sql/colexec/table_clone/table_clone_test.go b/pkg/sql/colexec/table_clone/table_clone_test.go
index 914561537ff61..28e98c166ad1e 100644
--- a/pkg/sql/colexec/table_clone/table_clone_test.go
+++ b/pkg/sql/colexec/table_clone/table_clone_test.go
@@ -16,10 +16,21 @@ package table_clone
 
 import (
 	"bytes"
+	"context"
+	"math"
 	"testing"
 
-	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/txn/client"
+	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 )
 
 func TestTableCloneOperatorMetadata(t *testing.T) {
@@ -33,3 +44,204 @@ func TestTableCloneOperatorMetadata(t *testing.T) {
 	require.NotEqual(t, vm.Top, tc.OpType())
 	require.Equal(t, "TableClone", tc.OpType().String())
 }
+
+type autoIncrementTestRelation struct {
+	engine.Relation
+	tableID uint64
+	name    string
+	def     *plan.TableDef
+}
+
+func (r *autoIncrementTestRelation) GetTableID(context.Context) uint64 {
+	return r.tableID
+}
+
+func (r *autoIncrementTestRelation) GetTableDef(context.Context) *plan.TableDef {
+	return r.def
+}
+
+func (r *autoIncrementTestRelation) GetTableName() string {
+	return r.name
+}
+
+func TestUpdateDstAutoIncrColumnsReconcilesAllSafeBounds(t *testing.T) {
+	tests := []struct {
+		name      string
+		requested uint64
+		copiedMax uint64
+		srcOffset uint64
+		want      uint64
+	}{
+		{name: "requested offset wins", requested: 99, copiedMax: 40, srcOffset: 50, want: 99},
+		{name: "copied maximum wins", requested: 99, copiedMax: 200, srcOffset: 50, want: 200},
+		{name: "source allocator wins", requested: 99, copiedMax: 40, srcOffset: 300, want: 300},
+		{name: "empty source", want: 0},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			proc := testutil.NewProcess(t)
+			ctrl := gomock.NewController(t)
+			incrSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+			proc.Base.IncrService = incrSvc
+
+			def := &plan.TableDef{
+				TblId: 42,
+				Cols: []*plan.ColDef{{
+					Name: "id",
+					Typ:  plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+				}},
+				Pkey: &plan.PrimaryKeyDef{PkeyColName: "id"},
+			}
+			tc := &TableClone{
+				Ctx: &TableCloneCtx{
+					RequestedAutoIncrOffset: tt.requested,
+					SrcAutoIncrMaxValues:    map[string]uint64{"id": tt.copiedMax},
+					SrcAutoIncrOffsets:      map[string]uint64{"id": tt.srcOffset},
+				},
+				dstMasterRel: &autoIncrementTestRelation{tableID: def.TblId, def: def},
+			}
+
+			incrSvc.EXPECT().SetOffset(
+				gomock.Any(), def.TblId, "id", tt.want, gomock.Any(),
+			)
+			require.NoError(t, tc.updateDstAutoIncrColumns(proc.Ctx, proc))
+		})
+	}
+}
+
+func TestUpdateDstAutoIncrColumnsKeepsHiddenAllocatorIndependent(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	ctrl := gomock.NewController(t)
+	incrSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+	proc.Base.IncrService = incrSvc
+
+	def := &plan.TableDef{
+		TblId: 42,
+		Cols: []*plan.ColDef{
+			{Name: "id", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+			{Name: "__mo_fake_pk_col", Hidden: true, Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+		},
+		Pkey: &plan.PrimaryKeyDef{PkeyColName: "id"},
+	}
+	tc := &TableClone{
+		Ctx: &TableCloneCtx{
+			RequestedAutoIncrOffset: 999,
+			SrcAutoIncrMaxValues:    map[string]uint64{"id": 40},
+			SrcAutoIncrOffsets: map[string]uint64{
+				"id":               50,
+				"__mo_fake_pk_col": 40,
+			},
+		},
+		dstMasterRel: &autoIncrementTestRelation{tableID: def.TblId, def: def},
+	}
+
+	gomock.InOrder(
+		incrSvc.EXPECT().SetOffset(gomock.Any(), def.TblId, "id", uint64(999), gomock.Any()),
+		incrSvc.EXPECT().SetOffset(gomock.Any(), def.TblId, "__mo_fake_pk_col", uint64(40), gomock.Any()),
+	)
+	require.NoError(t, tc.updateDstAutoIncrColumns(proc.Ctx, proc))
+}
+
+func TestUpdateDstAutoIncrColumnsReconcilesClonedIndexAllocator(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	ctrl := gomock.NewController(t)
+	incrSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+	proc.Base.IncrService = incrSvc
+
+	const indexTableName = "__mo_index_fulltext_target"
+	def := &plan.TableDef{
+		TblId: 84,
+		Name:  indexTableName,
+		Cols: []*plan.ColDef{{
+			Name:   "__mo_fake_pk_col",
+			Hidden: true,
+			Typ:    plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+		}},
+		Pkey: &plan.PrimaryKeyDef{PkeyColName: "__mo_fake_pk_col"},
+	}
+	tc := &TableClone{
+		Ctx: &TableCloneCtx{
+			IndexAutoIncrStates: map[string]AutoIncrementState{
+				"ftidx.": {
+					MaxValues: map[string]uint64{"__mo_fake_pk_col": 120},
+					Offsets:   map[string]uint64{"__mo_fake_pk_col": 200},
+				},
+				"p0.ftidx.": {
+					MaxValues: map[string]uint64{"__mo_fake_pk_col": 300},
+					Offsets:   map[string]uint64{"__mo_fake_pk_col": 250},
+				},
+				"p1.ftidx.": {
+					MaxValues: map[string]uint64{"__mo_fake_pk_col": 350},
+					Offsets:   map[string]uint64{"__mo_fake_pk_col": 400},
+				},
+			},
+		},
+		dstIdxRel: map[string]engine.Relation{
+			"ftidx.":    &autoIncrementTestRelation{tableID: 84, name: def.Name, def: def},
+			"p0.ftidx.": &autoIncrementTestRelation{tableID: 85, name: def.Name, def: def},
+			"p1.ftidx.": &autoIncrementTestRelation{tableID: 86, name: def.Name, def: def},
+		},
+	}
+
+	incrSvc.EXPECT().SetOffset(gomock.Any(), uint64(84), "__mo_fake_pk_col", uint64(200), gomock.Any())
+	incrSvc.EXPECT().SetOffset(gomock.Any(), uint64(85), "__mo_fake_pk_col", uint64(300), gomock.Any())
+	incrSvc.EXPECT().SetOffset(gomock.Any(), uint64(86), "__mo_fake_pk_col", uint64(400), gomock.Any())
+	require.NoError(t, tc.updateDstAutoIncrColumns(proc.Ctx, proc))
+}
+
+func TestUpdateDstAutoIncrColumnsRejectsOutOfRangeOffset(t *testing.T) {
+	tests := []struct {
+		name    string
+		oid     types.T
+		value   uint64
+		wantErr bool
+	}{
+		{name: "uint8 maximum", oid: types.T_uint8, value: math.MaxUint8},
+		{name: "uint8 overflow", oid: types.T_uint8, value: math.MaxUint8 + 1, wantErr: true},
+		{name: "int8 maximum", oid: types.T_int8, value: math.MaxInt8},
+		{name: "int8 overflow", oid: types.T_int8, value: math.MaxInt8 + 1, wantErr: true},
+		{name: "uint64 maximum", oid: types.T_uint64, value: math.MaxUint64},
+		{name: "int64 maximum", oid: types.T_int64, value: math.MaxInt64},
+		{name: "int64 overflow", oid: types.T_int64, value: uint64(math.MaxInt64) + 1, wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			proc := testutil.NewProcess(t)
+			ctrl := gomock.NewController(t)
+			incrSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+			proc.Base.IncrService = incrSvc
+			def := &plan.TableDef{
+				TblId: 42,
+				Cols: []*plan.ColDef{{
+					Name: "id",
+					Typ:  plan.Type{Id: int32(tt.oid), AutoIncr: true},
+				}},
+				Pkey: &plan.PrimaryKeyDef{PkeyColName: "id"},
+			}
+			tc := &TableClone{
+				Ctx: &TableCloneCtx{
+					RequestedAutoIncrOffset: tt.value,
+					SrcAutoIncrMaxValues:    map[string]uint64{"id": 0},
+				},
+				dstMasterRel: &autoIncrementTestRelation{tableID: def.TblId, def: def},
+			}
+
+			if !tt.wantErr {
+				incrSvc.EXPECT().SetOffset(
+					gomock.Any(), def.TblId, "id", tt.value, gomock.Any(),
+				).DoAndReturn(func(context.Context, uint64, string, uint64, client.TxnOperator) error {
+					return nil
+				})
+			}
+
+			err := tc.updateDstAutoIncrColumns(proc.Ctx, proc)
+			if tt.wantErr {
+				require.True(t, moerr.IsMoErrCode(err, moerr.ErrOutOfRange), err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
diff --git a/pkg/sql/colexec/table_clone/types.go b/pkg/sql/colexec/table_clone/types.go
index 29ab02d8158fb..00c32b705af50 100644
--- a/pkg/sql/colexec/table_clone/types.go
+++ b/pkg/sql/colexec/table_clone/types.go
@@ -37,7 +37,21 @@ type TableCloneCtx struct {
 	SrcCtx       context.Context
 	ScanSnapshot *plan.Snapshot
 
-	SrcAutoIncrOffsets map[int32]uint64
+	RequestedAutoIncrOffset uint64
+	// Source allocator state is keyed by lower-cased destination column name.
+	// COPY can reorder destination columns, so source indexes are not stable here.
+	SrcAutoIncrMaxValues map[string]uint64
+	SrcAutoIncrOffsets   map[string]uint64
+
+	// IndexAutoIncrStates uses the same lower-cased index key as dstIdxRel
+	// (partition.index.type for partitioned tables). Each hidden table owns an
+	// independent allocator and must be reconciled after its objects are cloned.
+	IndexAutoIncrStates map[string]AutoIncrementState
+}
+
+type AutoIncrementState struct {
+	MaxValues map[string]uint64
+	Offsets   map[string]uint64
 }
 
 type TableClone struct {
diff --git a/pkg/sql/compile/alter.go b/pkg/sql/compile/alter.go
index f785c0fbd260e..a928a7cfc339c 100644
--- a/pkg/sql/compile/alter.go
+++ b/pkg/sql/compile/alter.go
@@ -16,6 +16,7 @@ package compile
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"slices"
 	"strings"
@@ -23,7 +24,10 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/reuse"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/defines"
+	"github.com/matrixorigin/matrixone/pkg/incrservice"
 	indexplugin "github.com/matrixorigin/matrixone/pkg/indexplugin"
 	catalogplugin "github.com/matrixorigin/matrixone/pkg/indexplugin/catalog"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
@@ -57,6 +61,63 @@ func convertDBEOBToNoSuchTable(ctx context.Context, e error, dbName, tblName str
 	return e
 }
 
+type alterCopyAutoIncrementCleanup struct {
+	c        *Compile
+	tableIDs []uint64
+	tracked  map[uint64]struct{}
+}
+
+func newAlterCopyAutoIncrementCleanup(c *Compile) *alterCopyAutoIncrementCleanup {
+	return &alterCopyAutoIncrementCleanup{
+		c:       c,
+		tracked: make(map[uint64]struct{}),
+	}
+}
+
+func (cleanup *alterCopyAutoIncrementCleanup) track(tableID uint64) {
+	if _, ok := cleanup.tracked[tableID]; ok {
+		return
+	}
+	cleanup.tracked[tableID] = struct{}{}
+	cleanup.tableIDs = append(cleanup.tableIDs, tableID)
+}
+
+func (cleanup *alterCopyAutoIncrementCleanup) finish(statementErr *error) {
+	if *statementErr == nil && cleanup.c.proc.Ctx != nil {
+		*statementErr = cleanup.c.proc.Ctx.Err()
+	}
+	if *statementErr == nil || len(cleanup.tableIDs) == 0 {
+		return
+	}
+
+	ctx := cleanup.c.proc.Ctx
+	if ctx == nil {
+		ctx = context.Background()
+	} else {
+		ctx = context.WithoutCancel(ctx)
+	}
+	svc := incrservice.GetAutoIncrementService(cleanup.c.proc.GetService())
+	var cleanupErr error
+	for _, tableID := range cleanup.tableIDs {
+		cleanupErr = errors.Join(
+			cleanupErr,
+			svc.DiscardOffsetReset(ctx, tableID, cleanup.c.proc.GetTxnOperator()),
+		)
+	}
+	if cleanupErr == nil {
+		return
+	}
+	if _, ok := (*statementErr).(*moerr.Error); ok {
+		cleanup.c.proc.Error(
+			ctx,
+			"alter.table.copy.discard.auto.increment.reset",
+			zap.Error(cleanupErr),
+		)
+		return
+	}
+	*statementErr = errors.Join(*statementErr, cleanupErr)
+}
+
 func shouldEnableAlterCopyPipelineFlush(opt *plan.AlterCopyOpt) bool {
 	return opt != nil && opt.SkipPkDedup
 }
@@ -301,7 +362,10 @@ func (c *Compile) precheckAlterCopyPkDedup(dbName, tblName string, qry *plan.Alt
 	return opt, nil
 }
 
-func (s *Scope) AlterTableCopy(c *Compile) error {
+func (s *Scope) AlterTableCopy(c *Compile) (err error) {
+	cleanup := newAlterCopyAutoIncrementCleanup(c)
+	defer cleanup.finish(&err)
+
 	qry := s.Plan.GetDdl().GetAlterTable()
 	dbName := qry.Database
 
@@ -475,6 +539,15 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 			zap.Error(err))
 		return err
 	}
+	if err = c.reconcileAlterCopyAutoIncrement(
+		dbName,
+		qry.TableDef,
+		qry.CopyTableDef,
+		newRel,
+		cleanup,
+	); err != nil {
+		return err
+	}
 
 	//6. copy on writing unaffected index table
 	if err = cloneUnaffectedIndexes(
@@ -720,6 +793,112 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 	return nil
 }
 
+// reconcileAlterCopyAutoIncrement publishes allocator state for the temporary
+// table only after copied rows are visible in the ALTER transaction. Retained
+// source columns are matched by stable planner column ID, never by position or
+// a reused name.
+func (c *Compile) reconcileAlterCopyAutoIncrement(
+	dbName string,
+	srcDef *plan.TableDef,
+	copyDef *plan.TableDef,
+	newRel engine.Relation,
+	cleanup *alterCopyAutoIncrementCleanup,
+) error {
+	if err := c.proc.Ctx.Err(); err != nil {
+		return err
+	}
+	autoCols := incrservice.GetUserAutoColumnFromDef(copyDef)
+	if len(autoCols) == 0 {
+		return nil
+	}
+
+	sourceOffsets := make(map[string]uint64)
+	sourceNames := mapCloneAutoIncrColumns(srcDef, copyDef, true)
+	if len(sourceNames) > 0 {
+		sql := fmt.Sprintf(
+			"select col_index, offset from mo_catalog.mo_increment_columns where table_id = %d",
+			srcDef.TblId,
+		)
+		result, err := c.runSqlWithResultAndOptions(
+			sql,
+			NoAccountId,
+			executor.StatementOption{}.WithDisableLog(),
+		)
+		if err != nil {
+			result.Close()
+			return err
+		}
+		func() {
+			defer result.Close()
+			result.ReadRows(func(rows int, cols []*vector.Vector) bool {
+				colIndexes := vector.MustFixedColWithTypeCheck[int32](cols[0])
+				offsets := vector.MustFixedColWithTypeCheck[uint64](cols[1])
+				for i := 0; i < rows; i++ {
+					if name, ok := sourceNames[colIndexes[i]]; ok {
+						sourceOffsets[name] = offsets[i]
+					}
+				}
+				return true
+			})
+		}()
+	}
+
+	tableID := newRel.GetTableID(c.proc.Ctx)
+	svc := incrservice.GetAutoIncrementService(c.proc.GetService())
+	for _, col := range autoCols {
+		if err := c.proc.Ctx.Err(); err != nil {
+			return err
+		}
+		colIdent := quoteAlterCopyIdentifier(col.ColName)
+		maxSQL := fmt.Sprintf(
+			"select cast(coalesce(max(case when %s > 0 then %s else 0 end), 0) as unsigned) from %s",
+			colIdent,
+			colIdent,
+			quoteAlterCopyTableName(dbName, copyDef.Name),
+		)
+		result, err := c.runSqlWithResultAndOptions(
+			maxSQL,
+			NoAccountId,
+			executor.StatementOption{}.WithDisableLog(),
+		)
+		if err != nil {
+			result.Close()
+			return err
+		}
+		var copiedMax uint64
+		func() {
+			defer result.Close()
+			result.ReadRows(func(rows int, cols []*vector.Vector) bool {
+				if rows > 0 && len(cols) > 0 && !cols[0].IsNull(0) {
+					copiedMax = executor.GetFixedRows[uint64](cols[0])[0]
+				}
+				return false
+			})
+		}()
+
+		name := strings.ToLower(col.ColName)
+		effectiveOffset := max(copyDef.AutoIncrOffset, copiedMax, sourceOffsets[name])
+		if err := incrservice.ValidateAutoColumnOffset(
+			c.proc.Ctx,
+			types.T(copyDef.Cols[col.ColIndex].Typ.Id),
+			effectiveOffset,
+		); err != nil {
+			return err
+		}
+		if err := svc.SetOffset(
+			c.proc.Ctx,
+			tableID,
+			col.ColName,
+			effectiveOffset,
+			c.proc.GetTxnOperator(),
+		); err != nil {
+			return err
+		}
+		cleanup.track(tableID)
+	}
+	return nil
+}
+
 func (s *Scope) AlterTable(c *Compile) (err error) {
 	if s.ScopeAnalyzer == nil {
 		s.ScopeAnalyzer = NewScopeAnalyzer()
diff --git a/pkg/sql/compile/alter_test.go b/pkg/sql/compile/alter_test.go
index f1c4085e164cf..8c5a0efab0923 100644
--- a/pkg/sql/compile/alter_test.go
+++ b/pkg/sql/compile/alter_test.go
@@ -37,6 +37,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test"
 	mock_lock "github.com/matrixorigin/matrixone/pkg/frontend/test/mock_lock"
+	"github.com/matrixorigin/matrixone/pkg/incrservice"
 	"github.com/matrixorigin/matrixone/pkg/lockservice"
 	"github.com/matrixorigin/matrixone/pkg/objectio/ioutil"
 	"github.com/matrixorigin/matrixone/pkg/pb/api"
@@ -44,6 +45,7 @@ import (
 	plan2 "github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/txn/client"
 	"github.com/matrixorigin/matrixone/pkg/util/executor"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
@@ -70,6 +72,31 @@ func TestIsAlterAffectedPluginIndexMatchesIndexNamePartsAndIncludedColumns(t *te
 	require.False(t, isAlterAffectedPluginIndex(nil, []string{"idx_vec"}))
 }
 
+func TestAlterCopyAutoIncrementCleanupDiscardsTrackedReset(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	proc := testutil.NewProcess(t)
+	proc.Ctx = context.Background()
+	_, txnOp := newTestTxnClientAndOp(ctrl)
+	proc.Base.TxnOperator = txnOp
+
+	cleanupErr := errors.New("discard failed")
+	autoSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+	autoSvc.EXPECT().DiscardOffsetReset(gomock.Any(), uint64(11), txnOp).Return(cleanupErr)
+	autoSvc.EXPECT().DiscardOffsetReset(gomock.Any(), uint64(12), txnOp).Return(nil)
+	incrservice.SetAutoIncrementServiceByID(proc.GetService(), autoSvc)
+
+	cleanup := newAlterCopyAutoIncrementCleanup(&Compile{proc: proc})
+	cleanup.track(11)
+	cleanup.track(11)
+	cleanup.track(12)
+	originalErr := errors.New("statement failed")
+	statementErr := originalErr
+	cleanup.finish(&statementErr)
+
+	require.ErrorIs(t, statementErr, originalErr)
+	require.ErrorIs(t, statementErr, cleanupErr)
+}
+
 type alterCopyInsertSpyExecutor struct {
 	insertSQL    string
 	insertErr    error
@@ -80,6 +107,152 @@ type alterCopyInsertSpyExecutor struct {
 	executedSQLs []string
 }
 
+func TestReconcileAlterCopyAutoIncrementUsesStableIdentityAndSafeBounds(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	resultMP := mpool.MustNewZero()
+	sourceOffsetSQL := "select col_index, offset from mo_catalog.mo_increment_columns where table_id = 1"
+	renamedMaxSQL := "select cast(coalesce(max(case when `renamed_id` > 0 then `renamed_id` else 0 end), 0) as unsigned) from `test`.`dept_copy`"
+	reusedMaxSQL := "select cast(coalesce(max(case when `id` > 0 then `id` else 0 end), 0) as unsigned) from `test`.`dept_copy`"
+	spyExec := &alterCopyInsertSpyExecutor{results: map[string]executor.Result{
+		sourceOffsetSQL: newTableCloneOffsetResult(t, resultMP, 0, 500),
+		renamedMaxSQL:   newAlterCopyFixedResult(t, resultMP, types.T_uint64.ToType(), []uint64{40}),
+		reusedMaxSQL:    newAlterCopyFixedResult(t, resultMP, types.T_uint64.ToType(), []uint64{0}),
+	}}
+	c := newAlterCopyPrecheckCompile(t, ctrl, spyExec)
+
+	autoType := plan.Type{Id: int32(types.T_uint64), AutoIncr: true}
+	srcDef := &plan.TableDef{
+		TblId: 1,
+		Cols: []*plan.ColDef{
+			{ColId: 10, Name: "id", Typ: autoType},
+			{ColId: 11, Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+		},
+	}
+	copyDef := &plan.TableDef{
+		TblId:          2,
+		Name:           "dept_copy",
+		AutoIncrOffset: 99,
+		Cols: []*plan.ColDef{
+			{ColId: 12, Name: "id", Typ: autoType},
+			{ColId: 10, Name: "renamed_id", Typ: autoType},
+			{ColId: 11, Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+		},
+	}
+	copyRel := mock_frontend.NewMockRelation(ctrl)
+	copyRel.EXPECT().GetTableID(gomock.Any()).Return(copyDef.TblId).AnyTimes()
+	autoSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+	gomock.InOrder(
+		autoSvc.EXPECT().SetOffset(c.proc.Ctx, copyDef.TblId, "id", uint64(99), c.proc.GetTxnOperator()),
+		autoSvc.EXPECT().SetOffset(c.proc.Ctx, copyDef.TblId, "renamed_id", uint64(500), c.proc.GetTxnOperator()),
+		autoSvc.EXPECT().DiscardOffsetReset(gomock.Any(), copyDef.TblId, c.proc.GetTxnOperator()).Return(nil),
+	)
+	incrservice.SetAutoIncrementServiceByID(c.proc.GetService(), autoSvc)
+
+	cleanup := newAlterCopyAutoIncrementCleanup(c)
+	require.NoError(t, c.reconcileAlterCopyAutoIncrement(
+		"test", srcDef, copyDef, copyRel, cleanup,
+	))
+	require.Equal(t, []string{sourceOffsetSQL, reusedMaxSQL, renamedMaxSQL}, spyExec.executedSQLs)
+	require.Zero(t, resultMP.CurrNB(), "all internal SQL results must be closed")
+	laterErr := errors.New("later ALTER COPY step failed")
+	cleanup.finish(&laterErr)
+	require.ErrorContains(t, laterErr, "later ALTER COPY step failed")
+}
+
+func TestReconcileAlterCopyAutoIncrementSkipsHiddenAndRejectsNarrowedOverflow(t *testing.T) {
+	t.Run("hidden only", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		spyExec := &alterCopyInsertSpyExecutor{}
+		c := newAlterCopyPrecheckCompile(t, ctrl, spyExec)
+		copyDef := &plan.TableDef{
+			TblId: 2,
+			Name:  "dept_copy",
+			Cols: []*plan.ColDef{{
+				ColId: 1, Name: catalog.FakePrimaryKeyColName, Hidden: true,
+				Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+			}},
+		}
+		copyRel := mock_frontend.NewMockRelation(ctrl)
+		autoSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+		autoSvc.EXPECT().SetOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
+		incrservice.SetAutoIncrementServiceByID(c.proc.GetService(), autoSvc)
+
+		require.NoError(t, c.reconcileAlterCopyAutoIncrement(
+			"test", &plan.TableDef{}, copyDef, copyRel, newAlterCopyAutoIncrementCleanup(c),
+		))
+		require.Empty(t, spyExec.executedSQLs)
+	})
+
+	t.Run("source offset exceeds narrowed type", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		resultMP := mpool.MustNewZero()
+		sourceOffsetSQL := "select col_index, offset from mo_catalog.mo_increment_columns where table_id = 1"
+		maxSQL := "select cast(coalesce(max(case when `id` > 0 then `id` else 0 end), 0) as unsigned) from `test`.`dept_copy`"
+		spyExec := &alterCopyInsertSpyExecutor{results: map[string]executor.Result{
+			sourceOffsetSQL: newTableCloneOffsetResult(t, resultMP, 0, 300),
+			maxSQL:          newAlterCopyFixedResult(t, resultMP, types.T_uint64.ToType(), []uint64{40}),
+		}}
+		c := newAlterCopyPrecheckCompile(t, ctrl, spyExec)
+		srcDef := &plan.TableDef{TblId: 1, Cols: []*plan.ColDef{{
+			ColId: 10, Name: "id", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+		}}}
+		copyDef := &plan.TableDef{TblId: 2, Name: "dept_copy", Cols: []*plan.ColDef{{
+			ColId: 10, Name: "id", Typ: plan.Type{Id: int32(types.T_uint8), AutoIncr: true},
+		}}}
+		copyRel := mock_frontend.NewMockRelation(ctrl)
+		copyRel.EXPECT().GetTableID(gomock.Any()).Return(copyDef.TblId).AnyTimes()
+		autoSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+		autoSvc.EXPECT().SetOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
+		incrservice.SetAutoIncrementServiceByID(c.proc.GetService(), autoSvc)
+
+		err := c.reconcileAlterCopyAutoIncrement(
+			"test", srcDef, copyDef, copyRel, newAlterCopyAutoIncrementCleanup(c),
+		)
+		require.True(t, moerr.IsMoErrCode(err, moerr.ErrOutOfRange), err)
+		require.Zero(t, resultMP.CurrNB(), "all internal SQL results must be closed")
+	})
+}
+
+func TestReconcileAlterCopyAutoIncrementStopsAfterCancellation(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	resultMP := mpool.MustNewZero()
+	firstMaxSQL := "select cast(coalesce(max(case when `first` > 0 then `first` else 0 end), 0) as unsigned) from `test`.`dept_copy`"
+	spyExec := &alterCopyInsertSpyExecutor{results: map[string]executor.Result{
+		firstMaxSQL: newAlterCopyFixedResult(t, resultMP, types.T_uint64.ToType(), []uint64{40}),
+	}}
+	c := newAlterCopyPrecheckCompile(t, ctrl, spyExec)
+	ctx, cancel := context.WithCancel(c.proc.Ctx)
+	c.proc.Ctx = ctx
+	c.proc.ReplaceTopCtx(ctx)
+
+	copyDef := &plan.TableDef{
+		TblId:          2,
+		Name:           "dept_copy",
+		AutoIncrOffset: 99,
+		Cols: []*plan.ColDef{
+			{ColId: 20, Name: "first", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+			{ColId: 21, Name: "second", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+		},
+	}
+	copyRel := mock_frontend.NewMockRelation(ctrl)
+	copyRel.EXPECT().GetTableID(gomock.Any()).Return(copyDef.TblId).AnyTimes()
+	autoSvc := mock_frontend.NewMockAutoIncrementService(ctrl)
+	autoSvc.EXPECT().SetOffset(ctx, copyDef.TblId, "first", uint64(99), c.proc.GetTxnOperator()).DoAndReturn(
+		func(context.Context, uint64, string, uint64, client.TxnOperator) error {
+			cancel()
+			return nil
+		},
+	)
+	incrservice.SetAutoIncrementServiceByID(c.proc.GetService(), autoSvc)
+
+	err := c.reconcileAlterCopyAutoIncrement(
+		"test", &plan.TableDef{}, copyDef, copyRel, newAlterCopyAutoIncrementCleanup(c),
+	)
+	require.ErrorIs(t, err, context.Canceled)
+	require.Equal(t, []string{firstMaxSQL}, spyExec.executedSQLs)
+	require.Zero(t, resultMP.CurrNB())
+}
+
 const (
 	alterCopyTestPkNullCheckSQL      = "SELECT `col4` FROM `test`.`dept` WHERE `col4` IS NULL LIMIT 1"
 	alterCopyTestPkDuplicateCheckSQL = "SELECT `col4` FROM `test`.`dept` GROUP BY `col4` HAVING count(*) > 1 LIMIT 1"
diff --git a/pkg/sql/compile/operator.go b/pkg/sql/compile/operator.go
index b907b80daf839..1313ae42748d5 100644
--- a/pkg/sql/compile/operator.go
+++ b/pkg/sql/compile/operator.go
@@ -26,11 +26,13 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/bitmap"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/common/sqlquote"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	icebergapi "github.com/matrixorigin/matrixone/pkg/iceberg/api"
+	"github.com/matrixorigin/matrixone/pkg/incrservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/pipeline"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
@@ -2363,6 +2365,12 @@ func constructTableClone(
 ) (*table_clone.TableClone, error) {
 
 	metaCopy := table_clone.NewTableClone()
+	success := false
+	defer func() {
+		if !success {
+			metaCopy.Release()
+		}
+	}()
 
 	metaCopy.Ctx = &table_clone.TableCloneCtx{
 		Eng:       c.e,
@@ -2373,78 +2381,417 @@ func constructTableClone(
 		DstTblName:      clonePlan.DstTableName,
 		DstDatabaseName: clonePlan.DstDatabaseName,
 	}
+	dstTblDef := clonePlan.SrcTableDef
+	sameColumnIDSpace := true
+	var dstCreateTable *plan.CreateTable
+	metaCopy.Ctx.RequestedAutoIncrOffset = clonePlan.SrcTableDef.AutoIncrOffset
+	if createPlan := clonePlan.GetCreateTable(); createPlan != nil {
+		if createTable := createPlan.GetDdl().GetCreateTable(); createTable != nil && createTable.TableDef != nil {
+			dstCreateTable = createTable
+			dstTblDef = createTable.TableDef
+			sameColumnIDSpace = false
+			metaCopy.Ctx.RequestedAutoIncrOffset = createTable.TableDef.AutoIncrOffset
+		}
+	}
+	dstAutoIncrNames := mapCloneAutoIncrColumns(clonePlan.SrcTableDef, dstTblDef, sameColumnIDSpace)
+	mappedIndexAutoIncrTables := mapCloneIndexAutoIncrementTables(clonePlan.SrcTableDef, dstCreateTable)
 
 	var (
 		err error
 		ret executor.Result
 		sql string
 
-		account     = uint32(math.MaxUint32)
-		colOffset   map[int32]uint64
-		hasAutoIncr bool
+		account         = uint32(math.MaxUint32)
+		colMaxValue     map[string]uint64
+		autoIncrOffsets map[string]uint64
+		mainHasAutoIncr bool
 	)
 
 	for _, colDef := range clonePlan.SrcTableDef.Cols {
 		if colDef.Typ.AutoIncr {
-			hasAutoIncr = true
+			mainHasAutoIncr = true
 			break
 		}
 	}
 
-	if !hasAutoIncr {
+	if !mainHasAutoIncr && len(mappedIndexAutoIncrTables) == 0 {
+		success = true
 		return metaCopy, nil
 	}
 
-	sql = fmt.Sprintf(
-		"select col_index, offset from mo_catalog.mo_increment_columns where table_id = %d",
-		clonePlan.SrcTableDef.TblId,
-	)
-
-	if clonePlan.ScanSnapshot != nil {
+	if clonePlan.SrcObjDef != nil && clonePlan.SrcObjDef.PubInfo != nil {
+		account = uint32(clonePlan.SrcObjDef.PubInfo.TenantId)
+	} else if clonePlan.ScanSnapshot != nil {
 		if clonePlan.ScanSnapshot.Tenant != nil {
 			account = clonePlan.ScanSnapshot.Tenant.TenantID
 		}
+	}
+
+	if account == math.MaxUint32 {
+		if account, err = defines.GetAccountId(c.proc.Ctx); err != nil {
+			return nil, err
+		}
+	}
 
-		if clonePlan.ScanSnapshot.TS != nil {
+	if mainHasAutoIncr {
+		if err := c.proc.Ctx.Err(); err != nil {
+			return nil, err
+		}
+		sql = fmt.Sprintf(
+			"select col_index, offset from mo_catalog.mo_increment_columns where table_id = %d",
+			clonePlan.SrcTableDef.TblId,
+		)
+		if clonePlan.ScanSnapshot != nil && clonePlan.ScanSnapshot.TS != nil {
 			sql = fmt.Sprintf(
 				"select col_index, offset from mo_catalog.mo_increment_columns {MO_TS = %d} where table_id = %d",
-				clonePlan.ScanSnapshot.TS.PhysicalTime, clonePlan.SrcTableDef.TblId,
+				clonePlan.ScanSnapshot.TS.PhysicalTime,
+				clonePlan.SrcTableDef.TblId,
 			)
 		}
-	}
+		if ret, err = c.runSqlWithResultAndOptions(
+			sql,
+			int32(account),
+			executor.StatementOption{}.WithDisableLog(),
+		); err != nil {
+			ret.Close()
+			return nil, err
+		}
+		autoIncrOffsets = make(map[string]uint64)
+		func() {
+			defer ret.Close()
+			ret.ReadRows(func(rows int, cols []*vector.Vector) bool {
+				colIdxes := vector.MustFixedColWithTypeCheck[int32](cols[0])
+				offsets := vector.MustFixedColWithTypeCheck[uint64](cols[1])
+				for i := 0; i < rows; i++ {
+					if dstName, ok := dstAutoIncrNames[colIdxes[i]]; ok {
+						autoIncrOffsets[dstName] = offsets[i]
+					}
+				}
+				return true
+			})
+		}()
 
-	if account == math.MaxUint32 {
-		if account, err = defines.GetAccountId(c.proc.Ctx); err != nil {
+		colMaxValue = make(map[string]uint64)
+		for colIdx, colDef := range clonePlan.SrcTableDef.Cols {
+			if !colDef.Typ.AutoIncr || colDef.Hidden {
+				continue
+			}
+			if err := c.proc.Ctx.Err(); err != nil {
+				return nil, err
+			}
+
+			colIdent := sqlquote.Ident(colDef.Name)
+			tableIdent := sqlquote.QualifiedIdent(clonePlan.SrcTableDef.DbName, clonePlan.SrcTableDef.Name)
+			if (clonePlan.SrcObjDef == nil || clonePlan.SrcObjDef.PubInfo == nil) &&
+				clonePlan.ScanSnapshot != nil && clonePlan.ScanSnapshot.TS != nil {
+				tableIdent += fmt.Sprintf(" {MO_TS = %d}", clonePlan.ScanSnapshot.TS.PhysicalTime)
+			}
+			sql = fmt.Sprintf(
+				"select cast(coalesce(max(case when %s > 0 then %s else 0 end), 0) as unsigned) from %s",
+				colIdent,
+				colIdent,
+				tableIdent,
+			)
+
+			if ret, err = c.runSqlWithResultAndOptions(
+				sql,
+				int32(account),
+				executor.StatementOption{}.WithDisableLog(),
+			); err != nil {
+				ret.Close()
+				return nil, err
+			}
+			func() {
+				defer ret.Close()
+				ret.ReadRows(func(rows int, cols []*vector.Vector) bool {
+					if rows > 0 && len(cols) > 0 && !cols[0].IsNull(0) {
+						if dstName, ok := dstAutoIncrNames[int32(colIdx)]; ok {
+							colMaxValue[dstName] = executor.GetFixedRows[uint64](cols[0])[0]
+						}
+					}
+					return false
+				})
+			}()
+		}
+
+		metaCopy.Ctx.SrcAutoIncrMaxValues = colMaxValue
+		metaCopy.Ctx.SrcAutoIncrOffsets = autoIncrOffsets
+	}
+
+	indexAutoIncrTables := mappedIndexAutoIncrTables
+	if features.IsPartitioned(clonePlan.SrcTableDef.FeatureFlag) && len(mappedIndexAutoIncrTables) > 0 {
+		partitionTables, err := c.readClonePartitionIndexAutoIncrementTables(
+			clonePlan,
+			mappedIndexAutoIncrTables,
+			int32(account),
+		)
+		if err != nil {
 			return nil, err
 		}
+		indexAutoIncrTables = append(indexAutoIncrTables, partitionTables...)
 	}
+	if len(indexAutoIncrTables) > 0 {
+		metaCopy.Ctx.IndexAutoIncrStates = make(map[string]table_clone.AutoIncrementState, len(indexAutoIncrTables))
+		for _, table := range indexAutoIncrTables {
+			state, err := c.readCloneIndexAutoIncrementState(
+				clonePlan,
+				table.srcTableName,
+				int32(account),
+			)
+			if err != nil {
+				return nil, err
+			}
+			if len(state.Offsets) > 0 {
+				metaCopy.Ctx.IndexAutoIncrStates[strings.ToLower(table.stateKey)] = state
+			}
+		}
+	}
+	success = true
+	return metaCopy, nil
+}
 
-	if ret, err = c.runSqlWithResultAndOptions(
-		sql,
-		int32(account),
-		executor.StatementOption{}.WithDisableLog(),
-	); err != nil {
-		return nil, err
+type cloneIndexAutoIncrementTable struct {
+	srcTableName string
+	sourceKey    string
+	stateKey     string
+}
+
+func mapCloneIndexAutoIncrementTables(
+	srcDef *plan.TableDef,
+	dstCreate *plan.CreateTable,
+) []cloneIndexAutoIncrementTable {
+	if srcDef == nil || dstCreate == nil || dstCreate.TableDef == nil {
+		return nil
 	}
 
-	ret.ReadRows(func(rows int, cols []*vector.Vector) bool {
-		if colOffset == nil {
-			colOffset = make(map[int32]uint64)
+	dstIndexByKey := make(map[string]*plan.IndexDef, len(dstCreate.TableDef.Indexes))
+	for _, index := range dstCreate.TableDef.Indexes {
+		if index == nil {
+			continue
 		}
+		dstIndexByKey[cloneIndexTableKey(index)] = index
+	}
+	dstTableByName := make(map[string]*plan.TableDef, len(dstCreate.IndexTables))
+	for _, tableDef := range dstCreate.IndexTables {
+		if tableDef == nil {
+			continue
+		}
+		dstTableByName[strings.ToLower(tableDef.Name)] = tableDef
+	}
 
-		colIdxes := vector.MustFixedColWithTypeCheck[int32](cols[0])
-		offsets := vector.MustFixedColWithTypeCheck[uint64](cols[1])
-
-		for i := 0; i < rows; i++ {
-			colOffset[colIdxes[i]] = offsets[i]
+	result := make([]cloneIndexAutoIncrementTable, 0, len(srcDef.Indexes))
+	for _, srcIndex := range srcDef.Indexes {
+		if srcIndex == nil {
+			continue
+		}
+		dstIndex := dstIndexByKey[cloneIndexTableKey(srcIndex)]
+		if dstIndex == nil {
+			continue
+		}
+		dstTableDef := dstTableByName[strings.ToLower(dstIndex.IndexTableName)]
+		if dstTableDef == nil || len(incrservice.GetAutoColumnFromDef(dstTableDef)) == 0 {
+			continue
 		}
+		result = append(result, cloneIndexAutoIncrementTable{
+			srcTableName: srcIndex.IndexTableName,
+			sourceKey:    srcIndex.IndexName + "." + srcIndex.IndexAlgoTableType,
+			stateKey:     dstIndex.IndexName + "." + dstIndex.IndexAlgoTableType,
+		})
+	}
+	return result
+}
 
-		return true
-	})
+func cloneIndexTableKey(index *plan.IndexDef) string {
+	if index == nil {
+		return ""
+	}
+	return strings.ToLower(index.IndexName) + "\x00" + strings.ToLower(index.IndexAlgoTableType)
+}
 
-	ret.Close()
+func (c *Compile) readClonePartitionIndexAutoIncrementTables(
+	clonePlan *plan.CloneTable,
+	mapped []cloneIndexAutoIncrementTable,
+	account int32,
+) ([]cloneIndexAutoIncrementTable, error) {
+	if err := c.proc.Ctx.Err(); err != nil {
+		return nil, err
+	}
+	destinationKeyBySource := make(map[string]string, len(mapped))
+	for _, table := range mapped {
+		destinationKeyBySource[strings.ToLower(table.sourceKey)] = table.stateKey
+	}
+	hint := ""
+	if clonePlan.ScanSnapshot != nil && clonePlan.ScanSnapshot.TS != nil {
+		hint = fmt.Sprintf(" {MO_TS = %d}", clonePlan.ScanSnapshot.TS.PhysicalTime)
+	}
+	sql := fmt.Sprintf(
+		"select pt.partition_name, mi.name, mi.algo_table_type, mi.index_table_name "+
+			"from mo_catalog.%s%s pt join mo_catalog.%s%s mi on pt.partition_id = mi.table_id "+
+			"where pt.primary_table_id = %d",
+		catalog.MOPartitionTables,
+		hint,
+		catalog.MO_INDEXES,
+		hint,
+		clonePlan.SrcTableDef.TblId,
+	)
+	result, err := c.runSqlWithResultAndOptions(
+		sql,
+		account,
+		executor.StatementOption{}.WithDisableLog(),
+	)
+	if err != nil {
+		result.Close()
+		return nil, err
+	}
+	var tables []cloneIndexAutoIncrementTable
+	func() {
+		defer result.Close()
+		result.ReadRows(func(rows int, cols []*vector.Vector) bool {
+			partitionNames := executor.GetStringRows(cols[0])
+			indexNames := executor.GetStringRows(cols[1])
+			tableTypes := executor.GetStringRows(cols[2])
+			tableNames := executor.GetStringRows(cols[3])
+			for i := 0; i < rows; i++ {
+				sourceKey := indexNames[i] + "." + tableTypes[i]
+				destinationKey, ok := destinationKeyBySource[strings.ToLower(sourceKey)]
+				if !ok {
+					continue
+				}
+				tables = append(tables, cloneIndexAutoIncrementTable{
+					srcTableName: tableNames[i],
+					sourceKey:    sourceKey,
+					stateKey:     partitionNames[i] + "." + destinationKey,
+				})
+			}
+			return true
+		})
+	}()
+	return tables, nil
+}
 
-	metaCopy.Ctx.SrcAutoIncrOffsets = colOffset
+func (c *Compile) readCloneIndexAutoIncrementState(
+	clonePlan *plan.CloneTable,
+	srcTableName string,
+	account int32,
+) (table_clone.AutoIncrementState, error) {
+	state := table_clone.AutoIncrementState{
+		MaxValues: make(map[string]uint64),
+		Offsets:   make(map[string]uint64),
+	}
+	hint := ""
+	if clonePlan.ScanSnapshot != nil && clonePlan.ScanSnapshot.TS != nil {
+		hint = fmt.Sprintf(" {MO_TS = %d}", clonePlan.ScanSnapshot.TS.PhysicalTime)
+	}
+	offsetSQL := fmt.Sprintf(
+		"select col_name, offset from mo_catalog.mo_increment_columns%s where table_id = "+
+			"(select rel_id from mo_catalog.mo_tables%s where reldatabase = %s and relname = %s)",
+		hint,
+		hint,
+		sqlquote.String(clonePlan.SrcTableDef.DbName),
+		sqlquote.String(srcTableName),
+	)
+	result, err := c.runSqlWithResultAndOptions(
+		offsetSQL,
+		account,
+		executor.StatementOption{}.WithDisableLog(),
+	)
+	if err != nil {
+		result.Close()
+		return state, err
+	}
+	sourceColNames := make(map[string]string)
+	func() {
+		defer result.Close()
+		result.ReadRows(func(rows int, cols []*vector.Vector) bool {
+			names := executor.GetStringRows(cols[0])
+			offsets := executor.GetFixedRows[uint64](cols[1])
+			for i := 0; i < rows; i++ {
+				key := strings.ToLower(names[i])
+				sourceColNames[key] = names[i]
+				state.Offsets[key] = offsets[i]
+			}
+			return true
+		})
+	}()
+
+	for key, colName := range sourceColNames {
+		if err := c.proc.Ctx.Err(); err != nil {
+			return state, err
+		}
+		colIdent := sqlquote.Ident(colName)
+		tableIdent := sqlquote.QualifiedIdent(clonePlan.SrcTableDef.DbName, srcTableName)
+		if (clonePlan.SrcObjDef == nil || clonePlan.SrcObjDef.PubInfo == nil) && hint != "" {
+			tableIdent += hint
+		}
+		maxSQL := fmt.Sprintf(
+			"select cast(coalesce(max(case when %s > 0 then %s else 0 end), 0) as unsigned) from %s",
+			colIdent,
+			colIdent,
+			tableIdent,
+		)
+		result, err = c.runSqlWithResultAndOptions(
+			maxSQL,
+			account,
+			executor.StatementOption{}.WithDisableLog(),
+		)
+		if err != nil {
+			result.Close()
+			return state, err
+		}
+		func() {
+			defer result.Close()
+			result.ReadRows(func(rows int, cols []*vector.Vector) bool {
+				if rows > 0 && len(cols) > 0 && !cols[0].IsNull(0) {
+					state.MaxValues[key] = executor.GetFixedRows[uint64](cols[0])[0]
+				}
+				return false
+			})
+		}()
+	}
+	return state, nil
+}
 
-	return metaCopy, nil
+// mapCloneAutoIncrColumns translates source catalog indexes to destination
+// column names. ALTER COPY can reorder or rename columns while retaining their
+// planner column IDs, so source indexes must not be used as destination indexes.
+func mapCloneAutoIncrColumns(src, dst *plan.TableDef, sameColumnIDSpace bool) map[int32]string {
+	result := make(map[int32]string)
+	if src == nil || dst == nil {
+		return result
+	}
+
+	srcIDCounts := make(map[uint64]int, len(src.Cols))
+	dstByID := make(map[uint64][]*plan.ColDef, len(dst.Cols))
+	dstByName := make(map[string]*plan.ColDef, len(dst.Cols))
+	for _, col := range src.Cols {
+		srcIDCounts[col.ColId]++
+	}
+	for _, col := range dst.Cols {
+		dstByID[col.ColId] = append(dstByID[col.ColId], col)
+		dstByName[strings.ToLower(col.Name)] = col
+	}
+
+	for idx, srcCol := range src.Cols {
+		if !srcCol.Typ.AutoIncr {
+			continue
+		}
+		var dstCol *plan.ColDef
+		if sameColumnIDSpace && srcIDCounts[srcCol.ColId] == 1 {
+			if len(dstByID[srcCol.ColId]) == 1 {
+				dstCol = dstByID[srcCol.ColId][0]
+			} else {
+				// The source column was dropped. A new column reusing its name
+				// must not inherit the old allocator.
+				continue
+			}
+		} else {
+			// Fresh CREATE plans use a different column-ID coordinate system,
+			// so the cloned name is the stable identity.
+			dstCol = dstByName[strings.ToLower(srcCol.Name)]
+		}
+		if dstCol != nil && dstCol.Typ.AutoIncr {
+			result[int32(idx)] = strings.ToLower(dstCol.Name)
+		}
+	}
+	return result
 }
diff --git a/pkg/sql/compile/operator_table_clone_test.go b/pkg/sql/compile/operator_table_clone_test.go
new file mode 100644
index 0000000000000..2de3598cce763
--- /dev/null
+++ b/pkg/sql/compile/operator_table_clone_test.go
@@ -0,0 +1,444 @@
+// Copyright 2026 Matrix Origin
+//
+// 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 compile
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
+	"github.com/matrixorigin/matrixone/pkg/common/sqlquote"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/table_clone"
+	"github.com/matrixorigin/matrixone/pkg/sql/features"
+	mysqlparser "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect/mysql"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/util/executor"
+)
+
+type tableCloneRecordingExecutor struct {
+	result executor.Result
+	err    error
+	run    func(string) (executor.Result, error)
+	sql    string
+	sqls   []string
+	opts   executor.Options
+}
+
+func (e *tableCloneRecordingExecutor) Exec(
+	ctx context.Context,
+	sql string,
+	opts executor.Options,
+) (executor.Result, error) {
+	if err := ctx.Err(); err != nil {
+		return executor.Result{}, err
+	}
+	e.sql = sql
+	e.sqls = append(e.sqls, sql)
+	e.opts = opts
+	if e.run != nil {
+		return e.run(sql)
+	}
+	return e.result, e.err
+}
+
+func (e *tableCloneRecordingExecutor) ExecTxn(
+	ctx context.Context,
+	_ func(executor.TxnExecutor) error,
+	_ executor.Options,
+) error {
+	return ctx.Err()
+}
+
+func newTableCloneResult(t *testing.T, mp *mpool.MPool, value uint64) executor.Result {
+	t.Helper()
+	return newAlterCopyFixedResult(t, mp, types.T_uint64.ToType(), []uint64{value})
+}
+
+func newTableCloneOffsetResult(t *testing.T, mp *mpool.MPool, colIdx int32, offset uint64) executor.Result {
+	t.Helper()
+	memRes := executor.NewMemResult([]types.Type{types.T_int32.ToType(), types.T_uint64.ToType()}, mp)
+	memRes.NewBatchWithRowCount(1)
+	require.NoError(t, executor.AppendFixedRows(memRes, 0, []int32{colIdx}))
+	require.NoError(t, executor.AppendFixedRows(memRes, 1, []uint64{offset}))
+	return memRes.GetResult()
+}
+
+func newTableCloneNamedOffsetResult(t *testing.T, mp *mpool.MPool, colName string, offset uint64) executor.Result {
+	t.Helper()
+	memRes := executor.NewMemResult([]types.Type{types.T_varchar.ToType(), types.T_uint64.ToType()}, mp)
+	memRes.NewBatchWithRowCount(1)
+	require.NoError(t, executor.AppendStringRows(memRes, 0, []string{colName}))
+	require.NoError(t, executor.AppendFixedRows(memRes, 1, []uint64{offset}))
+	return memRes.GetResult()
+}
+
+func newTableClonePartitionIndexResult(
+	t *testing.T,
+	mp *mpool.MPool,
+	partitionNames, indexNames, tableTypes, tableNames []string,
+) executor.Result {
+	t.Helper()
+	memRes := executor.NewMemResult([]types.Type{
+		types.T_varchar.ToType(),
+		types.T_varchar.ToType(),
+		types.T_varchar.ToType(),
+		types.T_varchar.ToType(),
+	}, mp)
+	memRes.NewBatchWithRowCount(len(partitionNames))
+	require.NoError(t, executor.AppendStringRows(memRes, 0, partitionNames))
+	require.NoError(t, executor.AppendStringRows(memRes, 1, indexNames))
+	require.NoError(t, executor.AppendStringRows(memRes, 2, tableTypes))
+	require.NoError(t, executor.AppendStringRows(memRes, 3, tableNames))
+	return memRes.GetResult()
+}
+
+func cloneCreatePlan(dstDef *plan.TableDef) *plan.Plan {
+	return &plan.Plan{Plan: &plan.Plan_Ddl{Ddl: &plan.DataDefinition{
+		Definition: &plan.DataDefinition_CreateTable{
+			CreateTable: &plan.CreateTable{TableDef: dstDef},
+		},
+	}}}
+}
+
+func TestConstructTableCloneReadsAllocatorAndMaximumFromSnapshot(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	exec := &tableCloneRecordingExecutor{}
+	exec.run = func(sql string) (executor.Result, error) {
+		if strings.Contains(sql, "mo_catalog.mo_increment_columns") {
+			return newTableCloneOffsetResult(t, proc.Mp(), 1, 50), nil
+		}
+		return newTableCloneResult(t, proc.Mp(), 40), nil
+	}
+	runtime.ServiceRuntime(proc.GetService()).SetGlobalVariables(runtime.InternalSQLExecutor, exec)
+
+	srcDef := &plan.TableDef{
+		TblId:          7,
+		DbName:         "db-name",
+		Name:           "t`name",
+		AutoIncrOffset: 999,
+		Cols: []*plan.ColDef{
+			{ColId: 1, Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+			{ColId: 11, Name: "id`col", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+		},
+	}
+	dstDef := &plan.TableDef{
+		AutoIncrOffset: 99,
+		Cols: []*plan.ColDef{
+			{ColId: 0, Name: "id`col", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true}},
+			{ColId: 10, Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+		},
+	}
+	snapshot := &plan.Snapshot{
+		TS:     &timestamp.Timestamp{PhysicalTime: 123},
+		Tenant: &plan.SnapshotTenant{TenantID: 17},
+	}
+	tc, err := constructTableClone(&Compile{proc: proc, pn: &plan.Plan{}}, &plan.CloneTable{
+		SrcTableDef:  srcDef,
+		SrcObjDef:    &plan.ObjectRef{},
+		ScanSnapshot: snapshot,
+		CreateTable:  cloneCreatePlan(dstDef),
+	})
+	require.NoError(t, err)
+	t.Cleanup(tc.Release)
+	require.Equal(t, uint64(99), tc.Ctx.RequestedAutoIncrOffset)
+	require.Equal(t, map[string]uint64{"id`col": 40}, tc.Ctx.SrcAutoIncrMaxValues)
+	require.Equal(t, map[string]uint64{"id`col": 50}, tc.Ctx.SrcAutoIncrOffsets)
+	require.True(t, exec.opts.HasAccountID())
+	require.Equal(t, uint32(17), exec.opts.AccountID())
+
+	col := sqlquote.Ident(srcDef.Cols[1].Name)
+	table := sqlquote.QualifiedIdent(srcDef.DbName, srcDef.Name)
+	require.Equal(t,
+		"select cast(coalesce(max(case when "+col+" > 0 then "+col+" else 0 end), 0) as unsigned) from "+
+			table+" {MO_TS = 123}",
+		exec.sql,
+	)
+}
+
+func TestMapCloneAutoIncrColumnsAcrossSchemaChanges(t *testing.T) {
+	autoType := plan.Type{Id: int32(types.T_uint64), AutoIncr: true}
+	tests := []struct {
+		name        string
+		src         *plan.TableDef
+		dst         *plan.TableDef
+		sameIDSpace bool
+		want        map[int32]string
+	}{
+		{
+			name: "fresh clone maps by name despite numeric id collision",
+			src: &plan.TableDef{Cols: []*plan.ColDef{
+				{ColId: 1, Name: "id", Typ: autoType},
+				{ColId: 2, Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+			}},
+			dst: &plan.TableDef{Cols: []*plan.ColDef{
+				{ColId: 1, Name: "payload", Typ: plan.Type{Id: int32(types.T_int64)}},
+				{ColId: 0, Name: "id", Typ: autoType},
+			}},
+			want: map[int32]string{0: "id"},
+		},
+		{
+			name: "copy alter maps rename and reorder by stable id",
+			src: &plan.TableDef{Cols: []*plan.ColDef{
+				{ColId: 10, Name: "dropped", Typ: plan.Type{Id: int32(types.T_int64)}},
+				{ColId: 11, Name: "id", Typ: autoType},
+				{ColId: 12, Name: "__mo_fake_pk_col", Hidden: true, Typ: autoType},
+			}},
+			dst: &plan.TableDef{Cols: []*plan.ColDef{
+				{ColId: 13, Name: "added", Typ: plan.Type{Id: int32(types.T_int64)}},
+				{ColId: 12, Name: "__mo_fake_pk_col", Hidden: true, Typ: autoType},
+				{ColId: 11, Name: "renamed_id", Typ: autoType},
+			}},
+			sameIDSpace: true,
+			want:        map[int32]string{1: "renamed_id", 2: "__mo_fake_pk_col"},
+		},
+		{
+			name: "dropped source does not transfer allocator to reused name",
+			src: &plan.TableDef{Cols: []*plan.ColDef{
+				{ColId: 10, Name: "id", Typ: autoType},
+			}},
+			dst: &plan.TableDef{Cols: []*plan.ColDef{
+				{ColId: 12, Name: "id", Typ: autoType},
+			}},
+			sameIDSpace: true,
+			want:        map[int32]string{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, mapCloneAutoIncrColumns(tt.src, tt.dst, tt.sameIDSpace))
+		})
+	}
+}
+
+func TestConstructTableCloneDoesNotReadHiddenMaximum(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	wantErr := errors.New("hidden auto-increment column must not be queried")
+	exec := &tableCloneRecordingExecutor{}
+	exec.run = func(sql string) (executor.Result, error) {
+		if strings.Contains(sql, "__mo_fake_pk_col") {
+			return executor.Result{}, wantErr
+		}
+		return newTableCloneOffsetResult(t, proc.Mp(), 0, 40), nil
+	}
+	runtime.ServiceRuntime(proc.GetService()).SetGlobalVariables(runtime.InternalSQLExecutor, exec)
+
+	tc, err := constructTableClone(&Compile{proc: proc, pn: &plan.Plan{}}, &plan.CloneTable{
+		SrcTableDef: &plan.TableDef{
+			TblId: 7,
+			Cols: []*plan.ColDef{{
+				Name:   "__mo_fake_pk_col",
+				Hidden: true,
+				Typ:    plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+			}},
+		},
+		SrcObjDef: &plan.ObjectRef{},
+	})
+	require.NoError(t, err)
+	t.Cleanup(tc.Release)
+	require.Len(t, exec.sqls, 1)
+	require.Equal(t, map[string]uint64{"__mo_fake_pk_col": 40}, tc.Ctx.SrcAutoIncrOffsets)
+}
+
+func TestConstructTableCloneCapturesHiddenIndexAllocator(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	const (
+		srcIndexTable = "src'fulltext\\table"
+		dstIndexTable = "__mo_index_fulltext_target"
+		fakePK        = "__mo_fake_pk_col"
+	)
+	exec := &tableCloneRecordingExecutor{}
+	exec.run = func(sql string) (executor.Result, error) {
+		if strings.Contains(sql, "mo_catalog.mo_increment_columns") {
+			return newTableCloneNamedOffsetResult(t, proc.Mp(), fakePK, 200), nil
+		}
+		return newTableCloneResult(t, proc.Mp(), 120), nil
+	}
+	runtime.ServiceRuntime(proc.GetService()).SetGlobalVariables(runtime.InternalSQLExecutor, exec)
+
+	srcDef := &plan.TableDef{
+		TblId:  7,
+		DbName: "db'name\\path",
+		Name:   "src",
+		Indexes: []*plan.IndexDef{{
+			IndexName:      "ftidx",
+			IndexAlgo:      "fulltext",
+			IndexTableName: srcIndexTable,
+		}},
+	}
+	dstDef := &plan.TableDef{
+		Name: "dst",
+		Indexes: []*plan.IndexDef{{
+			IndexName:      "ftidx",
+			IndexAlgo:      "fulltext",
+			IndexTableName: dstIndexTable,
+		}},
+	}
+	dstIndexDef := &plan.TableDef{
+		Name: dstIndexTable,
+		Cols: []*plan.ColDef{{
+			Name:   fakePK,
+			Hidden: true,
+			Typ:    plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+		}},
+		Pkey: &plan.PrimaryKeyDef{PkeyColName: fakePK},
+	}
+	createPlan := cloneCreatePlan(dstDef)
+	createPlan.GetDdl().GetCreateTable().IndexTables = []*plan.TableDef{dstIndexDef}
+
+	tc, err := constructTableClone(&Compile{proc: proc, pn: &plan.Plan{}}, &plan.CloneTable{
+		SrcTableDef: srcDef,
+		SrcObjDef:   &plan.ObjectRef{},
+		CreateTable: createPlan,
+		ScanSnapshot: &plan.Snapshot{
+			TS: &timestamp.Timestamp{PhysicalTime: 123},
+		},
+	})
+	require.NoError(t, err)
+	t.Cleanup(tc.Release)
+	require.Equal(t, table_clone.AutoIncrementState{
+		MaxValues: map[string]uint64{fakePK: 120},
+		Offsets:   map[string]uint64{fakePK: 200},
+	}, tc.Ctx.IndexAutoIncrStates["ftidx."])
+	require.Len(t, exec.sqls, 2)
+	require.Contains(t, exec.sqls[0], "reldatabase = "+sqlquote.String(srcDef.DbName))
+	require.Contains(t, exec.sqls[0], "relname = "+sqlquote.String(srcIndexTable))
+	require.Contains(t, exec.sqls[1], "from "+sqlquote.QualifiedIdent(srcDef.DbName, srcIndexTable))
+	for _, sql := range exec.sqls {
+		_, err := mysqlparser.ParseOne(context.Background(), sql, 1)
+		require.NoError(t, err, sql)
+	}
+}
+
+func TestConstructTableCloneCapturesPartitionHiddenIndexAllocators(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	const fakePK = "__mo_fake_pk_col"
+	exec := &tableCloneRecordingExecutor{}
+	exec.run = func(sql string) (executor.Result, error) {
+		switch {
+		case strings.Contains(sql, "mo_partition_tables"):
+			return newTableClonePartitionIndexResult(
+				t,
+				proc.Mp(),
+				[]string{"p0", "p1"},
+				[]string{"ftidx", "ftidx"},
+				[]string{"", ""},
+				[]string{"__mo_index_fulltext_src_p0", "__mo_index_fulltext_src_p1"},
+			), nil
+		case strings.Contains(sql, "mo_increment_columns"):
+			return newTableCloneNamedOffsetResult(t, proc.Mp(), fakePK, 200), nil
+		default:
+			return newTableCloneResult(t, proc.Mp(), 120), nil
+		}
+	}
+	runtime.ServiceRuntime(proc.GetService()).SetGlobalVariables(runtime.InternalSQLExecutor, exec)
+
+	srcDef := &plan.TableDef{
+		TblId:       7,
+		DbName:      "db",
+		Name:        "src",
+		FeatureFlag: features.Partitioned,
+		Indexes: []*plan.IndexDef{{
+			IndexName:      "ftidx",
+			IndexAlgo:      "fulltext",
+			IndexTableName: "__mo_index_fulltext_src",
+		}},
+	}
+	dstDef := &plan.TableDef{
+		Name: "dst",
+		Indexes: []*plan.IndexDef{{
+			IndexName:      "ftidx",
+			IndexAlgo:      "fulltext",
+			IndexTableName: "__mo_index_fulltext_dst",
+		}},
+	}
+	dstIndexDef := &plan.TableDef{
+		Name: "__mo_index_fulltext_dst",
+		Cols: []*plan.ColDef{{
+			Name:   fakePK,
+			Hidden: true,
+			Typ:    plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+		}},
+		Pkey: &plan.PrimaryKeyDef{PkeyColName: fakePK},
+	}
+	createPlan := cloneCreatePlan(dstDef)
+	createPlan.GetDdl().GetCreateTable().IndexTables = []*plan.TableDef{dstIndexDef}
+
+	tc, err := constructTableClone(&Compile{proc: proc, pn: &plan.Plan{}}, &plan.CloneTable{
+		SrcTableDef: srcDef,
+		SrcObjDef:   &plan.ObjectRef{},
+		CreateTable: createPlan,
+	})
+	require.NoError(t, err)
+	t.Cleanup(tc.Release)
+	want := table_clone.AutoIncrementState{
+		MaxValues: map[string]uint64{fakePK: 120},
+		Offsets:   map[string]uint64{fakePK: 200},
+	}
+	require.Equal(t, want, tc.Ctx.IndexAutoIncrStates["p0.ftidx."])
+	require.Equal(t, want, tc.Ctx.IndexAutoIncrStates["p1.ftidx."])
+	require.Equal(t, want, tc.Ctx.IndexAutoIncrStates["ftidx."])
+	require.Len(t, exec.sqls, 7)
+	for _, sql := range exec.sqls {
+		_, err := mysqlparser.ParseOne(context.Background(), sql, 1)
+		require.NoError(t, err, sql)
+	}
+}
+
+func TestConstructTableCloneHonorsCancellationAndClosesErrorResult(t *testing.T) {
+	t.Run("canceled before read", func(t *testing.T) {
+		proc := testutil.NewProcess(t)
+		ctx, cancel := context.WithCancel(proc.Ctx)
+		cancel()
+		proc.Ctx = ctx
+		exec := &tableCloneRecordingExecutor{result: newTableCloneResult(t, proc.Mp(), 40)}
+		runtime.ServiceRuntime(proc.GetService()).SetGlobalVariables(runtime.InternalSQLExecutor, exec)
+
+		_, err := constructTableClone(&Compile{proc: proc, pn: &plan.Plan{}}, &plan.CloneTable{
+			SrcTableDef: &plan.TableDef{Cols: []*plan.ColDef{{
+				Name: "id", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+			}}},
+			SrcObjDef: &plan.ObjectRef{},
+		})
+		require.ErrorIs(t, err, context.Canceled)
+		require.Empty(t, exec.sql)
+	})
+
+	t.Run("result returned with error", func(t *testing.T) {
+		proc := testutil.NewProcess(t)
+		result := newTableCloneResult(t, proc.Mp(), 40)
+		resultBatch := result.Batches[0]
+		wantErr := errors.New("allocator read failed")
+		exec := &tableCloneRecordingExecutor{result: result, err: wantErr}
+		runtime.ServiceRuntime(proc.GetService()).SetGlobalVariables(runtime.InternalSQLExecutor, exec)
+
+		_, err := constructTableClone(&Compile{proc: proc, pn: &plan.Plan{}}, &plan.CloneTable{
+			SrcTableDef: &plan.TableDef{Cols: []*plan.ColDef{{
+				Name: "id", Typ: plan.Type{Id: int32(types.T_uint64), AutoIncr: true},
+			}}},
+			SrcObjDef: &plan.ObjectRef{},
+		})
+		require.ErrorIs(t, err, wantErr)
+		require.Nil(t, resultBatch.Vecs)
+	})
+}
diff --git a/test/distributed/cases/auto_increment/auto_increment_columns.result b/test/distributed/cases/auto_increment/auto_increment_columns.result
index e5acd0049f57a..5e0e79b28f4ed 100644
--- a/test/distributed/cases/auto_increment/auto_increment_columns.result
+++ b/test/distributed/cases/auto_increment/auto_increment_columns.result
@@ -830,3 +830,20 @@ select last_insert_id();
 last_insert_id()
 10
 Drop table auto_increment02;
+drop table if exists issue_23143_copy_auto_increment;
+create table issue_23143_copy_auto_increment(id bigint auto_increment primary key, payload int) auto_increment = 1000;
+insert into issue_23143_copy_auto_increment(payload) values (1), (2);
+delete from issue_23143_copy_auto_increment where payload = 2;
+select count(*) = 1 as deleted_value_removed, min(id) = 1000 as requested_start_used from issue_23143_copy_auto_increment;
+deleted_value_removed    requested_start_used
+1    1
+alter table issue_23143_copy_auto_increment add column copied_payload varchar(20) default 'copied';
+insert into issue_23143_copy_auto_increment(payload, copied_payload) values (3, 'after-copy');
+select count(*) = 2 as copied_rows_preserved, max(id) > 1001 as source_allocator_preserved from issue_23143_copy_auto_increment;
+copied_rows_preserved    source_allocator_preserved
+1    1
+select payload, copied_payload from issue_23143_copy_auto_increment order by payload;
+payload    copied_payload
+1    copied
+3    after-copy
+drop table issue_23143_copy_auto_increment;
diff --git a/test/distributed/cases/auto_increment/auto_increment_columns.sql b/test/distributed/cases/auto_increment/auto_increment_columns.sql
index 75cc6c3794bdc..51ded7a2086ef 100644
--- a/test/distributed/cases/auto_increment/auto_increment_columns.sql
+++ b/test/distributed/cases/auto_increment/auto_increment_columns.sql
@@ -479,3 +479,16 @@ Insert into auto_increment02 values(10);
 Select * from auto_increment02;
 select last_insert_id();
 Drop table auto_increment02;
+
+-- issue #23143 prerequisite: COPY ALTER must preserve the source allocator's
+-- reserved range, not rebuild it only from the copied row maximum.
+drop table if exists issue_23143_copy_auto_increment;
+create table issue_23143_copy_auto_increment(id bigint auto_increment primary key, payload int) auto_increment = 1000;
+insert into issue_23143_copy_auto_increment(payload) values (1), (2);
+delete from issue_23143_copy_auto_increment where payload = 2;
+select count(*) = 1 as deleted_value_removed, min(id) = 1000 as requested_start_used from issue_23143_copy_auto_increment;
+alter table issue_23143_copy_auto_increment add column copied_payload varchar(20) default 'copied';
+insert into issue_23143_copy_auto_increment(payload, copied_payload) values (3, 'after-copy');
+select count(*) = 2 as copied_rows_preserved, max(id) > 1001 as source_allocator_preserved from issue_23143_copy_auto_increment;
+select payload, copied_payload from issue_23143_copy_auto_increment order by payload;
+drop table issue_23143_copy_auto_increment;
diff --git a/test/distributed/cases/git4data/clone/table_clone.result b/test/distributed/cases/git4data/clone/table_clone.result
index c926dc9b042b0..0fe52bc66c62c 100644
--- a/test/distributed/cases/git4data/clone/table_clone.result
+++ b/test/distributed/cases/git4data/clone/table_clone.result
@@ -163,6 +163,10 @@ id    content
 1    this is a test for clone fulltext table 1
 2    this is a test for clone fulltext table 2
 3    this is a test for clone fulltext table 3
+insert into db1.t8_copy(content) values ("this is a test for clone fulltext table 4");
+select content from db1.t8_copy where content = "this is a test for clone fulltext table 4";
+content
+this is a test for clone fulltext table 4
 create table db1.t9(a int primary key, b vecf32(3));
 insert into db1.t9 values(1, "[1,2,3]");
 insert into db1.t9 values(2, "[1,2,4]");
diff --git a/test/distributed/cases/git4data/clone/table_clone.sql b/test/distributed/cases/git4data/clone/table_clone.sql
index 1a092426f1174..9f4394978b03b 100644
--- a/test/distributed/cases/git4data/clone/table_clone.sql
+++ b/test/distributed/cases/git4data/clone/table_clone.sql
@@ -92,6 +92,8 @@ insert into db1.t8(content) values ("this is a test for clone fulltext table 3")
 
 create table db1.t8_copy clone db1.t8;
 select * from db1.t8_copy order by id asc;
+insert into db1.t8_copy(content) values ("this is a test for clone fulltext table 4");
+select content from db1.t8_copy where content = "this is a test for clone fulltext table 4";
 
 
 create table db1.t9(a int primary key, b vecf32(3));
@@ -108,4 +110,4 @@ drop snapshot if exists sp0;
 drop snapshot if exists sp1;
 drop account if exists acc1;
 drop account if exists acc2;
-drop database if exists db1;
\ No newline at end of file
+drop database if exists db1;
diff --git a/test/distributed/cases/vector/vector_index.result b/test/distributed/cases/vector/vector_index.result
index f56107a0c0195..49ecbbbb6adb2 100644
--- a/test/distributed/cases/vector/vector_index.result
+++ b/test/distributed/cases/vector/vector_index.result
@@ -170,15 +170,15 @@ alter table vector_index_08 rename column d to e;
 select * from vector_index_08 where a>9775 order by  L2_DISTANCE(e,"[8.555,2.11,7.22]") desc limit 2;
 a    b    c    e
 9776    [10, 3, 8, 5, 48, 26, 5, 16, 17, 0, 0, 2, 132, 53, 1, 16, 112, 6, 0, 0, 7, 2, 1, 48, 48, 15, 18, 31, 3, 0, 0, 9, 6, 10, 19, 27, 50, 46, 17, 9, 18, 1, 4, 48, 132, 23, 3, 5, 132, 9, 4, 3, 11, 0, 2, 46, 84, 12, 10, 10, 1, 0, 12, 76, 26, 22, 16, 26, 35, 15, 3, 16, 15, 1, 51, 132, 125, 8, 1, 2, 132, 51, 67, 91, 8, 0, 0, 30, 126, 39, 32, 38, 4, 0, 1, 12, 24, 2, 2, 2, 4, 7, 2, 19, 93, 19, 70, 92, 2, 3, 1, 21, 36, 58, 132, 94, 0, 0, 0, 0, 21, 25, 57, 48, 1, 0, 0, 1]    3    [0, 0, 0]
-9777    null    null    [2.36, 5.021, 9.222]
+20001    null    null    [2.36, 5.021, 9.222]
 alter table vector_index_08 drop column e;
 select * from vector_index_08;
 a    b    c
 9774    [1, 0, 1, 6, 6, 17, 47, 39, 2, 0, 1, 25, 27, 10, 56, 130, 18, 5, 2, 6, 15, 2, 19, 130, 42, 28, 1, 1, 2, 1, 0, 5, 0, 2, 4, 4, 31, 34, 44, 35, 9, 3, 8, 11, 33, 12, 61, 130, 130, 17, 0, 1, 6, 2, 9, 130, 111, 36, 0, 0, 11, 9, 1, 12, 2, 100, 130, 28, 7, 2, 6, 7, 9, 27, 130, 83, 5, 0, 1, 18, 130, 130, 84, 9, 0, 0, 2, 24, 111, 24, 0, 1, 37, 24, 2, 10, 12, 62, 33, 3, 0, 0, 0, 1, 3, 16, 106, 28, 0, 0, 0, 0, 17, 46, 85, 10, 0, 0, 1, 4, 11, 4, 2, 2, 9, 14, 8, 8]    3
 9775    [0, 1, 1, 3, 0, 3, 46, 20, 1, 4, 17, 9, 1, 17, 108, 15, 0, 3, 37, 17, 6, 15, 116, 16, 6, 1, 4, 7, 7, 7, 9, 6, 0, 8, 10, 4, 26, 129, 27, 9, 0, 0, 5, 2, 11, 129, 129, 12, 103, 4, 0, 0, 2, 31, 129, 129, 94, 4, 0, 0, 0, 3, 13, 42, 0, 15, 38, 2, 70, 129, 1, 0, 5, 10, 40, 12, 74, 129, 6, 1, 129, 39, 6, 1, 2, 22, 9, 33, 122, 13, 0, 0, 0, 0, 5, 23, 4, 11, 9, 12, 45, 38, 1, 0, 0, 4, 36, 38, 57, 32, 0, 0, 82, 22, 9, 5, 13, 11, 3, 94, 35, 3, 0, 0, 0, 1, 16, 97]    5
 9776    [10, 3, 8, 5, 48, 26, 5, 16, 17, 0, 0, 2, 132, 53, 1, 16, 112, 6, 0, 0, 7, 2, 1, 48, 48, 15, 18, 31, 3, 0, 0, 9, 6, 10, 19, 27, 50, 46, 17, 9, 18, 1, 4, 48, 132, 23, 3, 5, 132, 9, 4, 3, 11, 0, 2, 46, 84, 12, 10, 10, 1, 0, 12, 76, 26, 22, 16, 26, 35, 15, 3, 16, 15, 1, 51, 132, 125, 8, 1, 2, 132, 51, 67, 91, 8, 0, 0, 30, 126, 39, 32, 38, 4, 0, 1, 12, 24, 2, 2, 2, 4, 7, 2, 19, 93, 19, 70, 92, 2, 3, 1, 21, 36, 58, 132, 94, 0, 0, 0, 0, 21, 25, 57, 48, 1, 0, 0, 1]    3
-9777    null    null
-9778    null    null
+20001    null    null
+20002    null    null
 create table vector_index_09(a int primary key, b vecf32(128),c int,key c_k(c));
 create index idx01 using ivfflat on vector_index_09(b) lists=3 op_type "vector_l2_ops";
 insert into vector_index_09 values(9774 ,NULL,3),(9775,NULL,10);