diff --git a/README b/README
index c054f1d..4255271 100644
--- a/README
+++ b/README
@@ -66,7 +66,7 @@ LINUX (Ubuntu/Debian)
 
 3. Clone MariaDB and copy TidesDB storage engine:
 
-   git clone --depth 1 --branch 12.1 https://github.com/MariaDB/server.git mariadb-server
+   git clone --depth 1 --branch mariadb-12.2.2 https://github.com/MariaDB/server.git mariadb-server
    cd mariadb-server
    git submodule update --init --recursive
    cp -r /path/to/tidesql/tidesdb storage/
@@ -105,7 +105,7 @@ MACOS
 
 3. Clone MariaDB and copy TidesDB storage engine:
 
-   git clone --depth 1 --branch 12.1 https://github.com/MariaDB/server.git mariadb-server
+   git clone --depth 1 --branch mariadb-12.2.2 https://github.com/MariaDB/server.git mariadb-server
    cd mariadb-server
    git submodule update --init --recursive
    cp -r /path/to/tidesql/tidesdb storage/
@@ -171,7 +171,7 @@ WINDOWS
 
 4. Clone MariaDB and copy TidesDB storage engine:
 
-   git clone --depth 1 --branch 12.1 https://github.com/MariaDB/server.git mariadb-server
+   git clone --depth 1 --branch mariadb-12.2.2 https://github.com/MariaDB/server.git mariadb-server
    cd mariadb-server
    git submodule update --init --recursive
    xcopy /E /I path\to\tidesql\tidesdb storage\tidesdb
@@ -210,17 +210,24 @@ Core:
 - MVCC transactions with per-table isolation (autocommit uses READ_COMMITTED;
   multi-statement transactions use the table's configured level)
 - SQL savepoints (SAVEPOINT / ROLLBACK TO / RELEASE SAVEPOINT)
+- START TRANSACTION WITH CONSISTENT SNAPSHOT
 - Lock-free concurrency (no THR_LOCK, TidesDB handles isolation internally)
+- Optional pessimistic row locking (tidesdb_pessimistic_locking=ON) for
+  workloads that depend on SELECT ... FOR UPDATE serialization (e.g. TPC-C)
 - LSM-tree storage with optional B+tree SSTable format
 - Compression (NONE, LZ4, LZ4_FAST, ZSTD, Snappy)
 - Bloom filters for fast key lookups
 - Block cache for frequently accessed data
 - Primary key (single and composite) and secondary index support
 - Index Condition Pushdown (ICP) for secondary index scans
+- REPLACE INTO and INSERT ... ON DUPLICATE KEY UPDATE
 - AUTO_INCREMENT with O(1) atomic counter (no iterator per INSERT)
 - TTL (time-to-live) per-row and per-table expiration
 - Virtual/generated columns
 - Online backup (SET GLOBAL tidesdb_backup_dir = '/path')
+- Hard-link checkpoint (SET GLOBAL tidesdb_checkpoint_dir = '/path')
+- OPTIMIZE TABLE (synchronous purge + compact via tidesdb_purge_cf)
+- SHOW ENGINE TIDESDB STATUS (DB stats, memory, cache, conflict info)
 - Partitioning (RANGE, LIST, HASH, KEY)
 - Data-at-rest encryption (MariaDB encryption plugin integration)
 - Online DDL (instant metadata, inplace add/drop index, copy for columns)
@@ -239,7 +246,7 @@ TidesDB stores its data as a sibling of the MariaDB data directory:
 SYSTEM VARIABLES (SET GLOBAL tidesdb_...)
 ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
 
-All are read-only (set at startup) unless noted otherwise.
+Read-only (set at startup):
 
   flush_threads             Background flush threads (default: 4)
   compaction_threads        Background compaction threads (default: 4)
@@ -247,8 +254,29 @@ All are read-only (set at startup) unless noted otherwise.
   block_cache_size          Global block cache in bytes (default: 256MB)
   max_open_sstables         Max cached SSTable files (default: 256)
   max_memory_usage          Global memory limit in bytes; 0 = auto (default: 0)
-  backup_dir                [dynamic] Set to a path to trigger online backup
-  debug_trace               [dynamic] Per-operation trace logging (default: OFF)
+  data_home_dir             Override TidesDB data directory (default: auto)
+  log_to_file               Write logs to file vs stderr (default: ON)
+  log_truncation_at         Log file truncation size (default: 24MB; 0 = off)
+  row_lock_stripes          Striped mutexes for pessimistic locking (default: 1024)
+
+Dynamic (SET GLOBAL at runtime):
+
+  backup_dir                Set to a path to trigger online backup
+  checkpoint_dir            Set to a path to trigger hard-link checkpoint
+  print_all_conflicts       Log all TDB_ERR_CONFLICT events (default: OFF)
+  pessimistic_locking       Enable plugin-level row locks for UPDATE/DELETE
+                            (default: OFF). Serializes concurrent writes to
+                            the same PK like InnoDB's row locks. Enable for
+                            TPC-C or workloads needing FOR UPDATE semantics.
+
+Session (SET SESSION tidesdb_...):
+
+  ttl                       Per-session TTL override in seconds (default: 0)
+  skip_unique_check         Skip PK/unique checks on INSERT (default: OFF)
+  default_compression       Default compression for new tables
+  default_write_buffer_size Default write buffer for new tables (32MB)
+  default_sync_mode         Default sync mode for new tables (FULL)
+  (and other default_* variables for all table options)
 
 Logging: TidesDB writes to <tidesdb_data>/LOG by default with automatic
 truncation at 24 MB.  Set log_level to WARN or higher in production to
@@ -258,9 +286,9 @@ reduce log volume.
 TABLE OPTIONS (CREATE TABLE ... ENGINE=TidesDB <option>=<value>)
 ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
 
-These are per-table options baked into the column family at creation time.
-Changing them via ALTER TABLE only updates the .frm; it does not reconfigure
-the live column family.
+These are per-table options set at creation time and applied to the column
+family configuration.  ALTER TABLE ... <option>=<value> updates both the .frm
+and the live column family via tidesdb_cf_update_runtime_config().
 
 Storage:
   WRITE_BUFFER_SIZE         Memtable size before flush (default: 32MB)
@@ -332,25 +360,37 @@ Run specific test:
 
   perl mtr --suite=tidesdb tidesdb_crud
 
-Available tests:
-  tidesdb_crud            Basic CRUD operations
-  tidesdb_pk_index        Primary key and secondary index scans
-  tidesdb_options         Table and field options
-  tidesdb_ttl             Time-to-live expiration
-  tidesdb_vcol            Virtual/generated columns
-  tidesdb_encryption      Data-at-rest encryption
-  tidesdb_backup          Online backup
-  tidesdb_partition       RANGE/LIST/HASH/KEY partitioning
-  tidesdb_online_ddl      Online DDL (instant, inplace, copy)
-  tidesdb_analyze         ANALYZE TABLE with CF statistics output
-  tidesdb_rename          Table rename
-  tidesdb_stress          Concurrent transactions, conflicts, truncate cycles
-  tidesdb_sql             Comprehensive SQL: 40 cases (aggregates, JOINs,
-                          subqueries, UNION, transactions, NULL handling, etc.)
-  tidesdb_json            JSON querying + generated-column JSON path indexing
-  tidesdb_savepoint       SQL SAVEPOINT / ROLLBACK TO / RELEASE SAVEPOINT
-  tidesdb_write_pressure  oltp_write_only OOM reproduction (multi-connection)
-  ...
+Available tests (30 total):
+  tidesdb_alter_crash         Crash safety during ALTER TABLE operations
+  tidesdb_analyze             ANALYZE TABLE with CF statistics output
+  tidesdb_backup              Online backup via tidesdb_backup_dir
+  tidesdb_concurrent_conflict Concurrent write-write conflict handling
+  tidesdb_concurrent_errors   Multi-connection error handling and recovery
+  tidesdb_consistent_snapshot START TRANSACTION WITH CONSISTENT SNAPSHOT
+  tidesdb_crud                Basic CRUD operations
+  tidesdb_data_home_dir       tidesdb_data_home_dir sysvar
+  tidesdb_drop_create         Repeated DROP/CREATE/TRUNCATE cycles
+  tidesdb_encryption          Data-at-rest encryption
+  tidesdb_engine_status       SHOW ENGINE TIDESDB STATUS
+  tidesdb_index_stats         Index statistics and optimizer cost model
+  tidesdb_info_schema         information_schema integration
+  tidesdb_insert_conflict     Duplicate key detection and handling
+  tidesdb_isolation           Isolation level behavior
+  tidesdb_json                JSON querying + generated-column JSON indexing
+  tidesdb_online_ddl          Online DDL (instant, inplace, copy)
+  tidesdb_options             Table and field options
+  tidesdb_partition           RANGE/LIST/HASH/KEY partitioning
+  tidesdb_per_index_btree     Per-index USE_BTREE option
+  tidesdb_pk_index            Primary key and secondary index scans
+  tidesdb_rename              Table rename
+  tidesdb_replace_iodku       REPLACE INTO and INSERT ON DUPLICATE KEY UPDATE
+  tidesdb_savepoint           SQL SAVEPOINT / ROLLBACK TO / RELEASE SAVEPOINT
+  tidesdb_sql                 40 SQL cases (aggregates, JOINs, subqueries, etc.)
+  tidesdb_stress              Concurrent transactions, conflicts, truncate cycles
+  tidesdb_tpcc_contention     TPC-C district counter contention (pessimistic locking)
+  tidesdb_ttl                 Time-to-live expiration
+  tidesdb_vcol                Virtual/generated columns
+  tidesdb_write_pressure      oltp_write_only pressure (multi-connection)
 
 Run with verbose output:
 
diff --git a/mysql-test/suite/tidesdb/r/tidesdb_tpcc_contention.result b/mysql-test/suite/tidesdb/r/tidesdb_tpcc_contention.result
new file mode 100644
index 0000000..bbb50d2
--- /dev/null
+++ b/mysql-test/suite/tidesdb/r/tidesdb_tpcc_contention.result
@@ -0,0 +1,108 @@
+#
+# === Setup: TPC-C district table (simplified) ===
+#
+CREATE TABLE district (
+d_w_id INT NOT NULL,
+d_id   INT NOT NULL,
+d_next_o_id INT NOT NULL,
+d_tax  DECIMAL(4,4),
+PRIMARY KEY (d_w_id, d_id)
+) ENGINE=TIDESDB;
+INSERT INTO district VALUES (1, 1, 3001, 0.1000);
+CREATE TABLE orders (
+o_id   INT NOT NULL,
+o_w_id INT NOT NULL,
+o_d_id INT NOT NULL,
+o_c_id INT NOT NULL,
+PRIMARY KEY (o_w_id, o_d_id, o_id)
+) ENGINE=TIDESDB;
+CREATE TABLE new_order (
+no_w_id INT NOT NULL,
+no_d_id INT NOT NULL,
+no_o_id INT NOT NULL,
+PRIMARY KEY (no_w_id, no_d_id, no_o_id)
+) ENGINE=TIDESDB;
+#
+# === TEST 1: Single-session NEWORD (baseline) ===
+#
+BEGIN;
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1 FOR UPDATE;
+d_next_o_id
+3001
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+INSERT INTO orders VALUES (3001, 1, 1, 42);
+INSERT INTO new_order VALUES (1, 1, 3001);
+COMMIT;
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+d_next_o_id
+3002
+#
+# === TEST 2: Two concurrent UPDATEs on same district row ===
+# With pessimistic_locking=ON, the second UPDATE blocks on the
+# row lock until the first commits.  Both succeed, counter
+# increments by 2 with no conflicts and no lost updates.
+#
+connect  connA, localhost, root,,;
+connect  connB, localhost, root,,;
+connection connA;
+BEGIN;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection connB;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection connA;
+COMMIT;
+connection connB;
+connection default;
+# Both UPDATEs succeeded: 3002 + 2 = 3004
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+d_next_o_id
+3004
+#
+# === TEST 3: Serial counter increment (10 iterations) ===
+# Verify the counter works correctly when serialized.
+#
+# Should be initial(3004) + 10 = 3014
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+d_next_o_id
+3014
+#
+# === TEST 4: 4 concurrent autocommit UPDATEs on same row ===
+# With pessimistic_locking=ON, all 4 serialize through the row lock.
+# Counter should advance by exactly 4.
+#
+UPDATE district SET d_next_o_id = 5001 WHERE d_w_id=1 AND d_id=1;
+connect  storm1, localhost, root,,;
+connect  storm2, localhost, root,,;
+connect  storm3, localhost, root,,;
+connect  storm4, localhost, root,,;
+connection storm1;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm2;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm3;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm4;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm1;
+connection storm2;
+connection storm3;
+connection storm4;
+connection default;
+# All 4 UPDATEs succeeded through serialized row locks: 5001 + 4 = 5005
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+d_next_o_id
+5005
+#
+# === Cleanup ===
+#
+disconnect connA;
+disconnect connB;
+disconnect storm1;
+disconnect storm2;
+disconnect storm3;
+disconnect storm4;
+connection default;
+DROP TABLE district;
+DROP TABLE orders;
+DROP TABLE new_order;
+# Done.
diff --git a/mysql-test/suite/tidesdb/t/tidesdb_tpcc_contention.opt b/mysql-test/suite/tidesdb/t/tidesdb_tpcc_contention.opt
new file mode 100644
index 0000000..917be65
--- /dev/null
+++ b/mysql-test/suite/tidesdb/t/tidesdb_tpcc_contention.opt
@@ -0,0 +1 @@
+--loose-tidesdb-pessimistic-locking=ON
diff --git a/mysql-test/suite/tidesdb/t/tidesdb_tpcc_contention.test b/mysql-test/suite/tidesdb/t/tidesdb_tpcc_contention.test
new file mode 100644
index 0000000..20466d6
--- /dev/null
+++ b/mysql-test/suite/tidesdb/t/tidesdb_tpcc_contention.test
@@ -0,0 +1,164 @@
+--source include/have_tidesdb.inc
+#
+# TidesDB TPC-C contention test -- reproduces the exact NEWORD district
+# counter read-modify-write pattern that causes 0 NOPM in HammerDB.
+#
+# The district row is a serial bottleneck: every New Order transaction
+# must SELECT d_next_o_id FOR UPDATE, then UPDATE d_next_o_id + 1.
+# With InnoDB this serializes via row locks. With TidesDB's optimistic
+# MVCC, two concurrent transactions both read the same value, both
+# write the incremented value, and the second to commit fails with
+# TDB_ERR_CONFLICT (mapped to ER_LOCK_DEADLOCK / ER_ERROR_DURING_COMMIT).
+#
+# This test verifies that concurrent counter increments produce correct
+# results without lost updates or permanent failures.
+#
+
+--echo #
+--echo # === Setup: TPC-C district table (simplified) ===
+--echo #
+
+CREATE TABLE district (
+  d_w_id INT NOT NULL,
+  d_id   INT NOT NULL,
+  d_next_o_id INT NOT NULL,
+  d_tax  DECIMAL(4,4),
+  PRIMARY KEY (d_w_id, d_id)
+) ENGINE=TIDESDB;
+
+INSERT INTO district VALUES (1, 1, 3001, 0.1000);
+
+CREATE TABLE orders (
+  o_id   INT NOT NULL,
+  o_w_id INT NOT NULL,
+  o_d_id INT NOT NULL,
+  o_c_id INT NOT NULL,
+  PRIMARY KEY (o_w_id, o_d_id, o_id)
+) ENGINE=TIDESDB;
+
+CREATE TABLE new_order (
+  no_w_id INT NOT NULL,
+  no_d_id INT NOT NULL,
+  no_o_id INT NOT NULL,
+  PRIMARY KEY (no_w_id, no_d_id, no_o_id)
+) ENGINE=TIDESDB;
+
+--echo #
+--echo # === TEST 1: Single-session NEWORD (baseline) ===
+--echo #
+
+BEGIN;
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1 FOR UPDATE;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+INSERT INTO orders VALUES (3001, 1, 1, 42);
+INSERT INTO new_order VALUES (1, 1, 3001);
+COMMIT;
+
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+
+--echo #
+--echo # === TEST 2: Two concurrent UPDATEs on same district row ===
+--echo # With pessimistic_locking=ON, the second UPDATE blocks on the
+--echo # row lock until the first commits.  Both succeed, counter
+--echo # increments by 2 with no conflicts and no lost updates.
+--echo #
+
+connect (connA, localhost, root,,);
+connect (connB, localhost, root,,);
+
+# Connection A: UPDATE district (acquires row lock, held until COMMIT)
+connection connA;
+BEGIN;
+UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+
+# Connection B: send UPDATE async -- will block on row lock until A commits
+connection connB;
+send UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+
+# Connection A: commit -- releases row lock, unblocks B
+connection connA;
+COMMIT;
+
+# Connection B: reap -- should succeed now that A released the lock
+connection connB;
+reap;
+
+# Check results
+connection default;
+--echo # Both UPDATEs succeeded: 3002 + 2 = 3004
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+
+--echo #
+--echo # === TEST 3: Serial counter increment (10 iterations) ===
+--echo # Verify the counter works correctly when serialized.
+--echo #
+
+--disable_query_log
+let $i= 0;
+while ($i < 10)
+{
+  BEGIN;
+  eval UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+  COMMIT;
+  inc $i;
+}
+--enable_query_log
+
+--echo # Should be initial(3004) + 10 = 3014
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+
+--echo #
+--echo # === TEST 4: 4 concurrent autocommit UPDATEs on same row ===
+--echo # With pessimistic_locking=ON, all 4 serialize through the row lock.
+--echo # Counter should advance by exactly 4.
+--echo #
+
+# Reset counter
+UPDATE district SET d_next_o_id = 5001 WHERE d_w_id=1 AND d_id=1;
+
+connect (storm1, localhost, root,,);
+connect (storm2, localhost, root,,);
+connect (storm3, localhost, root,,);
+connect (storm4, localhost, root,,);
+
+# Each connection does 5 serial increments (autocommit)
+connection storm1;
+send UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm2;
+send UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm3;
+send UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+connection storm4;
+send UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id=1 AND d_id=1;
+
+connection storm1;
+reap;
+connection storm2;
+reap;
+connection storm3;
+reap;
+connection storm4;
+reap;
+
+connection default;
+--echo # All 4 UPDATEs succeeded through serialized row locks: 5001 + 4 = 5005
+SELECT d_next_o_id FROM district WHERE d_w_id=1 AND d_id=1;
+
+--echo #
+--echo # === Cleanup ===
+--echo #
+
+disconnect connA;
+disconnect connB;
+disconnect storm1;
+disconnect storm2;
+disconnect storm3;
+disconnect storm4;
+
+connection default;
+DROP TABLE district;
+DROP TABLE orders;
+DROP TABLE new_order;
+
+--source suite/tidesdb/include/cleanup_tidesdb.inc
+--echo # Done.
diff --git a/tidesdb/ha_tidesdb.cc b/tidesdb/ha_tidesdb.cc
index 325558e..8eb6a6d 100644
--- a/tidesdb/ha_tidesdb.cc
+++ b/tidesdb/ha_tidesdb.cc
@@ -16,6 +16,12 @@
 */
 #include "ha_tidesdb.h"
 
+extern "C"
+{
+#define XXH_INLINE_ALL
+#include <tidesdb/xxhash.h>
+}
+
 #include <mysql/plugin.h>
 
 #include <cstring>
@@ -38,6 +44,7 @@
 
 /* Forward-declared for tdb_rc_to_ha(); defined with sysvars below */
 static my_bool srv_print_all_conflicts = 0;
+static my_bool srv_pessimistic_locking = 0;
 static mysql_mutex_t last_conflict_mutex;
 static char last_conflict_info[1024] = "";
 
@@ -96,6 +103,54 @@ static std::string tdb_path;
 
 static handlerton *tidesdb_hton;
 
+/* ******************** Plugin-level row locks ******************** */
+/*
+  Striped mutex array for emulating pessimistic row-level locking.
+  TidesDB library uses optimistic MVCC only -- write-write conflicts
+  are detected at commit time.  For workloads that depend on
+  SELECT ... FOR UPDATE semantics (e.g. TPC-C's district counter),
+  concurrent read-modify-write on the same row always conflicts.
+
+  This stripe array provides lightweight row-level serialization:
+  UPDATE/DELETE acquire the stripe mutex for their PK hash before
+  modifying the row.  The lock is held until COMMIT/ROLLBACK,
+  preventing a second transaction from reading a stale value and
+  creating a doomed write-write conflict.
+
+  The stripe count is configurable via tidesdb_row_lock_stripes (default 1024,
+  read-only).  1024 stripes provide sufficient parallelism (TPC-C with 1000
+  warehouses x 10 districts = 10000 rows spread across 1024 slots).
+  False sharing between different rows hashing to the same slot
+  causes brief serialization but not correctness issues.
+*/
+static ulong srv_row_lock_stripes = 1024;
+static mysql_mutex_t *row_lock_stripes = NULL;
+
+static inline uint32_t row_lock_stripe(const uchar *key, uint len)
+{
+    /* XXH3_64bits from TidesDB's bundled xxHash -- faster and better
+       distribution than FNV-1a, especially for short keys (PK bytes). */
+    uint64_t h = XXH3_64bits(key, len);
+    return (uint32_t)(h % srv_row_lock_stripes);
+}
+
+static void row_lock_acquire(tidesdb_trx_t *trx, const uchar *key, uint len)
+{
+    if (!row_lock_stripes) return; /* allocation failed at init */
+    uint32_t stripe = row_lock_stripe(key, len);
+    if (!trx->held_row_locks) trx->held_row_locks = new std::unordered_set<uint32_t>();
+    if (trx->held_row_locks->count(stripe)) return; /* already hold this stripe */
+    mysql_mutex_lock(&row_lock_stripes[stripe]);
+    trx->held_row_locks->insert(stripe);
+}
+
+static void row_locks_release_all(tidesdb_trx_t *trx)
+{
+    if (!trx->held_row_locks) return;
+    for (uint32_t s : *trx->held_row_locks) mysql_mutex_unlock(&row_lock_stripes[s]);
+    trx->held_row_locks->clear();
+}
+
 static handler *tidesdb_create_handler(handlerton *hton, TABLE_SHARE *table, MEM_ROOT *mem_root);
 
 /* File extensions -- TidesDB manages its own files */
@@ -258,6 +313,24 @@ static MYSQL_SYSVAR_BOOL(print_all_conflicts, srv_print_all_conflicts, PLUGIN_VA
                          "(similar to innodb_print_all_deadlocks)",
                          NULL, NULL, 0);
 
+static MYSQL_SYSVAR_BOOL(pessimistic_locking, srv_pessimistic_locking, PLUGIN_VAR_RQCMDARG,
+                         "Enable plugin-level row locks for UPDATE/DELETE. "
+                         "OFF (default): pure optimistic MVCC -- concurrent writers on the "
+                         "same row are detected at COMMIT time (TDB_ERR_CONFLICT). "
+                         "ON: writers acquire a striped mutex per PK hash, serializing "
+                         "concurrent access to the same row like InnoDB's row locks. "
+                         "Enable for TPC-C or legacy workloads that depend on "
+                         "SELECT ... FOR UPDATE semantics",
+                         NULL, NULL, 0);
+
+static MYSQL_SYSVAR_ULONG(row_lock_stripes, srv_row_lock_stripes,
+                          PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
+                          "Number of striped mutexes for pessimistic row locking. "
+                          "Higher values reduce false contention between unrelated rows "
+                          "at the cost of memory. Only meaningful when "
+                          "tidesdb_pessimistic_locking=ON",
+                          NULL, NULL, 1024, 64, 65536, 0);
+
 static MYSQL_SYSVAR_ULONGLONG(block_cache_size, srv_block_cache_size,
                               PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
                               "TidesDB global block cache size in bytes", NULL, NULL,
@@ -432,6 +505,8 @@ static struct st_mysql_sys_var *tidesdb_system_variables[] = {
     MYSQL_SYSVAR(backup_dir),
     MYSQL_SYSVAR(checkpoint_dir),
     MYSQL_SYSVAR(print_all_conflicts),
+    MYSQL_SYSVAR(pessimistic_locking),
+    MYSQL_SYSVAR(row_lock_stripes),
     MYSQL_SYSVAR(data_home_dir),
     MYSQL_SYSVAR(ttl),
     MYSQL_SYSVAR(skip_unique_check),
@@ -889,6 +964,7 @@ static int tidesdb_commit(handlerton *, THD *thd, bool all)
             trx->txn_generation++;
             trx->dirty = false;
             trx->stmt_savepoint_active = false;
+            row_locks_release_all(trx);
             return tdb_rc_to_ha(rc, "hton_commit");
         }
         tidesdb_txn_free(trx->txn);
@@ -905,6 +981,7 @@ static int tidesdb_commit(handlerton *, THD *thd, bool all)
     }
     trx->dirty = false;
     trx->stmt_savepoint_active = false;
+    row_locks_release_all(trx);
     return 0;
 }
 
@@ -952,6 +1029,7 @@ static int tidesdb_rollback(handlerton *, THD *thd, bool all)
     trx->txn_generation++;
     trx->dirty = false;
     trx->stmt_savepoint_active = false;
+    row_locks_release_all(trx);
     return 0;
 }
 
@@ -964,6 +1042,9 @@ static int tidesdb_close_connection(handlerton *, THD *thd)
     tidesdb_trx_t *trx = (tidesdb_trx_t *)thd_get_ha_data(thd, tidesdb_hton);
     if (trx)
     {
+        row_locks_release_all(trx);
+        delete trx->held_row_locks;
+        trx->held_row_locks = NULL;
         if (trx->txn)
         {
             tidesdb_txn_rollback(trx->txn);
@@ -1113,6 +1194,15 @@ static int tidesdb_init_func(void *p)
 
     mysql_mutex_init(0, &last_conflict_mutex, MY_MUTEX_INIT_FAST);
 
+    /* Initialize striped row lock mutexes for pessimistic locking mode */
+    row_lock_stripes = (mysql_mutex_t *)my_malloc(
+        PSI_NOT_INSTRUMENTED, srv_row_lock_stripes * sizeof(mysql_mutex_t), MYF(MY_ZEROFILL));
+    if (row_lock_stripes)
+    {
+        for (ulong i = 0; i < srv_row_lock_stripes; i++)
+            mysql_mutex_init(0, &row_lock_stripes[i], MY_MUTEX_INIT_FAST);
+    }
+
     /* Use tidesdb_data_home_dir if set, otherwise compute
        a sibling directory of the MariaDB data directory. */
     if (srv_data_home_dir && srv_data_home_dir[0])
@@ -1169,6 +1259,12 @@ static int tidesdb_deinit_func(void *p)
     }
 
     mysql_mutex_destroy(&last_conflict_mutex);
+    if (row_lock_stripes)
+    {
+        for (ulong i = 0; i < srv_row_lock_stripes; i++) mysql_mutex_destroy(&row_lock_stripes[i]);
+        my_free(row_lock_stripes);
+        row_lock_stripes = NULL;
+    }
 
     sql_print_information("TIDESDB: TidesDB closed");
     DBUG_RETURN(0);
@@ -2911,6 +3007,16 @@ int ha_tidesdb::index_read_map(uchar *buf, const uchar *key, key_part_map keypar
             uint full_pk_comp_len = share->idx_comp_key_len[share->pk_index];
             if (comp_len >= full_pk_comp_len)
             {
+                /* When pessimistic locking is ON and this is a write statement
+                   (UPDATE/DELETE/SELECT FOR UPDATE), acquire the row lock BEFORE
+                   reading the row.  This serializes the entire read-modify-write
+                   cycle on the same PK, preventing lost updates. */
+                if (unlikely(srv_pessimistic_locking) && stmt_has_write_lock_)
+                {
+                    tidesdb_trx_t *trx = (tidesdb_trx_t *)thd_get_ha_data(ha_thd(), ht);
+                    if (trx) row_lock_acquire(trx, comp_key, comp_len);
+                }
+
                 /* Full PK match -- point lookup only, no iterator needed.
                    If index_next is called later, ensure_scan_iter will create it. */
                 int ret = fetch_row_by_pk(scan_txn, comp_key, comp_len, buf);
@@ -3406,6 +3512,15 @@ int ha_tidesdb::update_row(const uchar *old_data, const uchar *new_data)
     uint old_pk_len = current_pk_len_;
     memcpy(old_pk, current_pk_buf_, old_pk_len);
 
+    /* When pessimistic locking is enabled, acquire plugin-level row lock
+       on old PK to serialize concurrent UPDATE on the same row.
+       Held until COMMIT/ROLLBACK.  OFF by default -- pure optimistic MVCC. */
+    if (unlikely(srv_pessimistic_locking))
+    {
+        tidesdb_trx_t *trx = (tidesdb_trx_t *)thd_get_ha_data(ha_thd(), ht);
+        if (trx) row_lock_acquire(trx, old_pk, old_pk_len);
+    }
+
     /* new_pk uses its own stack buffer so it survives the current_pk_buf_
        manipulations in the secondary index loop (avoids overlapping memcpy UB) */
     uchar new_pk[MAX_KEY_LENGTH];
@@ -3530,6 +3645,13 @@ int ha_tidesdb::delete_row(const uchar *buf)
 
     MY_BITMAP *old_map = tmp_use_all_columns(table, &table->read_set);
 
+    /* When pessimistic locking is enabled, acquire row lock before delete */
+    if (unlikely(srv_pessimistic_locking))
+    {
+        tidesdb_trx_t *trx = (tidesdb_trx_t *)thd_get_ha_data(ha_thd(), ht);
+        if (trx) row_lock_acquire(trx, current_pk_buf_, current_pk_len_);
+    }
+
     {
         int erc = ensure_stmt_txn();
         if (erc)
@@ -4306,13 +4428,25 @@ int ha_tidesdb::external_lock(THD *thd, int lock_type)
             sql_cmd == SQLCOM_OPTIMIZE || sql_cmd == SQLCOM_CREATE_TABLE ||
             sql_cmd == SQLCOM_DROP_TABLE)
             effective_iso = TDB_ISOLATION_READ_COMMITTED;
+        else if (unlikely(srv_pessimistic_locking))
+            /* When pessimistic locking is ON, the plugin's row-lock mutexes
+               already serialize concurrent access to the same row.  SNAPSHOT
+               conflict detection is redundant and causes spurious commit
+               failures on rows that were safely serialized by the mutex.
+               Use READ_COMMITTED so the library skips conflict checks. */
+            effective_iso = TDB_ISOLATION_READ_COMMITTED;
         else
             effective_iso = resolve_effective_isolation(thd, share->isolation_level);
         tidesdb_trx_t *trx = get_or_create_trx(thd, ht, effective_iso);
         if (!trx) DBUG_RETURN(HA_ERR_OUT_OF_MEM);
 
+        DBUG_PRINT("tidesdb_debug",
+                   ("external_lock: iso=%d txn=%p gen=%lu dirty=%d", (int)effective_iso, trx->txn,
+                    (unsigned long)trx->txn_generation, trx->dirty));
+
         stmt_txn = trx->txn;
         stmt_txn_dirty = false;
+        stmt_has_write_lock_ = (lock_type == F_WRLCK);
 
         /* We register at statement level (always) */
         trans_register_ha(thd, false, ht, 0);
@@ -5069,8 +5203,8 @@ maria_declare_plugin(tidesdb){
     PLUGIN_LICENSE_GPL,
     tidesdb_init_func,
     tidesdb_deinit_func,
-    0x30601,
+    0x30700,
     NULL,
     tidesdb_system_variables,
-    "3.6.1",
+    "3.7.0",
     MariaDB_PLUGIN_MATURITY_GAMMA} maria_declare_plugin_end;
diff --git a/tidesdb/ha_tidesdb.h b/tidesdb/ha_tidesdb.h
index fab5c04..b05eb0d 100644
--- a/tidesdb/ha_tidesdb.h
+++ b/tidesdb/ha_tidesdb.h
@@ -19,6 +19,7 @@
 #include <atomic>
 #include <mutex>
 #include <string>
+#include <unordered_set>
 #include <vector>
 
 #include "handler.h"
@@ -179,6 +180,13 @@ struct tidesdb_trx_t
     bool stmt_was_dirty;                       /* true if current stmt had writes */
     tidesdb_isolation_level_t isolation_level; /* from first table opened */
     uint64_t txn_generation; /* monotonic counter; incremented each time a new txn is created */
+
+    /* Plugin-level row locks -- striped mutex indices held by this txn.
+       Acquired during SELECT FOR UPDATE / UPDATE / DELETE on hot rows,
+       released on commit/rollback.  This emulates pessimistic row locking
+       for workloads like TPC-C that require read-modify-write serialization
+       on the same key (TidesDB library uses optimistic MVCC only). */
+    std::unordered_set<uint32_t> *held_row_locks; /* heap-allocated, NULL until first lock */
 };
 
 /*
@@ -258,6 +266,10 @@ class ha_tidesdb : public handler
     bool cached_skip_unique_;
     bool cached_thdvars_valid_;
 
+    /* Write-lock mode -- set when external_lock(F_WRLCK) is called.
+       Used to decide whether to acquire row locks in index_read_map. */
+    bool stmt_has_write_lock_;
+
     /* Bulk insert state */
     bool in_bulk_insert_;
     ha_rows bulk_insert_ops_; /* ops buffered since last mid-txn commit */