extract methods, remove duplication, etc from PhysicalPlanBuilder (co…

…nfluentinc#356)

ksql-engine/src/main/java/io/confluent/ksql/QueryEngine.java

@@ -28,6 +28,7 @@
 import io.confluent.ksql.ddl.commands.DropSourceCommand;
 import io.confluent.ksql.ddl.commands.DropTopicCommand;
 import io.confluent.ksql.ddl.commands.RegisterTopicCommand;
+import io.confluent.ksql.metastore.MetastoreUtil;
 import io.confluent.ksql.serde.DataSource;
 import io.confluent.ksql.metastore.KsqlStream;
 import io.confluent.ksql.metastore.KsqlTable;
@@ -216,7 +217,7 @@ public void buildQueryPhysicalPlan(final List<QueryMetadata> physicalPlans,
     KsqlConfig ksqlConfigClone = ksqlEngine.getKsqlConfig().clone();
 
     // Build a physical plan, in this case a Kafka Streams DSL
-    PhysicalPlanBuilder physicalPlanBuilder = new PhysicalPlanBuilder(builder, ksqlConfigClone, ksqlEngine.getTopicClient());
+    PhysicalPlanBuilder physicalPlanBuilder = new PhysicalPlanBuilder(builder, ksqlConfigClone, ksqlEngine.getTopicClient(), new MetastoreUtil());
     SchemaKStream schemaKStream = physicalPlanBuilder.buildPhysicalPlan(logicalPlan);
 
     OutputNode outputNode = physicalPlanBuilder.getPlanSink();

ksql-engine/src/main/java/io/confluent/ksql/physical/AddTimestampColumnValueTransformerSupplier.java

@@ -0,0 +1,53 @@
+/**
+ * Copyright 2017 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+package io.confluent.ksql.physical;
+
+import org.apache.kafka.streams.kstream.ValueTransformer;
+import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+import io.confluent.ksql.GenericRow;
+
+class AddTimestampColumnValueTransformerSupplier implements ValueTransformerSupplier<GenericRow, GenericRow> {
+  @Override
+  public ValueTransformer<GenericRow, GenericRow> get() {
+    return new ValueTransformer<GenericRow, GenericRow>() {
+      ProcessorContext processorContext;
+      @Override
+      public void init(ProcessorContext processorContext) {
+        this.processorContext = processorContext;
+      }
+
+      @Override
+      public GenericRow transform(GenericRow row) {
+        if (row != null) {
+          row.getColumns().add(0, processorContext.timestamp());
+        }
+        return row;
+      }
+
+      @Override
+      public GenericRow punctuate(long l) {
+        return null;
+      }
+
+      @Override
+      public void close() {
+
+      }
+    };
+  }
+}

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/JoinNode.java

@@ -26,6 +26,7 @@
 
 public class JoinNode extends PlanNode {
 
+
   public enum Type {
     CROSS, INNER, LEFT, RIGHT, FULL, IMPLICIT
   }
@@ -124,4 +125,8 @@ public String getRightAlias() {
   public Type getType() {
     return type;
   }
+
+  public boolean isLeftJoin() {
+    return type == Type.LEFT;
+  }
 }

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

@@ -19,6 +19,9 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.confluent.ksql.metastore.KsqlTopic;
+import io.confluent.ksql.serde.KsqlTopicSerDe;
+import io.confluent.ksql.util.KsqlConfig;
+
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
 
@@ -74,4 +77,91 @@ public Map<String, Object> getOutputProperties() {
     return outputProperties;
   }
 
+  public KsqlTopicSerDe getTopicSerde() {
+    return ksqlTopic.getKsqlTopicSerDe();
+  }
+
+  public static class Builder {
+    private PlanNodeId id;
+    private PlanNode source;
+    private Schema schema;
+    private Field timestampField;
+    private Field keyField;
+    private KsqlTopic ksqlTopic;
+    private String topicName;
+    private Map<String, Object> outputProperties;
+    private Optional<Integer> limit;
+
+    public KsqlStructuredDataOutputNode build() {
+      return new KsqlStructuredDataOutputNode(id,
+          source,
+          schema,
+          timestampField,
+          keyField,
+          ksqlTopic,
+          topicName,
+          outputProperties,
+          limit);
+    }
+
+    public static Builder from(final KsqlStructuredDataOutputNode original) {
+      return new Builder()
+          .withId(original.getId())
+          .withSource(original.getSource())
+          .withSchema(original.getSchema())
+          .withTimestampField(original.getTimestampField())
+          .withKeyField(original.getKeyField())
+          .withKsqlTopic(original.getKsqlTopic())
+          .withTopicName(original.getKafkaTopicName())
+          .withOutputProperties(original.getOutputProperties())
+          .withLimit(original.getLimit());
+    }
+
+    public Builder withLimit(final Optional<Integer> limit) {
+      this.limit = limit;
+      return this;
+    }
+
+    public Builder withOutputProperties(Map<String, Object> outputProperties) {
+      this.outputProperties = outputProperties;
+      return this;
+    }
+
+    public Builder withTopicName(String topicName) {
+      this.topicName = topicName;
+      return this;
+    }
+
+    public Builder withKsqlTopic(KsqlTopic ksqlTopic) {
+      this.ksqlTopic = ksqlTopic;
+      return this;
+    }
+
+    public Builder withKeyField(Field keyField) {
+      this.keyField = keyField;
+      return this;
+    }
+
+    public Builder withTimestampField(Field timestampField) {
+      this.timestampField = timestampField;
+      return this;
+    }
+
+    public Builder withSchema(Schema schema) {
+      this.schema = schema;
+      return this;
+    }
+
+    public Builder withSource(PlanNode source) {
+      this.source = source;
+      return this;
+    }
+
+    public Builder withId(final PlanNodeId id) {
+      this.id = id;
+      return this;
+    }
+
+  }
+
 }

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKStream.java

@@ -167,8 +167,7 @@ public SchemaKStream select(final List<Pair<String, Expression>> expressionPairL
                 newColumns.add(null);
               }
             }
-            GenericRow newRow = new GenericRow(newColumns);
-            return newRow;
+            return new GenericRow(newColumns);
           } catch (Exception e) {
             log.error("Projection exception for row: " + row.toString());
             log.error(e.getMessage(), e);

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java

@@ -22,6 +22,7 @@
 import io.confluent.ksql.analyzer.AnalysisContext;
 import io.confluent.ksql.analyzer.Analyzer;
 import io.confluent.ksql.metastore.MetaStore;
+import io.confluent.ksql.metastore.MetastoreUtil;
 import io.confluent.ksql.parser.KsqlParser;
 import io.confluent.ksql.util.AggregateExpressionRewriter;
 import io.confluent.ksql.parser.tree.Expression;
@@ -63,7 +64,7 @@ public void before() {
         configMap.put("commit.interval.ms", 0);
         configMap.put("cache.max.bytes.buffering", 0);
         configMap.put("auto.offset.reset", "earliest");
-        physicalPlanBuilder = new PhysicalPlanBuilder(kStreamBuilder, new KsqlConfig(configMap), new FakeKafkaTopicClient());
+        physicalPlanBuilder = new PhysicalPlanBuilder(kStreamBuilder, new KsqlConfig(configMap), new FakeKafkaTopicClient(), new MetastoreUtil());
     }
 
     private SchemaKStream buildPhysicalPlan(String queryStr) throws Exception {