diff --git a/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcMcpRuntimeStore.java b/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcMcpRuntimeStore.java index f0125f3dc..a4e035395 100644 --- a/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcMcpRuntimeStore.java +++ b/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcMcpRuntimeStore.java @@ -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); } diff --git a/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcResourceRegistryStore.java b/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcResourceRegistryStore.java index d0ce68b18..c3ad91303 100644 --- a/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcResourceRegistryStore.java +++ b/dc3-db/dc3-db-postgres-auth/src/main/java/io/github/pnoker/db/postgres/auth/R2dbcResourceRegistryStore.java @@ -281,9 +281,8 @@ public class R2dbcResourceRegistryStore implements ReactiveResourceRegistryStore @Override public Mono 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) diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertAnalyticsStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertAnalyticsStore.java index f50fcccac..31ac81fad 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertAnalyticsStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertAnalyticsStore.java @@ -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 countAll(long tenantId) { @@ -74,11 +71,7 @@ public class R2dbcAlertAnalyticsStore implements ReactiveAlertAnalyticsStore { @Override public Flux 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 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 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 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 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 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 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 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 diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertStore.java index 49c9a3d92..62a9dbc84 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcAlertStore.java @@ -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> 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) { diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityAlarmStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityAlarmStore.java index 8b0750988..599b870e5 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityAlarmStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityAlarmStore.java @@ -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()) diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityStateStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityStateStore.java index e6554062d..34b2f68c5 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityStateStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcEntityStateStore.java @@ -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)); } } diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcNotifyHistoryStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcNotifyHistoryStore.java index 455b1f2c4..3ded35fdc 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcNotifyHistoryStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcNotifyHistoryStore.java @@ -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 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 DatabaseClient.GenericExecuteSpec bindNullable( diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueIngestOutbox.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueIngestOutbox.java index 357f23075..1be4df7f8 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueIngestOutbox.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueIngestOutbox.java @@ -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 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 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)); } } diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueLatestStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueLatestStore.java index b55ff9423..2b58a4734 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueLatestStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcPointValueLatestStore.java @@ -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 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) { diff --git a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcTsdbStore.java b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcTsdbStore.java index 4a8aa0139..2b038301f 100644 --- a/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcTsdbStore.java +++ b/dc3-db/dc3-db-postgres-data/src/main/java/io/github/pnoker/db/postgres/data/R2dbcTsdbStore.java @@ -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); diff --git a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDeviceImportJobStore.java b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDeviceImportJobStore.java index 48e40898c..c60ce0b75 100644 --- a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDeviceImportJobStore.java +++ b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDeviceImportJobStore.java @@ -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 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)); } diff --git a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDriverLeaseStore.java b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDriverLeaseStore.java index 2c4f1a766..40cdf9706 100644 --- a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDriverLeaseStore.java +++ b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcDriverLeaseStore.java @@ -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 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()); - } } diff --git a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcGroupStore.java b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcGroupStore.java index 0e1ce16fd..a3eca54d1 100644 --- a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcGroupStore.java +++ b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcGroupStore.java @@ -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}) diff --git a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcLabelStore.java b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcLabelStore.java index e88d30c35..23cad8d26 100644 --- a/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcLabelStore.java +++ b/dc3-db/dc3-db-postgres-manager/src/main/java/io/github/pnoker/db/postgres/manager/R2dbcLabelStore.java @@ -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}) diff --git a/dc3-db/dc3-db-tck/src/test/java/io/github/pnoker/db/tck/AbstractDbDialectContractTest.java b/dc3-db/dc3-db-tck/src/test/java/io/github/pnoker/db/tck/AbstractDbDialectContractTest.java index f47556bc1..a51534184 100644 --- a/dc3-db/dc3-db-tck/src/test/java/io/github/pnoker/db/tck/AbstractDbDialectContractTest.java +++ b/dc3-db/dc3-db-tck/src/test/java/io/github/pnoker/db/tck/AbstractDbDialectContractTest.java @@ -118,7 +118,7 @@ abstract class AbstractDbDialectContractTest { } private R2dbcPointValueLatestStore latestStore() { - return new R2dbcPointValueLatestStore(client(), dialect(), tx()); + return new R2dbcPointValueLatestStore(client(), tx()); } private R2dbcEntityStateStore stateStore() {