Add per-connection driver properties (#99)

Co-authored-by: Joseph Grogan <[email protected]>

Author: ryannedolan

gradle/libs.versions.toml

@@ -26,8 +26,8 @@ junit = "junit:junit:4.12"
 kafka-clients = "org.apache.kafka:kafka-clients:3.2.0"
 kubernetes-client = "io.kubernetes:client-java:18.0.0"
 kubernetes-extended-client = "io.kubernetes:client-java-extended:18.0.0"
-slf4j-simple = "org.slf4j:slf4j-simple:1.7.30"
-slf4j-api = "org.slf4j:slf4j-api:1.7.30"
+slf4j-simple = "org.slf4j:slf4j-simple:1.7.36"
+slf4j-api = "org.slf4j:slf4j-api:1.7.36"
 sqlline = "sqlline:sqlline:1.12.0"
 quidem = "net.hydromatic:quidem:0.11"
 venice = "com.linkedin.venice:venice-common:0.4.376"

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

@@ -2,6 +2,7 @@
 
 import java.sql.SQLException;
 import java.sql.Wrapper;
+import java.util.Properties;
 
 
 /** Registers a set of tables, possibly within schemas and sub-schemas. */
@@ -11,5 +12,5 @@ public interface Catalog {
 
   String description();
 
-  void register(Wrapper parentSchema) throws SQLException;
+  void register(Wrapper parentSchema, Properties connectionProperties) throws SQLException;
 }

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

@@ -4,5 +4,5 @@
 
 public interface ConfigProvider {
 
-  Properties loadConfig(String namespace) throws Exception;
+  Properties loadConfig(Properties connectionProperties) throws Exception;
 }

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

@@ -1,9 +1,10 @@
 package com.linkedin.hoptimator;
 
 import java.util.Collection;
+import java.util.Properties;
 
 
 public interface ConnectorProvider {
 
-  <T> Collection<Connector<T>> connectors(Class<T> clazz);
+  <T> Collection<Connector<T>> connectors(Class<T> clazz, Properties connectionProperties);
 }

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

@@ -1,9 +1,10 @@
 package com.linkedin.hoptimator;
 
 import java.util.Collection;
+import java.util.Properties;
 
 
 public interface DeployerProvider {
 
-  <T> Collection<Deployer<T>> deployers(Class<T> clazz);
+  <T> Collection<Deployer<T>> deployers(Class<T> clazz, Properties connectionProperties);
 }

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

@@ -7,11 +7,11 @@
 import java.util.List;
 import java.util.Scanner;
 
-import org.apache.calcite.jdbc.CalciteConnection;
 import org.apache.calcite.rel.RelRoot;
 import org.jline.reader.Completer;
 
 import com.linkedin.hoptimator.SqlDialect;
+import com.linkedin.hoptimator.jdbc.HoptimatorConnection;
 import com.linkedin.hoptimator.jdbc.HoptimatorDriver;
 import com.linkedin.hoptimator.util.DeploymentService;
 import com.linkedin.hoptimator.util.planner.PipelineRel;
@@ -73,7 +73,7 @@ public String matches(String line) {
 
     @Override
     public void execute(String line, DispatchCallback dispatchCallback) {
-      if (!(sqlline.getConnection() instanceof CalciteConnection)) {
+      if (!(sqlline.getConnection() instanceof HoptimatorConnection)) {
         sqlline.error("This connection doesn't support `!pipeline`.");
         dispatchCallback.setToFailure();
         return;
@@ -85,10 +85,10 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         return;
       }
       String sql = split[1];
-      CalciteConnection conn = (CalciteConnection) sqlline.getConnection();
+      HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
       try {
-        RelRoot root = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root;
-        PipelineRel.Implementor plan = DeploymentService.plan(root);
+        RelRoot root = HoptimatorDriver.convert(conn, sql).root;
+        PipelineRel.Implementor plan = DeploymentService.plan(root, conn.connectionProperties());
         sqlline.output(plan.sql().apply(SqlDialect.ANSI));
       } catch (SQLException e) {
         sqlline.error(e);
@@ -142,7 +142,7 @@ public String matches(String line) {
 
     @Override
     public void execute(String line, DispatchCallback dispatchCallback) {
-      if (!(sqlline.getConnection() instanceof CalciteConnection)) {
+      if (!(sqlline.getConnection() instanceof HoptimatorConnection)) {
         sqlline.error("This connection doesn't support `!specify`.");
         dispatchCallback.setToFailure();
         return;
@@ -154,10 +154,11 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         return;
       }
       String sql = split[1];
-      CalciteConnection conn = (CalciteConnection) sqlline.getConnection();
-      RelRoot root = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root;
+      HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
+      RelRoot root = HoptimatorDriver.convert(conn, sql).root;
       try {
-        List<String> specs = DeploymentService.plan(root).pipeline("sink").specify();
+        List<String> specs = DeploymentService.plan(root, conn.connectionProperties())
+            .pipeline("sink").specify();
         specs.forEach(x -> sqlline.output(x + "\n\n---\n\n"));
       } catch (SQLException e) {
         sqlline.error(e);

hoptimator-demodb/build.gradle

@@ -1,9 +1,55 @@
 plugins {
   id 'java'
+  id 'maven-publish'
 }
 
 dependencies {
   implementation project(':hoptimator-api')
   implementation project(':hoptimator-util')
   implementation libs.calcite.core
 }
+
+publishing {
+  repositories {
+    maven {
+      name 'GitHubPackages'
+      url = 'https://maven.pkg.github.com/linkedin/Hoptimator'
+      credentials {
+        username = System.getenv('GITHUB_ACTOR')
+        password = System.getenv('GITHUB_TOKEN')
+      }
+    }
+    maven {
+      name 'LinkedInJFrog'
+      url 'https://linkedin.jfrog.io/artifactory/hoptimator'
+      credentials {
+        username = System.getenv('JFROG_USERNAME')
+        password = System.getenv('JFROG_API_KEY')
+      }
+    }
+  }
+  publications {
+    maven(MavenPublication) {
+      groupId = 'com.linkedin.hoptimator'
+      artifactId = 'hoptimator-demodb'
+      version = System.getenv('VERSION')
+      from components.java
+      pom {
+        name = 'hoptimator-api'
+        description = 'In-memory database driver for testing'
+        url = 'https://github.com/linkedin/Hoptimator'
+        licenses {
+          license {
+            name = 'BSD 2-Clause'
+            url = 'https://raw.githubusercontent.com/linkedin/Hoptimator/main/LICENSE'
+            }
+          }
+        scm {
+          connection = 'scm:git:git://github.com:linkedin/Hoptimator.git'
+          developerConnection = 'scm:git:ssh://github.com:linkedin/Hoptimator.git'
+          url = 'https://github.com/linkedin/Hoptimator'
+        }
+      }
+    }
+  }
+}

hoptimator-jdbc/build.gradle

@@ -54,6 +54,16 @@ java {
   targetCompatibility = JavaVersion.VERSION_11
 }
 
+// Work-around for gradle test fixtures bug, which spontaneously started publishing
+// junit-bom to maven, for some reason.
+components.java.withVariantsFromConfiguration(configurations.testFixturesApiElements) {
+  skip()
+}
+
+components.java.withVariantsFromConfiguration(configurations.testFixturesRuntimeElements) {
+  skip()
+}
+
 publishing {
   repositories {
     maven {
@@ -97,4 +107,4 @@ publishing {
       }
     }
   }
-}
+}

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

@@ -2,6 +2,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Properties;
 
 import com.linkedin.hoptimator.Catalog;
 import com.linkedin.hoptimator.CatalogProvider;

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

@@ -0,0 +1,41 @@
+package com.linkedin.hoptimator.jdbc;
+
+import java.sql.Connection;
+import java.sql.Statement;
+import java.sql.SQLException;
+import java.util.Properties;
+
+import org.apache.calcite.jdbc.CalciteConnection;
+import org.apache.calcite.jdbc.CalcitePrepare;
+
+import com.linkedin.hoptimator.util.DelegatingConnection;
+
+
+public class HoptimatorConnection extends DelegatingConnection {
+
+  private final CalciteConnection connection;
+  private final Properties connectionProperties;
+
+  public HoptimatorConnection(CalciteConnection connection, Properties connectionProperties) {
+    super(connection);
+    this.connection = connection;
+    this.connectionProperties = connectionProperties;
+  }
+
+  @Override
+  public Statement createStatement() throws SQLException {
+    return connection.createStatement(); 
+  }
+
+  public Properties connectionProperties() {
+    return connectionProperties;
+  }
+
+  public CalcitePrepare.Context createPrepareContext() {
+    return connection.createPrepareContext(); 
+  }
+
+  public CalciteConnection calciteConnection() {
+    return connection;
+  }
+}