diff --git a/br/pkg/lightning/backend/kv/allocator.go b/br/pkg/lightning/backend/kv/allocator.go
index 6792851ef9b06..38f6b45050bc6 100644
--- a/br/pkg/lightning/backend/kv/allocator.go
+++ b/br/pkg/lightning/backend/kv/allocator.go
@@ -31,6 +31,9 @@ type panickingAllocator struct {
 }
 
 // NewPanickingAllocators creates a PanickingAllocator shared by all allocation types.
+// we use this to collect the max id(either _tidb_rowid or auto_increment id or auto_random) used
+// during import, and we will use this info to do ALTER TABLE xxx AUTO_RANDOM_BASE or AUTO_INCREMENT
+// on post-process phase.
 func NewPanickingAllocators(base int64) autoid.Allocators {
 	sharedBase := &base
 	return autoid.NewAllocators(
diff --git a/br/pkg/lightning/backend/kv/base.go b/br/pkg/lightning/backend/kv/base.go
index a9185bc48166a..6ac223f764506 100644
--- a/br/pkg/lightning/backend/kv/base.go
+++ b/br/pkg/lightning/backend/kv/base.go
@@ -214,6 +214,7 @@ func (e *BaseKVEncoder) ProcessColDatum(col *table.Column, rowID int64, inputDat
 		}
 	}
 	if IsAutoIncCol(col.ToInfo()) {
+		// same as RowIDAllocType, since SepAutoInc is always false when initializing allocators of Table.
 		alloc := e.Table.Allocators(e.SessionCtx).Get(autoid.AutoIncrementType)
 		if err := alloc.Rebase(context.Background(), GetAutoRecordID(value, &col.FieldType), false); err != nil {
 			return value, errors.Trace(err)
diff --git a/br/pkg/lightning/common/BUILD.bazel b/br/pkg/lightning/common/BUILD.bazel
index c44c08d9b3a6c..15cc00048753d 100644
--- a/br/pkg/lightning/common/BUILD.bazel
+++ b/br/pkg/lightning/common/BUILD.bazel
@@ -90,6 +90,7 @@ go_test(
     name = "common_test",
     timeout = "short",
     srcs = [
+        "common_test.go",
         "errors_test.go",
         "main_test.go",
         "once_error_test.go",
@@ -101,15 +102,23 @@ go_test(
     ],
     embed = [":common"],
     flaky = True,
-    shard_count = 18,
+    shard_count = 19,
     deps = [
         "//br/pkg/errors",
         "//br/pkg/lightning/log",
+        "//ddl",
         "//errno",
+        "//kv",
+        "//meta",
+        "//meta/autoid",
         "//parser",
+        "//parser/ast",
+        "//parser/model",
         "//store/driver/error",
+        "//store/mockstore",
         "//testkit/testsetup",
         "//util/dbutil",
+        "//util/mock",
         "@com_github_data_dog_go_sqlmock//:go-sqlmock",
         "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_pingcap_errors//:errors",
diff --git a/br/pkg/lightning/common/common.go b/br/pkg/lightning/common/common.go
index aaf8860e4fb58..4b2f6698d80f0 100644
--- a/br/pkg/lightning/common/common.go
+++ b/br/pkg/lightning/common/common.go
@@ -52,24 +52,42 @@ var DefaultImportVariablesTiDB = map[string]string{
 // AllocGlobalAutoID allocs N consecutive autoIDs from TiDB.
 func AllocGlobalAutoID(ctx context.Context, n int64, store kv.Storage, dbID int64,
 	tblInfo *model.TableInfo) (autoIDBase, autoIDMax int64, err error) {
-	alloc, err := getGlobalAutoIDAlloc(store, dbID, tblInfo)
+	allocators, err := GetGlobalAutoIDAlloc(store, dbID, tblInfo)
 	if err != nil {
 		return 0, 0, err
 	}
-	return alloc.Alloc(ctx, uint64(n), 1, 1)
+	// there might be 2 allocators when tblInfo.SepAutoInc is true, and in this case
+	// RowIDAllocType will be the last one.
+	// we return the value of last Alloc as autoIDBase and autoIDMax, i.e. the value
+	// either comes from RowIDAllocType or AutoRandomType.
+	for _, alloc := range allocators {
+		autoIDBase, autoIDMax, err = alloc.Alloc(ctx, uint64(n), 1, 1)
+		if err != nil {
+			return 0, 0, err
+		}
+	}
+	return
 }
 
 // RebaseGlobalAutoID rebase the autoID base to newBase.
 func RebaseGlobalAutoID(ctx context.Context, newBase int64, store kv.Storage, dbID int64,
 	tblInfo *model.TableInfo) error {
-	alloc, err := getGlobalAutoIDAlloc(store, dbID, tblInfo)
+	allocators, err := GetGlobalAutoIDAlloc(store, dbID, tblInfo)
 	if err != nil {
 		return err
 	}
-	return alloc.Rebase(ctx, newBase, false)
+	for _, alloc := range allocators {
+		err = alloc.Rebase(ctx, newBase, false)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
-func getGlobalAutoIDAlloc(store kv.Storage, dbID int64, tblInfo *model.TableInfo) (autoid.Allocator, error) {
+// GetGlobalAutoIDAlloc returns the autoID allocators for a table.
+// export it for testing.
+func GetGlobalAutoIDAlloc(store kv.Storage, dbID int64, tblInfo *model.TableInfo) ([]autoid.Allocator, error) {
 	if store == nil {
 		return nil, errors.New("internal error: kv store should not be nil")
 	}
@@ -89,20 +107,29 @@ func getGlobalAutoIDAlloc(store kv.Storage, dbID int64, tblInfo *model.TableInfo
 	hasAutoIncID := tblInfo.GetAutoIncrementColInfo() != nil
 	hasAutoRandID := tblInfo.ContainsAutoRandomBits()
 
-	// Current TiDB has some limitations for auto ID.
+	// TiDB version <= 6.4.0 has some limitations for auto ID.
 	// 1. Auto increment ID and auto row ID are using the same RowID allocator.
 	//    See https://github.com/pingcap/tidb/issues/982.
 	// 2. Auto random column must be a clustered primary key. That is to say,
 	//    there is no implicit row ID for tables with auto random column.
 	// 3. There is at most one auto column in a table.
 	// Therefore, we assume there is only one auto column in a table and use RowID allocator if possible.
+	//
+	// Since TiDB 6.5.0, row ID and auto ID are using different allocators when tblInfo.SepAutoInc is true
 	switch {
 	case hasRowID || hasAutoIncID:
-		return autoid.NewAllocator(store, dbID, tblInfo.ID, tblInfo.IsAutoIncColUnsigned(),
-			autoid.RowIDAllocType, noCache, tblVer), nil
+		allocators := make([]autoid.Allocator, 0, 2)
+		if tblInfo.SepAutoInc() && hasAutoIncID {
+			allocators = append(allocators, autoid.NewAllocator(store, dbID, tblInfo.ID, tblInfo.IsAutoIncColUnsigned(),
+				autoid.AutoIncrementType, noCache, tblVer))
+		}
+		// this allocator is NOT used when SepAutoInc=true and auto increment column is clustered.
+		allocators = append(allocators, autoid.NewAllocator(store, dbID, tblInfo.ID, tblInfo.IsAutoIncColUnsigned(),
+			autoid.RowIDAllocType, noCache, tblVer))
+		return allocators, nil
 	case hasAutoRandID:
-		return autoid.NewAllocator(store, dbID, tblInfo.ID, tblInfo.IsAutoRandomBitColUnsigned(),
-			autoid.AutoRandomType, noCache, tblVer), nil
+		return []autoid.Allocator{autoid.NewAllocator(store, dbID, tblInfo.ID, tblInfo.IsAutoRandomBitColUnsigned(),
+			autoid.AutoRandomType, noCache, tblVer)}, nil
 	default:
 		return nil, errors.Errorf("internal error: table %s has no auto ID", tblInfo.Name)
 	}
diff --git a/br/pkg/lightning/common/common_test.go b/br/pkg/lightning/common/common_test.go
new file mode 100644
index 0000000000000..ef9b74ab7486f
--- /dev/null
+++ b/br/pkg/lightning/common/common_test.go
@@ -0,0 +1,161 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/br/pkg/lightning/common"
+	"github.com/pingcap/tidb/ddl"
+	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/meta"
+	"github.com/pingcap/tidb/meta/autoid"
+	"github.com/pingcap/tidb/parser"
+	"github.com/pingcap/tidb/parser/ast"
+	"github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/store/mockstore"
+	tmock "github.com/pingcap/tidb/util/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func newTableInfo(t *testing.T,
+	dbID, tableID int64,
+	createTableSql string, kvStore kv.Storage,
+) *model.TableInfo {
+	p := parser.New()
+	se := tmock.NewContext()
+
+	node, err := p.ParseOneStmt(createTableSql, "utf8mb4", "utf8mb4_bin")
+	require.NoError(t, err)
+	tableInfo, err := ddl.MockTableInfo(se, node.(*ast.CreateTableStmt), tableID)
+	require.NoError(t, err)
+	tableInfo.State = model.StatePublic
+
+	ctx := kv.WithInternalSourceType(context.Background(), "test")
+	err = kv.RunInNewTxn(ctx, kvStore, false, func(ctx context.Context, txn kv.Transaction) error {
+		m := meta.NewMeta(txn)
+		if err := m.CreateDatabase(&model.DBInfo{ID: dbID}); err != nil && !errors.ErrorEqual(err, meta.ErrDBExists) {
+			return err
+		}
+		return m.CreateTableOrView(dbID, tableInfo)
+	})
+	require.NoError(t, err)
+	return tableInfo
+}
+
+func TestAllocGlobalAutoID(t *testing.T) {
+	storePath := t.TempDir()
+	kvStore, err := mockstore.NewMockStore(mockstore.WithPath(storePath))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, kvStore.Close())
+	})
+
+	cases := []struct {
+		tableID              int64
+		createTableSQL       string
+		expectErrStr         string
+		expectAllocatorTypes []autoid.AllocatorType
+	}{
+		// autoID, autoIncrID = false, false
+		{
+			tableID:              11,
+			createTableSQL:       "create table t11 (a int primary key clustered)",
+			expectErrStr:         "has no auto ID",
+			expectAllocatorTypes: nil,
+		},
+		{
+			tableID:              12,
+			createTableSQL:       "create table t12 (a int primary key clustered) AUTO_ID_CACHE 1",
+			expectErrStr:         "has no auto ID",
+			expectAllocatorTypes: nil,
+		},
+		// autoID, autoIncrID = true, false
+		{
+			tableID:              21,
+			createTableSQL:       "create table t21 (a int)",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.RowIDAllocType},
+		},
+		{
+			tableID:              22,
+			createTableSQL:       "create table t22 (a int) AUTO_ID_CACHE 1",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.RowIDAllocType},
+		},
+		// autoID, autoIncrID = false, true
+		{
+			tableID:              31,
+			createTableSQL:       "create table t31 (a int primary key clustered auto_increment)",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.RowIDAllocType},
+		},
+		{
+			tableID:              32,
+			createTableSQL:       "create table t32 (a int primary key clustered auto_increment) AUTO_ID_CACHE 1",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.AutoIncrementType, autoid.RowIDAllocType},
+		},
+		// autoID, autoIncrID = true, true
+		{
+			tableID:              41,
+			createTableSQL:       "create table t41 (a int primary key nonclustered auto_increment)",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.RowIDAllocType},
+		},
+		{
+			tableID:              42,
+			createTableSQL:       "create table t42 (a int primary key nonclustered auto_increment) AUTO_ID_CACHE 1",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.AutoIncrementType, autoid.RowIDAllocType},
+		},
+		// autoRandomID
+		{
+			tableID:              51,
+			createTableSQL:       "create table t51 (a bigint primary key auto_random)",
+			expectErrStr:         "",
+			expectAllocatorTypes: []autoid.AllocatorType{autoid.AutoRandomType},
+		},
+	}
+	ctx := context.Background()
+	for _, c := range cases {
+		ti := newTableInfo(t, 1, c.tableID, c.createTableSQL, kvStore)
+		allocators, err := common.GetGlobalAutoIDAlloc(kvStore, 1, ti)
+		if c.expectErrStr == "" {
+			require.NoError(t, err, c.tableID)
+			require.NoError(t, common.RebaseGlobalAutoID(ctx, 123, kvStore, 1, ti))
+			base, idMax, err := common.AllocGlobalAutoID(ctx, 100, kvStore, 1, ti)
+			require.NoError(t, err, c.tableID)
+			require.Equal(t, int64(123), base, c.tableID)
+			require.Equal(t, int64(223), idMax, c.tableID)
+			// all allocators are rebased and allocated
+			for _, alloc := range allocators {
+				base2, max2, err := alloc.Alloc(ctx, 100, 1, 1)
+				require.NoError(t, err, c.tableID)
+				require.Equal(t, int64(223), base2, c.tableID)
+				require.Equal(t, int64(323), max2, c.tableID)
+			}
+		} else {
+			require.ErrorContains(t, err, c.expectErrStr, c.tableID)
+		}
+		var allocatorTypes []autoid.AllocatorType
+		for _, alloc := range allocators {
+			allocatorTypes = append(allocatorTypes, alloc.GetType())
+		}
+		require.Equal(t, c.expectAllocatorTypes, allocatorTypes, c.tableID)
+	}
+}
diff --git a/br/pkg/lightning/importer/meta_manager_test.go b/br/pkg/lightning/importer/meta_manager_test.go
index 8459c8f4d04ac..bbccb3ec5aead 100644
--- a/br/pkg/lightning/importer/meta_manager_test.go
+++ b/br/pkg/lightning/importer/meta_manager_test.go
@@ -35,43 +35,45 @@ type metaMgrSuite struct {
 	checksumMgr *testChecksumMgr
 }
 
-func newTableRestore(t *testing.T, kvStore kv.Storage) *TableImporter {
+func newTableRestore(t *testing.T,
+	db, table string,
+	dbID, tableID int64,
+	createTableSQL string, kvStore kv.Storage,
+) *TableImporter {
 	p := parser.New()
 	se := tmock.NewContext()
 
-	node, err := p.ParseOneStmt("CREATE TABLE `t1` (`c1` varchar(5) NOT NULL)", "utf8mb4", "utf8mb4_bin")
+	node, err := p.ParseOneStmt(createTableSQL, "utf8mb4", "utf8mb4_bin")
 	require.NoError(t, err)
-	tableInfo, err := ddl.MockTableInfo(se, node.(*ast.CreateTableStmt), int64(1))
+	tableInfo, err := ddl.MockTableInfo(se, node.(*ast.CreateTableStmt), tableID)
 	require.NoError(t, err)
 	tableInfo.State = model.StatePublic
 
-	schema := "test"
-	tb := "t1"
 	ti := &checkpoints.TidbTableInfo{
 		ID:   tableInfo.ID,
-		DB:   schema,
-		Name: tb,
+		DB:   db,
+		Name: table,
 		Core: tableInfo,
 	}
 	dbInfo := &checkpoints.TidbDBInfo{
-		ID:   1,
-		Name: schema,
+		ID:   dbID,
+		Name: db,
 		Tables: map[string]*checkpoints.TidbTableInfo{
-			tb: ti,
+			table: ti,
 		},
 	}
 
 	ctx := kv.WithInternalSourceType(context.Background(), "test")
 	err = kv.RunInNewTxn(ctx, kvStore, false, func(ctx context.Context, txn kv.Transaction) error {
 		m := meta.NewMeta(txn)
-		if err := m.CreateDatabase(&model.DBInfo{ID: dbInfo.ID}); err != nil {
+		if err := m.CreateDatabase(&model.DBInfo{ID: dbInfo.ID}); err != nil && !errors.ErrorEqual(err, meta.ErrDBExists) {
 			return err
 		}
 		return m.CreateTableOrView(dbInfo.ID, ti.Core)
 	})
 	require.NoError(t, err)
 
-	tableName := common.UniqueTable(schema, tb)
+	tableName := common.UniqueTable(db, table)
 	logger := log.With(zap.String("table", tableName))
 
 	return &TableImporter{
@@ -93,9 +95,10 @@ func newMetaMgrSuite(t *testing.T) *metaMgrSuite {
 
 	var s metaMgrSuite
 	s.mgr = &dbTableMetaMgr{
-		session:      db,
-		taskID:       1,
-		tr:           newTableRestore(t, kvStore),
+		session: db,
+		taskID:  1,
+		tr: newTableRestore(t, "test", "t1", 1, 1,
+			"CREATE TABLE `t1` (`c1` varchar(5) NOT NULL)", kvStore),
 		tableName:    common.UniqueTable("test", TableMetaTableName),
 		needChecksum: true,
 	}
diff --git a/br/pkg/lightning/importer/table_import.go b/br/pkg/lightning/importer/table_import.go
index 25ddc004dde59..15de04f004788 100644
--- a/br/pkg/lightning/importer/table_import.go
+++ b/br/pkg/lightning/importer/table_import.go
@@ -883,8 +883,21 @@ func (tr *TableImporter) postProcess(
 			maxCap := shardFmt.IncrementalBitsCapacity()
 			err = AlterAutoRandom(ctx, rc.db, tr.tableName, uint64(tr.alloc.Get(autoid.AutoRandomType).Base())+1, maxCap)
 		} else if common.TableHasAutoRowID(tblInfo) || tblInfo.GetAutoIncrementColInfo() != nil {
-			// only alter auto increment id iff table contains auto-increment column or generated handle
-			err = AlterAutoIncrement(ctx, rc.db, tr.tableName, uint64(tr.alloc.Get(autoid.RowIDAllocType).Base())+1)
+			// only alter auto increment id iff table contains auto-increment column or generated handle.
+			// ALTER TABLE xxx AUTO_INCREMENT = yyy has a bad naming.
+			// if a table has implicit _tidb_rowid column & tbl.SepAutoID=false, then it works on _tidb_rowid
+			// allocator, even if the table has NO auto-increment column.
+			newBase := uint64(tr.alloc.Get(autoid.RowIDAllocType).Base()) + 1
+			err = AlterAutoIncrement(ctx, rc.db, tr.tableName, newBase)
+
+			if err == nil && isLocalBackend(rc.cfg) {
+				// for TiDB version >= 6.5.0, a table might have separate allocators for auto_increment column and _tidb_rowid,
+				// especially when a table has auto_increment non-clustered PK, it will use both allocators.
+				// And in this case, ALTER TABLE xxx AUTO_INCREMENT = xxx only works on the allocator of auto_increment column,
+				// not for allocator of _tidb_rowid.
+				// So we need to rebase IDs for those 2 allocators explicitly.
+				err = common.RebaseGlobalAutoID(ctx, adjustIDBase(newBase), tr.kvStore, tr.dbInfo.ID, tr.tableInfo.Core)
+			}
 		}
 		rc.alterTableLock.Unlock()
 		saveCpErr := rc.saveStatusCheckpoint(ctx, tr.tableName, checkpoints.WholeTableEngineID, err, checkpoints.CheckpointStatusAlteredAutoInc)
diff --git a/br/pkg/lightning/importer/tidb.go b/br/pkg/lightning/importer/tidb.go
index ec57e0e3b8349..0b6174896eb3e 100644
--- a/br/pkg/lightning/importer/tidb.go
+++ b/br/pkg/lightning/importer/tidb.go
@@ -302,19 +302,18 @@ func ObtainNewCollationEnabled(ctx context.Context, db *sql.DB) (bool, error) {
 // AlterAutoIncrement rebase the table auto increment id
 //
 // NOTE: since tidb can make sure the auto id is always be rebase even if the `incr` value is smaller
-// the the auto incremanet base in tidb side, we needn't fetch currently auto increment value here.
+// than the auto increment base in tidb side, we needn't fetch currently auto increment value here.
 // See: https://github.com/pingcap/tidb/blob/64698ef9a3358bfd0fdc323996bb7928a56cadca/ddl/ddl_api.go#L2528-L2533
 func AlterAutoIncrement(ctx context.Context, db *sql.DB, tableName string, incr uint64) error {
-	var query string
 	logger := log.FromContext(ctx).With(zap.String("table", tableName), zap.Uint64("auto_increment", incr))
+	base := adjustIDBase(incr)
+	var forceStr string
 	if incr > math.MaxInt64 {
 		// automatically set max value
 		logger.Warn("auto_increment out of the maximum value TiDB supports, automatically set to the max", zap.Uint64("auto_increment", incr))
-		incr = math.MaxInt64
-		query = fmt.Sprintf("ALTER TABLE %s FORCE AUTO_INCREMENT=%d", tableName, incr)
-	} else {
-		query = fmt.Sprintf("ALTER TABLE %s AUTO_INCREMENT=%d", tableName, incr)
+		forceStr = "FORCE"
 	}
+	query := fmt.Sprintf("ALTER TABLE %s %s AUTO_INCREMENT=%d", tableName, forceStr, base)
 	task := logger.Begin(zap.InfoLevel, "alter table auto_increment")
 	exec := common.SQLWithRetry{DB: db, Logger: logger}
 	err := exec.Exec(ctx, "alter table auto_increment", query)
@@ -328,6 +327,13 @@ func AlterAutoIncrement(ctx context.Context, db *sql.DB, tableName string, incr
 	return errors.Annotatef(err, "%s", query)
 }
 
+func adjustIDBase(incr uint64) int64 {
+	if incr > math.MaxInt64 {
+		return math.MaxInt64
+	}
+	return int64(incr)
+}
+
 // AlterAutoRandom rebase the table auto random id
 func AlterAutoRandom(ctx context.Context, db *sql.DB, tableName string, randomBase uint64, maxAutoRandom uint64) error {
 	logger := log.FromContext(ctx).With(zap.String("table", tableName), zap.Uint64("auto_random", randomBase))
@@ -339,6 +345,7 @@ func AlterAutoRandom(ctx context.Context, db *sql.DB, tableName string, randomBa
 		logger.Warn("auto_random out of the maximum value TiDB supports")
 		return nil
 	}
+	// if new base is smaller than current, this query will success with a warning
 	query := fmt.Sprintf("ALTER TABLE %s AUTO_RANDOM_BASE=%d", tableName, randomBase)
 	task := logger.Begin(zap.InfoLevel, "alter table auto_random")
 	exec := common.SQLWithRetry{DB: db, Logger: logger}
diff --git a/br/tests/lightning_csv/data/auto_incr_id-schema-create.sql b/br/tests/lightning_csv/data/auto_incr_id-schema-create.sql
new file mode 100644
index 0000000000000..fdfcf79b5747f
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id-schema-create.sql
@@ -0,0 +1 @@
+CREATE DATABASE `auto_incr_id`;
diff --git a/br/tests/lightning_csv/data/auto_incr_id.clustered-schema.sql b/br/tests/lightning_csv/data/auto_incr_id.clustered-schema.sql
new file mode 100644
index 0000000000000..de9eaddf6ead7
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.clustered-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `clustered` (
+    `id` int(11) NOT NULL AUTO_INCREMENT,
+    v int,
+    PRIMARY KEY(`id`) CLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
diff --git a/br/tests/lightning_csv/data/auto_incr_id.clustered.0.csv b/br/tests/lightning_csv/data/auto_incr_id.clustered.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.clustered.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/auto_incr_id.clustered_cache1-schema.sql b/br/tests/lightning_csv/data/auto_incr_id.clustered_cache1-schema.sql
new file mode 100644
index 0000000000000..ab6a3b33f34e8
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.clustered_cache1-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `clustered_cache1` (
+    `id` int(11) NOT NULL AUTO_INCREMENT,
+    v int,
+    PRIMARY KEY(`id`) CLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin AUTO_ID_CACHE 1;
diff --git a/br/tests/lightning_csv/data/auto_incr_id.clustered_cache1.0.csv b/br/tests/lightning_csv/data/auto_incr_id.clustered_cache1.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.clustered_cache1.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/auto_incr_id.nonclustered-schema.sql b/br/tests/lightning_csv/data/auto_incr_id.nonclustered-schema.sql
new file mode 100644
index 0000000000000..178a0bf6fa457
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.nonclustered-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `nonclustered` (
+    `id` int(11) NOT NULL AUTO_INCREMENT,
+    v int,
+    PRIMARY KEY(`id`) NONCLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
diff --git a/br/tests/lightning_csv/data/auto_incr_id.nonclustered.0.csv b/br/tests/lightning_csv/data/auto_incr_id.nonclustered.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.nonclustered.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/auto_incr_id.nonclustered_cache1-schema.sql b/br/tests/lightning_csv/data/auto_incr_id.nonclustered_cache1-schema.sql
new file mode 100644
index 0000000000000..494af711683ce
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.nonclustered_cache1-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `nonclustered_cache1` (
+    `id` int(11) NOT NULL AUTO_INCREMENT,
+    v int,
+    PRIMARY KEY(`id`) NONCLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin AUTO_ID_CACHE 1;
diff --git a/br/tests/lightning_csv/data/auto_incr_id.nonclustered_cache1.0.csv b/br/tests/lightning_csv/data/auto_incr_id.nonclustered_cache1.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/auto_incr_id.nonclustered_cache1.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id-schema-create.sql b/br/tests/lightning_csv/data/no_auto_incr_id-schema-create.sql
new file mode 100644
index 0000000000000..03f54790ef319
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id-schema-create.sql
@@ -0,0 +1 @@
+CREATE DATABASE `no_auto_incr_id`;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.clustered-schema.sql b/br/tests/lightning_csv/data/no_auto_incr_id.clustered-schema.sql
new file mode 100644
index 0000000000000..b4da35f5678e8
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.clustered-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `clustered` (
+    `id` int(11) NOT NULL,
+    v int,
+    PRIMARY KEY(`id`) CLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.clustered.0.csv b/br/tests/lightning_csv/data/no_auto_incr_id.clustered.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.clustered.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.clustered_cache1-schema.sql b/br/tests/lightning_csv/data/no_auto_incr_id.clustered_cache1-schema.sql
new file mode 100644
index 0000000000000..25300a9f47512
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.clustered_cache1-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `clustered_cache1` (
+    `id` int(11) NOT NULL,
+    v int,
+    PRIMARY KEY(`id`) CLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin AUTO_ID_CACHE 1;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.clustered_cache1.0.csv b/br/tests/lightning_csv/data/no_auto_incr_id.clustered_cache1.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.clustered_cache1.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.no_pk-schema.sql b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk-schema.sql
new file mode 100644
index 0000000000000..02b79ed946810
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk-schema.sql
@@ -0,0 +1,4 @@
+CREATE TABLE `no_pk` (
+    `id` int(11) NOT NULL,
+    v int
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.no_pk.0.csv b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.no_pk_cache1-schema.sql b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk_cache1-schema.sql
new file mode 100644
index 0000000000000..eaf1fb5fa920c
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk_cache1-schema.sql
@@ -0,0 +1,4 @@
+CREATE TABLE `no_pk_cache1` (
+    `id` int(11) NOT NULL,
+    v int
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin AUTO_ID_CACHE 1;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.no_pk_cache1.0.csv b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk_cache1.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.no_pk_cache1.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered-schema.sql b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered-schema.sql
new file mode 100644
index 0000000000000..5b65440df3587
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `nonclustered` (
+    `id` int(11) NOT NULL,
+    v int,
+    PRIMARY KEY(`id`) NONCLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered.0.csv b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered_cache1-schema.sql b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered_cache1-schema.sql
new file mode 100644
index 0000000000000..8618402b8bb0e
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered_cache1-schema.sql
@@ -0,0 +1,5 @@
+CREATE TABLE `nonclustered_cache1` (
+    `id` int(11) NOT NULL,
+    v int,
+    PRIMARY KEY(`id`) NONCLUSTERED
+)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin AUTO_ID_CACHE 1;
diff --git a/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered_cache1.0.csv b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered_cache1.0.csv
new file mode 100644
index 0000000000000..8db79c216f250
--- /dev/null
+++ b/br/tests/lightning_csv/data/no_auto_incr_id.nonclustered_cache1.0.csv
@@ -0,0 +1,4 @@
+4,1
+5,2
+6,3
+
diff --git a/br/tests/lightning_csv/run.sh b/br/tests/lightning_csv/run.sh
index 2f35fd9eb7950..82d645cffd0b7 100755
--- a/br/tests/lightning_csv/run.sh
+++ b/br/tests/lightning_csv/run.sh
@@ -10,6 +10,8 @@ function run_with() {
   fi
 
   run_sql 'DROP DATABASE IF EXISTS csv'
+  run_sql 'DROP DATABASE IF EXISTS auto_incr_id'
+  run_sql 'DROP DATABASE IF EXISTS no_auto_incr_id'
 
   run_lightning --backend $backend --config $config_file
 
@@ -41,6 +43,24 @@ function run_with() {
   check_contains 'id: 3'
   run_sql 'SELECT id FROM csv.empty_strings WHERE b <> ""'
   check_not_contains 'id:'
+
+  for table in clustered nonclustered clustered_cache1 nonclustered_cache1; do
+    run_sql "select count(*) from auto_incr_id.$table"
+    check_contains 'count(*): 3'
+    # insert should work
+    run_sql "insert into auto_incr_id.$table(v) values(1)"
+    run_sql "select count(*) from auto_incr_id.$table"
+    check_contains 'count(*): 4'
+  done
+
+  for table in clustered nonclustered clustered_cache1 nonclustered_cache1 no_pk no_pk_cache1; do
+    run_sql "select count(*) from no_auto_incr_id.$table"
+    check_contains 'count(*): 3'
+    # insert should work
+    run_sql "insert into no_auto_incr_id.$table values(1, 1)"
+    run_sql "select count(*) from no_auto_incr_id.$table"
+    check_contains 'count(*): 4'
+  done
 }
 
 rm -rf $TEST_DIR/lightning.log
diff --git a/br/tests/lightning_tool_1472/run.sh b/br/tests/lightning_tool_1472/run.sh
index 71efe4a079a56..e8c2c5a8cfd08 100755
--- a/br/tests/lightning_tool_1472/run.sh
+++ b/br/tests/lightning_tool_1472/run.sh
@@ -25,8 +25,8 @@ run_lightning
 run_sql 'insert into EE1472.pk values ();'
 run_sql 'select count(a), max(a) from EE1472.pk;'
 check_contains 'count(a): 3'
-check_contains 'max(a): 5'
+check_contains 'max(a): 6'
 
 run_sql 'insert into EE1472.notpk (a) values (3333);'
 run_sql 'select b from EE1472.notpk where a = 3333;'
-check_contains 'b: 10'
+check_contains 'b: 11'