DTStack · weifuziang · Jan 19, 2021 · Jan 19, 2021 · Jan 19, 2021 · Jan 19, 2021
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
@@ -1,10 +1,22 @@
-build:
+stages:
+  - validate
+  - test
+
+test-job:
   stage: test
   script:
     - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
+  only:
+    - v1.10.0_dev
+  tags:
+    - dt-insight-engine
+
+validate-job:
+  stage: validate
+  script:
     - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL"  -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
     - sh ci/sonar_notify.sh
   only:
-    - v1.8.0_dev
+    - v1.10.0_dev
   tags:
-    - dt-insight-engine
+    - dt-insight-engine
diff --git a/README.md b/README.md
@@ -3,21 +3,32 @@ FlinkStreamSQL
 [![License](https://img.shields.io/badge/license-Apache%202-4EB1BA.svg)](https://www.apache.org/licenses/LICENSE-2.0.html)
 
 ## 技术交流
-- 招聘**大数据平台开发工程师**，想了解岗位详细信息可以添加本人微信号ysqwhiletrue，注明招聘，如有意者发送简历至[sishu@dtstack.com](mailto:sishu@dtstack.com)
+- 招聘**Flink开发工程师**，如果有兴趣，请联系思枢【微信号ysqwhiletrue】，注明招聘<BR>
+  Flink开发工程师JD要求：<BR>
+  1.负责袋鼠云基于Flink的衍生框架数据同步flinkx和实时计算flinkstreamsql框架的开发；<BR>
+  2.调研和把握当前最新大数据实时计算技术，将其中的合适技术引入到平台中，改善产品，提升竞争力；<BR>
+  职位要求：<BR>
+  1、本科及以上学历，3年及以上的Flink开发经验，精通Java，熟悉Scala、Python优先考虑；<BR>
+  2、熟悉Flink原理，有基于Flink做过二次源码的开发，在github上贡献者Flink源码者优先；<BR>
+  3、有机器学习、数据挖掘相关经验者优先；<BR>
+  4、对新技术有快速学习和上手能力，对代码有一定的洁癖；<BR>
+  加分项：<BR>
+  1.在GitHub或其他平台上有过开源项目;<BR>
+  可以添加本人微信号ysqwhiletrue，注明招聘，如有意者发送简历至sishu@dtstack.com<BR>
 - 我们使用[钉钉](https://www.dingtalk.com/)沟通交流，可以搜索群号[**30537511**]或者扫描下面的二维码进入钉钉群
 <div align=center>
      <img src=docs/images/streamsql_dd.jpg width=300 />
 </div>
 
 ## 介绍
- * 基于开源的flink，对其实时sql进行扩展   
-    * 自定义create table 语法（包括源表,输出表,维表）
-    * 自定义create view 语法
-    * 自定义create function 语法
-    * 实现了流与维表的join
-    * 支持原生FlinkSQL所有的语法
-    * 扩展了输入和输出的性能指标到Task metrics
-    
+* 基于开源的flink，对其实时sql进行扩展
+   * 自定义create table 语法（包括源表,输出表,维表）
+   * 自定义create view 语法
+   * 自定义create function 语法
+   * 实现了流与维表的join
+   * 支持原生FlinkSQL所有的语法
+   * 扩展了输入和输出的性能指标到Task metrics
+
 ## 目录
 
 [ 1.1 demo](docs/demo.md)  
@@ -26,11 +37,11 @@ FlinkStreamSQL
 [ 1.4 支持的插件介绍和demo](docs/pluginsInfo.md)     
 [ 1.5 指标参数](docs/newMetric.md)  
 [ 1.6 自定义函数](docs/function.md)  
-[ 1.7 自定义视图](docs/createView.md)  
-
-## 如何贡献FlinkStreamSQL  
-
+[ 1.7 自定义视图](docs/createView.md)
+
+## 如何贡献FlinkStreamSQL
+
+[pr规范](docs/pr.md)
+
 ## License
-FlinkStreamSQL is under the Apache 2.0 license. See the [LICENSE](http://www.apache.org/licenses/LICENSE-2.0) file for details.    
-
-
+FlinkStreamSQL is under the Apache 2.0 license. See the [LICENSE](http://www.apache.org/licenses/LICENSE-2.0) file for details.    
diff --git a/...a-all-side/src/test/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRowTest.java b/...a-all-side/src/test/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRowTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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.dtstack.flink.sql.side.cassandra;
+
+import com.dtstack.flink.sql.side.BaseAllReqRow;
+import com.dtstack.flink.sql.side.BaseSideInfo;
+import com.dtstack.flink.sql.side.FieldInfo;
+import com.dtstack.flink.sql.side.JoinInfo;
+import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.calcite.sql.JoinType;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.GenericRow;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.support.membermodification.MemberMatcher.constructor;
+import static org.powermock.api.support.membermodification.MemberModifier.suppress;
+
+/**
+ * @author: chuixue
+ * @create: 2020-07-28 10:58
+ * @description:
+ **/
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({CassandraAllReqRow.class,
+        BaseAllReqRow.class,
+        CassandraAllSideInfo.class})//要跳过的写在后面
+@PowerMockIgnore({"javax.*"})
+public class CassandraAllReqRowTest {
+
+    private CassandraAllReqRow cassandraAllReqRow;
+    private RowTypeInfo rowTypeInfo = new RowTypeInfo(new TypeInformation[]{TypeInformation.of(Integer.class), TypeInformation.of(String.class), TypeInformation.of(Integer.class)}, new String[]{"id", "bb", "PROCTIME"});
+    private JoinInfo joinInfo;
+    private List<FieldInfo> outFieldInfoList = new ArrayList<>();
+    private CassandraSideTableInfo sideTableInfo;
+    private BaseSideInfo sideInfo;
+    private AtomicReference<Map<String, List<Map<String, Object>>>> cacheRef = new AtomicReference<>();
+
+    @Before
+    public void setUp() {
+        joinInfo = mock(JoinInfo.class);
+        sideTableInfo = mock(CassandraSideTableInfo.class);
+        sideInfo = PowerMockito.mock(CassandraAllSideInfo.class);
+
+        Map<String, List<Map<String, Object>>> map = Maps.newHashMap();
+        cacheRef.set(map);
+
+        suppress(constructor(CassandraAllSideInfo.class));
+        suppress(constructor(BaseAllReqRow.class));
+        cassandraAllReqRow = new CassandraAllReqRow(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        Whitebox.setInternalState(cassandraAllReqRow, "sideInfo", sideInfo);
+        Whitebox.setInternalState(cassandraAllReqRow, "cacheRef", cacheRef);
+    }
+
+    @Test
+    public void testReloadCache() throws SQLException {
+        when(sideInfo.getSideTableInfo()).thenReturn(sideTableInfo);
+        when(sideTableInfo.getMaxRequestsPerConnection()).thenReturn(1);
+        when(sideTableInfo.getCoreConnectionsPerHost()).thenReturn(1);
+        when(sideTableInfo.getMaxConnectionsPerHost()).thenReturn(1);
+        when(sideTableInfo.getMaxQueueSize()).thenReturn(1);
+        when(sideTableInfo.getReadTimeoutMillis()).thenReturn(1);
+        when(sideTableInfo.getConnectTimeoutMillis()).thenReturn(1);
+        when(sideTableInfo.getPoolTimeoutMillis()).thenReturn(1);
+        when(sideTableInfo.getAddress()).thenReturn("12.12.12.12:9042,10.10.10.10:9042");
+        when(sideTableInfo.getUserName()).thenReturn("userName");
+        when(sideTableInfo.getPassword()).thenReturn("password");
+        when(sideTableInfo.getDatabase()).thenReturn("getDatabase");
+
+        cassandraAllReqRow.initCache();
+        cassandraAllReqRow.reloadCache();
+    }
+
+    @Test
+    public void testFlatmap() throws Exception {
+        GenericRow row = new GenericRow(3);
+        row.setField(0, 1);
+        row.setField(1, "bbbbbb");
+        row.setField(2, "2020-07-14 01:27:43.969");
+        Collector<BaseRow> out = mock(Collector.class);
+
+        List<String> equalFieldList = Lists.newArrayList();
+        equalFieldList.add("rowkey");
+        List<Integer> equalValIndex = Lists.newArrayList();
+        equalValIndex.add(0);
+
+        List<FieldInfo> outFieldInfoList = Lists.newArrayList();
+        FieldInfo fieldInfo = new FieldInfo();
+        fieldInfo.setTable("m");
+        fieldInfo.setFieldName("id");
+        fieldInfo.setTypeInformation(TypeInformation.of(Integer.class));
+        outFieldInfoList.add(fieldInfo);
+        outFieldInfoList.add(fieldInfo);
+        outFieldInfoList.add(fieldInfo);
+        outFieldInfoList.add(fieldInfo);
+
+        Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
+        inFieldIndex.put(0, 0);
+        inFieldIndex.put(1, 1);
+
+
+        Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();
+        sideFieldIndex.put(2, 0);
+        sideFieldIndex.put(3, 1);
+
+        Map<Integer, String> sideFieldNameIndex = Maps.newHashMap();
+        sideFieldNameIndex.put(2, "rowkey");
+        sideFieldNameIndex.put(3, "channel");
+
+        RowTypeInfo rowTypeInfo = new RowTypeInfo(new TypeInformation[]{TypeInformation.of(Integer.class), TypeInformation.of(String.class), TypeInformation.of(Timestamp.class)}, new String[]{"id", "bb", "PROCTIME"});
+
+        when(sideInfo.getEqualValIndex()).thenReturn(equalValIndex);
+        when(sideInfo.getJoinType()).thenReturn(JoinType.LEFT);
+        when(sideInfo.getOutFieldInfoList()).thenReturn(outFieldInfoList);
+        when(sideInfo.getInFieldIndex()).thenReturn(inFieldIndex);
+        when(sideInfo.getRowTypeInfo()).thenReturn(rowTypeInfo);
+        when(sideInfo.getSideFieldNameIndex()).thenReturn(sideFieldNameIndex);
+
+        cassandraAllReqRow.flatMap(row, out);
+    }
+
+}
diff --git a/...all-side/src/test/java/com/dtstack/flink/sql/side/cassandra/CassandraAllSideInfoTest.java b/...all-side/src/test/java/com/dtstack/flink/sql/side/cassandra/CassandraAllSideInfoTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.dtstack.flink.sql.side.cassandra;
+
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.BaseSideInfo;
+import com.dtstack.flink.sql.side.FieldInfo;
+import com.dtstack.flink.sql.side.JoinInfo;
+import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.support.membermodification.MemberMatcher.constructor;
+import static org.powermock.api.support.membermodification.MemberModifier.suppress;
+
+/**
+ * @author: chuixue
+ * @create: 2020-07-28 11:00
+ * @description:
+ **/
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({CassandraAllSideInfo.class,
+        BaseSideInfo.class})//要跳过的写在后面
+public class CassandraAllSideInfoTest {
+    private CassandraAllSideInfo cassandraAllSideInfo;
+    private JoinInfo joinInfo;
+    private AbstractSideTableInfo sideTableInfo;
+
+    @Before
+    public void setUp() {
+        joinInfo = mock(JoinInfo.class);
+        sideTableInfo = mock(AbstractSideTableInfo.class);
+
+        List<FieldInfo> outFieldInfoList = Lists.newArrayList();
+        FieldInfo fieldInfo = new FieldInfo();
+        fieldInfo.setTable("m");
+        fieldInfo.setFieldName("_id");
+        fieldInfo.setTypeInformation(TypeInformation.of(String.class));
+        outFieldInfoList.add(fieldInfo);
+
+        FieldInfo fieldInfo2 = new FieldInfo();
+        fieldInfo2.setTable("s");
+        fieldInfo2.setFieldName("name");
+        fieldInfo2.setTypeInformation(TypeInformation.of(String.class));
+        outFieldInfoList.add(fieldInfo2);
+
+        Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();
+        Map<Integer, String> sideFieldNameIndex = Maps.newHashMap();
+        RowTypeInfo rowTypeInfo = new RowTypeInfo(new TypeInformation[]{TypeInformation.of(Integer.class), TypeInformation.of(String.class), TypeInformation.of(Integer.class)}, new String[]{"id", "bb", "PROCTIME"});
+        Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
+        List<String> equalFieldList = Lists.newArrayList();
+        equalFieldList.add("_id");
+
+        suppress(constructor(BaseSideInfo.class));
+        cassandraAllSideInfo = new CassandraAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        Whitebox.setInternalState(cassandraAllSideInfo, "outFieldInfoList", outFieldInfoList);
+        Whitebox.setInternalState(cassandraAllSideInfo, "sideFieldIndex", sideFieldIndex);
+        Whitebox.setInternalState(cassandraAllSideInfo, "sideFieldNameIndex", sideFieldNameIndex);
+        Whitebox.setInternalState(cassandraAllSideInfo, "rowTypeInfo", rowTypeInfo);
+        Whitebox.setInternalState(cassandraAllSideInfo, "inFieldIndex", inFieldIndex);
+        Whitebox.setInternalState(cassandraAllSideInfo, "equalFieldList", equalFieldList);
+    }
+
+    @Test
+    public void testBuildEqualInfo() {
+        CassandraSideTableInfo sideTableInfo = mock(CassandraSideTableInfo.class);
+        when(sideTableInfo.getDatabase()).thenReturn("dd");
+        when(sideTableInfo.getTableName()).thenReturn("d");
+        Whitebox.setInternalState(cassandraAllSideInfo, "sideSelectFields", "cassandraAllSideInfo");
+        cassandraAllSideInfo.buildEqualInfo(joinInfo, sideTableInfo);
+    }
+
+    @Test
+    public void testParseSelectFields() throws NoSuchMethodException {
+        SqlBasicCall conditionNode = mock(SqlBasicCall.class);
+        when(joinInfo.getSideTableName()).thenReturn("s");
+        when(joinInfo.getNonSideTable()).thenReturn("m");
+        when(joinInfo.getCondition()).thenReturn(conditionNode);
+        suppress(BaseSideInfo.class.getMethod("dealOneEqualCon", SqlNode.class, String.class));
+
+        cassandraAllSideInfo.parseSelectFields(joinInfo);
+    }
+}