diff --git a/pkg/planner/core/plan_cache_utils.go b/pkg/planner/core/plan_cache_utils.go
index e6b4e00ff9e8c..9110d22ea711f 100644
--- a/pkg/planner/core/plan_cache_utils.go
+++ b/pkg/planner/core/plan_cache_utils.go
@@ -627,6 +627,57 @@ type PlanCacheStmt struct {
 	tbls   []table.Table
 }
 
+// PrepareStmtCacheEntry is stored in the session-level prepare dedup cache.
+// It holds the result of a full Prepare flow so that subsequent Prepares of
+// the same SQL text can skip the expensive Preprocess + PlanBuilder.Build.
+type PrepareStmtCacheEntry struct {
+	Stmt       *PlanCacheStmt
+	Fields     []*resolve.ResultField
+	ParamCount int
+}
+
+// ExtractAndSortParamMarkers extracts ParamMarkerExpr nodes from the AST,
+// sorts them by position, assigns their order indices, and initialises each
+// marker's Datum to NULL (matching the behaviour of GeneratePlanCacheStmtWithAST
+// for prepared statements where the actual parameter values are not yet known).
+func ExtractAndSortParamMarkers(stmtNode ast.StmtNode) []ast.ParamMarkerExpr {
+	var extractor paramMarkerExtractor
+	stmtNode.Accept(&extractor)
+	slices.SortFunc(extractor.markers, func(i, j ast.ParamMarkerExpr) int {
+		return cmp.Compare(i.(*driver.ParamMarkerExpr).Offset, j.(*driver.ParamMarkerExpr).Offset)
+	})
+	for i, m := range extractor.markers {
+		m.SetOrder(i)
+		p := m.(*driver.ParamMarkerExpr)
+		p.Datum.SetNull()
+		p.InExecute = false
+	}
+	return extractor.markers
+}
+
+// CollectPlanCacheStmtInfo walks the AST to populate the limits, hasSubquery,
+// and tables fields of stmt. It must be called on the fresh AST after a
+// re-parse so that limit nodes and table references point into the new tree.
+func CollectPlanCacheStmtInfo(ctx context.Context, is infoschema.InfoSchema, stmt *PlanCacheStmt, stmtNode ast.StmtNode) {
+	processor := &planCacheStmtProcessor{ctx: ctx, is: is, stmt: stmt}
+	stmtNode.Accept(processor)
+}
+
+// DBName returns the dbName field (used for metadata lock during Execute).
+func (s *PlanCacheStmt) DBName() []model.CIStr { return s.dbName }
+
+// Tbls returns the tbls field (used for metadata lock during Execute).
+func (s *PlanCacheStmt) Tbls() []table.Table { return s.tbls }
+
+// SetDBNameAndTbls sets the dbName and tbls fields, cloning the input slices
+// so that this PlanCacheStmt owns independent backing arrays. This is required
+// because planCachePreprocess replaces tbls[i] in-place during Execute, and
+// sharing the backing array with a cached template would cause cross-stmt contamination.
+func (s *PlanCacheStmt) SetDBNameAndTbls(dbName []model.CIStr, tbls []table.Table) {
+	s.dbName = slices.Clone(dbName)
+	s.tbls = slices.Clone(tbls)
+}
+
 // GetPreparedStmt extract the prepared statement from the execute statement.
 func GetPreparedStmt(stmt *ast.ExecuteStmt, vars *variable.SessionVars) (*PlanCacheStmt, error) {
 	if stmt.PrepStmt != nil {
diff --git a/pkg/server/internal/testserverclient/server_client.go b/pkg/server/internal/testserverclient/server_client.go
index 9dbaf855851ee..f731fbc4b52d3 100644
--- a/pkg/server/internal/testserverclient/server_client.go
+++ b/pkg/server/internal/testserverclient/server_client.go
@@ -2581,6 +2581,10 @@ func (cli *TestServerClient) RunTestInitConnect(t *testing.T) {
 // and not internal SQL statements. Thus, this test is in the server-test suite.
 func (cli *TestServerClient) RunTestInfoschemaClientErrors(t *testing.T) {
 	cli.RunTestsOnNewDB(t, nil, "clientErrors", func(dbt *testkit.DBTestKit) {
+		dbt.MustExec("set @@tidb_enable_cache_prepare_stmt = off")
+		defer func() {
+			dbt.MustExec("set @@tidb_enable_cache_prepare_stmt = default")
+		}()
 		clientErrors := []struct {
 			stmt              string
 			incrementWarnings bool
diff --git a/pkg/session/session.go b/pkg/session/session.go
index 74731d93a501f..7f189ef11a3e7 100644
--- a/pkg/session/session.go
+++ b/pkg/session/session.go
@@ -27,6 +27,7 @@ import (
 	stderrs "errors"
 	"fmt"
 	"iter"
+	"maps"
 	"math"
 	"math/rand"
 	"runtime/pprof"
@@ -2529,6 +2530,37 @@ func (s *session) PrepareStmt(sql string) (stmtID uint32, paramCount int, fields
 	if err = sessiontxn.GetTxnManager(s).AdviseWarmup(); err != nil {
 		return
 	}
+
+	var dedupKey string
+	if s.sessionVars.EnableCachePrepareStmt {
+		// Session-level prepare dedup cache: if the same SQL text has been prepared
+		// before in this session (with the same charset/collation/currentDB), reuse
+		// the already-built PlanCacheStmt and skip the expensive Preprocess+Build.
+		charset, collation := s.sessionVars.GetCharsetInfo()
+		dedupKey = variable.PrepareDedupCacheKey(sql, charset, collation, s.sessionVars.CurrentDB, s.sessionVars.SQLMode)
+		if v := s.sessionVars.GetPrepareStmtDedupCache(dedupKey); v != nil {
+			cached := v.(*plannercore.PrepareStmtCacheEntry)
+			is := sessiontxn.GetTxnManager(s).GetTxnInfoSchema()
+			if cached.Stmt.SchemaVersion == is.SchemaMetaVersion() {
+				newStmt, rebuildErr := s.rebuildFromPrepareCache(ctx, cached, sql, charset, collation)
+				if rebuildErr == nil {
+					stmtID = s.sessionVars.GetNextPreparedStmtID()
+					if err = s.sessionVars.AddPreparedStmt(stmtID, newStmt); err != nil {
+						s.rollbackOnError(ctx)
+						return
+					}
+					paramCount = cached.ParamCount
+					fields = cached.Fields
+					s.rollbackOnError(ctx)
+					return
+				}
+				// Re-parse or rebuild failed; fall through to the full prepare path.
+				logutil.Logger(ctx).Warn("prepare stmt dedup cache rebuild failed, fallback to full prepare", zap.Error(rebuildErr))
+			}
+			// Schema version changed; fall through and re-cache below.
+		}
+	}
+
 	prepareExec := executor.NewPrepareExec(s, sql)
 	err = prepareExec.Next(ctx, nil)
 	// Rollback even if err is nil.
@@ -2537,9 +2569,107 @@ func (s *session) PrepareStmt(sql string) (stmtID uint32, paramCount int, fields
 	if err != nil {
 		return
 	}
+
+	// Store the result in the dedup cache for future Prepares of the same SQL.
+	if s.sessionVars.EnableCachePrepareStmt {
+		if prepareExec.Stmt != nil {
+			s.sessionVars.SetPrepareStmtDedupCache(dedupKey, &plannercore.PrepareStmtCacheEntry{
+				Stmt:       prepareExec.Stmt.(*plannercore.PlanCacheStmt),
+				Fields:     prepareExec.Fields,
+				ParamCount: prepareExec.ParamCount,
+			})
+		}
+	}
 	return prepareExec.ID, prepareExec.ParamCount, prepareExec.Fields, nil
 }
 
+// rebuildFromPrepareCache constructs a new PlanCacheStmt from a cached entry,
+// re-parsing the SQL to obtain an independent AST (with fresh ParamMarkerExpr
+// nodes) while skipping only the expensive PlanBuilder.Build step.
+// Preprocess is still executed to build a fresh ResolveCtx whose tableNames map
+// is keyed by the new AST's TableName pointers; reusing the cached ResolveCtx
+// would cause nil-deref panics on plan-cache miss because the old pointer keys
+// would not match the newly-parsed AST nodes.
+func (s *session) rebuildFromPrepareCache(
+	ctx context.Context,
+	cached *plannercore.PrepareStmtCacheEntry,
+	sql, charset, collation string,
+) (*plannercore.PlanCacheStmt, error) {
+	stmts, _, err := s.ParseSQL(ctx, sql,
+		parser.CharsetConnection(charset),
+		parser.CollationConnection(collation),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if len(stmts) != 1 {
+		return nil, errors.New("unexpected statement count after re-parse")
+	}
+	stmtNode := stmts[0]
+
+	// Extract fresh param markers from the new AST and initialise them to NULL.
+	markers := plannercore.ExtractAndSortParamMarkers(stmtNode)
+
+	is := sessiontxn.GetTxnManager(s).GetTxnInfoSchema()
+
+	// Run Preprocess to build a fresh ResolveCtx aligned with the new AST.
+	// This is the only way to populate ResolveCtx.tableNames with the new
+	// AST's *ast.TableName pointer keys without re-running the full Build.
+	ret := &plannercore.PreprocessorReturn{InfoSchema: is}
+	nodeW := resolve.NewNodeW(stmtNode)
+	if err = plannercore.Preprocess(ctx, s, nodeW, plannercore.InPrepare,
+		plannercore.WithPreprocessorReturn(ret)); err != nil {
+		return nil, err
+	}
+	// Defensive: if schema changed between our earlier check and Preprocess,
+	// fall through to the full prepare path.
+	if ret.InfoSchema.SchemaMetaVersion() != cached.Stmt.SchemaVersion {
+		return nil, errors.New("schema version changed during rebuild")
+	}
+
+	newStmt := &plannercore.PlanCacheStmt{
+		// Fields derived from the new AST:
+		PreparedAst: &ast.Prepared{
+			Stmt:     stmtNode,
+			StmtType: cached.Stmt.PreparedAst.StmtType,
+		},
+		Params: markers,
+
+		// Fresh ResolveCtx whose tableNames keys match the new AST pointers.
+		ResolveCtx: nodeW.GetResolveContext(),
+
+		// Immutable fields – safe to share with the cached template:
+		StmtDB:              cached.Stmt.StmtDB,
+		StmtText:            cached.Stmt.StmtText,
+		VisitInfos:          cached.Stmt.VisitInfos,
+		NormalizedSQL:       cached.Stmt.NormalizedSQL,
+		SQLDigest:           cached.Stmt.SQLDigest,
+		ForUpdateRead:       cached.Stmt.ForUpdateRead,
+		SnapshotTSEvaluator: cached.Stmt.SnapshotTSEvaluator,
+		StmtCacheable:       cached.Stmt.StmtCacheable,
+		UncacheableReason:   cached.Stmt.UncacheableReason,
+		SchemaVersion:       cached.Stmt.SchemaVersion,
+
+		// Mutable containers – clone so each stmt has independent state:
+		RelateVersion: maps.Clone(cached.Stmt.RelateVersion),
+		// PointGet is zeroed (per-execution executor state must not leak).
+		// NormalizedPlan / PlanDigest are left as zero values; they will be
+		// populated on the first plan-cache miss during Execute.
+	}
+
+	// Walk the new AST to populate limits, hasSubquery, and tables.
+	// These fields hold pointers into the AST, so they must refer to the
+	// newly-parsed tree rather than the cached one.
+	plannercore.CollectPlanCacheStmtInfo(ctx, is, newStmt, stmtNode)
+
+	// dbName and tbls are only read during Execute (not written to by
+	// CollectPlanCacheStmtInfo since that populates tables, not tbls).
+	// Clone them so that planCachePreprocess can safely replace tbls[i].
+	newStmt.SetDBNameAndTbls(cached.Stmt.DBName(), cached.Stmt.Tbls())
+
+	return newStmt, nil
+}
+
 // ExecutePreparedStmt executes a prepared statement.
 func (s *session) ExecutePreparedStmt(ctx context.Context, stmtID uint32, params []expression.Expression) (sqlexec.RecordSet, error) {
 	prepStmt, err := s.sessionVars.GetPreparedStmtByID(stmtID)
diff --git a/pkg/session/test/common/BUILD.bazel b/pkg/session/test/common/BUILD.bazel
index 11b93655278c6..3d3dbe939a426 100644
--- a/pkg/session/test/common/BUILD.bazel
+++ b/pkg/session/test/common/BUILD.bazel
@@ -6,9 +6,10 @@ go_test(
     srcs = [
         "common_test.go",
         "main_test.go",
+        "prepare_dedup_cache_test.go",
     ],
     flaky = True,
-    shard_count = 10,
+    shard_count = 15,
     deps = [
         "//pkg/config",
         "//pkg/expression",
diff --git a/pkg/session/test/common/prepare_dedup_cache_test.go b/pkg/session/test/common/prepare_dedup_cache_test.go
new file mode 100644
index 0000000000000..aee65516987bd
--- /dev/null
+++ b/pkg/session/test/common/prepare_dedup_cache_test.go
@@ -0,0 +1,211 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/expression"
+	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/stretchr/testify/require"
+)
+
+// TestPrepareStmtDedupCacheBasic verifies that preparing the same SQL twice in
+// the same session reuses the cached PlanCacheStmt: both stmtIDs are distinct,
+// but paramCount and column metadata match.
+func TestPrepareStmtDedupCacheBasic(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t (id bigint primary key, age int, city varchar(32))")
+
+	sql := "select id, city from t where age > ? and city = ?"
+
+	id1, paramCount1, fields1, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.EqualValues(t, 2, paramCount1)
+	require.Len(t, fields1, 2)
+	require.Equal(t, "id", fields1[0].Column.Name.L)
+	require.Equal(t, "city", fields1[1].Column.Name.L)
+
+	// Second prepare of the same SQL — should hit the dedup cache.
+	id2, paramCount2, fields2, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.NotEqual(t, id1, id2, "each Prepare must return a distinct stmtID")
+	require.Equal(t, paramCount1, paramCount2)
+	require.Len(t, fields2, 2)
+	require.Equal(t, "id", fields2[0].Column.Name.L)
+	require.Equal(t, "city", fields2[1].Column.Name.L)
+}
+
+// TestPrepareStmtDedupCacheExecute verifies that stmts produced via the dedup
+// cache path execute correctly and return the right rows.
+func TestPrepareStmtDedupCacheExecute(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t2 (id bigint primary key, val int)")
+	tk.MustExec("insert into t2 values (1, 10), (2, 20), (3, 30)")
+
+	ctx := context.Background()
+	sql := "select id from t2 where val > ?"
+
+	runAndCollect := func(stmtID uint32, threshold int) []int64 {
+		rs, err := tk.Session().ExecutePreparedStmt(ctx, stmtID,
+			expression.Args2Expressions4Test(threshold))
+		require.NoError(t, err)
+		defer rs.Close()
+		var ids []int64
+		req := rs.NewChunk(nil)
+		for {
+			require.NoError(t, rs.Next(ctx, req))
+			if req.NumRows() == 0 {
+				break
+			}
+			for i := range req.NumRows() {
+				ids = append(ids, req.Column(0).GetInt64(i))
+			}
+			req.Reset()
+		}
+		return ids
+	}
+
+	// First prepare — full build path.
+	id1, _, _, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	ids := runAndCollect(id1, 15)
+	require.Equal(t, []int64{2, 3}, ids)
+
+	// Second prepare — dedup cache path.
+	id2, _, _, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.NotEqual(t, id1, id2)
+	ids = runAndCollect(id2, 5)
+	require.Equal(t, []int64{1, 2, 3}, ids)
+
+	// The original stmt must still work independently.
+	ids = runAndCollect(id1, 25)
+	require.Equal(t, []int64{3}, ids)
+}
+
+// TestPrepareStmtDedupCacheSchemaChange verifies that a DDL (schema version bump)
+// causes the dedup cache entry to be invalidated and the next Prepare to go
+// through the full build path, reflecting the updated schema.
+func TestPrepareStmtDedupCacheSchemaChange(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t3 (id bigint primary key, name varchar(32))")
+
+	sql := "select id, name from t3 where id = ?"
+
+	// Warm up the dedup cache.
+	id1, _, fields1, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.Len(t, fields1, 2)
+	_ = id1
+
+	// DDL that changes the schema version.
+	tk.MustExec("alter table t3 add column email varchar(64)")
+
+	// After DDL the dedup cache entry should be stale (schema version mismatch)
+	// and the full build path should be taken. The returned fields come from
+	// the original prepare (the client protocol doesn't return field info again
+	// on cache hit), but the resulting PlanCacheStmt must be valid.
+	id2, _, _, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.NotEqual(t, id1, id2)
+
+	// The new stmt must execute correctly against the updated schema.
+	ctx := context.Background()
+	tk.MustExec("insert into t3 (id, name, email) values (1, 'alice', 'alice@example.com')")
+	rs, err := tk.Session().ExecutePreparedStmt(ctx, id2, expression.Args2Expressions4Test(1))
+	require.NoError(t, err)
+	require.NoError(t, rs.Close())
+}
+
+// TestPrepareStmtDedupCacheIsolatedByDB verifies that prepare dedup cache entries
+// are isolated per database: the same SQL text prepared while connected to
+// different databases must not share a cache entry.
+func TestPrepareStmtDedupCacheIsolatedByDB(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("create database if not exists db1")
+	tk.MustExec("create database if not exists db2")
+	tk.MustExec("use db1")
+	tk.MustExec("create table tblx (id bigint primary key, v int)")
+	tk.MustExec("use db2")
+	tk.MustExec("create table tblx (id bigint primary key, v bigint)") // different column type
+
+	sql := "select v from tblx where id = ?"
+
+	// Prepare in db1.
+	tk.MustExec("use db1")
+	id1, _, fields1, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.Len(t, fields1, 1)
+	_ = id1
+
+	// Prepare in db2 — must NOT hit the db1 cache entry because currentDB differs.
+	tk.MustExec("use db2")
+	id2, _, fields2, err := tk.Session().PrepareStmt(sql)
+	require.NoError(t, err)
+	require.NotEqual(t, id1, id2)
+	// fields2 reflects db2.tblx.v which is bigint (not int), so a fresh build must have run.
+	require.Len(t, fields2, 1)
+}
+
+// TestPrepareStmtDedupCachePrepareExecuteCloseLoop verifies the prepare-per-request
+// anti-pattern: multiple Prepare→Execute→Close cycles for the same SQL all
+// produce correct results.
+func TestPrepareStmtDedupCachePrepareExecuteCloseLoop(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t4 (id bigint primary key, score int)")
+	tk.MustExec("insert into t4 values (1,100),(2,200),(3,300)")
+
+	ctx := context.Background()
+	sql := "select id from t4 where score >= ?"
+
+	thresholds := []int{50, 150, 250}
+	expected := [][]int64{{1, 2, 3}, {2, 3}, {3}}
+
+	for round, threshold := range thresholds {
+		id, _, _, err := tk.Session().PrepareStmt(sql)
+		require.NoError(t, err)
+
+		rs, err := tk.Session().ExecutePreparedStmt(ctx, id, expression.Args2Expressions4Test(threshold))
+		require.NoError(t, err)
+
+		var got []int64
+		req := rs.NewChunk(nil)
+		for {
+			require.NoError(t, rs.Next(ctx, req))
+			if req.NumRows() == 0 {
+				break
+			}
+			for i := range req.NumRows() {
+				got = append(got, req.Column(0).GetInt64(i))
+			}
+			req.Reset()
+		}
+		require.NoError(t, rs.Close())
+		require.Equal(t, expected[round], got, "round %d, threshold %d", round, threshold)
+
+		require.NoError(t, tk.Session().DropPreparedStmt(id))
+	}
+}
diff --git a/pkg/sessionctx/variable/session.go b/pkg/sessionctx/variable/session.go
index 332c1d9e06bba..88346f07def13 100644
--- a/pkg/sessionctx/variable/session.go
+++ b/pkg/sessionctx/variable/session.go
@@ -779,6 +779,10 @@ type SessionVars struct {
 	SysErrorCount uint16
 	// nonPreparedPlanCacheStmts stores PlanCacheStmts for non-prepared plan cache.
 	nonPreparedPlanCacheStmts *kvcache.SimpleLRUCache
+	// prepareStmtDedupCache caches PlanCacheStmt templates keyed by SQL text +
+	// charset + collation + currentDB to skip redundant Parse+Preprocess+Build
+	// on repeated COM_STMT_PREPARE for the same SQL within a session.
+	prepareStmtDedupCache *kvcache.SimpleLRUCache
 	// PreparedStmts stores prepared statement.
 	PreparedStmts        map[uint32]any
 	PreparedStmtNameToID map[string]uint32
@@ -1741,6 +1745,9 @@ type SessionVars struct {
 
 	// OutPacketBytes records the total outcoming packet bytes to clients for current session.
 	OutPacketBytes atomic.Uint64
+
+	// EnableCachePrepareStmt indicates whether to cache prepare stmt in plan cache.
+	EnableCachePrepareStmt bool
 }
 
 // GetSessionVars implements the `SessionVarsProvider` interface.
@@ -2280,6 +2287,7 @@ func NewSessionVars(hctx HookContext) *SessionVars {
 		OptimizerEnableNAAJ:            DefTiDBEnableNAAJ,
 		RegardNULLAsPoint:              DefTiDBRegardNULLAsPoint,
 		AllowProjectionPushDown:        DefOptEnableProjectionPushDown,
+		EnableCachePrepareStmt:         DefEnableCachePrepareStmt,
 	}
 	vars.TiFlashFineGrainedShuffleBatchSize = DefTiFlashFineGrainedShuffleBatchSize
 	vars.status.Store(uint32(mysql.ServerStatusAutocommit))
@@ -2708,6 +2716,37 @@ func (s *SessionVars) GetNonPreparedPlanCacheStmt(sql string) any {
 	return stmt
 }
 
+// PrepareDedupCacheKey builds the lookup key for the prepare dedup cache.
+// Including charset, collation, currentDB and sqlMode ensures that the cached
+// PlanCacheStmt is only reused when the session context that affects parsing,
+// name-resolution and cacheability decisions is identical. sqlMode is included
+// because flags like PIPES_AS_CONCAT and ANSI_QUOTES change AST shape, and
+// IsASTCacheable (which computes StmtCacheable) runs on that AST.
+func PrepareDedupCacheKey(sql, charset, collation, currentDB string, sqlMode mysql.SQLMode) string {
+	var modeBuf [8]byte
+	binary.LittleEndian.PutUint64(modeBuf[:], uint64(sqlMode))
+	return sql + "\x00" + charset + "\x00" + collation + "\x00" + currentDB + "\x00" + string(modeBuf[:])
+}
+
+// GetPrepareStmtDedupCache returns the cached PrepareStmtCacheEntry for the given key,
+// or nil when the cache is empty or the key is not found.
+func (s *SessionVars) GetPrepareStmtDedupCache(key string) any {
+	if s.prepareStmtDedupCache == nil {
+		return nil
+	}
+	v, _ := s.prepareStmtDedupCache.Get(planCacheStmtKey(key))
+	return v
+}
+
+// SetPrepareStmtDedupCache stores a PrepareStmtCacheEntry under the given key.
+// The cache is lazily initialized and bounded by SessionPlanCacheSize (LRU eviction).
+func (s *SessionVars) SetPrepareStmtDedupCache(key string, val any) {
+	if s.prepareStmtDedupCache == nil {
+		s.prepareStmtDedupCache = kvcache.NewSimpleLRUCache(uint(s.SessionPlanCacheSize), 0, 0)
+	}
+	s.prepareStmtDedupCache.Put(planCacheStmtKey(key), val)
+}
+
 // AddPreparedStmt adds prepareStmt to current session and count in global.
 func (s *SessionVars) AddPreparedStmt(stmtID uint32, stmt any) error {
 	if _, exists := s.PreparedStmts[stmtID]; !exists {
diff --git a/pkg/sessionctx/variable/setvar_affect.go b/pkg/sessionctx/variable/setvar_affect.go
index 323cb0548901a..c5cd69125c840 100644
--- a/pkg/sessionctx/variable/setvar_affect.go
+++ b/pkg/sessionctx/variable/setvar_affect.go
@@ -128,6 +128,7 @@ var isHintUpdatableVerified = map[string]struct{}{
 	"tiflash_fine_grained_shuffle_batch_size":         {},
 	"tiflash_fine_grained_shuffle_stream_count":       {},
 	"tidb_hash_join_version":                          {},
+	"tidb_enable_cache_prepare_stmt":                  {},
 	// Variables that is compatible with MySQL.
 	"cte_max_recursion_depth": {},
 	"sql_mode":                {},
diff --git a/pkg/sessionctx/variable/sysvar.go b/pkg/sessionctx/variable/sysvar.go
index dc5923faa5b9b..bddcec5c5ba90 100644
--- a/pkg/sessionctx/variable/sysvar.go
+++ b/pkg/sessionctx/variable/sysvar.go
@@ -3643,6 +3643,10 @@ var defaultSysVars = []*SysVar{
 			return (*SetPDClientDynamicOption.Load())(TiDBTSOClientRPCMode, val)
 		},
 	},
+	{Scope: ScopeGlobal | ScopeSession, Name: TiDBEnableCachePrepareStmt, Value: BoolToOnOff(DefEnableCachePrepareStmt), Type: TypeBool, SetSession: func(s *SessionVars, val string) error {
+		s.EnableCachePrepareStmt = TiDBOptOn(val)
+		return nil
+	}},
 }
 
 // GlobalSystemVariableInitialValue gets the default value for a system variable including ones that are dynamically set (e.g. based on the store)
diff --git a/pkg/sessionctx/variable/tidb_vars.go b/pkg/sessionctx/variable/tidb_vars.go
index 4be64269362bf..d1aa28a5c673f 100644
--- a/pkg/sessionctx/variable/tidb_vars.go
+++ b/pkg/sessionctx/variable/tidb_vars.go
@@ -1286,6 +1286,9 @@ const (
 
 	// MaxPreSplitRegions is the maximum number of regions that can be pre-split.
 	MaxPreSplitRegions = 15
+
+	// TiDBEnableCachePrepareStmt indicates whether to support cache prepare stmt in plan cache.
+	TiDBEnableCachePrepareStmt = "tidb_enable_cache_prepare_stmt"
 )
 
 // Default TiDB system variable values.
@@ -1654,6 +1657,7 @@ const (
 	DefTiDBEnableSharedLockPromotion                  = false
 	DefTiDBTSOClientRPCMode                           = TSOClientRPCModeDefault
 	DefTiDBLoadBindingTimeout                         = 200
+	DefEnableCachePrepareStmt                         = true
 )
 
 // Process global variables.
diff --git a/tests/integrationtest/r/sessionctx/setvar.result b/tests/integrationtest/r/sessionctx/setvar.result
index c848098efdbfb..7ce525e6adfa4 100644
--- a/tests/integrationtest/r/sessionctx/setvar.result
+++ b/tests/integrationtest/r/sessionctx/setvar.result
@@ -1713,3 +1713,35 @@ select @@max_execution_time;
 @@max_execution_time
 2000
 set @@global.max_execution_time=default;
+select /*+ set_var(tidb_enable_cache_prepare_stmt=0) */ @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+0
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
+set @@tidb_enable_cache_prepare_stmt=default;
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
+select /*+ set_var(tidb_enable_cache_prepare_stmt=1) */ @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
+set @@tidb_enable_cache_prepare_stmt=default;
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
+set @@tidb_enable_cache_prepare_stmt=off;
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+0
+set @@tidb_enable_cache_prepare_stmt=on;
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
+set @@tidb_enable_cache_prepare_stmt=default;
+select @@tidb_enable_cache_prepare_stmt;
+@@tidb_enable_cache_prepare_stmt
+1
diff --git a/tests/integrationtest/t/sessionctx/setvar.test b/tests/integrationtest/t/sessionctx/setvar.test
index d220bba452b11..07778bdda5320 100644
--- a/tests/integrationtest/t/sessionctx/setvar.test
+++ b/tests/integrationtest/t/sessionctx/setvar.test
@@ -703,3 +703,17 @@ select /*+ set_var(max_execution_time=100) */ @@max_execution_time;
 select @@max_execution_time;
 set @@global.max_execution_time=default;
 disconnect conn1;
+select /*+ set_var(tidb_enable_cache_prepare_stmt=0) */ @@tidb_enable_cache_prepare_stmt;
+select @@tidb_enable_cache_prepare_stmt;
+set @@tidb_enable_cache_prepare_stmt=default;
+select @@tidb_enable_cache_prepare_stmt;
+select /*+ set_var(tidb_enable_cache_prepare_stmt=1) */ @@tidb_enable_cache_prepare_stmt;
+select @@tidb_enable_cache_prepare_stmt;
+set @@tidb_enable_cache_prepare_stmt=default;
+select @@tidb_enable_cache_prepare_stmt;
+set @@tidb_enable_cache_prepare_stmt=off;
+select @@tidb_enable_cache_prepare_stmt;
+set @@tidb_enable_cache_prepare_stmt=on;
+select @@tidb_enable_cache_prepare_stmt;
+set @@tidb_enable_cache_prepare_stmt=default;
+select @@tidb_enable_cache_prepare_stmt;