add cql 2.0 version

Zhiqiang-He · Dec 5, 2016 · 3cf8bc8 · 3cf8bc8
1 parent 1006bb3
commit 3cf8bc8
Show file tree

Hide file tree

Showing 843 changed files with 34,014 additions and 5,528 deletions.
diff --git a/cql/logs/streaming-cql.log b/cql/logs/streaming-cql.log
diff --git a/cql/pom.xml b/cql/pom.xml
@@ -22,14 +22,26 @@
     <parent>
         <artifactId>streamCQL</artifactId>
         <groupId>com.huawei.streaming</groupId>
-        <version>1.0</version>
+        <version>2.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <artifactId>cql</artifactId>
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-jar-plugin</artifactId>
@@ -42,7 +54,7 @@
                     <excludes>
                         <exclude>com/huawei/streaming/cql/semanticanalyzer/parser/*.g4</exclude>
                         <exclude>streaming-site.xml</exclude>
-                        <exclude>logback.xml</exclude>
+                        <exclude>log4j2.xml</exclude>
                         <exclude>logback-console.xml</exclude>
                         <exclude>cqlclient</exclude>
                         <exclude>driver</exclude>
@@ -123,8 +135,19 @@
         <dependency>
             <groupId>org.apache.storm</groupId>
             <artifactId>storm-core</artifactId>
-            <version>0.10.0-beta</version>
+            <version>1.0.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>3.4.9</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 </project>
diff --git a/cql/src/main/java/com/huawei/streaming/api/AnnotationUtils.java b/cql/src/main/java/com/huawei/streaming/api/AnnotationUtils.java
@@ -19,7 +19,8 @@
 package com.huawei.streaming.api;
 
 import java.lang.reflect.Field;
-import java.util.Locale;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
@@ -50,18 +51,24 @@ public class AnnotationUtils
     /**
      * 将类中configannotation的申明转为streamingconfig中的属性
      *
-     * @param obj 待转义bean
-     * @return 配置属性对象
-     * @throws ApplicationBuildException 反射异常
      */
     public static TreeMap<String, String> getAnnotationsToConfig(Object obj)
         throws ApplicationBuildException
     {
         TreeMap<String, String> config = Maps.newTreeMap();
         Field[] fs = obj.getClass().getDeclaredFields();
-        for (Field f : fs)
+        for (final Field f : fs)
         {
-            f.setAccessible(true);
+            AccessController.doPrivileged(new PrivilegedAction<Object>()
+            {
+                public Object run()
+                {
+                    f.setAccessible(true);
+                    return null;
+                }
+
+            });
+
             ConfigAnnotation annotation = f.getAnnotation(ConfigAnnotation.class);
             if (null != annotation)
             {
@@ -110,19 +117,15 @@ private static Object getOperatorConfigValue(Object obj, Field field)
     /**
      * 获取API类中的fields和streamingconf的映射关系
      *
-     * @param clazz api类
-     * @return 配置属性映射关系
-     * @throws SemanticAnalyzerException 语义分析异常
      */
     public static Map<String, String> getConfigMapping(String clazz)
         throws SemanticAnalyzerException
     {
         Map<String, String> mapping = Maps.newHashMap();
-        Class< ? > apiClass = getaClass(clazz);
+        Class< ? > apiClass = getClass(clazz);
         Field[] fs = apiClass.getDeclaredFields();
         for (Field f : fs)
         {
-            f.setAccessible(true);
             ConfigAnnotation annotaion = f.getAnnotation(ConfigAnnotation.class);
             if (annotaion == null)
             {
@@ -134,13 +137,16 @@ public static Map<String, String> getConfigMapping(String clazz)
                 continue;
             }
 
-            String key = f.getName().toLowerCase(Locale.US);
+            String key = f.getName();
             mapping.put(key, value);
         }
         return mapping;
     }
-
-    private static Class< ? > getaClass(String clazz)
+
+    /**
+     * 获取一个类
+     */
+    public static Class< ? > getClass(String clazz)
         throws SemanticAnalyzerException
     {
         try
@@ -160,10 +166,6 @@ public static Map<String, String> getConfigMapping(String clazz)
     /**
      * 将config中的配置属性对应到该对象中
      *
-     * @param obj 待转义bean
-     * @param config 配置属性
-     * @return 配置属性对象
-     * @throws ApplicationBuildException 应用程序构建异常
      */
     public static Object setConfigToObject(Object obj, Map<String, String> config)
         throws ApplicationBuildException
@@ -187,10 +189,19 @@ public static Object setConfigToObject(Object obj, Map<String, String> config)
         return obj;
     }
 
-    private static void resetFieldValue(Object obj, Map<String, String> config, Field field)
+    private static void resetFieldValue(Object obj, Map<String, String> config, final Field field)
         throws IllegalAccessException
     {
-        field.setAccessible(true);
+        AccessController.doPrivileged(new PrivilegedAction<Object>()
+        {
+            public Object run()
+            {
+                field.setAccessible(true);
+                return null;
+            }
+
+        });
+
         ConfigAnnotation annotaion = field.getAnnotation(ConfigAnnotation.class);
         if (null != annotaion)
         {
@@ -213,8 +224,6 @@ private static void resetFieldValue(Object obj, Map<String, String> config, Fiel
     /**
      * 根据注解获取创建算子实例的类
      *
-     * @param clazz 注解所在类
-     * @return 算子实例类
      */
     public static Class< ? extends OperatorInfoCreator> getOperatorCreatorAnnotation(Class< ? > clazz)
     {
@@ -225,8 +234,6 @@ private static void resetFieldValue(Object obj, Map<String, String> config, Fiel
     /**
      * 获取类上面的序列化和反序列化申明
      *
-     * @param clazz 待获取的类
-     * @return 映射申明名称
      */
     public static Class< ? extends StreamSerDe> getStreamSerDeAnnoationOverClass(Class< ? > clazz)
     {
@@ -237,8 +244,6 @@ private static void resetFieldValue(Object obj, Map<String, String> config, Fiel
     /**
      * 获取类上面的表达式对象创建实例
      *
-     * @param clazz 待获取的类
-     * @return 映射申明名称
      */
     public static Class< ? extends ExpressionCreator> getExpressionCreatorAnnoationOverClass(Class< ? > clazz)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/Application.java b/cql/src/main/java/com/huawei/streaming/api/Application.java
@@ -92,7 +92,6 @@ public class Application
     /**
      * <默认构造函数>
      *
-     * @param applicationId 应用程序标示符
      */
     public Application(String applicationId)
     {
@@ -112,7 +111,7 @@ public void setConfs(TreeMap<String, String> confs)
 
     public String[] getUserFiles()
     {
-        return userFiles;
+        return userFiles == null ? new String[] {} : (String[])userFiles.clone();
     }
 
     public void setUserFiles(String[] userFiles)

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/AggregateOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/AggregateOperator.java
@@ -50,8 +50,6 @@ public class AggregateOperator extends BasicAggFunctionOperator
 
     /**
      * <默认构造函数>
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public AggregateOperator(String id, int parallelNumber)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/BaseDataSourceOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/BaseDataSourceOperator.java
@@ -65,8 +65,6 @@ public abstract class BaseDataSourceOperator extends BasicAggFunctionOperator
     /**
      * <默认构造函数>
      *
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public BaseDataSourceOperator(String id, int parallelNumber)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/BasicAggFunctionOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/BasicAggFunctionOperator.java
@@ -52,8 +52,6 @@ public class BasicAggFunctionOperator extends InnerFunctionOperator
 
     /**
      * <默认构造函数>
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public BasicAggFunctionOperator(String id, int parallelNumber)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/CombineOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/CombineOperator.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 com.huawei.streaming.api.opereators;
+
+import com.huawei.streaming.cql.executor.operatorinfocreater.CombineInfoCreator;
+import com.huawei.streaming.cql.executor.operatorinfocreater.OperatorInfoCreatorAnnotation;
+
+/**
+ * Combine算子，将多个流的数据合并成一个
+ * 从每个流中取出一个或者多列，依据combine的条件进行组合
+ * 
+ */
+@OperatorInfoCreatorAnnotation(CombineInfoCreator.class)
+public class CombineOperator extends InnerFunctionOperator
+{
+
+    /**
+     * combine条件
+     * 流的名称必须和
+     * 比如s1.id,s2.aid,s3.id
+     * 这里的s1,s2,s3既可以是schema名称，也可以是流名称，还有别名
+     */
+    private String combineProperties;
+
+    /**
+     * combine的流名称，如果多个流，用逗号隔离
+     * 这个流出现的顺序，决定了combine的最终输出的舒徐
+     * 这里的流出现顺序，必须和select中属性的流名称出现顺序一致。
+     */
+    private String orderedStreams;
+
+    /**
+     * <默认构造函数>
+     */
+    public CombineOperator(String id, int parallelNumber)
+    {
+        super(id, parallelNumber);
+    }
+
+    public String getCombineProperties()
+    {
+        return combineProperties;
+    }
+
+    public void setCombineProperties(String combineProperties)
+    {
+        this.combineProperties = combineProperties;
+    }
+
+    public String getOrderedStreams()
+    {
+        return orderedStreams;
+    }
+
+    public void setOrderedStreams(String orderedStreams)
+    {
+        this.orderedStreams = orderedStreams;
+    }
+
+}
diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/ConsoleInputOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/ConsoleInputOperator.java
diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/ConsoleOutputOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/ConsoleOutputOperator.java
@@ -37,8 +37,6 @@ public class ConsoleOutputOperator extends InnerOutputSourceOperator
 
     /**
      * <默认构造函数>
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public ConsoleOutputOperator(String id, int parallelNumber)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/DataSourceOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/DataSourceOperator.java
@@ -43,8 +43,6 @@ public class DataSourceOperator extends BaseDataSourceOperator
     /**
      * <默认构造函数>
      *
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public DataSourceOperator(String id, int parallelNumber)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/FilterOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/FilterOperator.java
@@ -43,8 +43,6 @@ public class FilterOperator extends InnerFunctionOperator
 
     /**
      * <默认构造函数>
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public FilterOperator(String id, int parallelNumber)
     {

diff --git a/cql/src/main/java/com/huawei/streaming/api/opereators/FunctionStreamOperator.java b/cql/src/main/java/com/huawei/streaming/api/opereators/FunctionStreamOperator.java
@@ -52,8 +52,6 @@ public class FunctionStreamOperator extends Operator
     /**
      * <默认构造函数>
      *
-     * @param id 算子id
-     * @param parallelNumber 算子并行度
      */
     public FunctionStreamOperator(String id, int parallelNumber)
     {