Master

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

@@ -30,14 +30,13 @@
 import java.util.List;
 import java.util.Map;
 
-import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.PersistentQueryMetadata;
 import io.confluent.ksql.util.QueryMetadata;
 
 public class KsqlContext {
 
   private static final Logger log = LoggerFactory.getLogger(KsqlContext.class);
-  final KsqlEngine ksqlEngine;
+  private final KsqlEngine ksqlEngine;
   private static final String APPLICATION_ID_OPTION_DEFAULT = "ksql_standalone_cli";
   private static final String KAFKA_BOOTSTRAP_SERVER_OPTION_DEFAULT = "localhost:9092";
   private final AdminClient adminClient;
@@ -53,7 +52,7 @@ public KsqlContext() {
    *
    * @param streamsProperties
    */
-  public KsqlContext(Map<String, Object> streamsProperties) {
+  KsqlContext(Map<String, Object> streamsProperties) {
     if (streamsProperties == null) {
       streamsProperties = new HashMap<>();
     }
@@ -104,24 +103,4 @@ public void close() throws IOException {
     topicClient.close();
     adminClient.close();
   }
-
-  /**
-   * Terminate a query with the given id.
-   *
-   * @param queryId
-   */
-  public void terminateQuery(long queryId) {
-    if (!ksqlEngine.getPersistentQueries().containsKey(queryId)) {
-      throw new KsqlException(String.format("Invalid query id. Query id, %d, does not exist.",
-                                            queryId));
-    }
-    PersistentQueryMetadata persistentQueryMetadata = ksqlEngine
-        .getPersistentQueries().get(queryId);
-    persistentQueryMetadata.getKafkaStreams().close();
-    ksqlEngine.getPersistentQueries().remove(queryId);
-  }
-
-  public Map<Long, PersistentQueryMetadata> getRunningQueries() {
-    return ksqlEngine.getPersistentQueries();
-  }
 }

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

@@ -370,10 +370,6 @@ public Map<Long, PersistentQueryMetadata> getPersistentQueries() {
     return new HashMap<>(persistentQueries);
   }
 
-  public Set<QueryMetadata> getLiveQueries() {
-    return new HashSet<>(liveQueries);
-  }
-
   public static List<String> getImmutableProperties() {
     return new ArrayList<>(IMMUTABLE_PROPERTIES);
   }

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

@@ -88,13 +88,13 @@ public class QueryEngine {
   private final KsqlEngine ksqlEngine;
 
 
-  public QueryEngine(final KsqlEngine ksqlEngine) {
+  QueryEngine(final KsqlEngine ksqlEngine) {
     this.queryIdCounter = new AtomicLong(1);
     this.ksqlEngine = ksqlEngine;
   }
 
 
-  public List<Pair<String, PlanNode>> buildLogicalPlans(
+  List<Pair<String, PlanNode>> buildLogicalPlans(
       final MetaStore metaStore,
       final List<Pair<String, Statement>> statementList) {
 
@@ -116,20 +116,20 @@ public List<Pair<String, PlanNode>> buildLogicalPlans(
     return logicalPlansList;
   }
 
-  public PlanNode buildQueryLogicalPlan(final Query query, final MetaStore tempMetaStore) {
+  private PlanNode buildQueryLogicalPlan(final Query query, final MetaStore tempMetaStore) {
 
     // Analyze the query to resolve the references and extract operations
     Analysis analysis = new Analysis();
     Analyzer analyzer = new Analyzer(analysis, tempMetaStore);
-    analyzer.process(query, new AnalysisContext(null, null));
+    analyzer.process(query, new AnalysisContext(null));
 
     AggregateAnalysis aggregateAnalysis = new AggregateAnalysis();
     AggregateAnalyzer aggregateAnalyzer = new
-        AggregateAnalyzer(aggregateAnalysis, tempMetaStore, analysis);
+        AggregateAnalyzer(aggregateAnalysis, analysis);
     AggregateExpressionRewriter aggregateExpressionRewriter = new AggregateExpressionRewriter();
     for (Expression expression: analysis.getSelectExpressions()) {
       aggregateAnalyzer
-          .process(expression, new AnalysisContext(null, null));
+          .process(expression, new AnalysisContext(null));
       if (!aggregateAnalyzer.isHasAggregateFunction()) {
         aggregateAnalysis.getNonAggResultColumns().add(expression);
       }
@@ -145,7 +145,7 @@ public PlanNode buildQueryLogicalPlan(final Query query, final MetaStore tempMet
     // TODO: make sure only aggregates are in the expression. For now we assume this is the case.
     if (analysis.getHavingExpression() != null) {
       aggregateAnalyzer.process(analysis.getHavingExpression(),
-                                new AnalysisContext(null, null));
+                                new AnalysisContext(null));
       if (!aggregateAnalyzer.isHasAggregateFunction()) {
         aggregateAnalysis.getNonAggResultColumns().add(analysis.getHavingExpression());
       }

ksql-engine/src/main/java/io/confluent/ksql/analyzer/AggregateAnalysis.java

@@ -26,18 +26,14 @@
 
 public class AggregateAnalysis {
 
+  private List<Expression> requiredColumnsList = new ArrayList<>();
+  private Expression havingExpression = null;
+  private Map<String, Expression> requiredColumnsMap = new HashMap<>();
+  private List<Expression> nonAggResultColumns = new ArrayList<>();
+  private List<Expression> finalSelectExpressions = new ArrayList<>();
   List<Expression> aggregateFunctionArguments = new ArrayList<>();
-  List<Expression> requiredColumnsList = new ArrayList<>();
-
-  Expression havingExpression = null;
-
-  Map<String, Expression> requiredColumnsMap = new HashMap<>();
-
   List<FunctionCall> functionList = new ArrayList<>();
 
-  List<Expression> nonAggResultColumns = new ArrayList<>();
-
-  List<Expression> finalSelectExpressions = new ArrayList<>();
 
   public List<Expression> getAggregateFunctionArguments() {
     return aggregateFunctionArguments;

ksql-engine/src/main/java/io/confluent/ksql/analyzer/AggregateAnalyzer.java

@@ -17,7 +17,6 @@
 package io.confluent.ksql.analyzer;
 
 import io.confluent.ksql.function.KsqlFunctions;
-import io.confluent.ksql.metastore.MetaStore;
 import io.confluent.ksql.parser.tree.DereferenceExpression;
 import io.confluent.ksql.parser.tree.Expression;
 import io.confluent.ksql.parser.tree.FunctionCall;
@@ -30,7 +29,6 @@
 public class AggregateAnalyzer extends DefaultTraversalVisitor<Node, AnalysisContext> {
 
   private AggregateAnalysis aggregateAnalysis;
-  private MetaStore metaStore;
   private Analysis analysis;
 
   private boolean hasAggregateFunction = false;
@@ -43,10 +41,9 @@ public void setHasAggregateFunction(boolean hasAggregateFunction) {
     this.hasAggregateFunction = hasAggregateFunction;
   }
 
-  public AggregateAnalyzer(AggregateAnalysis aggregateAnalysis, MetaStore metaStore,
+  public AggregateAnalyzer(AggregateAnalysis aggregateAnalysis,
                            Analysis analysis) {
     this.aggregateAnalysis = aggregateAnalysis;
-    this.metaStore = metaStore;
     this.analysis = analysis;
   }
 

ksql-engine/src/main/java/io/confluent/ksql/analyzer/Analysis.java

@@ -34,7 +34,6 @@ public class Analysis {
   private Map<String, Object> intoProperties = new HashMap<>();
   private String intoFormat = null;
   // TODO: Maybe have all as properties. At the moment this will only be set if format is avro.
-  private String intoAvroSchemaFilePath = null;
   private String intoKafkaTopicName = null;
   private List<Pair<StructuredDataSource, String>> fromDataSources = new ArrayList<>();
   private JoinNode join;
@@ -50,7 +49,7 @@ public class Analysis {
   private Optional<Integer> limitClause = Optional.empty();
 
 
-  public void addSelectItem(final Expression expression, final String alias) {
+  void addSelectItem(final Expression expression, final String alias) {
     selectExpressions.add(expression);
     selectExpressionAlias.add(alias);
   }
@@ -68,10 +67,6 @@ public List<Pair<StructuredDataSource, String>> getFromDataSources() {
     return fromDataSources;
   }
 
-  public void setFromDataSources(List<Pair<StructuredDataSource, String>> fromDataSources) {
-    this.fromDataSources = fromDataSources;
-  }
-
   public Expression getWhereExpression() {
     return whereExpression;
   }
@@ -84,18 +79,10 @@ public List<Expression> getSelectExpressions() {
     return selectExpressions;
   }
 
-  public void setSelectExpressions(List<Expression> selectExpressions) {
-    this.selectExpressions = selectExpressions;
-  }
-
   public List<String> getSelectExpressionAlias() {
     return selectExpressionAlias;
   }
 
-  public void setSelectExpressionAlias(List<String> selectExpressionAlias) {
-    this.selectExpressionAlias = selectExpressionAlias;
-  }
-
   public JoinNode getJoin() {
     return join;
   }
@@ -120,22 +107,10 @@ public String getIntoKafkaTopicName() {
     return intoKafkaTopicName;
   }
 
-  public String getIntoAvroSchemaFilePath() {
-    return intoAvroSchemaFilePath;
-  }
-
-  public void setIntoAvroSchemaFilePath(String intoAvroSchemaFilePath) {
-    this.intoAvroSchemaFilePath = intoAvroSchemaFilePath;
-  }
-
   public List<Expression> getGroupByExpressions() {
     return groupByExpressions;
   }
 
-  public void setGroupByExpressions(List<Expression> groupByExpressions) {
-    this.groupByExpressions = groupByExpressions;
-  }
-
   public WindowExpression getWindowExpression() {
     return windowExpression;
   }

ksql-engine/src/main/java/io/confluent/ksql/analyzer/AnalysisContext.java

@@ -16,8 +16,6 @@
 
 package io.confluent.ksql.analyzer;
 
-import io.confluent.ksql.parser.tree.Node;
-
 public class AnalysisContext {
 
   public enum ParentType {
@@ -38,18 +36,12 @@ public String getValue() {
     }
   }
 
-  final Node parentNode;
-  final ParentType parentType;
+  private final ParentType parentType;
 
-  public AnalysisContext(final Node parentNode, final ParentType parentType) {
-    this.parentNode = parentNode;
+  public AnalysisContext(final ParentType parentType) {
     this.parentType = parentType;
   }
 
-  public Node getParentNode() {
-    return parentNode;
-  }
-
   public ParentType getParentType() {
     return parentType;
   }

ksql-engine/src/main/java/io/confluent/ksql/analyzer/Analyzer.java

@@ -78,29 +78,29 @@ protected Node visitQuerySpecification(final QuerySpecification node,
                                          final AnalysisContext context) {
 
     process(node.getFrom().get(),
-            new AnalysisContext(null, AnalysisContext.ParentType.FROM));
+            new AnalysisContext(AnalysisContext.ParentType.FROM));
 
-    process(node.getInto().get(), new AnalysisContext(null,
-                                                      AnalysisContext.ParentType.INTO));
+    process(node.getInto().get(), new AnalysisContext(
+        AnalysisContext.ParentType.INTO));
     if (!(analysis.getInto() instanceof KsqlStdOut)) {
       analyzeNonStdOutSink();
     }
 
-    process(node.getSelect(), new AnalysisContext(null,
-                                                  AnalysisContext.ParentType.SELECT));
+    process(node.getSelect(), new AnalysisContext(
+        AnalysisContext.ParentType.SELECT));
     if (node.getWhere().isPresent()) {
-      analyzeWhere(node.getWhere().get(), context);
+      analyzeWhere(node.getWhere().get());
     }
     if (node.getGroupBy().isPresent()) {
-      analyzeGroupBy(node.getGroupBy().get(), context);
+      analyzeGroupBy(node.getGroupBy().get());
     }
 
     if (node.getWindowExpression().isPresent()) {
-      analyzeWindowExpression(node.getWindowExpression().get(), context);
+      analyzeWindowExpression(node.getWindowExpression().get());
     }
 
     if (node.getHaving().isPresent()) {
-      analyzeHaving(node.getHaving().get(), context);
+      analyzeHaving(node.getHaving().get());
     }
 
     if (node.getLimit().isPresent()) {
@@ -116,7 +116,7 @@ protected Node visitQuerySpecification(final QuerySpecification node,
   private void analyzeNonStdOutSink() {
     List<Pair<StructuredDataSource, String>> fromDataSources = analysis.getFromDataSources();
 
-    StructuredDataSource intoStructuredDataSource = (StructuredDataSource) analysis.getInto();
+    StructuredDataSource intoStructuredDataSource = analysis.getInto();
     String intoKafkaTopicName = analysis.getIntoKafkaTopicName();
     if (intoKafkaTopicName == null) {
       intoKafkaTopicName = intoStructuredDataSource.getName();
@@ -408,28 +408,22 @@ protected Node visitGroupBy(final GroupBy node, final AnalysisContext context) {
     return null;
   }
 
-  private StructuredDataSource analyzeFrom(final QuerySpecification node,
-                                           final AnalysisContext context) {
-    return null;
-  }
-
-  private void analyzeWhere(final Node node, final AnalysisContext context) {
+  private void analyzeWhere(final Node node) {
     analysis.setWhereExpression((Expression) node);
   }
 
-  private void analyzeGroupBy(final GroupBy groupBy, final AnalysisContext context) {
+  private void analyzeGroupBy(final GroupBy groupBy) {
     for (GroupingElement groupingElement : groupBy.getGroupingElements()) {
       Set<Expression> groupingSet = groupingElement.enumerateGroupingSets().get(0);
       analysis.getGroupByExpressions().addAll(groupingSet);
     }
   }
 
-  private void analyzeWindowExpression(final WindowExpression windowExpression,
-                                       final AnalysisContext context) {
+  private void analyzeWindowExpression(final WindowExpression windowExpression) {
     analysis.setWindowExpression(windowExpression);
   }
 
-  private void analyzeHaving(final Node node, final AnalysisContext context) {
+  private void analyzeHaving(final Node node) {
     analysis.setHavingExpression((Expression) node);
   }
 
@@ -526,7 +520,6 @@ private void setIntoTopicFormat(final StructuredDataSource into, final Table nod
         }
         avroSchemaFilePath = avroSchemaFilePath.substring(1, avroSchemaFilePath.length() - 1);
       }
-      analysis.setIntoAvroSchemaFilePath(avroSchemaFilePath);
       analysis.getIntoProperties().put(DdlConfig.AVRO_SCHEMA_FILE, avroSchemaFilePath);
     }
   }

ksql-engine/src/main/java/io/confluent/ksql/analyzer/ExpressionAnalyzer.java

@@ -16,8 +16,6 @@
 
 package io.confluent.ksql.analyzer;
 
-import io.confluent.ksql.function.KsqlFunction;
-import io.confluent.ksql.function.KsqlFunctions;
 import io.confluent.ksql.parser.tree.ArithmeticBinaryExpression;
 import io.confluent.ksql.parser.tree.AstVisitor;
 import io.confluent.ksql.parser.tree.Cast;
@@ -39,15 +37,15 @@
 
 
 public class ExpressionAnalyzer {
-  final Schema schema;
-  final boolean isJoinSchema;
+  private final Schema schema;
+  private final boolean isJoinSchema;
 
-  public ExpressionAnalyzer(Schema schema, boolean isJoinSchema) {
+  ExpressionAnalyzer(Schema schema, boolean isJoinSchema) {
     this.schema = schema;
     this.isJoinSchema = isJoinSchema;
   }
 
-  public void analyzeExpression(Expression expression) {
+  void analyzeExpression(Expression expression) {
     Visitor visitor = new Visitor(schema);
     visitor.process(expression, null);
   }
@@ -67,8 +65,6 @@ protected Object visitLikePredicate(LikePredicate node, Object context) {
     }
 
     protected Object visitFunctionCall(FunctionCall node, Object context) {
-      String functionName = node.getName().getSuffix();
-      KsqlFunction ksqlFunction = KsqlFunctions.getFunction(functionName);
       for (Expression argExpr : node.getArguments()) {
         process(argExpr, null);
       }

ksql-engine/src/main/java/io/confluent/ksql/codegen/SqlToJavaVisitor.java

@@ -42,7 +42,7 @@ public String process(final Expression expression, final Schema schema) {
     return formatExpression(expression, true);
   }
 
-  public String formatExpression(final Expression expression, final boolean unmangleNames) {
+  private String formatExpression(final Expression expression, final boolean unmangleNames) {
     Pair<String, Schema>
         expressionFormatterResult =
         new SqlToJavaVisitor.Formatter().process(expression, unmangleNames);