pkg/importsdk: fix wildcard generation for subdir csv files (pingcap#67472)

close pingcap#67471

Author: GMHDBJD

pkg/importsdk/pattern.go

@@ -158,18 +158,61 @@ func longestCommonSuffix(strs []string, prefixLen int) string {
 	return suffix
 }
 
-// generatePrefixSuffixPattern returns a wildcard pattern that matches all and only the given paths
-// by finding the longest common prefix and suffix among them, and placing a '*' wildcard in between.
-func generatePrefixSuffixPattern(paths []string) string {
+func generateFlatPrefixSuffixPattern(paths []string) string {
 	if len(paths) == 0 {
 		return ""
 	}
 	if len(paths) == 1 {
 		return paths[0]
 	}
+	allSame := true
+	for _, p := range paths[1:] {
+		if p != paths[0] {
+			allSame = false
+			break
+		}
+	}
+	if allSame {
+		return paths[0]
+	}
 
 	prefix := longestCommonPrefix(paths)
 	suffix := longestCommonSuffix(paths, len(prefix))
 
 	return prefix + "*" + suffix
 }
+
+// generatePrefixSuffixPattern returns a wildcard pattern that matches all and only the given paths.
+// When all paths have the same number of '/'-separated components, it generates the wildcard
+// component by component so every '*' stays within a single path segment, which is required by
+// filepath.Match.
+func generatePrefixSuffixPattern(paths []string) string {
+	if len(paths) <= 1 {
+		return generateFlatPrefixSuffixPattern(paths)
+	}
+
+	componentCount := -1
+	pathComponents := make([][]string, 0, len(paths))
+	for _, p := range paths {
+		components := strings.Split(p, "/")
+		if componentCount == -1 {
+			componentCount = len(components)
+		} else if len(components) != componentCount {
+			return generateFlatPrefixSuffixPattern(paths)
+		}
+		pathComponents = append(pathComponents, components)
+	}
+	if componentCount <= 1 {
+		return generateFlatPrefixSuffixPattern(paths)
+	}
+
+	componentPatterns := make([]string, 0, componentCount)
+	for componentIdx := range componentCount {
+		componentValues := make([]string, 0, len(pathComponents))
+		for _, components := range pathComponents {
+			componentValues = append(componentValues, components[componentIdx])
+		}
+		componentPatterns = append(componentPatterns, generateFlatPrefixSuffixPattern(componentValues))
+	}
+	return strings.Join(componentPatterns, "/")
+}

pkg/importsdk/pattern_test.go

@@ -131,6 +131,16 @@ func TestValidatePattern(t *testing.T) {
 	// If pattern doesn't match our table's file, it's also invalid
 	require.False(t, isValidPattern("*.csv", tableFiles, smallAll))
 
+	subdirTableFiles := map[string]struct{}{
+		"dir/subdir1/a.csv": {},
+		"dir/subdir2/b.csv": {},
+	}
+	subdirAll := map[string]mydump.FileInfo{
+		"dir/subdir1/a.csv": {},
+		"dir/subdir2/b.csv": {},
+	}
+	require.True(t, isValidPattern("dir/subdir*/*.csv", subdirTableFiles, subdirAll))
+
 	// empty pattern => invalid
 	require.False(t, isValidPattern("", tableFiles, smallAll))
 }
@@ -205,4 +215,48 @@ func TestGenerateWildcardPath(t *testing.T) {
 	_, err = generateWildcardPath(files4, allFiles4)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "cannot generate a unique wildcard pattern")
+
+	files5 := []mydump.FileInfo{
+		{
+			TableName: filter.Table{Schema: "db", Name: "tb"},
+			FileMeta: mydump.SourceFileMeta{
+				Path:        "dir/subdir1/a.csv",
+				Type:        mydump.SourceTypeCSV,
+				Compression: mydump.CompressionNone,
+			},
+		},
+		{
+			TableName: filter.Table{Schema: "db", Name: "tb"},
+			FileMeta: mydump.SourceFileMeta{
+				Path:        "dir/subdir1/b.csv",
+				Type:        mydump.SourceTypeCSV,
+				Compression: mydump.CompressionNone,
+			},
+		},
+		{
+			TableName: filter.Table{Schema: "db", Name: "tb"},
+			FileMeta: mydump.SourceFileMeta{
+				Path:        "dir/subdir2/c.csv",
+				Type:        mydump.SourceTypeCSV,
+				Compression: mydump.CompressionNone,
+			},
+		},
+		{
+			TableName: filter.Table{Schema: "db", Name: "tb"},
+			FileMeta: mydump.SourceFileMeta{
+				Path:        "dir/subdir2/d.csv",
+				Type:        mydump.SourceTypeCSV,
+				Compression: mydump.CompressionNone,
+			},
+		},
+	}
+	allFiles5 := map[string]mydump.FileInfo{
+		files5[0].FileMeta.Path: files5[0],
+		files5[1].FileMeta.Path: files5[1],
+		files5[2].FileMeta.Path: files5[2],
+		files5[3].FileMeta.Path: files5[3],
+	}
+	path5, err := generateWildcardPath(files5, allFiles5)
+	require.NoError(t, err)
+	require.Equal(t, "dir/subdir*/*.csv", path5)
 }