Skip to content

Commit b1189ae

Browse files
amansinghoriginalamansinghoriginal
authoredFeb 11, 2025··
Refactor Relational Source and add MySQL Support. (drasi-project#146)
1 parent d0d29ac commit b1189ae

21 files changed

+907
-574
lines changed
 

‎.github/workflows/build-test.yml

+4-4
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ jobs:
7373
install: true
7474

7575
- name: Cache Docker layers
76-
uses: actions/cache@v3
76+
uses: actions/cache@v4
7777
with:
7878
path: /tmp/.buildx-cache
7979
key: buildx-${{ matrix.component.name }}
@@ -124,7 +124,7 @@ jobs:
124124
install: true
125125

126126
- name: Cache Docker layers
127-
uses: actions/cache@v3
127+
uses: actions/cache@v4
128128
with:
129129
path: /tmp/.buildx-cache
130130
key: buildx-${{ matrix.component.name }}
@@ -231,7 +231,7 @@ jobs:
231231
install: true
232232

233233
- name: Cache Docker layers
234-
uses: actions/cache@v3
234+
uses: actions/cache@v4
235235
with:
236236
path: /tmp/.buildx-cache
237237
key: buildx-${{ matrix.component.name }}
@@ -316,7 +316,7 @@ jobs:
316316
install: true
317317

318318
- name: Cache Docker layers
319-
uses: actions/cache@v3
319+
uses: actions/cache@v4
320320
with:
321321
path: /tmp/.buildx-cache
322322
key: buildx-${{ matrix.component.name }}

‎.github/workflows/devskim.yml

+1-1
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ jobs:
3838
uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
3939

4040
- name: Run DevSkim scanner
41-
uses: microsoft/DevSkim-Action@914fa647b406c387000300b2f09bb28691be2b6d # v1.0.14
41+
uses: microsoft/DevSkim-Action@v1
4242

4343
- name: Upload DevSkim scan results to GitHub Security tab
4444
uses: github/codeql-action/upload-sarif@89036746af0bb9507d6f90289b0d5b97d5f44c0c # v2.26.4

‎cli/service/resources/default-source-providers.yaml

+52
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,58 @@ spec:
6666
---
6767
apiVersion: v1
6868
kind: SourceProvider
69+
name: MySQL
70+
spec:
71+
services:
72+
proxy:
73+
image: source-sql-proxy
74+
dapr:
75+
app-port: "4002"
76+
config_schema:
77+
type: object
78+
properties:
79+
connector:
80+
type: string
81+
default: MySQL
82+
reactivator:
83+
image: source-debezium-reactivator
84+
deprovisionHandler: true
85+
dapr:
86+
app-port: "80"
87+
config_schema:
88+
type: object
89+
properties:
90+
connector:
91+
type: string
92+
default: MySQL
93+
config_schema:
94+
type: object
95+
properties:
96+
database:
97+
type: string
98+
host:
99+
type: string
100+
password:
101+
type: string
102+
port:
103+
type: number
104+
ssl:
105+
type: boolean
106+
default: false
107+
user:
108+
type: string
109+
tables:
110+
type: array
111+
required:
112+
- database
113+
- host
114+
- port
115+
- password
116+
- user
117+
- tables
118+
---
119+
apiVersion: v1
120+
kind: SourceProvider
69121
name: SQLServer
70122
spec:
71123
services:

‎sources/relational/debezium-reactivator/pom.xml

+6-1
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
<artifactId>source.sdk</artifactId>
4040
<version>0.1.2</version>
4141
</dependency>
42-
42+
4343
<dependency>
4444
<groupId>io.debezium</groupId>
4545
<artifactId>debezium-api</artifactId>
@@ -55,6 +55,11 @@
5555
<artifactId>debezium-connector-postgres</artifactId>
5656
<version>${version.debezium}</version>
5757
</dependency>
58+
<dependency>
59+
<groupId>io.debezium</groupId>
60+
<artifactId>debezium-connector-mysql</artifactId>
61+
<version>${version.debezium}</version>
62+
</dependency>
5863
<dependency>
5964
<groupId>io.debezium</groupId>
6065
<artifactId>debezium-connector-sqlserver</artifactId>

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/DatabaseStrategy.java

+73
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
package io.drasi;
2+
3+
import com.fasterxml.jackson.databind.JsonNode;
4+
import io.debezium.config.Configuration;
5+
import io.debezium.jdbc.JdbcConnection;
6+
import io.drasi.models.NodeMapping;
7+
import io.drasi.models.RelationalGraphMapping;
8+
9+
import java.sql.Connection;
10+
import java.sql.SQLException;
11+
12+
/**
13+
* Interface for database-specific strategies.
14+
* Note: Might be broken down into smaller interfaces in the future.
15+
*/
16+
public interface DatabaseStrategy {
17+
/**
18+
* Gets a JDBC connection for the database
19+
*
20+
* @param config Configuration for the database
21+
*/
22+
JdbcConnection getConnection(Configuration config);
23+
24+
/**
25+
* Gets the name of the configuration property for the tables list
26+
*/
27+
String getTablesListConfigName();
28+
29+
/**
30+
* Gets the name of the configuration property for the database name
31+
*/
32+
String getDatabaseNameConfigName();
33+
34+
/**
35+
* Gets the node mapping for a table
36+
*
37+
* @param conn Connection to the database
38+
* @param schema Schema of the table
39+
* @param tableName Name of the table
40+
*/
41+
NodeMapping getNodeMapping(Connection conn, String schema, String tableName) throws SQLException;
42+
43+
/**
44+
* Extracts the LSN from a source change
45+
*
46+
* @param sourceChange Source change
47+
*/
48+
long extractLsn(JsonNode sourceChange);
49+
50+
/**
51+
* Extracts the fully qualified table name from a source change
52+
*
53+
* @param sourceChange Source change
54+
*/
55+
String extractTableName(JsonNode sourceChange);
56+
57+
/**
58+
* Creates a connector configuration for the database
59+
*
60+
* @param baseConfig Base configuration
61+
*/
62+
Configuration createConnectorConfig(Configuration baseConfig);
63+
64+
/**
65+
* Performs any database-specific initialization (e.g., publications for Postgres)
66+
*
67+
* @param config Configuration for the database
68+
* @param relationalGraphMapping Relational graph mapping
69+
*/
70+
default void initialize(Configuration config, RelationalGraphMapping relationalGraphMapping) {
71+
// Optional initialization
72+
}
73+
}

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/DebeziumReactivator.java

+9-3
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,9 @@
1616

1717
package io.drasi;
1818

19+
import io.drasi.databases.MySql;
20+
import io.drasi.databases.PostgreSql;
21+
import io.drasi.databases.SqlServer;
1922
import io.drasi.source.sdk.ChangeMonitor;
2023
import io.drasi.source.sdk.Reactivator;
2124
import org.slf4j.Logger;
@@ -32,12 +35,15 @@ public static void main(String[] args) throws IOException, SQLException {
3235

3336
log.info("Starting Debezium Reactivator");
3437

35-
ChangeMonitor monitor = switch (Reactivator.GetConfigValue("connector")) {
36-
case "PostgreSQL" -> new PostgresChangeMonitor();
37-
case "SQLServer" -> new SqlServerChangeMonitor();
38+
DatabaseStrategy strategy = switch (Reactivator.GetConfigValue("connector")) {
39+
case "PostgreSQL" -> new PostgreSql();
40+
case "SQLServer" -> new SqlServer();
41+
case "MySQL" -> new MySql();
3842
default -> throw new IllegalArgumentException("Unknown connector");
3943
};
4044

45+
ChangeMonitor monitor = new RelationalChangeMonitor(strategy);
46+
4147
var reactivator = Reactivator.builder()
4248
.withChangeMonitor(monitor)
4349
.withDeprovisionHandler((statestore) -> statestore.delete("offset"))

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/NoOpSchemaHistory.java

-3
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,11 @@
1919
import io.debezium.annotation.ThreadSafe;
2020
import io.debezium.relational.history.AbstractSchemaHistory;
2121
import io.debezium.relational.history.HistoryRecord;
22-
import org.slf4j.Logger;
23-
import org.slf4j.LoggerFactory;
2422

2523
import java.util.function.Consumer;
2624

2725
@ThreadSafe
2826
public final class NoOpSchemaHistory extends AbstractSchemaHistory {
29-
private static final Logger log = LoggerFactory.getLogger(NoOpSchemaHistory.class);
3027

3128
public NoOpSchemaHistory() {
3229
}

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/PostgresChangeConsumer.java

-33
This file was deleted.

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/PostgresChangeMonitor.java

-104
This file was deleted.

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/PostgresInitializer.java

-109
This file was deleted.

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/RelationalChangeConsumer.java

+90-70
Original file line numberDiff line numberDiff line change
@@ -1,51 +1,60 @@
11
/*
2-
* Copyright 2024 The Drasi Authors.
3-
*
4-
* Licensed under the Apache License, Version 2.0 (the "License");
5-
* you may not use this file except in compliance with the License.
6-
* You may obtain a copy of the License at
7-
*
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
*
10-
* Unless required by applicable law or agreed to in writing, software
11-
* distributed under the License is distributed on an "AS IS" BASIS,
12-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13-
* See the License for the specific language governing permissions and
14-
* limitations under the License.
15-
*/
2+
* Copyright 2024 The Drasi Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
1616

1717
package io.drasi;
1818

1919
import com.fasterxml.jackson.databind.JsonNode;
2020
import com.fasterxml.jackson.databind.ObjectMapper;
21-
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
22-
import com.fasterxml.jackson.databind.node.ObjectNode;
2321
import io.drasi.models.NodeMapping;
2422
import io.drasi.models.RelationalGraphMapping;
2523
import io.drasi.models.RelationshipMapping;
2624
import io.debezium.engine.ChangeEvent;
2725
import io.debezium.engine.DebeziumEngine;
2826
import io.drasi.source.sdk.ChangePublisher;
29-
import io.drasi.source.sdk.Reactivator;
3027
import io.drasi.source.sdk.models.*;
28+
import org.slf4j.Logger;
29+
import org.slf4j.LoggerFactory;
3130

3231
import java.io.IOException;
3332
import java.util.HashMap;
3433
import java.util.List;
3534
import java.util.Map;
3635

37-
abstract class RelationalChangeConsumer implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
38-
39-
private ObjectMapper objectMapper = new ObjectMapper();
40-
private Map<String, NodeMapping> tableToNodeMap;
41-
private Map<String, RelationshipMapping> tableToRelMap;
42-
private ChangePublisher changePublisher;
43-
44-
public RelationalChangeConsumer(RelationalGraphMapping mappings, ChangePublisher changePublisher) {
36+
/**
37+
* Processes change events from relational databases and publishes them using a change publisher.
38+
*/
39+
public class RelationalChangeConsumer implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
40+
private static final Logger log = LoggerFactory.getLogger(RelationalChangeConsumer.class);
41+
private final ObjectMapper objectMapper = new ObjectMapper();
42+
private final Map<String, NodeMapping> tableToNodeMap = new HashMap<>();
43+
private Map<String, RelationshipMapping> tableToRelMap = new HashMap<>();
44+
private final ChangePublisher changePublisher;
45+
private final DatabaseStrategy dbStrategy;
46+
47+
/**
48+
* Creates a new RelationalChangeConsumer instance.
49+
*
50+
* @param mappings Mappings between database tables and graph nodes.
51+
* @param changePublisher Publisher from Drasi Source SDK.
52+
* @param dbStrategy Strategy for the specific database in use.
53+
*/
54+
public RelationalChangeConsumer(RelationalGraphMapping mappings, ChangePublisher changePublisher, DatabaseStrategy dbStrategy) {
4555
this.changePublisher = changePublisher;
46-
tableToNodeMap = new HashMap<>();
47-
tableToRelMap = new HashMap<>();
48-
56+
this.dbStrategy = dbStrategy;
57+
4958
if (mappings.nodes != null)
5059
for (var nodeMapping : mappings.nodes) {
5160
tableToNodeMap.putIfAbsent(nodeMapping.tableName, nodeMapping);
@@ -58,72 +67,83 @@ public RelationalChangeConsumer(RelationalGraphMapping mappings, ChangePublisher
5867
}
5968

6069
@Override
61-
public void handleBatch(List<ChangeEvent<String, String>> records, DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> committer) throws InterruptedException {
62-
for (var record: records) {
63-
64-
if (record.value() == null)
65-
return;
70+
public void handleBatch(List<ChangeEvent<String, String>> records,
71+
DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> committer)
72+
throws InterruptedException {
73+
for (var record : records) {
74+
if (record.value() == null) {
75+
continue;
76+
}
6677

6778
try {
68-
var pgChange = objectMapper.readTree(record.value());
69-
var drasiChange = ExtractNodeChange(pgChange);
79+
var sourceChange = objectMapper.readTree(record.value());
80+
var drasiChange = ExtractDrasiChange(sourceChange);
7081
if (drasiChange != null) {
7182
changePublisher.Publish(drasiChange);
83+
} else {
84+
log.warn("Change not processed: {}", sourceChange);
7285
}
7386
} catch (IOException e) {
87+
log.error("Error processing change record: {}", e.getMessage());
7488
throw new InterruptedException(e.getMessage());
7589
}
90+
7691
committer.markProcessed(record);
7792
}
93+
7894
committer.markBatchFinished();
7995
}
8096

81-
private SourceChange ExtractNodeChange(JsonNode sourceChange) {
82-
var pgPayload = sourceChange.path("payload");
83-
84-
if (!pgPayload.has("op"))
97+
private SourceChange ExtractDrasiChange(JsonNode sourceChange) {
98+
var payload = sourceChange.path("payload");
99+
if (!payload.has("op")) {
85100
return null;
101+
}
86102

87-
var pgSource = pgPayload.path("source");
88-
var tableName = pgSource.path("schema").asText() + "." + pgSource.path("table").asText();
89-
90-
if (!tableToNodeMap.containsKey(tableName))
91-
return null;
103+
var source = payload.path("source");
104+
var tableName = dbStrategy.extractTableName(source);
92105

93106
var mapping = tableToNodeMap.get(tableName);
107+
108+
if (mapping == null) {
109+
log.warn("Table {} not found in mappings", tableName);
110+
return null;
111+
}
94112

95-
JsonNode item;
96-
switch (pgPayload.path("op").asText()) {
97-
case "c", "u":
98-
item = pgPayload.path("after");
99-
break;
100-
case "d":
101-
item = pgPayload.path("before");
102-
break;
103-
default:
104-
return null;
113+
var changeType = payload.path("op").asText();
114+
var item = getChangeData(payload, changeType);
115+
116+
if (item == null) {
117+
log.warn("No change data found for type: {}", changeType);
118+
return null;
105119
}
106-
var nodeId = SanitizeNodeId(mapping.tableName + ":" + item.path(mapping.keyField).asText());
120+
107121
if (!item.has(mapping.keyField)) {
122+
log.warn("Key field {} not found in change data", mapping.keyField);
108123
return null;
109124
}
110-
var tsMs = pgPayload.path("ts_ms").asLong();
111-
112-
switch (pgPayload.path("op").asText()) {
113-
case "c":
114-
return new SourceInsert(nodeId, tsMs, item, null, mapping.labels.stream().toList(), tsMs, ExtractLsn(pgSource));
115-
case "u":
116-
return new SourceUpdate(nodeId, tsMs, item, null, mapping.labels.stream().toList(), tsMs, ExtractLsn(pgSource));
117-
case "d":
118-
return new SourceDelete(nodeId, tsMs, null, mapping.labels.stream().toList(), tsMs, ExtractLsn(pgSource));
119-
}
120125

121-
return null;
126+
var nodeId = createNodeId(mapping.tableName, item.path(mapping.keyField).asText());
127+
var timestamp = payload.path("ts_ms").asLong();
128+
var lsn = dbStrategy.extractLsn(source);
129+
130+
return switch (changeType) {
131+
case "c" -> new SourceInsert(nodeId, timestamp, item, null, mapping.labels.stream().toList(), timestamp, lsn);
132+
case "u" -> new SourceUpdate(nodeId, timestamp, item, null, mapping.labels.stream().toList(), timestamp, lsn);
133+
case "d" -> new SourceDelete(nodeId, timestamp, null, mapping.labels.stream().toList(), timestamp, lsn);
134+
default -> null;
135+
};
122136
}
123137

124-
protected abstract long ExtractLsn(JsonNode sourceChange);
138+
private JsonNode getChangeData(JsonNode payload, String changeType) {
139+
return switch (changeType) {
140+
case "c", "u" -> payload.path("after");
141+
case "d" -> payload.path("before");
142+
default -> null;
143+
};
144+
}
125145

126-
private String SanitizeNodeId(String nodeId) {
127-
return nodeId.replace('.', ':');
146+
private String createNodeId(String tableName, String keyFieldValue) {
147+
return (tableName + ":" + keyFieldValue).replace('.', ':');
128148
}
129-
}
149+
}

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/RelationalChangeMonitor.java

+198
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,198 @@
1+
/*
2+
* Copyright 2024 The Drasi Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.drasi;
18+
19+
import io.drasi.models.NodeMapping;
20+
import io.drasi.models.RelationalGraphMapping;
21+
import io.drasi.models.RelationshipMapping;
22+
import io.debezium.config.Configuration;
23+
import io.debezium.engine.ChangeEvent;
24+
import io.debezium.engine.DebeziumEngine;
25+
import io.debezium.engine.format.Json;
26+
import io.debezium.engine.spi.OffsetCommitPolicy;
27+
import io.drasi.source.sdk.ChangeMonitor;
28+
import io.drasi.source.sdk.ChangePublisher;
29+
import io.drasi.source.sdk.Reactivator;
30+
import io.drasi.source.sdk.StateStore;
31+
import org.slf4j.Logger;
32+
import org.slf4j.LoggerFactory;
33+
34+
import java.sql.SQLException;
35+
import java.util.Collections;
36+
import java.util.LinkedList;
37+
import java.util.List;
38+
import java.util.Properties;
39+
40+
/**
41+
* Monitors database changes using Debezium and publishes them through a change consumer.
42+
*/
43+
public class RelationalChangeMonitor implements ChangeMonitor {
44+
private static final Logger log = LoggerFactory.getLogger(RelationalChangeMonitor.class);
45+
private final DatabaseStrategy dbStrategy;
46+
private DebeziumEngine<ChangeEvent<String, String>> engine;
47+
48+
/**
49+
* Creates a new RelationalChangeMonitor.
50+
*
51+
* @param dbStrategy Strategy for the specific database in use
52+
*/
53+
public RelationalChangeMonitor(DatabaseStrategy dbStrategy) {
54+
this.dbStrategy = dbStrategy;
55+
}
56+
57+
@Override
58+
public void run(ChangePublisher changePublisher, StateStore stateStore) throws Exception {
59+
var baseConfig = createBaseConfig();
60+
var connectorConfig = dbStrategy.createConnectorConfig(baseConfig);
61+
62+
var mappings = createRelationalGraphMapping(connectorConfig);
63+
dbStrategy.initialize(connectorConfig, mappings);
64+
startEngine(changePublisher, connectorConfig, mappings);
65+
}
66+
67+
@Override
68+
public void close() throws Exception {
69+
if (engine != null) {
70+
engine.close();
71+
}
72+
}
73+
74+
private Configuration createBaseConfig() {
75+
var sourceId = Reactivator.SourceId();
76+
var cleanSourceId = sourceId.replace("-", "_");
77+
78+
// Some relational stores offer alternate authentication options, in which case theses are set to empty string.
79+
var dbUser = Reactivator.GetConfigValue("user", "");
80+
var dbPassword = Reactivator.GetConfigValue("password", "");
81+
82+
var dbHost = Reactivator.GetConfigValue("host");
83+
if (dbHost == null || dbHost.isEmpty())
84+
Reactivator.TerminalError(new IllegalArgumentException("Database host is required."));
85+
86+
var dbPort = Reactivator.GetConfigValue("port");
87+
if (dbPort == null || dbPort.isEmpty())
88+
Reactivator.TerminalError(new IllegalArgumentException("Database port is required."));
89+
90+
var tableListStr = Reactivator.GetConfigValue("tables");
91+
if (tableListStr == null || tableListStr.isEmpty())
92+
Reactivator.TerminalError(new IllegalArgumentException("Tables are required."));
93+
94+
var dbName = Reactivator.GetConfigValue("database");
95+
if (dbName == null || dbName.isEmpty())
96+
Reactivator.TerminalError(new IllegalArgumentException("Database name is required."));
97+
98+
var databaseNameConfigStr = dbStrategy.getDatabaseNameConfigName();
99+
var tableListConfigStr = dbStrategy.getTablesListConfigName();
100+
101+
return Configuration.create()
102+
// Name of the database from which to stream the changes.
103+
.with(databaseNameConfigStr, dbName)
104+
// List of tables to include in the connector.
105+
.with(tableListConfigStr, tableListStr)
106+
// Registers custom converters in a comma-separated list.
107+
.with("converters", "temporalConverter")
108+
// Hostname of the database server.
109+
.with("database.hostname", dbHost)
110+
// Password to be used when connecting to the database server.
111+
.with("database.password", dbPassword)
112+
// Port of the database server.
113+
.with("database.port", dbPort)
114+
// Username to be used when connecting to the database server.
115+
.with("database.user", dbUser)
116+
// Represent all DECIMAL, NUMERIC and MONEY values as `doubles`.
117+
.with("decimal.handling.mode", "double")
118+
// Retry a fixed number of times. Default is -1 (infinite).
119+
.with("errors.max.retries", "10")
120+
// Unique name for the connector instance.
121+
.with("name", cleanSourceId)
122+
// Interval at which to try committing offsets. Default is 1 minute.
123+
.with("offset.flush.interval.ms", 5000)
124+
// Class responsible for persistence of connector offsets.
125+
.with("offset.storage", "io.drasi.OffsetBackingStore")
126+
// Class responsible for persistence of database schema history.
127+
.with("schema.history.internal", "io.drasi.NoOpSchemaHistory")
128+
// Define custom converter for temporal types.
129+
.with("temporalConverter.type", "io.drasi.TemporalConverter")
130+
// Determine the type for temporal types based on DB column's type.
131+
// This ensures that all TIME fields captured as microseconds.
132+
.with("time.precision.mode", "adaptive_time_microseconds")
133+
// No subsequent tombstone events will be generated for delete events.
134+
.with("tombstones.on.delete", false)
135+
// Used as a namespace for the connector storage.
136+
.with("topic.prefix", cleanSourceId)
137+
.build();
138+
}
139+
140+
private void startEngine(ChangePublisher changePublisher, Configuration config, RelationalGraphMapping mappings) {
141+
final Properties props = config.asProperties();
142+
engine = DebeziumEngine.create(Json.class)
143+
.using(props)
144+
.using(OffsetCommitPolicy.always())
145+
.using((success, message, error) -> {
146+
if (!success && error != null) {
147+
log.error("Error in Debezium engine: {}", error.getMessage());
148+
Reactivator.TerminalError(error);
149+
}
150+
})
151+
.notifying(new RelationalChangeConsumer(mappings, changePublisher, dbStrategy))
152+
.build();
153+
154+
engine.run();
155+
}
156+
157+
private RelationalGraphMapping createRelationalGraphMapping(Configuration config) {
158+
159+
var result = new RelationalGraphMapping();
160+
result.nodes = readNodeMappingsFromSchema(config);
161+
result.relationships = readRelationshipMappingsFromSchema(config);
162+
163+
return result;
164+
}
165+
166+
private List<RelationshipMapping> readRelationshipMappingsFromSchema(Configuration config) {
167+
return Collections.emptyList();
168+
}
169+
170+
private List<NodeMapping> readNodeMappingsFromSchema(Configuration config) {
171+
var tableNames = config.getString(dbStrategy.getTablesListConfigName());
172+
String[] tables = tableNames.split(",");
173+
var result = new LinkedList<NodeMapping>();
174+
175+
try (var conn = dbStrategy.getConnection(config);
176+
var connection = conn.connection()) {
177+
178+
for (var table : tables) {
179+
table = table.trim();
180+
String schemaName = null;
181+
String tableName = table;
182+
183+
if (table.contains(".")) {
184+
var parts = table.split("\\.");
185+
schemaName = parts[0];
186+
tableName = parts[1];
187+
}
188+
189+
var nodeMapping = dbStrategy.getNodeMapping(connection, schemaName, tableName);
190+
result.add(nodeMapping);
191+
}
192+
} catch (SQLException e) {
193+
Reactivator.TerminalError(e);
194+
}
195+
196+
return result;
197+
}
198+
}

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/SchemaReader.java

-96
This file was deleted.

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/SqlServerChangeConsumer.java

-44
This file was deleted.

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/SqlServerChangeMonitor.java

-97
This file was deleted.

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/databases/MySql.java

+124
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
/*
2+
* Copyright 2024 The Drasi Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.drasi.databases;
18+
19+
import com.fasterxml.jackson.databind.JsonNode;
20+
import io.debezium.config.Configuration;
21+
import io.debezium.connector.mysql.MySqlConnectorConfig;
22+
import io.debezium.connector.mysql.jdbc.MySqlConnection;
23+
import io.debezium.connector.mysql.jdbc.MySqlConnectionConfiguration;
24+
import io.debezium.connector.mysql.jdbc.MySqlFieldReaderResolver;
25+
import io.debezium.jdbc.JdbcConnection;
26+
import io.drasi.DatabaseStrategy;
27+
import io.drasi.models.NodeMapping;
28+
29+
import org.slf4j.Logger;
30+
import org.slf4j.LoggerFactory;
31+
32+
import java.sql.Connection;
33+
import java.sql.SQLException;
34+
import java.util.Collections;
35+
36+
public class MySql implements DatabaseStrategy {
37+
private static final Logger log = LoggerFactory.getLogger(MySql.class);
38+
39+
@Override
40+
public JdbcConnection getConnection(Configuration config) {
41+
var connectionConfig = new MySqlConnectionConfiguration(config);
42+
var connectorConfig = new MySqlConnectorConfig(config);
43+
return new MySqlConnection(connectionConfig, MySqlFieldReaderResolver.resolve(connectorConfig));
44+
}
45+
46+
@Override
47+
public String getDatabaseNameConfigName() {
48+
return "database.include.list";
49+
}
50+
51+
@Override
52+
public String getTablesListConfigName() {
53+
return "table.include.list";
54+
}
55+
56+
@Override
57+
public long extractLsn(JsonNode sourceChange) {
58+
// Get binlog position which is always present
59+
long position = sourceChange.path("pos").asLong(0);
60+
61+
// Get binlog file number from filename (ex: "mysql-bin.000003")
62+
String binlogFile = sourceChange.path("file").asText("");
63+
long fileNumber = 0;
64+
if (!binlogFile.isEmpty()) {
65+
try {
66+
// Extract number from end of filename
67+
String numberPart = binlogFile.substring(binlogFile.lastIndexOf(".") + 1);
68+
fileNumber = Long.parseLong(numberPart);
69+
} catch (NumberFormatException e) {
70+
// If parsing fails, use 0
71+
}
72+
}
73+
74+
// Combine file number and position into single LSN.
75+
// Binlog file numbers have a specific format:
76+
// they're 6-digit numbers padded with zeroes (e.g., "mysql-bin.000001").
77+
// Thus, atmost they will need 20 bits. They are always increasing, and
78+
// to maintain replication order, we can give them the higher 20 bits.
79+
long lsn = (fileNumber << 44) | position;
80+
return lsn;
81+
}
82+
83+
@Override
84+
public String extractTableName(JsonNode sourceChange) {
85+
return sourceChange.path("db").asText() + "." + sourceChange.path("table").asText();
86+
}
87+
88+
@Override
89+
public Configuration createConnectorConfig(Configuration baseConfig) {
90+
return Configuration.create()
91+
// Start with the base configuration.
92+
.with(baseConfig)
93+
// Specify the MySQL connector class.
94+
.with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
95+
// Numeric ID of this database client. No default.
96+
.with("database.server.id", "1")
97+
// Immediately bgin to stream changes without performing a snapshot.
98+
// Note: Might be deprecated in future. no_data needs extra permissions for locking.
99+
// For no_data to work, debezium engine needs (at least one of) the RELOAD or FLUSH_TABLES privilege(s).
100+
// RELOAD allows flushing logs, caches, privileges, and tables.
101+
// If misused, flushing logs could disrupt replication and CDC.
102+
// FLUSH_TABLES allows locking tables and refreshing metadata.
103+
// If misused, it can block writes to tables, impacting performance.
104+
.with("snapshot.mode", "never")
105+
.build();
106+
}
107+
108+
@Override
109+
public NodeMapping getNodeMapping(Connection conn, String schema, String tableName) throws SQLException {
110+
// MySQL uses catalog instead of schema
111+
try (var rs = conn.getMetaData().getPrimaryKeys(schema, null, tableName)) {
112+
if (!rs.next()) {
113+
throw new SQLException("No primary key found for " + tableName);
114+
}
115+
116+
var mapping = new NodeMapping();
117+
mapping.tableName = rs.getString("TABLE_CAT") + "." + tableName;
118+
mapping.keyField = rs.getString("COLUMN_NAME");
119+
mapping.labels = Collections.singleton(tableName);
120+
121+
return mapping;
122+
}
123+
}
124+
}

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/databases/PostgreSql.java

+188
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,188 @@
1+
/*
2+
* Copyright 2024 The Drasi Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.drasi.databases;
18+
19+
import com.fasterxml.jackson.databind.JsonNode;
20+
import io.debezium.config.Configuration;
21+
import io.debezium.connector.postgresql.PostgresConnectorConfig;
22+
import io.debezium.connector.postgresql.connection.PostgresConnection;
23+
import io.debezium.jdbc.JdbcConnection;
24+
import io.drasi.DatabaseStrategy;
25+
import io.drasi.models.NodeMapping;
26+
import io.drasi.models.RelationalGraphMapping;
27+
import io.drasi.source.sdk.Reactivator;
28+
29+
import org.checkerframework.checker.units.qual.t;
30+
import org.slf4j.Logger;
31+
import org.slf4j.LoggerFactory;
32+
33+
import java.sql.Connection;
34+
import java.sql.SQLException;
35+
import java.util.Collections;
36+
import java.util.HashSet;
37+
import java.util.List;
38+
import java.util.Set;
39+
40+
public class PostgreSql implements DatabaseStrategy {
41+
private static final Logger log = LoggerFactory.getLogger(PostgreSql.class);
42+
43+
@Override
44+
public JdbcConnection getConnection(Configuration config) {
45+
var pgConfig = new PostgresConnectorConfig(config);
46+
var jdbcConfig = pgConfig.getJdbcConfig();
47+
48+
var connection = new PostgresConnection(jdbcConfig, "drasi");
49+
return connection;
50+
}
51+
52+
@Override
53+
public NodeMapping getNodeMapping(Connection conn, String schema, String tableName) throws SQLException {
54+
try (var rs = conn.getMetaData().getPrimaryKeys(null, schema, tableName)) {
55+
if (!rs.next()) {
56+
throw new SQLException("No primary key found for " + tableName);
57+
}
58+
59+
var mapping = new NodeMapping();
60+
mapping.tableName = rs.getString("TABLE_SCHEM") + "." + tableName;
61+
mapping.keyField = rs.getString("COLUMN_NAME");
62+
mapping.labels = Collections.singleton(tableName);
63+
64+
return mapping;
65+
}
66+
}
67+
68+
@Override
69+
public long extractLsn(JsonNode sourceChange) {
70+
return sourceChange.get("lsn").asLong();
71+
}
72+
73+
@Override
74+
public String extractTableName(JsonNode sourceChange) {
75+
var schema = sourceChange.path("schema").asText();
76+
var table = sourceChange.path("table").asText();
77+
return schema + "." + table;
78+
}
79+
80+
@Override
81+
public String getDatabaseNameConfigName() {
82+
return "database.dbname";
83+
}
84+
85+
@Override
86+
public String getTablesListConfigName() {
87+
return "table.include.list";
88+
}
89+
90+
@Override
91+
public Configuration createConnectorConfig(Configuration baseConfig) {
92+
var publicationSlotName = "rg_" + baseConfig.getString("name");
93+
return Configuration.create()
94+
// Start with the base configuration.
95+
.with(baseConfig)
96+
// Specify the Postgres connector class.
97+
.with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
98+
// Default is decoder-bufs.
99+
.with("plugin.name", "pgoutput")
100+
// Default is all_tables.
101+
.with("publication.autocreate.mode", "filtered")
102+
// Name of publication created when using pgoutput plugin. Deault is dbz_publication.
103+
.with("publication.name", publicationSlotName)
104+
// Name of replication slot for streaming changes from the database. Default is debezium.
105+
.with("slot.name", publicationSlotName)
106+
// If started first time, start from beginning, else start from last stored LSN.
107+
.with("snapshot.mode", "no_data")
108+
.build();
109+
}
110+
111+
@Override
112+
public void initialize(Configuration config, RelationalGraphMapping relationalGraphMapping) {
113+
// For PostgreSql DBs, Debezium Engine creates a publication on startup
114+
// for tables in the include list, as we've set the publication.autocreate.mode
115+
// to `filtered`. However, if we restart the Debezium engine with changes to
116+
// the `table.include.list`, it doesn't seem to update the publication. So,
117+
// we need to check if the publication tables match the config and update them.
118+
try (var conn = getConnection(config).connection()) {
119+
var pubName = config.getString("publication.name");
120+
var tableList = config.getString(getTablesListConfigName()).split(",");
121+
122+
// Debezium Engine executes `CREATE PUBLICATION {publication.name}`
123+
// on startup for tables in the include list, as we've set the
124+
// publication.autocreate.mode to `filtered`.
125+
if (!publicationExists(conn, pubName)) {
126+
log.warn("Publication {} does not exist, skipping initialization", pubName);
127+
return;
128+
}
129+
130+
var currentTables = getPublicationTables(conn, pubName);
131+
var expectedTables = Set.of(tableList);
132+
133+
if (!currentTables.containsAll(expectedTables) || !expectedTables.containsAll(currentTables)) {
134+
log.warn("Publication {} tables do not match config", pubName);
135+
setPublicationTables(conn, pubName, relationalGraphMapping.nodes);
136+
}
137+
} catch (SQLException e) {
138+
log.error("Error initializing publication: {}", e.getMessage());
139+
Reactivator.TerminalError(e);
140+
}
141+
}
142+
143+
private boolean publicationExists(Connection conn, String pubName) throws SQLException {
144+
try (var stmt = conn.prepareStatement("select * from pg_publication where pubname = ?")) {
145+
stmt.setString(1, pubName);
146+
try (var rs = stmt.executeQuery()) {
147+
return rs.next();
148+
}
149+
}
150+
}
151+
152+
private Set<String> getPublicationTables(Connection conn, String pubName) throws SQLException {
153+
var result = new HashSet<String>();
154+
try (var stmt = conn.prepareStatement("select * from pg_publication_tables where pubname = ?")) {
155+
stmt.setString(1, pubName);
156+
try (var rs = stmt.executeQuery()) {
157+
while (rs.next()) {
158+
result.add(rs.getString("schemaname") + "." + rs.getString("tablename"));
159+
}
160+
}
161+
}
162+
return result;
163+
}
164+
165+
// Alter the publication to include all the tables in the include list.
166+
private void setPublicationTables(Connection conn, String publicationName, List<NodeMapping> mappings) throws SQLException {
167+
var tableList = "";
168+
for (var mapping : mappings) {
169+
if (tableList == "")
170+
tableList += formatTableName(mapping.tableName);
171+
else
172+
tableList += ", " + formatTableName(mapping.tableName);
173+
}
174+
175+
try (var stmt = conn.prepareStatement("ALTER PUBLICATION \"" + publicationName + "\" SET TABLE " + tableList)) {
176+
stmt.execute();
177+
log.info("Updated publication " + publicationName);
178+
}
179+
}
180+
181+
// Quote the table name without quoting the schema name.
182+
private String formatTableName(String name) {
183+
if (!name.contains("."))
184+
return "\"" + name + "\"";
185+
var comps = name.split("\\.");
186+
return comps[0] + "." + "\"" + comps[1] + "\"";
187+
}
188+
}

‎sources/relational/debezium-reactivator/src/main/java/io/drasi/databases/SqlServer.java

+135
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
/*
2+
* Copyright 2024 The Drasi Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.drasi.databases;
18+
19+
import com.fasterxml.jackson.databind.JsonNode;
20+
import io.debezium.config.Configuration;
21+
import io.debezium.connector.sqlserver.SqlServerConnection;
22+
import io.debezium.connector.sqlserver.SqlServerConnectorConfig;
23+
import io.debezium.jdbc.JdbcConnection;
24+
import io.drasi.DatabaseStrategy;
25+
import io.drasi.models.NodeMapping;
26+
import io.drasi.source.sdk.Reactivator;
27+
28+
import java.sql.Connection;
29+
import java.sql.SQLException;
30+
import java.util.Collections;
31+
import java.util.HashSet;
32+
33+
import org.slf4j.Logger;
34+
import org.slf4j.LoggerFactory;
35+
36+
/**
37+
* Implements the strategy to connect to a SQL Server database.
38+
*/
39+
public class SqlServer implements DatabaseStrategy {
40+
private static final Logger log = LoggerFactory.getLogger(SqlServer.class);
41+
42+
@Override
43+
public JdbcConnection getConnection(Configuration config) {
44+
var sqlConfig = new SqlServerConnectorConfig(config);
45+
return new SqlServerConnection(sqlConfig, null, new HashSet<>(), true);
46+
}
47+
48+
@Override
49+
public NodeMapping getNodeMapping(Connection conn, String schema, String tableName) throws SQLException {
50+
try (var rs = conn.getMetaData().getPrimaryKeys(null, schema, tableName)) {
51+
if (!rs.next()) {
52+
throw new SQLException("No primary key found for " + tableName);
53+
}
54+
55+
var mapping = new NodeMapping();
56+
mapping.tableName = rs.getString("TABLE_SCHEM") + "." + tableName;
57+
mapping.keyField = rs.getString("COLUMN_NAME");
58+
mapping.labels = Collections.singleton(tableName);
59+
60+
return mapping;
61+
}
62+
}
63+
64+
@Override
65+
public String getTablesListConfigName() {
66+
return "table.include.list";
67+
}
68+
69+
@Override
70+
public String getDatabaseNameConfigName() {
71+
return "database.names";
72+
}
73+
74+
@Override
75+
public long extractLsn(JsonNode sourceChange) {
76+
var lsn = sourceChange.get("change_lsn").asText();
77+
if (lsn == null || lsn.isEmpty()) {
78+
return 0;
79+
}
80+
81+
String[] parts = lsn.split(":");
82+
if (parts.length != 3) {
83+
return 0;
84+
}
85+
86+
// VLF = Virtual Log File. Transaction log is divided into multiple VLFs.
87+
// Each VLF is contiguous section of log file.
88+
long vlfSeqNo = Long.parseLong(parts[0], 16);
89+
90+
// Log block offset is the offset of the log block within the VLF.
91+
// Typically log blocks are 512 bytes or multiples of 512 bytes.
92+
long logBlockOffset = Long.parseLong(parts[1], 16);
93+
94+
// Pinpoints the exact record within the log block.
95+
// A typical record size can be assumed to be around 64 bytes.
96+
// Then for a 512 byte block, there can be 8 records (0 to 7).
97+
long slotNo = Long.parseLong(parts[2], 16);
98+
99+
// We are allocating 32 bytes for VLF sequence number,
100+
// 16 bytes for log block offset and 16 bytes for slot number.
101+
return (vlfSeqNo << 32) | (logBlockOffset << 16) | slotNo;
102+
}
103+
104+
@Override
105+
public String extractTableName(JsonNode sourceChange) {
106+
return sourceChange.path("schema").asText() + "." + sourceChange.path("table").asText();
107+
}
108+
109+
@Override
110+
public Configuration createConnectorConfig(Configuration baseConfig) {
111+
112+
var encryptConfigValue = Reactivator.GetConfigValue("encrypt");
113+
if (encryptConfigValue == null || encryptConfigValue.isEmpty()) {
114+
Reactivator.TerminalError(new IllegalArgumentException("Encrypt setting is required."));
115+
}
116+
117+
var trustServerCertConfigValue = Reactivator.GetConfigValue("trustServerCertificate", "false");
118+
var authenticationConfigValue = Reactivator.GetConfigValue("authentication", "NotSpecified");
119+
120+
return Configuration.create()
121+
// Start with the base configuration.
122+
.with(baseConfig)
123+
// Specify the SQL Server connector class.
124+
.with("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector")
125+
// Whether JDBC connections to SQL Server should be encrypted. By default this is true.
126+
.with("database.encrypt", encryptConfigValue)
127+
// Capture structure of relevant tables, but do not capture data.
128+
.with("snapshot.mode", "no_data")
129+
// Whether to trust the server certificate. By default this is false.
130+
.with("driver.trustServerCertificate", trustServerCertConfigValue)
131+
// Authentication method to use. By default this is NotSpecified.
132+
.with("driver.authentication", authenticationConfigValue)
133+
.build();
134+
}
135+
}

‎sources/relational/sql-proxy/pom.xml

+6
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,12 @@
3939
<version>0.1.0</version>
4040
</dependency>
4141

42+
<dependency>
43+
<groupId>mysql</groupId>
44+
<artifactId>mysql-connector-java</artifactId>
45+
<version>8.0.33</version>
46+
</dependency>
47+
4248
<dependency>
4349
<groupId>org.postgresql</groupId>
4450
<artifactId>postgresql</artifactId>

‎sources/relational/sql-proxy/src/main/java/io/drasi/ResultStream.java

+14-5
Original file line numberDiff line numberDiff line change
@@ -33,19 +33,19 @@ public ResultStream(BootstrapRequest request) {
3333

3434
public SourceElement next() {
3535
var cursor = cursors.peek();
36-
if (cursor == null) {
36+
if (cursor == null)
3737
return null;
38-
}
38+
3939
var next = cursor.next(connection);
4040
while (next == null) {
4141
cursors.poll().close();
4242
cursor = cursors.peek();
43-
if (cursor == null) {
43+
if (cursor == null)
4444
return null;
45-
}
45+
4646
next = cursor.next(connection);
4747
}
48-
48+
4949
return next;
5050
}
5151

@@ -68,6 +68,15 @@ private static Connection getConnection() throws SQLException {
6868
propsPG.setProperty("sslmode", SourceProxy.GetConfigValue("sslMode", "prefer"));
6969

7070
return DriverManager.getConnection("jdbc:postgresql://" + SourceProxy.GetConfigValue("host") + ":" + SourceProxy.GetConfigValue("port") + "/" + SourceProxy.GetConfigValue("database"), propsPG);
71+
case "MySQL":
72+
var propsMySql = new Properties();
73+
propsMySql.setProperty("user", SourceProxy.GetConfigValue("user"));
74+
propsMySql.setProperty("password", SourceProxy.GetConfigValue("password"));
75+
propsMySql.setProperty("sslmode", SourceProxy.GetConfigValue("sslMode", "prefer"));
76+
77+
var jdbcConnectionString = "jdbc:mysql://" + SourceProxy.GetConfigValue("host") + ":" + SourceProxy.GetConfigValue("port") + "/" + SourceProxy.GetConfigValue("database");
78+
79+
return DriverManager.getConnection(jdbcConnectionString, propsMySql);
7180
case "SQLServer":
7281
var propsSQL = new Properties();
7382
propsSQL.setProperty("user", SourceProxy.GetConfigValue("user"));

‎sources/relational/sql-proxy/src/main/java/io/drasi/TableCursor.java

+7-4
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
44
import com.fasterxml.jackson.databind.node.ObjectNode;
5+
6+
import io.drasi.source.sdk.SourceProxy;
57
import io.drasi.source.sdk.models.SourceElement;
68
import org.slf4j.Logger;
79
import org.slf4j.LoggerFactory;
@@ -28,8 +30,10 @@ private void Init(Connection connection) throws SQLException {
2830
if (resultSet == null) {
2931
mapping = ReadMappingFromSchema(tableName, connection);
3032
var statement = connection.createStatement();
31-
var sanitizedTableName = tableName.replace("\"", "").replace(";", "");
32-
resultSet = statement.executeQuery("SELECT * FROM \"" + sanitizedTableName + "\"");
33+
34+
String quote = SourceProxy.GetConfigValue("connector").equalsIgnoreCase("MySQL") ? "`" : "\"";
35+
var sanitizedTableName = tableName.replace(quote, "").replace(";", "");
36+
resultSet = statement.executeQuery("SELECT * FROM " + quote + sanitizedTableName + quote);
3337
metaData = resultSet.getMetaData();
3438
columnCount = metaData.getColumnCount();
3539
}
@@ -66,9 +70,8 @@ public SourceElement next(Connection connection) {
6670

6771
public void close() {
6872
try {
69-
if (resultSet != null) {
73+
if (resultSet != null)
7074
resultSet.close();
71-
}
7275
} catch (SQLException e) {
7376
log.error("Error closing result set", e);
7477
}

0 commit comments

Comments
 (0)
Please sign in to comment.