From cc695733f0a414681b878b3b97b949babbe7f08a Mon Sep 17 00:00:00 2001
From: tangenta <tangenta@126.com>
Date: Tue, 6 Dec 2022 21:17:01 +0800
Subject: [PATCH] ddl: assign table record prefix to start/end key if it is
 empty

---
 ddl/backfilling.go                               |  9 ++++++++-
 .../addindextest/integration_test.go             | 16 ++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/ddl/backfilling.go b/ddl/backfilling.go
index 76e1fc2aaf5fd..5bf21250eadd8 100644
--- a/ddl/backfilling.go
+++ b/ddl/backfilling.go
@@ -502,6 +502,7 @@ func (dc *ddlCtx) handleRangeTasks(scheduler *backfillScheduler, t table.Table,
 	// Build reorg tasks.
 	job := reorgInfo.Job
 	for i, keyRange := range kvRanges {
+		startKey := keyRange.StartKey
 		endKey := keyRange.EndKey
 		endK, err := getRangeEndKey(scheduler.jobCtx, dc.store, job.Priority, prefix, keyRange.StartKey, endKey)
 		if err != nil {
@@ -511,11 +512,17 @@ func (dc *ddlCtx) handleRangeTasks(scheduler *backfillScheduler, t table.Table,
 				zap.String("end key", hex.EncodeToString(endKey)), zap.String("current end key", hex.EncodeToString(endK)))
 			endKey = endK
 		}
+		if len(startKey) == 0 {
+			startKey = prefix
+		}
+		if len(endKey) == 0 {
+			endKey = prefix.PrefixNext()
+		}
 
 		task := &reorgBackfillTask{
 			id:              i,
 			physicalTableID: physicalTableID,
-			startKey:        keyRange.StartKey,
+			startKey:        startKey,
 			endKey:          endKey,
 			// If the boundaries overlap, we should ignore the preceding endKey.
 			endInclude: endK.Cmp(keyRange.EndKey) != 0 || i == len(kvRanges)-1}
diff --git a/tests/realtikvtest/addindextest/integration_test.go b/tests/realtikvtest/addindextest/integration_test.go
index 70da49e58364b..945bf37dda44d 100644
--- a/tests/realtikvtest/addindextest/integration_test.go
+++ b/tests/realtikvtest/addindextest/integration_test.go
@@ -256,3 +256,19 @@ func TestAddIndexIngestGeneratedColumns(t *testing.T) {
 	tk.MustQuery("select * from t;").Check(testkit.Rows("1 1 1 2 1", "2 2 2 4 2", "3 3 3 6 3"))
 	assertLastNDDLUseIngest(4)
 }
+
+func TestAddIndexIngestEmptyTable(t *testing.T) {
+	store := realtikvtest.CreateMockStoreAndSetup(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("drop database if exists addindexlit;")
+	tk.MustExec("create database addindexlit;")
+	tk.MustExec("use addindexlit;")
+	tk.MustExec("create table t (a int);")
+	tk.MustExec(`set global tidb_ddl_enable_fast_reorg=on;`)
+	tk.MustExec("alter table t add index idx(a);")
+
+	rows := tk.MustQuery("admin show ddl jobs 1;").Rows()
+	require.Len(t, rows, 1)
+	jobTp := rows[0][3].(string)
+	require.True(t, strings.Contains(jobTp, "ingest"), jobTp)
+}