Repository: pingcap/tidb-lightning
Pull request #366 (2020-08-07 07:53:06)

restore: support file level routing
<!--
Thank you for contributing to TiDB-Lightning! Please read the [CONTRIBUTING](https://github.com/pingcap/tidb-lightning/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
Implement table level routing, see `Lighting RFC 5` for more detals

### What is changed and how it works?
With now configurations `mydumper.files`, user can privide regexp pattern to match source files and conf the source relate schema/table/type either by constant string or template value expand from the regexp pattern.

NOTE: this pr doesn't implement following features:
- file sorting key. relate to the `mydumper.files.key` conf. User can still set this conf, but is currently useless.
- file sorting key transform. relate to the `mydumper.files.key-transform`. This conf is not provided.
- compression. relate to the `mydumper.files.compression`. As we don't support read compression files yet.  User can still set this conf, but is currently useless.

We should deprecate the feature `table router` and remote the `lightning.router` config fileds in the future. 


### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)
 - No code

Side effects

 - Increased code complexity
 - Increased config complexity

Related changes

 - Need to cherry-pick to the release branch
 - Need to update the documentation
 - Need to be included in the release note

kennytm: ```
[2020-08-07T10:28:24.499Z] Error: restore table `error_summary`.`a` failed: checksum mismatched remote vs local => (checksum: 10176589659643620760 vs 5152985213657377814) (total_kvs: 3 vs 2) (total_bytes:87 vs 58)
```
glorv: > remote

This is expected. Because target table contains 1 row before restore. 
glorv: @kennytm  PTAL again.
BTW, shall we add the new config field `path` to [files] to allow user set a file path instead of regexp pattern?
kennytm: >  BTW, shall we add the new config field `path` to `[files]` to allow user set a file path instead of regexp pattern?

do it if it's easy

---

i still think there's no point having a `constExpander`. 

---

you are also making `extractSubPatterns` way too complicated by using `strings.IndexByte`. try https://play.golang.org/p/pxPaTbLPuWH.

```go
var expandVariablePattern = regexp.MustCompile(`\$(?:\$|[\pL\p{Nd}_]+|\{[\pL\p{Nd}_]+\})`)

func VerifyExpansion(re *regexp.Regexp, template string) error {
	for _, variable := range expandVariablePattern.FindAllString(template, -1) {
		var name string
		switch {
		case variable == "$$":
			continue
		case strings.HasPrefix(variable, "${"):
			name = variable[2 : len(variable)-1]
		default:
			name = variable[1:]
		}
		if number, err := strconv.Atoi(name); err == nil {
			if number > re.NumSubexp() {
				return fmt.Errorf("sub pattern capture '%s' out of range", variable)
			}
		} else if re.SubexpIndex(name) < 0 {
			return fmt.Errorf("invalid named capture '%s'", variable)
		}
	}
	return nil
}
```

(btw a trailing `$` and malformed `${...` is fine by go standard)
glorv: @kennytm  I updated the pattern check logic and add support for `path` config. PTAL, again 😂

BTW:
after #361  , the file path traversed by `WalkDir` will be relative path, but currently, they are all abs path. So I decide to support both abs path and relative path for `files.path` in that pr, because there are a lot of tests to modify and I don't want to do again in this pr
kennytm: PTAL @3pointer 

Patch:
diff --git a/lightning/checkpoints/checkpoints.go b/lightning/checkpoints/checkpoints.go
index d78ea7c..e8d9ca9 100644
--- a/lightning/checkpoints/checkpoints.go
+++ b/lightning/checkpoints/checkpoints.go
@@ -61,7 +61,7 @@ const (
 	// remember to increase the version number in case of incompatible change.
 	CheckpointTableNameTable  = "table_v6"
 	CheckpointTableNameEngine = "engine_v5"
-	CheckpointTableNameChunk  = "chunk_v4"
+	CheckpointTableNameChunk  = "chunk_v5"
 )
 
 func IsCheckpointTable(name string) bool {
@@ -115,6 +115,7 @@ func (key *ChunkCheckpointKey) less(other *ChunkCheckpointKey) bool {
 
 type ChunkCheckpoint struct {
 	Key               ChunkCheckpointKey
+	FileMeta          mydump.SourceFileMeta
 	ColumnPermutation []int
 	Chunk             mydump.Chunk
 	Checksum          verify.KVChecksum
@@ -126,6 +127,7 @@ func (ccp *ChunkCheckpoint) DeepCopy() *ChunkCheckpoint {
 	colPerm = append(colPerm, ccp.ColumnPermutation...)
 	return &ChunkCheckpoint{
 		Key:               ccp.Key,
+		FileMeta:          ccp.FileMeta,
 		ColumnPermutation: colPerm,
 		Chunk:             ccp.Chunk,
 		Checksum:          ccp.Checksum,
@@ -435,6 +437,9 @@ func NewMySQLCheckpointsDB(ctx context.Context, db *sql.DB, schemaName string, t
 			engine_id int unsigned NOT NULL,
 			path varchar(2048) NOT NULL,
 			offset bigint NOT NULL,
+			type int NOT NULL,
+			compression int NOT NULL,
+			sort_key varchar(256) NOT NULL,
 			columns text NULL,
 			should_include_row_id BOOL NOT NULL,
 			end_offset bigint NOT NULL,
@@ -547,7 +552,7 @@ func (cpdb *MySQLCheckpointsDB) Get(ctx context.Context, tableName string) (*Tab
 
 		chunkQuery := fmt.Sprintf(`
 			SELECT
-				engine_id, path, offset, columns,
+				engine_id, path, offset, type, compression, sort_key, columns,
 				pos, end_offset, prev_rowid_max, rowid_max,
 				kvc_bytes, kvc_kvs, kvc_checksum, unix_timestamp(create_time)
 			FROM %s.%s WHERE table_name = ?
@@ -560,7 +565,7 @@ func (cpdb *MySQLCheckpointsDB) Get(ctx context.Context, tableName string) (*Tab
 		defer chunkRows.Close()
 		for chunkRows.Next() {
 			var (
-				value       = new(ChunkCheckpoint)
+				value       = &ChunkCheckpoint{}
 				colPerm     []byte
 				engineID    int32
 				kvcBytes    uint64
@@ -568,12 +573,14 @@ func (cpdb *MySQLCheckpointsDB) Get(ctx context.Context, tableName string) (*Tab
 				kvcChecksum uint64
 			)
 			if err := chunkRows.Scan(
-				&engineID, &value.Key.Path, &value.Key.Offset, &colPerm,
-				&value.Chunk.Offset, &value.Chunk.EndOffset, &value.Chunk.PrevRowIDMax, &value.Chunk.RowIDMax,
-				&kvcBytes, &kvcKVs, &kvcChecksum, &value.Timestamp,
+				&engineID, &value.Key.Path, &value.Key.Offset, &value.FileMeta.Type, &value.FileMeta.Compression,
+				&value.FileMeta.SortKey, &colPerm, &value.Chunk.Offset, &value.Chunk.EndOffset,
+				&value.Chunk.PrevRowIDMax, &value.Chunk.RowIDMax, &kvcBytes, &kvcKVs, &kvcChecksum,
+				&value.Timestamp,
 			); err != nil {
 				return errors.Trace(err)
 			}
+			value.FileMeta.Path = value.Key.Path
 			value.Checksum = verify.MakeKVChecksum(kvcBytes, kvcKVs, kvcChecksum)
 			if err := json.Unmarshal(colPerm, &value.ColumnPermutation); err != nil {
 				return errors.Trace(err)
@@ -622,12 +629,12 @@ func (cpdb *MySQLCheckpointsDB) InsertEngineCheckpoints(ctx context.Context, tab
 		chunkStmt, err := tx.PrepareContext(c, fmt.Sprintf(`
 			REPLACE INTO %s.%s (
 				table_name, engine_id,
-				path, offset, columns, should_include_row_id,
+				path, offset, type, compression, sort_key, columns, should_include_row_id,
 				pos, end_offset, prev_rowid_max, rowid_max,
 				kvc_bytes, kvc_kvs, kvc_checksum, create_time
 			) VALUES (
 				?, ?,
-				?, ?, ?, FALSE,
+				?, ?, ?, ?, ?, ?, FALSE,
 				?, ?, ?, ?,
 				0, 0, 0, from_unixtime(?)
 			);
@@ -649,9 +656,9 @@ func (cpdb *MySQLCheckpointsDB) InsertEngineCheckpoints(ctx context.Context, tab
 				}
 				_, err = chunkStmt.ExecContext(
 					c, tableName, engineID,
-					value.Key.Path, value.Key.Offset, columnPerm,
-					value.Chunk.Offset, value.Chunk.EndOffset, value.Chunk.PrevRowIDMax, value.Chunk.RowIDMax,
-					value.Timestamp,
+					value.Key.Path, value.Key.Offset, value.FileMeta.Type, value.FileMeta.Compression,
+					value.FileMeta.SortKey, columnPerm, value.Chunk.Offset, value.Chunk.EndOffset,
+					value.Chunk.PrevRowIDMax, value.Chunk.RowIDMax, value.Timestamp,
 				)
 				if err != nil {
 					return errors.Trace(err)
@@ -865,6 +872,12 @@ func (cpdb *FileCheckpointsDB) Get(_ context.Context, tableName string) (*TableC
 					Path:   chunkModel.Path,
 					Offset: chunkModel.Offset,
 				},
+				FileMeta: mydump.SourceFileMeta{
+					Path:        chunkModel.Path,
+					Type:        mydump.SourceType(chunkModel.Type),
+					Compression: mydump.Compression(chunkModel.Compression),
+					SortKey:     chunkModel.SortKey,
+				},
 				ColumnPermutation: colPerm,
 				Chunk: mydump.Chunk{
 					Offset:       chunkModel.Pos,
@@ -907,6 +920,9 @@ func (cpdb *FileCheckpointsDB) InsertEngineCheckpoints(_ context.Context, tableN
 				}
 				engineModel.Chunks[key] = chunk
 			}
+			chunk.Type = int32(value.FileMeta.Type)
+			chunk.Compression = int32(value.FileMeta.Compression)
+			chunk.SortKey = value.FileMeta.SortKey
 			chunk.Pos = value.Chunk.Offset
 			chunk.EndOffset = value.Chunk.EndOffset
 			chunk.PrevRowidMax = value.Chunk.PrevRowIDMax
@@ -1196,6 +1212,9 @@ func (cpdb *MySQLCheckpointsDB) DumpChunks(ctx context.Context, writer io.Writer
 			table_name,
 			path,
 			offset,
+			type,
+			compression,
+			sort_key,
 			columns,
 			pos,
 			end_offset,
diff --git a/lightning/checkpoints/file_checkpoints.pb.go b/lightning/checkpoints/file_checkpoints.pb.go
index 6dbcf90..b5b8f39 100644
--- a/lightning/checkpoints/file_checkpoints.pb.go
+++ b/lightning/checkpoints/file_checkpoints.pb.go
@@ -6,12 +6,11 @@ package checkpoints
 import (
 	encoding_binary "encoding/binary"
 	fmt "fmt"
+	_ "github.com/gogo/protobuf/gogoproto"
+	proto "github.com/gogo/protobuf/proto"
 	io "io"
 	math "math"
 	math_bits "math/bits"
-
-	_ "github.com/gogo/protobuf/gogoproto"
-	proto "github.com/gogo/protobuf/proto"
 )
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -155,6 +154,9 @@ type ChunkCheckpointModel struct {
 	KvcKvs            uint64  `protobuf:"varint,10,opt,name=kvc_kvs,json=kvcKvs,proto3" json:"kvc_kvs,omitempty"`
 	KvcChecksum       uint64  `protobuf:"fixed64,11,opt,name=kvc_checksum,json=kvcChecksum,proto3" json:"kvc_checksum,omitempty"`
 	Timestamp         int64   `protobuf:"fixed64,13,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+	Type              int32   `protobuf:"varint,14,opt,name=type,proto3" json:"type,omitempty"`
+	Compression       int32   `protobuf:"varint,15,opt,name=compression,proto3" json:"compression,omitempty"`
+	SortKey           string  `protobuf:"bytes,16,opt,name=sort_key,json=sortKey,proto3" json:"sort_key,omitempty"`
 }
 
 func (m *ChunkCheckpointModel) Reset()         { *m = ChunkCheckpointModel{} }
@@ -205,44 +207,47 @@ func init() {
 }
 
 var fileDescriptor_deb32a9bf46ada61 = []byte{
-	// 587 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0xcb, 0x6e, 0xd3, 0x40,
-	0x14, 0xcd, 0xc4, 0x6d, 0x1e, 0x37, 0x29, 0x4a, 0x47, 0x6d, 0x19, 0x05, 0xb0, 0x4c, 0xc5, 0xc2,
-	0x88, 0x36, 0x91, 0xca, 0x06, 0x55, 0xac, 0xfa, 0x58, 0x20, 0x54, 0x51, 0x8d, 0x60, 0xc3, 0xc6,
-	0x9a, 0x38, 0x13, 0xdb, 0xf2, 0x63, 0x2c, 0xcf, 0xd8, 0xb4, 0x1f, 0x81, 0xc4, 0x9f, 0xf0, 0x05,
-	0xec, 0xbb, 0xec, 0x92, 0x05, 0x0b, 0x68, 0x7f, 0x04, 0x79, 0xec, 0x2a, 0x2e, 0x8a, 0x2a, 0x76,
-	0xf7, 0x9e, 0x73, 0xee, 0x99, 0x9c, 0xb9, 0x19, 0xc3, 0x5e, 0x14, 0x78, 0xbe, 0x4a, 0x82, 0xc4,
-	0x9b, 0xba, 0x3e, 0x77, 0xc3, 0x54, 0x04, 0x89, 0x92, 0xd3, 0x45, 0x10, 0x71, 0xa7, 0x01, 0x4c,
-	0xd2, 0x4c, 0x28, 0x31, 0xde, 0xf7, 0x02, 0xe5, 0xe7, 0xb3, 0x89, 0x2b, 0xe2, 0xa9, 0x27, 0x3c,
-	0x31, 0xd5, 0xf0, 0x2c, 0x5f, 0xe8, 0x4e, 0x37, 0xba, 0xaa, 0xe4, 0xbb, 0xdf, 0x11, 0x8c, 0x8e,
-	0x97, 0x26, 0x67, 0x62, 0xce, 0x23, 0x7c, 0x02, 0x83, 0x86, 0x31, 0x41, 0x96, 0x61, 0x0f, 0x0e,
-	0x76, 0x27, 0xff, 0xea, 0x9a, 0xc0, 0x69, 0xa2, 0xb2, 0x4b, 0xda, 0x1c, 0x1b, 0x7f, 0xba, 0xe7,
-	0xac, 0x05, 0x78, 0x04, 0x46, 0xc8, 0x2f, 0x09, 0xb2, 0x90, 0xdd, 0xa7, 0x65, 0x89, 0x5f, 0xc1,
-	0x7a, 0xc1, 0xa2, 0x9c, 0x93, 0xb6, 0x85, 0xec, 0xc1, 0xc1, 0xf6, 0xe4, 0x23, 0x9b, 0x45, 0x7c,
-	0x39, 0xa8, 0x4f, 0xa2, 0x95, 0xe6, 0xb0, 0xfd, 0x06, 0xed, 0x7e, 0x6d, 0xc3, 0xd6, 0x2a, 0x0d,
-	0xc6, 0xb0, 0xe6, 0x33, 0xe9, 0x6b, 0xf3, 0x21, 0xd5, 0x35, 0xde, 0x81, 0x8e, 0x54, 0x4c, 0xe5,
-	0x92, 0x18, 0x16, 0xb2, 0x37, 0x68, 0xdd, 0xe1, 0x67, 0x00, 0x2c, 0x8a, 0x84, 0xeb, 0xcc, 0x98,
-	0xe4, 0x64, 0xcd, 0x42, 0xb6, 0x41, 0xfb, 0x1a, 0x39, 0x62, 0x92, 0xe3, 0xb7, 0xd0, 0xe5, 0x89,
-	0x17, 0x24, 0x5c, 0x92, 0x5e, 0x1d, 0x7e, 0xd5, 0x91, 0x93, 0xd3, 0x4a, 0x54, 0x85, 0xbf, 0x1b,
-	0xc1, 0x04, 0xba, 0xaa, 0x54, 0xbf, 0x3b, 0x21, 0x7d, 0xed, 0x7c, 0xd7, 0x8e, 0x29, 0x0c, 0x9b,
-	0x23, 0xcd, 0xeb, 0xd8, 0xac, 0xae, 0x63, 0xef, 0xfe, 0x75, 0xec, 0xd4, 0x47, 0x3c, 0x70, 0x1f,
-	0x3f, 0x10, 0x6c, 0xaf, 0x14, 0x35, 0xc2, 0xa3, 0x7b, 0xe1, 0x0f, 0xa1, 0xe3, 0xfa, 0x79, 0x12,
-	0x4a, 0xd2, 0xae, 0xc3, 0xad, 0x9c, 0x9f, 0x1c, 0x6b, 0x51, 0x15, 0xae, 0x9e, 0x18, 0x9f, 0xc3,
-	0xa0, 0x01, 0xff, 0xcf, 0x3e, 0xb5, 0xfc, 0x81, 0xdf, 0xff, 0xab, 0x0d, 0x5b, 0xab, 0x34, 0xe5,
-	0x3e, 0x53, 0xa6, 0xfc, 0xda, 0x5c, 0xd7, 0x65, 0x24, 0xb1, 0x58, 0x48, 0xae, 0xb4, 0xbd, 0x41,
-	0xeb, 0x0e, 0xef, 0x03, 0x76, 0x45, 0x94, 0xc7, 0x89, 0x93, 0xf2, 0x2c, 0xce, 0x15, 0x53, 0x81,
-	0x48, 0xc8, 0xd0, 0x32, 0xec, 0x75, 0xba, 0x59, 0x31, 0xe7, 0x4b, 0xa2, 0x5c, 0x3f, 0x4f, 0xe6,
-	0x4e, 0x6d, 0xb5, 0x5e, 0xad, 0x9f, 0x27, 0xf3, 0x0f, 0x95, 0xdb, 0x08, 0x8c, 0x54, 0x48, 0xd2,
-	0xd1, 0x78, 0x59, 0xe2, 0x17, 0xf0, 0x28, 0xcd, 0x78, 0xe1, 0x64, 0xe2, 0x4b, 0x30, 0x77, 0x62,
-	0x76, 0x41, 0xba, 0x9a, 0x1c, 0x96, 0x28, 0x2d, 0xc1, 0x33, 0x76, 0x81, 0x9f, 0x40, 0x7f, 0x29,
-	0xe8, 0x69, 0x41, 0x2f, 0x6b, 0x90, 0x61, 0xe1, 0x3a, 0xb3, 0x4b, 0xc5, 0xa5, 0xfe, 0x5f, 0xac,
-	0xd1, 0x5e, 0x58, 0xb8, 0x47, 0x65, 0x8f, 0x1f, 0x43, 0xb7, 0x24, 0xc3, 0x42, 0x12, 0xd0, 0x54,
-	0x27, 0x2c, 0xdc, 0xf7, 0x85, 0xc4, 0xcf, 0x61, 0x58, 0x12, 0xfa, 0x5d, 0xc9, 0x3c, 0x26, 0x03,
-	0x0b, 0xd9, 0x1d, 0x3a, 0x08, 0x0b, 0xf7, 0xb8, 0x86, 0xf0, 0x53, 0xe8, 0xab, 0x20, 0xe6, 0x52,
-	0xb1, 0x38, 0x25, 0x1b, 0x16, 0xb2, 0x47, 0x74, 0x09, 0x1c, 0xbd, 0xbc, 0xfa, 0x63, 0xb6, 0xae,
-	0x6e, 0x4c, 0x74, 0x7d, 0x63, 0xa2, 0xdf, 0x37, 0x26, 0xfa, 0x76, 0x6b, 0xb6, 0xae, 0x6f, 0xcd,
-	0xd6, 0xcf, 0x5b, 0xb3, 0xf5, 0xb9, 0xf9, 0x60, 0x67, 0x1d, 0xfd, 0x49, 0x78, 0xfd, 0x37, 0x00,
-	0x00, 0xff, 0xff, 0xda, 0x1a, 0x6c, 0x68, 0x71, 0x04, 0x00, 0x00,
+	// 629 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0xcd, 0x6e, 0xd3, 0x40,
+	0x10, 0xce, 0xc6, 0xcd, 0xdf, 0x24, 0x2d, 0xe9, 0xaa, 0x2d, 0x4b, 0x80, 0xc8, 0x54, 0x1c, 0x8c,
+	0x68, 0x13, 0xa9, 0x5c, 0x50, 0xc5, 0xa9, 0x3f, 0x07, 0x54, 0x55, 0x54, 0x2b, 0xb8, 0x70, 0xb1,
+	0x1c, 0x67, 0x13, 0x5b, 0xfe, 0x59, 0xcb, 0xbb, 0x36, 0xcd, 0x43, 0x20, 0xf1, 0x22, 0x88, 0x27,
+	0xe0, 0xde, 0x63, 0x8f, 0x1c, 0xa1, 0x7d, 0x11, 0xb4, 0x6b, 0x57, 0x71, 0x51, 0x54, 0x71, 0x9b,
+	0xf9, 0xbe, 0x6f, 0xbe, 0xc9, 0xcc, 0xc4, 0x0b, 0x7b, 0xa1, 0x3f, 0xf7, 0x64, 0xec, 0xc7, 0xf3,
+	0xb1, 0xeb, 0x31, 0x37, 0x48, 0xb8, 0x1f, 0x4b, 0x31, 0x9e, 0xf9, 0x21, 0xb3, 0x2b, 0xc0, 0x28,
+	0x49, 0xb9, 0xe4, 0x83, 0xfd, 0xb9, 0x2f, 0xbd, 0x6c, 0x32, 0x72, 0x79, 0x34, 0x9e, 0xf3, 0x39,
+	0x1f, 0x6b, 0x78, 0x92, 0xcd, 0x74, 0xa6, 0x13, 0x1d, 0x15, 0xf2, 0xdd, 0x1f, 0x08, 0xfa, 0xc7,
+	0x4b, 0x93, 0x73, 0x3e, 0x65, 0x21, 0x3e, 0x81, 0x6e, 0xc5, 0x98, 0x20, 0xd3, 0xb0, 0xba, 0x07,
+	0xbb, 0xa3, 0x7f, 0x75, 0x55, 0xe0, 0x34, 0x96, 0xe9, 0x82, 0x56, 0xcb, 0x06, 0x9f, 0xee, 0x39,
+	0x6b, 0x01, 0xee, 0x83, 0x11, 0xb0, 0x05, 0x41, 0x26, 0xb2, 0x3a, 0x54, 0x85, 0xf8, 0x35, 0x34,
+	0x72, 0x27, 0xcc, 0x18, 0xa9, 0x9b, 0xc8, 0xea, 0x1e, 0x6c, 0x8f, 0x3e, 0x3a, 0x93, 0x90, 0x2d,
+	0x0b, 0x75, 0x27, 0x5a, 0x68, 0x0e, 0xeb, 0x6f, 0xd1, 0xee, 0xd7, 0x3a, 0x6c, 0xad, 0xd2, 0x60,
+	0x0c, 0x6b, 0x9e, 0x23, 0x3c, 0x6d, 0xde, 0xa3, 0x3a, 0xc6, 0x3b, 0xd0, 0x14, 0xd2, 0x91, 0x99,
+	0x20, 0x86, 0x89, 0xac, 0x75, 0x5a, 0x66, 0xf8, 0x39, 0x80, 0x13, 0x86, 0xdc, 0xb5, 0x27, 0x8e,
+	0x60, 0x64, 0xcd, 0x44, 0x96, 0x41, 0x3b, 0x1a, 0x39, 0x72, 0x04, 0xc3, 0xef, 0xa0, 0xc5, 0xe2,
+	0xb9, 0x1f, 0x33, 0x41, 0xda, 0xe5, 0xf0, 0xab, 0x5a, 0x8e, 0x4e, 0x0b, 0x51, 0x31, 0xfc, 0x5d,
+	0x09, 0x26, 0xd0, 0x92, 0x4a, 0xfd, 0xfe, 0x84, 0x74, 0xb4, 0xf3, 0x5d, 0x3a, 0xa0, 0xd0, 0xab,
+	0x96, 0x54, 0xd7, 0xb1, 0x59, 0xac, 0x63, 0xef, 0xfe, 0x3a, 0x76, 0xca, 0x16, 0x0f, 0xec, 0xe3,
+	0x27, 0x82, 0xed, 0x95, 0xa2, 0xca, 0xf0, 0xe8, 0xde, 0xf0, 0x87, 0xd0, 0x74, 0xbd, 0x2c, 0x0e,
+	0x04, 0xa9, 0x97, 0xc3, 0xad, 0xac, 0x1f, 0x1d, 0x6b, 0x51, 0x31, 0x5c, 0x59, 0x31, 0xb8, 0x80,
+	0x6e, 0x05, 0xfe, 0x9f, 0x7b, 0x6a, 0xf9, 0x03, 0xbf, 0xff, 0xbb, 0x01, 0x5b, 0xab, 0x34, 0xea,
+	0x9e, 0x89, 0x23, 0xbd, 0xd2, 0x5c, 0xc7, 0x6a, 0x24, 0x3e, 0x9b, 0x09, 0x26, 0xb5, 0xbd, 0x41,
+	0xcb, 0x0c, 0xef, 0x03, 0x76, 0x79, 0x98, 0x45, 0xb1, 0x9d, 0xb0, 0x34, 0xca, 0xa4, 0x23, 0x7d,
+	0x1e, 0x93, 0x9e, 0x69, 0x58, 0x0d, 0xba, 0x59, 0x30, 0x17, 0x4b, 0x42, 0x9d, 0x9f, 0xc5, 0x53,
+	0xbb, 0xb4, 0x6a, 0x14, 0xe7, 0x67, 0xf1, 0xf4, 0x43, 0xe1, 0xd6, 0x07, 0x23, 0xe1, 0x82, 0x34,
+	0x35, 0xae, 0x42, 0xfc, 0x12, 0x36, 0x92, 0x94, 0xe5, 0x76, 0xca, 0xbf, 0xf8, 0x53, 0x3b, 0x72,
+	0x2e, 0x49, 0x4b, 0x93, 0x3d, 0x85, 0x52, 0x05, 0x9e, 0x3b, 0x97, 0xf8, 0x29, 0x74, 0x96, 0x82,
+	0xb6, 0x16, 0xb4, 0xd3, 0x0a, 0x19, 0xe4, 0xae, 0x3d, 0x59, 0x48, 0x26, 0xf4, 0xff, 0x62, 0x8d,
+	0xb6, 0x83, 0xdc, 0x3d, 0x52, 0x39, 0x7e, 0x0c, 0x2d, 0x45, 0x06, 0xb9, 0x20, 0xa0, 0xa9, 0x66,
+	0x90, 0xbb, 0x67, 0xb9, 0xc0, 0x2f, 0xa0, 0xa7, 0x08, 0xfd, 0x5d, 0x89, 0x2c, 0x22, 0x5d, 0x13,
+	0x59, 0x4d, 0xda, 0x0d, 0x72, 0xf7, 0xb8, 0x84, 0xf0, 0x33, 0xe8, 0x48, 0x3f, 0x62, 0x42, 0x3a,
+	0x51, 0x42, 0xd6, 0x4d, 0x64, 0xf5, 0xe9, 0x12, 0x50, 0x5b, 0x94, 0x8b, 0x84, 0x91, 0x0d, 0x13,
+	0x59, 0x0d, 0xaa, 0x63, 0x6c, 0x42, 0xd7, 0xe5, 0x51, 0x92, 0x32, 0x21, 0xd4, 0x9a, 0x1e, 0x69,
+	0xaa, 0x0a, 0xe1, 0x27, 0xd0, 0x16, 0x3c, 0x95, 0xb6, 0x3a, 0x6e, 0x5f, 0xef, 0xbf, 0xa5, 0xf2,
+	0x33, 0xb6, 0x38, 0x7a, 0x75, 0xf5, 0x67, 0x58, 0xbb, 0xba, 0x19, 0xa2, 0xeb, 0x9b, 0x21, 0xfa,
+	0x7d, 0x33, 0x44, 0xdf, 0x6e, 0x87, 0xb5, 0xeb, 0xdb, 0x61, 0xed, 0xd7, 0xed, 0xb0, 0xf6, 0xb9,
+	0xfa, 0x02, 0x4c, 0x9a, 0xfa, 0x8d, 0x79, 0xf3, 0x37, 0x00, 0x00, 0xff, 0xff, 0xd8, 0xcb, 0xe2,
+	0xb3, 0xc2, 0x04, 0x00, 0x00,
 }
 
 func (m *CheckpointsModel) Marshal() (dAtA []byte, err error) {
@@ -437,6 +442,25 @@ func (m *ChunkCheckpointModel) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if len(m.SortKey) > 0 {
+		i -= len(m.SortKey)
+		copy(dAtA[i:], m.SortKey)
+		i = encodeVarintFileCheckpoints(dAtA, i, uint64(len(m.SortKey)))
+		i--
+		dAtA[i] = 0x1
+		i--
+		dAtA[i] = 0x82
+	}
+	if m.Compression != 0 {
+		i = encodeVarintFileCheckpoints(dAtA, i, uint64(m.Compression))
+		i--
+		dAtA[i] = 0x78
+	}
+	if m.Type != 0 {
+		i = encodeVarintFileCheckpoints(dAtA, i, uint64(m.Type))
+		i--
+		dAtA[i] = 0x70
+	}
 	if m.Timestamp != 0 {
 		i -= 8
 		encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(m.Timestamp))
@@ -650,6 +674,16 @@ func (m *ChunkCheckpointModel) Size() (n int) {
 	if m.Timestamp != 0 {
 		n += 9
 	}
+	if m.Type != 0 {
+		n += 1 + sovFileCheckpoints(uint64(m.Type))
+	}
+	if m.Compression != 0 {
+		n += 1 + sovFileCheckpoints(uint64(m.Compression))
+	}
+	l = len(m.SortKey)
+	if l > 0 {
+		n += 2 + l + sovFileCheckpoints(uint64(l))
+	}
 	return n
 }
 
@@ -1594,6 +1628,76 @@ func (m *ChunkCheckpointModel) Unmarshal(dAtA []byte) error {
 			}
 			m.Timestamp = int64(encoding_binary.LittleEndian.Uint64(dAtA[iNdEx:]))
 			iNdEx += 8
+		case 14:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Type", wireType)
+			}
+			m.Type = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowFileCheckpoints
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Type |= int32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 15:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Compression", wireType)
+			}
+			m.Compression = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowFileCheckpoints
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Compression |= int32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 16:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field SortKey", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowFileCheckpoints
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthFileCheckpoints
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthFileCheckpoints
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.SortKey = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipFileCheckpoints(dAtA[iNdEx:])
diff --git a/lightning/checkpoints/file_checkpoints.proto b/lightning/checkpoints/file_checkpoints.proto
index 9e11173..e854dac 100644
--- a/lightning/checkpoints/file_checkpoints.proto
+++ b/lightning/checkpoints/file_checkpoints.proto
@@ -49,4 +49,7 @@ message ChunkCheckpointModel {
     uint64 kvc_kvs = 10;
     fixed64 kvc_checksum = 11;
     sfixed64 timestamp = 13;
+    int32 type = 14;
+    int32 compression = 15;
+    string sort_key = 16;
 }
diff --git a/lightning/config/config.go b/lightning/config/config.go
index 7bdc2b1..f6f48a8 100644
--- a/lightning/config/config.go
+++ b/lightning/config/config.go
@@ -140,17 +140,29 @@ type CSVConfig struct {
 }
 
 type MydumperRuntime struct {
-	ReadBlockSize    int64     `toml:"read-block-size" json:"read-block-size"`
-	BatchSize        int64     `toml:"batch-size" json:"batch-size"`
-	BatchImportRatio float64   `toml:"batch-import-ratio" json:"batch-import-ratio"`
-	SourceDir        string    `toml:"data-source-dir" json:"data-source-dir"`
-	NoSchema         bool      `toml:"no-schema" json:"no-schema"`
-	CharacterSet     string    `toml:"character-set" json:"character-set"`
-	CSV              CSVConfig `toml:"csv" json:"csv"`
-	CaseSensitive    bool      `toml:"case-sensitive" json:"case-sensitive"`
-	StrictFormat     bool      `toml:"strict-format" json:"strict-format"`
-	MaxRegionSize    int64     `toml:"max-region-size" json:"max-region-size"`
-	Filter           []string  `toml:"filter" json:"filter"`
+	ReadBlockSize    int64            `toml:"read-block-size" json:"read-block-size"`
+	BatchSize        int64            `toml:"batch-size" json:"batch-size"`
+	BatchImportRatio float64          `toml:"batch-import-ratio" json:"batch-import-ratio"`
+	SourceDir        string           `toml:"data-source-dir" json:"data-source-dir"`
+	NoSchema         bool             `toml:"no-schema" json:"no-schema"`
+	CharacterSet     string           `toml:"character-set" json:"character-set"`
+	CSV              CSVConfig        `toml:"csv" json:"csv"`
+	CaseSensitive    bool             `toml:"case-sensitive" json:"case-sensitive"`
+	StrictFormat     bool             `toml:"strict-format" json:"strict-format"`
+	MaxRegionSize    int64            `toml:"max-region-size" json:"max-region-size"`
+	Filter           []string         `toml:"filter" json:"filter"`
+	FileRouters      []*FileRouteRule `toml:"files" json:"files"`
+	DefaultFileRules bool             `toml:"default-file-rules" json:"default-file-rules"`
+}
+
+type FileRouteRule struct {
+	Pattern     string `json:"pattern" toml:"pattern" yaml:"pattern"`
+	Path        string `json:"path" toml:"path" yaml:"path"`
+	Schema      string `json:"schema" toml:"schema" yaml:"schema"`
+	Table       string `json:"table" toml:"table" yaml:"table"`
+	Type        string `json:"type" toml:"type" yaml:"type"`
+	Key         string `json:"key" toml:"key" yaml:"key"`
+	Compression string `json:"compression" toml:"compression" yaml:"compression"`
 }
 
 type TikvImporter struct {
@@ -387,6 +399,11 @@ func (cfg *Config) Adjust() error {
 		}
 	}
 
+	// enable default file route rule if no rules are set
+	if len(cfg.Mydumper.FileRouters) == 0 {
+		cfg.Mydumper.DefaultFileRules = true
+	}
+
 	cfg.TikvImporter.Backend = strings.ToLower(cfg.TikvImporter.Backend)
 	mustHaveInternalConnections := true
 	switch cfg.TikvImporter.Backend {
diff --git a/lightning/mydump/loader.go b/lightning/mydump/loader.go
index 4c8a0c1..e141fe7 100644
--- a/lightning/mydump/loader.go
+++ b/lightning/mydump/loader.go
@@ -16,15 +16,13 @@ package mydump
 import (
 	"os"
 	"path/filepath"
-	"regexp"
 	"sort"
-	"strings"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb-lightning/lightning/common"
 	"github.com/pingcap/tidb-lightning/lightning/config"
 	"github.com/pingcap/tidb-lightning/lightning/log"
-	"github.com/pingcap/tidb-tools/pkg/table-filter"
+	filter "github.com/pingcap/tidb-tools/pkg/table-filter"
 	router "github.com/pingcap/tidb-tools/pkg/table-router"
 	"go.uber.org/zap"
 )
@@ -40,11 +38,18 @@ type MDTableMeta struct {
 	DB         string
 	Name       string
 	SchemaFile string
-	DataFiles  []string
+	DataFiles  []FileInfo
 	charSet    string
 	TotalSize  int64
 }
 
+type SourceFileMeta struct {
+	Path        string
+	Type        SourceType
+	Compression Compression
+	SortKey     string
+}
+
 func (m *MDTableMeta) GetSchema() string {
 	schema, err := ExportStatement(m.SchemaFile, m.charSet)
 	if err != nil {
@@ -61,19 +66,20 @@ func (m *MDTableMeta) GetSchema() string {
 	Mydumper File Loader
 */
 type MDLoader struct {
-	dir      string
-	noSchema bool
-	dbs      []*MDDatabaseMeta
-	filter   filter.Filter
-	router   *router.Table
-	charSet  string
+	dir        string
+	noSchema   bool
+	dbs        []*MDDatabaseMeta
+	filter     filter.Filter
+	router     *router.Table
+	fileRouter FileRouter
+	charSet    string
 }
 
 type mdLoaderSetup struct {
 	loader        *MDLoader
-	dbSchemas     []fileInfo
-	tableSchemas  []fileInfo
-	tableDatas    []fileInfo
+	dbSchemas     []FileInfo
+	tableSchemas  []FileInfo
+	tableDatas    []FileInfo
 	dbIndexMap    map[string]int
 	tableIndexMap map[filter.Table]int
 }
@@ -81,6 +87,11 @@ type mdLoaderSetup struct {
 func NewMyDumpLoader(cfg *config.Config) (*MDLoader, error) {
 	var r *router.Table
 	var err error
+
+	if len(cfg.Routes) > 0 && len(cfg.Mydumper.FileRouters) > 0 {
+		return nil, errors.New("table route is deprecated, can't config both [routes] and [mydumper.files]")
+	}
+
 	if len(cfg.Routes) > 0 {
 		r, err = router.NewTableRouter(cfg.Mydumper.CaseSensitive, cfg.Routes)
 		if err != nil {
@@ -102,12 +113,23 @@ func NewMyDumpLoader(cfg *config.Config) (*MDLoader, error) {
 		f = filter.CaseInsensitive(f)
 	}
 
+	fileRouteRules := cfg.Mydumper.FileRouters
+	if cfg.Mydumper.DefaultFileRules {
+		fileRouteRules = append(fileRouteRules, defaultFileRouteRules...)
+	}
+
+	fileRouter, err := NewFileRouter(fileRouteRules)
+	if err != nil {
+		return nil, err
+	}
+
 	mdl := &MDLoader{
-		dir:      cfg.Mydumper.SourceDir,
-		noSchema: cfg.Mydumper.NoSchema,
-		filter:   f,
-		router:   r,
-		charSet:  cfg.Mydumper.CharacterSet,
+		dir:        cfg.Mydumper.SourceDir,
+		noSchema:   cfg.Mydumper.NoSchema,
+		filter:     f,
+		router:     r,
+		charSet:    cfg.Mydumper.CharacterSet,
+		fileRouter: fileRouter,
 	}
 
 	setup := mdLoaderSetup{
@@ -144,14 +166,12 @@ func (ftype fileType) String() string {
 	}
 }
 
-type fileInfo struct {
-	tableName filter.Table
-	path      string
-	size      int64
+type FileInfo struct {
+	TableName filter.Table
+	FileMeta  SourceFileMeta
+	Size      int64
 }
 
-var tableNameRegexp = regexp.MustCompile(`^([^.]+)\.(.*?)(?:\.[0-9]+)?$`)
-
 // setup the `s.loader.dbs` slice by scanning all *.sql files inside `dir`.
 //
 // The database and tables are inserted in a consistent order, so creating an
@@ -190,48 +210,57 @@ func (s *mdLoaderSetup) setup(dir string) error {
 			return errors.New("missing {schema}-schema-create.sql")
 		}
 		for _, fileInfo := range s.dbSchemas {
-			if _, dbExists := s.insertDB(fileInfo.tableName.Schema, fileInfo.path); dbExists && s.loader.router == nil {
-				return errors.Errorf("invalid database schema file, duplicated item - %s", fileInfo.path)
+			if _, dbExists := s.insertDB(fileInfo.TableName.Schema, fileInfo.FileMeta.Path); dbExists && s.loader.router == nil {
+				return errors.Errorf("invalid database schema file, duplicated item - %s", fileInfo.FileMeta.Path)
 			}
 		}
 
 		// setup table schema
 		for _, fileInfo := range s.tableSchemas {
-			_, dbExists, tableExists := s.insertTable(fileInfo.tableName, fileInfo.path)
+			_, dbExists, tableExists := s.insertTable(fileInfo.TableName, fileInfo.FileMeta.Path)
 			if !dbExists {
-				return errors.Errorf("invalid table schema file, cannot find db - %s", fileInfo.path)
+				return errors.Errorf("invalid table schema file, cannot find db '%s' - %s", fileInfo.TableName.Schema, fileInfo.FileMeta.Path)
 			} else if tableExists && s.loader.router == nil {
-				return errors.Errorf("invalid table schema file, duplicated item - %s", fileInfo.path)
+				return errors.Errorf("invalid table schema file, duplicated item - %s", fileInfo.FileMeta.Path)
 			}
 		}
 	}
 
 	// Sql file for restore data
 	for _, fileInfo := range s.tableDatas {
-		tableMeta, dbExists, tableExists := s.insertTable(fileInfo.tableName, "")
+		tableMeta, dbExists, tableExists := s.insertTable(fileInfo.TableName, "")
 		if !s.loader.noSchema {
 			if !dbExists {
-				return errors.Errorf("invalid data file, miss host db - %s", fileInfo.path)
+				return errors.Errorf("invalid data file, miss host db '%s' - %s", fileInfo.TableName.Schema, fileInfo.FileMeta.Path)
 			} else if !tableExists {
-				return errors.Errorf("invalid data file, miss host table - %s", fileInfo.path)
+				return errors.Errorf("invalid data file, miss host table '%s' - %s", fileInfo.TableName.Name, fileInfo.FileMeta.Path)
 			}
 		}
-		tableMeta.DataFiles = append(tableMeta.DataFiles, fileInfo.path)
-		tableMeta.TotalSize += fileInfo.size
+		tableMeta.DataFiles = append(tableMeta.DataFiles, fileInfo)
+		tableMeta.TotalSize += fileInfo.Size
 	}
 
-	// Put the small table in the front of the slice which can avoid large table
-	// take a long time to import and block small table to release index worker.
 	for _, dbMeta := range s.loader.dbs {
+		// Put the small table in the front of the slice which can avoid large table
+		// take a long time to import and block small table to release index worker.
 		sort.SliceStable(dbMeta.Tables, func(i, j int) bool {
 			return dbMeta.Tables[i].TotalSize < dbMeta.Tables[j].TotalSize
 		})
+
+		// sort each table source files by sort-key
+		for _, tbMeta := range dbMeta.Tables {
+			dataFiles := tbMeta.DataFiles
+			sort.SliceStable(dataFiles, func(i, j int) bool {
+				return dataFiles[i].FileMeta.SortKey < dataFiles[j].FileMeta.SortKey
+			})
+		}
 	}
 
 	return nil
 }
 
 func (s *mdLoaderSetup) listFiles(dir string) error {
+	fr := s.loader.fileRouter
 	// `filepath.Walk` yields the paths in a deterministic (lexicographical) order,
 	// meaning the file and chunk orders will be the same everytime it is called
 	// (as long as the source is immutable).
@@ -244,61 +273,41 @@ func (s *mdLoaderSetup) listFiles(dir string) error {
 			return nil
 		}
 
-		fname := strings.TrimSpace(f.Name())
-		lowerFName := strings.ToLower(fname)
-
-		info := fileInfo{path: path, size: f.Size()}
 		logger := log.With(zap.String("path", path))
 
-		var (
-			ftype         fileType
-			qualifiedName string
-		)
-		switch {
-		case strings.HasSuffix(lowerFName, "-schema-create.sql"):
-			ftype = fileTypeDatabaseSchema
-			qualifiedName = fname[:len(fname)-18] + "."
-
-		case strings.HasSuffix(lowerFName, "-schema.sql"):
-			ftype = fileTypeTableSchema
-			qualifiedName = fname[:len(fname)-11]
-
-			// ignore functionality :
-			// 		- view
-			//		- triggers
-		case strings.HasSuffix(lowerFName, "-schema-view.sql"),
-			strings.HasSuffix(lowerFName, "-schema-trigger.sql"),
-			strings.HasSuffix(lowerFName, "-schema-post.sql"):
-			logger.Warn("[loader] ignore unsupport view/trigger")
-			return nil
-		case strings.HasSuffix(lowerFName, ".sql"), strings.HasSuffix(lowerFName, ".csv"):
-			ftype = fileTypeTableData
-			qualifiedName = fname[:len(fname)-4]
-		default:
+		res, err := fr.Route(path)
+		if err != nil {
+			return errors.Annotatef(err, "apply file routing on file '%s' failed", path)
+		}
+		if res == nil {
+			logger.Info("[loader] file is filtered by file router")
 			return nil
 		}
 
-		matchRes := tableNameRegexp.FindStringSubmatch(qualifiedName)
-		if len(matchRes) != 3 {
-			logger.Debug("[loader] ignore almost " + ftype.String() + " file")
-			return nil
+		info := FileInfo{
+			TableName: filter.Table{Schema: res.Schema, Name: res.Name},
+			FileMeta:  SourceFileMeta{Path: path, Type: res.Type, Compression: res.Compression, SortKey: res.Key},
+			Size:      f.Size(),
 		}
-		info.tableName.Schema = matchRes[1]
-		info.tableName.Name = matchRes[2]
 
-		if s.loader.shouldSkip(&info.tableName) {
+		if s.loader.shouldSkip(&info.TableName) {
 			logger.Debug("[filter] ignoring table file")
+
 			return nil
 		}
 
-		switch ftype {
-		case fileTypeDatabaseSchema:
+		switch res.Type {
+		case SourceTypeSchemaSchema:
 			s.dbSchemas = append(s.dbSchemas, info)
-		case fileTypeTableSchema:
+		case SourceTypeTableSchema:
 			s.tableSchemas = append(s.tableSchemas, info)
-		case fileTypeTableData:
+		case SourceTypeSQL, SourceTypeCSV:
 			s.tableDatas = append(s.tableDatas, info)
 		}
+
+		logger.Info("file route result", zap.String("schema", res.Schema),
+			zap.String("table", res.Name), zap.Stringer("type", res.Type))
+
 		return nil
 	})
 
@@ -319,46 +328,46 @@ func (s *mdLoaderSetup) route() error {
 	}
 
 	type dbInfo struct {
-		path  string
-		count int
+		fileMeta SourceFileMeta
+		count    int
 	}
 
 	knownDBNames := make(map[string]dbInfo)
 	for _, info := range s.dbSchemas {
-		knownDBNames[info.tableName.Schema] = dbInfo{
-			path:  info.path,
-			count: 1,
+		knownDBNames[info.TableName.Schema] = dbInfo{
+			fileMeta: info.FileMeta,
+			count:    1,
 		}
 	}
 	for _, info := range s.tableSchemas {
-		dbInfo := knownDBNames[info.tableName.Schema]
+		dbInfo := knownDBNames[info.TableName.Schema]
 		dbInfo.count++
-		knownDBNames[info.tableName.Schema] = dbInfo
+		knownDBNames[info.TableName.Schema] = dbInfo
 	}
 
-	run := func(arr []fileInfo) error {
+	run := func(arr []FileInfo) error {
 		for i, info := range arr {
-			dbName, tableName, err := r.Route(info.tableName.Schema, info.tableName.Name)
+			dbName, tableName, err := r.Route(info.TableName.Schema, info.TableName.Name)
 			if err != nil {
 				return errors.Trace(err)
 			}
-			if dbName != info.tableName.Schema {
-				oldInfo := knownDBNames[info.tableName.Schema]
+			if dbName != info.TableName.Schema {
+				oldInfo := knownDBNames[info.TableName.Schema]
 				oldInfo.count--
-				knownDBNames[info.tableName.Schema] = oldInfo
+				knownDBNames[info.TableName.Schema] = oldInfo
 
 				newInfo, ok := knownDBNames[dbName]
 				newInfo.count++
 				if !ok {
-					newInfo.path = oldInfo.path
-					s.dbSchemas = append(s.dbSchemas, fileInfo{
-						tableName: filter.Table{Schema: dbName},
-						path:      oldInfo.path,
+					newInfo.fileMeta = oldInfo.fileMeta
+					s.dbSchemas = append(s.dbSchemas, FileInfo{
+						TableName: filter.Table{Schema: dbName},
+						FileMeta:  oldInfo.fileMeta,
 					})
 				}
 				knownDBNames[dbName] = newInfo
 			}
-			arr[i].tableName = filter.Table{Schema: dbName, Name: tableName}
+			arr[i].TableName = filter.Table{Schema: dbName, Name: tableName}
 		}
 		return nil
 	}
@@ -374,7 +383,7 @@ func (s *mdLoaderSetup) route() error {
 	// https://github.com/golang/go/wiki/SliceTricks#filtering-without-allocating
 	remainingSchemas := s.dbSchemas[:0]
 	for _, info := range s.dbSchemas {
-		if knownDBNames[info.tableName.Schema].count > 0 {
+		if knownDBNames[info.TableName.Schema].count > 0 {
 			remainingSchemas = append(remainingSchemas, info)
 		}
 	}
@@ -410,7 +419,7 @@ func (s *mdLoaderSetup) insertTable(tableName filter.Table, path string) (*MDTab
 			DB:         tableName.Schema,
 			Name:       tableName.Name,
 			SchemaFile: path,
-			DataFiles:  make([]string, 0, 16),
+			DataFiles:  make([]FileInfo, 0, 16),
 			charSet:    s.loader.charSet,
 		}
 		dbMeta.Tables = append(dbMeta.Tables, ptr)
diff --git a/lightning/mydump/reader.go b/lightning/mydump/reader.go
index 66a9221..a13afcb 100644
--- a/lightning/mydump/reader.go
+++ b/lightning/mydump/reader.go
@@ -21,10 +21,11 @@ import (
 	"strings"
 	"unicode/utf8"
 
+	"go.uber.org/zap"
+
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb-lightning/lightning/log"
 	"github.com/pingcap/tidb-lightning/lightning/worker"
-	"go.uber.org/zap"
 	"golang.org/x/text/encoding/simplifiedchinese"
 )
 
diff --git a/lightning/mydump/region.go b/lightning/mydump/region.go
index 9e16b0a..b609bd3 100644
--- a/lightning/mydump/region.go
+++ b/lightning/mydump/region.go
@@ -16,9 +16,7 @@ package mydump
 import (
 	"math"
 	"os"
-	"strings"
 
-	"github.com/pingcap/errors"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tidb-lightning/lightning/config"
@@ -31,9 +29,9 @@ const tableRegionSizeWarningThreshold int64 = 1024 * 1024 * 1024
 type TableRegion struct {
 	EngineID int32
 
-	DB    string
-	Table string
-	File  string
+	DB       string
+	Table    string
+	FileMeta SourceFileMeta
 
 	Chunk Chunk
 }
@@ -51,21 +49,6 @@ func (reg *TableRegion) Size() int64 {
 	return reg.Chunk.EndOffset - reg.Chunk.Offset
 }
 
-type regionSlice []*TableRegion
-
-func (rs regionSlice) Len() int {
-	return len(rs)
-}
-func (rs regionSlice) Swap(i, j int) {
-	rs[i], rs[j] = rs[j], rs[i]
-}
-func (rs regionSlice) Less(i, j int) bool {
-	if rs[i].File == rs[j].File {
-		return rs[i].Chunk.Offset < rs[j].Chunk.Offset
-	}
-	return rs[i].File < rs[j].File
-}
-
 ////////////////////////////////////////////////////////////////
 
 func AllocateEngineIDs(
@@ -146,18 +129,15 @@ func MakeTableRegions(
 	ioWorkers *worker.Pool,
 ) ([]*TableRegion, error) {
 	// Split files into regions
-	filesRegions := make(regionSlice, 0, len(meta.DataFiles))
+	filesRegions := make([]*TableRegion, 0, len(meta.DataFiles))
 	dataFileSizes := make([]float64, 0, len(meta.DataFiles))
 	prevRowIDMax := int64(0)
+	var err error
 	for _, dataFile := range meta.DataFiles {
-		dataFileInfo, err := os.Stat(dataFile)
-		if err != nil {
-			return nil, errors.Annotatef(err, "cannot stat %s", dataFile)
-		}
-		dataFileSize := dataFileInfo.Size()
+		dataFileSize := dataFile.Size
 
 		divisor := int64(columns)
-		isCsvFile := strings.HasSuffix(strings.ToLower(dataFile), ".csv")
+		isCsvFile := dataFile.FileMeta.Type == SourceTypeCSV
 		if !isCsvFile {
 			divisor += 2
 		}
@@ -168,7 +148,7 @@ func MakeTableRegions(
 				regions      []*TableRegion
 				subFileSizes []float64
 			)
-			prevRowIDMax, regions, subFileSizes, err = SplitLargeFile(meta, cfg, dataFile, dataFileSize, divisor, prevRowIDMax, ioWorkers)
+			prevRowIDMax, regions, subFileSizes, err = SplitLargeFile(meta, cfg, dataFile, divisor, prevRowIDMax, ioWorkers)
 			if err != nil {
 				return nil, err
 			}
@@ -178,9 +158,9 @@ func MakeTableRegions(
 		}
 		rowIDMax := prevRowIDMax + dataFileSize/divisor
 		tableRegion := &TableRegion{
-			DB:    meta.DB,
-			Table: meta.Name,
-			File:  dataFile,
+			DB:       meta.DB,
+			Table:    meta.Name,
+			FileMeta: dataFile.FileMeta,
 			Chunk: Chunk{
 				Offset:       0,
 				EndOffset:    dataFileSize,
@@ -192,7 +172,7 @@ func MakeTableRegions(
 		if tableRegion.Size() > tableRegionSizeWarningThreshold {
 			log.L().Warn(
 				"file is too big to be processed efficiently; we suggest splitting it at 256 MB each",
-				zap.String("file", dataFile),
+				zap.String("file", dataFile.FileMeta.Path),
 				zap.Int64("size", dataFileSize))
 		}
 		prevRowIDMax = rowIDMax
@@ -217,18 +197,17 @@ func MakeTableRegions(
 func SplitLargeFile(
 	meta *MDTableMeta,
 	cfg *config.Config,
-	dataFilePath string,
-	dataFileSize int64,
+	dataFile FileInfo,
 	divisor int64,
 	prevRowIdxMax int64,
 	ioWorker *worker.Pool,
 ) (prevRowIdMax int64, regions []*TableRegion, dataFileSizes []float64, err error) {
 	maxRegionSize := cfg.Mydumper.MaxRegionSize
-	dataFileSizes = make([]float64, 0, dataFileSize/maxRegionSize+1)
+	dataFileSizes = make([]float64, 0, dataFile.Size/maxRegionSize+1)
 	startOffset, endOffset := int64(0), maxRegionSize
 	var columns []string
 	if cfg.Mydumper.CSV.Header {
-		reader, err := os.Open(dataFilePath)
+		reader, err := os.Open(dataFile.FileMeta.Path)
 		if err != nil {
 			return 0, nil, nil, err
 		}
@@ -243,8 +222,8 @@ func SplitLargeFile(
 	for {
 		curRowsCnt := (endOffset - startOffset) / divisor
 		rowIDMax := prevRowIdxMax + curRowsCnt
-		if endOffset != dataFileSize {
-			reader, err := os.Open(dataFilePath)
+		if endOffset != dataFile.Size {
+			reader, err := os.Open(dataFile.FileMeta.Path)
 			if err != nil {
 				return 0, nil, nil, err
 			}
@@ -259,9 +238,9 @@ func SplitLargeFile(
 		}
 		regions = append(regions,
 			&TableRegion{
-				DB:    meta.DB,
-				Table: meta.Name,
-				File:  dataFilePath,
+				DB:       meta.DB,
+				Table:    meta.Name,
+				FileMeta: dataFile.FileMeta,
 				Chunk: Chunk{
 					Offset:       startOffset,
 					EndOffset:    endOffset,
@@ -272,12 +251,12 @@ func SplitLargeFile(
 			})
 		dataFileSizes = append(dataFileSizes, float64(endOffset-startOffset))
 		prevRowIdxMax = rowIDMax
-		if endOffset == dataFileSize {
+		if endOffset == dataFile.Size {
 			break
 		}
 		startOffset = endOffset
-		if endOffset += maxRegionSize; endOffset > dataFileSize {
-			endOffset = dataFileSize
+		if endOffset += maxRegionSize; endOffset > dataFile.Size {
+			endOffset = dataFile.Size
 		}
 	}
 	return prevRowIdxMax, regions, dataFileSizes, nil
diff --git a/lightning/mydump/router.go b/lightning/mydump/router.go
new file mode 100644
index 0000000..6d62380
--- /dev/null
+++ b/lightning/mydump/router.go
@@ -0,0 +1,326 @@
+package mydump
+
+import (
+	"regexp"
+	"strconv"
+	"strings"
+
+	"github.com/pingcap/pd/v4/pkg/slice"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb-lightning/lightning/config"
+	"github.com/pingcap/tidb-tools/pkg/filter"
+)
+
+type SourceType int
+
+const (
+	SourceTypeIgnore SourceType = iota
+	SourceTypeSchemaSchema
+	SourceTypeTableSchema
+	SourceTypeSQL
+	SourceTypeCSV
+)
+
+const (
+	SchemaSchema = "schema-schema"
+	TableSchema  = "table-schema"
+	TypeSQL      = "sql"
+	TypeCSV      = "csv"
+	TypeIgnore   = "ignore"
+)
+
+type Compression int
+
+const (
+	CompressionNone Compression = iota
+	CompressionGZ
+	CompressionLZ4
+	CompressionZStd
+	CompressionXZ
+)
+
+func parseSourceType(t string) (SourceType, error) {
+	switch strings.ToLower(strings.TrimSpace(t)) {
+	case SchemaSchema:
+		return SourceTypeSchemaSchema, nil
+	case TableSchema:
+		return SourceTypeTableSchema, nil
+	case TypeSQL:
+		return SourceTypeSQL, nil
+	case TypeCSV:
+		return SourceTypeCSV, nil
+	case TypeIgnore:
+		return SourceTypeIgnore, nil
+	default:
+		return SourceTypeIgnore, errors.Errorf("unknown source type '%s'", t)
+	}
+}
+
+func (s SourceType) String() string {
+	switch s {
+	case SourceTypeSchemaSchema:
+		return SchemaSchema
+	case SourceTypeTableSchema:
+		return TableSchema
+	case SourceTypeCSV:
+		return TypeCSV
+	case SourceTypeSQL:
+		return TypeSQL
+	default:
+		return TypeIgnore
+	}
+}
+
+func parseCompressionType(t string) (Compression, error) {
+	switch strings.ToLower(strings.TrimSpace(t)) {
+	case "gz":
+		return CompressionGZ, nil
+	case "lz4":
+		return CompressionLZ4, nil
+	case "zstd":
+		return CompressionZStd, nil
+	case "xz":
+		return CompressionXZ, nil
+	case "":
+		return CompressionNone, nil
+	default:
+		return CompressionNone, errors.Errorf("invalid compression type '%s'", t)
+	}
+}
+
+var (
+	expandVariablePattern = regexp.MustCompile(`\$(?:\$|[\pL\p{Nd}_]+|\{[\pL\p{Nd}_]+\})`)
+)
+
+var (
+	defaultFileRouteRules = []*config.FileRouteRule{
+		// ignore *-schema-view.sql,-schema-trigger.sql,-schema-post.sql files
+		{Pattern: `(?i).*(-schema-view|-schema-trigger|-schema-post)\.sql`, Type: "ignore"},
+		// db schema create file pattern, matches files like '{schema}-schema-create.sql'
+		{Pattern: `(?i)^(?:[^/]*/)*([^/.]+)-schema-create\.sql`, Schema: "$1", Table: "", Type: SchemaSchema},
+		// table schema create file pattern, matches files like '{schema}.{table}-schema.sql'
+		{Pattern: `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema\.sql`, Schema: "$1", Table: "$2", Type: TableSchema},
+		// source file pattern, matches files like '{schema}.{table}.0001.{sql|csv}'
+		{Pattern: `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)(?:\.([0-9]+))?\.(sql|csv)$`, Schema: "$1", Table: "$2", Type: "$4", Key: "$3"},
+	}
+)
+
+// // RouteRule is a rule to route file path to target schema/table
+type FileRouter interface {
+	// Route apply rule to path. Return nil if path doesn't math route rule;
+	// return error if path match route rule but the captured value for field is invalid
+	Route(path string) (*RouteResult, error)
+}
+
+// chainRouters aggregates multi `FileRouter` as a router
+type chainRouters []FileRouter
+
+func (c chainRouters) Route(path string) (*RouteResult, error) {
+	for _, r := range c {
+		res, err := r.Route(path)
+		if err != nil {
+			return nil, err
+		}
+		if res != nil {
+			return res, nil
+		}
+	}
+	return nil, nil
+}
+
+func NewFileRouter(cfg []*config.FileRouteRule) (FileRouter, error) {
+	res := make([]FileRouter, 0, len(cfg))
+	p := regexRouterParser{}
+	for _, c := range cfg {
+		rule, err := p.Parse(c)
+		if err != nil {
+			return nil, err
+		}
+		res = append(res, rule)
+	}
+	return chainRouters(res), nil
+}
+
+// `RegexRouter` is a `FileRouter` implement that apply specific regex pattern to filepath.
+// if regex pattern match, then each extractors with capture the matched regexp pattern and
+// set value to target field in `RouteResult`
+type RegexRouter struct {
+	pattern    *regexp.Regexp
+	extractors []patExpander
+}
+
+func (r *RegexRouter) Route(path string) (*RouteResult, error) {
+	indexes := r.pattern.FindStringSubmatchIndex(path)
+	if len(indexes) == 0 {
+		return nil, nil
+	}
+	result := &RouteResult{}
+	for _, e := range r.extractors {
+		err := e.Expand(r.pattern, path, indexes, result)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return result, nil
+}
+
+type regexRouterParser struct{}
+
+func (p regexRouterParser) Parse(r *config.FileRouteRule) (*RegexRouter, error) {
+	rule := &RegexRouter{}
+	if r.Path != "" && r.Pattern != "" {
+		return nil, errors.New("can't set both `path` and `pattern` field in [[mydumper.files]]")
+	}
+	if r.Path != "" {
+		// convert constant string as a regexp pattern
+		r.Pattern = regexp.QuoteMeta(r.Path)
+		// escape all '$' by '$$' in match templates
+		quoteTmplFn := func(t string) string { return strings.ReplaceAll(t, "$", "$$") }
+		r.Table = quoteTmplFn(r.Table)
+		r.Schema = quoteTmplFn(r.Schema)
+		r.Type = quoteTmplFn(r.Type)
+		r.Compression = quoteTmplFn(r.Compression)
+		r.Key = quoteTmplFn(r.Key)
+
+	}
+	pattern, err := regexp.Compile(r.Pattern)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	rule.pattern = pattern
+
+	err = p.parseFieldExtractor(rule, "type", r.Type, func(result *RouteResult, value string) error {
+		ty, err := parseSourceType(value)
+		if err != nil {
+			return err
+		}
+		result.Type = ty
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	// ignore pattern needn't parse other fields
+	if r.Type == TypeIgnore {
+		return rule, nil
+	}
+
+	err = p.parseFieldExtractor(rule, "schema", r.Schema, func(result *RouteResult, value string) error {
+		result.Schema = value
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// special case: when the pattern is for db schema, should not parse table name
+	if r.Type != SchemaSchema {
+		err = p.parseFieldExtractor(rule, "table", r.Table, func(result *RouteResult, value string) error {
+			result.Name = value
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if len(r.Key) > 0 {
+		err = p.parseFieldExtractor(rule, "key", r.Key, func(result *RouteResult, value string) error {
+			result.Key = value
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if len(r.Compression) > 0 {
+		err = p.parseFieldExtractor(rule, "compression", r.Compression, func(result *RouteResult, value string) error {
+			// TODO: should support restore compressed source files
+			compression, err := parseCompressionType(value)
+			if err != nil {
+				return err
+			}
+			if compression != CompressionNone {
+				return errors.New("Currently we don't support restore compressed source file yet")
+
+			}
+			result.Compression = compression
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return rule, nil
+}
+
+// parse each field extractor in `p.r` and set them to p.rule
+func (p regexRouterParser) parseFieldExtractor(
+	rule *RegexRouter,
+	field,
+	fieldPattern string,
+	applyFn func(result *RouteResult, value string) error,
+) error {
+	// pattern is empty, return default rule
+	if len(fieldPattern) == 0 {
+		return errors.Errorf("field '%s' match pattern can't be empty", field)
+	}
+
+	// check and parse regexp template
+	if err := p.checkSubPatterns(rule.pattern, fieldPattern); err != nil {
+		return errors.Trace(err)
+	}
+	rule.extractors = append(rule.extractors, patExpander{
+		template: fieldPattern,
+		applyFn:  applyFn,
+	})
+	return nil
+}
+
+func (p regexRouterParser) checkSubPatterns(pat *regexp.Regexp, t string) error {
+	subPats := expandVariablePattern.FindAllString(t, -1)
+	for _, subVar := range subPats {
+		var tmplName string
+		switch {
+		case subVar == "$$":
+			continue
+		case strings.HasPrefix(subVar, "${"):
+			tmplName = subVar[2 : len(subVar)-1]
+		default:
+			tmplName = subVar[1:]
+		}
+		if number, err := strconv.Atoi(tmplName); err == nil {
+			if number > pat.NumSubexp() {
+				return errors.Errorf("sub pattern capture '%s' out of range", subVar)
+			}
+		} else if !slice.AnyOf(pat.SubexpNames(), func(i int) bool {
+			// FIXME: we should use re.SubexpIndex here, but not supported in go1.13 yet
+			return pat.SubexpNames()[i] == tmplName
+		}) {
+			return errors.Errorf("invalid named capture '%s'", subVar)
+		}
+	}
+
+	return nil
+}
+
+// patExpander extract string by expanding template with the regexp pattern
+type patExpander struct {
+	template string
+	applyFn  func(result *RouteResult, value string) error
+}
+
+func (p *patExpander) Expand(pattern *regexp.Regexp, path string, matchIndex []int, result *RouteResult) error {
+	value := pattern.ExpandString([]byte{}, p.template, path, matchIndex)
+	return p.applyFn(result, string(value))
+}
+
+type RouteResult struct {
+	filter.Table
+	Key         string
+	Compression Compression
+	Type        SourceType
+}
diff --git a/lightning/restore/restore.go b/lightning/restore/restore.go
index 9507f32..f01a2ff 100644
--- a/lightning/restore/restore.go
+++ b/lightning/restore/restore.go
@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"path"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -151,11 +150,20 @@ type RestoreController struct {
 	closedEngineLimit *worker.Pool
 }
 
-func NewRestoreController(ctx context.Context, dbMetas []*mydump.MDDatabaseMeta, cfg *config.Config) (*RestoreController, error) {
+func NewRestoreController(
+	ctx context.Context,
+	dbMetas []*mydump.MDDatabaseMeta,
+	cfg *config.Config,
+) (*RestoreController, error) {
 	return NewRestoreControllerWithPauser(ctx, dbMetas, cfg, DeliverPauser)
 }
 
-func NewRestoreControllerWithPauser(ctx context.Context, dbMetas []*mydump.MDDatabaseMeta, cfg *config.Config, pauser *common.Pauser) (*RestoreController, error) {
+func NewRestoreControllerWithPauser(
+	ctx context.Context,
+	dbMetas []*mydump.MDDatabaseMeta,
+	cfg *config.Config,
+	pauser *common.Pauser,
+) (*RestoreController, error) {
 	tls, err := cfg.ToTLS()
 	if err != nil {
 		return nil, err
@@ -339,14 +347,11 @@ func (rc *RestoreController) estimateChunkCountIntoMetrics() {
 	estimatedChunkCount := 0
 	for _, dbMeta := range rc.dbMetas {
 		for _, tableMeta := range dbMeta.Tables {
-			for _, tablePath := range tableMeta.DataFiles {
-				isCsvFile := strings.HasSuffix(strings.ToLower(tablePath), ".csv")
-				if isCsvFile {
-					f, _ := os.Stat(tablePath)
-					dataFileSize := f.Size()
+			for _, fileMeta := range tableMeta.DataFiles {
+				if fileMeta.FileMeta.Type == mydump.SourceTypeCSV {
 					cfg := rc.cfg.Mydumper
-					if dataFileSize > cfg.MaxRegionSize && cfg.StrictFormat && !cfg.CSV.Header {
-						estimatedChunkCount += int(dataFileSize / cfg.MaxRegionSize)
+					if fileMeta.Size > cfg.MaxRegionSize && cfg.StrictFormat && !cfg.CSV.Header {
+						estimatedChunkCount += int(fileMeta.Size / cfg.MaxRegionSize)
 					} else {
 						estimatedChunkCount += 1
 					}
@@ -1267,12 +1272,14 @@ func newChunkRestore(
 	}
 
 	var parser mydump.Parser
-	switch path.Ext(strings.ToLower(chunk.Key.Path)) {
-	case ".csv":
+	switch chunk.FileMeta.Type {
+	case mydump.SourceTypeCSV:
 		hasHeader := cfg.Mydumper.CSV.Header && chunk.Chunk.Offset == 0
 		parser = mydump.NewCSVParser(&cfg.Mydumper.CSV, reader, blockBufSize, ioWorkers, hasHeader)
-	default:
+	case mydump.SourceTypeSQL:
 		parser = mydump.NewChunkParser(cfg.TiDB.SQLMode, reader, blockBufSize, ioWorkers)
+	default:
+		panic(fmt.Sprintf("file '%s' with unknown source type '%s'", chunk.Key.Path, chunk.FileMeta.Type.String()))
 	}
 
 	if len(chunk.ColumnPermutation) > 0 {
@@ -1350,9 +1357,10 @@ func (t *TableRestore) populateChunks(rc *RestoreController, cp *TableCheckpoint
 			}
 			ccp := &ChunkCheckpoint{
 				Key: ChunkCheckpointKey{
-					Path:   chunk.File,
+					Path:   chunk.FileMeta.Path,
 					Offset: chunk.Chunk.Offset,
 				},
+				FileMeta:          chunk.FileMeta,
 				ColumnPermutation: nil,
 				Chunk:             chunk.Chunk,
 				Timestamp:         timestamp,
diff --git a/tidb-lightning.toml b/tidb-lightning.toml
index de88248..565710d 100644
--- a/tidb-lightning.toml
+++ b/tidb-lightning.toml
@@ -133,6 +133,16 @@ strict-format = false
 # will restore in parallel. The size of each chunk is `max-region-size`, where the default is 256 MiB.
 #max-region-size = 268_435_456
 
+# enable file router to use the default rules. By default, it will be set to true if no `mydumper.files`
+# rule is provided, else false. You can explicitly set it to `true` to enable the default rules, they will
+# take effect on files that on other rules are match.
+# The default file routing rules' behavior is the same as former versions without this conf, that is:
+#   {schema}-schema-create.sql --> schema create sql file
+#   {schema}.{table}-schema.sql --> table schema sql file
+#   {schema}.{table}.{0001}.{sql|csv} --> data source file
+#   *-schema-view.sql, *-schema-trigger.sql, *-schema-post.sql --> ignore all the sql files end with these pattern
+#default-file-rules = false
+
 # only import tables if the wildcard rules are matched. See documention for details.
 filter = ['*.*']
 
@@ -153,6 +163,24 @@ backslash-escape = true
 # if a line ends with a separator, remove it.
 trim-last-separator = false
 
+# file level routing rule that map file path to schema,table,type,sort-key
+# The schema, table , type and key can be either a constant string or template strings
+# supported by go regexp. 
+#[[mydumper.files]]
+# pattern and path determine target source files, you can use either of them but not both.
+# pattern is a regexp in Go syntax that can match one or more files in `source-dir`.
+#pattern = '(?i)^(?:[^/]*/)(?P<schema>[^/.]+)\.([^/.]+)(?:\.([0-9]+))?\.(sql|csv)$'
+# path is the target file path, both absolute file path or relative path to `mydump.source-dir` are supported.
+#path = "schema_name.table_name.00001.sql"
+# schema(database) name
+#schema = "$schema"
+# table name
+#table = "$2"
+# file type, can be one of schema-schema, table-schema, sql, csv
+#type = "$4"
+# an arbitrary string used to maintain the sort order among the files for row ID allocation and checkpoint resumption
+#key = "$3"
+
 # configuration for tidb server address(one is enough) and pd server address(one is enough).
 [tidb]
 host = "127.0.0.1"
@@ -215,12 +243,3 @@ analyze = true
 switch-mode = "5m"
 # the duration which the an import progress will be printed to the log.
 log-progress = "5m"
-
-## Rules to rename existing databases/tables and possibly merge them together.
-## The patterns support wildcards with `*` and `?`.
-## Routes are applied _after_ filter.
-# [[routes]]
-# schema-pattern = "shard_db_*"
-# table-pattern = "shard_table_*"
-# target-schema = "shard_db"
-# target-table = "shard_table"


Test patch:
diff --git a/lightning/checkpoints/checkpoints_file_test.go b/lightning/checkpoints/checkpoints_file_test.go
index 0cbc818..129cd8e 100644
--- a/lightning/checkpoints/checkpoints_file_test.go
+++ b/lightning/checkpoints/checkpoints_file_test.go
@@ -59,6 +59,10 @@ func (s *cpFileSuite) SetUpTest(c *C) {
 					Path:   "/tmp/path/1.sql",
 					Offset: 0,
 				},
+				FileMeta: mydump.SourceFileMeta{
+					Path: "/tmp/path/1.sql",
+					Type: mydump.SourceTypeSQL,
+				},
 				Chunk: mydump.Chunk{
 					Offset:       12,
 					EndOffset:    102400,
@@ -152,6 +156,10 @@ func (s *cpFileSuite) TestGet(c *C) {
 						Path:   "/tmp/path/1.sql",
 						Offset: 0,
 					},
+					FileMeta: mydump.SourceFileMeta{
+						Path: "/tmp/path/1.sql",
+						Type: mydump.SourceTypeSQL,
+					},
 					ColumnPermutation: []int{},
 					Chunk: mydump.Chunk{
 						Offset:       55904,
diff --git a/lightning/checkpoints/checkpoints_sql_test.go b/lightning/checkpoints/checkpoints_sql_test.go
index 0078329..1acfaa3 100644
--- a/lightning/checkpoints/checkpoints_sql_test.go
+++ b/lightning/checkpoints/checkpoints_sql_test.go
@@ -111,7 +111,7 @@ func (s *cpSQLSuite) TestNormalOperations(c *C) {
 		ExpectPrepare("REPLACE INTO `mock-schema`\\.chunk_v\\d+ .+")
 	insertChunkStmt.
 		ExpectExec().
-		WithArgs("`db1`.`t2`", 0, "/tmp/path/1.sql", 0, []byte("null"), 12, 102400, 1, 5000, 1234567890).
+		WithArgs("`db1`.`t2`", 0, "/tmp/path/1.sql", 0, mydump.SourceTypeSQL, 0, "", []byte("null"), 12, 102400, 1, 5000, 1234567890).
 		WillReturnResult(sqlmock.NewResult(10, 1))
 	s.mock.ExpectCommit()
 
@@ -124,6 +124,10 @@ func (s *cpSQLSuite) TestNormalOperations(c *C) {
 					Path:   "/tmp/path/1.sql",
 					Offset: 0,
 				},
+				FileMeta: mydump.SourceFileMeta{
+					Path: "/tmp/path/1.sql",
+					Type: mydump.SourceTypeSQL,
+				},
 				Chunk: mydump.Chunk{
 					Offset:       12,
 					EndOffset:    102400,
@@ -215,12 +219,12 @@ func (s *cpSQLSuite) TestNormalOperations(c *C) {
 		WithArgs("`db1`.`t2`").
 		WillReturnRows(
 			sqlmock.NewRows([]string{
-				"engine_id", "path", "offset", "columns",
+				"engine_id", "path", "offset", "type", "compression", "sort_key", "columns",
 				"pos", "end_offset", "prev_rowid_max", "rowid_max",
 				"kvc_bytes", "kvc_kvs", "kvc_checksum", "unix_timestamp(create_time)",
 			}).
 				AddRow(
-					0, "/tmp/path/1.sql", 0, "[]",
+					0, "/tmp/path/1.sql", 0, mydump.SourceTypeSQL, 0, "", "[]",
 					55904, 102400, 681, 5000,
 					4491, 586, 486070148917, 1234567894,
 				),
@@ -249,6 +253,10 @@ func (s *cpSQLSuite) TestNormalOperations(c *C) {
 						Path:   "/tmp/path/1.sql",
 						Offset: 0,
 					},
+					FileMeta: mydump.SourceFileMeta{
+						Path: "/tmp/path/1.sql",
+						Type: mydump.SourceTypeSQL,
+					},
 					ColumnPermutation: []int{},
 					Chunk: mydump.Chunk{
 						Offset:       55904,
@@ -396,12 +404,12 @@ func (s *cpSQLSuite) TestDump(c *C) {
 		ExpectQuery("SELECT (?s:.+) FROM `mock-schema`\\.chunk_v\\d+").
 		WillReturnRows(
 			sqlmock.NewRows([]string{
-				"table_name", "path", "offset", "columns",
+				"table_name", "path", "offset", "type", "compression", "sort_key", "columns",
 				"pos", "end_offset", "prev_rowid_max", "rowid_max",
 				"kvc_bytes", "kvc_kvs", "kvc_checksum",
 				"create_time", "update_time",
 			}).AddRow(
-				"`db1`.`t2`", "/tmp/path/1.sql", 0, "[]",
+				"`db1`.`t2`", "/tmp/path/1.sql", 0, mydump.SourceTypeSQL, mydump.CompressionNone, "", "[]",
 				55904, 102400, 681, 5000,
 				4491, 586, 486070148917,
 				t, t,
@@ -412,8 +420,8 @@ func (s *cpSQLSuite) TestDump(c *C) {
 	err := s.cpdb.DumpChunks(ctx, &csvBuilder)
 	c.Assert(err, IsNil)
 	c.Assert(csvBuilder.String(), Equals,
-		"table_name,path,offset,columns,pos,end_offset,prev_rowid_max,rowid_max,kvc_bytes,kvc_kvs,kvc_checksum,create_time,update_time\n"+
-			"`db1`.`t2`,/tmp/path/1.sql,0,[],55904,102400,681,5000,4491,586,486070148917,2019-04-18 02:45:55 +0000 UTC,2019-04-18 02:45:55 +0000 UTC\n",
+		"table_name,path,offset,type,compression,sort_key,columns,pos,end_offset,prev_rowid_max,rowid_max,kvc_bytes,kvc_kvs,kvc_checksum,create_time,update_time\n"+
+			"`db1`.`t2`,/tmp/path/1.sql,0,3,0,,[],55904,102400,681,5000,4491,586,486070148917,2019-04-18 02:45:55 +0000 UTC,2019-04-18 02:45:55 +0000 UTC\n",
 	)
 
 	s.mock.
diff --git a/lightning/lightning_test.go b/lightning/lightning_test.go
index 07503f4..a998f73 100644
--- a/lightning/lightning_test.go
+++ b/lightning/lightning_test.go
@@ -63,8 +63,9 @@ func (s *lightningSuite) TestRun(c *C) {
 
 	err = lightning.run(&config.Config{
 		Mydumper: config.MydumperRuntime{
-			SourceDir: ".",
-			Filter:    []string{"*.*"},
+			SourceDir:        ".",
+			Filter:           []string{"*.*"},
+			DefaultFileRules: true,
 		},
 		Checkpoint: config.Checkpoint{
 			Enable: true,
diff --git a/lightning/mydump/loader_test.go b/lightning/mydump/loader_test.go
index 6faadce..38ac7da 100644
--- a/lightning/mydump/loader_test.go
+++ b/lightning/mydump/loader_test.go
@@ -20,9 +20,11 @@ import (
 	"testing"
 
 	. "github.com/pingcap/check"
+	filter "github.com/pingcap/tidb-tools/pkg/table-filter"
+	router "github.com/pingcap/tidb-tools/pkg/table-router"
+
 	"github.com/pingcap/tidb-lightning/lightning/config"
 	md "github.com/pingcap/tidb-lightning/lightning/mydump"
-	router "github.com/pingcap/tidb-tools/pkg/table-router"
 )
 
 var _ = Suite(&testMydumpLoaderSuite{})
@@ -41,8 +43,9 @@ func (s *testMydumpLoaderSuite) TearDownSuite(c *C) {}
 func newConfigWithSourceDir(sourceDir string) *config.Config {
 	return &config.Config{
 		Mydumper: config.MydumperRuntime{
-			SourceDir: sourceDir,
-			Filter:    []string{"*.*"},
+			SourceDir:        sourceDir,
+			Filter:           []string{"*.*"},
+			DefaultFileRules: true,
 		},
 	}
 }
@@ -134,7 +137,7 @@ func (s *testMydumpLoaderSuite) TestTableNoHostDB(c *C) {
 	c.Assert(err, IsNil)
 
 	_, err = md.NewMyDumpLoader(s.cfg)
-	c.Assert(err, ErrorMatches, `invalid table schema file, cannot find db - .*[/\\]db\.tbl-schema\.sql`)
+	c.Assert(err, ErrorMatches, `invalid table schema file, cannot find db 'db' - .*[/\\]db\.tbl-schema\.sql`)
 }
 
 func (s *testMydumpLoaderSuite) TestDuplicatedTable(c *C) {
@@ -168,7 +171,7 @@ func (s *testMydumpLoaderSuite) TestDataNoHostDB(c *C) {
 	s.touch(c, "db.tbl.sql")
 
 	_, err := md.NewMyDumpLoader(s.cfg)
-	c.Assert(err, ErrorMatches, `invalid data file, miss host db - .*[/\\]db\.tbl\.sql`)
+	c.Assert(err, ErrorMatches, `invalid data file, miss host db 'db' - .*[/\\]db\.tbl\.sql`)
 }
 
 func (s *testMydumpLoaderSuite) TestDataNoHostTable(c *C) {
@@ -182,7 +185,7 @@ func (s *testMydumpLoaderSuite) TestDataNoHostTable(c *C) {
 	s.touch(c, "db.tbl.sql")
 
 	_, err := md.NewMyDumpLoader(s.cfg)
-	c.Assert(err, ErrorMatches, `invalid data file, miss host table - .*[/\\]db\.tbl\.sql`)
+	c.Assert(err, ErrorMatches, `invalid data file, miss host table 'tbl' - .*[/\\]db\.tbl\.sql`)
 }
 
 func (s *testMydumpLoaderSuite) TestDataWithoutSchema(c *C) {
@@ -203,7 +206,7 @@ func (s *testMydumpLoaderSuite) TestDataWithoutSchema(c *C) {
 			DB:         "db",
 			Name:       "tbl",
 			SchemaFile: "",
-			DataFiles:  []string{p},
+			DataFiles:  []md.FileInfo{{TableName: filter.Table{Schema: "db", Name: "tbl"}, FileMeta: md.SourceFileMeta{Path: p, Type: md.SourceTypeSQL}}},
 		}},
 	}})
 }
@@ -224,7 +227,6 @@ func (s *testMydumpLoaderSuite) TestTablesWithDots(c *C) {
 
 	mdl, err := md.NewMyDumpLoader(s.cfg)
 	c.Assert(err, IsNil)
-
 	c.Assert(mdl.GetDatabases(), DeepEquals, []*md.MDDatabaseMeta{{
 		Name:       "db",
 		SchemaFile: pDBSchema,
@@ -233,13 +235,13 @@ func (s *testMydumpLoaderSuite) TestTablesWithDots(c *C) {
 				DB:         "db",
 				Name:       "0002",
 				SchemaFile: pT2Schema,
-				DataFiles:  []string{pT2Data},
+				DataFiles:  []md.FileInfo{{TableName: filter.Table{Schema: "db", Name: "0002"}, FileMeta: md.SourceFileMeta{Path: pT2Data, Type: md.SourceTypeSQL}}},
 			},
 			{
 				DB:         "db",
 				Name:       "tbl.with.dots",
 				SchemaFile: pT1Schema,
-				DataFiles:  []string{pT1Data},
+				DataFiles:  []md.FileInfo{{TableName: filter.Table{Schema: "db", Name: "tbl.with.dots"}, FileMeta: md.SourceFileMeta{Path: pT1Data, Type: md.SourceTypeSQL, SortKey: "0001"}}},
 			},
 		},
 	}})
@@ -297,7 +299,6 @@ func (s *testMydumpLoaderSuite) TestRouter(c *C) {
 
 	mdl, err := md.NewMyDumpLoader(s.cfg)
 	c.Assert(err, IsNil)
-
 	c.Assert(mdl.GetDatabases(), DeepEquals, []*md.MDDatabaseMeta{
 		{
 			Name:       "a1",
@@ -307,7 +308,7 @@ func (s *testMydumpLoaderSuite) TestRouter(c *C) {
 					DB:         "a1",
 					Name:       "s1",
 					SchemaFile: pA1S1Schema,
-					DataFiles:  []string{pA1S1Data},
+					DataFiles:  []md.FileInfo{{TableName: filter.Table{Schema: "a1", Name: "s1"}, FileMeta: md.SourceFileMeta{Path: pA1S1Data, Type: md.SourceTypeSQL, SortKey: "1"}}},
 				},
 			},
 		},
@@ -323,7 +324,11 @@ func (s *testMydumpLoaderSuite) TestRouter(c *C) {
 					DB:         "b",
 					Name:       "u",
 					SchemaFile: pA0T0Schema,
-					DataFiles:  []string{pA0T0Data, pA0T1Data, pA1T2Data},
+					DataFiles: []md.FileInfo{
+						{TableName: filter.Table{Schema: "b", Name: "u"}, FileMeta: md.SourceFileMeta{Path: pA0T0Data, Type: md.SourceTypeSQL, SortKey: "1"}},
+						{TableName: filter.Table{Schema: "b", Name: "u"}, FileMeta: md.SourceFileMeta{Path: pA0T1Data, Type: md.SourceTypeSQL, SortKey: "1"}},
+						{TableName: filter.Table{Schema: "b", Name: "u"}, FileMeta: md.SourceFileMeta{Path: pA1T2Data, Type: md.SourceTypeSQL, SortKey: "1"}},
+					},
 				},
 			},
 		},
@@ -335,7 +340,7 @@ func (s *testMydumpLoaderSuite) TestRouter(c *C) {
 					DB:         "c",
 					Name:       "t3",
 					SchemaFile: pC0T3Schema,
-					DataFiles:  []string{pC0T3Data},
+					DataFiles:  []md.FileInfo{{TableName: filter.Table{Schema: "c", Name: "t3"}, FileMeta: md.SourceFileMeta{Path: pC0T3Data, Type: md.SourceTypeSQL, SortKey: "1"}}},
 				},
 			},
 		},
@@ -348,8 +353,80 @@ func (s *testMydumpLoaderSuite) TestBadRouterRule(c *C) {
 		TargetSchema:  "ab",
 	}}
 
-	s.touch(c, "a1b-schema-create.sql")
-
 	_, err := md.NewMyDumpLoader(s.cfg)
 	c.Assert(err, ErrorMatches, `.*pattern a\*b not valid`)
 }
+
+func (s *testMydumpLoaderSuite) TestFileRouting(c *C) {
+	s.cfg.Mydumper.DefaultFileRules = false
+	s.cfg.Mydumper.FileRouters = []*config.FileRouteRule{
+		{
+			Pattern: `(?i)^(?:[^./]*/)*([a-z0-9_]+)/schema\.sql$`,
+			Schema:  "$1",
+			Type:    "schema-schema",
+		},
+		{
+			Pattern: `(?i)^(?:[^./]*/)*([a-z0-9]+)/([a-z0-9_]+)-table\.sql$`,
+			Schema:  "$1",
+			Table:   "$2",
+			Type:    "table-schema",
+		},
+		{
+			Pattern: `(?i)^(?:[^./]*/)*([a-z][a-z0-9_]*)/([a-z]+)[0-9]*(?:\.([0-9]+))?\.(sql|csv)$`,
+			Schema:  "$1",
+			Table:   "$2",
+			Type:    "$4",
+		},
+		{
+			Pattern: `^(?:[^./]*/)*([a-z]+)(?:\.([0-9]+))?\.(sql|csv)$`,
+			Schema:  "d2",
+			Table:   "$1",
+			Type:    "$3",
+		},
+	}
+
+	s.mkdir(c, "d1")
+	s.mkdir(c, "d2")
+	d1Schema := s.touch(c, "d1/schema.sql")
+	d1TestTable := s.touch(c, "d1/test-table.sql")
+	d1TestData0 := s.touch(c, "d1/test0.sql")
+	d1TestData1 := s.touch(c, "d1/test1.sql")
+	d1TestData2 := s.touch(c, "d1/test2.001.sql")
+	_ = s.touch(c, "d1/t1-schema-create.sql")
+	d2Schema := s.touch(c, "d2/schema.sql")
+	d2TestTable := s.touch(c, "d2/abc-table.sql")
+	d2AbcData0 := s.touch(c, "abc.1.sql")
+
+	mdl, err := md.NewMyDumpLoader(s.cfg)
+	c.Assert(err, IsNil)
+	c.Assert(mdl.GetDatabases(), DeepEquals, []*md.MDDatabaseMeta{
+		{
+			Name:       "d1",
+			SchemaFile: d1Schema,
+			Tables: []*md.MDTableMeta{
+				{
+					DB:         "d1",
+					Name:       "test",
+					SchemaFile: d1TestTable,
+					DataFiles: []md.FileInfo{
+						{TableName: filter.Table{Schema: "d1", Name: "test"}, FileMeta: md.SourceFileMeta{Path: d1TestData0, Type: md.SourceTypeSQL}},
+						{TableName: filter.Table{Schema: "d1", Name: "test"}, FileMeta: md.SourceFileMeta{Path: d1TestData1, Type: md.SourceTypeSQL}},
+						{TableName: filter.Table{Schema: "d1", Name: "test"}, FileMeta: md.SourceFileMeta{Path: d1TestData2, Type: md.SourceTypeSQL}},
+					},
+				},
+			},
+		},
+		{
+			Name:       "d2",
+			SchemaFile: d2Schema,
+			Tables: []*md.MDTableMeta{
+				{
+					DB:         "d2",
+					Name:       "abc",
+					SchemaFile: d2TestTable,
+					DataFiles:  []md.FileInfo{{TableName: filter.Table{Schema: "d2", Name: "abc"}, FileMeta: md.SourceFileMeta{Path: d2AbcData0, Type: md.SourceTypeSQL}}},
+				},
+			},
+		},
+	})
+}
diff --git a/lightning/mydump/region_test.go b/lightning/mydump/region_test.go
index 4b54108..fe93fa9 100644
--- a/lightning/mydump/region_test.go
+++ b/lightning/mydump/region_test.go
@@ -71,7 +71,7 @@ func (s *testMydumpRegionSuite) TestTableRegion(c *C) {
 		table := meta.Name
 		fmt.Printf("[%s] region count ===============> %d\n", table, len(regions))
 		for _, region := range regions {
-			fname := filepath.Base(region.File)
+			fname := filepath.Base(region.FileMeta.Path)
 			fmt.Printf("[%s] rowID = %5d / rows = %5d / offset = %10d / size = %10d \n",
 				fname,
 				region.RowIDMin(),
@@ -83,9 +83,7 @@ func (s *testMydumpRegionSuite) TestTableRegion(c *C) {
 		// check - region-size vs file-size
 		var tolFileSize int64 = 0
 		for _, file := range meta.DataFiles {
-			fileSize, err := getFileSize(file)
-			c.Assert(err, IsNil)
-			tolFileSize += fileSize
+			tolFileSize += file.Size
 		}
 		var tolRegionSize int64 = 0
 		for _, region := range regions {
@@ -105,7 +103,7 @@ func (s *testMydumpRegionSuite) TestTableRegion(c *C) {
 		preReg := regions[0]
 		for i := 1; i < regionNum; i++ {
 			reg := regions[i]
-			if preReg.File == reg.File {
+			if preReg.FileMeta.Path == reg.FileMeta.Path {
 				c.Assert(reg.Offset(), Equals, preReg.Offset()+preReg.Size())
 				c.Assert(reg.RowIDMin(), Equals, preReg.RowIDMin()+preReg.Rows())
 			} else {
@@ -225,6 +223,7 @@ func (s *testMydumpRegionSuite) TestSplitLargeFile(c *C) {
 		log.Fatal(err)
 	}
 	fileSize := dataFileInfo.Size()
+	fileInfo := FileInfo{FileMeta: SourceFileMeta{Path: filePath, Type: SourceTypeCSV}, Size: fileSize}
 	colCnt := int64(3)
 	columns := []string{"a", "b", "c"}
 	for _, tc := range []struct {
@@ -243,7 +242,7 @@ func (s *testMydumpRegionSuite) TestSplitLargeFile(c *C) {
 		cfg.Mydumper.MaxRegionSize = tc.maxRegionSize
 		prevRowIdxMax := int64(0)
 		ioWorker := worker.NewPool(context.Background(), 4, "io")
-		_, regions, _, err := SplitLargeFile(meta, cfg, filePath, fileSize, colCnt, prevRowIdxMax, ioWorker)
+		_, regions, _, err := SplitLargeFile(meta, cfg, fileInfo, colCnt, prevRowIdxMax, ioWorker)
 		c.Assert(err, IsNil)
 		c.Assert(len(regions), Equals, tc.chkCnt)
 		for i := range tc.offsets {
diff --git a/lightning/mydump/router_test.go b/lightning/mydump/router_test.go
new file mode 100644
index 0000000..3be8e6e
--- /dev/null
+++ b/lightning/mydump/router_test.go
@@ -0,0 +1,223 @@
+package mydump
+
+import (
+	"strings"
+
+	. "github.com/pingcap/check"
+	"github.com/pingcap/tidb-tools/pkg/filter"
+
+	"github.com/pingcap/tidb-lightning/lightning/config"
+)
+
+var _ = Suite(&testFileRouterSuite{})
+
+type testFileRouterSuite struct{}
+
+func (t *testFileRouterSuite) TestRouteParser(c *C) {
+	// valid rules
+	rules := []*config.FileRouteRule{
+		{Pattern: `^(?:[^/]*/)*([^/.]+)\.([^./]+)(?:\.[0-9]+)?\.(csv|sql)`, Schema: "$1", Table: "$2", Type: "$3"},
+		{Pattern: `^.+\.(csv|sql)`, Schema: "test", Table: "t", Type: "$1"},
+		{Pattern: `^(?:[^/]*/)*(?P<schema>[^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "$schema", Table: "$table", Type: "$type", Key: "$key", Compression: "$cp"},
+		{Pattern: `^(?:[^/]*/)*(?P<schema>[^/.]+)\.(?P<table>[^./]+)(?:\.([0-9]+))?\.(csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "${schema}s", Table: "$table", Type: "${3}_0", Key: "$4", Compression: "$cp"},
+		{Pattern: `^(?:[^/]*/)*([^/.]+)\.(?P<table>[^./]+)(?:\.([0-9]+))?\.(csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "${1}s", Table: "$table", Type: "${3}_0", Key: "$4", Compression: "$cp"},
+		{Pattern: `^(?:[^/]*/)*([^/.]+)\.([^./]+)(?:\.[0-9]+)?\.(csv|sql)`, Schema: "$1-schema", Table: "$1-table", Type: "$2"},
+	}
+	for _, r := range rules {
+		_, err := NewFileRouter([]*config.FileRouteRule{r})
+		c.Assert(err, IsNil)
+	}
+
+	// invalid rules
+	invalidRules := []*config.FileRouteRule{
+		{Pattern: `^(?:[^/]*/)*(?P<schema>\.(?P<table>[^./]+).*$`, Schema: "$test", Table: "$table"},
+		{Pattern: `^(?:[^/]*/)*(?P<schema>[^/.]+)\.(?P<table>[^./]+).*$`, Schema: "$schemas", Table: "$table"},
+		{Pattern: `^(?:[^/]*/)*([^/.]+)\.([^./]+)(?:\.[0-9]+)?\.(csv|sql)`, Schema: "$1", Table: "$2", Type: "$3", Key: "$4"},
+	}
+	for _, r := range invalidRules {
+		_, err := NewFileRouter([]*config.FileRouteRule{r})
+		c.Assert(err, NotNil)
+	}
+}
+
+func (t *testFileRouterSuite) TestSingleRouteRule(c *C) {
+	rules := []*config.FileRouteRule{
+		{Pattern: `^(?:[^/]*/)*([^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "$1", Table: "$table", Type: "$type", Key: "$key", Compression: "$cp"},
+	}
+
+	r, err := NewFileRouter(rules)
+	c.Assert(err, IsNil)
+
+	inputOutputMap := map[string][]string{
+		"my_schema.my_table.sql":           {"my_schema", "my_table", "", "", "sql"},
+		"/test/123/my_schema.my_table.sql": {"my_schema", "my_table", "", "", "sql"},
+		"my_dir/my_schema.my_table.csv":    {"my_schema", "my_table", "", "", "csv"},
+		"my_schema.my_table.0001.sql":      {"my_schema", "my_table", "0001", "", "sql"},
+	}
+	for path, fields := range inputOutputMap {
+		res, err := r.Route(path)
+		c.Assert(err, IsNil)
+		compress, e := parseCompressionType(fields[3])
+		c.Assert(e, IsNil)
+		ty, e := parseSourceType(fields[4])
+		c.Assert(e, IsNil)
+		exp := &RouteResult{filter.Table{Schema: fields[0], Name: fields[1]}, fields[2], compress, ty}
+		c.Assert(res, DeepEquals, exp)
+	}
+
+	notMatchPaths := []string{
+		"my_table.sql",
+		"/schema/table.sql",
+		"my_schema.my_table.txt",
+		"my_schema.my_table.001.txt",
+		"my_schema.my_table.0001-002.sql",
+	}
+	for _, p := range notMatchPaths {
+		res, err := r.Route(p)
+		c.Assert(res, IsNil)
+		c.Assert(err, IsNil)
+	}
+
+	rule := &config.FileRouteRule{Pattern: `^(?:[^/]*/)*([^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>\w+)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "$1", Table: "$table", Type: "$type", Key: "$key", Compression: "$cp"}
+	r, err = NewFileRouter([]*config.FileRouteRule{rule})
+	c.Assert(err, IsNil)
+	c.Assert(r, NotNil)
+	invalidMatchPaths := []string{
+		"my_schema.my_table.sql.gz",
+		"my_schema.my_table.sql.rar",
+		"my_schema.my_table.txt",
+	}
+	for _, p := range invalidMatchPaths {
+		res, err := r.Route(p)
+		c.Assert(res, IsNil)
+		c.Assert(err, NotNil)
+	}
+}
+
+func (t *testFileRouterSuite) TestMultiRouteRule(c *C) {
+	// multi rule don't intersect with each other
+	rules := []*config.FileRouteRule{
+		{Pattern: `(?:[^/]*/)*([^/.]+)-schema-create\.sql`, Schema: "$1", Type: SchemaSchema},
+		{Pattern: `(?:[^/]*/)*([^/.]+)\.([^/.]+)-schema\.sql`, Schema: "$1", Table: "$2", Type: TableSchema},
+		{Pattern: `^(?:[^/]*/)*(?P<schema>[^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "$schema", Table: "$table", Type: "$type", Key: "$key", Compression: "$cp"},
+	}
+
+	r, err := NewFileRouter(rules)
+	c.Assert(err, IsNil)
+
+	inputOutputMap := map[string][]string{
+		"test-schema-create.sql":           {"test", "", "", "", SchemaSchema},
+		"test.t-schema.sql":                {"test", "t", "", "", TableSchema},
+		"my_schema.my_table.sql":           {"my_schema", "my_table", "", "", "sql"},
+		"/test/123/my_schema.my_table.sql": {"my_schema", "my_table", "", "", "sql"},
+		"my_dir/my_schema.my_table.csv":    {"my_schema", "my_table", "", "", "csv"},
+		"my_schema.my_table.0001.sql":      {"my_schema", "my_table", "0001", "", "sql"},
+		//"my_schema.my_table.0001.sql.gz":      {"my_schema", "my_table", "0001", "gz", "sql"},
+	}
+	for path, fields := range inputOutputMap {
+		res, err := r.Route(path)
+		c.Assert(err, IsNil)
+		if len(fields) == 0 {
+			c.Assert(res, IsNil)
+		} else {
+			compress, e := parseCompressionType(fields[3])
+			c.Assert(e, IsNil)
+			ty, e := parseSourceType(fields[4])
+			c.Assert(e, IsNil)
+			exp := &RouteResult{filter.Table{Schema: fields[0], Name: fields[1]}, fields[2], compress, ty}
+			c.Assert(res, DeepEquals, exp)
+		}
+	}
+
+	// multi rule don't intersect with each other
+	// add another rule that math same pattern with the third rule, the result should be no different
+	p := &config.FileRouteRule{Pattern: `^(?P<schema>[^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "test_schema", Table: "test_table", Type: "$type", Key: "$key", Compression: "$cp"}
+	rules = append(rules, p)
+	r, err = NewFileRouter(rules)
+	c.Assert(err, IsNil)
+	for path, fields := range inputOutputMap {
+		res, err := r.Route(path)
+		c.Assert(err, IsNil)
+		if len(fields) == 0 {
+			c.Assert(res, IsNil)
+		} else {
+			compress, e := parseCompressionType(fields[3])
+			c.Assert(e, IsNil)
+			ty, e := parseSourceType(fields[4])
+			c.Assert(e, IsNil)
+			exp := &RouteResult{filter.Table{Schema: fields[0], Name: fields[1]}, fields[2], compress, ty}
+			c.Assert(res, DeepEquals, exp)
+		}
+	}
+}
+
+func (t *testFileRouterSuite) TestRouteExpanding(c *C) {
+	rule := &config.FileRouteRule{
+		Pattern:     `^(?:[^/]*/)*(?P<schema>[^/.]+)\.(?P<table_name>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`,
+		Schema:      "$schema",
+		Type:        "$type",
+		Key:         "$key",
+		Compression: "$cp",
+	}
+	path := "db.table.001.sql"
+	tablePatternResMap := map[string]string{
+		"$schema":             "db",
+		"$table_name":         "table",
+		"$schema.$table_name": "db.table",
+		"${1}":                "db",
+		"${1}_$table_name":    "db_table",
+		"${2}.schema":         "table.schema",
+		"$${2}":               "${2}",
+		"$$table_name":        "$table_name",
+		"$table_name-123":     "table-123",
+		"$$12$1$schema":       "$12dbdb",
+		"${table_name}$$2":    "table$2",
+		"${table_name}$$":     "table$",
+		"{1}$$":               "{1}$",
+		"my_table":            "my_table",
+	}
+
+	for pat, value := range tablePatternResMap {
+		rule.Table = pat
+		router, err := NewFileRouter([]*config.FileRouteRule{rule})
+		c.Assert(err, IsNil)
+		res, err := router.Route(path)
+		c.Assert(err, IsNil)
+		c.Assert(res, NotNil)
+		c.Assert(res.Name, Equals, value)
+	}
+
+	invalidPatterns := []string{"$1_$schema", "$schema_$table_name", "$6"}
+	for _, pat := range invalidPatterns {
+		rule.Table = pat
+		_, err := NewFileRouter([]*config.FileRouteRule{rule})
+		c.Assert(err, NotNil)
+	}
+}
+
+func (t *testFileRouterSuite) TestRouteWithPath(c *C) {
+	fileName := "myschema.(my_table$1).000.sql"
+	rule := &config.FileRouteRule{
+		Path:   fileName,
+		Schema: "schema",
+		Table:  "my_table$1",
+		Type:   "sql",
+		Key:    "$key",
+	}
+	r := *rule
+	router, err := NewFileRouter([]*config.FileRouteRule{&r})
+	c.Assert(err, IsNil)
+	res, err := router.Route(fileName)
+	c.Assert(err, IsNil)
+	c.Assert(res, NotNil)
+	c.Assert(res.Schema, Equals, rule.Schema)
+	c.Assert(res.Name, Equals, rule.Table)
+	ty, _ := parseSourceType(rule.Type)
+	c.Assert(res.Type, Equals, ty)
+	c.Assert(res.Key, Equals, rule.Key)
+
+	// replace all '.' by '-', if with plain regex pattern, will still match
+	res, err = router.Route(strings.ReplaceAll(fileName, ".", "-"))
+	c.Assert(err, IsNil)
+	c.Assert(res, IsNil)
+}
diff --git a/lightning/restore/restore_test.go b/lightning/restore/restore_test.go
index 73af368..ffd9e4e 100644
--- a/lightning/restore/restore_test.go
+++ b/lightning/restore/restore_test.go
@@ -16,6 +16,7 @@ package restore
 import (
 	"context"
 	"fmt"
+	"github.com/pingcap/tidb-tools/pkg/filter"
 	"io/ioutil"
 	"path/filepath"
 	"sort"
@@ -304,12 +305,12 @@ func (s *tableRestoreSuite) SetUpSuite(c *C) {
 	fakeDataFilesCount := 6
 	fakeDataFilesContent := []byte("INSERT INTO `table` VALUES (1, 2, 3);")
 	c.Assert(len(fakeDataFilesContent), Equals, 37)
-	fakeDataFiles := make([]string, 0, fakeDataFilesCount)
+	fakeDataFiles := make([]mydump.FileInfo, 0, fakeDataFilesCount)
 	for i := 1; i <= fakeDataFilesCount; i++ {
 		fakeDataPath := filepath.Join(fakeDataDir, fmt.Sprintf("db.table.%d.sql", i))
 		err = ioutil.WriteFile(fakeDataPath, fakeDataFilesContent, 0644)
 		c.Assert(err, IsNil)
-		fakeDataFiles = append(fakeDataFiles, fakeDataPath)
+		fakeDataFiles = append(fakeDataFiles, mydump.FileInfo{TableName: filter.Table{"db", "table"}, FileMeta: mydump.SourceFileMeta{Path: fakeDataPath, Type: mydump.SourceTypeSQL, SortKey: fmt.Sprintf("%d", i)}, Size: 37})
 	}
 
 	s.tableMeta = &mydump.MDTableMeta{
@@ -352,7 +353,8 @@ func (s *tableRestoreSuite) TestPopulateChunks(c *C) {
 			Status: CheckpointStatusLoaded,
 			Chunks: []*ChunkCheckpoint{
 				{
-					Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[0], Offset: 0},
+					Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[0].FileMeta.Path, Offset: 0},
+					FileMeta: s.tr.tableMeta.DataFiles[0].FileMeta,
 					Chunk: mydump.Chunk{
 						Offset:       0,
 						EndOffset:    37,
@@ -362,7 +364,8 @@ func (s *tableRestoreSuite) TestPopulateChunks(c *C) {
 					Timestamp: 1234567897,
 				},
 				{
-					Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[1], Offset: 0},
+					Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[1].FileMeta.Path, Offset: 0},
+					FileMeta: s.tr.tableMeta.DataFiles[1].FileMeta,
 					Chunk: mydump.Chunk{
 						Offset:       0,
 						EndOffset:    37,
@@ -372,7 +375,8 @@ func (s *tableRestoreSuite) TestPopulateChunks(c *C) {
 					Timestamp: 1234567897,
 				},
 				{
-					Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[2], Offset: 0},
+					Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[2].FileMeta.Path, Offset: 0},
+					FileMeta: s.tr.tableMeta.DataFiles[2].FileMeta,
 					Chunk: mydump.Chunk{
 						Offset:       0,
 						EndOffset:    37,
@@ -387,7 +391,8 @@ func (s *tableRestoreSuite) TestPopulateChunks(c *C) {
 			Status: CheckpointStatusLoaded,
 			Chunks: []*ChunkCheckpoint{
 				{
-					Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[3], Offset: 0},
+					Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[3].FileMeta.Path, Offset: 0},
+					FileMeta: s.tr.tableMeta.DataFiles[3].FileMeta,
 					Chunk: mydump.Chunk{
 						Offset:       0,
 						EndOffset:    37,
@@ -397,7 +402,8 @@ func (s *tableRestoreSuite) TestPopulateChunks(c *C) {
 					Timestamp: 1234567897,
 				},
 				{
-					Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[4], Offset: 0},
+					Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[4].FileMeta.Path, Offset: 0},
+					FileMeta: s.tr.tableMeta.DataFiles[4].FileMeta,
 					Chunk: mydump.Chunk{
 						Offset:       0,
 						EndOffset:    37,
@@ -407,7 +413,8 @@ func (s *tableRestoreSuite) TestPopulateChunks(c *C) {
 					Timestamp: 1234567897,
 				},
 				{
-					Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[5], Offset: 0},
+					Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[5].FileMeta.Path, Offset: 0},
+					FileMeta: s.tr.tableMeta.DataFiles[5].FileMeta,
 					Chunk: mydump.Chunk{
 						Offset:       0,
 						EndOffset:    37,
@@ -571,7 +578,8 @@ func (s *chunkRestoreSuite) SetUpTest(c *C) {
 	w := worker.NewPool(ctx, 5, "io")
 
 	chunk := ChunkCheckpoint{
-		Key: ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[1], Offset: 0},
+		Key:      ChunkCheckpointKey{Path: s.tr.tableMeta.DataFiles[1].FileMeta.Path, Offset: 0},
+		FileMeta: s.tr.tableMeta.DataFiles[1].FileMeta,
 		Chunk: mydump.Chunk{
 			Offset:       0,
 			EndOffset:    37,
diff --git a/tests/file_routing/config.toml b/tests/file_routing/config.toml
new file mode 100644
index 0000000..94eaf7e
--- /dev/null
+++ b/tests/file_routing/config.toml
@@ -0,0 +1,31 @@
+[lightning]
+table-concurrency = 1
+
+[checkpoint]
+enable = false
+
+[mydumper]
+default-file-rules = false
+
+[[mydumper.files]]
+pattern = "(?i)^(?:[^/]*/)*([a-z0-9]+)/schema\\.sql$"
+schema = "$1"
+type = "schema-schema"
+
+[[mydumper.files]]
+pattern = "(?i)^(?:[^/]*/)*([a-z0-9]+)/([a-z0-9]+)-table\\.sql$"
+schema = "$1"
+table = "$2"
+type = "table-schema"
+
+[[mydumper.files]]
+pattern = "(?i)^(?:[^/]*/)*([^./]+)/([a-z]+)[0-9]*\\.(sql|csv)(?:\\.([0-9]+))?$"
+schema = "$1"
+table = "$2"
+type = "$3"
+
+[[mydumper.files]]
+pattern = "(?i)^(?:[^/]*/)*([a-z]+)[0-9]*\\.(sql|csv)(?:\\.([0-9]+))?$"
+schema = "fr"
+table = "$1"
+type = "$2"
diff --git a/tests/file_routing/run.sh b/tests/file_routing/run.sh
new file mode 100755
index 0000000..cdf5eee
--- /dev/null
+++ b/tests/file_routing/run.sh
@@ -0,0 +1,44 @@
+#!/bin/sh
+#
+# Copyright 2019 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -euE
+
+# Populate the mydumper source
+DBPATH="$TEST_DIR/fr.mydump"
+
+mkdir -p $DBPATH $DBPATH/fr $DBPATH/ff
+echo 'CREATE DATABASE fr;' > "$DBPATH/fr/schema.sql"
+echo "CREATE TABLE tbl(i TINYINT PRIMARY KEY, j INT);" > "$DBPATH/fr/tbl-table.sql"
+# the column orders in data file is different from table schema order.
+echo "INSERT INTO tbl (i, j) VALUES (1, 1),(2, 2);" > "$DBPATH/fr/tbl1.sql.0"
+echo "INSERT INTO tbl (i, j) VALUES (3, 3),(4, 4);" > "$DBPATH/fr/tbl2.sql.0"
+echo "INSERT INTO tbl (i, j) VALUES (5, 5);" > "$DBPATH/fr/tbl.sql"
+echo "INSERT INTO tbl (i, j) VALUES (6, 6), (7, 7), (8, 8), (9, 9);" > "$DBPATH/tbl1.sql.1"
+echo "INSERT INTO tbl (i, j) VALUES (11, 11);" > "$DBPATH/fr/tbl-noused.sql"
+
+# Set minDeliverBytes to a small enough number to only write only 1 row each time
+# Set the failpoint to kill the lightning instance as soon as one row is written
+
+# Start importing the tables.
+run_sql 'DROP DATABASE IF EXISTS fr'
+
+set +e
+run_lightning -d "$DBPATH" --backend local 2> /dev/null
+set -e
+run_sql 'SELECT count(*) FROM `fr`.tbl'
+check_contains "count(*): 9"
+
+run_sql 'SELECT sum(j) FROM `fr`.tbl'
+check_contains "sum(j): 45"