Repository: juicedata/juicefs
Pull request #2091 (2022-05-24 08:19:23)

cmd/objbench: add content check 
add content check for downloaded files in object storage performance tests


Patch:
diff --git a/.github/workflows/tpch_clickhouse.yml b/.github/workflows/tpch_clickhouse.yml
index 3f51f13c..9ab20e26 100644
--- a/.github/workflows/tpch_clickhouse.yml
+++ b/.github/workflows/tpch_clickhouse.yml
@@ -44,7 +44,7 @@ jobs:
 
       - name: Juicefs Format
         run: |
-          sudo ./juicefs format redis://127.0.0.1:6379/1 pics
+          sudo ./juicefs format --trash-days 0 redis://127.0.0.1:6379/1 pics
 
       - name: Juicefs Mount
         run: |
@@ -72,14 +72,16 @@ jobs:
       - name: Generate Data
         run: |
           cd /data/jfs/
-          git clone https://github.com/sanwan/ssb-dbgen.git
-          cd ssb-dbgen
-          make
-          ./dbgen -s 10 -T l -q
+          git clone https://github.com/sanwan/actiondata.git
+          cd actiondata
+          cat lineorder.tbl.gz* >lineorder.tbl.gz
+          gzip -dk lineorder.tbl.gz
           git clone https://github.com/sanwan/actionfile.git
           mv actionfile/*.tbl .
+          rm lineorder.tbl.gz*
           ls -lrth
           sudo du -sh /data/jfs/
+          df -lh
 
       - name: Create Table
         run: |
@@ -144,11 +146,14 @@ jobs:
 
       - name: Import Data
         run: |
-          cd /data/jfs/ssb-dbgen/
+          cd /data/jfs/actiondata/
           clickhouse-client --query "INSERT INTO customer FORMAT CSV" < customer.tbl
           clickhouse-client --query "INSERT INTO part FORMAT CSV" < part.tbl
           clickhouse-client --query "INSERT INTO supplier FORMAT CSV" < supplier.tbl
           time clickhouse-client --query "INSERT INTO lineorder FORMAT CSV" < lineorder.tbl
+          sudo du -sh /data/jfs/
+          sudo du -sh /var/jfsCache/
+          df -lh
 
 
       - name: ClickHouse Test
@@ -250,7 +255,9 @@ jobs:
                                      ORDER BY
                                          year ASC,
                                          C_NATION ASC;"
-
+          sudo du -sh /var/jfsCache/
+          sudo du -sh /data/jfs/
+          df -lh
 
 
       - name: Send Slack Notification
diff --git a/.github/workflows/tpch_clickhouse_mode.yml b/.github/workflows/tpch_clickhouse_limitcache.yml
similarity index 95%
rename from .github/workflows/tpch_clickhouse_mode.yml
rename to .github/workflows/tpch_clickhouse_limitcache.yml
index a0f45bb3..bd32c08e 100644
--- a/.github/workflows/tpch_clickhouse_mode.yml
+++ b/.github/workflows/tpch_clickhouse_limitcache.yml
@@ -1,4 +1,4 @@
-name: "tpch_clickhouse_mode"
+name: "tpch_clickhouse_limitcache"
 
 on:
   push:
@@ -18,7 +18,7 @@ on:
   workflow_dispatch:
 
 jobs:
-  tpchchmode:
+  tpchchlimitcache:
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go
@@ -44,11 +44,11 @@ jobs:
 
       - name: Juicefs Format
         run: |
-          sudo ./juicefs format redis://127.0.0.1:6379/1 pics
+          sudo ./juicefs format --trash-days 0 redis://127.0.0.1:6379/1 pics
 
       - name: Juicefs Mount
         run: |
-          sudo ./juicefs mount -d redis://127.0.0.1:6379/1 /data/jfs --no-usage-report &
+          sudo ./juicefs mount -d redis://127.0.0.1:6379/1 /data/jfs  --cache-size 3072 --no-usage-report &
 
       - name: Install ClickHouse
         run: |
@@ -72,14 +72,16 @@ jobs:
       - name: Generate Data
         run: |
           cd /data/jfs/
-          git clone https://github.com/sanwan/ssb-dbgen.git
-          cd ssb-dbgen
-          make
-          ./dbgen -s 10 -T l -q
+          git clone https://github.com/sanwan/actiondata.git
+          cd actiondata
+          cat lineorder.tbl.gz* >lineorder.tbl.gz
+          gzip -dk lineorder.tbl.gz
           git clone https://github.com/sanwan/actionfile.git
           mv actionfile/*.tbl .
+          rm lineorder.tbl.gz*
           ls -lrth
           sudo du -sh /data/jfs/
+          df -lh
 
       - name: Create Table
         run: |
@@ -144,12 +146,15 @@ jobs:
 
       - name: Import Data
         run: |
-          cd /data/jfs/ssb-dbgen/
+          df -lh
+          cd /data/jfs/actiondata/
           clickhouse-client --query "INSERT INTO customer FORMAT CSV" < customer.tbl
           clickhouse-client --query "INSERT INTO part FORMAT CSV" < part.tbl
           clickhouse-client --query "INSERT INTO supplier FORMAT CSV" < supplier.tbl
           time clickhouse-client --query "INSERT INTO lineorder FORMAT CSV" < lineorder.tbl
-
+          sudo du -sh /data/jfs/
+          sudo du -sh /var/jfsCache/
+          df -lh
 
       - name: ClickHouse Test
         run: |
@@ -250,7 +255,9 @@ jobs:
                                      ORDER BY
                                          year ASC,
                                          C_NATION ASC;"
-
+          sudo du -sh /var/jfsCache/
+          sudo du -sh /data/jfs/
+          df -lh
 
 
       - name: Send Slack Notification
diff --git a/.github/workflows/tpch_clickhouse_mode_local.yml b/.github/workflows/tpch_clickhouse_mode_local.yml
index 41023c55..f92178c4 100644
--- a/.github/workflows/tpch_clickhouse_mode_local.yml
+++ b/.github/workflows/tpch_clickhouse_mode_local.yml
@@ -74,14 +74,16 @@ jobs:
       - name: Generate Data
         run: |
           cd /home/runner/work/
-          git clone https://github.com/sanwan/ssb-dbgen.git
-          cd ssb-dbgen
-          make
-          ./dbgen -s 10 -T l -q
+          git clone https://github.com/sanwan/actiondata.git
+          cd actiondata
+          cat lineorder.tbl.gz* >lineorder.tbl.gz
+          gzip -dk lineorder.tbl.gz
           git clone https://github.com/sanwan/actionfile.git
           mv actionfile/*.tbl .
+          rm lineorder.tbl.gz*
           ls -lrth
-          sudo du -sh /home/runner/work/ssb-dbgen/
+          sudo du -sh /home/runner/work/actiondata/
+          df -lh
 
 
       - name: Create Table
@@ -145,7 +147,7 @@ jobs:
                                      ENGINE = MergeTree ORDER BY S_SUPPKEY;"
       - name: Import Data
         run: |
-          cd /home/runner/work/ssb-dbgen/
+          cd /home/runner/work/actiondata/
           clickhouse-client --query "INSERT INTO customer FORMAT CSV" < customer.tbl
           clickhouse-client --query "INSERT INTO part FORMAT CSV" < part.tbl
           clickhouse-client --query "INSERT INTO supplier FORMAT CSV" < supplier.tbl
@@ -250,6 +252,7 @@ jobs:
                                      ORDER BY
                                          year ASC,
                                          C_NATION ASC;"
+          df -lh
 
       - name: Send Slack Notification
         if: ${{ failure() }}
diff --git a/cmd/objbench.go b/cmd/objbench.go
index 85f5c7ae..60fb920c 100644
--- a/cmd/objbench.go
+++ b/cmd/objbench.go
@@ -79,10 +79,19 @@ $ ACCESS_KEY=myAccessKey SECRET_KEY=mySecretKey juicefs objbench --storage s3 --
 				Value: 128,
 				Usage: "size of each small object in KiB",
 			},
+			&cli.UintFlag{
+				Name:  "small-objects",
+				Value: 100,
+				Usage: "number of small object",
+			},
+			&cli.BoolFlag{
+				Name:  "skip-functional-tests",
+				Usage: "skip functional tests",
+			},
 			&cli.UintFlag{
 				Name:    "threads",
 				Aliases: []string{"p"},
-				Value:   1,
+				Value:   4,
 				Usage:   "number of concurrent threads",
 			},
 		},
@@ -98,6 +107,11 @@ var (
 
 func objbench(ctx *cli.Context) error {
 	setup(ctx, 1)
+	for _, name := range []string{"big-object-size", "small-object-size", "block-size", "small-objects", "threads"} {
+		if ctx.Uint(name) == 0 {
+			logger.Fatalf("%s should not be set to zero", name)
+		}
+	}
 	ak, sk := ctx.String("access-key"), ctx.String("secret-key")
 	if ak == "" {
 		ak = os.Getenv("ACCESS_KEY")
@@ -119,6 +133,7 @@ func objbench(ctx *cli.Context) error {
 	fsize := int(ctx.Uint("big-object-size")) << 20
 	smallBSize := int(ctx.Uint("small-object-size")) << 10
 	bCount := int(math.Ceil(float64(fsize) / float64(bSize)))
+	sCount := int(ctx.Uint("small-objects"))
 	threads := int(ctx.Uint("threads"))
 	tty := isatty.IsTerminal(os.Stdout.Fd())
 	progress := utils.NewProgress(!tty, false)
@@ -132,124 +147,70 @@ func objbench(ctx *cli.Context) error {
 
 	var result [][]string
 	result = append(result, []string{"CATEGORY", "TEST", "RESULT"})
-	functionalTesting(blob, fsize, &result, tty)
+	if !ctx.IsSet("skip-functional-tests") {
+		functionalTesting(blob, &result, tty)
+	}
 	printResult(result, tty)
 	fmt.Println("\nStart Performance Testing ...")
 	var pResult [][]string
 	pResult = append(pResult, []string{"ITEM", "VALUE", "COST"})
 
-	{
-		multiUploadFunc := func(key string, content [][]byte) error {
-			total := len(content)
-			bar := progress.AddCountBar("multi-upload", int64(total))
-			defer bar.Done()
-			upload, err := blob.CreateMultipartUpload(key)
-			if err != nil {
-				return err
-			}
-			parts := make([]*object.Part, total)
-			pool := make(chan struct{}, threads)
-			var wg sync.WaitGroup
-			for i := 1; i <= total; i++ {
-				pool <- struct{}{}
-				wg.Add(1)
-				num := i
-				go func() {
-					defer func() {
-						<-pool
-						wg.Done()
-						bar.Increment()
-					}()
-					parts[num-1], err = blob.UploadPart(key, upload.UploadID, num, content[num-1])
-					if err != nil {
-						logger.Fatalf("multipart upload error: %v", err)
-					}
-				}()
-			}
-			wg.Wait()
-			return blob.CompleteUpload(key, upload.UploadID, parts)
-		}
-
-		var uploadResult = []string{"multi-upload", nspt, nspt}
-		fname := fmt.Sprintf("__multi_upload__test__%d__", time.Now().UnixNano())
-		if err := blob.CompleteUpload("test", "fakeUploadId", []*object.Part{}); err != utils.ENOTSUP {
-			part := make([]byte, 5<<20)
-			total := int(math.Ceil(float64(fsize) / float64(len(part))))
-			rand.Read(part)
-			content := make([][]byte, total)
-			for i := 0; i < total; i++ {
-				content[i] = part
-			}
-			start := time.Now()
-			if err := multiUploadFunc(fname, content); err != nil {
-				logger.Fatalf("multipart upload error: %v", err)
-			}
-			cost := time.Since(start).Seconds()
-			uploadResult[1], uploadResult[2] = colorize("multi-upload", float64(total*5)/cost, cost, 2, tty)
-			uploadResult[1] += " MiB/s"
-			uploadResult[2] += " s/object"
-		}
-		pResult = append(pResult, uploadResult)
-		if err = blob.Delete(fname); err != nil {
-			logger.Fatalf("delete %s error: %v", fname, err)
-		}
-	}
-
 	apis := []apiInfo{
 		{
-			name:  "put",
-			count: bCount,
-			title: "upload objects",
+			name:  "smallput",
+			count: sCount,
+			title: "put small objects",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("put", float64(bSize>>20*bCount)/cost, cost*1000/float64(bCount), 2, tty)
-					line[1] += " MiB/s"
+					line[1], line[2] = colorize("smallput", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
 				return line
 			},
 		}, {
-			name:  "get",
-			count: bCount,
-			title: "download objects",
+			name:  "smallget",
+			count: sCount,
+			title: "get small objects",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("get", float64(bSize>>20*bCount)/cost, cost*1000/float64(bCount), 2, tty)
-					line[1] += " MiB/s"
+					line[1], line[2] = colorize("smallget", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
 				return line
 			},
 		}, {
-			name:     "smallput",
+			name:     "put",
 			count:    bCount,
-			title:    "put small objects",
-			startKey: bCount,
+			title:    "upload objects",
+			startKey: sCount,
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("smallput", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
-					line[1] += " objects/s"
+					line[1], line[2] = colorize("put", float64(bSize>>20*bCount)/cost, cost*1000/float64(bCount), 2, tty)
+					line[1] += " MiB/s"
 					line[2] += " ms/object"
 				}
 				return line
 			},
 		}, {
-			name:  "smallget",
-			count: bCount,
-			title: "get small objects",
+			name:     "get",
+			count:    bCount,
+			title:    "download objects",
+			startKey: sCount,
 			after: func(blob object.ObjectStorage) {
-				for i := bCount; i < bCount*2; i++ {
+				for i := sCount; i < sCount+bCount; i++ {
 					_ = blob.Delete(strconv.Itoa(i))
 				}
 			},
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("smallget", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
-					line[1] += " objects/s"
+					line[1], line[2] = colorize("get", float64(bSize>>20*bCount)/cost, cost*1000/float64(bCount), 2, tty)
+					line[1] += " MiB/s"
 					line[2] += " ms/object"
 				}
 				return line
@@ -261,7 +222,7 @@ func objbench(ctx *cli.Context) error {
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("list", float64(bCount)*100/cost, cost*10, 2, tty)
+					line[1], line[2] = colorize("list", float64(sCount)*100/cost, cost*10, 2, tty)
 					line[1] += " objects/s"
 					line[2] += " ms/op"
 				}
@@ -269,12 +230,12 @@ func objbench(ctx *cli.Context) error {
 			},
 		}, {
 			name:  "head",
-			count: bCount,
+			count: sCount,
 			title: "head objects",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("head", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
+					line[1], line[2] = colorize("head", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
 					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
@@ -282,12 +243,12 @@ func objbench(ctx *cli.Context) error {
 			},
 		}, {
 			name:  "chtimes",
-			count: bCount,
+			count: sCount,
 			title: "update mtime",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("chtimes", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
+					line[1], line[2] = colorize("chtimes", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
 					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
@@ -295,12 +256,12 @@ func objbench(ctx *cli.Context) error {
 			},
 		}, {
 			name:  "chmod",
-			count: bCount,
+			count: sCount,
 			title: "change permissions",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("chmod", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
+					line[1], line[2] = colorize("chmod", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
 					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
@@ -308,12 +269,12 @@ func objbench(ctx *cli.Context) error {
 			},
 		}, {
 			name:  "chown",
-			count: bCount,
+			count: sCount,
 			title: "change owner/group",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("chown", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
+					line[1], line[2] = colorize("chown", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
 					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
@@ -321,12 +282,12 @@ func objbench(ctx *cli.Context) error {
 			},
 		}, {
 			name:  "delete",
-			count: bCount,
+			count: sCount,
 			title: "delete objects",
 			getResult: func(cost float64) []string {
 				line := []string{"", nspt, nspt}
 				if cost > 0 {
-					line[1], line[2] = colorize("delete", float64(bCount)/cost, cost*1000/float64(bCount), 1, tty)
+					line[1], line[2] = colorize("delete", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
 					line[1] += " objects/s"
 					line[2] += " ms/object"
 				}
@@ -336,14 +297,14 @@ func objbench(ctx *cli.Context) error {
 	}
 
 	bm := &benchMarkObj{
-		blob:         blob,
-		progressBar:  progress,
-		threads:      threads,
-		content:      make([]byte, bSize),
-		smallContent: make([]byte, smallBSize),
+		blob:        blob,
+		progressBar: progress,
+		threads:     threads,
+		seed:        make([]byte, bSize),
+		smallSeed:   make([]byte, smallBSize),
 	}
-	rand.Read(bm.content)
-	rand.Read(bm.smallContent)
+	rand.Read(bm.seed)
+	rand.Read(bm.smallSeed)
 
 	for _, api := range apis {
 		pResult = append(pResult, bm.run(api))
@@ -351,15 +312,16 @@ func objbench(ctx *cli.Context) error {
 	progress.Done()
 
 	for i := bCount; i < bCount*2; i++ {
-		_ = bm.delete(strconv.Itoa(i))
+		_ = bm.delete(strconv.Itoa(i), 0)
 	}
 
-	fmt.Printf("Benchmark finished! block-size: %d KiB, big-object-size: %d MiB, small-object-size: %d KiB, NumThreads: %d\n",
-		ctx.Uint("block-size"), ctx.Uint("big-object-size"), ctx.Uint("small-object-size"), ctx.Uint("threads"))
+	fmt.Printf("Benchmark finished! block-size: %d KiB, big-object-size: %d MiB, small-object-size: %d KiB, small-objects: %d, NumThreads: %d\n",
+		ctx.Uint("block-size"), ctx.Uint("big-object-size"), ctx.Uint("small-object-size"), sCount, threads)
 
 	// adjust the print order
-	pResult[1], pResult[2] = pResult[2], pResult[1]
-	pResult[8], pResult[11] = pResult[11], pResult[8]
+	pResult[1], pResult[3] = pResult[3], pResult[1]
+	pResult[2], pResult[4] = pResult[4], pResult[2]
+	pResult[7], pResult[10] = pResult[10], pResult[7]
 	printResult(pResult, tty)
 	return nil
 }
@@ -417,15 +379,15 @@ type apiInfo struct {
 }
 
 type benchMarkObj struct {
-	progressBar           *utils.Progress
-	blob                  object.ObjectStorage
-	threads               int
-	content, smallContent []byte
+	progressBar     *utils.Progress
+	blob            object.ObjectStorage
+	threads         int
+	seed, smallSeed []byte
 }
 
 func (bm *benchMarkObj) run(api apiInfo) []string {
 	if api.name == "chown" || api.name == "chmod" || api.name == "chtimes" {
-		if err := bm.chmod("not_exists"); err == utils.ENOTSUP {
+		if err := bm.chmod("not_exists", 0); err == utils.ENOTSUP {
 			line := api.getResult(-1)
 			line[0] = api.title
 			return line
@@ -435,14 +397,16 @@ func (bm *benchMarkObj) run(api apiInfo) []string {
 			return []string{api.title, skipped, skipped}
 		}
 	}
-	var fn func(key string) error
+	var fn func(key string, startKey int) error
 	switch api.name {
 	case "put":
 		fn = bm.put
-	case "smallget", "get":
+	case "get":
 		fn = bm.get
 	case "smallput":
 		fn = bm.smallPut
+	case "smallget":
+		fn = bm.smallGet
 	case "delete":
 		fn = bm.delete
 	case "head":
@@ -458,7 +422,6 @@ func (bm *benchMarkObj) run(api apiInfo) []string {
 	}
 
 	var wg sync.WaitGroup
-	start := time.Now()
 	pool := make(chan struct{}, bm.threads)
 	count := api.count
 	if api.count != 0 {
@@ -466,6 +429,7 @@ func (bm *benchMarkObj) run(api apiInfo) []string {
 	}
 	bar := bm.progressBar.AddCountBar(api.title, int64(count))
 	var err error
+	start := time.Now()
 	for i := api.startKey; i < api.startKey+count; i++ {
 		pool <- struct{}{}
 		wg.Add(1)
@@ -474,7 +438,7 @@ func (bm *benchMarkObj) run(api apiInfo) []string {
 				<-pool
 				wg.Done()
 			}()
-			if e := fn(strconv.Itoa(key)); e != nil {
+			if e := fn(strconv.Itoa(key), api.startKey); e != nil {
 				err = e
 			}
 			bar.Increment()
@@ -482,6 +446,7 @@ func (bm *benchMarkObj) run(api apiInfo) []string {
 	}
 	wg.Wait()
 	bar.Done()
+	line := api.getResult(time.Since(start).Seconds())
 	if api.after != nil {
 		api.after(bm.blob)
 	}
@@ -489,54 +454,96 @@ func (bm *benchMarkObj) run(api apiInfo) []string {
 		logger.Errorf("%s test failed: %s", api.name, err)
 		return []string{api.title, failed, failed}
 	}
-	line := api.getResult(time.Since(start).Seconds())
 	line[0] = api.title
 	return line
 }
 
-func (bm *benchMarkObj) put(key string) error {
-	return bm.blob.Put(key, bytes.NewReader(bm.content))
+func getMockData(seed []byte, idx int) []byte {
+	size := len(seed)
+	if size == 0 {
+		return nil
+	}
+	content := make([]byte, size)
+	if idx == 0 {
+		content = seed
+	} else {
+		i := idx % size
+		copy(content[:size-i], seed[i:size])
+		copy(content[size-i:size], seed[:i])
+	}
+	return content
+}
+
+func (bm *benchMarkObj) put(key string, startKey int) error {
+	idx, _ := strconv.Atoi(key)
+	return bm.blob.Put(key, bytes.NewReader(getMockData(bm.seed, idx-startKey)))
 }
 
-func (bm *benchMarkObj) smallPut(key string) error {
-	return bm.blob.Put(key, bytes.NewReader(bm.smallContent))
+func (bm *benchMarkObj) smallPut(key string, startKey int) error {
+	idx, _ := strconv.Atoi(key)
+	return bm.blob.Put(key, bytes.NewReader(getMockData(bm.smallSeed, idx)))
 }
 
-func (bm *benchMarkObj) get(key string) error {
-	r, err := bm.blob.Get(key, 0, -1)
+func getAndCheckN(blob object.ObjectStorage, key string, seed []byte, getOrgIdx func(idx int) int) error {
+	idx, _ := strconv.Atoi(key)
+	r, err := blob.Get(key, 0, -1)
 	if err != nil {
 		return err
 	}
 	defer r.Close()
-	_, err = io.Copy(io.Discard, r)
-	return err
+	content, err := io.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	orgIdx := getOrgIdx(idx)
+	checkN := 10
+	l := len(seed)
+	if l < checkN {
+		checkN = l
+	}
+	if len(content) != len(seed) || !bytes.Equal(content[:checkN], getMockData(seed, orgIdx)[:checkN]) {
+		return fmt.Errorf("the downloaded content is incorrect")
+	}
+	return nil
 }
 
-func (bm *benchMarkObj) delete(key string) error {
+func (bm *benchMarkObj) get(key string, startKey int) error {
+	return getAndCheckN(bm.blob, key, bm.seed, func(idx int) int {
+		return idx - startKey
+	})
+}
+
+func (bm *benchMarkObj) smallGet(key string, startKey int) error {
+	return getAndCheckN(bm.blob, key, bm.smallSeed, func(idx int) int {
+		return idx
+	})
+}
+
+func (bm *benchMarkObj) delete(key string, startKey int) error {
 	return bm.blob.Delete(key)
 }
 
-func (bm *benchMarkObj) head(key string) error {
+func (bm *benchMarkObj) head(key string, startKey int) error {
 	_, err := bm.blob.Head(key)
 	return err
 }
 
-func (bm *benchMarkObj) list(key string) error {
+func (bm *benchMarkObj) list(key string, startKey int) error {
 	result, err := osync.ListAll(bm.blob, "", "")
 	for range result {
 	}
 	return err
 }
 
-func (bm *benchMarkObj) chown(key string) error {
+func (bm *benchMarkObj) chown(key string, startKey int) error {
 	return bm.blob.(object.FileSystem).Chown(key, "nobody", "nogroup")
 }
 
-func (bm *benchMarkObj) chmod(key string) error {
+func (bm *benchMarkObj) chmod(key string, startKey int) error {
 	return bm.blob.(object.FileSystem).Chmod(key, 0755)
 }
 
-func (bm *benchMarkObj) chtimes(key string) error {
+func (bm *benchMarkObj) chtimes(key string, startKey int) error {
 	return bm.blob.(object.FileSystem).Chtimes(key, time.Now())
 }
 
@@ -564,7 +571,7 @@ var syncTests = map[string]bool{
 	"multipart upload":    true,
 }
 
-func functionalTesting(blob object.ObjectStorage, fsize int, result *[][]string, tty bool) {
+func functionalTesting(blob object.ObjectStorage, result *[][]string, tty bool) {
 	runCase := func(title string, fn func(blob object.ObjectStorage) error) {
 		r := pass
 		if err := fn(blob); err == utils.ENOTSUP {
@@ -806,7 +813,8 @@ func functionalTesting(blob object.ObjectStorage, fsize int, result *[][]string,
 	})
 
 	runCase("put a big object", func(blob object.ObjectStorage) error {
-		buffL := 1 << 20
+		fsize := 256 << 20
+		buffL := 4 << 20
 		buff := make([]byte, buffL)
 		rand.Read(buff)
 		count := int(math.Floor(float64(fsize) / float64(buffL)))
@@ -840,38 +848,57 @@ func functionalTesting(blob object.ObjectStorage, fsize int, result *[][]string,
 		key := "multi_test_file"
 		if err := blob.CompleteUpload(key, "notExistsUploadId", []*object.Part{}); err != utils.ENOTSUP {
 			defer blob.Delete(key) //nolint:errcheck
-			if uploader, err := blob.CreateMultipartUpload(key); err == nil {
-				partSize := uploader.MinPartSize
-				uploadID := uploader.UploadID
-				defer blob.AbortUpload(key, uploadID)
-
-				part1, err := blob.UploadPart(key, uploadID, 1, make([]byte, partSize))
-				if err != nil {
-					return fmt.Errorf("uploadPart 1 failed: %s", err)
-				}
-				part2Size := 1 << 20
-				_, err = blob.UploadPart(key, uploadID, 2, make([]byte, part2Size))
-				if err != nil {
-					return fmt.Errorf("uploadPart 2 failed: %s", err)
-				}
-				part2Size = 2 << 20
-				part2, err := blob.UploadPart(key, uploadID, 2, make([]byte, part2Size))
-				if err != nil {
-					return fmt.Errorf("uploadPart 2 failed: %s", err)
-				}
-
-				if err := blob.CompleteUpload(key, uploadID, []*object.Part{part1, part2}); err != nil {
-					return fmt.Errorf("completeMultipart failed: %s", err)
-				}
-				if in, err := blob.Get(key, 0, -1); err != nil {
-					return fmt.Errorf("failed to download file,key=%s", key)
-				} else if d, err := ioutil.ReadAll(in); err != nil {
-					return fmt.Errorf("failed to read downloaded content key=%s", key)
-				} else if len(d) != partSize+part2Size {
-					return fmt.Errorf("size of %s file: %d != %d", key, len(d), partSize+part2Size)
-				}
-				return nil
+			partSize := 5 << 20
+			total := 10
+			seed := make([]byte, partSize)
+			rand.Read(seed)
+			upload, err := blob.CreateMultipartUpload(key)
+			if err != nil {
+				return fmt.Errorf("create multipart upload failed: %s", err)
+			}
+			parts := make([]*object.Part, total)
+			content := make([][]byte, total)
+			for i := 0; i < total; i++ {
+				content[i] = getMockData(seed, i)
+			}
+			pool := make(chan struct{}, 4)
+			var wg sync.WaitGroup
+			for i := 1; i <= total; i++ {
+				pool <- struct{}{}
+				wg.Add(1)
+				num := i
+				go func() {
+					defer func() {
+						<-pool
+						wg.Done()
+					}()
+					parts[num-1], err = blob.UploadPart(key, upload.UploadID, num, content[num-1])
+					if err != nil {
+						logger.Fatalf("multipart upload error: %v", err)
+					}
+				}()
+			}
+			wg.Wait()
+			// overwrite part
+			if parts[total-1], err = blob.UploadPart(key, upload.UploadID, total, getMockData(seed, total)); err != nil {
+				logger.Fatalf("multipart upload error: %v", err)
+			}
+			if err = blob.CompleteUpload(key, upload.UploadID, parts); err != nil {
+				logger.Fatalf("failed to complete multipart upload: %v", err)
+			}
+			r, err := blob.Get(key, 0, -1)
+			if err != nil {
+				logger.Fatalf("failed to get multipart upload file: %v", err)
+			}
+			cnt, err := io.ReadAll(r)
+			if err != nil {
+				logger.Fatalf("failed to get multipart upload file: %v", err)
+			}
+			content[total-1] = getMockData(seed, total)
+			if !bytes.Equal(cnt, bytes.Join(content, nil)) {
+				logger.Fatal("the content of the multipart upload file is incorrect")
 			}
+			return nil
 		}
 		return utils.ENOTSUP
 	})
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index 4b238040..adc49d46 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -70,6 +70,7 @@ type engine interface {
 	doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst Ino, nameDst string, flags uint32, inode *Ino, attr *Attr) syscall.Errno
 	doSetXattr(ctx Context, inode Ino, name string, value []byte, flags uint32) syscall.Errno
 	doRemoveXattr(ctx Context, inode Ino, name string) syscall.Errno
+	doGetParents(ctx Context, inode Ino) map[Ino]int
 
 	GetSession(sid uint64, detail bool) (*Session, error)
 }
@@ -846,6 +847,22 @@ func (m *baseMeta) RemoveXattr(ctx Context, inode Ino, name string) syscall.Errn
 	return m.en.doRemoveXattr(ctx, m.checkRoot(inode), name)
 }
 
+func (m *baseMeta) GetParents(ctx Context, inode Ino) map[Ino]int {
+	if inode == 1 {
+		return map[Ino]int{1: 1}
+	}
+	var attr Attr
+	if st := m.GetAttr(ctx, inode, &attr); st != 0 {
+		logger.Warnf("GetAttr inode %d: %s", inode, st)
+		return nil
+	}
+	if attr.Parent > 0 {
+		return map[Ino]int{attr.Parent: 1}
+	} else {
+		return m.en.doGetParents(ctx, inode)
+	}
+}
+
 func (m *baseMeta) fileDeleted(opened bool, inode Ino, length uint64) {
 	if opened {
 		m.Lock()
diff --git a/pkg/meta/dump.go b/pkg/meta/dump.go
index 08d5e6b3..224ebef1 100644
--- a/pkg/meta/dump.go
+++ b/pkg/meta/dump.go
@@ -88,7 +88,7 @@ type DumpedXattr struct {
 
 type DumpedEntry struct {
 	Name    string                  `json:"-"`
-	Parent  Ino                     `json:"-"`
+	Parents []Ino                   `json:"-"`
 	Attr    *DumpedAttr             `json:"attr,omitempty"`
 	Symlink string                  `json:"symlink,omitempty"`
 	Xattrs  []*DumpedXattr          `json:"xattrs,omitempty"`
@@ -317,8 +317,10 @@ func collectEntry(e *DumpedEntry, entries map[Ino]*DumpedEntry, showProgress fun
 			return fmt.Errorf("inode conflict: %d", inode)
 		}
 		eattr.Nlink++
+		exist.Parents = append(exist.Parents, e.Parents...)
 		if eattr.Ctime*1e9+int64(eattr.Ctimensec) < attr.Ctime*1e9+int64(attr.Ctimensec) {
 			attr.Nlink = eattr.Nlink
+			e.Parents = exist.Parents
 			entries[inode] = e
 		}
 		return nil
@@ -329,12 +331,12 @@ func collectEntry(e *DumpedEntry, entries map[Ino]*DumpedEntry, showProgress fun
 		e.Attr.Nlink = 1 // reset
 	} else if typ == TypeDirectory {
 		if inode == 1 || inode == TrashInode { // root or trash inode
-			e.Parent = 1
+			e.Parents = []Ino{1}
 		}
 		e.Attr.Nlink = 2
 		for name, child := range e.Entries {
 			child.Name = name
-			child.Parent = inode
+			child.Parents = []Ino{inode}
 			if child.Attr == nil {
 				logger.Warnf("ignore empty entry: %s/%s", inode, name)
 				continue
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index 5962a0e3..2e7a3289 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -106,7 +106,7 @@ type Attr struct {
 	Nlink     uint32 // number of links (sub-directories or hardlinks)
 	Length    uint64 // length of regular file
 
-	Parent    Ino  // inode of parent, only for Directory
+	Parent    Ino  // inode of parent; 0 means tracked by parentKey (for hardlinks)
 	Full      bool // the attributes are completed or not
 	KeepCache bool // whether to keep the cached page or not
 }
@@ -311,6 +311,8 @@ type Meta interface {
 	InvalidateChunkCache(ctx Context, inode Ino, indx uint32) syscall.Errno
 	// CopyFileRange copies part of a file to another one.
 	CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, offOut uint64, size uint64, flags uint32, copied *uint64) syscall.Errno
+	// GetParents returns a map of node parents (> 1 parents if hardlinked)
+	GetParents(ctx Context, inode Ino) map[Ino]int
 
 	// GetXattr returns the value of extended attribute for given name.
 	GetXattr(ctx Context, inode Ino, name string, vbuff *[]byte) syscall.Errno
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index 7587acbc..3a7faab2 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -49,6 +49,7 @@ import (
 /*
 	Node: i$inode -> Attribute{type,mode,uid,gid,atime,mtime,ctime,nlink,length,rdev}
 	Dir:   d$inode -> {name -> {inode,type}}
+	Parent: p$inode -> {parent -> count} // for hard links
 	File:  c$inode_$indx -> [Slice{pos,id,length,off,len}]
 	Symlink: s$inode -> target
 	Xattr: x$inode -> {name -> value}
@@ -487,6 +488,10 @@ func (m *redisMeta) entryKey(parent Ino) string {
 	return m.prefix + "d" + parent.String()
 }
 
+func (m *redisMeta) parentKey(inode Ino) string {
+	return m.prefix + "p" + inode.String()
+}
+
 func (m *redisMeta) chunkKey(inode Ino, indx uint32) string {
 	return m.prefix + "c" + inode.String() + "_" + strconv.FormatInt(int64(indx), 10)
 }
@@ -1245,13 +1250,14 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 				if _type == TypeFile && attr.Nlink == 0 {
 					opened = m.of.IsOpen(inode)
 				}
-			} else if attr.Nlink == 1 { // don't change parent if it has hard links
+			} else if attr.Parent > 0 {
 				attr.Parent = trash
 			}
 		} else {
 			logger.Warnf("no attribute for inode %d (%d, %s)", inode, parent, name)
 			trash = 0
 		}
+
 		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
 			pipe.HDel(ctx, m.entryKey(parent), name)
 			if updateParent {
@@ -1261,6 +1267,12 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 				pipe.Set(ctx, m.inodeKey(inode), m.marshal(&attr), 0)
 				if trash > 0 {
 					pipe.HSet(ctx, m.entryKey(trash), m.trashEntry(parent, inode, name), buf)
+					if attr.Parent == 0 {
+						pipe.HIncrBy(ctx, m.parentKey(inode), trash.String(), 1)
+					}
+				}
+				if attr.Parent == 0 {
+					pipe.HIncrBy(ctx, m.parentKey(inode), parent.String(), -1)
 				}
 			} else {
 				switch _type {
@@ -1285,6 +1297,9 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 					pipe.Decr(ctx, m.totalInodesKey())
 				}
 				pipe.Del(ctx, m.xattrKey(inode))
+				if attr.Parent == 0 {
+					pipe.Del(ctx, m.parentKey(inode))
+				}
 			}
 			return nil
 		})
@@ -1480,11 +1495,15 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			tattr.Ctime = now.Unix()
 			tattr.Ctimensec = uint32(now.Nanosecond())
 			if exchange {
-				tattr.Parent = parentSrc
-				if dtyp == TypeDirectory && parentSrc != parentDst {
-					dattr.Nlink--
-					sattr.Nlink++
-					supdate, dupdate = true, true
+				if parentSrc != parentDst {
+					if dtyp == TypeDirectory {
+						tattr.Parent = parentSrc
+						dattr.Nlink--
+						sattr.Nlink++
+						supdate, dupdate = true, true
+					} else if tattr.Parent > 0 {
+						tattr.Parent = parentSrc
+					}
 				}
 			} else {
 				if dtyp == TypeDirectory {
@@ -1507,7 +1526,7 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 							opened = m.of.IsOpen(dino)
 						}
 						defer func() { m.of.InvalidateChunk(dino, 0xFFFFFFFE) }()
-					} else if tattr.Nlink == 1 {
+					} else if tattr.Parent > 0 {
 						tattr.Parent = trash
 					}
 				}
@@ -1525,10 +1544,15 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			return syscall.EACCES
 		}
 
-		if typ == TypeDirectory && parentSrc != parentDst {
-			sattr.Nlink--
-			dattr.Nlink++
-			supdate, dupdate = true, true
+		if parentSrc != parentDst {
+			if typ == TypeDirectory {
+				iattr.Parent = parentDst
+				sattr.Nlink--
+				dattr.Nlink++
+				supdate, dupdate = true, true
+			} else if iattr.Parent > 0 {
+				iattr.Parent = parentDst
+			}
 		}
 		if supdate || now.Sub(time.Unix(sattr.Mtime, int64(sattr.Mtimensec))) >= minUpdateTime {
 			sattr.Mtime = now.Unix()
@@ -1544,7 +1568,6 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			dattr.Ctimensec = uint32(now.Nanosecond())
 			dupdate = true
 		}
-		iattr.Parent = parentDst
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		if inode != nil {
@@ -1557,14 +1580,25 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			if exchange { // dbuf, tattr are valid
 				pipe.HSet(ctx, m.entryKey(parentSrc), nameSrc, dbuf)
 				pipe.Set(ctx, m.inodeKey(dino), m.marshal(&tattr), 0)
+				if parentSrc != parentDst && tattr.Parent == 0 {
+					pipe.HIncrBy(ctx, m.parentKey(dino), parentSrc.String(), 1)
+					pipe.HIncrBy(ctx, m.parentKey(dino), parentDst.String(), -1)
+				}
 			} else {
 				pipe.HDel(ctx, m.entryKey(parentSrc), nameSrc)
 				if dino > 0 {
 					if trash > 0 {
 						pipe.Set(ctx, m.inodeKey(dino), m.marshal(&tattr), 0)
 						pipe.HSet(ctx, m.entryKey(trash), m.trashEntry(parentDst, dino, nameDst), dbuf)
+						if tattr.Parent == 0 {
+							pipe.HIncrBy(ctx, m.parentKey(dino), trash.String(), 1)
+							pipe.HIncrBy(ctx, m.parentKey(dino), parentDst.String(), -1)
+						}
 					} else if dtyp != TypeDirectory && tattr.Nlink > 0 {
 						pipe.Set(ctx, m.inodeKey(dino), m.marshal(&tattr), 0)
+						if tattr.Parent == 0 {
+							pipe.HIncrBy(ctx, m.parentKey(dino), parentDst.String(), -1)
+						}
 					} else {
 						if dtyp == TypeFile {
 							if opened {
@@ -1587,11 +1621,20 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 							pipe.Decr(ctx, m.totalInodesKey())
 						}
 						pipe.Del(ctx, m.xattrKey(dino))
+						if tattr.Parent == 0 {
+							pipe.Del(ctx, m.parentKey(dino))
+						}
 					}
 				}
 			}
-			if parentDst != parentSrc && !isTrash(parentSrc) && supdate {
-				pipe.Set(ctx, m.inodeKey(parentSrc), m.marshal(&sattr), 0)
+			if parentDst != parentSrc {
+				if !isTrash(parentSrc) && supdate {
+					pipe.Set(ctx, m.inodeKey(parentSrc), m.marshal(&sattr), 0)
+				}
+				if iattr.Parent == 0 {
+					pipe.HIncrBy(ctx, m.parentKey(ino), parentDst.String(), 1)
+					pipe.HIncrBy(ctx, m.parentKey(ino), parentSrc.String(), -1)
+				}
 			}
 			pipe.Set(ctx, m.inodeKey(ino), m.marshal(&iattr), 0)
 			pipe.HSet(ctx, m.entryKey(parentDst), nameDst, buf)
@@ -1638,6 +1681,8 @@ func (m *redisMeta) doLink(ctx Context, inode, parent Ino, name string, attr *At
 		if iattr.Typ == TypeDirectory {
 			return syscall.EPERM
 		}
+		oldParent := iattr.Parent
+		iattr.Parent = 0
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		iattr.Nlink++
@@ -1657,6 +1702,10 @@ func (m *redisMeta) doLink(ctx Context, inode, parent Ino, name string, attr *At
 				pipe.Set(ctx, m.inodeKey(parent), m.marshal(&pattr), 0)
 			}
 			pipe.Set(ctx, m.inodeKey(inode), m.marshal(&iattr), 0)
+			if oldParent > 0 {
+				pipe.HIncrBy(ctx, m.parentKey(inode), oldParent.String(), 1)
+			}
+			pipe.HIncrBy(ctx, m.parentKey(inode), parent.String(), 1)
 			return nil
 		})
 		if err == nil && attr != nil {
@@ -2090,6 +2139,22 @@ func (m *redisMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino,
 	return errno(err)
 }
 
+func (m *redisMeta) doGetParents(ctx Context, inode Ino) map[Ino]int {
+	vals, err := m.rdb.HGetAll(ctx, m.parentKey(inode)).Result()
+	if err != nil {
+		logger.Warnf("Scan parent key of inode %d: %s", inode, err)
+		return nil
+	}
+	ps := make(map[Ino]int)
+	for k, v := range vals {
+		if n, _ := strconv.Atoi(v); n > 0 {
+			ino, _ := strconv.ParseUint(k, 10, 64)
+			ps[Ino(ino)] = n
+		}
+	}
+	return ps
+}
+
 // For now only deleted files
 func (m *redisMeta) cleanupLegacies() {
 	for {
@@ -3196,7 +3261,14 @@ func (m *redisMeta) loadEntry(e *DumpedEntry, p redis.Pipeliner, tryExec func(),
 	logger.Debugf("Loading entry inode %d name %s", inode, unescape(e.Name))
 	ctx := Background
 	attr := loadAttr(e.Attr)
-	attr.Parent = e.Parent
+	numParents := len(e.Parents)
+	if numParents == 1 {
+		attr.Parent = e.Parents[0]
+	} else if numParents == 0 {
+		logger.Fatalf("No parent for inode: %d", inode)
+	} else if attr.Typ == TypeDirectory {
+		logger.Fatalf("Too many parents for directory inode %d: %v", inode, e.Parents)
+	}
 	batch := 100
 	if attr.Typ == TypeFile {
 		attr.Length = e.Attr.Length
@@ -3258,6 +3330,11 @@ func (m *redisMeta) loadEntry(e *DumpedEntry, p redis.Pipeliner, tryExec func(),
 		}
 		p.HSet(ctx, m.xattrKey(inode), xattrs)
 	}
+	if numParents > 1 {
+		for _, parent := range e.Parents {
+			p.HIncrBy(ctx, m.parentKey(inode), parent.String(), 1)
+		}
+	}
 	p.Set(ctx, m.inodeKey(inode), m.marshal(attr), 0)
 }
 
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index f14571dd..b4f565a2 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -61,6 +61,12 @@ type edge struct {
 	Type   uint8  `xorm:"notnull"`
 }
 
+type linkParent struct {
+	Id     int64 `xorm:"pk bigserial"`
+	Inode  Ino   `xorm:"index notnull"`
+	Parent Ino   `xorm:"notnull"`
+}
+
 type node struct {
 	Inode  Ino    `xorm:"pk"`
 	Type   uint8  `xorm:"notnull"`
@@ -224,23 +230,31 @@ func (m *dbMeta) doDeleteSlice(chunkid uint64, size uint32) error {
 	})
 }
 
+func (m *dbMeta) syncTable(beans ...interface{}) error {
+	err := m.db.Sync2(beans...)
+	if err != nil && strings.Contains(err.Error(), "Duplicate key") {
+		err = nil
+	}
+	return err
+}
+
 func (m *dbMeta) Init(format Format, force bool) error {
-	if err := m.db.Sync2(new(setting), new(counter)); err != nil {
+	if err := m.syncTable(new(setting), new(counter)); err != nil {
 		return fmt.Errorf("create table setting, counter: %s", err)
 	}
-	if err := m.db.Sync2(new(edge)); err != nil && !strings.Contains(err.Error(), "Duplicate entry") {
-		return fmt.Errorf("create table edge: %s", err)
+	if err := m.syncTable(new(edge), new(linkParent)); err != nil {
+		return fmt.Errorf("create table edge, linkParent: %s", err)
 	}
-	if err := m.db.Sync2(new(node), new(symlink), new(xattr)); err != nil {
+	if err := m.syncTable(new(node), new(symlink), new(xattr)); err != nil {
 		return fmt.Errorf("create table node, symlink, xattr: %s", err)
 	}
-	if err := m.db.Sync2(new(chunk), new(chunkRef), new(delslices)); err != nil {
+	if err := m.syncTable(new(chunk), new(chunkRef), new(delslices)); err != nil {
 		return fmt.Errorf("create table chunk, chunk_ref, delslices: %s", err)
 	}
-	if err := m.db.Sync2(new(session2), new(sustained), new(delfile)); err != nil {
+	if err := m.syncTable(new(session2), new(sustained), new(delfile)); err != nil {
 		return fmt.Errorf("create table session2, sustaind, delfile: %s", err)
 	}
-	if err := m.db.Sync2(new(flock), new(plock)); err != nil {
+	if err := m.syncTable(new(flock), new(plock)); err != nil {
 		return fmt.Errorf("create table flock, plock: %s", err)
 	}
 
@@ -316,7 +330,7 @@ func (m *dbMeta) Init(format Format, force bool) error {
 
 func (m *dbMeta) Reset() error {
 	return m.db.DropTables(&setting{}, &counter{},
-		&node{}, &edge{}, &symlink{}, &xattr{},
+		&node{}, &edge{}, &linkParent{}, &symlink{}, &xattr{},
 		&chunk{}, &chunkRef{}, &delslices{},
 		&session{}, &session2{}, &sustained{}, &delfile{},
 		&flock{}, &plock{})
@@ -340,16 +354,17 @@ func (m *dbMeta) doLoad() (data []byte, err error) {
 }
 
 func (m *dbMeta) doNewSession(sinfo []byte) error {
-	err := m.db.Sync2(new(session2), new(delslices))
+	// add new table
+	err := m.syncTable(new(session2), new(delslices), new(linkParent))
 	if err != nil {
-		return fmt.Errorf("update table session2, delslices: %s", err)
+		return fmt.Errorf("update table session2, delslices, linkParent: %s", err)
 	}
 	// add primary key
-	if err = m.db.Sync2(new(edge), new(chunk), new(xattr), new(sustained)); err != nil && !strings.Contains(err.Error(), "Duplicate entry") {
+	if err = m.syncTable(new(edge), new(chunk), new(xattr), new(sustained)); err != nil {
 		return fmt.Errorf("update table edge, chunk, xattr, sustained: %s", err)
 	}
 	// update the owner from uint64 to int64
-	if err = m.db.Sync2(new(flock), new(plock)); err != nil && !strings.Contains(err.Error(), "Duplicate entry") {
+	if err = m.syncTable(new(flock), new(plock)); err != nil {
 		return fmt.Errorf("update table flock, plock: %s", err)
 	}
 
@@ -708,11 +723,15 @@ func (m *dbMeta) flushStats() {
 func (m *dbMeta) doLookup(ctx Context, parent Ino, name string, inode *Ino, attr *Attr) syscall.Errno {
 	return errno(m.roTxn(func(s *xorm.Session) error {
 		s = s.Table(&edge{})
+		nn := namedNode{node: node{Parent: parent}, Name: []byte(name)}
+		var exist bool
+		var err error
 		if attr != nil {
 			s = s.Join("INNER", &node{}, "jfs_edge.inode=jfs_node.inode")
+			exist, err = s.Select("jfs_node.*").Get(&nn)
+		} else {
+			exist, err = s.Select("*").Get(&nn)
 		}
-		nn := namedNode{node: node{Parent: parent}, Name: []byte(name)}
-		exist, err := s.Select("*").Get(&nn)
 		if err != nil {
 			return err
 		}
@@ -1203,7 +1222,7 @@ func (m *dbMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				if n.Type == TypeFile && n.Nlink == 0 {
 					opened = m.of.IsOpen(e.Inode)
 				}
-			} else if n.Nlink == 1 {
+			} else if n.Parent > 0 {
 				n.Parent = trash
 			}
 		} else {
@@ -1235,6 +1254,16 @@ func (m *dbMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				if err = mustInsert(s, &edge{Parent: trash, Name: []byte(m.trashEntry(parent, e.Inode, string(e.Name))), Inode: e.Inode, Type: e.Type}); err != nil {
 					return err
 				}
+				if n.Parent == 0 {
+					if err = mustInsert(s, &linkParent{Inode: e.Inode, Parent: trash}); err != nil {
+						return err
+					}
+				}
+			}
+			if n.Parent == 0 {
+				if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: e.Inode, Parent: parent}); err != nil {
+					return err
+				}
 			}
 		} else {
 			switch e.Type {
@@ -1269,6 +1298,11 @@ func (m *dbMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 			if _, err := s.Delete(&xattr{Inode: e.Inode}); err != nil {
 				return err
 			}
+			if n.Parent == 0 {
+				if _, err = s.Delete(&linkParent{Inode: e.Inode}); err != nil {
+					return err
+				}
+			}
 		}
 		return err
 	})
@@ -1473,11 +1507,15 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			}
 			dn.Ctime = now
 			if exchange {
-				dn.Parent = parentSrc
-				if de.Type == TypeDirectory && parentSrc != parentDst {
-					dpn.Nlink--
-					spn.Nlink++
-					supdate, dupdate = true, true
+				if parentSrc != parentDst {
+					if de.Type == TypeDirectory {
+						dn.Parent = parentSrc
+						dpn.Nlink--
+						spn.Nlink++
+						supdate, dupdate = true, true
+					} else if dn.Parent > 0 {
+						dn.Parent = parentSrc
+					}
 				}
 			} else {
 				if de.Type == TypeDirectory {
@@ -1500,7 +1538,7 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 							opened = m.of.IsOpen(dn.Inode)
 						}
 						defer func() { m.of.InvalidateChunk(dino, 0xFFFFFFFE) }()
-					} else if dn.Nlink == 1 {
+					} else if dn.Parent > 0 {
 						dn.Parent = trash
 					}
 				}
@@ -1518,10 +1556,15 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			return syscall.EACCES
 		}
 
-		if se.Type == TypeDirectory && parentSrc != parentDst {
-			spn.Nlink--
-			dpn.Nlink++
-			supdate, dupdate = true, true
+		if parentSrc != parentDst {
+			if se.Type == TypeDirectory {
+				sn.Parent = parentDst
+				spn.Nlink--
+				dpn.Nlink++
+				supdate, dupdate = true, true
+			} else if sn.Parent > 0 {
+				sn.Parent = parentDst
+			}
 		}
 		if supdate || time.Duration(now-spn.Mtime)*1e3 >= minUpdateTime {
 			spn.Mtime = now
@@ -1533,7 +1576,6 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			dpn.Ctime = now
 			dupdate = true
 		}
-		sn.Parent = parentDst
 		sn.Ctime = now
 		if inode != nil {
 			*inode = sn.Inode
@@ -1550,6 +1592,14 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			if _, err := s.Cols("ctime", "parent").Update(dn, &node{Inode: dino}); err != nil {
 				return err
 			}
+			if parentSrc != parentDst && dn.Parent == 0 {
+				if err = mustInsert(s, &linkParent{Inode: dino, Parent: parentSrc}); err != nil {
+					return err
+				}
+				if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: dino, Parent: parentDst}); err != nil {
+					return err
+				}
+			}
 		} else {
 			if n, err := s.Delete(&edge{Parent: parentSrc, Name: se.Name}); err != nil {
 				return err
@@ -1565,10 +1615,23 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 					if err = mustInsert(s, &edge{Parent: trash, Name: []byte(name), Inode: dino, Type: de.Type}); err != nil {
 						return err
 					}
+					if dn.Parent == 0 {
+						if err = mustInsert(s, &linkParent{Inode: dino, Parent: trash}); err != nil {
+							return err
+						}
+						if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: dino, Parent: parentDst}); err != nil {
+							return err
+						}
+					}
 				} else if de.Type != TypeDirectory && dn.Nlink > 0 {
-					if _, err := s.Cols("ctime", "nlink").Update(dn, &node{Inode: dino}); err != nil {
+					if _, err := s.Cols("ctime", "nlink", "parent").Update(dn, &node{Inode: dino}); err != nil {
 						return err
 					}
+					if dn.Parent == 0 {
+						if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: dino, Parent: parentDst}); err != nil {
+							return err
+						}
+					}
 				} else {
 					if de.Type == TypeFile {
 						if opened {
@@ -1601,6 +1664,11 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 					if _, err := s.Delete(&xattr{Inode: dino}); err != nil {
 						return err
 					}
+					if dn.Parent == 0 {
+						if _, err = s.Delete(&linkParent{Inode: dino}); err != nil {
+							return err
+						}
+					}
 				}
 				if _, err := s.Delete(&edge{Parent: parentDst, Name: de.Name}); err != nil {
 					return err
@@ -1610,9 +1678,19 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 				return err
 			}
 		}
-		if parentDst != parentSrc && !isTrash(parentSrc) && supdate {
-			if _, err := s.Cols("nlink", "mtime", "ctime").Update(&spn, &node{Inode: parentSrc}); err != nil {
-				return err
+		if parentDst != parentSrc {
+			if !isTrash(parentSrc) && supdate {
+				if _, err := s.Cols("nlink", "mtime", "ctime").Update(&spn, &node{Inode: parentSrc}); err != nil {
+					return err
+				}
+			}
+			if sn.Parent == 0 {
+				if err = mustInsert(s, &linkParent{Inode: sn.Inode, Parent: parentDst}); err != nil {
+					return err
+				}
+				if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: sn.Inode, Parent: parentSrc}); err != nil {
+					return err
+				}
 			}
 		}
 		if _, err := s.Cols("ctime", "parent").Update(&sn, &node{Inode: sn.Inode}); err != nil {
@@ -1675,6 +1753,8 @@ func (m *dbMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 			pn.Ctime = now
 			updateParent = true
 		}
+		oldParent := n.Parent
+		n.Parent = 0
 		n.Nlink++
 		n.Ctime = now
 
@@ -1686,7 +1766,15 @@ func (m *dbMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 				return err
 			}
 		}
-		if _, err := s.Cols("nlink", "ctime").Update(&n, node{Inode: inode}); err != nil {
+		if _, err := s.Cols("nlink", "ctime", "parent").Update(&n, node{Inode: inode}); err != nil {
+			return err
+		}
+		if oldParent > 0 {
+			if err = mustInsert(s, &linkParent{Inode: inode, Parent: oldParent}); err != nil {
+				return err
+			}
+		}
+		if err = mustInsert(s, &linkParent{Inode: inode, Parent: parent}); err != nil {
 			return err
 		}
 		if err == nil {
@@ -2092,6 +2180,22 @@ func (m *dbMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, off
 	return errno(err)
 }
 
+func (m *dbMeta) doGetParents(ctx Context, inode Ino) map[Ino]int {
+	var rows []linkParent
+	if err := m.roTxn(func(s *xorm.Session) error {
+		rows = nil
+		return s.Find(&rows, &linkParent{Inode: inode})
+	}); err != nil {
+		logger.Warnf("Scan parent key of inode %d: %s", inode, err)
+		return nil
+	}
+	ps := make(map[Ino]int)
+	for _, row := range rows {
+		ps[row.Parent]++
+	}
+	return ps
+}
+
 func (m *dbMeta) doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) {
 	files := make(map[Ino]uint64)
 	err := m.roTxn(func(s *xorm.Session) error {
@@ -2698,7 +2802,6 @@ func (m *dbMeta) dumpDir(inode Ino, tree *DumpedEntry, bw *bufio.Writer, depth i
 
 		entry.Name = string(e.Name)
 		if e.Type == TypeDirectory {
-			logger.Infof("dump dir %d %s -> %d", inode, e.Name, e.Inode)
 			err = m.dumpDir(e.Inode, entry, bw, depth+2, showProgress)
 		} else {
 			err = entry.writeJSON(bw, depth+2)
@@ -2916,18 +3019,25 @@ func (m *dbMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[uint64]*
 	logger.Debugf("Loading entry inode %d name %s", inode, unescape(e.Name))
 	attr := e.Attr
 	n := &node{
-		Inode:  inode,
-		Type:   typeFromString(attr.Type),
-		Mode:   attr.Mode,
-		Uid:    attr.Uid,
-		Gid:    attr.Gid,
-		Atime:  attr.Atime*1e6 + int64(attr.Atimensec)/1e3,
-		Mtime:  attr.Mtime*1e6 + int64(attr.Atimensec)/1e3,
-		Ctime:  attr.Ctime*1e6 + int64(attr.Atimensec)/1e3,
-		Nlink:  attr.Nlink,
-		Rdev:   attr.Rdev,
-		Parent: e.Parent,
-	} // Length not set
+		Inode: inode,
+		Type:  typeFromString(attr.Type),
+		Mode:  attr.Mode,
+		Uid:   attr.Uid,
+		Gid:   attr.Gid,
+		Atime: attr.Atime*1e6 + int64(attr.Atimensec)/1e3,
+		Mtime: attr.Mtime*1e6 + int64(attr.Atimensec)/1e3,
+		Ctime: attr.Ctime*1e6 + int64(attr.Atimensec)/1e3,
+		Nlink: attr.Nlink,
+		Rdev:  attr.Rdev,
+	} // Length and Parent not set
+	numParents := len(e.Parents)
+	if numParents == 1 {
+		n.Parent = e.Parents[0]
+	} else if numParents == 0 {
+		logger.Fatalf("No parent for inode: %d", inode)
+	} else if n.Type == TypeDirectory {
+		logger.Fatalf("Too many parents for directory inode %d: %v", inode, e.Parents)
+	}
 	if n.Type == TypeFile {
 		n.Length = attr.Length
 		bar.IncrTotal(int64(len(e.Chunks)))
@@ -2988,6 +3098,12 @@ func (m *dbMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[uint64]*
 			beansCh <- &xattr{Inode: inode, Name: x.Name, Value: unescape(x.Value)}
 		}
 	}
+	if numParents > 1 {
+		bar.IncrTotal(int64(len(e.Parents)))
+		for _, parent := range e.Parents {
+			beansCh <- &linkParent{Inode: inode, Parent: parent}
+		}
+	}
 	beansCh <- n
 }
 
@@ -2999,19 +3115,19 @@ func (m *dbMeta) LoadMeta(r io.Reader) error {
 	if len(tables) > 0 {
 		return fmt.Errorf("Database %s is not empty", m.Name())
 	}
-	if err = m.db.Sync2(new(setting), new(counter)); err != nil {
+	if err = m.syncTable(new(setting), new(counter)); err != nil {
 		return fmt.Errorf("create table setting, counter: %s", err)
 	}
-	if err = m.db.Sync2(new(node), new(edge), new(symlink), new(xattr)); err != nil {
-		return fmt.Errorf("create table node, edge, symlink, xattr: %s", err)
+	if err = m.syncTable(new(node), new(edge), new(linkParent), new(symlink), new(xattr)); err != nil {
+		return fmt.Errorf("create table node, edge, linkParent, symlink, xattr: %s", err)
 	}
-	if err = m.db.Sync2(new(chunk), new(chunkRef), new(delslices)); err != nil {
+	if err = m.syncTable(new(chunk), new(chunkRef), new(delslices)); err != nil {
 		return fmt.Errorf("create table chunk, chunk_ref, delslices: %s", err)
 	}
-	if err = m.db.Sync2(new(session2), new(sustained), new(delfile)); err != nil {
+	if err = m.syncTable(new(session2), new(sustained), new(delfile)); err != nil {
 		return fmt.Errorf("create table session2, sustaind, delfile: %s", err)
 	}
-	if err = m.db.Sync2(new(flock), new(plock)); err != nil {
+	if err = m.syncTable(new(flock), new(plock)); err != nil {
 		return fmt.Errorf("create table flock, plock: %s", err)
 	}
 
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index abc5b94e..e3d3aeeb 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -160,6 +160,7 @@ All keys:
   C...               counter
   AiiiiiiiiI         inode attribute
   AiiiiiiiiD...      dentry
+  AiiiiiiiiPiiiiiiii parents // for hard links
   AiiiiiiiiCnnnn     file chunks
   AiiiiiiiiS         symlink target
   AiiiiiiiiX...      extented attribute
@@ -182,6 +183,10 @@ func (m *kvMeta) entryKey(parent Ino, name string) []byte {
 	return m.fmtKey("A", parent, "D", name)
 }
 
+func (m *kvMeta) parentKey(inode, parent Ino) []byte {
+	return m.fmtKey("A", inode, "P", parent)
+}
+
 func (m *kvMeta) chunkKey(inode Ino, indx uint32) []byte {
 	return m.fmtKey("A", inode, "C", indx)
 }
@@ -1117,13 +1122,14 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				if _type == TypeFile && attr.Nlink == 0 {
 					opened = m.of.IsOpen(inode)
 				}
-			} else if attr.Nlink == 1 {
+			} else if attr.Parent > 0 {
 				attr.Parent = trash
 			}
 		} else {
 			logger.Warnf("no attribute for inode %d (%d, %s)", inode, parent, name)
 			trash = 0
 		}
+
 		defer func() { m.of.InvalidateChunk(inode, 0xFFFFFFFE) }()
 		var updateParent bool
 		if !isTrash(parent) && now.Sub(time.Unix(pattr.Mtime, int64(pattr.Mtimensec))) >= minUpdateTime {
@@ -1142,6 +1148,12 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 			tx.set(m.inodeKey(inode), m.marshal(&attr))
 			if trash > 0 {
 				tx.set(m.entryKey(trash, m.trashEntry(parent, inode, name)), buf)
+				if attr.Parent == 0 {
+					tx.incrBy(m.parentKey(inode, trash), 1)
+				}
+			}
+			if attr.Parent == 0 {
+				tx.incrBy(m.parentKey(inode, parent), -1)
 			}
 		} else {
 			switch _type {
@@ -1162,6 +1174,9 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				newSpace, newInode = -align4K(0), -1
 			}
 			tx.dels(tx.scanKeys(m.xattrKey(inode, ""))...)
+			if attr.Parent == 0 {
+				tx.dels(tx.scanKeys(m.fmtKey("A", inode, "P"))...)
+			}
 		}
 		return nil
 	})
@@ -1316,11 +1331,15 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			tattr.Ctime = now.Unix()
 			tattr.Ctimensec = uint32(now.Nanosecond())
 			if exchange {
-				tattr.Parent = parentSrc
-				if dtyp == TypeDirectory && parentSrc != parentDst {
-					dattr.Nlink--
-					sattr.Nlink++
-					supdate, dupdate = true, true
+				if parentSrc != parentDst {
+					if dtyp == TypeDirectory {
+						tattr.Parent = parentSrc
+						dattr.Nlink--
+						sattr.Nlink++
+						supdate, dupdate = true, true
+					} else if tattr.Parent > 0 {
+						tattr.Parent = parentSrc
+					}
 				}
 			} else {
 				if dtyp == TypeDirectory {
@@ -1339,7 +1358,7 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 							opened = m.of.IsOpen(dino)
 						}
 						defer func() { m.of.InvalidateChunk(dino, 0xFFFFFFFE) }()
-					} else if tattr.Nlink == 1 {
+					} else if tattr.Parent > 0 {
 						tattr.Parent = trash
 					}
 				}
@@ -1357,10 +1376,15 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			return syscall.EACCES
 		}
 
-		if typ == TypeDirectory && parentSrc != parentDst {
-			sattr.Nlink--
-			dattr.Nlink++
-			supdate, dupdate = true, true
+		if parentSrc != parentDst {
+			if typ == TypeDirectory {
+				iattr.Parent = parentDst
+				sattr.Nlink--
+				dattr.Nlink++
+				supdate, dupdate = true, true
+			} else if iattr.Parent > 0 {
+				iattr.Parent = parentDst
+			}
 		}
 		if supdate || now.Sub(time.Unix(sattr.Mtime, int64(sattr.Mtimensec))) >= minUpdateTime {
 			sattr.Mtime = now.Unix()
@@ -1376,7 +1400,6 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			dattr.Ctimensec = uint32(now.Nanosecond())
 			dupdate = true
 		}
-		iattr.Parent = parentDst
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		if inode != nil {
@@ -1389,14 +1412,25 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 		if exchange { // dino > 0
 			tx.set(m.entryKey(parentSrc, nameSrc), dbuf)
 			tx.set(m.inodeKey(dino), m.marshal(&tattr))
+			if parentSrc != parentDst && tattr.Parent == 0 {
+				tx.incrBy(m.parentKey(dino, parentSrc), 1)
+				tx.incrBy(m.parentKey(dino, parentDst), -1)
+			}
 		} else {
 			tx.dels(m.entryKey(parentSrc, nameSrc))
 			if dino > 0 {
 				if trash > 0 {
 					tx.set(m.inodeKey(dino), m.marshal(&tattr))
 					tx.set(m.entryKey(trash, m.trashEntry(parentDst, dino, nameDst)), dbuf)
+					if tattr.Parent == 0 {
+						tx.incrBy(m.parentKey(dino, trash), 1)
+						tx.incrBy(m.parentKey(dino, parentDst), -1)
+					}
 				} else if dtyp != TypeDirectory && tattr.Nlink > 0 {
 					tx.set(m.inodeKey(dino), m.marshal(&tattr))
+					if tattr.Parent == 0 {
+						tx.incrBy(m.parentKey(dino, parentDst), -1)
+					}
 				} else {
 					if dtyp == TypeFile {
 						if opened {
@@ -1415,11 +1449,20 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 						newSpace, newInode = -align4K(0), -1
 					}
 					tx.dels(tx.scanKeys(m.xattrKey(dino, ""))...)
+					if tattr.Parent == 0 {
+						tx.dels(tx.scanKeys(m.fmtKey("A", dino, "P"))...)
+					}
 				}
 			}
 		}
-		if parentDst != parentSrc && !isTrash(parentSrc) && supdate {
-			tx.set(m.inodeKey(parentSrc), m.marshal(&sattr))
+		if parentDst != parentSrc {
+			if !isTrash(parentSrc) && supdate {
+				tx.set(m.inodeKey(parentSrc), m.marshal(&sattr))
+			}
+			if iattr.Parent == 0 {
+				tx.incrBy(m.parentKey(ino, parentDst), 1)
+				tx.incrBy(m.parentKey(ino, parentSrc), -1)
+			}
 		}
 		tx.set(m.inodeKey(ino), m.marshal(&iattr))
 		tx.set(m.entryKey(parentDst, nameDst), buf)
@@ -1466,6 +1509,8 @@ func (m *kvMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 			pattr.Ctimensec = uint32(now.Nanosecond())
 			updateParent = true
 		}
+		oldParent := iattr.Parent
+		iattr.Parent = 0
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		iattr.Nlink++
@@ -1474,6 +1519,10 @@ func (m *kvMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 			tx.set(m.inodeKey(parent), m.marshal(&pattr))
 		}
 		tx.set(m.inodeKey(inode), m.marshal(&iattr))
+		if oldParent > 0 {
+			tx.incrBy(m.parentKey(inode, oldParent), 1)
+		}
+		tx.incrBy(m.parentKey(inode, parent), 1)
 		if attr != nil {
 			*attr = iattr
 		}
@@ -1762,6 +1811,22 @@ func (m *kvMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, off
 	return errno(err)
 }
 
+func (m *kvMeta) doGetParents(ctx Context, inode Ino) map[Ino]int {
+	vals, err := m.scanValues(m.fmtKey("A", inode, "P"), -1, func(k, v []byte) bool {
+		// parents: AiiiiiiiiPiiiiiiii
+		return len(k) == 1+8+1+8 && parseCounter(v) > 0
+	})
+	if err != nil {
+		logger.Warnf("Scan parent key of inode %d: %s", inode, err)
+		return nil
+	}
+	ps := make(map[Ino]int)
+	for k, v := range vals {
+		ps[m.decodeInode([]byte(k[10:]))] = int(parseCounter(v))
+	}
+	return ps
+}
+
 func (m *kvMeta) doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) {
 	klen := 1 + 8 + 8
 	vals, err := m.scanValues(m.fmtKey("D"), limit, func(k, v []byte) bool {
@@ -2448,7 +2513,14 @@ func (m *kvMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[string]i
 	inode := e.Attr.Inode
 	logger.Debugf("Loading entry inode %d name %s", inode, unescape(e.Name))
 	attr := loadAttr(e.Attr)
-	attr.Parent = e.Parent
+	numParents := len(e.Parents)
+	if numParents == 1 {
+		attr.Parent = e.Parents[0]
+	} else if numParents == 0 {
+		logger.Fatalf("No parent for inode: %d", inode)
+	} else if attr.Typ == TypeDirectory {
+		logger.Fatalf("Too many parents for directory inode %d: %v", inode, e.Parents)
+	}
 	return m.txn(func(tx kvTxn) error {
 		if attr.Typ == TypeFile {
 			attr.Length = e.Attr.Length
@@ -2497,6 +2569,11 @@ func (m *kvMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[string]i
 		for _, x := range e.Xattrs {
 			tx.set(m.xattrKey(inode, x.Name), []byte(unescape(x.Value)))
 		}
+		if numParents > 1 {
+			for _, parent := range e.Parents {
+				tx.incrBy(m.parentKey(inode, parent), 1)
+			}
+		}
 		tx.set(m.inodeKey(inode), m.marshal(attr))
 		return nil
 	})


Test patch:
diff --git a/pkg/meta/redis_test.go b/pkg/meta/redis_test.go
index 40f28412..5f081316 100644
--- a/pkg/meta/redis_test.go
+++ b/pkg/meta/redis_test.go
@@ -21,6 +21,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"reflect"
 	"runtime"
 	"strconv"
 	"strings"
@@ -66,6 +67,7 @@ func testMeta(t *testing.T, m Meta) {
 	testMetaClient(t, m)
 	testTruncateAndDelete(t, m)
 	testTrash(t, m)
+	testParents(t, m)
 	testRemove(t, m)
 	testStickyBit(t, m)
 	testLocks(t, m)
@@ -1250,6 +1252,79 @@ func testTrash(t *testing.T, m Meta) {
 	}
 }
 
+func testParents(t *testing.T, m Meta) {
+	if err := m.Init(Format{Name: "test"}, false); err != nil {
+		t.Fatalf("init: %s", err)
+	}
+	ctx := Background
+	var inode, parent Ino
+	var attr = &Attr{}
+	if st := m.Create(ctx, 1, "f", 0644, 022, 0, &inode, attr); st != 0 {
+		t.Fatalf("create f: %s", st)
+	}
+	if attr.Parent != 1 {
+		t.Fatalf("expect parent 1, but got %d", attr.Parent)
+	}
+	checkParents := func(inode Ino, expect map[Ino]int) {
+		if ps := m.GetParents(ctx, inode); ps == nil {
+			t.Fatalf("get parents of inode %d returns nil", inode)
+		} else if !reflect.DeepEqual(ps, expect) {
+			t.Fatalf("expect parents %v, but got %v", expect, ps)
+		}
+	}
+	checkParents(inode, map[Ino]int{1: 1})
+
+	if st := m.Link(ctx, inode, 1, "l1", attr); st != 0 {
+		t.Fatalf("link l1 -> f: %s", st)
+	}
+	if attr.Parent != 0 {
+		t.Fatalf("expect parent 0, but got %d", attr.Parent)
+	}
+	checkParents(inode, map[Ino]int{1: 2})
+
+	if st := m.Mkdir(ctx, 1, "d", 0755, 022, 0, &parent, attr); st != 0 {
+		t.Fatalf("mkdir d: %s", st)
+	}
+	if st := m.Link(ctx, inode, parent, "l2", attr); st != 0 {
+		t.Fatalf("link l2 -> f: %s", st)
+	}
+	if st := m.Link(ctx, inode, parent, "l3", attr); st != 0 {
+		t.Fatalf("link l3 -> f: %s", st)
+	}
+	checkParents(inode, map[Ino]int{1: 2, parent: 2})
+
+	if st := m.Unlink(ctx, 1, "f"); st != 0 {
+		t.Fatalf("unlink f: %s", st)
+	}
+	if st := m.Create(ctx, 1, "f2", 0644, 022, 0, &inode, attr); st != 0 {
+		t.Fatalf("create f2: %s", st)
+	}
+	if st := m.Rename(ctx, 1, "f2", 1, "l1", 0, &inode, attr); st != 0 {
+		t.Fatalf("rename f2 -> l1: %s", st)
+	}
+	if st := m.Lookup(ctx, parent, "l2", &inode, attr); st != 0 {
+		t.Fatalf("lookup d/l2: %s", st)
+	}
+	if attr.Parent != 0 {
+		t.Fatalf("expect parent 0, but got %d", attr.Parent)
+	}
+	if st := m.Unlink(ctx, parent, "l2"); st != 0 {
+		t.Fatalf("unlink d/l2: %s", st)
+	}
+	checkParents(inode, map[Ino]int{parent: 1})
+
+	// clean up
+	if st := m.Unlink(ctx, 1, "l1"); st != 0 {
+		t.Fatalf("unlink l1: %s", st)
+	}
+	if st := m.Unlink(ctx, parent, "l3"); st != 0 {
+		t.Fatalf("unlink d/l3: %s", st)
+	}
+	if st := m.Rmdir(ctx, 1, "d"); st != 0 {
+		t.Fatalf("rmdir d: %s", st)
+	}
+}
+
 func testOpenCache(t *testing.T, m Meta) {
 	ctx := Background
 	var inode Ino
@@ -1292,7 +1367,7 @@ func testReadOnly(t *testing.T, m Meta) {
 	var inode Ino
 	var attr = &Attr{}
 	if st := m.GetAttr(ctx, 1, attr); st != 0 {
-		t.Fatalf("mkdir d: %s", st)
+		t.Fatalf("getattr 1: %s", st)
 	}
 	if st := m.Mkdir(ctx, 1, "d", 0640, 022, 0, &inode, attr); st != syscall.EROFS {
 		t.Fatalf("mkdir d: %s", st)