Fix multiple owner references (#1)

* Fix multiple owner references

* Add additional printer columns to subscription CRD

Author: ryannedolan

Makefile

@@ -1,5 +1,5 @@
 
-
+bounce: build undeploy deploy
 
 build:
 	./gradlew build
@@ -15,6 +15,9 @@ integration-tests:
 clean:
 	./gradlew clean
 
+undeploy:
+	kubectl delete -f ./deploy || echo "skipping"
+
 quickstart: build deploy-dev-environment deploy
 
 deploy-dev-environment: 
@@ -29,13 +32,13 @@ deploy-dev-environment:
 deploy:
 	kubectl apply -f ./deploy/
 
-undeploy:
-	kubectl delete -f ./deploy || echo "skipping"
-
 deploy-samples:
 	kubectl apply -f ./deploy/samples
 
 release:
 	./gradlew publish
 
-.PHONY: build clean quickstart deploy-dev-environment deploy deploy-samples integration-tests
+generate-models:
+	./models/generate-models.sh
+
+.PHONY: build clean quickstart deploy-dev-environment deploy deploy-samples integration-tests bounce generate-models

deploy/kafkatopics.crd.yaml

@@ -32,7 +32,7 @@ spec:
               description: Desired Kafka topic configuration.
               type: object
               properties:
-                name:
+                topicName:
                   description: The topic name.
                   type: string
                 numPartitions:
@@ -67,7 +67,7 @@ spec:
                   additionalProperties:
                     type: string
               required:
-              - name
+              - topicName
             status:
               description: Current state of the topic.
               type: object

deploy/samples/kafkatopic.yaml

@@ -0,0 +1,29 @@
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: Subscription
+metadata:
+  name: people
+spec:
+  sql: SELECT NAME FROM DATAGEN.PERSON
+  database: RAWKAFKA
+
+---
+
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: Subscription
+metadata:
+  name: companies
+spec:
+  sql: SELECT NAME FROM DATAGEN.COMPANY
+  database: RAWKAFKA
+
+---
+
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: Subscription
+metadata:
+  name: names
+spec:
+  sql: SELECT p.PAYLOAD || c.PAYLOAD FROM RAWKAFKA."people" p, RAWKAFKA."companies" c
+  database: RAWKAFKA
+
+

deploy/samples/subscription.yaml

@@ -53,3 +53,13 @@ spec:
                   type: string
       subresources:
         status: {}
+      additionalPrinterColumns:
+      - name: DB
+        type: string
+        description: The database where the subscription is materialized.
+        jsonPath: .spec.database
+      - name: SQL
+        type: string
+        description: The SQL query that the subscription materializes.
+        jsonPath: .spec.sql
+

deploy/samples/subscriptions.yaml

@@ -21,11 +21,11 @@ public final class AvroConverter {
   private AvroConverter() {
   }
 
-  public static Schema avro(String name, String namespace, RelDataType dataType) {
+  public static Schema avro(String namespace, String name, RelDataType dataType) {
     if (dataType.isStruct()) {
       List<Schema.Field> fields = dataType.getFieldList().stream()
         .filter(x -> !x.getName().startsWith("__")) // don't write out hidden fields
-        .map(x -> new Schema.Field(sanitize(x.getName()), avro(x.getName(), namespace, x.getType()), describe(x), null))
+        .map(x -> new Schema.Field(sanitize(x.getName()), avro(namespace, x.getName(), x.getType()), describe(x), null))
         .collect(Collectors.toList());
       return Schema.createRecord(sanitize(name), dataType.toString(), namespace, false, fields);
     } else {
@@ -49,9 +49,9 @@ public static Schema avro(String name, String namespace, RelDataType dataType) {
     }
   }
 
-  public static Schema avro(String name, String namespace, RelProtoDataType relProtoDataType) {
+  public static Schema avro(String namespace, String name, RelProtoDataType relProtoDataType) {
     RelDataTypeFactory factory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
-    return avro(name, namespace, relProtoDataType.apply(factory));
+    return avro(namespace, name, relProtoDataType.apply(factory));
   }
 
   private static Schema createAvroTypeWithNullability(Schema.Type rawType, boolean nullable) {

deploy/subscriptions.crd.yaml

@@ -28,12 +28,6 @@ public abstract class Resource {
   private final String template;
   private final Map<String, Supplier<String>> properties = new HashMap<>();
 
-  /** A Resource that will be rendered with the specified template */
-  public Resource(String name, String template) {
-    this(template);
-    export("name", name);
-  }
-
   /** A Resource that will be rendered with the specified template */
   public Resource(String template) {
     this.template = template;
@@ -73,10 +67,6 @@ public Set<String> keys() {
     return properties.keySet();
   }
 
-  public String name() {
-    return getOrDefault("name", () -> "(no name)");
-  }
-
   /** Render this Resource using the given TemplateFactory */
   public String render(TemplateFactory templateFactory) {
     return templateFactory.get(this).render(this);
@@ -103,7 +93,7 @@ public String toString() {
   public String getOrDefault(String key, Supplier<String> f) {
     String computed = properties.getOrDefault(key, f).get();
     if (computed == null) {
-      throw new IllegalArgumentException("Resource '" + name() + "' missing variable '" + key + "'");
+      throw new IllegalArgumentException("Resource missing variable '" + key + "'");
     }
     return computed;
   }

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

@@ -98,7 +98,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
       try {
         HoptimatorPlanner planner = HoptimatorPlanner.fromModelFile(connectionUrl, new Properties());
         RelNode plan = planner.logical(sql);
-        String avroSchema = AvroConverter.avro("OutputRecord", "OutputNamespace", plan.getRowType()).toString(true);
+        String avroSchema = AvroConverter.avro("OutputNamespace", "OutputName", plan.getRowType()).toString(true);
         sqlline.output(avroSchema); 
         dispatchCallback.setToSuccess();
       } catch (Exception e) {

hoptimator-catalog/src/main/java/com/linkedin/hoptimator/catalog/Resource.java

@@ -8,14 +8,11 @@
 
 public class KafkaTopic extends Resource {
   public KafkaTopic(String name, Integer numPartitions, Map<String, String> clientOverrides) {
-    super(name, "KafkaTopic");
+    super("KafkaTopic");
+    export("name", name);
     export("numPartitions", Optional.ofNullable(numPartitions)
       .map(x -> Integer.toString(x)).orElse("null"));
     export("clientOverrides", clientOverrides);
   }
-
-  public KafkaTopic(String name) {
-    this(name, null, Collections.emptyMap());
-  }
 }