refactor(db)!: drop mysql/mariadb dialect branches from postgres stores

The flag-day migration to reactive PostgreSQL left dead mysql/mariadb arms behind dialect.name() switches in the per-domain postgres stores. Every branch now collapses to its PostgreSQL arm: ON DUPLICATE KEY upserts, DATE_FORMAT/TIMESTAMPDIFF analytics buckets, JSON_EXTRACT expressions, INSERT IGNORE locks, and the UUID-to-BINARY(16) codec. R2dbcPointValueLatestStore no longer silently falls back to MySQL syntax for any non-postgres dialect name, and stores whose dialect field became unused lose it (TCK constructor updated accordingly).
2026-11-02 19:21:51 +08:00 · 2026-09-19 19:25:24 +08:00
parent c840fbe85f
commit 0f6a4bfa13
15 changed files with 108 additions and 345 deletions
@@ -162,11 +162,8 @@ public class R2dbcMcpRuntimeStore implements ReactiveOAuthMcpStore {
        String base =
                "INSERT INTO dc3_auth.dc3_oauth_authorization_consent (id,registered_client_id,client_id,principal_id,tenant_id,scopes,consent_ext,deleted) VALUES (:id,:registered_client_id,:client_id,:principal_id,:tenant_id,:scopes,"
                        + dialect.jsonWriteExpression(":consent_ext") + ",0) ";
-        String sql = "postgres".equalsIgnoreCase(dialect.name())
-                ? base
-                        + "ON CONFLICT (registered_client_id,principal_id,tenant_id) WHERE deleted=0 DO UPDATE SET client_id=EXCLUDED.client_id,scopes=EXCLUDED.scopes,consent_ext=EXCLUDED.consent_ext,deleted=0,operate_time=CURRENT_TIMESTAMP"
-                : base
-                        + "ON DUPLICATE KEY UPDATE client_id=VALUES(client_id),scopes=VALUES(scopes),consent_ext=VALUES(consent_ext),deleted=0,operate_time=CURRENT_TIMESTAMP";
+        String sql = base
+                + "ON CONFLICT (registered_client_id,principal_id,tenant_id) WHERE deleted=0 DO UPDATE SET client_id=EXCLUDED.client_id,scopes=EXCLUDED.scopes,consent_ext=EXCLUDED.consent_ext,deleted=0,operate_time=CURRENT_TIMESTAMP";
        return client.sql(sql)
                .bind("id", value.getId())
                .bind("registered_client_id", value.getRegisteredClientId())
@@ -549,24 +546,14 @@ public class R2dbcMcpRuntimeStore implements ReactiveOAuthMcpStore {
        String values =
                " VALUES (:id,:tool_id,:tool_name,:tool_title,:tool_category,:service_name,:api_code,:permission_code,:http_method,:api_path,:schema_hash,:risk_level,:read_only_hint,:destructive_hint,:idempotent_hint,:open_world_hint,:enable_flag,:remark,"
                        + dialect.jsonWriteExpression(":tool_ext") + ",0) ";
-        String sql;
-        if ("postgres".equalsIgnoreCase(dialect.name())) {
-            sql = "INSERT INTO " + TOOL + columns + values
-                    + "ON CONFLICT (tool_id) WHERE deleted=0 AND tool_id<>'' DO UPDATE SET "
-                    + "tool_name=EXCLUDED.tool_name,tool_title=EXCLUDED.tool_title,tool_category=EXCLUDED.tool_category,"
-                    + "service_name=EXCLUDED.service_name,api_code=EXCLUDED.api_code,permission_code=EXCLUDED.permission_code,"
-                    + "http_method=EXCLUDED.http_method,api_path=EXCLUDED.api_path,schema_hash=EXCLUDED.schema_hash,"
-                    + "risk_level=EXCLUDED.risk_level,read_only_hint=EXCLUDED.read_only_hint,destructive_hint=EXCLUDED.destructive_hint,"
-                    + "idempotent_hint=EXCLUDED.idempotent_hint,open_world_hint=EXCLUDED.open_world_hint,enable_flag=EXCLUDED.enable_flag,"
-                    + "remark=EXCLUDED.remark,tool_ext=EXCLUDED.tool_ext,deleted=0,operate_time=CURRENT_TIMESTAMP";
-        } else {
-            sql = "INSERT INTO " + TOOL + columns + values
-                    + "ON DUPLICATE KEY UPDATE tool_name=VALUES(tool_name),tool_title=VALUES(tool_title),tool_category=VALUES(tool_category),"
-                    + "service_name=VALUES(service_name),api_code=VALUES(api_code),permission_code=VALUES(permission_code),"
-                    + "http_method=VALUES(http_method),api_path=VALUES(api_path),schema_hash=VALUES(schema_hash),risk_level=VALUES(risk_level),"
-                    + "read_only_hint=VALUES(read_only_hint),destructive_hint=VALUES(destructive_hint),idempotent_hint=VALUES(idempotent_hint),"
-                    + "open_world_hint=VALUES(open_world_hint),enable_flag=VALUES(enable_flag),remark=VALUES(remark),tool_ext=VALUES(tool_ext),deleted=0,operate_time=CURRENT_TIMESTAMP";
-        }
+        String sql = "INSERT INTO " + TOOL + columns + values
+                + "ON CONFLICT (tool_id) WHERE deleted=0 AND tool_id<>'' DO UPDATE SET "
+                + "tool_name=EXCLUDED.tool_name,tool_title=EXCLUDED.tool_title,tool_category=EXCLUDED.tool_category,"
+                + "service_name=EXCLUDED.service_name,api_code=EXCLUDED.api_code,permission_code=EXCLUDED.permission_code,"
+                + "http_method=EXCLUDED.http_method,api_path=EXCLUDED.api_path,schema_hash=EXCLUDED.schema_hash,"
+                + "risk_level=EXCLUDED.risk_level,read_only_hint=EXCLUDED.read_only_hint,destructive_hint=EXCLUDED.destructive_hint,"
+                + "idempotent_hint=EXCLUDED.idempotent_hint,open_world_hint=EXCLUDED.open_world_hint,enable_flag=EXCLUDED.enable_flag,"
+                + "remark=EXCLUDED.remark,tool_ext=EXCLUDED.tool_ext,deleted=0,operate_time=CURRENT_TIMESTAMP";
        return toolMutationStatement(sql, value).fetch().rowsUpdated().map(Long::intValue);
    }

@@ -281,9 +281,8 @@ public class R2dbcResourceRegistryStore implements ReactiveResourceRegistryStore

    @Override
    public Mono<Long> acquireLock(String lockName) {
-        String insert = dialect.name().toLowerCase().contains("postgres")
-                ? "INSERT INTO dc3_platform_lock(lock_name,fencing_token,expires_at) VALUES (:lock_name,0,:expires_at) ON CONFLICT (lock_name) DO NOTHING"
-                : "INSERT IGNORE INTO dc3_platform_lock(lock_name,fencing_token,expires_at) VALUES (:lock_name,0,:expires_at)";
+        String insert =
+                "INSERT INTO dc3_platform_lock(lock_name,fencing_token,expires_at) VALUES (:lock_name,0,:expires_at) ON CONFLICT (lock_name) DO NOTHING";
        Instant expires = Instant.now().plusSeconds(300);
        return databaseClient
                .sql(insert)
@@ -31,13 +31,11 @@ import io.github.pnoker.common.data.entity.bo.dashboard.RecentChangeRow;
 import io.github.pnoker.common.data.entity.bo.dashboard.SourceCountRow;
 import io.github.pnoker.common.data.entity.bo.dashboard.SourceStatsRow;
 import io.github.pnoker.common.data.repository.ReactiveAlertAnalyticsStore;
-import io.github.pnoker.db.r2dbc.core.dialect.R2dbcDialect;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
-import java.util.Locale;
 import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.r2dbc.core.DatabaseClient;
@@ -47,14 +45,13 @@ import reactor.core.publisher.Mono;

 /** Explicit SQL adapter for dashboard alert aggregates. */
@Repository
-@ConditionalOnClass({DatabaseClient.class, R2dbcDialect.class})
+@ConditionalOnClass(DatabaseClient.class)
@RequiredArgsConstructor
 public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    private static final String TABLE = "dc3_data.dc3_entity_alarm";

    private final DatabaseClient databaseClient;
-    private final R2dbcDialect dialect;

    @Override
    public Mono<AlertCountersRow> countAll(long tenantId) {
@@ -74,11 +71,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<BucketRow> countByType(long tenantId) {
-        String typeExpression =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "CAST(alarm_type_flag AS CHAR)";
-                    default -> "CAST(alarm_type_flag AS TEXT)";
-                };
+        String typeExpression = "CAST(alarm_type_flag AS TEXT)";
        return databaseClient
                .sql("SELECT " + typeExpression + " AS bucket_key, COUNT(*) AS count FROM " + TABLE
                        + " WHERE tenant_id=:tenant_id GROUP BY alarm_type_flag ORDER BY count DESC")
@@ -103,11 +96,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<HourCountRow> hourlyCounts(long tenantId, LocalDateTime from) {
-        String bucket =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "DATE_FORMAT(create_time, '%Y-%m-%d %H:00:00')";
-                    default -> "date_trunc('hour', create_time)";
-                };
+        String bucket = "date_trunc('hour', create_time)";
        return databaseClient
                .sql("SELECT " + bucket + " AS bucket, COUNT(*) AS count FROM " + TABLE
                        + " WHERE tenant_id=:tenant_id AND create_time>=:from_time GROUP BY " + bucket
@@ -136,11 +125,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<AlertTrendRow> dailyTrend(long tenantId, LocalDateTime from) {
-        String day =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "DATE(create_time)";
-                    default -> "CAST(create_time AS DATE)";
-                };
+        String day = "CAST(create_time AS DATE)";
        return databaseClient
                .sql("SELECT " + day + " AS day, "
                        + "COALESCE(SUM(CASE WHEN alarm_target_type_flag IN (0,1) THEN 1 ELSE 0 END),0) AS device_count, "
@@ -172,16 +157,8 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<ActivityCellRow> activityHeatmap(long tenantId, LocalDateTime from) {
-        String dow =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "DAYOFWEEK(create_time)-1";
-                    default -> "EXTRACT(DOW FROM create_time)::int";
-                };
-        String hour =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "HOUR(create_time)";
-                    default -> "EXTRACT(HOUR FROM create_time)::int";
-                };
+        String dow = "EXTRACT(DOW FROM create_time)::int";
+        String hour = "EXTRACT(HOUR FROM create_time)::int";
        return databaseClient
                .sql("SELECT " + dow + " AS dow, " + hour + " AS hour, COUNT(*) AS count FROM " + TABLE
                        + " WHERE tenant_id=:tenant_id AND create_time>=:from_time GROUP BY " + dow + ", " + hour)
@@ -199,11 +176,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<BucketRow> typeDistribution(long tenantId, LocalDateTime from) {
-        String type =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "JSON_UNQUOTE(JSON_EXTRACT(alarm_ext, '$.type'))";
-                    default -> "alarm_ext ->> 'type'";
-                };
+        String type = "alarm_ext ->> 'type'";
        return databaseClient
                .sql("SELECT " + type + " AS bucket_key, COUNT(*) AS count FROM " + TABLE
                        + " WHERE tenant_id=:tenant_id AND create_time>=:from_time AND " + type + " IS NOT NULL"
@@ -255,12 +228,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<CorrelationPairRow> correlationPairs(long tenantId, LocalDateTime from, int windowSec, int limit) {
-        String distance =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" ->
-                        "ABS(TIMESTAMPDIFF(MICROSECOND, e2.create_time, e1.create_time)) <= :window_micros";
-                    default -> "ABS(EXTRACT(EPOCH FROM (e1.create_time-e2.create_time))) <= :window_seconds";
-                };
+        String distance = "ABS(EXTRACT(EPOCH FROM (e1.create_time-e2.create_time))) <= :window_seconds";
        String sql = "WITH ev AS (SELECT " + sourceExpression()
                + " AS source, entity_id AS source_id, alarm_type_flag, create_time FROM " + TABLE
                + " WHERE tenant_id=:tenant_id AND create_time>=:from_time) SELECT e1.source AS a_source, e1.source_id AS a_source_id,"
@@ -274,10 +242,8 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {
                .sql(sql)
                .bind("tenant_id", tenantId)
                .bind("from_time", from)
-                .bind("limit", limit);
-        spec = "mysql".equalsIgnoreCase(dialect.name()) || "mariadb".equalsIgnoreCase(dialect.name())
-                ? spec.bind("window_micros", windowSec * 1_000_000L)
-                : spec.bind("window_seconds", windowSec);
+                .bind("limit", limit)
+                .bind("window_seconds", windowSec);
        return spec.map((row, metadata) -> {
                    CorrelationPairRow value = new CorrelationPairRow();
                    value.setASource(text(row.get("a_source")));
@@ -314,11 +280,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Mono<AgingBucketRow> agingBuckets(long tenantId) {
-        String age =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "TIMESTAMPDIFF(SECOND, create_time, CURRENT_TIMESTAMP)";
-                    default -> "EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP-create_time))";
-                };
+        String age = "EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP-create_time))";
        return databaseClient
                .sql("SELECT COALESCE(SUM(CASE WHEN age<3600 THEN 1 ELSE 0 END),0) AS under_1h, "
                        + "COALESCE(SUM(CASE WHEN age>=3600 AND age<21600 THEN 1 ELSE 0 END),0) AS h1_to_6, "
@@ -342,16 +304,8 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore {

    @Override
    public Flux<MttaTrendRow> mttaByDay(long tenantId, LocalDateTime from) {
-        String day =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "DATE(create_time)";
-                    default -> "CAST(create_time AS DATE)";
-                };
-        String latency =
-                switch (dialect.name().toLowerCase(Locale.ROOT)) {
-                    case "mysql", "mariadb" -> "TIMESTAMPDIFF(MICROSECOND, create_time, operate_time)/1000.0";
-                    default -> "EXTRACT(EPOCH FROM (operate_time-create_time))*1000.0";
-                };
+        String day = "CAST(create_time AS DATE)";
+        String latency = "EXTRACT(EPOCH FROM (operate_time-create_time))*1000.0";
        String sql = "WITH ranked AS (SELECT " + day + " AS day, " + latency + " AS latency_ms, "
                + "ROW_NUMBER() OVER (PARTITION BY " + day + " ORDER BY " + latency + ") AS rn, "
                + "COUNT(*) OVER (PARTITION BY " + day + ") AS cnt FROM " + TABLE
@@ -18,7 +18,6 @@ package io.github.pnoker.db.postgres.data;

 import io.github.pnoker.common.data.entity.bo.dashboard.AlertItemRow;
 import io.github.pnoker.common.data.repository.ReactiveAlertStore;
-import io.github.pnoker.db.r2dbc.core.dialect.R2dbcDialect;
 import io.github.pnoker.db.r2dbc.core.page.OffsetPage;
 import io.github.pnoker.db.r2dbc.core.page.PageRequest;
 import io.github.pnoker.db.r2dbc.core.page.SortSpec;
@@ -29,7 +28,6 @@ import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Locale;
 import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.r2dbc.core.DatabaseClient;
@@ -38,7 +36,7 @@ import reactor.core.publisher.Mono;

 /** Explicit SQL adapter for dashboard alert reads and confirmation updates. */
@Repository
-@ConditionalOnClass({DatabaseClient.class, PageTransaction.class, R2dbcDialect.class})
+@ConditionalOnClass({DatabaseClient.class, PageTransaction.class})
@RequiredArgsConstructor
 public class R2dbcAlertStore implements ReactiveAlertStore {

@@ -48,7 +46,6 @@ public class R2dbcAlertStore implements ReactiveAlertStore {

    private final DatabaseClient databaseClient;
    private final PageTransaction pageTransaction;
-    private final R2dbcDialect dialect;

    @Override
    public Mono<OffsetPage<AlertItemRow>> list(
@@ -167,10 +164,7 @@ public class R2dbcAlertStore implements ReactiveAlertStore {
    }

    private String messageExpression() {
-        return switch (dialect.name().toLowerCase(Locale.ROOT)) {
-            case "mysql", "mariadb" -> "COALESCE(JSON_UNQUOTE(JSON_EXTRACT(a.alarm_ext, '$.content')), '')";
-            default -> "COALESCE(a.alarm_ext ->> 'content', '')";
-        };
+        return "COALESCE(a.alarm_ext ->> 'content', '')";
    }

    private int sourceIndex(String source) {
@@ -70,9 +70,7 @@ public class R2dbcEntityAlarmStore implements ReactiveEntityAlarmStore {
                + "(:id,:alarm_target_type_flag,:entity_id,:driver_id,:device_id,:point_id,:rule_id,:rule_state_id,:dedupe_key,"
                + ":alarm_type_flag,:alarm_source_flag,:alarm_level_flag," + dialect.jsonWriteExpression(":alarm_ext")
                + ",:expired_time,:confirm_flag,:tenant_id,:create_time,:operate_time)";
-        String sql = dialect.name().equalsIgnoreCase("postgres")
-                ? insert + " ON CONFLICT (tenant_id,dedupe_key) DO NOTHING"
-                : insert + " ON DUPLICATE KEY UPDATE id=id";
+        String sql = insert + " ON CONFLICT (tenant_id,dedupe_key) DO NOTHING";
        DatabaseClient.GenericExecuteSpec spec = databaseClient
                .sql(sql)
                .bind("id", alarm.getId())
@@ -134,30 +134,19 @@ public class R2dbcEntityStateStore implements ReactiveEntityStateStore {
                + "timeout_source_flag,entity_state_ext,tenant_id,create_time,operate_time) VALUES "
                + "(:id,:entity_type,:entity_id,:parent_entity,:state,:last_state,1,:expire,:timeout,:heartbeat,0,:timeout_source,"
                + dialect.jsonWriteExpression(":state_ext") + ",:tenant_id,:create_time,:operate_time) ";
-        String stateExtensionUpdate = dialect.name().equals("postgres")
-                ? "CASE WHEN CAST(:state_ext AS JSONB)->>'content' <> '' THEN jsonb_build_object('type',"
-                        + TABLE
-                        + ".entity_state_ext->>'type','content',CAST(:state_ext AS JSONB)->>'content','version',"
-                        + "COALESCE((" + TABLE + ".entity_state_ext->>'version')::int,0)+1) ELSE " + TABLE
-                        + ".entity_state_ext END"
-                : "CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(:state_ext,'$.content')) <> '' THEN JSON_OBJECT('type',"
-                        + "JSON_UNQUOTE(JSON_EXTRACT(entity_state_ext,'$.type')),'content',JSON_UNQUOTE(JSON_EXTRACT(:state_ext,'$.content')),'version',"
-                        + "COALESCE(CAST(JSON_UNQUOTE(JSON_EXTRACT(entity_state_ext,'$.version')) AS SIGNED),0)+1) ELSE entity_state_ext END";
-        String upsert = dialect.name().equals("postgres")
-                ? insert
-                        + "ON CONFLICT (tenant_id,entity_type_flag,entity_id) DO UPDATE SET parent_entity_id=EXCLUDED.parent_entity_id,"
-                        + "entity_state_flag=EXCLUDED.entity_state_flag,last_state_flag=" + TABLE
-                        + ".entity_state_flag,"
-                        + "lease_version=" + TABLE
-                        + ".lease_version+1,expire_time=EXCLUDED.expire_time,timeout_seconds=EXCLUDED.timeout_seconds,"
-                        + "last_heartbeat_time=EXCLUDED.last_heartbeat_time,timeout_source_flag=EXCLUDED.timeout_source_flag,"
-                        + "entity_state_ext=" + stateExtensionUpdate + ",operate_time=EXCLUDED.operate_time"
-                : insert
-                        + "ON DUPLICATE KEY UPDATE parent_entity_id=VALUES(parent_entity_id),last_state_flag=entity_state_flag,"
-                        + "entity_state_flag=VALUES(entity_state_flag),lease_version=lease_version+1,expire_time=VALUES(expire_time),"
-                        + "timeout_seconds=VALUES(timeout_seconds),last_heartbeat_time=VALUES(last_heartbeat_time),"
-                        + "timeout_source_flag=VALUES(timeout_source_flag),entity_state_ext=" + stateExtensionUpdate
-                        + ",operate_time=VALUES(operate_time)";
+        String stateExtensionUpdate = "CASE WHEN CAST(:state_ext AS JSONB)->>'content' <> '' THEN jsonb_build_object('type',"
+                + TABLE
+                + ".entity_state_ext->>'type','content',CAST(:state_ext AS JSONB)->>'content','version',"
+                + "COALESCE((" + TABLE + ".entity_state_ext->>'version')::int,0)+1) ELSE " + TABLE
+                + ".entity_state_ext END";
+        String upsert = insert
+                + "ON CONFLICT (tenant_id,entity_type_flag,entity_id) DO UPDATE SET parent_entity_id=EXCLUDED.parent_entity_id,"
+                + "entity_state_flag=EXCLUDED.entity_state_flag,last_state_flag=" + TABLE
+                + ".entity_state_flag,"
+                + "lease_version=" + TABLE
+                + ".lease_version+1,expire_time=EXCLUDED.expire_time,timeout_seconds=EXCLUDED.timeout_seconds,"
+                + "last_heartbeat_time=EXCLUDED.last_heartbeat_time,timeout_source_flag=EXCLUDED.timeout_source_flag,"
+                + "entity_state_ext=" + stateExtensionUpdate + ",operate_time=EXCLUDED.operate_time";
        DatabaseClient.GenericExecuteSpec spec = databaseClient
                .sql(upsert)
                .bind("id", stateId)
@@ -211,9 +200,7 @@ public class R2dbcEntityStateStore implements ReactiveEntityStateStore {
                || renewSeconds < 1) {
            return Mono.empty();
        }
-        String renewExpression = "postgres".equals(dialect.name())
-                ? "CURRENT_TIMESTAMP + (:renew * INTERVAL '1 second')"
-                : "DATE_ADD(CURRENT_TIMESTAMP, INTERVAL :renew SECOND)";
+        String renewExpression = "CURRENT_TIMESTAMP + (:renew * INTERVAL '1 second')";
        return transactionalOperator.transactional(databaseClient
                .sql("UPDATE " + TABLE
                        + " SET last_state_flag=entity_state_flag,entity_state_flag=:offline,"
@@ -249,77 +236,27 @@ public class R2dbcEntityStateStore implements ReactiveEntityStateStore {
        byte maintain = EntityStatusEnum.MAINTAIN.getIndex();
        byte fault = EntityStatusEnum.FAULT.getIndex();
        byte offline = EntityStatusEnum.OFFLINE.getIndex();
-        if ("postgres".equals(dialect.name())) {
-            String sql = "WITH candidates AS (SELECT id FROM " + TABLE
-                    + " WHERE entity_type_flag=:entity_type AND entity_state_flag IN (:online,:maintain,:fault)"
-                    + " AND expire_time <= CURRENT_TIMESTAMP ORDER BY expire_time LIMIT :limit FOR UPDATE SKIP LOCKED)"
-                    + " UPDATE " + TABLE + " s SET last_state_flag=s.entity_state_flag,entity_state_flag=:offline,"
-                    + " lease_version=s.lease_version+1,expire_time=CURRENT_TIMESTAMP + (:renew * INTERVAL '1 second'),"
-                    + " operate_time=CURRENT_TIMESTAMP FROM candidates c WHERE s.id=c.id RETURNING "
-                    + "s.id,s.tenant_id,s.entity_type_flag,s.entity_id,s.parent_entity_id,s.entity_state_flag,s.last_state_flag,"
-                    + "s.lease_version,s.expire_time,s.timeout_seconds,s.last_heartbeat_time,s.last_alarm_id,s.timeout_source_flag,s.entity_state_ext";
-            return transactionalOperator
-                    .transactional(databaseClient
-                            .sql(sql)
-                            .bind("entity_type", type.getIndex())
-                            .bind("online", online)
-                            .bind("maintain", maintain)
-                            .bind("fault", fault)
-                            .bind("offline", offline)
-                            .bind("limit", limit)
-                            .bind("renew", renewSeconds)
-                            .map((row, metadata) -> lease(row, type))
-                            .all()
-                            .collectList())
-                    .flatMapMany(Flux::fromIterable);
-        }
-        String select =
-                "SELECT id,tenant_id,entity_type_flag,entity_id,parent_entity_id,entity_state_flag,last_state_flag,"
-                        + "lease_version,expire_time,timeout_seconds,last_heartbeat_time,last_alarm_id,timeout_source_flag,entity_state_ext"
-                        + " FROM " + TABLE
-                        + " WHERE entity_type_flag=:entity_type AND entity_state_flag IN (:online,:maintain,:fault)"
-                        + " AND expire_time <= CURRENT_TIMESTAMP ORDER BY expire_time LIMIT :limit FOR UPDATE SKIP LOCKED";
+        String sql = "WITH candidates AS (SELECT id FROM " + TABLE
+                + " WHERE entity_type_flag=:entity_type AND entity_state_flag IN (:online,:maintain,:fault)"
+                + " AND expire_time <= CURRENT_TIMESTAMP ORDER BY expire_time LIMIT :limit FOR UPDATE SKIP LOCKED)"
+                + " UPDATE " + TABLE + " s SET last_state_flag=s.entity_state_flag,entity_state_flag=:offline,"
+                + " lease_version=s.lease_version+1,expire_time=CURRENT_TIMESTAMP + (:renew * INTERVAL '1 second'),"
+                + " operate_time=CURRENT_TIMESTAMP FROM candidates c WHERE s.id=c.id RETURNING "
+                + "s.id,s.tenant_id,s.entity_type_flag,s.entity_id,s.parent_entity_id,s.entity_state_flag,s.last_state_flag,"
+                + "s.lease_version,s.expire_time,s.timeout_seconds,s.last_heartbeat_time,s.last_alarm_id,s.timeout_source_flag,s.entity_state_ext";
        return transactionalOperator
                .transactional(databaseClient
-                        .sql(select)
+                        .sql(sql)
                        .bind("entity_type", type.getIndex())
                        .bind("online", online)
                        .bind("maintain", maintain)
                        .bind("fault", fault)
+                        .bind("offline", offline)
                        .bind("limit", limit)
+                        .bind("renew", renewSeconds)
                        .map((row, metadata) -> lease(row, type))
                        .all()
-                        .collectList()
-                        .flatMap(rows -> Flux.fromIterable(rows)
-                                .concatMap(row -> databaseClient
-                                        .sql("UPDATE " + TABLE
-                                                + " SET last_state_flag=entity_state_flag,entity_state_flag=:offline,lease_version=lease_version+1,"
-                                                + "expire_time=DATE_ADD(CURRENT_TIMESTAMP, INTERVAL :renew SECOND),operate_time=CURRENT_TIMESTAMP"
-                                                + " WHERE id=:id AND lease_version=:lease_version")
-                                        .bind("offline", offline)
-                                        .bind("renew", renewSeconds)
-                                        .bind("id", row.id())
-                                        .bind("lease_version", row.leaseVersion())
-                                        .fetch()
-                                        .rowsUpdated()
-                                        .filter(updated -> updated == 1)
-                                        .flatMapMany(ignored -> getLease(row.tenantId(), type, row.entityId()))
-                                        .map(lease -> new EntityStateLease(
-                                                lease.id(),
-                                                lease.tenantId(),
-                                                lease.type(),
-                                                lease.entityId(),
-                                                lease.parentEntityId(),
-                                                lease.stateFlag(),
-                                                row.stateFlag(),
-                                                lease.leaseVersion(),
-                                                lease.expireTime(),
-                                                lease.timeoutSeconds(),
-                                                lease.lastHeartbeatTime(),
-                                                lease.lastAlarmId(),
-                                                lease.timeoutSourceFlag(),
-                                                lease.stateExt())))
-                                .collectList()))
+                        .collectList())
                .flatMapMany(Flux::fromIterable);
    }

@@ -384,8 +321,6 @@ public class R2dbcEntityStateStore implements ReactiveEntityStateStore {

    private DatabaseClient.GenericExecuteSpec bindTime(
            DatabaseClient.GenericExecuteSpec spec, String name, LocalDateTime value) {
-        return "postgres".equalsIgnoreCase(dialect.name())
-                ? spec.bind(name, value.atOffset(ZoneOffset.UTC))
-                : spec.bind(name, value);
+        return spec.bind(name, value.atOffset(ZoneOffset.UTC));
    }
 }
@@ -182,11 +182,7 @@ public class R2dbcNotifyHistoryStore implements ReactiveNotifyHistoryStore {
                + ",:error_message,:retry_count,:tenant_id,:remark,:creator_id,:creator_name,:create_time,:operator_id,:operator_name,:operate_time)";
        boolean deduplicated =
                history.getDedupeKey() != null && !history.getDedupeKey().isBlank();
-        String sql = deduplicated
-                ? ("postgres".equals(dialect.name())
-                        ? baseSql + " ON CONFLICT (tenant_id,dedupe_key) DO NOTHING"
-                        : baseSql + " ON DUPLICATE KEY UPDATE id=id")
-                : baseSql;
+        String sql = deduplicated ? baseSql + " ON CONFLICT (tenant_id,dedupe_key) DO NOTHING" : baseSql;
        DatabaseClient.GenericExecuteSpec spec = insertSpec(history, sql);
        Mono<Long> affected = spec.fetch().rowsUpdated();
        if (!deduplicated) {
@@ -371,9 +367,7 @@ public class R2dbcNotifyHistoryStore implements ReactiveNotifyHistoryStore {

    private DatabaseClient.GenericExecuteSpec bindTime(
            DatabaseClient.GenericExecuteSpec spec, String name, LocalDateTime value) {
-        return "postgres".equals(dialect.name())
-                ? spec.bind(name, OffsetDateTime.of(value, ZoneOffset.UTC))
-                : spec.bind(name, value);
+        return spec.bind(name, OffsetDateTime.of(value, ZoneOffset.UTC));
    }

    private <T> DatabaseClient.GenericExecuteSpec bindNullable(
@@ -96,9 +96,7 @@ public class R2dbcPointValueIngestOutbox implements ReactivePointValueIngestOutb
                + ",status,attempts,available_at,claimed_at,claimed_by) VALUES "
                + "(:tenant_id,:message_id,:schema_version,:driver_node,:sequence,:fencing_token,:device_id,:point_id,"
                + ":raw_value,:cal_value,:num_value,:driver_id,:create_time,:operate_time,'CLAIMED',1,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,:claimed_by) "
-                + (postgres()
-                        ? "ON CONFLICT (tenant_id,message_id) DO NOTHING"
-                        : "ON DUPLICATE KEY UPDATE message_id=message_id");
+                + "ON CONFLICT (tenant_id,message_id) DO NOTHING";
        DatabaseClient.GenericExecuteSpec spec = databaseClient
                .sql(sql)
                .bind("tenant_id", value.getTenantId())
@@ -135,13 +133,9 @@ public class R2dbcPointValueIngestOutbox implements ReactivePointValueIngestOutb
    public Flux<PointValueDO> claim(String owner, int limit) {
        if (owner == null || owner.isBlank() || limit < 1) return Flux.empty();
        int bounded = Math.min(limit, 500);
-        String expired = postgres()
-                ? "UPDATE " + table() + " SET status='PENDING',claimed_at=NULL,claimed_by=NULL "
-                        + "WHERE status='CLAIMED' AND claimed_at < CURRENT_TIMESTAMP - INTERVAL '30 seconds'"
-                : "UPDATE " + table() + " SET status='PENDING',claimed_at=NULL,claimed_by=NULL "
-                        + "WHERE status='CLAIMED' AND claimed_at < CURRENT_TIMESTAMP - INTERVAL 30 SECOND";
-        String persistedLeaseExpired =
-                postgres() ? "CURRENT_TIMESTAMP - INTERVAL '30 seconds'" : "CURRENT_TIMESTAMP - INTERVAL 30 SECOND";
+        String expired = "UPDATE " + table() + " SET status='PENDING',claimed_at=NULL,claimed_by=NULL "
+                + "WHERE status='CLAIMED' AND claimed_at < CURRENT_TIMESTAMP - INTERVAL '30 seconds'";
+        String persistedLeaseExpired = "CURRENT_TIMESTAMP - INTERVAL '30 seconds'";
        String select = "SELECT " + COLUMNS + " FROM " + table()
                + " WHERE (status='PENDING' AND available_at<=CURRENT_TIMESTAMP)"
                + " OR (status='PERSISTED' AND (claimed_at IS NULL OR claimed_at < " + persistedLeaseExpired + "))"
@@ -189,9 +183,7 @@ public class R2dbcPointValueIngestOutbox implements ReactivePointValueIngestOutb
    public Mono<Integer> markFailed(PointValueDO value, String owner, String error) {
        if (value == null || value.getTenantId() == null || value.getMessageId() == null) return Mono.just(0);
        if (owner == null || owner.isBlank()) return Mono.just(0);
-        String backoff = postgres()
-                ? "CURRENT_TIMESTAMP + (LEAST(300, POWER(2, attempts)) * INTERVAL '1 second')"
-                : "DATE_ADD(CURRENT_TIMESTAMP, INTERVAL LEAST(300, POW(2, attempts)) SECOND)";
+        String backoff = "CURRENT_TIMESTAMP + (LEAST(300, POWER(2, attempts)) * INTERVAL '1 second')";
        String sql = "UPDATE " + table() + " SET status=CASE WHEN attempts>=10 THEN 'FAILED' ELSE 'PENDING' END, "
                + "available_at=" + backoff + ",claimed_at=NULL,claimed_by=NULL,last_error=:last_error "
                + "WHERE tenant_id=:tenant_id AND message_id=:message_id AND status IN ('CLAIMED','PERSISTED') AND claimed_by=:owner";
@@ -225,10 +217,6 @@ public class R2dbcPointValueIngestOutbox implements ReactivePointValueIngestOutb
        return value;
    }

-    private boolean postgres() {
-        return "postgres".equalsIgnoreCase(dialect.name());
-    }
-
    private void require(PointValueDO value) {
        if (value.getTenantId() == null
                || value.getTenantId() <= 0
@@ -261,6 +249,6 @@ public class R2dbcPointValueIngestOutbox implements ReactivePointValueIngestOutb

    private DatabaseClient.GenericExecuteSpec bindTime(
            DatabaseClient.GenericExecuteSpec spec, String name, LocalDateTime value) {
-        return postgres() ? spec.bind(name, value.atOffset(ZoneOffset.UTC)) : spec.bind(name, value);
+        return spec.bind(name, value.atOffset(ZoneOffset.UTC));
    }
 }
@@ -18,7 +18,6 @@ package io.github.pnoker.db.postgres.data;

 import io.github.pnoker.common.data.entity.model.PointValueDO;
 import io.github.pnoker.common.data.repository.ReactivePointValueLatestStore;
-import io.github.pnoker.db.r2dbc.core.dialect.R2dbcDialect;
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
@@ -44,7 +43,6 @@ public class R2dbcPointValueLatestStore implements ReactivePointValueLatestStore
            + "sequence, fencing_token, raw_value, cal_value, num_value, driver_id, create_time, operate_time";

    private final DatabaseClient databaseClient;
-    private final R2dbcDialect dialect;
    private final TransactionalOperator transactionalOperator;

    @Override
@@ -105,7 +103,7 @@ public class R2dbcPointValueLatestStore implements ReactivePointValueLatestStore

    private Mono<Integer> upsert(PointValueDO value) {
        requireKey(value);
-        String sql = "postgres".equalsIgnoreCase(dialect.name()) ? postgresUpsert() : mysqlUpsert();
+        String sql = upsertSql();
        DatabaseClient.GenericExecuteSpec spec = databaseClient
                .sql(sql)
                .bind("tenant_id", value.getTenantId())
@@ -125,7 +123,7 @@ public class R2dbcPointValueLatestStore implements ReactivePointValueLatestStore
        return spec.fetch().rowsUpdated().map(Long::intValue);
    }

-    private String postgresUpsert() {
+    private String upsertSql() {
        return "INSERT INTO " + TABLE + " (" + COLUMNS + ") VALUES (:tenant_id,:device_id,:point_id,:message_id,"
                + ":schema_version,:driver_node,:sequence,:fencing_token,:raw_value,:cal_value,:num_value,:driver_id,:create_time,:operate_time) "
                + "ON CONFLICT (tenant_id,device_id,point_id) DO UPDATE SET message_id=EXCLUDED.message_id,"
@@ -136,23 +134,6 @@ public class R2dbcPointValueLatestStore implements ReactivePointValueLatestStore
                + "(dc3_point_latest.fencing_token,dc3_point_latest.create_time,dc3_point_latest.sequence,dc3_point_latest.message_id)";
    }

-    private String mysqlUpsert() {
-        return "INSERT INTO " + TABLE + " (" + COLUMNS + ") VALUES (:tenant_id,:device_id,:point_id,:message_id,"
-                + ":schema_version,:driver_node,:sequence,:fencing_token,:raw_value,:cal_value,:num_value,:driver_id,:create_time,:operate_time) "
-                + "ON DUPLICATE KEY UPDATE message_id=IF((VALUES(fencing_token),VALUES(create_time),VALUES(sequence),VALUES(message_id)) > "
-                + "(fencing_token,create_time,sequence,message_id),VALUES(message_id),message_id), "
-                + "schema_version=IF(VALUES(message_id)=message_id,VALUES(schema_version),schema_version), "
-                + "driver_node=IF(VALUES(message_id)=message_id,VALUES(driver_node),driver_node), "
-                + "sequence=IF(VALUES(message_id)=message_id,VALUES(sequence),sequence), "
-                + "fencing_token=IF(VALUES(message_id)=message_id,VALUES(fencing_token),fencing_token), "
-                + "raw_value=IF(VALUES(message_id)=message_id,VALUES(raw_value),raw_value), "
-                + "cal_value=IF(VALUES(message_id)=message_id,VALUES(cal_value),cal_value), "
-                + "num_value=IF(VALUES(message_id)=message_id,VALUES(num_value),num_value), "
-                + "driver_id=IF(VALUES(message_id)=message_id,VALUES(driver_id),driver_id), "
-                + "create_time=IF(VALUES(message_id)=message_id,VALUES(create_time),create_time), "
-                + "operate_time=IF(VALUES(message_id)=message_id,VALUES(operate_time),operate_time)";
-    }
-
    private PointValueDO map(io.r2dbc.spi.Row row, io.r2dbc.spi.RowMetadata metadata) {
        PointValueDO value = new PointValueDO();
        value.setTenantId(number(row, "tenant_id", Long.class));
@@ -200,9 +181,7 @@ public class R2dbcPointValueLatestStore implements ReactivePointValueLatestStore

    private DatabaseClient.GenericExecuteSpec bindTime(
            DatabaseClient.GenericExecuteSpec spec, String name, LocalDateTime value) {
-        return "postgres".equalsIgnoreCase(dialect.name())
-                ? spec.bind(name, value.atOffset(ZoneOffset.UTC))
-                : spec.bind(name, value);
+        return spec.bind(name, value.atOffset(ZoneOffset.UTC));
    }

    private boolean validKey(Long tenantId, Long deviceId) {
@@ -144,13 +144,8 @@ public class R2dbcTsdbStore implements ReactiveTsdbStore {
                    .append(i)
                    .append(')');
        }
-        if (dialect.name().toLowerCase().contains("mysql")) {
-            sql.append(
-                    " ON DUPLICATE KEY UPDATE message_id=VALUES(message_id),schema_version=VALUES(schema_version),driver_node=VALUES(driver_node),sequence=VALUES(sequence),fencing_token=VALUES(fencing_token),raw_value=VALUES(raw_value),cal_value=VALUES(cal_value),num_value=VALUES(num_value),quality=VALUES(quality),driver_id=VALUES(driver_id),operate_time=VALUES(operate_time)");
-        } else {
-            sql.append(
-                    " ON CONFLICT (tenant_id,device_id,point_id,create_time) DO UPDATE SET message_id=EXCLUDED.message_id,schema_version=EXCLUDED.schema_version,driver_node=EXCLUDED.driver_node,sequence=EXCLUDED.sequence,fencing_token=EXCLUDED.fencing_token,raw_value=EXCLUDED.raw_value,cal_value=EXCLUDED.cal_value,num_value=EXCLUDED.num_value,quality=EXCLUDED.quality,driver_id=EXCLUDED.driver_id,operate_time=EXCLUDED.operate_time");
-        }
+        sql.append(
+                " ON CONFLICT (tenant_id,device_id,point_id,create_time) DO UPDATE SET message_id=EXCLUDED.message_id,schema_version=EXCLUDED.schema_version,driver_node=EXCLUDED.driver_node,sequence=EXCLUDED.sequence,fencing_token=EXCLUDED.fencing_token,raw_value=EXCLUDED.raw_value,cal_value=EXCLUDED.cal_value,num_value=EXCLUDED.num_value,quality=EXCLUDED.quality,driver_id=EXCLUDED.driver_id,operate_time=EXCLUDED.operate_time");
        DatabaseClient.GenericExecuteSpec spec = databaseClient.sql(sql.toString());
        for (int i = 0; i < rows.size(); i++) {
            PointValueSample row = rows.get(i);
@@ -54,7 +54,7 @@ public class R2dbcDeviceImportJobStore implements ReactiveDeviceImportJobStore {
                .sql("INSERT INTO " + TABLE + " (" + COLUMNS
                        + ", created_at) VALUES (:operation_id,:tenant_id,:driver_id,:profile_id,:operator_id,"
                        + ":operator_name,:file_name,:file_data,:claimed_by,:claimed_until,:attempts,:created_at)")
-                .bind("operation_id", uuid(job.operationId()))
+                .bind("operation_id", job.operationId())
                .bind("tenant_id", job.tenantId())
                .bind("driver_id", job.driverId())
                .bind("profile_id", job.profileId())
@@ -83,7 +83,7 @@ public class R2dbcDeviceImportJobStore implements ReactiveDeviceImportJobStore {
                        + " WHERE operation_id=:operation_id AND (claimed_until IS NULL OR claimed_until<:now)")
                .bind("worker_id", workerId)
                .bind("claimed_until", dialect.bindInstant(claimedUntil))
-                .bind("operation_id", uuid(operationId))
+                .bind("operation_id", operationId)
                .bind("now", dialect.bindInstant(now))
                .fetch()
                .rowsUpdated()
@@ -107,7 +107,7 @@ public class R2dbcDeviceImportJobStore implements ReactiveDeviceImportJobStore {
                .sql("UPDATE " + TABLE + " SET claimed_until=:claimed_until"
                        + " WHERE operation_id=:operation_id AND claimed_by=:worker_id")
                .bind("claimed_until", dialect.bindInstant(claimedUntil))
-                .bind("operation_id", uuid(operationId))
+                .bind("operation_id", operationId)
                .bind("worker_id", workerId)
                .fetch()
                .rowsUpdated()
@@ -119,7 +119,7 @@ public class R2dbcDeviceImportJobStore implements ReactiveDeviceImportJobStore {
        if (operationId == null || tenantId == null) return Mono.empty();
        return databaseClient
                .sql("DELETE FROM " + TABLE + " WHERE operation_id=:operation_id AND tenant_id=:tenant_id")
-                .bind("operation_id", uuid(operationId))
+                .bind("operation_id", operationId)
                .bind("tenant_id", tenantId)
                .fetch()
                .rowsUpdated()
@@ -129,7 +129,7 @@ public class R2dbcDeviceImportJobStore implements ReactiveDeviceImportJobStore {
    private Mono<DeviceImportJob> get(UUID operationId) {
        return databaseClient
                .sql("SELECT " + COLUMNS + " FROM " + TABLE + " WHERE operation_id=:operation_id")
-                .bind("operation_id", uuid(operationId))
+                .bind("operation_id", operationId)
                .map((row, metadata) -> new DeviceImportJob(
                        readUuid(row.get("operation_id")),
                        row.get("tenant_id", Long.class),
@@ -145,31 +145,8 @@ public class R2dbcDeviceImportJobStore implements ReactiveDeviceImportJobStore {
                .one();
    }

-    private Object uuid(UUID value) {
-        if (dialect.name().equals("postgres")) return value;
-        byte[] bytes = new byte[16];
-        long most = value.getMostSignificantBits();
-        long least = value.getLeastSignificantBits();
-        for (int index = 7; index >= 0; index--) {
-            bytes[index] = (byte) most;
-            most >>>= 8;
-            bytes[index + 8] = (byte) least;
-            least >>>= 8;
-        }
-        return bytes;
-    }
-
    private UUID readUuid(Object value) {
        if (value instanceof UUID uuid) return uuid;
-        if (value instanceof byte[] bytes && bytes.length == 16) {
-            long most = 0;
-            long least = 0;
-            for (int index = 0; index < 8; index++) {
-                most = (most << 8) | (bytes[index] & 0xffL);
-                least = (least << 8) | (bytes[index + 8] & 0xffL);
-            }
-            return new UUID(most, least);
-        }
        return UUID.fromString(Objects.toString(value));
    }

@@ -19,7 +19,6 @@ package io.github.pnoker.db.postgres.manager;
 import io.github.pnoker.common.manager.entity.model.DeviceLeaseDO;
 import io.github.pnoker.common.manager.entity.model.DriverLeaseStateDO;
 import io.github.pnoker.common.manager.repository.ReactiveDriverLeaseStore;
-import io.github.pnoker.db.r2dbc.core.dialect.R2dbcDialect;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
@@ -27,14 +26,13 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.r2dbc.core.DatabaseClient;
 import org.springframework.stereotype.Repository;
-import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;

 /** R2DBC implementation for driver leases and assignments. */
@Repository
@RequiredArgsConstructor
-@ConditionalOnClass({DatabaseClient.class, R2dbcDialect.class, TransactionalOperator.class})
+@ConditionalOnClass(DatabaseClient.class)
 public class R2dbcDriverLeaseStore implements ReactiveDriverLeaseStore {
    private static final String DRIVER_TABLE = "dc3_manager.dc3_driver";
    private static final String INSTANCE_TABLE = "dc3_manager.dc3_driver_instance";
@@ -44,8 +42,6 @@ public class R2dbcDriverLeaseStore implements ReactiveDriverLeaseStore {
    private static final String REVISION_TABLE = "dc3_manager.dc3_driver_device_revision";

    private final DatabaseClient databaseClient;
-    private final TransactionalOperator transactionalOperator;
-    private final R2dbcDialect dialect;

    @Override
    public Mono<Void> acquireDriverLock(Long tenantId, Long driverId) {
@@ -69,18 +65,12 @@ public class R2dbcDriverLeaseStore implements ReactiveDriverLeaseStore {
        if (!valid(tenantId) || !valid(driverId) || blank(node) || blank(client) || blank(host) || leaseUntil == null) {
            return Mono.error(new IllegalArgumentException("invalid driver instance lease"));
        }
-        String sql = postgres()
-                ? "INSERT INTO " + INSTANCE_TABLE
-                        + " (tenant_id, driver_id, node_id, client_id, service_host, started_at, last_heartbeat, lease_until) "
-                        + "VALUES (:tenant_id, :driver_id, :node_id, :client_id, :service_host, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, :lease_until) "
-                        + "ON CONFLICT (tenant_id, driver_id, node_id) DO UPDATE SET "
-                        + "client_id = EXCLUDED.client_id, service_host = EXCLUDED.service_host, "
-                        + "last_heartbeat = CURRENT_TIMESTAMP, lease_until = EXCLUDED.lease_until"
-                : "INSERT INTO " + INSTANCE_TABLE
-                        + " (tenant_id, driver_id, node_id, client_id, service_host, started_at, last_heartbeat, lease_until) "
-                        + "VALUES (:tenant_id, :driver_id, :node_id, :client_id, :service_host, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, :lease_until) "
-                        + "ON DUPLICATE KEY UPDATE client_id = VALUES(client_id), service_host = VALUES(service_host), "
-                        + "last_heartbeat = CURRENT_TIMESTAMP, lease_until = VALUES(lease_until)";
+        String sql = "INSERT INTO " + INSTANCE_TABLE
+                + " (tenant_id, driver_id, node_id, client_id, service_host, started_at, last_heartbeat, lease_until) "
+                + "VALUES (:tenant_id, :driver_id, :node_id, :client_id, :service_host, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, :lease_until) "
+                + "ON CONFLICT (tenant_id, driver_id, node_id) DO UPDATE SET "
+                + "client_id = EXCLUDED.client_id, service_host = EXCLUDED.service_host, "
+                + "last_heartbeat = CURRENT_TIMESTAMP, lease_until = EXCLUDED.lease_until";
        return databaseClient
                .sql(sql)
                .bind("tenant_id", tenantId)
@@ -164,34 +154,18 @@ public class R2dbcDriverLeaseStore implements ReactiveDriverLeaseStore {
        String sql = "INSERT INTO " + STATE_TABLE
                + " (tenant_id, driver_id, membership_hash, device_revision, assignment_version, operate_time) "
                + "VALUES (:tenant_id, :driver_id, :membership_hash, :device_revision, 1, CURRENT_TIMESTAMP) "
-                + (postgres()
-                        ? "ON CONFLICT (tenant_id, driver_id) DO UPDATE SET membership_hash=EXCLUDED.membership_hash,"
-                                + " device_revision=EXCLUDED.device_revision, assignment_version=" + STATE_TABLE
-                                + ".assignment_version + 1,"
-                                + " operate_time=CURRENT_TIMESTAMP RETURNING assignment_version"
-                        : "ON DUPLICATE KEY UPDATE membership_hash=VALUES(membership_hash), device_revision=VALUES(device_revision),"
-                                + " assignment_version=assignment_version + 1, operate_time=CURRENT_TIMESTAMP");
-        DatabaseClient.GenericExecuteSpec statement = databaseClient
+                + "ON CONFLICT (tenant_id, driver_id) DO UPDATE SET membership_hash=EXCLUDED.membership_hash,"
+                + " device_revision=EXCLUDED.device_revision, assignment_version=" + STATE_TABLE
+                + ".assignment_version + 1,"
+                + " operate_time=CURRENT_TIMESTAMP RETURNING assignment_version";
+        return databaseClient
                .sql(sql)
                .bind("tenant_id", tenantId)
                .bind("driver_id", driverId)
                .bind("membership_hash", membershipHash)
-                .bind("device_revision", deviceRevision);
-        if (postgres()) {
-            return statement
-                    .map((row, metadata) -> row.get("assignment_version", Long.class))
-                    .one();
-        }
-        return transactionalOperator.transactional(statement
-                .fetch()
-                .rowsUpdated()
-                .then(databaseClient
-                        .sql("SELECT assignment_version FROM " + STATE_TABLE
-                                + " WHERE tenant_id=:tenant_id AND driver_id=:driver_id LIMIT 1")
-                        .bind("tenant_id", tenantId)
-                        .bind("driver_id", driverId)
-                        .map((row, metadata) -> row.get("assignment_version", Long.class))
-                        .one()));
+                .bind("device_revision", deviceRevision)
+                .map((row, metadata) -> row.get("assignment_version", Long.class))
+                .one();
    }

    @Override
@@ -232,30 +206,23 @@ public class R2dbcDriverLeaseStore implements ReactiveDriverLeaseStore {
            values.add(lease.getDeviceId());
            values.add(lease.getOwnerNode());
        }
-        if (postgres()) {
-            sql.append(" ON CONFLICT (tenant_id, device_id) DO UPDATE SET ")
-                    .append("fencing_token = CASE WHEN ")
-                    .append(LEASE_TABLE)
-                    .append(".driver_id <> EXCLUDED.driver_id " + "OR ")
-                    .append(LEASE_TABLE)
-                    .append(".owner_node <> EXCLUDED.owner_node THEN ")
-                    .append(LEASE_TABLE)
-                    .append(".fencing_token + 1 ELSE ")
-                    .append(LEASE_TABLE)
-                    .append(".fencing_token END, ")
-                    .append("operate_time = CASE WHEN ")
-                    .append(LEASE_TABLE)
-                    .append(".driver_id <> EXCLUDED.driver_id " + "OR ")
-                    .append(LEASE_TABLE)
-                    .append(".owner_node <> EXCLUDED.owner_node THEN CURRENT_TIMESTAMP ELSE ")
-                    .append(LEASE_TABLE)
-                    .append(".operate_time END, driver_id=EXCLUDED.driver_id, owner_node=EXCLUDED.owner_node");
-        } else {
-            sql.append(
-                    " ON DUPLICATE KEY UPDATE fencing_token = IF(driver_id <> VALUES(driver_id) OR owner_node <> VALUES(owner_node),"
-                            + " fencing_token + 1, fencing_token), operate_time = IF(driver_id <> VALUES(driver_id) OR owner_node <> VALUES(owner_node),"
-                            + " CURRENT_TIMESTAMP, operate_time), driver_id=VALUES(driver_id), owner_node=VALUES(owner_node)");
-        }
+        sql.append(" ON CONFLICT (tenant_id, device_id) DO UPDATE SET ")
+                .append("fencing_token = CASE WHEN ")
+                .append(LEASE_TABLE)
+                .append(".driver_id <> EXCLUDED.driver_id " + "OR ")
+                .append(LEASE_TABLE)
+                .append(".owner_node <> EXCLUDED.owner_node THEN ")
+                .append(LEASE_TABLE)
+                .append(".fencing_token + 1 ELSE ")
+                .append(LEASE_TABLE)
+                .append(".fencing_token END, ")
+                .append("operate_time = CASE WHEN ")
+                .append(LEASE_TABLE)
+                .append(".driver_id <> EXCLUDED.driver_id " + "OR ")
+                .append(LEASE_TABLE)
+                .append(".owner_node <> EXCLUDED.owner_node THEN CURRENT_TIMESTAMP ELSE ")
+                .append(LEASE_TABLE)
+                .append(".operate_time END, driver_id=EXCLUDED.driver_id, owner_node=EXCLUDED.owner_node");

        DatabaseClient.GenericExecuteSpec spec = databaseClient.sql(sql.toString());
        for (int index = 0; index < leases.size(); index++) {
@@ -338,8 +305,4 @@ public class R2dbcDriverLeaseStore implements ReactiveDriverLeaseStore {
    private boolean blank(String value) {
        return value == null || value.isBlank();
    }
-
-    private boolean postgres() {
-        return "postgres".equalsIgnoreCase(dialect.name());
-    }
 }
@@ -39,7 +39,7 @@ import org.springframework.stereotype.Repository;
 import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Mono;

-/** R2DBC group repository shared by PostgreSQL, MySQL and MariaDB. */
+/** Explicit SQL adapter for tenant-scoped groups. */
@Repository
@RequiredArgsConstructor
@ConditionalOnClass({DatabaseClient.class, R2dbcDialect.class, TransactionalOperator.class})
@@ -39,7 +39,7 @@ import org.springframework.stereotype.Repository;
 import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Mono;

-/** R2DBC label repository shared by PostgreSQL, MySQL and MariaDB. */
+/** Explicit SQL adapter for tenant-scoped labels. */
@Repository
@RequiredArgsConstructor
@ConditionalOnClass({DatabaseClient.class, R2dbcDialect.class, TransactionalOperator.class})
@@ -118,7 +118,7 @@ abstract class AbstractDbDialectContractTest {
    }

    private R2dbcPointValueLatestStore latestStore() {
-        return new R2dbcPointValueLatestStore(client(), dialect(), tx());
+        return new R2dbcPointValueLatestStore(client(), tx());
    }

    private R2dbcEntityStateStore stateStore() {