1 package de.dlr.shepard.common.neo4j.migrations;
2
3 import static org.neo4j.cypherdsl.core.Cypher.node;
4
5 import ac.simons.neo4j.migrations.core.JavaBasedMigration;
6 import ac.simons.neo4j.migrations.core.MigrationContext;
7 import de.dlr.shepard.common.util.Neo4jLabels;
8 import de.dlr.shepard.data.timeseries.model.enums.DataPointValueType;
9 import java.sql.Connection;
10 import java.sql.DriverManager;
11 import java.sql.ResultSet;
12 import java.sql.SQLException;
13 import java.sql.Statement;
14 import java.util.ArrayList;
15 import java.util.List;
16 import org.eclipse.microprofile.config.ConfigProvider;
17 import org.jspecify.annotations.NonNull;
18 import org.neo4j.cypherdsl.core.Cypher;
19 import org.neo4j.cypherdsl.core.Node;
20
21 public class V13__Timescale2Neo4j implements JavaBasedMigration {
22
23 private static final Node TSC_NODE = node("TimeseriesContainer").named("tsc");
24
25 private record TimescaleTimeseries(
26 long id,
27 long containerId,
28 String measurement,
29 String device,
30 String location,
31 String symbolicName,
32 String field,
33 DataPointValueType valueType
34 ) {}
35
36 private static final String TIMESERIES_UNIQUE_CONSTRAINT =
37 "CREATE CONSTRAINT timeseries_id_unique IF NOT EXISTS FOR (ts:Timeseries) REQUIRE ts.timeseriesId IS UNIQUE";
38
39 public Connection createPostgresConnection() throws ClassNotFoundException, SQLException {
40 Class.forName("org.postgresql.Driver");
41 return DriverManager.getConnection(
42 ConfigProvider.getConfig().getValue("quarkus.datasource.jdbc.url", String.class),
43 ConfigProvider.getConfig().getValue("quarkus.datasource.username", String.class),
44 ConfigProvider.getConfig().getValue("quarkus.datasource.password", String.class)
45 );
46 }
47
48 @Override
49 public void apply(MigrationContext context) {
50 try (var connection = createPostgresConnection()) {
51 assert isTimescaleOld(connection);
52
53 context.getSession().run(TIMESERIES_UNIQUE_CONSTRAINT);
54 var tsList = getTimeseriesListFromTimescale(connection);
55 migrateTimeseriesMetadataToNeo4(context, tsList);
56 deleteMetadataFromTimescale(connection);
57 } catch (ClassNotFoundException | SQLException e) {
58 throw new MigrationFailureException(e);
59 }
60 }
61
62 private static void migrateTimeseriesMetadataToNeo4(MigrationContext context, List<TimescaleTimeseries> tsList) {
63 try (var session = context.getSession(); var tx = session.beginTransaction()) {
64 tsList.stream().map(V13__Timescale2Neo4j::ts2InsertQuery).forEach(tx::run);
65 tx.commit();
66 }
67 }
68
69 private static void deleteMetadataFromTimescale(Connection connection) throws SQLException {
70 try (
71
72 var dropConstraint = connection.prepareStatement(
73 "alter table timeseries_data_points drop constraint FKog3jr0iowrx3wkun79k0ihs6o"
74 );
75 var dropTimeseries = connection.prepareStatement("drop table timeseries")
76 ) {
77 connection.setAutoCommit(false);
78 dropConstraint.executeUpdate();
79 dropTimeseries.executeUpdate();
80 connection.commit();
81 }
82 }
83
84 private static String ts2InsertQuery(TimescaleTimeseries ts) {
85 var tsNodeToCreate = node("Timeseries")
86 .withProperties(
87 "timeseriesId",
88 Cypher.literalOf(ts.id()),
89 "valueType",
90 Cypher.literalOf(ts.valueType().toString()),
91 "deleted",
92 Cypher.literalOf(false)
93 )
94 .named("ts");
95 var tsTupleNode = node("TimeseriesTuple")
96 .withProperties(
97 "measurement",
98 Cypher.literalOf(ts.measurement()),
99 "device",
100 Cypher.literalOf(ts.device()),
101 "location",
102 Cypher.literalOf(ts.location()),
103 "symbolicName",
104 Cypher.literalOf(ts.symbolicName()),
105 "field",
106 Cypher.literalOf(ts.field())
107 )
108 .named("tst");
109 return Cypher.match(TSC_NODE.where(TSC_NODE.internalId().eq(Cypher.literalOf(ts.containerId()))))
110 .with(TSC_NODE)
111 .create(tsNodeToCreate)
112 .merge(tsTupleNode)
113 .merge(tsNodeToCreate.relationshipTo(tsTupleNode, Neo4jLabels.HAS_TIMESERIES_TUPLE))
114 .merge(tsNodeToCreate.relationshipTo(TSC_NODE, Neo4jLabels.IS_IN_CONTAINER))
115 .build()
116 .getCypher();
117 }
118
119 private @NonNull List<TimescaleTimeseries> getTimeseriesListFromTimescale(Connection connection) throws SQLException {
120 try (
121 var res = connection
122 .prepareStatement(
123 "select id, container_id, measurement, device, location, symbolic_name, field, value_type from timeseries"
124 )
125 .executeQuery()
126 ) {
127 var resList = new ArrayList<TimescaleTimeseries>();
128 while (res.next()) {
129 resList.add(
130 new TimescaleTimeseries(
131 res.getLong(1),
132 res.getLong(2),
133 res.getString(3),
134 res.getString(4),
135 res.getString(5),
136 res.getString(6),
137 res.getString(7),
138 dbValueType2Java(res.getString(8))
139 )
140 );
141 }
142 return resList;
143 }
144 }
145
146 private DataPointValueType dbValueType2Java(String valueType) {
147 return switch (valueType) {
148 case "Boolean" -> DataPointValueType.Boolean;
149 case "Integer" -> DataPointValueType.Integer;
150 case "String" -> DataPointValueType.String;
151 case "Double" -> DataPointValueType.Double;
152 default -> throw new MigrationFailureException("Value type from timescale not assignable!");
153 };
154 }
155
156 private boolean isTimescaleOld(Connection connection) throws SQLException {
157 try (
158 Statement statement = connection.createStatement();
159 ResultSet res = statement.executeQuery("select * from pg_tables");
160 ) {
161 while (res.next()) {
162 String tablename = res.getString("tablename");
163 if (tablename.equals("timeseries") || tablename.equals("timeseries_data_points")) return true;
164 }
165 return false;
166 }
167 }
168 }