Add support for partial views (#92)

Author: ryannedolan

hoptimator-api/src/main/java/com/linkedin/hoptimator/Job.java

@@ -5,14 +5,20 @@
 
 public class Job {
 
+  private final String name;
   private final Sink sink;
   private final Function<SqlDialect, String> sql;
 
-  public Job(Sink sink, Function<SqlDialect, String> sql) {
+  public Job(String name, Sink sink, Function<SqlDialect, String> sql) {
+    this.name = name;
     this.sink = sink;
     this.sql = sql;
   }
 
+  public String name() {
+    return name;
+  }
+
   public Sink sink() {
     return sink;
   }

hoptimator-cli/src/main/java/sqlline/HoptimatorAppConfig.java

@@ -157,7 +157,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
       CalciteConnection conn = (CalciteConnection) sqlline.getConnection();
       RelRoot root = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root;
       try {
-        List<String> specs = DeploymentService.plan(root).pipeline().specify();
+        List<String> specs = DeploymentService.plan(root).pipeline("sink").specify();
         specs.forEach(x -> sqlline.output(x + "\n\n---\n\n"));
       } catch (SQLException e) {
         sqlline.error(e);

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

@@ -164,13 +164,20 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
       MaterializedViewTable materializedViewTable = new MaterializedViewTable(viewTableMacro);
       RelDataType rowType = materializedViewTable.getRowType(new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT));
 
+      // Suport "partial views", i.e. CREATE VIEW FOO$BAR, where the view name
+      // is "foo-bar" and the sink is just FOO.
+      String sinkName = viewName.split("\\$", 2)[0];
+      List<String> sinkPath = new ArrayList<>();
+      sinkPath.addAll(schemaPath);
+      sinkPath.add(sinkName); 
+
       // Plan a pipeline to materialize the view.
       RelRoot root = HoptimatorDriver.convert(context, sql).root;
       PipelineRel.Implementor plan = DeploymentService.plan(root);
-      plan.setSink(database, viewPath, rowType, Collections.emptyMap());
-      Pipeline pipeline = plan.pipeline();
+      plan.setSink(database, sinkPath, rowType, Collections.emptyMap());
+      Pipeline pipeline = plan.pipeline(viewName);
 
-      MaterializedView hook = new MaterializedView(database, viewPath, sql, plan.sql(), plan.pipeline());
+      MaterializedView hook = new MaterializedView(database, viewPath, sql, plan.sql(), plan.pipeline(viewName));
       // TODO support CREATE ... WITH (options...)
       ValidationService.validateOrThrow(hook, MaterializedView.class);
       pipeline.update();

hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

@@ -74,8 +74,10 @@ public void execute(Context context, boolean execute) throws Exception {
                 String sql = context.previousSqlCommand().sql;
                 CalciteConnection conn = (CalciteConnection) connection;
                 RelRoot root = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root;
+                String []parts = line.split(" ", 2);
+                String pipelineName = parts.length == 2 ? parts[1] : "test";
                 String specs =
-                    DeploymentService.plan(root).pipeline().specify().stream().sorted()
+                    DeploymentService.plan(root).pipeline(pipelineName).specify().stream().sorted()
                         .collect(Collectors.joining("---\n"));
                 String[] lines = specs.replaceAll(";\n", "\n").split("\n");
                 context.echo(Arrays.asList(lines));

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sJobDeployer.java

@@ -26,8 +26,9 @@ class K8sJobDeployer extends K8sYamlDeployer<Job> {
   @Override
   public List<String> specify(Job job) throws SQLException {
     Function<SqlDialect, String> sql = job.sql();
-    Template.Environment env = new Template.SimpleEnvironment().with("name",
-            job.sink().database() + "-" + job.sink().table().toLowerCase(Locale.ROOT))
+    String name = K8sUtils.canonicalizeName(job.sink().database(), job.name());
+    Template.Environment env = new Template.SimpleEnvironment()
+        .with("name", name)
         .with("database", job.sink().database())
         .with("schema", job.sink().schema())
         .with("table", job.sink().table())

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sSourceDeployer.java

@@ -23,8 +23,10 @@ class K8sSourceDeployer extends K8sYamlDeployer<Source> {
 
   @Override
   public List<String> specify(Source source) throws SQLException {
+    String name = K8sUtils.canonicalizeName(source.database(), source.table());
     Template.Environment env =
-        new Template.SimpleEnvironment().with("name", source.database() + "-" + source.table().toLowerCase(Locale.ROOT))
+        new Template.SimpleEnvironment()
+            .with("name", name)
             .with("database", source.database())
             .with("schema", source.schema())
             .with("table", source.table())

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sUtils.java

@@ -2,6 +2,7 @@
 
 import java.util.Collection;
 import java.util.Locale;
+import java.util.stream.Stream;
 import java.util.stream.Collectors;
 
 import io.kubernetes.client.common.KubernetesType;
@@ -22,7 +23,12 @@ public static String canonicalizeName(Collection<String> parts) {
 
   // TODO: Robust and reversible canonicalization
   public static String canonicalizeName(String name) {
-    return name.toLowerCase(Locale.ROOT).replace("_", "");
+    return name.toLowerCase(Locale.ROOT).replace("_", "").replace("$", "-");
+  }
+
+  public static String canonicalizeName(String database, String name) {
+    return Stream.of(database, name).filter(x -> x != null).map(x -> canonicalizeName(x))
+        .collect(Collectors.joining("-"));
   }
 
   // see:

hoptimator-kafka/src/test/resources/kafka-ddl.id

@@ -51,4 +51,4 @@ spec:
   config:
     retention.ms: 7200000
     segment.bytes: 1073741824
-!specify
+!specify existing-topic-1

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/PipelineRel.java

@@ -126,7 +126,7 @@ public void setQuery(RelNode query) {
     }
 
     /** Combine Deployables into a Pipeline */
-    public Pipeline pipeline() throws SQLException {
+    public Pipeline pipeline(String name) throws SQLException {
       List<Deployable> deployables = new ArrayList<>();
       for (Source source : sources.keySet()) {
         deployables.addAll(DeploymentService.deployables(source, Source.class));
@@ -138,7 +138,7 @@ public Pipeline pipeline() throws SQLException {
       }
       Sink sink = new Sink(sinkDatabase, sinkPath, sinkOptions);
       ConnectionService.configure(sink, Sink.class);
-      Job job = new Job(sink, sql());
+      Job job = new Job(name, sink, sql());
       RelOptUtil.equal(sink.table(), targetRowType, "pipeline", query.getRowType(), Litmus.THROW);
       deployables.addAll(DeploymentService.deployables(sink, Sink.class));
       deployables.addAll(DeploymentService.deployables(job, Job.class));

hoptimator-venice/src/test/resources/venice-ddl-insert-all.id

@@ -21,4 +21,4 @@ spec:
     parallelism: 1
     upgradeMode: stateless
     state: running
-!specify
+!specify test-store-1