cdc: adjust sorter options to avoid Seek CPU usage exploding (#11099)

cdc/processor/sourcemanager/sorter/pebble/db.go

@@ -118,7 +118,8 @@ func buildPebbleOption(cfg *config.DBConfig) (opts *pebble.Options) {
 	opts.DisableWAL = false // Delete range requires WAL.
 	opts.MaxOpenFiles = cfg.MaxOpenFiles / cfg.Count
 	opts.MaxConcurrentCompactions = func() int { return 6 }
-	opts.L0CompactionThreshold = cfg.CompactionL0Trigger
+	opts.L0CompactionThreshold = 4 // Default for PebbleDB.
+	opts.L0CompactionFileThreshold = cfg.CompactionL0Trigger
 	opts.L0StopWritesThreshold = cfg.WriteL0PauseTrigger
 	opts.LBaseMaxBytes = 64 << 20 // 64 MB
 	opts.MemTableSize = uint64(cfg.WriterBufferSize)

pkg/cmd/server/server_test.go

@@ -178,7 +178,7 @@ func TestParseCfg(t *testing.T) {
 				WriterBufferSize:    8388608,
 				Compression:         "snappy",
 				WriteL0PauseTrigger: math.MaxInt32,
-				CompactionL0Trigger: 160,
+				CompactionL0Trigger: 16,
 			},
 			// We expect the default configuration here.
 			Messages: &config.MessagesConfig{
@@ -464,7 +464,7 @@ cert-allowed-cn = ["dd","ee"]
 				WriterBufferSize:    8388608,
 				Compression:         "snappy",
 				WriteL0PauseTrigger: math.MaxInt32,
-				CompactionL0Trigger: 160,
+				CompactionL0Trigger: 16,
 			},
 			// We expect the default configuration here.
 			Messages: &config.MessagesConfig{
@@ -529,7 +529,7 @@ unknown3 = 3
 			WriterBufferSize:    8388608,
 			Compression:         "snappy",
 			WriteL0PauseTrigger: math.MaxInt32,
-			CompactionL0Trigger: 160,
+			CompactionL0Trigger: 16,
 		},
 		// We expect the default configuration here.
 		Messages: &config.MessagesConfig{

pkg/config/config_test_data.go

@@ -162,7 +162,7 @@ const (
       "writer-buffer-size": 8388608,
       "compression": "snappy",
       "write-l0-pause-trigger": 2147483647,
-      "compaction-l0-trigger": 160
+      "compaction-l0-trigger": 16
     },
     "messages": {
       "client-max-batch-interval": 10000000,

pkg/config/db.go

@@ -47,7 +47,7 @@ type DBConfig struct {
 	// CompactionL0Trigger defines number of db sst file at level-0 that will
 	// trigger compaction.
 	//
-	// The default value is 160.
+	// The default value is 16, which is based on a performance test on 4K tables.
 	CompactionL0Trigger int `toml:"compaction-l0-trigger" json:"compaction-l0-trigger"`
 }
 

pkg/config/server_config.go

@@ -134,7 +134,7 @@ var defaultServerConfig = &ServerConfig{
 			WriterBufferSize:    8388608,
 			Compression:         "snappy",
 			WriteL0PauseTrigger: math.MaxInt32,
-			CompactionL0Trigger: 160,
+			CompactionL0Trigger: 16, // Based on a performance test on 4K tables.
 		},
 		Messages: defaultMessageConfig.Clone(),
 

pkg/spanz/hash_map.go

@@ -15,8 +15,8 @@
 
 import (
 	"encoding/binary"
-	"hash/fnv"
 
+	"blainsmith.com/go/seahash"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 )
 
@@ -80,10 +80,8 @@
 
 // HashTableSpan hashes the given span to a slot offset.
 func HashTableSpan(span tablepb.Span, slots int) int {
-	h := fnv.New64()
 	b := make([]byte, 8+len(span.StartKey))
 	binary.LittleEndian.PutUint64(b[0:8], uint64(span.TableID))
 	copy(b[8:], span.StartKey)
-	h.Write(b[:])
-	return int(h.Sum64() % uint64(slots))
+	return int(seahash.Sum64(b) % uint64(slots))
 }