Add ResolvedTable (#151)

Author: ryannedolan

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

@@ -2,6 +2,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -19,6 +20,7 @@
 import com.linkedin.hoptimator.SqlDialect;
 import com.linkedin.hoptimator.jdbc.HoptimatorConnection;
 import com.linkedin.hoptimator.jdbc.HoptimatorDriver;
+import com.linkedin.hoptimator.jdbc.ResolvedTable;
 import com.linkedin.hoptimator.util.DeploymentService;
 import com.linkedin.hoptimator.util.planner.PipelineRel;
 
@@ -41,6 +43,7 @@ public Collection<CommandHandler> getCommandHandlers(SqlLine sqlline) {
     Collection<CommandHandler> list = new ArrayList<>(super.getCommandHandlers(sqlline));
     list.add(new IntroCommandHandler(sqlline));
     list.add(new PipelineCommandHandler(sqlline));
+    list.add(new ResolveCommandHandler(sqlline));
     list.add(new SpecifyCommandHandler(sqlline));
     return list;
   }
@@ -118,6 +121,75 @@ public boolean echoToFile() {
     }
   }
 
+  private static final class ResolveCommandHandler implements CommandHandler {
+
+    private final SqlLine sqlline;
+
+    private ResolveCommandHandler(SqlLine sqlline) {
+      this.sqlline = sqlline;
+    }
+
+    @Override
+    public String getName() {
+      return "resolve";
+    }
+
+    @Override
+    public List<String> getNames() {
+      return Collections.singletonList(getName());
+    }
+
+    @Override
+    public String getHelpText() {
+      return "Resolve a table path.";
+    }
+
+    @Override
+    public String matches(String line) {
+      if (startsWith(line, "!resolve") || startsWith(line, "resolve")) {
+        return line;
+      } else {
+        return null;
+      }
+    }
+
+    @Override
+    public void execute(String line, DispatchCallback dispatchCallback) {
+      if (!(sqlline.getConnection() instanceof HoptimatorConnection)) {
+        sqlline.error("This connection doesn't support `!resolve`.");
+        dispatchCallback.setToFailure();
+        return;
+      }
+      String[] split = line.split("\\s+", 2);
+      if (split.length < 2) {
+        sqlline.error("Missing argument.");
+        dispatchCallback.setToFailure();
+        return;
+      }
+      List<String> tablePath = Arrays.asList(split[1].split("\\."));
+      HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
+      try {
+        ResolvedTable resolved = conn.resolve(tablePath, Collections.emptyMap());
+        sqlline.output("Avro schema:\n");
+        sqlline.output(resolved.avroSchemaString());
+      } catch (SQLException e) {
+        sqlline.error(e);
+        dispatchCallback.setToFailure();
+      }
+    }
+
+    @Override
+    public List<Completer> getParameterCompleters() {
+      return Collections.emptyList();
+    }
+
+    @Override
+    public boolean echoToFile() {
+      return false;
+    }
+  }
+
+
   private static final class SpecifyCommandHandler implements CommandHandler {
 
     private final SqlLine sqlline;

hoptimator-jdbc/build.gradle

@@ -6,7 +6,9 @@ plugins {
 
 dependencies {
   implementation project(':hoptimator-api')
+  implementation project(':hoptimator-avro')
   implementation project(':hoptimator-util')
+  implementation libs.avro
   implementation libs.calcite.core
   implementation libs.calcite.server
   implementation libs.slf4j.api

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

@@ -4,7 +4,10 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
 
@@ -13,6 +16,7 @@
 import org.apache.calcite.plan.RelOptMaterialization;
 import org.apache.calcite.rel.RelNode;
 
+import com.linkedin.hoptimator.avro.AvroConverter;
 import com.linkedin.hoptimator.util.DelegatingConnection;
 
 
@@ -28,6 +32,26 @@ public HoptimatorConnection(CalciteConnection connection, Properties connectionP
     this.connectionProperties = connectionProperties;
   }
 
+  public ResolvedTable resolve(List<String> tablePath, Map<String, String> hints) throws SQLException {
+    try {
+      String tableSql = "SELECT * FROM " + tablePath.stream()
+          .map(x -> "\"" + x + "\"")
+          .collect(Collectors.joining("."));
+      RelNode tableRel = HoptimatorDriver.convert(this, tableSql).root.rel;
+      String namespace = "com.linkedin."
+            + tablePath.stream()
+            .map(x -> x.toLowerCase(Locale.ROOT))
+            .limit(tablePath.size() - 1)
+            .collect(Collectors.joining("."));
+      String schemaName = tablePath.get(tablePath.size() - 1).toLowerCase(Locale.ROOT);
+      org.apache.avro.Schema avroSchema = AvroConverter.avro(namespace, schemaName, tableRel.getRowType());
+      // TODO: generate source and sink configs via ConnectionService
+      return new ResolvedTable(tablePath, avroSchema, Collections.emptyMap(), Collections.emptyMap());
+    } catch (Exception e) {
+      throw new SQLException("Failed to resolve " + String.join(".", tablePath) + ": " + e.getMessage(), e);
+    }
+  }
+
   @Override
   public Statement createStatement() throws SQLException {
     return connection.createStatement();

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

@@ -0,0 +1,37 @@
+package com.linkedin.hoptimator.jdbc;
+
+import java.util.List;
+import java.util.Map;
+
+/** Metadata required to access a table in the data plane. */
+public class ResolvedTable {
+
+  private final List<String> tablePath;
+  private org.apache.avro.Schema avroSchema;
+  private final Map<String, String> sourceConnectorConfigs;
+  private final Map<String, String> sinkConnectorConfigs;
+
+  public ResolvedTable(List<String> tablePath, org.apache.avro.Schema avroSchema,
+      Map<String, String> sourceConnectorConfigs, Map<String, String> sinkConnectorConfigs) {
+    this.tablePath = tablePath;
+    this.avroSchema = avroSchema;
+    this.sourceConnectorConfigs = sourceConnectorConfigs;
+    this.sinkConnectorConfigs = sinkConnectorConfigs;
+  }
+
+  public Map<String, String> sourceConnectorConfigs() {
+    return sourceConnectorConfigs;
+  }
+
+  public Map<String, String> sinkConnectorConfigs() {
+    return sinkConnectorConfigs;
+  }
+
+  public org.apache.avro.Schema avroSchema() {
+    return avroSchema;
+  }
+
+  public String avroSchemaString() {
+    return avroSchema.toString(true);
+  }
+}