refactor: Extract deployment file model classes to sqrl-deployment-model

pom.xml

@@ -69,6 +69,7 @@
     <module>sqrl-cli</module>
     <module>sqrl-discovery</module>
     <module>sqrl-planner</module>
+    <module>sqrl-deployment-model</module>
     <module>sqrl-server</module>
     <module>sqrl-testing</module>
   </modules>
@@ -220,6 +221,11 @@
       </dependency>
 
       <!-- SQRL deps, please keep this in order -->
+      <dependency>
+        <groupId>com.datasqrl</groupId>
+        <artifactId>sqrl-deployment-model</artifactId>
+        <version>${project.version}</version>
+      </dependency>
       <dependency>
         <groupId>com.datasqrl</groupId>
         <artifactId>sqrl-server-core</artifactId>

sqrl-cli/src/main/java/com/datasqrl/cli/DatasqrlRun.java

@@ -21,6 +21,7 @@
 import static com.datasqrl.env.EnvVariableNames.POSTGRES_USERNAME;
 
 import com.datasqrl.config.PackageJson;
+import com.datasqrl.deployment.model.KafkaNewTopicModel;
 import com.datasqrl.engine.server.VertxEngineFactory;
 import com.datasqrl.flinkrunner.SqrlRunner;
 import com.datasqrl.flinkrunner.utils.EnvUtils;
@@ -260,7 +261,7 @@ private void initKafka() {
     var topicsToCreate = new HashSet<String>();
 
     Stream.concat(kafkaPlan.topics().stream(), kafkaPlan.testRunnerTopics().stream())
-        .map(com.datasqrl.engine.log.kafka.NewTopic::topicName)
+        .map(KafkaNewTopicModel::topicName)
         .forEach(topicsToCreate::add);
 
     var bootstrapServers = getenv(KAFKA_BOOTSTRAP_SERVERS);
@@ -304,9 +305,9 @@ private void initPostgres() {
         DriverManager.getConnection(
             getenv(POSTGRES_JDBC_URL), getenv(POSTGRES_USERNAME), getenv(POSTGRES_PASSWORD))) {
       for (var jdbcStmt : statements) {
-        log.info("Executing statement {} of type {}", jdbcStmt.getName(), jdbcStmt.getType());
+        log.info("Executing statement {} of type {}", jdbcStmt.name(), jdbcStmt.type());
         try (Statement stmt = connection.createStatement()) {
-          stmt.execute(jdbcStmt.getSql());
+          stmt.execute(jdbcStmt.sql());
         } catch (Exception e) {
           e.printStackTrace();
           assert false : e.getMessage();

sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java

@@ -18,9 +18,9 @@
 import com.datasqrl.config.GraphqlSourceLoader;
 import com.datasqrl.config.PackageJson;
 import com.datasqrl.config.WorkspacePaths;
+import com.datasqrl.deployment.model.JdbcStatementModel.Type;
 import com.datasqrl.engine.PhysicalPlan;
 import com.datasqrl.engine.database.relational.JdbcPhysicalPlan;
-import com.datasqrl.engine.database.relational.JdbcStatement;
 import com.datasqrl.engine.server.ServerPhysicalPlan;
 import com.datasqrl.engine.stream.flink.FlinkStreamEngine;
 import com.datasqrl.error.ErrorCode;
@@ -31,6 +31,7 @@
 import com.datasqrl.planner.SqlScriptPlanner;
 import com.datasqrl.planner.Sqrl2FlinkSQLTranslator;
 import com.datasqrl.planner.dag.DAGPlanner;
+import com.datasqrl.planner.dag.plan.MutationDatabase;
 import com.datasqrl.server.GenerateServerModel;
 import com.datasqrl.util.ServiceLoaderDiscovery;
 import java.nio.file.Path;
@@ -111,7 +112,7 @@ public Pair<PhysicalPlan, TestPlan> executeCompilation(Optional<Path> testsPath)
         var jdbcViews =
             physicalPlan
                 .getPlans(JdbcPhysicalPlan.class)
-                .map(p -> p.getStatementsForType(JdbcStatement.Type.VIEW))
+                .map(p -> p.getStatementsForType(Type.VIEW))
                 .findFirst()
                 .orElse(List.of());
 
@@ -126,7 +127,8 @@ public Pair<PhysicalPlan, TestPlan> executeCompilation(Optional<Path> testsPath)
         .ifPresent(
             compareDb ->
                 errors.checkFatal(
-                    mutationDatabase.isBackwardsCompatible(compareDb, environment, errors),
+                    MutationDatabase.isBackwardsCompatible(
+                        mutationDatabase, compareDb, environment, errors),
                     "The mutation tables defined by the script are not backwards compatible with the provided database. See warnings above for incompatibilities."));
     return Pair.of(physicalPlan, testPlan);
   }

sqrl-cli/src/main/java/com/datasqrl/compile/DagWriter.java

@@ -18,9 +18,9 @@
 import com.datasqrl.config.PackageJson.CompilerConfig;
 import com.datasqrl.config.PackageJson.ExplainConfig;
 import com.datasqrl.config.WorkspacePaths;
+import com.datasqrl.deployment.model.MutationDatabaseModel;
 import com.datasqrl.plan.global.PipelineDAGExporter;
 import com.datasqrl.planner.dag.PipelineDAG;
-import com.datasqrl.planner.dag.plan.MutationDatabase;
 import com.datasqrl.serializer.Deserializer;
 import com.google.common.io.Resources;
 import java.io.IOException;
@@ -55,12 +55,12 @@ public class DagWriter {
   private final CompilerConfig compilerConfig;
 
   @SneakyThrows
-  void run(PipelineDAG dag, String source, MutationDatabase mutationDatabase) {
+  void run(PipelineDAG dag, String source, MutationDatabaseModel mutationDatabaseModel) {
     writeExplain(dag);
     writeFile(workspacePaths.buildDir().resolve(FULL_SOURCE_FILENAME), source);
     writeFile(
         workspacePaths.buildDir().resolve(DATABASE_FILENAME),
-        Deserializer.INSTANCE.toJson(mutationDatabase));
+        Deserializer.INSTANCE.toJson(mutationDatabaseModel));
   }
 
   void writeInferredSchema(String inferredSchema) {

sqrl-cli/src/main/java/com/datasqrl/packager/Packager.java

@@ -142,7 +142,8 @@ private void writePlan(String name, EnginePhysicalPlan plan, Path planDir) {
 
     var artifacts =
         ListUtils.union(
-            plan.getDeploymentArtifacts(), List.of(new DeploymentArtifact(".json", plan)));
+            plan.getDeploymentArtifacts(),
+            List.of(new DeploymentArtifact(".json", plan.toFileModel())));
 
     for (var artifact : artifacts) {
       if (artifact.isEmpty()) {

sqrl-cli/src/test/java/com/datasqrl/cli/DatasqrlTestTest.java

@@ -23,7 +23,11 @@
 
 import com.datasqrl.cli.output.NoOutputFormatter;
 import com.datasqrl.cli.output.TestOutputManager;
+import com.datasqrl.compile.TestPlan;
 import com.datasqrl.config.PackageJson;
+import com.datasqrl.deployment.model.JdbcStatementModel;
+import com.datasqrl.engine.database.relational.GenericJdbcStatement;
+import com.datasqrl.util.SqrlObjectMapper;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Map;
@@ -68,4 +72,28 @@ void run_whenPipelineFailsToStart_recordsFailureWithNonZeroExit() throws Excepti
     var exitCode = underTest.run();
     assertThat(exitCode).isNotZero();
   }
+
+  @Test
+  void givenTestPlanWithJdbcViews_whenDeserialized_thenUsesGenericJdbcStatements()
+      throws Exception {
+    var json =
+        """
+        {
+          "jdbcViews": [{
+            "name": "orders_view",
+            "type": "VIEW",
+            "sql": "CREATE VIEW orders_view AS SELECT 1"
+          }],
+          "queries": [],
+          "mutations": [],
+          "subscriptions": []
+        }
+        """;
+
+    var plan = SqrlObjectMapper.INSTANCE.readValue(json, TestPlan.class);
+
+    assertThat(plan.getJdbcViews()).hasSize(1);
+    assertThat(plan.getJdbcViews().get(0)).isInstanceOf(GenericJdbcStatement.class);
+    assertThat(plan.getJdbcViews().get(0).getType()).isEqualTo(JdbcStatementModel.Type.VIEW);
+  }
 }

sqrl-cli/src/test/java/com/datasqrl/util/OsProcessManagerTest.java

@@ -22,11 +22,10 @@
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.when;
 
-import com.datasqrl.engine.database.relational.GenericJdbcStatement;
-import com.datasqrl.engine.database.relational.JdbcPhysicalPlan;
-import com.datasqrl.engine.database.relational.JdbcStatement;
-import com.datasqrl.engine.log.kafka.KafkaPhysicalPlan;
-import com.datasqrl.engine.log.kafka.NewTopic;
+import com.datasqrl.deployment.model.JdbcPlanModel;
+import com.datasqrl.deployment.model.JdbcStatementModel;
+import com.datasqrl.deployment.model.KafkaNewTopicModel;
+import com.datasqrl.deployment.model.KafkaPlanModel;
 import com.datasqrl.env.GlobalEnvironmentStore;
 import java.io.File;
 import java.io.IOException;
@@ -78,7 +77,7 @@ void givenLogFileExists_whenReadServiceLogFile_thenReturnsContent() throws Excep
       filesMocked.when(() -> Files.exists(mockLogFile)).thenReturn(true);
       filesMocked
           .when(() -> Files.readAllLines(mockLogFile))
-          .thenReturn(java.util.List.of("Line 1", "Line 2", "Line 3"));
+          .thenReturn(List.of("Line 1", "Line 2", "Line 3"));
 
       // When
       String result = serviceManager.readServiceLogFile(serviceName);
@@ -147,7 +146,7 @@ void givenCustomEnvironmentVariables_whenStartDependentServices_thenSetsSystemPr
       // Mock that no services are needed
       configMocked
           .when(() -> ConfigLoaderUtils.loadKafkaPhysicalPlan(mockPlanDir))
-          .thenReturn(Optional.of(new KafkaPhysicalPlan(List.of(), List.of())));
+          .thenReturn(Optional.of(new KafkaPlanModel(List.of(), List.of())));
       configMocked
           .when(() -> ConfigLoaderUtils.loadPostgresPhysicalPlan(mockPlanDir))
           .thenReturn(Optional.empty());
@@ -342,10 +341,10 @@ void givenPlanDirWithKafkaTopics_whenStartDependentServices_thenStartsRedpanda()
       filesMocked.when(() -> Files.createDirectories(any(Path.class))).thenReturn(mockPath);
 
       // Mock that Kafka topics are found but no Postgres statements
-      var mockTopic = mock(NewTopic.class);
+      var mockTopic = mock(KafkaNewTopicModel.class);
       configMocked
           .when(() -> ConfigLoaderUtils.loadKafkaPhysicalPlan(mockPlanDir))
-          .thenReturn(Optional.of(new KafkaPhysicalPlan(List.of(mockTopic), List.of())));
+          .thenReturn(Optional.of(new KafkaPlanModel(List.of(mockTopic), List.of())));
       configMocked
           .when(() -> ConfigLoaderUtils.loadPostgresPhysicalPlan(mockPlanDir))
           .thenReturn(Optional.empty());
@@ -396,8 +395,9 @@ void givenPlanDirWithPostgresStatements_whenStartDependentServices_thenStartsPos
           .when(() -> ConfigLoaderUtils.loadKafkaPhysicalPlan(mockPlanDir))
           .thenReturn(Optional.empty());
       var mockStatement =
-          new GenericJdbcStatement("test", JdbcStatement.Type.TABLE, "CREATE TABLE test");
-      var mockJdbcPlan = JdbcPhysicalPlan.builder().statement(mockStatement).build();
+          new JdbcStatementModel(
+              "test", JdbcStatementModel.Type.TABLE, "CREATE TABLE test", null, null);
+      var mockJdbcPlan = new JdbcPlanModel(List.of(mockStatement), List.of());
       configMocked
           .when(() -> ConfigLoaderUtils.loadPostgresPhysicalPlan(mockPlanDir))
           .thenReturn(Optional.of(mockJdbcPlan));
@@ -442,13 +442,14 @@ void givenPlanDirWithBothServices_whenStartDependentServices_thenStartsBothServi
       filesMocked.when(() -> Files.list(any(Path.class))).thenReturn(Stream.of(mockPath));
 
       // Mock that both Kafka topics and Postgres statements are found
-      var mockTopic = mock(NewTopic.class);
+      var mockTopic = mock(KafkaNewTopicModel.class);
       configMocked
           .when(() -> ConfigLoaderUtils.loadKafkaPhysicalPlan(mockPlanDir))
-          .thenReturn(Optional.of(new KafkaPhysicalPlan(List.of(mockTopic), List.of())));
+          .thenReturn(Optional.of(new KafkaPlanModel(List.of(mockTopic), List.of())));
       var mockStatement =
-          new GenericJdbcStatement("test", JdbcStatement.Type.TABLE, "CREATE TABLE test");
-      var mockJdbcPlan = JdbcPhysicalPlan.builder().statement(mockStatement).build();
+          new JdbcStatementModel(
+              "test", JdbcStatementModel.Type.TABLE, "CREATE TABLE test", null, null);
+      var mockJdbcPlan = new JdbcPlanModel(List.of(mockStatement), List.of());
       configMocked
           .when(() -> ConfigLoaderUtils.loadPostgresPhysicalPlan(mockPlanDir))
           .thenReturn(Optional.of(mockJdbcPlan));

sqrl-deployment-model/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright © 2021 DataSQRL (contact@datasqrl.com)
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>com.datasqrl</groupId>
+    <artifactId>sqrl-root</artifactId>
+    <version>1.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>sqrl-deployment-model</artifactId>
+  <name>SQRL :: Deployment Model</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-annotations</artifactId>
+    </dependency>
+  </dependencies>
+</project>

sqrl-deployment-model/src/main/java/com/datasqrl/deployment/model/CombinedPlanModel.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.deployment.model;
+
+import java.util.Map;
+
+/** The contents of a deployment file that combines plans from multiple database engines. */
+public record CombinedPlanModel(Map<String, Object> plans) {
+
+  public CombinedPlanModel {
+    plans = Map.copyOf(plans);
+  }
+}

sqrl-deployment-model/src/main/java/com/datasqrl/deployment/model/FlinkPlanModel.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.deployment.model;
+
+import java.util.List;
+import java.util.Set;
+
+/** The contents of the {@code flink.json} deployment file. */
+public record FlinkPlanModel(
+    List<String> flinkSql, Set<String> connectors, Set<String> formats, Set<String> functions) {}

sqrl-deployment-model/src/main/java/com/datasqrl/deployment/model/JdbcPlanModel.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.deployment.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import java.util.List;
+
+/** The contents of a JDBC database plan file such as {@code postgres.json}. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public record JdbcPlanModel(
+    List<JdbcStatementModel> statements, List<JdbcStatementModel> standaloneExtensionStatements) {
+
+  public JdbcPlanModel {
+    statements = statements == null ? List.of() : List.copyOf(statements);
+    standaloneExtensionStatements =
+        standaloneExtensionStatements == null
+            ? List.of()
+            : List.copyOf(standaloneExtensionStatements);
+  }
+}