transpiler/cdc: wire schema-aware upsert conflicts and remove forced WAL checkpoints

This commit completes the in-flight replication/transpiler fixes and cleans up CDC apply behavior.

Transpiler and parse pipeline:
- Add ParseOptions.SchemaProvider and thread it through parser/query context.
- Refactor InsertOnDuplicateKeyRule to return conflict metadata instead of mutating serializer state.
- Make SQLite serializer stateless for conflict targets via SerializeWithOpts.
- Consolidate transpiler flow to single final serialization, qualifier stripping, and literal extraction ordering.
- Preserve schema-derived ON CONFLICT targets when literal extraction is enabled.
- Add/adjust tests for literal extraction + ON DUPLICATE KEY, serializer behavior, and ANSI-quote/backslash handling.
- Tighten auto-increment injection behavior to skip unsafe injection when INSERT has no explicit column list.

Runtime wiring for P1:
- Extend coordinator DatabaseManager contract with GetTranspilerSchema.
- Implement GetTranspilerSchema in db.DatabaseManager using cached schema.
- Wire schema provider into coordinator and read-only replica ParseStatementWithOptions paths.
- Filter out SchemaCache rowid sentinel from conflict targeting so fallback behavior remains correct when no explicit PK exists.
- Update coordinator-facing test database manager mocks and add focused DB unit tests for GetTranspilerSchema.

CDC / replication visibility and P2 rollback:
- Keep shared-cache DSN usage across write/read/reopen and batch committer DB connections.
- Ensure snapshot/read query paths use readDB.
- Remove synchronous PRAGMA wal_checkpoint(PASSIVE) calls from CDC apply commit and batch flush commit hot paths.
- Retain adaptive/background checkpoint behavior where applicable.

Functional impact:
- INSERT ... ON DUPLICATE KEY now resolves ON CONFLICT targets from real schema at runtime (not parser-local fallback only).
- Literal extraction no longer drops schema-derived conflict targets.
- Replica local reads stay local while forwarded writes can still opt into visibility guarantees via wait-for-replication semantics, without per-commit forced checkpoint overhead.

Author: Zohaib Sibte Hassan

coordinator/handler.go

@@ -34,6 +34,9 @@ type DatabaseManager interface {
 	// GetAutoIncrementColumn returns the auto-increment column name for a table.
 	// Uses cached schema - does NOT query SQLite PRAGMA.
 	GetAutoIncrementColumn(database, table string) (string, error)
+	// GetTranspilerSchema returns schema information for transpiler conflict resolution.
+	// Uses cached schema - does NOT query SQLite PRAGMA.
+	GetTranspilerSchema(database, table string) (*transform.SchemaInfo, error)
 }
 
 // ReplicatedDatabaseProvider provides access to replicated database operations
@@ -213,7 +216,7 @@ func (h *CoordinatorHandler) HandleQuery(session *protocol.ConnectionSession, sq
 	// Build schema lookup function for auto-increment ID injection.
 	// Uses cached schema via DatabaseManager - does NOT query SQLite PRAGMA.
 	var schemaLookup protocol.SchemaLookupFunc
-	if session.CurrentDatabase != "" {
+	if h.dbManager != nil && session.CurrentDatabase != "" {
 		dbName := session.CurrentDatabase
 		schemaLookup = func(table string) string {
 			col, err := h.dbManager.GetAutoIncrementColumn(dbName, table)
@@ -223,10 +226,29 @@ func (h *CoordinatorHandler) HandleQuery(session *protocol.ConnectionSession, sq
 			return col
 		}
 	}
+	schemaProvider := func(database, table string) *transform.SchemaInfo {
+		if table == "" || h.dbManager == nil {
+			return nil
+		}
+		dbName := database
+		if dbName == "" {
+			dbName = session.CurrentDatabase
+		}
+		if dbName == "" {
+			return nil
+		}
+
+		schemaInfo, err := h.dbManager.GetTranspilerSchema(dbName, table)
+		if err != nil {
+			return nil
+		}
+		return schemaInfo
+	}
 
 	// Parse with options based on session state
 	stmt := protocol.ParseStatementWithOptions(sql, protocol.ParseOptions{
 		SchemaLookup:      schemaLookup,
+		SchemaProvider:    schemaProvider,
 		SkipTranspilation: !session.TranspilationEnabled,
 		ExtractLiterals:   true, // Enable literal extraction for parameterized execution
 	})

db/database_manager.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/maxpert/marmot/coordinator"
 	"github.com/maxpert/marmot/hlc"
+	"github.com/maxpert/marmot/protocol/query/transform"
 	"github.com/maxpert/marmot/publisher"
 	"github.com/rs/zerolog/log"
 )
@@ -1291,6 +1292,33 @@ func (dm *DatabaseManager) GetAutoIncrementColumn(database, table string) (strin
 	return schema.GetAutoIncrementCol(), nil
 }
 
+// GetTranspilerSchema returns schema information used by SQL transpilation rules.
+// Uses cached schema - does NOT query SQLite PRAGMA.
+func (dm *DatabaseManager) GetTranspilerSchema(database, table string) (*transform.SchemaInfo, error) {
+	db, err := dm.GetDatabase(database)
+	if err != nil {
+		return nil, err
+	}
+
+	schema, err := db.GetCachedTableSchema(table)
+	if err != nil {
+		return nil, fmt.Errorf("schema not cached for table %s: %w", table, err)
+	}
+
+	info := &transform.SchemaInfo{
+		AutoIncrementColumn: schema.GetAutoIncrementCol(),
+	}
+
+	// PrimaryKeys uses "rowid" sentinel when no explicit PRIMARY KEY is defined.
+	// Keep it out of transpiler conflict targeting to preserve fallback behavior.
+	if len(schema.PrimaryKeys) > 0 &&
+		!(len(schema.PrimaryKeys) == 1 && strings.EqualFold(schema.PrimaryKeys[0], "rowid")) {
+		info.PrimaryKey = append([]string(nil), schema.PrimaryKeys...)
+	}
+
+	return info, nil
+}
+
 // GetCommittedTxnCount returns the count of committed transactions in a database
 // This is used by anti-entropy to compare data completeness between nodes
 func (dm *DatabaseManager) GetCommittedTxnCount(database string) (int64, error) {

db/database_manager_transpiler_schema_test.go

@@ -0,0 +1,74 @@
+//go:build sqlite_preupdate_hook
+// +build sqlite_preupdate_hook
+
+package db
+
+import "testing"
+
+func TestGetTranspilerSchema_PrimaryKey(t *testing.T) {
+	dm, _ := setupTestDatabaseManager(t)
+	defer dm.Close()
+
+	if err := dm.CreateDatabase("testdb"); err != nil {
+		t.Fatalf("CreateDatabase failed: %v", err)
+	}
+
+	mdb, err := dm.GetDatabase("testdb")
+	if err != nil {
+		t.Fatalf("GetDatabase failed: %v", err)
+	}
+
+	cache, ok := mdb.GetSchemaCache().(*SchemaCache)
+	if !ok || cache == nil {
+		t.Fatal("schema cache is unavailable")
+	}
+
+	cache.Update("users", &TableSchema{
+		PrimaryKeys:      []string{"tenant_id", "user_id"},
+		AutoIncrementCol: "user_id",
+	})
+
+	info, err := dm.GetTranspilerSchema("testdb", "users")
+	if err != nil {
+		t.Fatalf("GetTranspilerSchema failed: %v", err)
+	}
+
+	if len(info.PrimaryKey) != 2 || info.PrimaryKey[0] != "tenant_id" || info.PrimaryKey[1] != "user_id" {
+		t.Fatalf("unexpected primary key: %#v", info.PrimaryKey)
+	}
+	if info.AutoIncrementColumn != "user_id" {
+		t.Fatalf("unexpected auto-increment column: %q", info.AutoIncrementColumn)
+	}
+}
+
+func TestGetTranspilerSchema_IgnoresRowIDSentinel(t *testing.T) {
+	dm, _ := setupTestDatabaseManager(t)
+	defer dm.Close()
+
+	if err := dm.CreateDatabase("testdb"); err != nil {
+		t.Fatalf("CreateDatabase failed: %v", err)
+	}
+
+	mdb, err := dm.GetDatabase("testdb")
+	if err != nil {
+		t.Fatalf("GetDatabase failed: %v", err)
+	}
+
+	cache, ok := mdb.GetSchemaCache().(*SchemaCache)
+	if !ok || cache == nil {
+		t.Fatal("schema cache is unavailable")
+	}
+
+	cache.Update("logs", &TableSchema{
+		PrimaryKeys: []string{"rowid"},
+	})
+
+	info, err := dm.GetTranspilerSchema("testdb", "logs")
+	if err != nil {
+		t.Fatalf("GetTranspilerSchema failed: %v", err)
+	}
+
+	if len(info.PrimaryKey) != 0 {
+		t.Fatalf("expected empty primary key, got %#v", info.PrimaryKey)
+	}
+}

db/db_integration.go

@@ -70,12 +70,13 @@ func NewReplicatedDatabase(dbPath string, nodeID uint64, clock *hlc.Clock, metaS
 	// === WRITE CONNECTION ===
 	// Single connection for all writes (SQLite allows only one writer at a time)
 	// Uses _txlock=immediate to acquire write lock at BEGIN, avoiding deadlocks
+	// Uses cache=shared so all local connections share a single page cache.
 	writeDSN := dbPath
 	if !isMemoryDB {
 		if strings.Contains(writeDSN, "?") {
-			writeDSN += fmt.Sprintf("&_journal_mode=WAL&_busy_timeout=%d&_txlock=immediate", busyTimeoutMS)
+			writeDSN += fmt.Sprintf("&_journal_mode=WAL&_busy_timeout=%d&_txlock=immediate&cache=shared", busyTimeoutMS)
 		} else {
-			writeDSN += fmt.Sprintf("?_journal_mode=WAL&_busy_timeout=%d&_txlock=immediate", busyTimeoutMS)
+			writeDSN += fmt.Sprintf("?_journal_mode=WAL&_busy_timeout=%d&_txlock=immediate&cache=shared", busyTimeoutMS)
 		}
 	}
 
@@ -108,12 +109,13 @@ func NewReplicatedDatabase(dbPath string, nodeID uint64, clock *hlc.Clock, metaS
 	// Multiple connections for concurrent reads (WAL mode supports this)
 	// No _txlock needed for reads - they don't acquire write locks
 	// Note: Don't use mode=ro as it can interfere with WAL checkpointing
+	// Uses cache=shared so all local connections share a single page cache.
 	readDSN := dbPath
 	if !isMemoryDB {
 		if strings.Contains(readDSN, "?") {
-			readDSN += fmt.Sprintf("&_journal_mode=WAL&_busy_timeout=%d", busyTimeoutMS)
+			readDSN += fmt.Sprintf("&_journal_mode=WAL&_busy_timeout=%d&cache=shared", busyTimeoutMS)
 		} else {
-			readDSN += fmt.Sprintf("?_journal_mode=WAL&_busy_timeout=%d", busyTimeoutMS)
+			readDSN += fmt.Sprintf("?_journal_mode=WAL&_busy_timeout=%d&cache=shared", busyTimeoutMS)
 		}
 	}
 
@@ -357,7 +359,7 @@ func (mdb *ReplicatedDatabase) OpenSQLiteConnections(dbPath string) error {
 	poolCfg := cfg.Config.ConnectionPool
 
 	// Open write connection
-	writeDSN := fmt.Sprintf("%s?_journal_mode=WAL&_busy_timeout=%d&_txlock=immediate", dbPath, busyTimeoutMS)
+	writeDSN := fmt.Sprintf("%s?_journal_mode=WAL&_busy_timeout=%d&_txlock=immediate&cache=shared", dbPath, busyTimeoutMS)
 	writeDB, err := sql.Open(SQLiteDriverName, writeDSN)
 	if err != nil {
 		return fmt.Errorf("failed to open write connection: %w", err)
@@ -383,7 +385,7 @@ func (mdb *ReplicatedDatabase) OpenSQLiteConnections(dbPath string) error {
 	hookDB.SetConnMaxLifetime(0)
 
 	// Open read connection pool
-	readDSN := fmt.Sprintf("%s?_journal_mode=WAL&_busy_timeout=%d", dbPath, busyTimeoutMS)
+	readDSN := fmt.Sprintf("%s?_journal_mode=WAL&_busy_timeout=%d&cache=shared", dbPath, busyTimeoutMS)
 	readDB, err := sql.Open(SQLiteDriverName, readDSN)
 	if err != nil {
 		writeDB.Close()
@@ -717,14 +719,12 @@ func (mdb *ReplicatedDatabase) ExecuteTransaction(ctx context.Context, statement
 	return nil
 }
 
-// ExecuteQuery executes a read query with snapshot isolation
-// Uses the read connection pool for concurrent read access
+// ExecuteQuery executes a read query with snapshot isolation.
+// Uses the read connection pool for concurrent read access.
 func (mdb *ReplicatedDatabase) ExecuteQuery(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
 	// Get current snapshot timestamp
 	snapshotTS := mdb.clock.Now()
 
-	// Execute on read connection pool for concurrent reads
-	// SQLite's WAL mode provides snapshot isolation at the connection level
 	rows, err := mdb.readDB.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, fmt.Errorf("query failed: %w", err)
@@ -736,11 +736,9 @@ func (mdb *ReplicatedDatabase) ExecuteQuery(ctx context.Context, query string, a
 	return rows, nil
 }
 
-// ExecuteSnapshotRead executes a read query with full transactional support
-// Uses rqlite/sql AST parser for proper SQL analysis
-// Uses the read connection pool for concurrent read access
+// ExecuteSnapshotRead executes a read query with full transactional support.
+// Uses the read connection pool for concurrent read access.
 func (mdb *ReplicatedDatabase) ExecuteSnapshotRead(ctx context.Context, query string, args ...interface{}) ([]string, []map[string]interface{}, error) {
-	// SQLite WAL mode provides snapshot isolation at connection level
 	rows, err := mdb.readDB.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, nil, err
@@ -776,8 +774,8 @@ func (mdb *ReplicatedDatabase) ExecuteSnapshotRead(ctx context.Context, query st
 	return columns, results, nil
 }
 
-// ExecuteQueryRow executes a single-row query with snapshot isolation
-// Uses the read connection pool for concurrent read access
+// ExecuteQueryRow executes a single-row query with snapshot isolation.
+// Uses the read connection pool for concurrent read access.
 func (mdb *ReplicatedDatabase) ExecuteQueryRow(ctx context.Context, query string, args ...interface{}) *sql.Row {
 	// Get current snapshot timestamp
 	// Note: SQLite's WAL mode provides snapshot isolation at transaction level.

db/sqlite_batch_committer.go

@@ -177,12 +177,13 @@ func (bc *SQLiteBatchCommitter) openOptimizedConnection() (*sql.DB, error) {
 	// Build DSN with batch-optimized settings
 	// WAL mode for compatibility with other connections
 	// _txlock=immediate to acquire write lock at BEGIN
+	// cache=shared ensures writes are immediately visible to other connections
 	dsn := bc.dbPath
 	if !strings.Contains(dsn, ":memory:") {
 		if strings.Contains(dsn, "?") {
-			dsn += "&_journal_mode=WAL&_txlock=immediate"
+			dsn += "&_journal_mode=WAL&_txlock=immediate&cache=shared"
 		} else {
-			dsn += "?_journal_mode=WAL&_txlock=immediate"
+			dsn += "?_journal_mode=WAL&_txlock=immediate&cache=shared"
 		}
 	}
 
@@ -400,7 +401,7 @@ func (bc *SQLiteBatchCommitter) flush(batch map[uint64]*pendingCommit, trigger s
 		return
 	}
 
-	// Adaptive checkpoint
+	// Adaptive checkpoint for larger WAL sizes
 	if bc.checkpointEnabled {
 		walSizeMB := bc.checkWALSize()
 		if walSizeMB >= bc.checkpointPassiveThreshMB {

db/transaction.go

@@ -391,6 +391,7 @@ func (tm *TransactionManager) applyCDCEntries(txnID uint64, entries []*IntentEnt
 	if err := tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit CDC transaction: %w", err)
 	}
+
 	return nil
 }
 

protocol/parser.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/maxpert/marmot/id"
 	"github.com/maxpert/marmot/protocol/query"
+	"github.com/maxpert/marmot/protocol/query/transform"
 	"github.com/rs/zerolog/log"
 )
 
@@ -114,6 +115,7 @@ type SchemaLookupFunc func(table string) string
 // ParseOptions holds options for parsing SQL statements.
 type ParseOptions struct {
 	SchemaLookup      SchemaLookupFunc
+	SchemaProvider    transform.SchemaProvider // For ON CONFLICT target resolution
 	SkipTranspilation bool
 	ExtractLiterals   bool // Enable literal extraction for parameterized execution
 }
@@ -129,6 +131,7 @@ func ParseStatement(sql string) Statement {
 func ParseStatementWithOptions(sql string, opts ParseOptions) Statement {
 	ctx := query.NewContext(sql, nil)
 	ctx.SchemaLookup = opts.SchemaLookup
+	ctx.SchemaProvider = opts.SchemaProvider
 	ctx.SkipTranspilation = opts.SkipTranspilation
 	ctx.ExtractLiterals = opts.ExtractLiterals
 

protocol/query/context.go

@@ -108,9 +108,8 @@ type QueryOutput struct {
 
 // TranspiledStatement represents a single transpiled SQL statement with parameters.
 type TranspiledStatement struct {
-	SQL             string
-	Params          []interface{}
-	RequiresPrepare bool
+	SQL    string
+	Params []interface{}
 }
 
 // MySQLParseState holds MySQL-specific parsing state and metadata.
@@ -128,6 +127,9 @@ type MySQLParseState struct {
 	TableName string
 	// ShowFilter holds the LIKE pattern for SHOW TABLES LIKE queries
 	ShowFilter string
+	// ConflictColumns holds columns for ON CONFLICT clause.
+	// Set by InsertOnDuplicateKeyRule during transpilation.
+	ConflictColumns []string
 }
 
 // QueryContext holds all state for processing a single query through the pipeline.