fix: skip OnlineDDL filter during cancelMigrate (cancellation path)

When cancelMigrate=true, we should not exclude stopped OnlineDDL
streams from the full read, as the cancellation path needs to see
all streams to properly clean up.

Co-Authored-By: Claude <svc-devxp-claude@slack-corp.com>

Author: yushuqin

go/vt/vtctl/workflow/stream_migrator.go

@@ -469,7 +469,11 @@ func (sm *StreamMigrator) legacyReadSourceStreams(ctx context.Context, cancelMig
 			}
 		}
 
-		tabletStreams, err := sm.legacyReadTabletStreams(ctx, source.GetPrimary(), "NOT (workflow_type = 5 AND state = 'Stopped')")
+		constraint := ""
+		if !cancelMigrate {
+			constraint = "NOT (workflow_type = 5 AND state = 'Stopped')"
+		}
+		tabletStreams, err := sm.legacyReadTabletStreams(ctx, source.GetPrimary(), constraint)
 		if err != nil {
 			return err
 		}
@@ -575,7 +579,7 @@ func (sm *StreamMigrator) readSourceStreams(ctx context.Context, cancelMigrate b
 			}
 		}
 
-		tabletStreams, err := sm.readTabletStreams(ctx, source.GetPrimary(), nil, nil, false, true)
+		tabletStreams, err := sm.readTabletStreams(ctx, source.GetPrimary(), nil, nil, false, !cancelMigrate)
 		if err != nil {
 			return err
 		}

go/vt/wrangler/stream_migrater_test.go

@@ -1156,7 +1156,7 @@ func TestStreamMigrateCancelWithStoppedStreams(t *testing.T) {
 
 		for i, dbclient := range tme.dbSourceClients {
 			// sm.stopStreams->sm.readSourceStreams->readTabletStreams('') and VReplicationExec(_vt.copy_state)
-			dbclient.addQuery("select id, workflow, source, pos, workflow_type, workflow_sub_type, defer_secondary_keys from _vt.vreplication where db_name='vt_ks' and workflow != 'test_reverse' and NOT (workflow_type = 5 AND state = 'Stopped')", sqltypes.MakeTestResult(sqltypes.MakeTestFields(
+			dbclient.addQuery("select id, workflow, source, pos, workflow_type, workflow_sub_type, defer_secondary_keys from _vt.vreplication where db_name='vt_ks' and workflow != 'test_reverse'", sqltypes.MakeTestResult(sqltypes.MakeTestFields(
 				"id|workflow|source|pos|workflow_type|workflow_sub_type|defer_secondary_keys",
 				"int64|varbinary|varchar|varbinary|int64|int64|int64"),
 				sourceRows[i]...),