Skip to content

Commit 73767c6

Browse files
devhawkdevhawk
committed
WIP
1 parent de16d45 commit 73767c6

4 files changed

Lines changed: 136 additions & 135 deletions

File tree

‎transact-jdbi-step-factory/src/main/java/dev/dbos/transact/jdbi/JdbiStepFactory.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import dev.dbos.transact.database.SystemDatabase;
55
import dev.dbos.transact.json.DBOSSerializer;
66
import dev.dbos.transact.json.SerializationUtil;
7-
import dev.dbos.transact.txstep.PostgresStepFactoryHelpers;
7+
import dev.dbos.transact.txstep.JdbcStepFactory;
88
import dev.dbos.transact.workflow.internal.StepResult;
99

1010
import java.util.Objects;
@@ -97,9 +97,9 @@ public JdbiStepFactory(DBOS dbos, Jdbi jdbi, String schema, DBOSSerializer seria
9797
try {
9898
jdbi.useHandle(
9999
handle -> {
100-
PostgresStepFactoryHelpers.ensurePostgres(handle.getConnection());
101-
PostgresStepFactoryHelpers.ensureSchema(handle.getConnection(), this.schema);
102-
PostgresStepFactoryHelpers.ensureTxOutputTable(handle.getConnection(), this.schema);
100+
JdbcStepFactory.ensurePostgres(handle.getConnection());
101+
JdbcStepFactory.ensureSchema(handle.getConnection(), this.schema);
102+
JdbcStepFactory.ensureTxOutputTable(handle.getConnection(), this.schema);
103103
});
104104
} catch (Exception e) {
105105
if (e instanceof RuntimeException re) {
@@ -178,7 +178,7 @@ public <X extends Exception> void useStep(final HandleConsumer<X> callback, Stri
178178
}
179179

180180
private Optional<StepResult> checkExecution(String workflowId, int stepId, String stepName) {
181-
var sql = PostgresStepFactoryHelpers.CHECK_SQL_TEMPLATE.formatted(schema);
181+
var sql = JdbcStepFactory.CHECK_SQL_TEMPLATE.formatted(schema);
182182
return jdbi.withHandle(
183183
h ->
184184
h.createQuery(sql)
@@ -220,7 +220,7 @@ private void recordResult(
220220
if (output != null && error != null) {
221221
throw new IllegalArgumentException("attempted to record non null output and error result");
222222
}
223-
var sql = PostgresStepFactoryHelpers.UPSERT_SQL_TEMPLATE.formatted(schema);
223+
var sql = JdbcStepFactory.UPSERT_SQL_TEMPLATE.formatted(schema);
224224
handle
225225
.createUpdate(sql)
226226
.bind(0, workflowId)

‎transact-jooq-step-factory/src/main/java/dev/dbos/transact/jooq/JooqStepFactory.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import dev.dbos.transact.database.SystemDatabase;
55
import dev.dbos.transact.json.DBOSSerializer;
66
import dev.dbos.transact.json.SerializationUtil;
7-
import dev.dbos.transact.txstep.PostgresStepFactoryHelpers;
7+
import dev.dbos.transact.txstep.JdbcStepFactory;
88
import dev.dbos.transact.workflow.internal.StepResult;
99

1010
import java.util.Objects;
@@ -47,9 +47,9 @@ public JooqStepFactory(DBOS dbos, DSLContext dsl, String schema, DBOSSerializer
4747
try {
4848
dsl.connection(
4949
conn -> {
50-
PostgresStepFactoryHelpers.ensurePostgres(conn);
51-
PostgresStepFactoryHelpers.ensureSchema(conn, this.schema);
52-
PostgresStepFactoryHelpers.ensureTxOutputTable(conn, this.schema);
50+
JdbcStepFactory.ensurePostgres(conn);
51+
JdbcStepFactory.ensureSchema(conn, this.schema);
52+
JdbcStepFactory.ensureTxOutputTable(conn, this.schema);
5353
});
5454
} catch (Exception e) {
5555
if (e instanceof RuntimeException re) {
@@ -95,7 +95,7 @@ public void txStep(TransactionalRunnable transactional, String stepName) {
9595
}
9696

9797
private Optional<StepResult> checkExecution(String workflowId, int stepId, String stepName) {
98-
var sql = PostgresStepFactoryHelpers.CHECK_SQL_TEMPLATE.formatted(this.schema);
98+
var sql = JdbcStepFactory.CHECK_SQL_TEMPLATE.formatted(this.schema);
9999
return dsl.fetchOptional(sql, workflowId, stepId)
100100
.map(
101101
r ->
@@ -132,7 +132,7 @@ private void recordResult(
132132
if (output != null && error != null) {
133133
throw new IllegalArgumentException("attempted to record non null output and error result");
134134
}
135-
var sql = PostgresStepFactoryHelpers.UPSERT_SQL_TEMPLATE.formatted(schema);
135+
var sql = JdbcStepFactory.UPSERT_SQL_TEMPLATE.formatted(schema);
136136
trx.dsl().execute(sql, workflowId, stepId, output, error, serialization);
137137
}
138138
}

‎transact/src/main/java/dev/dbos/transact/txstep/JdbcStepFactory.java‎

Lines changed: 124 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import dev.dbos.transact.database.SystemDatabase;
55
import dev.dbos.transact.json.DBOSSerializer;
66
import dev.dbos.transact.json.SerializationUtil;
7+
import dev.dbos.transact.txstep.TransactionalRunnable.WrappedSqlException;
78
import dev.dbos.transact.workflow.internal.StepResult;
89

910
import java.sql.Connection;
@@ -13,6 +14,9 @@
1314

1415
import javax.sql.DataSource;
1516

17+
import org.slf4j.Logger;
18+
import org.slf4j.LoggerFactory;
19+
1620
/**
1721
* A {@link PostgresStepFactoryHelpers} implementation backed by plain JDBC {@link Connection}
1822
* objects.
@@ -34,6 +38,110 @@
3438
*/
3539
public class JdbcStepFactory {
3640

41+
private static final Logger logger = LoggerFactory.getLogger(JdbcStepFactory.class);
42+
43+
public static final String CHECK_SQL_TEMPLATE =
44+
"""
45+
SELECT output, error, serialization
46+
FROM "%s".tx_step_outputs
47+
WHERE workflow_id = ? AND step_id = ?
48+
""";
49+
50+
public static final String UPSERT_SQL_TEMPLATE =
51+
"""
52+
INSERT INTO "%s".tx_step_outputs
53+
(workflow_id, step_id, output, error, serialization)
54+
VALUES (?, ?, ?, ?, ?)
55+
ON CONFLICT DO NOTHING
56+
""";
57+
58+
/**
59+
* Verifies that the given connection is to a PostgreSQL database.
60+
*
61+
* @throws IllegalArgumentException if the database is not PostgreSQL
62+
* @throws SQLException if database metadata cannot be read
63+
*/
64+
public static void ensurePostgres(Connection conn) throws SQLException {
65+
var productName = conn.getMetaData().getDatabaseProductName();
66+
if (!productName.equalsIgnoreCase("PostgreSQL")) {
67+
throw new IllegalArgumentException(
68+
"PostgresStepFactory requires a PostgreSQL datasource, got: " + productName);
69+
}
70+
}
71+
72+
/**
73+
* Returns {@code true} if the named schema exists in the database.
74+
*
75+
* @throws SQLException if the query fails
76+
*/
77+
public static boolean schemaExists(Connection conn, String schema) throws SQLException {
78+
var sql = "SELECT schema_name FROM information_schema.schemata WHERE schema_name = ?";
79+
try (var stmt = conn.prepareStatement(sql)) {
80+
stmt.setString(1, Objects.requireNonNull(schema, "schema must not be null"));
81+
try (var rs = stmt.executeQuery()) {
82+
return rs.next();
83+
}
84+
}
85+
}
86+
87+
/**
88+
* Creates the named schema if it does not already exist.
89+
*
90+
* @throws SQLException if the DDL fails
91+
*/
92+
public static void ensureSchema(Connection conn, String schema) throws SQLException {
93+
Objects.requireNonNull(schema, "schema must not be null");
94+
if (!schemaExists(conn, schema)) {
95+
try (var stmt = conn.createStatement()) {
96+
stmt.execute("CREATE SCHEMA IF NOT EXISTS \"%s\"".formatted(schema));
97+
}
98+
}
99+
}
100+
101+
/**
102+
* Returns {@code true} if the {@code tx_step_outputs} table exists in the named schema.
103+
*
104+
* @throws SQLException if the query fails
105+
*/
106+
public static boolean tableExists(Connection conn, String schema) throws SQLException {
107+
var sql = "SELECT 1 FROM information_schema.tables WHERE table_schema = ? AND table_name = ?";
108+
try (var stmt = conn.prepareStatement(sql)) {
109+
stmt.setString(1, Objects.requireNonNull(schema, "schema must not be null"));
110+
stmt.setString(2, Objects.requireNonNull("tx_step_outputs", "tableName must not be null"));
111+
try (var rs = stmt.executeQuery()) {
112+
return rs.next();
113+
}
114+
}
115+
}
116+
117+
/**
118+
* Creates the {@code tx_step_outputs} table in the named schema if it does not already exist.
119+
*
120+
* @throws SQLException if the DDL fails
121+
*/
122+
public static void ensureTxOutputTable(Connection conn, String schema) throws SQLException {
123+
Objects.requireNonNull(schema, "schema must not be null");
124+
if (tableExists(conn, schema)) {
125+
return;
126+
}
127+
logger.debug("Creating tx_step_outputs table in schema={}", schema);
128+
try (var stmt = conn.createStatement()) {
129+
var ddlSql =
130+
"""
131+
CREATE TABLE IF NOT EXISTS "%1$s".tx_step_outputs (
132+
workflow_id TEXT NOT NULL,
133+
step_id INT NOT NULL,
134+
output TEXT,
135+
error TEXT,
136+
serialization TEXT,
137+
created_at BIGINT NOT NULL DEFAULT (EXTRACT(EPOCH FROM now())*1000)::bigint,
138+
PRIMARY KEY (workflow_id, step_id)
139+
)"""
140+
.formatted(schema);
141+
stmt.execute(ddlSql);
142+
}
143+
}
144+
37145
private final DBOS dbos;
38146
private final DataSource dataSource;
39147
private final String schema;
@@ -65,9 +173,9 @@ public JdbcStepFactory(DBOS dbos, DataSource dataSource, String schema, DBOSSeri
65173
this.serializer = serializer == null ? config.serializer() : serializer;
66174

67175
try (var conn = dataSource.getConnection()) {
68-
PostgresStepFactoryHelpers.ensurePostgres(conn);
69-
PostgresStepFactoryHelpers.ensureSchema(conn, this.schema);
70-
PostgresStepFactoryHelpers.ensureTxOutputTable(conn, this.schema);
176+
ensurePostgres(conn);
177+
ensureSchema(conn, this.schema);
178+
ensureTxOutputTable(conn, this.schema);
71179
}
72180
}
73181

@@ -183,7 +291,7 @@ private static void close(Connection conn) {
183291
}
184292

185293
private Optional<StepResult> checkExecution(String workflowId, int stepId, String stepName) {
186-
var sql = PostgresStepFactoryHelpers.CHECK_SQL_TEMPLATE.formatted(this.schema);
294+
var sql = CHECK_SQL_TEMPLATE.formatted(this.schema);
187295
try (var conn = dataSource.getConnection();
188296
var stmt = conn.prepareStatement(sql)) {
189297
stmt.setString(1, workflowId);
@@ -209,7 +317,8 @@ private Optional<StepResult> checkExecution(String workflowId, int stepId, Strin
209317

210318
private <R> void recordOutput(Connection conn, String workflowId, int stepId, R result) {
211319
var value = SerializationUtil.serializeValue(result, null, serializer);
212-
recordResult(conn, workflowId, stepId, value.serializedValue(), null, value.serialization());
320+
recordResult(
321+
conn, schema, workflowId, stepId, value.serializedValue(), null, value.serialization());
213322
}
214323

215324
private <X extends Exception> void recordError(String workflowId, int stepId, X exception) {
@@ -218,13 +327,20 @@ private <X extends Exception> void recordError(String workflowId, int stepId, X
218327
dataSource,
219328
(Connection conn) -> {
220329
recordResult(
221-
conn, workflowId, stepId, null, value.serializedValue(), value.serialization());
330+
conn,
331+
schema,
332+
workflowId,
333+
stepId,
334+
null,
335+
value.serializedValue(),
336+
value.serialization());
222337
return null;
223338
});
224339
}
225340

226-
private void recordResult(
341+
public static void recordResult(
227342
Connection conn,
343+
String schema,
228344
String workflowId,
229345
int stepId,
230346
String output,
@@ -233,7 +349,7 @@ private void recordResult(
233349
if (output != null && error != null) {
234350
throw new IllegalArgumentException("attempted to record non null output and error result");
235351
}
236-
var sql = PostgresStepFactoryHelpers.UPSERT_SQL_TEMPLATE.formatted(schema);
352+
var sql = UPSERT_SQL_TEMPLATE.formatted(schema);
237353
try (var stmt = conn.prepareStatement(sql)) {
238354
stmt.setString(1, workflowId);
239355
stmt.setInt(2, stepId);

‎transact/src/main/java/dev/dbos/transact/txstep/PostgresStepFactoryHelpers.java‎

Lines changed: 0 additions & 115 deletions
This file was deleted.

0 commit comments

Comments
 (0)