br: fix table filter nil pointer issue (pingcap#61225)

close pingcap#61226

Author: Tristan1900

br/pkg/restore/log_client/batch_meta_processor.go

@@ -60,6 +60,7 @@ func (rp *RestoreMetaKVProcessor) RestoreAndRewriteMetaKVFiles(
 	ctx context.Context,
 	hasExplicitFilter bool,
 	files []*backuppb.DataFileInfo,
+	schemasReplace *stream.SchemasReplace,
 ) error {
 	// starts gc row collector
 	rp.client.RunGCRowsLoader(ctx)
@@ -89,7 +90,12 @@ func (rp *RestoreMetaKVProcessor) RestoreAndRewriteMetaKVFiles(
 			return errors.Trace(err)
 		}
 	} else {
-		log.Info("skip doing full reload for filtered PiTR")
+		// refresh metadata will sync data from TiKV to info schema one table at a time.
+		// this must succeed to ensure schema consistency
+		log.Info("refreshing schema meta")
+		if err := rp.client.RefreshMetaForTables(ctx, schemasReplace); err != nil {
+			return errors.Trace(err)
+		}
 	}
 	return nil
 }

br/pkg/task/stream.go

@@ -1579,7 +1579,7 @@ func restoreStream(
 	var rp *logclient.RestoreMetaKVProcessor
 	if err = glue.WithProgress(ctx, g, "Restore Meta Files", int64(len(ddlFiles)), !cfg.LogProgress, func(p glue.Progress) error {
 		rp = logclient.NewRestoreMetaKVProcessor(client, schemasReplace, updateStats, p.Inc)
-		return rp.RestoreAndRewriteMetaKVFiles(ctx, cfg.ExplicitFilter, ddlFiles)
+		return rp.RestoreAndRewriteMetaKVFiles(ctx, cfg.ExplicitFilter, ddlFiles, schemasReplace)
 	}); err != nil {
 		return errors.Annotate(err, "failed to restore meta files")
 	}
@@ -1663,12 +1663,6 @@ func restoreStream(
 	}
 
 	if cfg.ExplicitFilter {
-		// refresh metadata will sync data from TiKV to info schema one table at a time.
-		// this must succeed to ensure schema consistency
-		if err = client.RefreshMetaForTables(ctx, schemasReplace); err != nil {
-			return errors.Trace(err)
-		}
-
 		failpoint.Inject("before-set-table-mode-to-normal", func(_ failpoint.Value) {
 			failpoint.Return(errors.New("fail before setting table mode to normal"))
 		})
@@ -2131,7 +2125,9 @@ func buildAndSaveIDMapIfNeeded(ctx context.Context, client *logclient.LogClient,
 
 	// either getting base id map from previous pitr or this is a new task and get base map from snapshot restore phase
 	// do filter
-	cfg.tableMappingManager.ApplyFilterToDBReplaceMap(cfg.PiTRTableTracker)
+	if cfg.PiTRTableTracker != nil {
+		cfg.tableMappingManager.ApplyFilterToDBReplaceMap(cfg.PiTRTableTracker)
+	}
 	// replace temp id with read global id
 	err = cfg.tableMappingManager.ReplaceTemporaryIDs(ctx, client.GenGlobalIDs)
 	if err != nil {

br/tests/br_pitr_table_filter/run.sh

@@ -1553,16 +1553,124 @@ test_log_compaction() {
     echo "log compaction with filter test passed"
 }
 
-#test_basic_filter
-#test_with_full_backup_filter
+test_pitr_chaining() {
+    restart_services || { echo "Failed to restart services"; exit 1; }
+
+    echo "case 21: start testing PITR chaining (sequential restores without cleaning up)"
+
+    run_sql "create schema $DB;"
+
+    echo "creating tables for initial state..."
+    run_sql "CREATE TABLE $DB.table_a (
+        id INT PRIMARY KEY,
+        value VARCHAR(50)
+    );"
+    run_sql "CREATE TABLE $DB.table_b (
+        id INT PRIMARY KEY,
+        value VARCHAR(50)
+    );"
+
+    run_br --pd $PD_ADDR log start --task-name $TASK_NAME -s "local://$TEST_DIR/$TASK_NAME/log"
+
+    run_sql "INSERT INTO $DB.table_a VALUES (1, 'initial data 1'), (2, 'initial data 2');"
+    run_sql "INSERT INTO $DB.table_b VALUES (1, 'initial data 1'), (2, 'initial data 2');"
+    
+    run_br backup full -s "local://$TEST_DIR/$TASK_NAME/full" --pd $PD_ADDR
+
+    run_sql "INSERT INTO $DB.table_a VALUES (3, 'post-backup data 1');"
+    run_sql "INSERT INTO $DB.table_b VALUES (3, 'post-backup data 1');"
+    
+    . "$CUR/../br_test_utils.sh" && wait_log_checkpoint_advance "$TASK_NAME"
+    first_restore_ts=$(python3 -c "import time; print(int(time.time() * 1000) << 18)")
+    echo "Captured first checkpoint timestamp: $first_restore_ts"
+    
+    run_sql "INSERT INTO $DB.table_a VALUES (4, 'post-first-checkpoint data');"
+    run_sql "INSERT INTO $DB.table_b VALUES (4, 'post-first-checkpoint data');"
+    
+    run_sql "CREATE TABLE $DB.table_c (
+        id INT PRIMARY KEY,
+        value VARCHAR(50)
+    );"
+    run_sql "INSERT INTO $DB.table_c VALUES (1, 'created after first checkpoint');"
+    
+    . "$CUR/../br_test_utils.sh" && wait_log_checkpoint_advance "$TASK_NAME"
+
+    run_br --pd $PD_ADDR log stop --task-name $TASK_NAME
+    
+    run_sql "drop schema if exists $DB;"
+    
+    echo "Step 1: First restore with full backup to first checkpoint timestamp"
+    run_br --pd "$PD_ADDR" restore point -s "local://$TEST_DIR/$TASK_NAME/log" \
+        --full-backup-storage "local://$TEST_DIR/$TASK_NAME/full" \
+        --restored-ts $first_restore_ts \
+        -f "$DB.*"
+    
+    run_sql "SELECT COUNT(*) = 3 FROM $DB.table_a" || {
+        echo "table_a doesn't have expected row count after first restore"
+        exit 1
+    }
+    
+    run_sql "SELECT COUNT(*) = 3 FROM $DB.table_b" || {
+        echo "table_b doesn't have expected row count after first restore"
+        exit 1
+    }
+    
+    if run_sql "SELECT * FROM $DB.table_c" 2>/dev/null; then
+        echo "table_c exists after first restore but shouldn't"
+        exit 1
+    fi
+    
+    echo "Step 2: Second restore with log only using first checkpoint timestamp as startTS"
+    run_br --pd "$PD_ADDR" restore point -s "local://$TEST_DIR/$TASK_NAME/log" \
+        --start-ts $first_restore_ts \
+        -f "$DB.*"
+    
+    # Verify data after second restore
+    run_sql "SELECT COUNT(*) = 4 FROM $DB.table_a" || {
+        echo "table_a doesn't have expected row count after second restore"
+        exit 1
+    }
+    
+    run_sql "SELECT COUNT(*) = 4 FROM $DB.table_b" || {
+        echo "table_b doesn't have expected row count after second restore"
+        exit 1
+    }
+    
+    run_sql "SELECT COUNT(*) = 1 FROM $DB.table_c" || {
+        echo "table_c doesn't have expected row count after second restore"
+        exit 1
+    }
+
+    # make sure able to write data after restore
+    run_sql "INSERT INTO $DB.table_a VALUES (5, 'post-second-checkpoint data');"
+    run_sql "CREATE TABLE $DB.table_d (
+        id INT PRIMARY KEY,
+        value VARCHAR(50)
+    );"
+    run_sql "INSERT INTO $DB.table_d VALUES (1, 'created after second checkpoint');"
+
+    verify_no_unexpected_tables 4 "$DB" || {
+        echo "Wrong number of tables after all restores"
+        exit 1
+    }
+
+    run_sql "drop schema if exists $DB;"
+    rm -rf "$TEST_DIR/$TASK_NAME"
+    
+    echo "PITR sequential restore test passed"
+}
+
+test_basic_filter
+test_with_full_backup_filter
 test_table_rename
-#test_with_checkpoint
-#test_partition_exchange
-#test_system_tables
-#test_foreign_keys
-#test_index_filter
-#test_table_truncation
-#test_sequential_restore
-#test_log_compaction
+test_with_checkpoint
+test_partition_exchange
+test_system_tables
+test_foreign_keys
+test_index_filter
+test_table_truncation
+test_sequential_restore
+test_log_compaction
+test_pitr_chaining
 
 echo "br pitr table filter all tests passed"