Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand Down
7 changes: 4 additions & 3 deletions sqrl-cli/src/main/java/com/datasqrl/cli/DatasqrlRun.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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());

Expand All @@ -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);
}
Expand Down
6 changes: 3 additions & 3 deletions sqrl-cli/src/main/java/com/datasqrl/compile/DagWriter.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down
28 changes: 28 additions & 0 deletions sqrl-cli/src/test/java/com/datasqrl/cli/DatasqrlTestTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
}
31 changes: 16 additions & 15 deletions sqrl-cli/src/test/java/com/datasqrl/util/OsProcessManagerTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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));
Expand Down
36 changes: 36 additions & 0 deletions sqrl-deployment-model/pom.xml
Original file line number Diff line number Diff line change
@@ -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>
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.datasqrl.engine.stream;
package com.datasqrl.deployment.model;

import com.datasqrl.engine.EnginePhysicalPlan;
import java.util.List;
import java.util.Set;

public interface StreamPhysicalPlan extends EnginePhysicalPlan {}
/** The contents of the {@code flink.json} deployment file. */
public record FlinkPlanModel(
List<String> flinkSql, Set<String> connectors, Set<String> formats, Set<String> functions) {}
Original file line number Diff line number Diff line change
@@ -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);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
/*
* 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 com.fasterxml.jackson.annotation.JsonInclude;
import java.time.Duration;
import java.util.List;

/** A rendered statement in a JDBC deployment file. */
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public record JdbcStatementModel(
String name,
Type type,
String sql,
String description,
List<Field> fields,
List<String> primaryKey,
List<String> partitionKey,
PartitionType partitionType,
Integer numPartitions,
Duration ttl) {

public JdbcStatementModel(
String name, Type type, String sql, String description, List<Field> fields) {
this(name, type, sql, description, fields, null, null, null, null, null);
}

public enum Type {
TABLE,
VIEW,
QUERY,
INDEX,
EXTENSION
}

public enum PartitionType {
NONE,
HASH,
LIST,
RANGE
}

public record Field(String name, String type, boolean nullable, String description) {}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is component-for-component identical to JdbcStatement.Field, and JdbcPhysicalPlan.toStatementModel maps between them field by field.

Since Field is already pure data with no planner dependencies, moving the existing record into this module and having JdbcStatement reference it would delete both the duplicate type and the mapping loop. Same question applies to MutationDatabaseModel.Table/TableDefinition/ColumnDefinition, which are straight copies of the originals.

The cost of keeping both is that every future field is two edits plus a mapper line, and forgetting the mapper line compiles fine and silently drops the field from the deployment file.

}
Loading