Ksql 386 ksql functions non static

ksql-cli/src/main/java/io/confluent/ksql/cli/StandaloneExecutor.java

@@ -23,6 +23,8 @@
 import io.confluent.ksql.util.Pair;
 import io.confluent.ksql.util.PersistentQueryMetadata;
 import io.confluent.ksql.util.QueryMetadata;
+
+import org.apache.kafka.clients.admin.AdminClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,7 +44,7 @@ public class StandaloneExecutor {
 
   public StandaloneExecutor(Map streamProperties) {
     KsqlConfig ksqlConfig = new KsqlConfig(streamProperties);
-    ksqlEngine = new KsqlEngine(ksqlConfig, new KafkaTopicClientImpl(ksqlConfig.getKsqlAdminClientConfigProps()));
+    ksqlEngine = new KsqlEngine(ksqlConfig, new KafkaTopicClientImpl(AdminClient.create(ksqlConfig.getKsqlAdminClientConfigProps())));
   }
 
   public void executeStatements(String queries) throws Exception {

ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -49,7 +49,11 @@ public class KsqlConfig extends AbstractConfig {
   public static final String DEFAULT_SINK_WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION =
       "ksql.sink.window.change.log.additional.retention.default";
 
-  public static final String FAIL_ON_DESERIALIZATION_ERROR_CONFIG = KSQL_CONFIG_PREPERTY_PREFIX + ".fail.on.deserialization.error";
+  public static final String STREAM_INTERNAL_CHANGELOG_TOPIC_SUFFIX = "-changelog";
+
+  public static final String STREAM_INTERNAL_REPARTITION_TOPIC_SUFFIX = "-repartition";
+
+  public static final String FAIL_ON_DESERIALIZATION_ERROR_CONFIG = "fail.on.deserialization.error";
 
   public static final String
       KSQL_SERVICE_ID_CONFIG = "ksql.service.id";

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

@@ -17,10 +17,14 @@
 package io.confluent.ksql;
 
 import io.confluent.ksql.metastore.MetaStore;
+import io.confluent.ksql.util.KafkaTopicClient;
 import io.confluent.ksql.util.KafkaTopicClientImpl;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.PersistentQueryMetadata;
 import io.confluent.ksql.util.QueryMetadata;
+
+
+import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.streams.StreamsConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,19 +41,15 @@ public class KsqlContext {
   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 KafkaTopicClientImpl topicClient;
+  private final AdminClient adminClient;
+  private final KafkaTopicClient topicClient;
+
 
-  public KsqlContext() {
-    this(null);
+  public static KsqlContext create() {
+    return create(Collections.emptyMap());
   }
 
-  /**
-   * Create a KSQL context object with the given properties.
-   * A KSQL context has it's own metastore valid during the life of the object.
-   *
-   * @param streamsProperties
-   */
-  public KsqlContext(Map<String, Object> streamsProperties) {
+  public static KsqlContext create(Map<String, Object> streamsProperties) {
     if (streamsProperties == null) {
       streamsProperties = new HashMap<>();
     }
@@ -59,12 +59,29 @@ public KsqlContext(Map<String, Object> streamsProperties) {
     if (!streamsProperties.containsKey(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) {
       streamsProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP_SERVER_OPTION_DEFAULT);
     }
-    KsqlConfig ksqlConfig = new KsqlConfig(streamsProperties);
+    return new KsqlContext(streamsProperties);
+  }
+
 
-    topicClient = new KafkaTopicClientImpl(ksqlConfig.getKsqlAdminClientConfigProps());
+  /**
+   * Create a KSQL context object with the given properties.
+   * A KSQL context has it's own metastore valid during the life of the object.
+   *
+   * @param streamsProperties
+   */
+  private KsqlContext(Map<String, Object> streamsProperties) {
+    KsqlConfig ksqlConfig = new KsqlConfig(streamsProperties);
+    adminClient = AdminClient.create(ksqlConfig.getKsqlAdminClientConfigProps());
+    topicClient = new KafkaTopicClientImpl(adminClient);
     ksqlEngine = new KsqlEngine(ksqlConfig, topicClient);
   }
 
+  protected KsqlContext(KsqlEngine ksqlEngin, AdminClient adminClient, KafkaTopicClient
+      topicClient) {
+    this.ksqlEngine = ksqlEngin;
+    this.adminClient = adminClient;
+    this.topicClient = topicClient;
+  }
 
   public MetaStore getMetaStore() {
     return ksqlEngine.getMetaStore();
@@ -97,5 +114,6 @@ public void sql(String sql) throws Exception {
   public void close() throws IOException {
     ksqlEngine.close();
     topicClient.close();
+    adminClient.close();
   }
 }

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

@@ -18,6 +18,7 @@
 
 import io.confluent.ksql.ddl.DdlConfig;
 import io.confluent.ksql.ddl.commands.*;
+import io.confluent.ksql.function.FunctionRegistry;
 import io.confluent.ksql.parser.exception.ParseFailedException;
 import io.confluent.ksql.metastore.*;
 import io.confluent.ksql.parser.KsqlParser;
@@ -57,7 +58,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 
 public class KsqlEngine implements Closeable {
 
@@ -78,6 +78,8 @@ public class KsqlEngine implements Closeable {
   private final Map<Long, PersistentQueryMetadata> persistentQueries;
   private final Set<QueryMetadata> liveQueries;
 
+  public final FunctionRegistry functionRegistry;
+
   public KsqlEngine(final KsqlConfig ksqlConfig, final KafkaTopicClient topicClient) {
     Objects.requireNonNull(ksqlConfig, "Streams properties map cannot be null as it may be mutated later on");
 
@@ -90,6 +92,7 @@ public KsqlEngine(final KsqlConfig ksqlConfig, final KafkaTopicClient topicClien
 
     this.persistentQueries = new HashMap<>();
     this.liveQueries = new HashSet<>();
+    this.functionRegistry = new FunctionRegistry();
   }
 
   /**
@@ -128,9 +131,7 @@ public List<QueryMetadata> buildMultipleQueries(
   public List<QueryMetadata> planQueries(final boolean createNewAppId,
                                          final List<Pair<String, Statement>> statementList,
                                          final Map<String, Object> overriddenProperties,
-                                         final MetaStore tempMetaStore)
-          throws Exception {
-
+                                         final MetaStore tempMetaStore) throws Exception {
     // Logical plan creation from the ASTs
     List<Pair<String, PlanNode>> logicalPlans = queryEngine.buildLogicalPlans(tempMetaStore, statementList);
 
@@ -345,6 +346,10 @@ public MetaStore getMetaStore() {
     return metaStore;
   }
 
+  public FunctionRegistry getFunctionRegistry() {
+    return functionRegistry;
+  }
+
   public KafkaTopicClient getTopicClient() {
     return topicClient;
   }
@@ -360,8 +365,7 @@ public boolean terminateQuery(final long queryId, final boolean closeStreams) {
     }
     liveQueries.remove(queryMetadata);
     if (closeStreams) {
-      queryMetadata.getKafkaStreams().close(100L, TimeUnit.MILLISECONDS);
-      queryMetadata.getKafkaStreams().cleanUp();
+      queryMetadata.close();
     }
     return true;
   }
@@ -388,8 +392,7 @@ public KsqlConfig getKsqlConfig() {
   @Override
   public void close() throws IOException {
     for (QueryMetadata queryMetadata : liveQueries) {
-      queryMetadata.getKafkaStreams().close(100L, TimeUnit.MILLISECONDS);
-      queryMetadata.getKafkaStreams().cleanUp();
+      queryMetadata.close();
     }
     topicClient.close();
   }
@@ -403,8 +406,8 @@ public boolean terminateAllQueries() {
       for (QueryMetadata queryMetadata : liveQueries) {
         if (queryMetadata instanceof PersistentQueryMetadata) {
           PersistentQueryMetadata persistentQueryMetadata = (PersistentQueryMetadata) queryMetadata;
-          persistentQueryMetadata.getKafkaStreams().close(100L, TimeUnit.MILLISECONDS);
-          persistentQueryMetadata.getKafkaStreams().cleanUp();
+          persistentQueryMetadata.close();
+
         }
       }
     } catch (Exception e) {

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

@@ -126,8 +126,9 @@ private PlanNode buildQueryLogicalPlan(final Query query, final MetaStore tempMe
 
     AggregateAnalysis aggregateAnalysis = new AggregateAnalysis();
     AggregateAnalyzer aggregateAnalyzer = new
-        AggregateAnalyzer(aggregateAnalysis, analysis);
-    AggregateExpressionRewriter aggregateExpressionRewriter = new AggregateExpressionRewriter();
+        AggregateAnalyzer(aggregateAnalysis, analysis, ksqlEngine.getFunctionRegistry());
+    AggregateExpressionRewriter aggregateExpressionRewriter =
+        new AggregateExpressionRewriter(ksqlEngine.getFunctionRegistry());
     for (Expression expression: analysis.getSelectExpressions()) {
       aggregateAnalyzer
           .process(expression, new AnalysisContext(null));
@@ -160,7 +161,7 @@ private PlanNode buildQueryLogicalPlan(final Query query, final MetaStore tempMe
 
 
     // Build a logical plan
-    PlanNode logicalPlan = new LogicalPlanner(analysis, aggregateAnalysis).buildPlan();
+    PlanNode logicalPlan = new LogicalPlanner(analysis, aggregateAnalysis, ksqlEngine.getFunctionRegistry()).buildPlan();
     if (logicalPlan instanceof KsqlStructuredDataOutputNode) {
       KsqlStructuredDataOutputNode ksqlStructuredDataOutputNode =
           (KsqlStructuredDataOutputNode) logicalPlan;
@@ -217,7 +218,9 @@ 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(), new MetastoreUtil());
+    PhysicalPlanBuilder physicalPlanBuilder = new PhysicalPlanBuilder(builder, ksqlConfigClone,
+                                                                      ksqlEngine.getTopicClient(),
+                                                                      new MetastoreUtil(), ksqlEngine.getFunctionRegistry());
     SchemaKStream schemaKStream = physicalPlanBuilder.buildPhysicalPlan(logicalPlan);
 
     OutputNode outputNode = physicalPlanBuilder.getPlanSink();
@@ -291,7 +294,8 @@ private QueryMetadata buildPlanForBareQuery(boolean addUniqueTimeSuffix,
         schemaKStream.getExecutionPlan(""),
         schemaKStream.getQueue(),
         (sourceSchemaKstream instanceof SchemaKTable) ?
-        DataSource.DataSourceType.KTABLE : DataSource.DataSourceType.KSTREAM
+        DataSource.DataSourceType.KTABLE : DataSource.DataSourceType.KSTREAM,
+        applicationId
     );
   }
 
@@ -308,9 +312,15 @@ private QueryMetadata buildPlanForBareQuery(boolean addUniqueTimeSuffix,
    * @param persistanceQueryPrefix
    */
   private QueryMetadata buildPlanForStructuredOutputNode(boolean addUniqueTimeSuffix,
-                                                         Pair<String, PlanNode> statementPlanPair, Map<String, Object> overriddenStreamsProperties,
-                                                         boolean updateMetastore, StreamsBuilder builder, KsqlConfig ksqlConfigClone, SchemaKStream schemaKStream,
-                                                         KsqlStructuredDataOutputNode outputNode, String serviceId, String persistanceQueryPrefix) {
+                                                         Pair<String, PlanNode> statementPlanPair,
+                                                         Map<String, Object> overriddenStreamsProperties,
+                                                         boolean updateMetastore,
+                                                         StreamsBuilder builder,
+                                                         KsqlConfig ksqlConfigClone,
+                                                         SchemaKStream schemaKStream,
+                                                         KsqlStructuredDataOutputNode outputNode,
+                                                         String serviceId,
+                                                         String persistanceQueryPrefix) {
 
     long queryId = getNextQueryId();
 
@@ -353,7 +363,9 @@ private QueryMetadata buildPlanForStructuredOutputNode(boolean addUniqueTimeSuff
                                        streams, outputNode, schemaKStream
                                            .getExecutionPlan(""), queryId,
                                        (schemaKStream instanceof SchemaKTable) ? DataSource
-                                           .DataSourceType.KTABLE : DataSource.DataSourceType.KSTREAM);
+                                           .DataSourceType.KTABLE : DataSource.DataSourceType
+                                           .KSTREAM,
+                                       applicationId);
   }
 
 

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

@@ -16,7 +16,7 @@
 
 package io.confluent.ksql.analyzer;
 
-import io.confluent.ksql.function.KsqlFunctions;
+import io.confluent.ksql.function.FunctionRegistry;
 import io.confluent.ksql.parser.tree.DereferenceExpression;
 import io.confluent.ksql.parser.tree.Expression;
 import io.confluent.ksql.parser.tree.FunctionCall;
@@ -30,6 +30,7 @@ public class AggregateAnalyzer extends DefaultTraversalVisitor<Node, AnalysisCon
 
   private AggregateAnalysis aggregateAnalysis;
   private Analysis analysis;
+  private FunctionRegistry functionRegistry;
 
   private boolean hasAggregateFunction = false;
 
@@ -42,15 +43,17 @@ public void setHasAggregateFunction(boolean hasAggregateFunction) {
   }
 
   public AggregateAnalyzer(AggregateAnalysis aggregateAnalysis,
-                           Analysis analysis) {
+                           Analysis analysis,
+                           FunctionRegistry functionRegistry) {
     this.aggregateAnalysis = aggregateAnalysis;
     this.analysis = analysis;
+    this.functionRegistry = functionRegistry;
   }
 
   @Override
   protected Node visitFunctionCall(final FunctionCall node, final AnalysisContext context) {
     String functionName = node.getName().getSuffix();
-    if (KsqlFunctions.isAnAggregateFunction(functionName)) {
+    if (functionRegistry.isAnAggregateFunction(functionName)) {
       if (node.getArguments().isEmpty()) {
         Expression argExpression;
         if (analysis.getJoin() != null) {

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

@@ -17,7 +17,7 @@
 package io.confluent.ksql.codegen;
 
 import io.confluent.ksql.function.KsqlFunction;
-import io.confluent.ksql.function.KsqlFunctions;
+import io.confluent.ksql.function.FunctionRegistry;
 import io.confluent.ksql.function.udf.Kudf;
 import io.confluent.ksql.parser.tree.AstVisitor;
 import io.confluent.ksql.parser.tree.Expression;
@@ -36,17 +36,24 @@
 
 public class CodeGenRunner {
 
-  public Map<String, Class> getParameterInfo(final Expression expression, final Schema schema) {
-    Visitor visitor = new Visitor(schema);
+  final Schema schema;
+  final FunctionRegistry functionRegistry;
+
+  public CodeGenRunner(Schema schema, FunctionRegistry functionRegistry) {
+    this.functionRegistry = functionRegistry;
+    this.schema = schema;
+  }
+
+  public Map<String, Class> getParameterInfo(final Expression expression) {
+    Visitor visitor = new Visitor(schema, functionRegistry);
     visitor.process(expression, null);
     return visitor.parameterMap;
   }
 
   public ExpressionMetadata buildCodeGenFromParseTree(
-      final Expression expression,
-      final Schema schema) throws Exception {
-    CodeGenRunner codeGenRunner = new CodeGenRunner();
-    Map<String, Class> parameterMap = codeGenRunner.getParameterInfo(expression, schema);
+      final Expression expression) throws Exception {
+    CodeGenRunner codeGenRunner = new CodeGenRunner(schema, functionRegistry);
+    Map<String, Class> parameterMap = codeGenRunner.getParameterInfo(expression);
 
     String[] parameterNames = new String[parameterMap.size()];
     Class[] parameterTypes = new Class[parameterMap.size()];
@@ -66,15 +73,16 @@ public ExpressionMetadata buildCodeGenFromParseTree(
       index++;
     }
 
-    String javaCode = new SqlToJavaVisitor().process(expression, schema);
+    String javaCode = new SqlToJavaVisitor(schema, functionRegistry).process(expression);
 
     IExpressionEvaluator ee = CompilerFactoryFactory.getDefaultCompilerFactory().newExpressionEvaluator();
 
     // The expression will have two "int" parameters: "a" and "b".
     ee.setParameters(parameterNames, parameterTypes);
 
     // And the expression (i.e. "result") type is also "int".
-    ExpressionTypeManager expressionTypeManager = new ExpressionTypeManager(schema);
+    ExpressionTypeManager expressionTypeManager = new ExpressionTypeManager(schema,
+                                                                            functionRegistry);
     Schema expressionType = expressionTypeManager.getExpressionType(expression);
 
     ee.setExpressionType(SchemaUtil.getJavaType(expressionType));
@@ -89,10 +97,12 @@ private class Visitor extends AstVisitor<Object, Object> {
 
     final Schema schema;
     final Map<String, Class> parameterMap;
+    final FunctionRegistry functionRegistry;
 
-    Visitor(Schema schema) {
+    Visitor(Schema schema, FunctionRegistry functionRegistry) {
       this.schema = schema;
       this.parameterMap = new HashMap<>();
+      this.functionRegistry = functionRegistry;
     }
 
     protected Object visitLikePredicate(LikePredicate node, Object context) {
@@ -102,7 +112,7 @@ protected Object visitLikePredicate(LikePredicate node, Object context) {
 
     protected Object visitFunctionCall(FunctionCall node, Object context) {
       String functionName = node.getName().getSuffix();
-      KsqlFunction ksqlFunction = KsqlFunctions.getFunction(functionName);
+      KsqlFunction ksqlFunction = functionRegistry.getFunction(functionName);
       parameterMap.put(node.getName().getSuffix(),
                        ksqlFunction.getKudfClass());
       for (Expression argExpr : node.getArguments()) {