fix: table sync should reset the shadow index map (#662)

if the table are full backup & restore.

Author: w41ter

pkg/ccr/job.go

@@ -936,13 +936,17 @@ func (j *Job) partialSync() error {
 			j.progress.TableMapping[tableId] = destTable.Id
 			j.progress.NextWithPersist(j.progress.CommitSeq, DBTablesIncrementalSync, Done, "")
 		case TableSync:
-			commitSeq, ok := j.progress.TableCommitSeqMap[j.Src.TableId]
-			if !ok {
-				return xerror.Errorf(xerror.Normal, "table id %d, commit seq not found", j.Src.TableId)
-			}
+			var commitSeq int64
 			if len(partitions) > 0 {
 				// Only commit partition commit seq map, instead of the entire table commit seq.
 				commitSeq = j.progress.CommitSeq
+			} else if seq, ok := j.progress.TableCommitSeqMap[j.Src.TableId]; !ok {
+				return xerror.Errorf(xerror.Normal, "table id %d, commit seq not found", j.Src.TableId)
+			} else {
+				commitSeq = seq
+				// The entire table has been restored, clear the partition commit seq map and shadow index.
+				j.progress.PartitionCommitSeqMap = nil
+				j.progress.ShadowIndexes = nil
 			}
 			j.Dest.TableId = destTable.Id
 			j.progress.TableMapping = nil

pkg/rpc/rpc_factory.go

@@ -66,12 +66,12 @@ func (rf *RpcFactory) NewFeRpc(spec *base.Spec) (IFeRpc, error) {
 	rf.feRpcsLock.Lock()
 	if feRpc, ok := rf.feRpcs[key]; ok {
 		rf.feRpcsLock.Unlock()
-		log.Debugf("RpcFactory: reused cached FeRpc for %s (cache hit)", key)
+		log.Tracef("RpcFactory: reused cached FeRpc for %s (cache hit)", key)
 		return feRpc, nil
 	}
 	rf.feRpcsLock.Unlock()
 
-	log.Debugf("RpcFactory: creating new FeRpc for %s (cache miss)", key)
+	log.Tracef("RpcFactory: creating new FeRpc for %s (cache miss)", key)
 	feRpc, err := NewFeRpc(spec)
 	if err != nil {
 		return nil, err

regression-test/suites/db_sync_dep/index/drop_bf/test_dsd_index_drop_bf.groovy

@@ -149,4 +149,14 @@ suite("test_dsd_index_drop_bf") {
     assertTrue(helper.checkShowTimesOf(
         """ select * from ${tableName} where username = 'post_drop_unique' """,
         has_count(1), 30, "target"))
+
+    // Insert some more data to verify downstream is still functional
+    sql """
+        INSERT INTO ${tableName} VALUES
+        (2, 200, "final_user_1", "final_data_1"),
+        (2, 201, "final_user_2", "final_data_2")
+    """
+    assertTrue(helper.checkShowTimesOf(
+        """ select * from ${tableName} """,
+        has_count(insert_num + 5), 60, "target"))
 }

regression-test/suites/table_sync_dep/index/drop_bf/test_tsd_index_drop_bf.groovy

@@ -108,22 +108,32 @@ suite("test_tsd_index_drop_bf") {
         ALTER TABLE ${tableName}
         SET ("bloom_filter_columns" = "")
     """
+
+    // Insert more data in parallel, to create the shadow index.
+    def num_values = insert_num
+    insert_num *= 2
+    for (int index = num_values; index < insert_num; index++) {
+        sql """
+            INSERT INTO ${tableName} VALUES (1, ${index}, 'user_${index}', 'data_${index}')
+        """
+    }
     sql "sync"
 
     // Verify bloom filter was removed upstream
     assertTrue(helper.checkShowTimesOf("""
         SHOW CREATE TABLE ${context.dbName}.${tableName}
-        """, 
+        """,
         noBloomFilter, 30, "sql"))
 
     // 4. Insert more data after dropping bloom filter
     sql """
         INSERT INTO ${tableName} VALUES
-        (1, 100, "post_drop_user_1", "extra_data_1"),
-        (1, 101, "post_drop_user_2", "extra_data_2"),
-        (1, 102, "post_drop_unique", "extra_data_3")
+        (2, 100, "post_drop_user_1", "extra_data_1"),
+        (2, 101, "post_drop_user_2", "extra_data_2"),
+        (2, 102, "post_drop_unique", "extra_data_3")
     """
     sql "sync"
+    insert_num += 3
 
     // Verify query still works without bloom filter
     assertTrue(helper.checkShowTimesOf(
@@ -137,7 +147,7 @@ suite("test_tsd_index_drop_bf") {
     // Check all data is synced
     assertTrue(helper.checkShowTimesOf(
         """ select * from ${tableName} """,
-        has_count(insert_num + 3), 60, "target"))
+        has_count(insert_num), 60, "target"))
 
     // Check bloom filter is removed downstream
     assertTrue(helper.checkShowTimesOf("""
@@ -149,4 +159,14 @@ suite("test_tsd_index_drop_bf") {
     assertTrue(helper.checkShowTimesOf(
         """ select * from ${tableName} where username = 'post_drop_unique' """,
         has_count(1), 30, "target"))
+        
+    // Insert some more data to verify downstream is still functional
+    sql """
+        INSERT INTO ${tableName} VALUES
+        (3, 200, "final_user_1", "final_data_1"),
+        (3, 201, "final_user_2", "final_data_2")
+    """
+    assertTrue(helper.checkShowTimesOf(
+        """ select * from ${tableName} """,
+        has_count(insert_num + 2), 60, "target"))
 }