Dynamic partition

core/src/main/java/org/apache/flink/streaming/siddhi/router/AddRouteOperator.java

@@ -169,8 +169,14 @@ private void handleExecutionPlan(String executionPlanId, String executionPlan) t
                 executionPlanIdToPartitionKeys.put(executionPlanId, new ArrayList<>());
             }
             inputStreamToExecutionPlans.get(inputStreamId).add(executionPlanId);
-            executionPlanIdToPartitionKeys.get(executionPlanId).addAll(
-                streamPartitions.get(inputStreamId).getGroupByList());
+
+            if(streamPartitions.get(inputStreamId).getPartitonWithList().isEmpty()){
+                executionPlanIdToPartitionKeys.get(executionPlanId).addAll(
+                        streamPartitions.get(inputStreamId).getGroupByList());
+            }else{
+                executionPlanIdToPartitionKeys.get(executionPlanId).addAll(
+                        streamPartitions.get(inputStreamId).getPartitonWithList());
+            }
         }
     }
 }

core/src/main/java/org/apache/flink/streaming/siddhi/utils/SiddhiExecutionPlanner.java

@@ -21,6 +21,8 @@
 import io.siddhi.query.api.definition.StreamDefinition;
 import io.siddhi.query.api.execution.ExecutionElement;
 import io.siddhi.query.api.execution.partition.Partition;
+import io.siddhi.query.api.execution.partition.PartitionType;
+import io.siddhi.query.api.execution.partition.ValuePartitionType;
 import io.siddhi.query.api.execution.query.Query;
 import io.siddhi.query.api.execution.query.input.handler.StreamHandler;
 import io.siddhi.query.api.execution.query.input.handler.Window;
@@ -31,6 +33,7 @@
 import io.siddhi.query.api.execution.query.output.stream.OutputStream;
 import io.siddhi.query.api.execution.query.selection.OutputAttribute;
 import io.siddhi.query.api.execution.query.selection.Selector;
+import io.siddhi.query.api.expression.Expression;
 import io.siddhi.query.api.expression.Variable;
 import io.siddhi.query.compiler.SiddhiCompiler;
 import org.apache.commons.collections.ListUtils;
@@ -82,7 +85,14 @@ private void parse() throws Exception {
             if (executionElement instanceof Query) {
                query = (Query) executionElement;
             }else{
-               query = ((Partition) executionElement).getQueryList().get(0);
+                Partition partition = (Partition) executionElement;
+                Map<String, PartitionType> partitionTypeMap = partition.getPartitionTypeMap();
+                query = partition.getQueryList().get(0);
+                for(Map.Entry<String, PartitionType> partitionType : partitionTypeMap.entrySet()){
+                    if(partitionType.getValue() instanceof ValuePartitionType){
+                        retrievePartition(findStreamPartition(partitionType.getKey(),(ValuePartitionType) partitionType.getValue()));
+                    }
+                }
             }
             //--FIX END
             InputStream inputStream = query.getInputStream();
@@ -198,7 +208,7 @@ private void retrievePartition(StreamPartition partition) throws Exception {
             StreamPartition existingPartition = streamPartitions.get(partition.getInputStreamId());
             if (existingPartition.getType().equals(partition.getType())
                 && ListUtils.isEqualList(existingPartition.getGroupByList(), partition.getGroupByList())
-                || existingPartition.getType().equals(StreamPartition.Type.SHUFFLE)) {
+                || existingPartition.getType().equals(StreamPartition.Type.SHUFFLE) || existingPartition.getType().equals(StreamPartition.Type.PARTITIONWITH)) {
                 streamPartitions.put(partition.getInputStreamId(), partition);
             } else {
                 throw new Exception("You have incompatible partitions on stream " + partition.getInputStreamId()
@@ -225,6 +235,20 @@ private StreamPartition findStreamPartition(SingleInputStream inputStream, Selec
         }
     }
 
+    private StreamPartition findStreamPartition(String streamId, ValuePartitionType value){
+        StreamPartition partition = new StreamPartition(streamId);
+        if(value != null){
+            Expression expression = value.getExpression();
+            if(expression instanceof Variable){
+                String attributeName = ((Variable) expression).getAttributeName();
+                partition.setPartitionWithList(Collections.singletonList(attributeName));
+                partition.setType(StreamPartition.Type.PARTITIONWITH);
+                return partition;
+            }
+        }
+        return null;
+    }
+
     private StreamPartition generatePartition(String streamId, List<Window> windows, List<Variable> groupBy) {
         StreamPartition partition = new StreamPartition(streamId);
 
@@ -259,12 +283,14 @@ public Map<String, StreamPartition> getStreamPartitions() throws Exception {
     public static class StreamPartition {
         public enum Type {
             GROUPBY,
-            SHUFFLE
+            SHUFFLE,
+            PARTITIONWITH,
         }
 
         private String inputStreamId;
         private Type type;
         private List<String> groupByList = new ArrayList<>();
+        private List<String> partitionWithList = new ArrayList<>();
 
         public StreamPartition(String inputStreamId) {
             this.inputStreamId = inputStreamId;
@@ -289,5 +315,11 @@ public List<String> getGroupByList() {
         public void setGroupByList(List<String> groupByList) {
             this.groupByList = groupByList;
         }
+
+        public List<String> getPartitonWithList() {
+            return partitionWithList;
+        }
+
+        public void setPartitionWithList(List<String> partitionWithList){ this.partitionWithList = partitionWithList;}
     }
 }