diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go
index 5ad27f803c..67e9d46884 100644
--- a/logservice/schemastore/disk_format.go
+++ b/logservice/schemastore/disk_format.go
@@ -15,8 +15,10 @@ package schemastore
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"encoding/json"
+	"io"
 	"math"
 	"strings"
 	"time"
@@ -25,6 +27,8 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+	"github.com/pingcap/ticdc/pkg/encryption"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/filter"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta/model"
@@ -44,14 +48,93 @@ import (
 //     the valid data keys on disk includes snapshot data at `snapshot_ts`
 //     and ddl jobs in the range (snapshot_ts, max_finished_ddl_ts]
 //     and we will pull ddl job from `resolved_ts` at restart if the current gc ts is smaller than resolved_ts.
+//
+// Persisted schema and DDL keys append an 8-byte mask. Bit 0 marks that the
+// value passed through the encryption layer; other bits are reserved. Readers
+// also accept legacy keys without the mask.
 
 const (
 	snapshotSchemaKeyPrefix    = "ss_"
 	snapshotTableKeyPrefix     = "st_"
 	snapshotPartitionKeyPrefix = "sp_"
 	ddlKeyPrefix               = "ds_"
+
+	schemaStoreKeyMaskLen = 8
+)
+
+const (
+	// encryptionLayerKeyMask indicates that the value passed through the encryption layer.
+	// Other mask bits are reserved.
+	encryptionLayerKeyMask uint64 = 1 << iota
 )
 
+func keyWithMask(key []byte, mask uint64) []byte {
+	result := make([]byte, len(key)+schemaStoreKeyMaskLen)
+	copy(result, key)
+	binary.BigEndian.PutUint64(result[len(key):], mask)
+	return result
+}
+
+// keyWithEncryptionLayer marks that the value stored under key passed through
+// the encryption layer. The marker is persisted in the key so readers do not
+// infer the value format from whether an encryption manager is configured.
+func keyWithEncryptionLayer(key []byte) []byte {
+	return keyWithMask(key, encryptionLayerKeyMask)
+}
+
+func keyUsesEncryptionLayer(key []byte) bool {
+	var plainKeyLen int
+	switch {
+	case bytes.HasPrefix(key, []byte(snapshotSchemaKeyPrefix)):
+		plainKeyLen = len(snapshotSchemaKeyPrefix) + 8 + 8
+	case bytes.HasPrefix(key, []byte(snapshotTableKeyPrefix)):
+		plainKeyLen = len(snapshotTableKeyPrefix) + 8 + 8
+	case bytes.HasPrefix(key, []byte(ddlKeyPrefix)):
+		plainKeyLen = len(ddlKeyPrefix) + 8
+	default:
+		return false
+	}
+	return len(key) == plainKeyLen+schemaStoreKeyMaskLen &&
+		binary.BigEndian.Uint64(key[plainKeyLen:])&encryptionLayerKeyMask != 0
+}
+
+func decryptValueIfNeeded(
+	key []byte,
+	value []byte,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
+) ([]byte, error) {
+	if !keyUsesEncryptionLayer(key) {
+		return value, nil
+	}
+	if encMgr == nil {
+		log.Panic("encountered encryption-layer value but no encryption manager is configured",
+			zap.Uint32("keyspaceID", keyspaceID),
+			zap.Binary("key", append([]byte(nil), key...)))
+	}
+	return encMgr.DecryptData(context.Background(), keyspaceID, value)
+}
+
+func getValueWithMaskKey(
+	snap *pebble.Snapshot,
+	plainKey []byte,
+) (value []byte, closer io.Closer, key []byte, err error) {
+	encryptionLayerKey := keyWithEncryptionLayer(plainKey)
+	value, closer, err = snap.Get(encryptionLayerKey)
+	if !errors.Is(err, pebble.ErrNotFound) {
+		return value, closer, encryptionLayerKey, err
+	}
+
+	unencryptedKey := keyWithMask(plainKey, 0)
+	value, closer, err = snap.Get(unencryptedKey)
+	if !errors.Is(err, pebble.ErrNotFound) {
+		return value, closer, unencryptedKey, err
+	}
+
+	value, closer, err = snap.Get(plainKey)
+	return value, closer, plainKey, err
+}
+
 func gcTsKey() []byte {
 	return []byte("gc")
 }
@@ -163,6 +246,11 @@ func writeUpperBoundMeta(db *pebble.DB, upperBound UpperBoundMeta) {
 }
 
 func loadDatabasesInKVSnap(snap *pebble.Snapshot, gcTs uint64) (map[int64]*BasicDatabaseInfo, error) {
+	return loadDatabasesInKVSnapWithEncryption(snap, gcTs, nil, 0)
+}
+
+// loadDatabasesInKVSnapWithEncryption decrypts and loads databases from snapshot if encryption is enabled
+func loadDatabasesInKVSnapWithEncryption(snap *pebble.Snapshot, gcTs uint64, encMgr encryption.EncryptionManager, keyspaceID uint32) (map[int64]*BasicDatabaseInfo, error) {
 	databaseMap := make(map[int64]*BasicDatabaseInfo)
 
 	startKey, err := schemaInfoKey(gcTs, 0)
@@ -182,8 +270,15 @@ func loadDatabasesInKVSnap(snap *pebble.Snapshot, gcTs uint64) (map[int64]*Basic
 	}
 	defer snapIter.Close()
 	for snapIter.First(); snapIter.Valid(); snapIter.Next() {
+		value := snapIter.Value()
+
+		value, err = decryptValueIfNeeded(snapIter.Key(), value, encMgr, keyspaceID)
+		if err != nil {
+			log.Fatal("decrypt db info failed", zap.Error(err))
+		}
+
 		var dbInfo model.DBInfo
-		if err := json.Unmarshal(snapIter.Value(), &dbInfo); err != nil {
+		if err := json.Unmarshal(value, &dbInfo); err != nil {
 			log.Fatal("unmarshal db info failed", zap.Error(err))
 		}
 
@@ -201,6 +296,16 @@ func loadTablesInKVSnap(
 	snap *pebble.Snapshot,
 	gcTs uint64,
 	databaseMap map[int64]*BasicDatabaseInfo,
+) (map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) {
+	return loadTablesInKVSnapWithEncryption(snap, gcTs, databaseMap, nil, 0)
+}
+
+func loadTablesInKVSnapWithEncryption(
+	snap *pebble.Snapshot,
+	gcTs uint64,
+	databaseMap map[int64]*BasicDatabaseInfo,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) (map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) {
 	tablesInKVSnap := make(map[int64]*BasicTableInfo)
 	partitionsInKVSnap := make(map[int64]BasicPartitionInfo)
@@ -222,8 +327,14 @@ func loadTablesInKVSnap(
 	}
 	defer snapIter.Close()
 	for snapIter.First(); snapIter.Valid(); snapIter.Next() {
+		value := snapIter.Value()
+		value, err = decryptValueIfNeeded(snapIter.Key(), value, encMgr, keyspaceID)
+		if err != nil {
+			log.Fatal("decrypt table info failed", zap.Error(err))
+		}
+
 		var table_info_entry PersistedTableInfoEntry
-		if _, err := table_info_entry.UnmarshalMsg(snapIter.Value()); err != nil {
+		if _, err := table_info_entry.UnmarshalMsg(value); err != nil {
 			log.Fatal("unmarshal table info entry failed", zap.Error(err))
 		}
 
@@ -259,6 +370,16 @@ func loadFullTablesInKVSnap(
 	snap *pebble.Snapshot,
 	gcTs uint64,
 	databaseMap map[int64]*BasicDatabaseInfo,
+) (map[int64]*model.TableInfo, map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) {
+	return loadFullTablesInKVSnapWithEncryption(snap, gcTs, databaseMap, nil, 0)
+}
+
+func loadFullTablesInKVSnapWithEncryption(
+	snap *pebble.Snapshot,
+	gcTs uint64,
+	databaseMap map[int64]*BasicDatabaseInfo,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) (map[int64]*model.TableInfo, map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) {
 	tableInfosInKVSnap := make(map[int64]*model.TableInfo)
 	tablesInKVSnap := make(map[int64]*BasicTableInfo)
@@ -281,8 +402,14 @@ func loadFullTablesInKVSnap(
 	}
 	defer snapIter.Close()
 	for snapIter.First(); snapIter.Valid(); snapIter.Next() {
+		value := snapIter.Value()
+		value, err = decryptValueIfNeeded(snapIter.Key(), value, encMgr, keyspaceID)
+		if err != nil {
+			log.Fatal("decrypt table info failed", zap.Error(err))
+		}
+
 		var table_info_entry PersistedTableInfoEntry
-		if _, err := table_info_entry.UnmarshalMsg(snapIter.Value()); err != nil {
+		if _, err := table_info_entry.UnmarshalMsg(value); err != nil {
 			log.Fatal("unmarshal table info entry failed", zap.Error(err))
 		}
 
@@ -323,6 +450,8 @@ func loadAndApplyDDLHistory(
 	databaseMap map[int64]*BasicDatabaseInfo,
 	tableMap map[int64]*BasicTableInfo,
 	partitionMap map[int64]BasicPartitionInfo,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) (map[int64][]uint64, []uint64, error) {
 	tablesDDLHistory := make(map[int64][]uint64)
 	tableTriggerDDLHistory := make([]uint64, 0)
@@ -345,7 +474,15 @@ func loadAndApplyDDLHistory(
 	}
 	defer snapIter.Close()
 	for snapIter.First(); snapIter.Valid(); snapIter.Next() {
-		ddlEvent := unmarshalPersistedDDLEvent(snapIter.Value())
+		ddlValue := snapIter.Value()
+		ddlValue, err = decryptValueIfNeeded(snapIter.Key(), ddlValue, encMgr, keyspaceID)
+		if err != nil {
+			log.Fatal("decrypt ddl job failed when loading ddl history",
+				zap.Uint32("keyspaceID", keyspaceID),
+				zap.Binary("ddlKey", append([]byte(nil), snapIter.Key()...)),
+				zap.Error(err))
+		}
+		ddlEvent := unmarshalPersistedDDLEvent(ddlValue)
 		// Note: no need to skip ddl here
 		// 1. for create table and create tables, we always store the ddl with smaller commit ts, so the ddl won't conflict with the tables in the gc snapshot.
 		// 2. for other ddls to be ignored, they are already filtered before write to disk.
@@ -396,12 +533,22 @@ func tryReadLogicalTableID(snap *pebble.Snapshot, tableID int64, version uint64)
 }
 
 func readTableInfoInKVSnap(snap *pebble.Snapshot, tableID int64, version uint64) *common.TableInfo {
+	return readTableInfoInKVSnapWithEncryption(snap, tableID, version, nil, 0)
+}
+
+func readTableInfoInKVSnapWithEncryption(
+	snap *pebble.Snapshot,
+	tableID int64,
+	version uint64,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
+) *common.TableInfo {
 	readRawTableInfo := func(targetTableID int64) (string, *model.TableInfo) {
-		targetKey, err := tableInfoKey(version, targetTableID)
+		plainKey, err := tableInfoKey(version, targetTableID)
 		if err != nil {
 			log.Fatal("generate table info failed", zap.Error(err))
 		}
-		value, closer, err := snap.Get(targetKey)
+		value, closer, targetKey, err := getValueWithMaskKey(snap, plainKey)
 		if err == pebble.ErrNotFound {
 			return "", nil
 		}
@@ -409,6 +556,10 @@ func readTableInfoInKVSnap(snap *pebble.Snapshot, tableID int64, version uint64)
 			log.Fatal("get table info failed", zap.Error(err))
 		}
 		defer closer.Close()
+		value, err = decryptValueIfNeeded(targetKey, value, encMgr, keyspaceID)
+		if err != nil {
+			log.Fatal("decrypt table info failed", zap.Error(err))
+		}
 
 		var table_info_entry PersistedTableInfoEntry
 		if _, err := table_info_entry.UnmarshalMsg(value); err != nil {
@@ -472,21 +623,41 @@ func unmarshalPersistedDDLEvent(value []byte) PersistedDDLEvent {
 }
 
 func readPersistedDDLEvent(snap *pebble.Snapshot, version uint64) PersistedDDLEvent {
-	ddlKey, err := ddlJobKey(version)
+	return readPersistedDDLEventWithEncryption(snap, version, nil, 0)
+}
+
+// readPersistedDDLEventWithEncryption reads a DDL event and decrypts it when
+// its persisted key says the value passed through the encryption layer.
+func readPersistedDDLEventWithEncryption(snap *pebble.Snapshot, version uint64, encMgr encryption.EncryptionManager, keyspaceID uint32) PersistedDDLEvent {
+	plainKey, err := ddlJobKey(version)
 	if err != nil {
 		log.Fatal("generate ddl job key failed", zap.Error(err))
 	}
-	ddlValue, closer, err := snap.Get(ddlKey)
+	ddlValue, closer, ddlKey, err := getValueWithMaskKey(snap, plainKey)
 	if err != nil {
 		log.Fatal("get ddl job failed",
 			zap.Uint64("version", version),
 			zap.Error(err))
 	}
 	defer closer.Close()
+
+	ddlValue, err = decryptValueIfNeeded(ddlKey, ddlValue, encMgr, keyspaceID)
+	if err != nil {
+		log.Fatal("decrypt ddl event failed",
+			zap.Uint64("version", version),
+			zap.Uint32("keyspaceID", keyspaceID),
+			zap.Error(err))
+	}
+
 	return unmarshalPersistedDDLEvent(ddlValue)
 }
 
 func writePersistedDDLEvent(db *pebble.DB, ddlEvent *PersistedDDLEvent) error {
+	return writePersistedDDLEventWithEncryption(db, ddlEvent, nil, 0)
+}
+
+// writePersistedDDLEventWithEncryption encrypts and writes DDL event if encryption is enabled
+func writePersistedDDLEventWithEncryption(db *pebble.DB, ddlEvent *PersistedDDLEvent, encMgr encryption.EncryptionManager, keyspaceID uint32) error {
 	batch := db.NewBatch()
 	ddlKey, err := ddlJobKey(ddlEvent.FinishedTs)
 	if err != nil {
@@ -515,6 +686,19 @@ func writePersistedDDLEvent(db *pebble.DB, ddlEvent *PersistedDDLEvent) error {
 	if err != nil {
 		return err
 	}
+
+	keyMask := uint64(0)
+	// Encrypt if encryption is enabled
+	if encMgr != nil {
+		encryptedValue, err := encMgr.EncryptData(context.Background(), keyspaceID, ddlValue)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		ddlValue = encryptedValue
+		keyMask |= encryptionLayerKeyMask
+	}
+
+	ddlKey = keyWithMask(ddlKey, keyMask)
 	batch.Set(ddlKey, ddlValue, pebble.NoSync)
 	return batch.Commit(pebble.NoSync)
 }
@@ -526,6 +710,11 @@ func isTableRawKey(key []byte) bool {
 }
 
 func addSchemaInfoToBatch(batch *pebble.Batch, ts uint64, info *model.DBInfo) {
+	addSchemaInfoToBatchWithEncryption(batch, ts, info, nil, 0)
+}
+
+// addSchemaInfoToBatchWithEncryption encrypts and adds schema info to batch if encryption is enabled
+func addSchemaInfoToBatchWithEncryption(batch *pebble.Batch, ts uint64, info *model.DBInfo, encMgr encryption.EncryptionManager, keyspaceID uint32) {
 	schemaKey, err := schemaInfoKey(ts, info.ID)
 	if err != nil {
 		log.Fatal("generate schema key failed", zap.Error(err))
@@ -534,6 +723,19 @@ func addSchemaInfoToBatch(batch *pebble.Batch, ts uint64, info *model.DBInfo) {
 	if err != nil {
 		log.Fatal("marshal schema info failed", zap.Error(err))
 	}
+
+	keyMask := uint64(0)
+	// Encrypt if encryption is enabled
+	if encMgr != nil {
+		encryptedValue, err := encMgr.EncryptData(context.Background(), keyspaceID, schemaValue)
+		if err != nil {
+			log.Fatal("encrypt schema info failed", zap.Error(err))
+		}
+		schemaValue = encryptedValue
+		keyMask |= encryptionLayerKeyMask
+	}
+
+	schemaKey = keyWithMask(schemaKey, keyMask)
 	batch.Set(schemaKey, schemaValue, pebble.NoSync)
 }
 
@@ -542,6 +744,18 @@ func addTableInfoToBatch(
 	ts uint64,
 	dbInfo *model.DBInfo,
 	tableInfoValue []byte,
+) (int64, string, []int64) {
+	return addTableInfoToBatchWithEncryption(batch, ts, dbInfo, tableInfoValue, nil, 0)
+}
+
+// addTableInfoToBatchWithEncryption encrypts and adds table info to batch if encryption is enabled
+func addTableInfoToBatchWithEncryption(
+	batch *pebble.Batch,
+	ts uint64,
+	dbInfo *model.DBInfo,
+	tableInfoValue []byte,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) (int64, string, []int64) {
 	tableInfo := model.TableInfo{}
 	if err := json.Unmarshal(tableInfoValue, &tableInfo); err != nil {
@@ -561,6 +775,19 @@ func addTableInfoToBatch(
 	if err != nil {
 		log.Fatal("marshal table info entry failed", zap.Error(err))
 	}
+
+	keyMask := uint64(0)
+	// Encrypt if encryption is enabled
+	if encMgr != nil {
+		encryptedValue, err := encMgr.EncryptData(context.Background(), keyspaceID, tableInfoEntryValue)
+		if err != nil {
+			log.Fatal("encrypt table info entry failed", zap.Error(err))
+		}
+		tableInfoEntryValue = encryptedValue
+		keyMask |= encryptionLayerKeyMask
+	}
+
+	tableKey = keyWithMask(tableKey, keyMask)
 	batch.Set(tableKey, tableInfoEntryValue, pebble.NoSync)
 
 	// write partition info to batch if the table is a partition table
@@ -589,6 +816,17 @@ func persistSchemaSnapshot(
 	tiStore kv.Storage,
 	snapTs uint64,
 	collectMetaInfo bool,
+) (map[int64]*BasicDatabaseInfo, map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) {
+	return persistSchemaSnapshotWithEncryption(db, tiStore, snapTs, collectMetaInfo, nil, 0)
+}
+
+func persistSchemaSnapshotWithEncryption(
+	db *pebble.DB,
+	tiStore kv.Storage,
+	snapTs uint64,
+	collectMetaInfo bool,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) (map[int64]*BasicDatabaseInfo, map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) {
 	for {
 		meta := getSnapshotMeta(tiStore, snapTs)
@@ -617,7 +855,7 @@ func persistSchemaSnapshot(
 				continue
 			}
 			batch := db.NewBatch()
-			addSchemaInfoToBatch(batch, snapTs, dbInfo)
+			addSchemaInfoToBatchWithEncryption(batch, snapTs, dbInfo, encMgr, keyspaceID)
 			for {
 				rawTables, err := meta.GetMetasByDBID(dbInfo.ID)
 				if err == nil {
@@ -629,7 +867,8 @@ func persistSchemaSnapshot(
 						if !isTableRawKey(rawTable.Field) {
 							continue
 						}
-						tableID, tableName, partitionIDs := addTableInfoToBatch(batch, snapTs, dbInfo, rawTable.Value)
+						tableID, tableName, partitionIDs := addTableInfoToBatchWithEncryption(
+							batch, snapTs, dbInfo, rawTable.Value, encMgr, keyspaceID)
 						if collectMetaInfo {
 							tableMap[tableID] = &BasicTableInfo{
 								SchemaID: dbInfo.ID,
@@ -738,14 +977,17 @@ func loadAllPhysicalTablesAtTs(
 	gcTs uint64,
 	snapVersion uint64,
 	tableFilter filter.Filter,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) ([]commonEvent.Table, error) {
 	// TODO: respect tableFilter(filter table in kv snap is easy, filter ddl jobs need more attention)
-	databaseMap, err := loadDatabasesInKVSnap(storageSnap, gcTs)
+	databaseMap, err := loadDatabasesInKVSnapWithEncryption(storageSnap, gcTs, encMgr, keyspaceID)
 	if err != nil {
 		return nil, err
 	}
 
-	tableInfoMap, tableMap, partitionMap, err := loadFullTablesInKVSnap(storageSnap, gcTs, databaseMap)
+	tableInfoMap, tableMap, partitionMap, err := loadFullTablesInKVSnapWithEncryption(
+		storageSnap, gcTs, databaseMap, encMgr, keyspaceID)
 	if err != nil {
 		return nil, err
 	}
@@ -772,7 +1014,15 @@ func loadAllPhysicalTablesAtTs(
 	}
 	defer snapIter.Close()
 	for snapIter.First(); snapIter.Valid(); snapIter.Next() {
-		ddlEvent := unmarshalPersistedDDLEvent(snapIter.Value())
+		ddlValue := snapIter.Value()
+		ddlValue, err = decryptValueIfNeeded(snapIter.Key(), ddlValue, encMgr, keyspaceID)
+		if err != nil {
+			log.Fatal("decrypt ddl job failed when loading all physical tables",
+				zap.Uint32("keyspaceID", keyspaceID),
+				zap.Binary("ddlKey", append([]byte(nil), snapIter.Key()...)),
+				zap.Error(err))
+		}
+		ddlEvent := unmarshalPersistedDDLEvent(ddlValue)
 		handler, ok := allDDLHandlers[model.ActionType(ddlEvent.Type)]
 		if !ok {
 			log.Panic("unknown ddl type", zap.Any("ddlType", ddlEvent.Type), zap.String("query", ddlEvent.Query))
diff --git a/logservice/schemastore/disk_format_test.go b/logservice/schemastore/disk_format_test.go
index d2a53d87ba..1a6412ba10 100644
--- a/logservice/schemastore/disk_format_test.go
+++ b/logservice/schemastore/disk_format_test.go
@@ -14,14 +14,41 @@
 package schemastore
 
 import (
+	"bytes"
 	"context"
+	"encoding/binary"
+	"encoding/json"
+	"path/filepath"
 	"testing"
 
 	"github.com/cockroachdb/pebble"
 	"github.com/pingcap/tidb/pkg/kv"
+	"github.com/pingcap/tidb/pkg/meta"
+	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/store/mockstore"
 	"github.com/stretchr/testify/require"
 )
 
+type countingEncryptionManagerForTest struct {
+	encryptCalls int
+	decryptCalls int
+}
+
+func (m *countingEncryptionManagerForTest) EncryptData(
+	ctx context.Context, keyspaceID uint32, data []byte,
+) ([]byte, error) {
+	m.encryptCalls++
+	return data, nil
+}
+
+func (m *countingEncryptionManagerForTest) DecryptData(
+	ctx context.Context, keyspaceID uint32, data []byte,
+) ([]byte, error) {
+	m.decryptCalls++
+	return data, nil
+}
+
 func TestPersistSchemaSnapshotReturnsWhenListDatabasesSnapshotLost(t *testing.T) {
 	db, err := pebble.Open(t.TempDir(), &pebble.Options{})
 	require.NoError(t, err)
@@ -32,6 +59,228 @@ func TestPersistSchemaSnapshotReturnsWhenListDatabasesSnapshotLost(t *testing.T)
 	require.ErrorIs(t, err, snapshotLostErr)
 }
 
+func TestSchemaStoreDoesNotDecryptLegacyValuesWhenManagerIsConfigured(t *testing.T) {
+	db, err := pebble.Open(t.TempDir(), &pebble.Options{})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
+	const snapshotTs = uint64(100)
+	dbInfo := &model.DBInfo{ID: 100, Name: ast.NewCIStr("test")}
+	tableInfo := newEligibleTableInfoForTest(200, "t1")
+	tableInfoValue, err := json.Marshal(tableInfo)
+	require.NoError(t, err)
+	schemaValue, err := json.Marshal(dbInfo)
+	require.NoError(t, err)
+	schemaKey, err := schemaInfoKey(snapshotTs, dbInfo.ID)
+	require.NoError(t, err)
+	tableEntryValue, err := (&PersistedTableInfoEntry{
+		SchemaID:       dbInfo.ID,
+		SchemaName:     dbInfo.Name.O,
+		TableInfoValue: tableInfoValue,
+	}).MarshalMsg(nil)
+	require.NoError(t, err)
+	tableKey, err := tableInfoKey(snapshotTs, tableInfo.ID)
+	require.NoError(t, err)
+
+	batch := db.NewBatch()
+	require.NoError(t, batch.Set(schemaKey, schemaValue, pebble.NoSync))
+	require.NoError(t, batch.Set(tableKey, tableEntryValue, pebble.NoSync))
+	require.NoError(t, batch.Commit(pebble.NoSync))
+
+	manager := &countingEncryptionManagerForTest{}
+	snapshot := db.NewSnapshot()
+	defer snapshot.Close()
+
+	databaseMap, err := loadDatabasesInKVSnapWithEncryption(snapshot, snapshotTs, manager, 42)
+	require.NoError(t, err)
+	_, _, err = loadTablesInKVSnapWithEncryption(snapshot, snapshotTs, databaseMap, manager, 42)
+	require.NoError(t, err)
+	require.Zero(t, manager.decryptCalls)
+}
+
+func TestSchemaStoreDoesNotDecryptLegacyDDLWhenManagerIsConfigured(t *testing.T) {
+	db, err := pebble.Open(t.TempDir(), &pebble.Options{})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
+	ddlEvent := &PersistedDDLEvent{
+		FinishedTs: 100,
+		TableInfo:  newEligibleTableInfoForTest(200, "t1"),
+	}
+	require.NoError(t, writePersistedDDLEvent(db, ddlEvent))
+	plainKey, err := ddlJobKey(ddlEvent.FinishedTs)
+	require.NoError(t, err)
+	maskedKey := keyWithMask(plainKey, 0)
+	value, closer, err := db.Get(maskedKey)
+	require.NoError(t, err)
+	legacyValue := append([]byte(nil), value...)
+	require.NoError(t, closer.Close())
+	batch := db.NewBatch()
+	require.NoError(t, batch.Delete(maskedKey, pebble.NoSync))
+	require.NoError(t, batch.Set(plainKey, legacyValue, pebble.NoSync))
+	require.NoError(t, batch.Commit(pebble.NoSync))
+
+	manager := &countingEncryptionManagerForTest{}
+	snapshot := db.NewSnapshot()
+	defer snapshot.Close()
+	readPersistedDDLEventWithEncryption(snapshot, ddlEvent.FinishedTs, manager, 42)
+	require.Zero(t, manager.decryptCalls)
+}
+
+func TestEncryptionLayerKeyUsesEightByteMask(t *testing.T) {
+	plainKey, err := ddlJobKey(100)
+	require.NoError(t, err)
+
+	unencryptedKey := keyWithMask(plainKey, 0)
+	require.Len(t, unencryptedKey, len(plainKey)+8)
+	require.Zero(t, binary.BigEndian.Uint64(unencryptedKey[len(plainKey):]))
+	require.False(t, keyUsesEncryptionLayer(unencryptedKey))
+
+	encryptionLayerKey := keyWithEncryptionLayer(plainKey)
+	require.Len(t, encryptionLayerKey, len(plainKey)+8)
+	require.Equal(t, uint64(1), binary.BigEndian.Uint64(encryptionLayerKey[len(plainKey):]))
+	require.True(t, keyUsesEncryptionLayer(encryptionLayerKey))
+
+	binary.BigEndian.PutUint64(encryptionLayerKey[len(plainKey):], uint64(1)<<1)
+	require.False(t, keyUsesEncryptionLayer(encryptionLayerKey))
+	binary.BigEndian.PutUint64(encryptionLayerKey[len(plainKey):], uint64(1)|(uint64(1)<<1))
+	require.True(t, keyUsesEncryptionLayer(encryptionLayerKey))
+}
+
+func TestPersistedDDLEventUsesZeroMaskWithoutEncryption(t *testing.T) {
+	db, err := pebble.Open(t.TempDir(), &pebble.Options{})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
+	ddlEvent := &PersistedDDLEvent{
+		FinishedTs: 100,
+		TableInfo:  newEligibleTableInfoForTest(200, "t1"),
+	}
+	require.NoError(t, writePersistedDDLEvent(db, ddlEvent))
+
+	plainKey, err := ddlJobKey(ddlEvent.FinishedTs)
+	require.NoError(t, err)
+	value, closer, err := db.Get(keyWithMask(plainKey, 0))
+	require.NoError(t, err)
+	require.NotEmpty(t, value)
+	require.NoError(t, closer.Close())
+	_, plainCloser, err := db.Get(plainKey)
+	require.ErrorIs(t, err, pebble.ErrNotFound)
+	require.Nil(t, plainCloser)
+}
+
+func TestPersistedDDLEventUsesEncryptionLayerKey(t *testing.T) {
+	db, err := pebble.Open(t.TempDir(), &pebble.Options{})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
+	ddlEvent := &PersistedDDLEvent{
+		FinishedTs: 100,
+		TableInfo:  newEligibleTableInfoForTest(200, "t1"),
+	}
+	manager := &prefixEncryptionManagerForTest{}
+	require.NoError(t, writePersistedDDLEventWithEncryption(db, ddlEvent, manager, 42))
+
+	plainKey, err := ddlJobKey(ddlEvent.FinishedTs)
+	require.NoError(t, err)
+	_, plainCloser, err := db.Get(plainKey)
+	require.ErrorIs(t, err, pebble.ErrNotFound)
+	require.Nil(t, plainCloser)
+
+	encryptionLayerKey := keyWithEncryptionLayer(plainKey)
+	require.True(t, keyUsesEncryptionLayer(encryptionLayerKey))
+	value, closer, err := db.Get(encryptionLayerKey)
+	require.NoError(t, err)
+	require.True(t, bytes.HasPrefix(value, encryptedPrefixForTest))
+	require.NoError(t, closer.Close())
+
+	snapshot := db.NewSnapshot()
+	defer snapshot.Close()
+	readEvent := readPersistedDDLEventWithEncryption(snapshot, ddlEvent.FinishedTs, manager, 42)
+	require.Equal(t, ddlEvent.FinishedTs, readEvent.FinishedTs)
+}
+
+func TestPersistSchemaSnapshotEncryptionRoundTrip(t *testing.T) {
+	tikvStore, err := mockstore.NewMockStore()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, tikvStore.Close())
+	}()
+
+	dbInfo := &model.DBInfo{ID: 100, Name: ast.NewCIStr("test")}
+	tableInfo := newEligibleTableInfoForTest(200, "t1")
+	txn, err := tikvStore.Begin()
+	require.NoError(t, err)
+	metaMutator := meta.NewMutator(txn)
+	require.NoError(t, metaMutator.CreateDatabase(dbInfo))
+	require.NoError(t, metaMutator.CreateTableOrView(dbInfo.ID, tableInfo))
+	require.NoError(t, txn.Commit(context.Background()))
+
+	version, err := tikvStore.CurrentVersion(kv.GlobalTxnScope)
+	require.NoError(t, err)
+	encMgr := &prefixEncryptionManagerForTest{}
+	storage := &persistentStorage{
+		keyspaceID:        42,
+		kvStorage:         tikvStore,
+		encryptionManager: encMgr,
+	}
+	storage.initializeFromKVStorage(filepath.Join(t.TempDir(), "schema-store"), version.Ver)
+	defer func() {
+		require.NoError(t, storage.db.Close())
+	}()
+
+	schemaKey, err := schemaInfoKey(version.Ver, dbInfo.ID)
+	require.NoError(t, err)
+	schemaKey = keyWithEncryptionLayer(schemaKey)
+	require.True(t, keyUsesEncryptionLayer(schemaKey))
+	schemaValue, schemaCloser, err := storage.db.Get(schemaKey)
+	require.NoError(t, err)
+	schemaEncrypted := bytes.HasPrefix(schemaValue, encryptedPrefixForTest)
+	require.NoError(t, schemaCloser.Close())
+	require.True(t, schemaEncrypted)
+
+	tableKey, err := tableInfoKey(version.Ver, tableInfo.ID)
+	require.NoError(t, err)
+	tableKey = keyWithEncryptionLayer(tableKey)
+	require.True(t, keyUsesEncryptionLayer(tableKey))
+	tableValue, tableCloser, err := storage.db.Get(tableKey)
+	require.NoError(t, err)
+	tableEncrypted := bytes.HasPrefix(tableValue, encryptedPrefixForTest)
+	require.NoError(t, tableCloser.Close())
+	require.True(t, tableEncrypted)
+
+	storage.databaseMap = make(map[int64]*BasicDatabaseInfo)
+	storage.tableMap = make(map[int64]*BasicTableInfo)
+	storage.partitionMap = make(map[int64]BasicPartitionInfo)
+	storage.tablesDDLHistory = make(map[int64][]uint64)
+	storage.tableTriggerDDLHistory = nil
+	storage.initializeFromDisk()
+	require.Contains(t, storage.databaseMap, dbInfo.ID)
+	require.Contains(t, storage.tableMap, tableInfo.ID)
+	storageSnapshot := storage.db.NewSnapshot()
+	storedTableInfo := readTableInfoInKVSnapWithEncryption(
+		storageSnapshot, tableInfo.ID, version.Ver, encMgr, storage.keyspaceID,
+	)
+	require.NoError(t, storageSnapshot.Close())
+	require.NotNil(t, storedTableInfo)
+	require.Equal(t, tableInfo.ID, storedTableInfo.TableName.TableID)
+
+	tables, err := storage.getAllPhysicalTables(version.Ver, nil)
+	require.NoError(t, err)
+	require.Len(t, tables, 1)
+	require.Equal(t, int64(200), tables[0].TableID)
+	require.Equal(t, "test", tables[0].SchemaName)
+	require.Equal(t, "t1", tables[0].TableName)
+}
+
 type snapshotLostByGCError struct{}
 
 func (snapshotLostByGCError) Error() string {
diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go
index 6d94e1e2f3..2dcf8b00aa 100644
--- a/logservice/schemastore/persist_storage.go
+++ b/logservice/schemastore/persist_storage.go
@@ -27,8 +27,10 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
+	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/encryption"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/filter"
 	"github.com/pingcap/ticdc/pkg/txnutil/gc"
@@ -99,6 +101,9 @@ type persistentStorage struct {
 
 	// tableID -> total registered count
 	tableRegisteredCount map[int64]int
+
+	// encryptionManager for encrypting/decrypting data (optional)
+	encryptionManager encryption.EncryptionManager
 }
 
 func exists(path string) bool {
@@ -144,6 +149,9 @@ func newPersistentStorage(
 	storage kv.Storage,
 	gcSafePoint uint64,
 ) (*persistentStorage, error) {
+	// Try to get encryption manager from appcontext (optional)
+	encMgr, _ := appcontext.TryGetService[encryption.EncryptionManager](appcontext.EncryptionManager)
+
 	dataStorage := &persistentStorage{
 		rootDir:                root,
 		keyspaceID:             keyspaceID,
@@ -156,6 +164,7 @@ func newPersistentStorage(
 		tableTriggerDDLHistory: make([]uint64, 0),
 		tableInfoStoreMap:      make(map[int64]*versionedTableInfoStore),
 		tableRegisteredCount:   make(map[int64]int),
+		encryptionManager:      encMgr,
 	}
 	dataStorage.ctx, dataStorage.cancel = context.WithCancel(ctx)
 	err := dataStorage.initialize(gcSafePoint)
@@ -225,7 +234,8 @@ func (p *persistentStorage) initializeFromKVStorage(dbPath string, gcTs uint64)
 		zap.Uint64("snapTs", gcTs))
 
 	var err error
-	if p.databaseMap, p.tableMap, p.partitionMap, err = persistSchemaSnapshot(p.db, p.kvStorage, gcTs, true); err != nil {
+	if p.databaseMap, p.tableMap, p.partitionMap, err = persistSchemaSnapshotWithEncryption(
+		p.db, p.kvStorage, gcTs, true, p.encryptionManager, p.keyspaceID); err != nil {
 		// TODO: retry
 		log.Fatal("fail to initialize from kv snapshot", zap.Error(err))
 	}
@@ -250,11 +260,13 @@ func (p *persistentStorage) initializeFromDisk() {
 	defer storageSnap.Close()
 
 	var err error
-	if p.databaseMap, err = loadDatabasesInKVSnap(storageSnap, p.gcTs); err != nil {
+	if p.databaseMap, err = loadDatabasesInKVSnapWithEncryption(
+		storageSnap, p.gcTs, p.encryptionManager, p.keyspaceID); err != nil {
 		log.Fatal("load database info from disk failed")
 	}
 
-	if p.tableMap, p.partitionMap, err = loadTablesInKVSnap(storageSnap, p.gcTs, p.databaseMap); err != nil {
+	if p.tableMap, p.partitionMap, err = loadTablesInKVSnapWithEncryption(
+		storageSnap, p.gcTs, p.databaseMap, p.encryptionManager, p.keyspaceID); err != nil {
 		log.Fatal("load tables in kv snapshot failed")
 	}
 
@@ -264,7 +276,9 @@ func (p *persistentStorage) initializeFromDisk() {
 		p.upperBound.FinishedDDLTs,
 		p.databaseMap,
 		p.tableMap,
-		p.partitionMap); err != nil {
+		p.partitionMap,
+		p.encryptionManager,
+		p.keyspaceID); err != nil {
 		log.Fatal("fail to initialize from disk")
 	}
 }
@@ -305,7 +319,7 @@ func (p *persistentStorage) getAllPhysicalTables(snapTs uint64, tableFilter filt
 		log.Debug("getAllPhysicalTables finish",
 			zap.Any("duration(s)", time.Since(start).Seconds()))
 	}()
-	return loadAllPhysicalTablesAtTs(storageSnap, gcTs, snapTs, tableFilter)
+	return loadAllPhysicalTablesAtTs(storageSnap, gcTs, snapTs, tableFilter, p.encryptionManager, p.keyspaceID)
 }
 
 // only return when table info is initialized
@@ -431,7 +445,7 @@ func (p *persistentStorage) fetchTableDDLEvents(dispatcherID common.DispatcherID
 	// TODO: if the first event is a create table ddl, return error?
 	events := make([]commonEvent.DDLEvent, 0, len(allTargetTs))
 	for _, ts := range allTargetTs {
-		rawEvent := readPersistedDDLEvent(storageSnap, ts)
+		rawEvent := readPersistedDDLEventWithEncryption(storageSnap, ts, p.encryptionManager, p.keyspaceID)
 		ddlEvent, ok, err := buildDDLEvent(&rawEvent, tableFilter, tableID)
 		if err != nil {
 			return nil, errors.Trace(err)
@@ -501,7 +515,7 @@ func (p *persistentStorage) fetchTableTriggerDDLEvents(tableFilter filter.Filter
 		}
 		p.mu.RUnlock()
 		for _, ts := range allTargetTs {
-			rawEvent := readPersistedDDLEvent(storageSnap, ts)
+			rawEvent := readPersistedDDLEventWithEncryption(storageSnap, ts, p.encryptionManager, p.keyspaceID)
 			// the tableID of buildDDLEvent is not used in this function, set it to 0
 			ddlEvent, ok, err := buildDDLEvent(&rawEvent, tableFilter, 0)
 			if err != nil {
@@ -531,12 +545,14 @@ func (p *persistentStorage) buildVersionedTableInfoStore(store *versionedTableIn
 	allDDLFinishedTs = append(allDDLFinishedTs, p.tablesDDLHistory[tableID]...)
 	p.mu.RUnlock()
 
-	if err := addTableInfoFromKVSnap(store, kvSnapVersion, storageSnap); err != nil {
+	if err := addTableInfoFromKVSnap(
+		store, kvSnapVersion, storageSnap, p.encryptionManager, p.keyspaceID,
+	); err != nil {
 		return err
 	}
 
 	for _, version := range allDDLFinishedTs {
-		ddlEvent := readPersistedDDLEvent(storageSnap, version)
+		ddlEvent := readPersistedDDLEventWithEncryption(storageSnap, version, p.encryptionManager, p.keyspaceID)
 		store.applyDDLFromPersistStorage(&ddlEvent)
 	}
 	store.setTableInfoInitialized()
@@ -547,8 +563,12 @@ func addTableInfoFromKVSnap(
 	store *versionedTableInfoStore,
 	kvSnapVersion uint64,
 	snap *pebble.Snapshot,
+	encMgr encryption.EncryptionManager,
+	keyspaceID uint32,
 ) error {
-	tableInfo := readTableInfoInKVSnap(snap, store.getTableID(), kvSnapVersion)
+	tableInfo := readTableInfoInKVSnapWithEncryption(
+		snap, store.getTableID(), kvSnapVersion, encMgr, keyspaceID,
+	)
 	if tableInfo != nil {
 		store.addInitialTableInfo(tableInfo, kvSnapVersion)
 	}
@@ -596,7 +616,8 @@ func (p *persistentStorage) doGc(gcTs uint64) {
 	}
 
 	start := time.Now()
-	_, _, _, err := persistSchemaSnapshot(p.db, p.kvStorage, gcTs, false)
+	_, _, _, err := persistSchemaSnapshotWithEncryption(
+		p.db, p.kvStorage, gcTs, false, p.encryptionManager, p.keyspaceID)
 	if err != nil {
 		log.Warn("fail to write kv snapshot during gc",
 			zap.Uint64("gcTs", gcTs), zap.Error(err))
@@ -735,7 +756,10 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error {
 
 	// Note: need write ddl event to disk before update ddl history,
 	// because other goroutines may read ddl events from disk according to ddl history
-	writePersistedDDLEvent(p.db, &ddlEvent)
+	err := writePersistedDDLEventWithEncryption(p.db, &ddlEvent, p.encryptionManager, p.keyspaceID)
+	if err != nil {
+		return errors.Trace(err)
+	}
 
 	p.mu.Lock()
 	defer p.mu.Unlock()
diff --git a/logservice/schemastore/schema_store.go b/logservice/schemastore/schema_store.go
index 13d10cff64..e25cd78df0 100644
--- a/logservice/schemastore/schema_store.go
+++ b/logservice/schemastore/schema_store.go
@@ -116,7 +116,18 @@ func (s *keyspaceSchemaStore) tryUpdateResolvedTs() {
 		return
 	}
 	resolvedEvents := s.unsortedCache.fetchSortedDDLEventBeforeTS(pendingTs)
-	for _, event := range resolvedEvents {
+	newResolvedTs := s.resolvedTs.Load()
+	advanceResolvedTsAfterGroup := func(idx int) {
+		commitTs := resolvedEvents[idx].CommitTs
+		if idx+1 < len(resolvedEvents) && resolvedEvents[idx+1].CommitTs == commitTs {
+			return
+		}
+		if commitTs > newResolvedTs {
+			newResolvedTs = commitTs
+		}
+	}
+	applyFailed := false
+	for idx, event := range resolvedEvents {
 		if event.Job.BinlogInfo.SchemaVersion == 0 /* means the ddl is ignored in upstream */ {
 			log.Info("skip ddl job with empty SchemaVersion",
 				zap.Any("type", event.Job.Type),
@@ -126,6 +137,7 @@ func (s *keyspaceSchemaStore) tryUpdateResolvedTs() {
 				zap.Uint64("jobCommitTs", event.CommitTs),
 				zap.Any("storeSchemaVersion", s.schemaVersion),
 				zap.Uint64("storeFinishedDDLTS", s.finishedDDLTs))
+			advanceResolvedTsAfterGroup(idx)
 			continue
 		}
 		if event.Job.BinlogInfo.FinishedTS <= s.finishedDDLTs {
@@ -137,6 +149,7 @@ func (s *keyspaceSchemaStore) tryUpdateResolvedTs() {
 				zap.Uint64("jobCommitTs", event.CommitTs),
 				zap.Any("storeSchemaVersion", s.schemaVersion),
 				zap.Uint64("storeFinishedDDLTS", s.finishedDDLTs))
+			advanceResolvedTsAfterGroup(idx)
 			continue
 		}
 		log.Info("handle a ddl job",
@@ -151,20 +164,46 @@ func (s *keyspaceSchemaStore) tryUpdateResolvedTs() {
 			zap.Any("storeSchemaVersion", s.schemaVersion),
 			zap.Uint64("storeFinishedDDLTS", s.finishedDDLTs))
 
-		// need to update the following two members for every event to filter out later duplicate events
+		err := s.dataStorage.handleDDLJob(event.Job)
+		if err != nil {
+			applyFailed = true
+			// Keep failed and later events for retry. They are removed from cache when fetched.
+			for _, pendingEvent := range resolvedEvents[idx:] {
+				s.unsortedCache.addDDLEvent(pendingEvent)
+			}
+			log.Error("handle ddl job failed, retry later",
+				zap.Int64("schemaID", event.Job.SchemaID),
+				zap.String("schemaName", event.Job.SchemaName),
+				zap.Int64("tableID", event.Job.TableID),
+				zap.String("tableName", event.Job.TableName),
+				zap.Any("type", event.Job.Type),
+				zap.String("DDL", event.Job.Query),
+				zap.Int64("jobSchemaVersion", event.Job.BinlogInfo.SchemaVersion),
+				zap.Uint64("jobFinishTs", event.Job.BinlogInfo.FinishedTS),
+				zap.Uint64("jobCommitTs", event.CommitTs),
+				zap.Any("storeSchemaVersion", s.schemaVersion),
+				zap.Uint64("storeFinishedDDLTS", s.finishedDDLTs),
+				zap.Error(err))
+			break
+		}
+
+		// Update dedup watermark only after DDL is persisted successfully.
 		s.schemaVersion = event.Job.BinlogInfo.SchemaVersion
 		s.finishedDDLTs = event.Job.BinlogInfo.FinishedTS
-
-		s.dataStorage.handleDDLJob(event.Job)
+		advanceResolvedTsAfterGroup(idx)
 	}
-	// When register a new table, it will load all ddl jobs from disk for the table,
-	// so we can only update resolved ts after all ddl jobs are written to disk
-	// Can we optimize it to update resolved ts more eagerly?
-	s.resolvedTs.Store(pendingTs)
+	if !applyFailed {
+		// When register a new table, it will load all ddl jobs from disk for the table,
+		// so we can only update resolved ts after all ddl jobs are written to disk
+		// Can we optimize it to update resolved ts more eagerly?
+		newResolvedTs = pendingTs
+	}
+
+	s.resolvedTs.Store(newResolvedTs)
 	s.dataStorage.updateUpperBound(UpperBoundMeta{
 		FinishedDDLTs: s.finishedDDLTs,
 		SchemaVersion: s.schemaVersion,
-		ResolvedTs:    pendingTs,
+		ResolvedTs:    newResolvedTs,
 	})
 }
 
diff --git a/logservice/schemastore/schema_store_test.go b/logservice/schemastore/schema_store_test.go
index 6168bef86f..1c30d52f95 100644
--- a/logservice/schemastore/schema_store_test.go
+++ b/logservice/schemastore/schema_store_test.go
@@ -14,6 +14,7 @@
 package schemastore
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -26,6 +27,47 @@ import (
 	"go.uber.org/zap"
 )
 
+type flakyEncryptionManagerForTest struct {
+	failTimes  int
+	failOnCall int
+	calls      int
+}
+
+func (m *flakyEncryptionManagerForTest) EncryptData(ctx context.Context, keyspaceID uint32, data []byte) ([]byte, error) {
+	m.calls++
+	if m.calls <= m.failTimes || m.calls == m.failOnCall {
+		return nil, cerror.New("inject encryption failure")
+	}
+	return data, nil
+}
+
+func (m *flakyEncryptionManagerForTest) DecryptData(ctx context.Context, keyspaceID uint32, encryptedData []byte) ([]byte, error) {
+	return encryptedData, nil
+}
+
+type prefixEncryptionManagerForTest struct{}
+
+var encryptedPrefixForTest = []byte("enc:")
+
+func (m *prefixEncryptionManagerForTest) EncryptData(ctx context.Context, keyspaceID uint32, data []byte) ([]byte, error) {
+	encrypted := make([]byte, 0, len(encryptedPrefixForTest)+len(data))
+	encrypted = append(encrypted, encryptedPrefixForTest...)
+	encrypted = append(encrypted, data...)
+	return encrypted, nil
+}
+
+func (m *prefixEncryptionManagerForTest) DecryptData(ctx context.Context, keyspaceID uint32, encryptedData []byte) ([]byte, error) {
+	if len(encryptedData) < len(encryptedPrefixForTest) {
+		return nil, cerror.New("encrypted data too short")
+	}
+	if string(encryptedData[:len(encryptedPrefixForTest)]) != string(encryptedPrefixForTest) {
+		return nil, cerror.New("invalid encrypted prefix")
+	}
+	plaintext := make([]byte, len(encryptedData)-len(encryptedPrefixForTest))
+	copy(plaintext, encryptedData[len(encryptedPrefixForTest):])
+	return plaintext, nil
+}
+
 func TestIgnoreDDLByCommitTs(t *testing.T) {
 	// 1. Setup a mock SchemaStore.
 	// We don't need a real puller or kv storage for this test.
@@ -110,6 +152,153 @@ func TestIgnoreDDLByCommitTs(t *testing.T) {
 	require.NotContains(t, tableNames, "t2")
 }
 
+func TestTryUpdateResolvedTsRetryAfterDDLHandleFailure(t *testing.T) {
+	mockPDClock := pdutil.NewClock4Test()
+	appcontext.SetService(appcontext.DefaultPDClock, mockPDClock)
+
+	dir := t.TempDir()
+	pstorage := newPersistentStorageForTest(dir, nil)
+	defer func() {
+		err := pstorage.close()
+		require.NoError(t, err)
+	}()
+	pstorage.encryptionManager = &flakyEncryptionManagerForTest{failTimes: 1}
+
+	store := &keyspaceSchemaStore{
+		pdClock:       mockPDClock,
+		unsortedCache: newDDLCache(),
+		dataStorage:   pstorage,
+		notifyCh:      make(chan any, 1),
+	}
+	store.resolvedTs.Store(pstorage.gcTs)
+	store.pendingResolvedTs.Store(pstorage.gcTs)
+
+	createSchema := buildCreateSchemaJobForTest(100, "test", 1000)
+	createSchema.BinlogInfo.SchemaVersion = 1
+	createTable := buildCreateTableJobForTest(100, 200, "t1", 1010)
+	createTable.BinlogInfo.SchemaVersion = 2
+
+	store.writeDDLEvent(DDLJobWithCommitTs{
+		Job:      createSchema,
+		CommitTs: 1000,
+	})
+	store.writeDDLEvent(DDLJobWithCommitTs{
+		Job:      createTable,
+		CommitTs: 1010,
+	})
+	store.advancePendingResolvedTs(1010)
+
+	// First attempt fails when writing create schema; dedup watermarks should not advance.
+	store.tryUpdateResolvedTs()
+	require.Equal(t, int64(0), store.schemaVersion)
+	require.Equal(t, uint64(0), store.finishedDDLTs)
+	require.Equal(t, uint64(0), store.resolvedTs.Load())
+
+	// Second attempt retries failed DDLs and succeeds.
+	store.tryUpdateResolvedTs()
+	require.Equal(t, int64(2), store.schemaVersion)
+	require.Equal(t, uint64(1010), store.finishedDDLTs)
+	require.Equal(t, uint64(1010), store.resolvedTs.Load())
+
+	tables, err := pstorage.getAllPhysicalTables(1010, nil)
+	require.NoError(t, err)
+	require.Len(t, tables, 1)
+	require.Equal(t, "t1", tables[0].TableName)
+}
+
+func TestTryUpdateResolvedTsDoesNotAdvancePastFailedDDLAtSameCommitTs(t *testing.T) {
+	mockPDClock := pdutil.NewClock4Test()
+	appcontext.SetService(appcontext.DefaultPDClock, mockPDClock)
+
+	pstorage := newPersistentStorageForTest(t.TempDir(), nil)
+	defer func() {
+		require.NoError(t, pstorage.close())
+	}()
+	pstorage.encryptionManager = &flakyEncryptionManagerForTest{failOnCall: 2}
+
+	store := &keyspaceSchemaStore{
+		pdClock:       mockPDClock,
+		unsortedCache: newDDLCache(),
+		dataStorage:   pstorage,
+		notifyCh:      make(chan any, 1),
+	}
+	store.resolvedTs.Store(pstorage.gcTs)
+	store.pendingResolvedTs.Store(pstorage.gcTs)
+
+	const commitTs = uint64(1010)
+	createSchema := buildCreateSchemaJobForTest(100, "test", 1000)
+	createSchema.BinlogInfo.SchemaVersion = 1
+	createTable := buildCreateTableJobForTest(100, 200, "t1", 1010)
+	createTable.BinlogInfo.SchemaVersion = 2
+	store.writeDDLEvent(DDLJobWithCommitTs{Job: createSchema, CommitTs: commitTs})
+	store.writeDDLEvent(DDLJobWithCommitTs{Job: createTable, CommitTs: commitTs})
+	store.advancePendingResolvedTs(commitTs)
+
+	store.tryUpdateResolvedTs()
+	require.Equal(t, int64(1), store.schemaVersion)
+	require.Equal(t, uint64(1000), store.finishedDDLTs)
+	require.Less(t, store.resolvedTs.Load(), commitTs)
+
+	store.tryUpdateResolvedTs()
+	require.Equal(t, int64(2), store.schemaVersion)
+	require.Equal(t, uint64(1010), store.finishedDDLTs)
+	require.Equal(t, commitTs, store.resolvedTs.Load())
+}
+
+func TestGetAllPhysicalTablesDecryptsEncryptedDDLEvents(t *testing.T) {
+	dir := t.TempDir()
+	pstorage := newPersistentStorageForTest(dir, nil)
+	defer func() {
+		err := pstorage.close()
+		require.NoError(t, err)
+	}()
+	pstorage.encryptionManager = &prefixEncryptionManagerForTest{}
+
+	createSchema := buildCreateSchemaJobForTest(100, "test", 1000)
+	createSchema.BinlogInfo.SchemaVersion = 1
+	createTable := buildCreateTableJobForTest(100, 200, "t1", 1010)
+	createTable.BinlogInfo.SchemaVersion = 2
+
+	err := pstorage.handleDDLJob(createSchema)
+	require.NoError(t, err)
+	err = pstorage.handleDDLJob(createTable)
+	require.NoError(t, err)
+
+	tables, err := pstorage.getAllPhysicalTables(1010, nil)
+	require.NoError(t, err)
+	require.Len(t, tables, 1)
+	require.Equal(t, int64(200), tables[0].TableID)
+	require.Equal(t, int64(100), tables[0].SchemaID)
+	require.Equal(t, "t1", tables[0].TableName)
+}
+
+func TestRegisterTableDecryptsEncryptedDDLEvents(t *testing.T) {
+	dir := t.TempDir()
+	pstorage := newPersistentStorageForTest(dir, nil)
+	defer func() {
+		err := pstorage.close()
+		require.NoError(t, err)
+	}()
+	pstorage.encryptionManager = &prefixEncryptionManagerForTest{}
+
+	createSchema := buildCreateSchemaJobForTest(100, "test", 1000)
+	createSchema.BinlogInfo.SchemaVersion = 1
+	createTable := buildCreateTableJobForTest(100, 200, "t1", 1010)
+	createTable.BinlogInfo.SchemaVersion = 2
+
+	err := pstorage.handleDDLJob(createSchema)
+	require.NoError(t, err)
+	err = pstorage.handleDDLJob(createTable)
+	require.NoError(t, err)
+
+	err = pstorage.registerTable(200, 1010)
+	require.NoError(t, err)
+
+	tableInfo, err := pstorage.getTableInfo(200, 1010)
+	require.NoError(t, err)
+	require.NotNil(t, tableInfo)
+}
+
 func TestGetAllPhysicalTablesReturnsSnapshotLostByGCError(t *testing.T) {
 	dir := t.TempDir()
 	pstorage := newPersistentStorageForTest(dir, nil)
diff --git a/pkg/common/table_info_shared_schema_guard_test.go b/pkg/common/table_info_shared_schema_guard_test.go
index 7e807fcfd6..b7070369b5 100644
--- a/pkg/common/table_info_shared_schema_guard_test.go
+++ b/pkg/common/table_info_shared_schema_guard_test.go
@@ -83,7 +83,7 @@ func TestLatestTiDBTableInfoSharedSchemaGuard(t *testing.T) {
 				"Partition", "Compression", "View", "Sequence", "Lock", "Version", "TiFlashReplica", "IsColumnar",
 				"TempTableType", "TableCacheStatusType", "PlacementPolicyRef", "StatsOptions",
 				"ExchangePartitionInfo", "TTLInfo", "IsActiveActive", "SoftdeleteInfo", "Affinity",
-				"Revision", "DBID", "Mode",
+				"Revision", "DBID", "Mode", "EngineAttribute", "StorageClassTier", "StorageClassTransitions",
 			},
 		},
 		{