eventstore: add config for zstd compression

logservice/eventstore/event_store.go

@@ -236,6 +236,8 @@ type eventStore struct {
 
 	// compressionThreshold is the size in bytes above which a value will be compressed.
 	compressionThreshold int
+	// enableZstdCompression controls whether to enable zstd compression for large values.
+	enableZstdCompression bool
 }
 
 const (
@@ -276,7 +278,8 @@ func New(
 				return decoder
 			},
 		},
-		compressionThreshold: config.GetGlobalServerConfig().Debug.EventStore.CompressionThreshold,
+		compressionThreshold:  config.GetGlobalServerConfig().Debug.EventStore.CompressionThreshold,
+		enableZstdCompression: config.GetGlobalServerConfig().Debug.EventStore.EnableZstdCompression,
 	}
 	store.gcManager = newGCManager(store.dbs, deleteDataRange, compactDataRange)
 
@@ -1310,7 +1313,7 @@ func (e *eventStore) writeEvents(
 			valueBytesBefore := int64(len(rawValue))
 			valueBytesAfter := valueBytesBefore
 			value := rawValue
-			if len(rawValue) > e.compressionThreshold {
+			if e.enableZstdCompression && len(rawValue) > e.compressionThreshold {
 				maxEncodedSize := encoder.MaxEncodedSize(len(rawValue))
 				if cap(dstBuf) < maxEncodedSize {
 					dstBuf = make([]byte, 0, maxEncodedSize)

logservice/eventstore/event_store_test.go

@@ -115,6 +115,17 @@ func setDataSharingForTest(t *testing.T, enable bool) func() {
 	}
 }
 
+func setZstdCompressionForTest(t *testing.T, enable bool) func() {
+	t.Helper()
+	originalCfg := config.GetGlobalServerConfig().Clone()
+	updatedCfg := originalCfg.Clone()
+	updatedCfg.Debug.EventStore.EnableZstdCompression = enable
+	config.StoreGlobalServerConfig(updatedCfg)
+	return func() {
+		config.StoreGlobalServerConfig(originalCfg)
+	}
+}
+
 func TestEventStoreInteractionWithSubClient(t *testing.T) {
 	restoreCfg := setDataSharingForTest(t, true)
 	defer restoreCfg()
@@ -881,6 +892,57 @@ func TestWriteToEventStore(t *testing.T) {
 	require.True(t, foundLarge, "large value entry not found")
 }
 
+func TestWriteToEventStoreZstdCompressionDisabled(t *testing.T) {
+	restoreCfg := setZstdCompressionForTest(t, false)
+	defer restoreCfg()
+
+	dir := t.TempDir()
+	_, storeInt := newEventStoreForTest(dir)
+	store := storeInt.(*eventStore)
+	defer store.Close(context.Background())
+
+	key := []byte("large-key")
+	value := bytes.Repeat([]byte("a"), store.compressionThreshold+1)
+	entry := common.RawKVEntry{
+		OpType:  common.OpTypePut,
+		StartTs: 200,
+		CRTs:    210,
+		Key:     key,
+		Value:   value,
+	}
+	events := []eventWithCallback{{
+		subID:    1,
+		tableID:  1,
+		kvs:      []common.RawKVEntry{entry},
+		callback: func() {},
+	}}
+
+	encoder, err := zstd.NewWriter(nil)
+	require.NoError(t, err)
+	defer encoder.Close()
+
+	var compressionBuf []byte
+	err = store.writeEvents(store.dbs[0], events, encoder, &compressionBuf)
+	require.NoError(t, err)
+
+	iter, err := store.dbs[0].NewIter(&pebble.IterOptions{})
+	require.NoError(t, err)
+	defer iter.Close()
+
+	count := 0
+	for iter.First(); iter.Valid(); iter.Next() {
+		_, compressionType := DecodeKeyMetas(iter.Key())
+		require.Equal(t, CompressionNone, compressionType)
+
+		readEntry := &common.RawKVEntry{}
+		err = readEntry.Decode(iter.Value())
+		require.NoError(t, err)
+		require.Equal(t, value, readEntry.Value)
+		count++
+	}
+	require.Equal(t, 1, count)
+}
+
 func TestEventStoreCompressionAndIterDecodeBufferReuse(t *testing.T) {
 	dir := t.TempDir()
 	_, storeInt := newEventStoreForTest(dir)

pkg/config/debug.go

@@ -49,9 +49,6 @@ func (c *DebugConfig) ValidateAndAdjust() error {
 	if err := c.Scheduler.ValidateAndAdjust(); err != nil {
 		return errors.Trace(err)
 	}
-	if c.EventStore == nil {
-		c.EventStore = NewDefaultEventStoreConfig()
-	}
 
 	return nil
 }
@@ -85,14 +82,18 @@ func NewDefaultPullerConfig() *PullerConfig {
 type EventStoreConfig struct {
 	CompressionThreshold int `toml:"compression-threshold" json:"compression_threshold"`
 
+	// EnableZstdCompression controls whether to enable zstd compression for large values.
+	EnableZstdCompression bool `toml:"enable-zstd-compression" json:"enable_zstd_compression"`
+
 	EnableDataSharing bool `toml:"enable-data-sharing" json:"enable_data_sharing"`
 }
 
 // NewDefaultEventStoreConfig returns the default event store configuration.
 func NewDefaultEventStoreConfig() *EventStoreConfig {
 	return &EventStoreConfig{
-		CompressionThreshold: 16384, // 16KB
-		EnableDataSharing:    false,
+		CompressionThreshold:  16384, // 16KB
+		EnableZstdCompression: true,
+		EnableDataSharing:     false,
 	}
 }