Support hints via jdbc connection properties (#116)

* Support hints via jdbc connection properties

* Fix PipelineReconciler deletion errors

* checkstyle

* fixes

* more checkstyle

Author: jogrogan

README.md

@@ -111,21 +111,79 @@ To push a Flink job directly to the Flink deployment created above, `kubectl app
 
 ## The SQL CLI
 
-The `./hoptimator` script launches the [sqlline](https://github.com/julianhyde/sqlline) SQL CLI pre-configured to connect to `jdbc:hoptimator://`. The CLI includes some additional commands. See `!intro`.
+The `./hoptimator` script launches the [sqlline](https://github.com/julianhyde/sqlline) SQL CLI pre-configured to connect to `jdbc:hoptimator://`.
+The CLI includes some additional commands. See `!intro`.
 
 ## The JDBC Driver
 
 To use Hoptimator from Java code, or from anything that supports JDBC, use the `jdbc:hoptimator://` JDBC driver.
 
 ## The Operator
 
-`hoptimator-operator` turns materialized views into real data pipelines.
+`hoptimator-operator` turns materialized views into real data pipelines. The name operator comes from the Kubernetes Operator pattern.
+`PipelineOperatorApp` is intended to be an entry point for a running application that can listen to and reconcile the resources created in Kubernetes by the K8s Deployers.
+See [hoptimator-operator-deployment.yaml](deploy/hoptimator-operator-deployment.yaml) for K8s pod deployment of the operator.
 
 ## Extending Hoptimator
 
-Hoptimator can be extended via `TableTemplates`:
+Hoptimator is extensible via `hoptimator-api`, which provides hooks for deploying, validating, and configuring the elements of a pipeline,
+including external objects (e.g. Kafka topics) and data plane connectors (e.g. Flink connectors).
+To deploy a source or sink, implement `Deployer<Source>`.
+To deploy a job, implement `Deployer<Job>`.
 
+In addition, the `k8s` catalog is itself highly extensible via `TableTemplates` and `JobTemplates`.
+Generally, you can get Hoptimator to do what you want without writing any new code.
+
+### Table Templates
+
+`TableTemplates` let you specify how sources and sinks should be included in a pipeline. For example see [kafkadb.yaml](deploy/samples/kafkadb.yaml).
+
+In this case, any tables within `kafka-database` will get deployed as `KafkaTopics` and use `kafka` connectors.
+
+### Job Templates
+
+`JobTemplates` are similar, but can embed SQL. For example see [flink-template.yaml](deploy/samples/flink-template.yaml).
+
+In this case, any jobs created with this template will get deployed as `FlinkSessionJobs` within the `flink` namespace.
+
+### Configuration
+
+The ``{{ }}`` sections you see in the templates are variable placeholders that will be filled in by the Deployer.
+See [Template.java](hoptimator-util/src/main/java/com/linkedin/hoptimator/util/Template.java) for how to specify templates.
+
+While Deployers are extensible, today the primary deployer is to Kubernetes. These deployers 
+[K8sSourceDeployer](hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sSourceDeployer.java) (for table-templates)
+and [K8sJobDeployer](hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sJobDeployer.java) (for job-templates)
+provide a few template defaults that you can choose to include in your templates:
+
+K8sSourceDeployer: `name, database, schema, table`
+
+K8sJobDeployer: `name, database, schema, table, sql, flinksql, flinkconfigs`
+
+However, it is often a case where you want to add additional information to the templates that will be passed through during Source or Job creation.
+There are two mechanisms to achieve this:
+
+#### ConfigProvider
+
+The ConfigProvider interface allows you to load additional configuration information that will be used during Source or Job creation.
+
+The [K8sConfigProvider](hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sConfigProvider.java) is the default one used in the K8s deployers.
+K8sConfigProvider contains the ability to read configuration information via a Kubernetes configmap, `hoptimator-configmap`.
+See [hoptimator-configmap.yaml](deploy/config/hoptimator-configmap.yaml) for information on how to use configmaps.
+
+Configmaps are meant to be used for static configuration information applicable to the namespace `hoptimator-configmap` belongs to.
+
+#### Hints
+
+Users may want to provide additional information for their Job or Sink creation at runtime.
+This can be done by adding hints as JDBC properties.
+
+Hints are key-value pairs separated by an equals sign. Multiple hints are separated by a comma.
+
+For example, to specify the number of kafka partitions and the flink parallelism, you could add the following hints to the query:
 ```
- $ kubectl apply -f my-table-template.yaml
+jdbc:hoptimator://hints=kafka.partitions=4,flink.parallelism=2
 ```
+These fields can then be added to templates as `{{kafka.partitions}}` or `{{flink.parallelism}}` where applicable.
 
+Note that hints are simply recommendations, if the planner plans a different pipeline, they will be ignored.

deploy/samples/flink-template.yaml

@@ -17,6 +17,6 @@ spec:
         args:
         - {{flinksql}}
         jarURI: file:///opt/hoptimator-flink-runner.jar
-        parallelism: 1
+        parallelism: {{flink.parallelism:1}}
         upgradeMode: stateless
         state: running

deploy/samples/kafkadb.yaml

@@ -25,7 +25,7 @@ spec:
         strimzi.io/cluster: one
     spec:
       topicName: {{table}}
-      partitions: 1
+      partitions: {{kafka.partitions:1}}
       replicas: 1
       config:
         retention.ms: 7200000

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

@@ -1,6 +1,5 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Wrapper;
 

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

@@ -2,7 +2,6 @@
 
 import java.util.List;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 
 
 public class MaterializedView extends View implements Deployable {

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

@@ -1,6 +1,5 @@
 package com.linkedin.hoptimator;
 
-import java.sql.SQLException;
 import java.util.Collection;
 
 
@@ -9,9 +8,9 @@
  */
 public class Pipeline {
 
-  private Collection<Source> sources;
-  private Sink sink;
-  private Job job;
+  private final Collection<Source> sources;
+  private final Sink sink;
+  private final Job job;
 
   public Pipeline(Collection<Source> sources, Sink sink, Job job) {
     this.sources = sources;

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

@@ -11,9 +11,7 @@
 import org.apache.calcite.rel.RelRoot;
 import org.jline.reader.Completer;
 
-import com.linkedin.hoptimator.Job;
 import com.linkedin.hoptimator.Pipeline;
-import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;
 import com.linkedin.hoptimator.SqlDialect;
 import com.linkedin.hoptimator.jdbc.HoptimatorConnection;
@@ -93,7 +91,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
       HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
       try {
         RelRoot root = HoptimatorDriver.convert(conn, sql).root;
-        PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations());
+        PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations(), conn.connectionProperties());
         sqlline.output(plan.sql(conn.connectionProperties()).apply(SqlDialect.ANSI));
       } catch (SQLException e) {
         sqlline.error(e);
@@ -163,7 +161,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
       RelRoot root = HoptimatorDriver.convert(conn, sql).root;
       try {
         Properties connectionProperties = conn.connectionProperties();
-        Pipeline pipeline = DeploymentService.plan(root, conn.materializations()).pipeline("sink", connectionProperties);
+        Pipeline pipeline = DeploymentService.plan(root, conn.materializations(), conn.connectionProperties()).pipeline("sink", connectionProperties);
         List<String> specs = new ArrayList<>();
         for (Source source : pipeline.sources()) {
           specs.addAll(DeploymentService.specify(source, connectionProperties));

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

@@ -31,14 +31,13 @@
 import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeImpl;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeImpl;
 import org.apache.calcite.rel.type.RelDataTypeSystem;
 import org.apache.calcite.rel.type.RelProtoDataType;
 import org.apache.calcite.schema.Function;
 import org.apache.calcite.schema.SchemaPlus;
 import org.apache.calcite.schema.Table;
-import org.apache.calcite.schema.TranslatableTable;
 import org.apache.calcite.schema.impl.ViewTable;
 import org.apache.calcite.server.DdlExecutor;
 import org.apache.calcite.server.ServerDdlExecutor;
@@ -205,10 +204,10 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
 
       // Plan a pipeline to materialize the view.
       RelRoot root = new HoptimatorDriver.Prepare(connection).convert(context, sql).root;
-      PipelineRel.Implementor plan = DeploymentService.plan(root, connection.materializations());
+      PipelineRel.Implementor plan = DeploymentService.plan(root, connection.materializations(), connectionProperties);
       plan.setSink(database, sinkPath, rowType, Collections.emptyMap());
       Pipeline pipeline = plan.pipeline(viewName, connectionProperties);
- 
+
       MaterializedView hook = new MaterializedView(database, viewPath, sql, plan.sql(connectionProperties), pipeline);
       // TODO support CREATE ... WITH (options...)
       ValidationService.validateOrThrow(hook);
@@ -217,8 +216,8 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
         DeploymentService.update(hook, connectionProperties);
       } else {
         DeploymentService.create(hook, connectionProperties);
-      } 
-      
+      }
+
       schemaPlus.add(viewName, materializedViewTable);
     } catch (Exception e) {
       throw new RuntimeException("Cannot CREATE MATERIALIZED VIEW in " + schemaName + ": " + e.getMessage(), e);

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

@@ -2,11 +2,11 @@
 
 import java.io.IOException;
 import java.sql.Connection;
-import java.sql.SQLException;
 import java.sql.DriverManager;
 import java.sql.DriverPropertyInfo;
+import java.sql.SQLException;
 import java.util.Properties;
-import java.util.function.Supplier;
+import java.util.logging.LogManager;
 
 import org.apache.calcite.avatica.ConnectStringParser;
 import org.apache.calcite.jdbc.CalciteConnection;
@@ -17,10 +17,8 @@
 import org.apache.calcite.schema.impl.AbstractSchema;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.parser.SqlParser;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import java.util.logging.LogManager;
 
 import com.linkedin.hoptimator.Catalog;
 
@@ -32,13 +30,15 @@ public class HoptimatorDriver implements java.sql.Driver {
 
   public static final String CONNECTION_PREFIX = "jdbc:hoptimator://";
 
-  static {{
-    try {
-      DriverManager.registerDriver(INSTANCE);
-    } catch (SQLException e) {
-      throw new RuntimeException("Failed to register Hoptimator driver.", e);
+  static {
+    {
+      try {
+        DriverManager.registerDriver(INSTANCE);
+      } catch (SQLException e) {
+        throw new RuntimeException("Failed to register Hoptimator driver.", e);
+      }
     }
-  }}
+  }
 
   public static CalcitePrepare.ConvertResult convert(HoptimatorConnection conn, String sql) {
     CalcitePrepare.Context context = conn.createPrepareContext();
@@ -77,7 +77,7 @@ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) throws
 
   @Override
   public boolean jdbcCompliant() {
-    return false; 
+    return false;
   }
 
   @Override
@@ -109,7 +109,7 @@ public Connection connect(String url, Properties props) throws SQLException {
       rootSchema.add("DEFAULT", new AbstractSchema());
 
       calciteConnection.setSchema("DEFAULT");
-      
+
       HoptimatorConnection hoptimatorConnection = new HoptimatorConnection(calciteConnection, properties);
       holder.connection = hoptimatorConnection;
 

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

@@ -1,6 +1,5 @@
 package com.linkedin.hoptimator.jdbc;
 
-import java.util.Collections;
 import java.util.List;
 
 import org.apache.calcite.plan.RelOptTable;

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

@@ -12,7 +12,6 @@
 import org.apache.calcite.schema.SchemaPlus;
 import org.apache.calcite.schema.Table;
 
-import com.linkedin.hoptimator.Validated;
 import com.linkedin.hoptimator.Validator;
 import com.linkedin.hoptimator.ValidatorProvider;
 

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/schema/UtilityCatalog.java

@@ -1,11 +1,9 @@
 package com.linkedin.hoptimator.jdbc.schema;
 
-import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Wrapper;
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.Properties;
 
 import org.apache.calcite.schema.SchemaPlus;
 import org.apache.calcite.schema.Table;

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

@@ -49,7 +49,7 @@ protected void run(URI resource, String jdbcProperties) throws IOException {
       Quidem.Config config = Quidem.configBuilder()
           .withReader(r)
           .withWriter(w)
-          .withConnectionFactory((x, y) -> DriverManager.getConnection("jdbc:hoptimator://catalogs=" + x + jdbcProperties))
+          .withConnectionFactory((x, y) -> DriverManager.getConnection("jdbc:hoptimator://catalogs=" + x + ";" + jdbcProperties))
           .withCommandHandler(new CustomCommandHandler())
           .build();
       new Quidem(config).execute();
@@ -82,7 +82,7 @@ public void execute(Context context, boolean execute) throws Exception {
                 RelRoot root = HoptimatorDriver.convert(conn, sql).root;
                 String []parts = line.split(" ", 2);
                 String pipelineName = parts.length == 2 ? parts[1] : "test";
-                Pipeline pipeline = DeploymentService.plan(root, Collections.emptyList())
+                Pipeline pipeline = DeploymentService.plan(root, Collections.emptyList(), conn.connectionProperties())
                     .pipeline(pipelineName, conn.connectionProperties());
                 List<String> specs = new ArrayList<>();
                 for (Source source : pipeline.sources()) {

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

@@ -1,11 +1,9 @@
 package com.linkedin.hoptimator.k8s;
 
 import java.sql.SQLException;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 import io.kubernetes.client.common.KubernetesListObject;
@@ -158,7 +156,7 @@ public void updateStatus(T obj, Object status) throws SQLException {
 
   private void checkResponse(KubernetesApiResponse<?> resp) throws SQLException {
     if (!resp.isSuccess()) {
-      throw new SQLException(resp.getStatus().getMessage() + ": " + context);
+      throw new SQLException(resp.getStatus().getMessage() + ": " + context, null, resp.getHttpStatusCode());
     }
   }
 }

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

@@ -2,7 +2,6 @@
 
 import java.sql.SQLException;
 import java.sql.Wrapper;
-import java.util.Properties;
 
 import org.apache.calcite.schema.SchemaPlus;
 import org.slf4j.Logger;

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

@@ -52,7 +52,8 @@ public K8sDatabaseTable(K8sContext context, K8sEngineTable engines) {
   public void addDatabases(SchemaPlus parentSchema, Properties connectionProperties) {
     for (Row row : rows()) {
       parentSchema.add(schemaName(row),
-          HoptimatorJdbcSchema.create(row.NAME, row.SCHEMA, dataSource(row, connectionProperties), parentSchema, dialect(row), engines.forDatabase(row.NAME), connectionProperties));
+          HoptimatorJdbcSchema.create(row.NAME, row.SCHEMA, dataSource(row, connectionProperties), parentSchema,
+              dialect(row), engines.forDatabase(row.NAME), connectionProperties));
     }
   }
 

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

@@ -6,9 +6,8 @@
 
 import io.kubernetes.client.common.KubernetesListObject;
 import io.kubernetes.client.common.KubernetesObject;
-import io.kubernetes.client.util.Yaml;
 import io.kubernetes.client.openapi.models.V1OwnerReference;
-
+import io.kubernetes.client.util.Yaml;
 
 import com.linkedin.hoptimator.Deployer;
 

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

@@ -25,7 +25,7 @@ public <T extends Deployable> Collection<Deployer> deployers(T obj, Properties c
     } else if (obj instanceof View) {
       list.add(new K8sViewDeployer((View) obj, false, context));
     } else if (obj instanceof Job) {
-      list.add(new K8sJobDeployer((Job) obj, context));
+      list.add(new K8sJobDeployer((Job) obj, context, connectionProperties));
     } else if (obj instanceof Source) {
       list.add(new K8sSourceDeployer((Source) obj, context));
     }

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

@@ -19,18 +19,20 @@ class K8sJobDeployer extends K8sYamlDeployer {
 
   private static final String FLINK_CONFIG = "flink.config";
 
+  private final Properties connectionProperties;
   private final Job job;
   private final K8sApi<V1alpha1JobTemplate, V1alpha1JobTemplateList> jobTemplateApi;
 
-  K8sJobDeployer(Job job, K8sContext context) {
+  K8sJobDeployer(Job job, K8sContext context, Properties connectionProperties) {
     super(context);
+    this.connectionProperties = connectionProperties;
     this.job = job;
     this.jobTemplateApi = new K8sApi<>(context, K8sApiEndpoints.JOB_TEMPLATES);
   }
 
   @Override
   public List<String> specify() throws SQLException {
-    Properties properties = ConfigService.config(null, false, FLINK_CONFIG);
+    Properties properties = ConfigService.config(this.connectionProperties, false, FLINK_CONFIG);
     properties.putAll(job.sink().options());
     Function<SqlDialect, String> sql = job.sql();
     String name = K8sUtils.canonicalizeName(job.sink().database(), job.name());