diff --git a/AGENTS.md b/AGENTS.md
index 03ae8c3a9c..62cf5b3333 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -22,4 +22,17 @@ Go is the primary language. Follow [Go Code Review Comments](https://github.com/
 Add unit tests for package-level logic and integration cases for end-to-end replication behavior. New DM shell cases should be added under `dm/tests/<case>/run.sh` and registered in `dm/tests/run_group.sh`. Coverage artifacts are written to `/tmp/dm_test`. If a change affects compatibility behavior, use the existing compatibility workflow instead of only adding a unit test.
 
 ## Commit & Pull Request Guidelines
-Use the established DM commit style: `<subsystem>(dm): <what changed>`, for example `worker(dm): improve retry logging`. Keep the subject within 70 characters and explain the reason in the body. PRs should follow [the pull request template](.github/pull_request_template.md): link the issue with `Issue Number: close #12345`, list the DM tests you ran, note compatibility impact, mention doc updates, and provide a release note or `None`. DM fixes and features are expected to include tests, and PRs normally require two maintainer LGTMs.
+Use the repository commit style from [CONTRIBUTING.md](CONTRIBUTING.md) and existing history: `<subsystem>(ticdc|dm|engine|all): <what changed>` for TiCDC, DM, engine, or shared code, and `sync-diff-inspector: <what changed>` for sync-diff-inspector changes. Match the scope to the code you touch, for example `worker(dm): improve retry logging` for DM-only changes, `capture(ticdc): add comment for variable declaration` for TiCDC changes, and `sync-diff-inspector: update splitter behavior` for sync-diff-inspector changes. Keep the subject within 70 characters and explain the reason in the body.
+
+PR descriptions MUST follow [.github/pull_request_template.md](.github/pull_request_template.md). If you create a PR with GitHub CLI, start from the template with `gh pr create -T .github/pull_request_template.md`, then fill in the fields. Do not replace the template with an informal summary.
+
+Before reporting that a PR is ready or updated, re-read the published PR body and verify it still contains:
+
+- `### What problem does this PR solve?`
+- one line starting with `Issue Number:`. Prefer `Issue Number: close #12345` or `Issue Number: ref #12345`; use `Issue Number: None` only for test-only or housekeeping PRs with no issue.
+- `### What is changed and how it works?`
+- `### Check List`, with at least one applicable test category kept and the exact commands or CI jobs used for verification.
+- `#### Questions`, explicitly answering compatibility/performance impact and documentation impact.
+- `### Release note` with a fenced `release-note` block containing the release note or `None`.
+
+DM fixes and features are expected to include tests.
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index 193d79a388..a31a14eae5 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -32,6 +32,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	tidbddl "github.com/pingcap/tidb/pkg/ddl"
+	"github.com/pingcap/tidb/pkg/lightning/mydump"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -101,6 +102,8 @@ const (
 
 	unhandledEventSampleInterval = 5 * time.Minute
 	unhandledEventSampleFirst    = 1
+
+	maxSchemaFileReadConcurrency = 16
 )
 
 // waitXIDStatus represents the status for waiting XID event when pause/stop task.
@@ -1914,7 +1917,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 
 	if cleanDumpFile {
 		s.tctx.L().Info("try to remove all dump files")
-		if err = storage.RemoveAll(ctx, s.cfg.Dir, nil); err != nil {
+		if err = storage.RemoveAll(ctx, s.cfg.Dir, s.cfg.ExtStorage); err != nil {
 			s.tctx.L().Warn("error when remove loaded dump folder", zap.String("data folder", s.cfg.Dir), zap.Error(err))
 		}
 	}
@@ -3041,12 +3044,17 @@ func (s *Syncer) genRouter() error {
 
 func (s *Syncer) loadTableStructureFromDump(ctx context.Context) error {
 	logger := s.tctx.L()
-	// TODO: delete this check after we support parallel reading the files to improve load speed
-	if !storage.IsLocalDiskPath(s.cfg.LoaderConfig.Dir) {
-		logger.Warn("skip load table structure from dump files for non-local-dir loader because it may be slow", zap.String("loaderDir", s.cfg.LoaderConfig.Dir))
-		return nil
+	dumpStorage := s.cfg.ExtStorage
+	if dumpStorage == nil {
+		var err error
+		dumpStorage, err = storage.CreateStorage(ctx, s.cfg.LoaderConfig.Dir)
+		if err != nil {
+			logger.Warn("fail to create dump storage", zap.Error(err))
+			return err
+		}
+		defer dumpStorage.Close()
 	}
-	files, err := storage.CollectDirFiles(ctx, s.cfg.LoaderConfig.Dir, nil)
+	files, err := storage.CollectDirFiles(ctx, s.cfg.LoaderConfig.Dir, dumpStorage)
 	if err != nil {
 		logger.Warn("fail to get dump files", zap.Error(err))
 		return err
@@ -3092,9 +3100,29 @@ func (s *Syncer) loadTableStructureFromDump(ctx context.Context) error {
 		return err
 	}
 
-	for _, dbAndFile := range tableFiles {
+	type schemaFileReadResult struct {
+		content []byte
+		err     error
+	}
+	readConcurrency := min(max(s.cfg.LoaderConfig.PoolSize, 1), maxSchemaFileReadConcurrency)
+	readResults, err := mydump.ParallelProcess(
+		ctx,
+		tableFiles,
+		readConcurrency,
+		func(ctx context.Context, dbAndFile [2]string) (schemaFileReadResult, error) {
+			content, readErr := storage.ReadFile(ctx, s.cfg.LoaderConfig.Dir, dbAndFile[1], dumpStorage)
+			// Keep reading the remaining schema files and report the first error,
+			// which preserves the existing best-effort behavior.
+			return schemaFileReadResult{content: content, err: readErr}, nil
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	for i, dbAndFile := range tableFiles {
 		db, file := dbAndFile[0], dbAndFile[1]
-		content, err2 := storage.ReadFile(ctx, s.cfg.LoaderConfig.Dir, file, nil)
+		content, err2 := readResults[i].content, readResults[i].err
 		if err2 != nil {
 			logger.Warn("fail to read file for creating table in schema tracker",
 				zap.String("db", db),
diff --git a/dm/syncer/syncer_test.go b/dm/syncer/syncer_test.go
index 5dde6faa51..71e097dea5 100644
--- a/dm/syncer/syncer_test.go
+++ b/dm/syncer/syncer_test.go
@@ -16,6 +16,7 @@ package syncer
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -33,6 +34,7 @@ import (
 	"github.com/pingcap/failpoint"
 	pclog "github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/infoschema"
+	"github.com/pingcap/tidb/pkg/objstore/storeapi"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	pmysql "github.com/pingcap/tidb/pkg/parser/mysql"
@@ -52,6 +54,7 @@ import (
 	parserpkg "github.com/pingcap/tiflow/dm/pkg/parser"
 	"github.com/pingcap/tiflow/dm/pkg/retry"
 	"github.com/pingcap/tiflow/dm/pkg/schema"
+	"github.com/pingcap/tiflow/dm/pkg/storage"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/dm/syncer/binlogstream"
@@ -1326,6 +1329,77 @@ func (s *testSyncerSuite) TestRemoveMetadataIsFine(c *check.C) {
 	c.Assert(fresh, check.IsFalse)
 }
 
+func TestLoadTableStructureFromExternalStorage(t *testing.T) {
+	ctx := context.Background()
+	localDir := t.TempDir()
+	dumpStorage, err := storage.CreateStorage(ctx, localDir)
+	require.NoError(t, err)
+	t.Cleanup(dumpStorage.Close)
+	require.NoError(t, dumpStorage.WriteFile(ctx, "source_db-schema-create.sql", []byte("CREATE DATABASE `source_db`;\n")))
+	require.NoError(t, dumpStorage.WriteFile(ctx, "source_db.t-schema.sql", []byte(
+		"CREATE TABLE `t` (`a` BIGINT, `b` VARCHAR(20));\n")))
+	// A data file must never be read as a schema file.
+	require.NoError(t, dumpStorage.WriteFile(ctx, "source_db.t.0.sql", []byte("not valid SQL")))
+
+	syncer := newTestSyncerForExternalStorage(t, dumpStorage)
+	require.NoError(t, syncer.loadTableStructureFromDump(ctx))
+
+	tableInfo, err := syncer.schemaTracker.GetTableInfo(&filter.Table{Schema: "source_db", Name: "t"})
+	require.NoError(t, err)
+	require.Equal(t, "a", tableInfo.Columns[0].Name.O)
+	require.Equal(t, "b", tableInfo.Columns[1].Name.O)
+}
+
+type readErrorStorage struct {
+	storeapi.Storage
+	failFile string
+}
+
+func (s *readErrorStorage) ReadFile(ctx context.Context, name string) ([]byte, error) {
+	if name == s.failFile {
+		return nil, errors.New("injected read error")
+	}
+	return s.Storage.ReadFile(ctx, name)
+}
+
+func TestLoadTableStructureFromExternalStorageReadError(t *testing.T) {
+	ctx := context.Background()
+	dumpStorage, err := storage.CreateStorage(ctx, t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(dumpStorage.Close)
+	require.NoError(t, dumpStorage.WriteFile(ctx, "source_db-schema-create.sql", []byte("CREATE DATABASE `source_db`;\n")))
+	require.NoError(t, dumpStorage.WriteFile(ctx, "source_db.bad-schema.sql", []byte("CREATE TABLE `bad` (`a` BIGINT);\n")))
+	require.NoError(t, dumpStorage.WriteFile(ctx, "source_db.good-schema.sql", []byte("CREATE TABLE `good` (`a` BIGINT);\n")))
+
+	syncer := newTestSyncerForExternalStorage(t, &readErrorStorage{
+		Storage:  dumpStorage,
+		failFile: "source_db.bad-schema.sql",
+	})
+	require.EqualError(t, syncer.loadTableStructureFromDump(ctx), "injected read error")
+
+	_, err = syncer.schemaTracker.GetTableInfo(&filter.Table{Schema: "source_db", Name: "good"})
+	require.NoError(t, err)
+}
+
+func newTestSyncerForExternalStorage(t *testing.T, dumpStorage storeapi.Storage) *Syncer {
+	t.Helper()
+	ctx := context.Background()
+	cfg := genDefaultSubTaskConfig4Test()
+	localLoaderDir := cfg.LoaderConfig.Dir
+	t.Cleanup(func() { require.NoError(t, os.RemoveAll(localLoaderDir)) })
+	cfg.LoaderConfig.Dir = "s3://unused/dump"
+	cfg.LoaderConfig.PoolSize = 4
+	cfg.ExtStorage = dumpStorage
+
+	syncer := NewSyncer(cfg, nil, nil)
+	var err error
+	syncer.schemaTracker, err = schema.NewTestTracker(ctx, cfg.Name, nil, log.L())
+	require.NoError(t, err)
+	t.Cleanup(func() { syncer.schemaTracker.Close() })
+	require.NoError(t, syncer.genRouter())
+	return syncer
+}
+
 func (s *testSyncerSuite) TestTrackDDL(c *check.C) {
 	var (
 		testDB   = "test_db"
diff --git a/dm/tests/many_tables/run.sh b/dm/tests/many_tables/run.sh
index 59cd24af06..50a6078e1b 100644
--- a/dm/tests/many_tables/run.sh
+++ b/dm/tests/many_tables/run.sh
@@ -73,7 +73,7 @@ function run() {
 	echo "finish prepare_data"
 
 	# we will check metrics, so don't clean metrics
-	export GO_FAILPOINTS='github.com/pingcap/tiflow/dm/loader/DontUnregister=return();github.com/pingcap/tiflow/dm/syncer/IOTotalBytes=return("uuid")'
+	export GO_FAILPOINTS='github.com/pingcap/tiflow/dm/dumpling/SleepBeforeDumplingClose=return(10);github.com/pingcap/tiflow/dm/loader/DontUnregister=return();github.com/pingcap/tiflow/dm/syncer/IOTotalBytes=return("uuid")'
 
 	run_dm_master $WORK_DIR/master $MASTER_PORT $cur/conf/dm-master.toml
 	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT
diff --git a/dm/tests/s3_dumpling_lightning/data/db1.increment.sql b/dm/tests/s3_dumpling_lightning/data/db1.increment.sql
index cd650ef011..3726881d57 100644
--- a/dm/tests/s3_dumpling_lightning/data/db1.increment.sql
+++ b/dm/tests/s3_dumpling_lightning/data/db1.increment.sql
@@ -1,5 +1,5 @@
 use `s3_dumpling_lightning1`;
 insert into t1 (id, name) values (15, '15'), (16, '16');
-insert into t1 (id, name) values (17, '17'), (18, '18');
+insert into t1 (id, name) values (17, '17'), (18, 'name-18');
 insert into t11 (id, name) values (115, '115'), (116, '116');
-insert into t11 (id, name) values (117, '117'), (118, '118');
\ No newline at end of file
+insert into t11 (id, name) values (117, '117'), (118, '118');
diff --git a/dm/tests/s3_dumpling_lightning/data/downstream.prepare.sql b/dm/tests/s3_dumpling_lightning/data/downstream.prepare.sql
new file mode 100644
index 0000000000..618c2f6950
--- /dev/null
+++ b/dm/tests/s3_dumpling_lightning/data/downstream.prepare.sql
@@ -0,0 +1,5 @@
+create database `s3_dumpling_lightning`;
+use `s3_dumpling_lightning`;
+-- Keep the same columns as the upstream tables, but reverse their physical
+-- order to verify that Syncer uses the Dumpling schema for row events.
+create table t (name varchar(20), id int, primary key(`id`));
diff --git a/dm/tests/s3_dumpling_lightning/run.sh b/dm/tests/s3_dumpling_lightning/run.sh
index a07506967e..66661ef724 100755
--- a/dm/tests/s3_dumpling_lightning/run.sh
+++ b/dm/tests/s3_dumpling_lightning/run.sh
@@ -117,6 +117,7 @@ function run_test() {
 	run_sql_file $cur/data/clean_data.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
 	run_sql_file $cur/data/db1.prepare.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
 	run_sql_file $cur/data/db2.prepare.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+	run_sql_file $cur/data/downstream.prepare.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
 
 	echo "start task"
 	cp $cur/conf/dm-task.yaml $WORK_DIR/dm-task.yaml
@@ -139,6 +140,7 @@ function run_test() {
 
 	# check table data (full dump + increments replicated via sync)
 	run_sql_tidb_with_retry "select count(1) from ${db}.${tb};" "count(1): 25"
+	run_sql_tidb_with_retry "select name from ${db}.${tb} where id = 18;" "name: name-18"
 
 	# check dump file
 	if $1; then
diff --git a/dm/tests/shardddl1/run.sh b/dm/tests/shardddl1/run.sh
index b22e51f713..db8b1e185b 100644
--- a/dm/tests/shardddl1/run.sh
+++ b/dm/tests/shardddl1/run.sh
@@ -559,6 +559,7 @@ function DM_COMPACT_CASE() {
 		run_sql_source1 "delete from ${shardddl1}.${tb1} where a=$((i + 100))"
 		run_sql_source1 "insert into ${shardddl1}.${tb1}(a,b) values($i,$i)"
 	done
+	run_sql_tidb_with_retry_times "select count(1) from ${shardddl}.${tb};" "count(1): 100" 120
 	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml 30
 	compactCnt=$(cat $WORK_DIR/worker1/log/dm-worker.log $WORK_DIR/worker2/log/dm-worker.log | grep "finish to compact" | wc -l)
 	if [[ "$compactCnt" -le 100 ]]; then
@@ -677,7 +678,7 @@ function DM_MULTIPLE_ROWS_CASE() {
 		($((i + 5)),$((i + 5))),($((i + 6)),$((i + 6))),($((i + 7)),$((i + 7))),($((i + 8)),$((i + 8))),($((i + 9)),$((i + 9)))"
 	done
 
-	run_sql_tidb_with_retry "select count(1) from ${shardddl}.${tb} where a>100 and a<=200;" "count(1): 100"
+	run_sql_tidb_with_retry_times "select count(1) from ${shardddl}.${tb} where a>100 and a<=200;" "count(1): 100" 60
 	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml 30
 	insertMergeCnt=$(cat $WORK_DIR/worker1/log/dm-worker.log $WORK_DIR/worker2/log/dm-worker.log | grep '\[op=DMLInsert\]' | wc -l)
 	replaceMergeCnt=$(cat $WORK_DIR/worker1/log/dm-worker.log $WORK_DIR/worker2/log/dm-worker.log | grep '\[op=DMLReplace\]' | wc -l)
diff --git a/dm/tests/shardddl2/run.sh b/dm/tests/shardddl2/run.sh
index 1be8dde17e..99772c933c 100644
--- a/dm/tests/shardddl2/run.sh
+++ b/dm/tests/shardddl2/run.sh
@@ -126,8 +126,8 @@ function DM_040_CASE() {
 	else
 		run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 			"query-status test" \
-			'ALTER TABLE `shardddl`.`tb` ADD COLUMN `col1` VARCHAR(10) CHARACTER SET UTF8' 1 \
-			"\`${shardddl1}\`.\`${tb1}\`\"" 1
+			"\"result\": true" 3
+		check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
 	fi
 }