meta: Fix CaseInsensi in MySQL (#591)

* meta/redis: check CaseInsensi in Link

* meta/test: add more CaseInsensi cases

* meta/sql: change mysql edge table to case-sensitive

* meta/sql: change charset and together with collate

* fix unlink/rmdir with case-insensitive
Co-authored-by: NDavies Liu <davies@juicedata.io>

pkg/meta/redis.go

@@ -1746,6 +1746,8 @@ func (r *redisMeta) Link(ctx Context, inode, parent Ino, name string, attr *Attr
 			return err
 		} else if err == nil {
 			return syscall.EEXIST
+		} else if err == redis.Nil && r.conf.CaseInsensi && r.resolveCase(ctx, parent, name) != nil {
+			return syscall.EEXIST
 		}
 
 		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {

pkg/meta/redis_test.go

@@ -184,6 +184,10 @@ func testMetaClient(t *testing.T, m Meta) {
 		t.Fatalf("link f3 -> f: %s", st)
 	}
 	defer m.Unlink(ctx, 1, "f3")
+	if st := m.Link(ctx, inode, 1, "F3", attr); st != 0 { // CaseInsensi = false
+		t.Fatalf("link F3 -> f: %s", st)
+	}
+	_ = m.Unlink(ctx, 1, "F3")
 	if st := m.Link(ctx, parent, 1, "d2", attr); st != syscall.EPERM {
 		t.Fatalf("link d2 -> d: %s", st)
 	}
@@ -545,14 +549,23 @@ func testCaseIncensi(t *testing.T, m Meta) {
 	if st := m.Rename(ctx, 1, "Foo", 1, "bar", &inode, attr); st != 0 {
 		t.Fatalf("rename Foo to bar should be OK, but got %s", st)
 	}
+	if st := m.Create(ctx, 1, "Foo", 0755, 0, &inode, attr); st != 0 {
+		t.Fatalf("create Foo should be OK")
+	}
 	if st := m.Lookup(ctx, 1, "Bar", &inode, attr); st != 0 {
 		t.Fatalf("lookup Bar should be OK")
 	}
+	if st := m.Link(ctx, inode, 1, "foo", attr); st != syscall.EEXIST {
+		t.Fatalf("link should fail with EEXIST")
+	}
 	if st := m.Unlink(ctx, 1, "Bar"); st != 0 {
 		t.Fatalf("unlink Bar should be OK")
 	}
+	if st := m.Unlink(ctx, 1, "foo"); st != 0 {
+		t.Fatalf("unlink foo should be OK")
+	}
 	if st := m.Mkdir(ctx, 1, "Foo", 0755, 0, 0, &inode, attr); st != 0 {
-		t.Fatalf("mkdir Foo should be OK")
+		t.Fatalf("mkdir Foo should be OK, but got %s", st)
 	}
 	if st := m.Rmdir(ctx, 1, "foo"); st != 0 {
 		t.Fatalf("rmdir foo should be OK")

pkg/meta/sql.go

@@ -216,6 +216,11 @@ func (m *dbMeta) Init(format Format, force bool) error {
 	if err := m.engine.Sync2(new(flock), new(plock)); err != nil {
 		logger.Fatalf("create table flock, plock: %s", err)
 	}
+	if m.engine.DriverName() == "mysql" {
+		if _, err := m.engine.Exec("alter table jfs_edge modify name varchar (255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL"); err != nil {
+			logger.Fatalf("alter collate for edge: %s", err)
+		}
+	}
 
 	old, err := m.Load()
 	if err != nil {
@@ -296,6 +301,11 @@ func (m *dbMeta) NewSession() error {
 	if err := m.engine.Sync2(new(session)); err != nil { // old client has no info field
 		return err
 	}
+	if m.engine.DriverName() == "mysql" {
+		if _, err := m.engine.Exec("alter table jfs_edge modify name varchar (255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL"); err != nil {
+			return err
+		}
+	}
 	// update the owner from uint64 to int64
 	if err := m.engine.Sync2(new(flock), new(plock)); err != nil {
 		logger.Fatalf("update table flock, plock: %s", err)
@@ -1120,6 +1130,7 @@ func (m *dbMeta) Unlink(ctx Context, parent Ino, name string) syscall.Errno {
 		if !ok && m.conf.CaseInsensi {
 			if ee := m.resolveCase(ctx, parent, name); ee != nil {
 				ok = true
+				e.Name = string(ee.Name)
 				e.Inode = ee.Inode
 				e.Type = ee.Attr.Typ
 			}
@@ -1154,7 +1165,7 @@ func (m *dbMeta) Unlink(ctx Context, parent Ino, name string) syscall.Errno {
 			opened = m.of.IsOpen(e.Inode)
 		}
 
-		if _, err := s.Delete(&edge{Parent: parent, Name: name}); err != nil {
+		if _, err := s.Delete(&edge{Parent: parent, Name: e.Name}); err != nil {
 			return err
 		}
 		if _, err := s.Delete(&xattr{Inode: e.Inode}); err != nil {
@@ -1246,6 +1257,7 @@ func (m *dbMeta) Rmdir(ctx Context, parent Ino, name string) syscall.Errno {
 			if ee := m.resolveCase(ctx, parent, name); ee != nil {
 				ok = true
 				e.Inode = ee.Inode
+				e.Name = string(ee.Name)
 				e.Type = ee.Attr.Typ
 			}
 		}
@@ -1280,7 +1292,7 @@ func (m *dbMeta) Rmdir(ctx Context, parent Ino, name string) syscall.Errno {
 		pn.Nlink--
 		pn.Mtime = now
 		pn.Ctime = now
-		if _, err := s.Delete(&edge{Parent: parent, Name: name}); err != nil {
+		if _, err := s.Delete(&edge{Parent: parent, Name: e.Name}); err != nil {
 			return err
 		}
 		if _, err := s.Delete(&node{Inode: e.Inode}); err != nil {
@@ -1324,6 +1336,7 @@ func (m *dbMeta) Rename(ctx Context, parentSrc Ino, nameSrc string, parentDst In
 				ok = true
 				se.Inode = e.Inode
 				se.Name = string(e.Name)
+				se.Type = e.Attr.Typ
 			}
 		}
 		if !ok {