Add create table support, venice deployer, and various fixes (#182)

* Add base CREATE TABLE

* Add Venice Deployer

* Fix PipelineReconciler and AvroConverter defaults

* Add drop tests

* Fix failing tests caused by specify closing connections prematurely

* Add back some comments about copy-pasted code

* Add log

Author: jogrogan

.gitignore

@@ -3,6 +3,7 @@
 /build
 */build/
 */*.iml
+*bin/
 ./models/external/
 .DS_Store
 

deploy/config/hoptimator-configmap.yaml

@@ -22,3 +22,7 @@ data:
   flink.config: |
     flink.app.name=hoptimator-flink-runner
     flink.app.type=SQL
+
+  venice.config: |
+    router.url=http://localhost:7777
+    clusters=venice-cluster0

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroConverter.java

@@ -35,7 +35,15 @@ public static Schema avro(String namespace, String name, RelDataType dataType) {
     String newNamespace = namespace + "." + name;
     if (dataType.isStruct()) {
       List<Schema.Field> fields = dataType.getFieldList().stream()
-          .map(x -> new Schema.Field(sanitize(x.getName()), avro(newNamespace, x.getName(), x.getType()), describe(x), null))
+          .map(x -> {
+            Schema innerField = avro(newNamespace, x.getName(), x.getType());
+            Object defaultValue = null;
+            // For unions containing null, defaults are specified in a specific way
+            if (innerField.isUnion() && innerField.isNullable()) {
+              defaultValue = Schema.Field.NULL_DEFAULT_VALUE;
+            }
+            return new Schema.Field(sanitize(x.getName()), innerField, describe(x), defaultValue);
+          })
           .collect(Collectors.toList());
       return createAvroSchemaWithNullability(Schema.createRecord(sanitize(name), dataType.toString(), newNamespace, false, fields),
           dataType.isNullable());
@@ -150,6 +158,7 @@ public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory) {
   /** Converts Avro Schema to RelDataType.
    * Nullability is preserved except for array types, JDBC is incapable of interpreting e.g. "FLOAT NOT NULL ARRAY"
    * causing "NOT NULL" arrays to get demoted to "ANY ARRAY" which is not desired.
+   * TODO: default field values are lost when converting from Avro to RelDataType
    */
   public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory, boolean nullable) {
     RelDataType unknown = typeFactory.createUnknownType();

hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroConverterTest.java

@@ -4,6 +4,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import org.apache.avro.JsonProperties;
 import org.apache.avro.Schema;
 import org.apache.calcite.plan.RelOptUtil;
 import org.apache.calcite.rel.type.RelDataType;
@@ -47,6 +48,7 @@ public void convertsNestedSchemas() {
     Schema avroSchema4 = AvroConverter.avro("NS", "R", rel1);
     assertFalse(avroSchema4.isNullable());
     assertEquals(avroSchema4.getFields().size(), rel1.getFieldCount());
+    assertEquals(JsonProperties.NULL_VALUE, avroSchema4.getField("h").defaultVal());
     Schema avroSchema5 = AvroConverter.avro("NS", "R", rel2);
     assertTrue(avroSchema5.isNullable());
     assertEquals(avroSchema5.getTypes().get(1).getFields().size(), rel2.getFieldCount());

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

@@ -60,7 +60,7 @@ private HoptimatorDdlUtils() {
   // N.B. copy-pasted from Apache Calcite
   /** Returns the schema in which to create an object;
    * the left part is null if the schema does not exist. */
-  static Pair<CalciteSchema, String> schema(CalcitePrepare.Context context, boolean mutable, SqlIdentifier id) {
+  public static Pair<CalciteSchema, String> schema(CalcitePrepare.Context context, boolean mutable, SqlIdentifier id) {
     final String name;
     final List<String> path;
     if (id.isSimple()) {
@@ -92,7 +92,7 @@ public static SqlNode renameColumns(SqlNodeList columnList, SqlNode query) {
   }
 
   // N.B. copy-pasted from Apache Calcite
-  static ViewTable viewTable(CalcitePrepare.Context context, String sql, CalcitePrepareImpl impl,
+  public static ViewTable viewTable(CalcitePrepare.Context context, String sql, CalcitePrepareImpl impl,
       List<String> schemaPath, List<String> viewPath) {
     CalcitePrepare.AnalyzeViewResult analyzed = impl.analyzeView(context, sql, false);
     RelProtoDataType protoType = RelDataTypeImpl.proto(analyzed.rowType);

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

@@ -45,12 +45,20 @@ public void createView() throws Exception {
     var expectedLogs = List.of(
         "[HoptimatorDdlExecutor] Validating statement: CREATE VIEW `V` AS\nSELECT *\nFROM `T`",
         "[HoptimatorDdlExecutor] Validated sql statement. The view is named V and has path [DEFAULT, V]",
-        "[HoptimatorDdlExecutor] Validating view V with deployers",
+        "[HoptimatorDdlExecutor] Validating deployable resources for view V",
         "[HoptimatorDdlExecutor] Validated view V",
         "[HoptimatorDdlExecutor] Deploying create view V",
         "[HoptimatorDdlExecutor] Deployed view V",
         "[HoptimatorDdlExecutor] Added view V to schema DEFAULT",
-        "[HoptimatorDdlExecutor] CREATE VIEW V completed");
+        "[HoptimatorDdlExecutor] CREATE VIEW V completed",
+        "[HoptimatorDdlExecutor] Validating statement: DROP VIEW `V`",
+        "[HoptimatorDdlExecutor] Deleting view V",
+        "[HoptimatorDdlExecutor] Removed view V from schema DEFAULT",
+        "[HoptimatorDdlExecutor] DROP VIEW `V` completed",
+        "[HoptimatorDdlExecutor] Validating statement: DROP TABLE `T`",
+        "[HoptimatorDdlExecutor] Deleting table T",
+        "[HoptimatorDdlExecutor] Removed table T from schema DEFAULT",
+        "[HoptimatorDdlExecutor] DROP TABLE `T` completed");
     Assertions.assertEquals(expectedLogs, logs);
   }
 
@@ -64,7 +72,7 @@ public void dropNonExistentViewHandlesNullSchema() throws Exception {
       sql("DROP VIEW non_existing_schema.non_existing_view");
     });
     Assertions.assertTrue(
-        ex.getMessage().matches("(?s).*Cannot DROP VIEW .*?: View .*? not found\\..*"),
+        ex.getMessage().matches("(?s).*Cannot DROP VIEW .*?: Element .*? not found\\..*"),
         "Error message should match regex, but was: " + ex.getMessage()
     );
   }

hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TestBasicSql.java

@@ -118,6 +118,25 @@ select * from ads."PAGE_VIEWS$filter";
 
 !ok
 
+create or replace table ads."newtable" ("i" int, "s" VARCHAR);
+(0 rows modified)
+
+!update
+
+select * from ads."newtable";
++---+---+
+| i | s |
++---+---+
++---+---+
+(0 rows)
+
+!ok
+
+drop table ads."newtable";
+(0 rows modified)
+
+!update
+
 drop materialized view ads."PAGE_VIEWS$filter";
 (0 rows modified)
 

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

@@ -115,12 +115,12 @@ drop materialized view ads.pages;
 
 !update
 
-drop materialized view ads.audience;
+drop materialized view ads.audience2;
 (0 rows modified)
 
 !update
 
-drop materialized view ads.audience2;
+drop view ads.audience;
 (0 rows modified)
 
 !update

hoptimator-k8s/src/test/resources/k8s-metadata.id

@@ -5,6 +5,10 @@ create or replace materialized view ads.page_views as select first_name, last_na
 Cannot overwrite physical table PAGE_VIEWS with a view.
 !error
 
+create table ads.page_views ("i" int);
+Table PAGE_VIEWS already exists.
+!error
+
 create or replace materialized view invalid.page_views as select first_name, last_name from profile.members;
 Schema for INVALID.PAGE_VIEWS not found.
 !error
@@ -18,11 +22,23 @@ create materialized view ads.audience as select first_name, last_name from ads.p
 View AUDIENCE already exists. Use CREATE OR REPLACE to update.
 !error
 
+drop table ads.audience;
+AUDIENCE is a materialized view and does not correspond to DROP TABLE
+!error
+
 drop materialized view ads.audience;
 (0 rows modified)
 
 !update
 
+drop materialized view ads.audience;
+Element AUDIENCE not found
+!error
+
+drop materialized view ads.page_views;
+PAGE_VIEWS is a table and does not correspond to DROP MATERIALIZED VIEW
+!error
+
 create or replace materialized view ads."PAGE_VIEWS$myview" as select first_name from profile.members;
 Field FIRST_NAME not found in sink schema
 !error