code review

Author: dapeng

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -28,10 +28,10 @@
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
-import com.dtstack.flink.sql.side.AllReqRow;
+import com.dtstack.flink.sql.side.BaseAllReqRow;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.SideTableInfo;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
@@ -60,14 +60,12 @@
  *
  * @author xuqianjin
  */
-public class CassandraAllReqRow extends AllReqRow {
+public class CassandraAllReqRow extends BaseAllReqRow {
 
     private static final long serialVersionUID = 54015343561288219L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraAllReqRow.class);
 
-    private static final String cassandra_DRIVER = "com.cassandra.jdbc.Driver";
-
     private static final int CONN_RETRY_NUM = 3;
 
     private static final int FETCH_SIZE = 1000;
@@ -77,7 +75,7 @@ public class CassandraAllReqRow extends AllReqRow {
 
     private AtomicReference<Map<String, List<Map<String, Object>>>> cacheRef = new AtomicReference<>();
 
-    public CassandraAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
+    public CassandraAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(new com.dtstack.flink.sql.side.cassandra.CassandraAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllSideInfo.java

@@ -20,8 +20,8 @@
 
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.SideInfo;
-import com.dtstack.flink.sql.side.SideTableInfo;
+import com.dtstack.flink.sql.side.BaseSideInfo;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
 import com.dtstack.flink.sql.util.ParseUtils;
 import org.apache.calcite.sql.SqlNode;
@@ -37,16 +37,16 @@
  *
  * @author xuqianjin
  */
-public class CassandraAllSideInfo extends SideInfo {
+public class CassandraAllSideInfo extends BaseSideInfo {
 
     private static final long serialVersionUID = -8690814317653033557L;
 
-    public CassandraAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
+    public CassandraAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
 
     @Override
-    public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
+    public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
         CassandraSideTableInfo cassandraSideTableInfo = (CassandraSideTableInfo) sideTableInfo;
 
         sqlCondition = "select ${selectField} from ${tableName} ";

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -30,11 +30,11 @@
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
 import com.dtstack.flink.sql.enums.ECacheContentType;
-import com.dtstack.flink.sql.side.AsyncReqRow;
+import com.dtstack.flink.sql.side.BaseAsyncReqRow;
 import com.dtstack.flink.sql.side.CacheMissVal;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.SideTableInfo;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
 import com.google.common.base.Function;
@@ -56,7 +56,6 @@
 import java.net.InetAddress;
 import java.sql.Timestamp;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -66,7 +65,7 @@
  *
  * @author xuqianjin
  */
-public class CassandraAsyncReqRow extends AsyncReqRow {
+public class CassandraAsyncReqRow extends BaseAsyncReqRow {
 
     private static final long serialVersionUID = 6631584128079864735L;
 
@@ -82,7 +81,7 @@ public class CassandraAsyncReqRow extends AsyncReqRow {
     private transient ListenableFuture session;
     private transient CassandraSideTableInfo cassandraSideTableInfo;
 
-    public CassandraAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
+    public CassandraAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(new com.dtstack.flink.sql.side.cassandra.CassandraAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncSideInfo.java

@@ -20,8 +20,8 @@
 
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.SideInfo;
-import com.dtstack.flink.sql.side.SideTableInfo;
+import com.dtstack.flink.sql.side.BaseSideInfo;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
 import com.dtstack.flink.sql.util.ParseUtils;
 import org.apache.calcite.sql.SqlBasicCall;
@@ -39,16 +39,16 @@
  *
  * @author xuqianjin
  */
-public class CassandraAsyncSideInfo extends SideInfo {
+public class CassandraAsyncSideInfo extends BaseSideInfo {
 
     private static final long serialVersionUID = -4403313049809013362L;
 
-    public CassandraAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
+    public CassandraAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
 
     @Override
-    public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
+    public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
         CassandraSideTableInfo cassandraSideTableInfo = (CassandraSideTableInfo) sideTableInfo;
 
         String sideTableName = joinInfo.getSideTableName();

cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java

@@ -19,26 +19,24 @@
 
 package com.dtstack.flink.sql.side.cassandra.table;
 
-import com.dtstack.flink.sql.table.AbsSideTableParser;
-import com.dtstack.flink.sql.table.TableInfo;
+import com.dtstack.flink.sql.table.AbstractSideTableParser;
+import com.dtstack.flink.sql.table.AbstractTableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 
-import java.math.BigDecimal;
-import java.sql.Date;
 import java.sql.Timestamp;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
+import static com.dtstack.flink.sql.table.AbstractTableInfo.PARALLELISM_KEY;
 
 /**
  * Reason:
  * Date: 2018/11/22
  *
  * @author xuqianjin
  */
-public class CassandraSideParser extends AbsSideTableParser {
+public class CassandraSideParser extends AbstractSideTableParser {
 
     private final static String SIDE_SIGN_KEY = "sideSignKey";
 
@@ -73,7 +71,7 @@ public CassandraSideParser() {
     }
 
     @Override
-    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+    public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
         com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo cassandraSideTableInfo = new com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo();
         cassandraSideTableInfo.setName(tableName);
         parseFieldsInfo(fieldsInfo, cassandraSideTableInfo);
@@ -96,7 +94,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         return cassandraSideTableInfo;
     }
 
-    private void dealSideSign(Matcher matcher, TableInfo tableInfo) {
+    private void dealSideSign(Matcher matcher, AbstractTableInfo tableInfo) {
     }
 
     @Override

cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideTableInfo.java

@@ -19,7 +19,7 @@
 
 package com.dtstack.flink.sql.side.cassandra.table;
 
-import com.dtstack.flink.sql.side.SideTableInfo;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.google.common.base.Preconditions;
 
 /**
@@ -28,7 +28,7 @@
  *
  * @author xuqianjin
  */
-public class CassandraSideTableInfo extends SideTableInfo {
+public class CassandraSideTableInfo extends AbstractSideTableInfo {
 
     private static final long serialVersionUID = -5556431094535478915L;
 

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -48,7 +48,7 @@
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
-import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
+import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends DtRichOutputFormat<Tuple2> {
+public class CassandraOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraSink.java

@@ -22,7 +22,7 @@
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.cassandra.table.CassandraTableInfo;
-import com.dtstack.flink.sql.table.TargetTableInfo;
+import com.dtstack.flink.sql.table.AbstractTargetTableInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -63,7 +63,7 @@ public CassandraSink() {
     }
 
     @Override
-    public CassandraSink genStreamSink(TargetTableInfo targetTableInfo) {
+    public CassandraSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         CassandraTableInfo cassandraTableInfo = (CassandraTableInfo) targetTableInfo;
         this.address = cassandraTableInfo.getAddress();
         this.tableName = cassandraTableInfo.getTableName();

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/table/CassandraSinkParser.java

@@ -19,21 +19,21 @@
 
 package com.dtstack.flink.sql.sink.cassandra.table;
 
-import com.dtstack.flink.sql.table.AbsTableParser;
-import com.dtstack.flink.sql.table.TableInfo;
+import com.dtstack.flink.sql.table.AbstractTableParser;
+import com.dtstack.flink.sql.table.AbstractTableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 
 import java.util.Map;
 
-import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
+import static com.dtstack.flink.sql.table.AbstractTableInfo.PARALLELISM_KEY;
 
 /**
  * Reason:
  * Date: 2018/11/22
  *
  * @author xuqianjin
  */
-public class CassandraSinkParser extends AbsTableParser {
+public class CassandraSinkParser extends AbstractTableParser {
 
     public static final String ADDRESS_KEY = "address";
 
@@ -60,7 +60,7 @@ public class CassandraSinkParser extends AbsTableParser {
     public static final String POOL_TIMEOUT_MILLIS_KEY = "poolTimeoutMillis";
 
     @Override
-    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+    public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
         CassandraTableInfo cassandraTableInfo = new CassandraTableInfo();
         cassandraTableInfo.setName(tableName);
         parseFieldsInfo(fieldsInfo, cassandraTableInfo);

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/table/CassandraTableInfo.java

@@ -19,7 +19,7 @@
 
 package com.dtstack.flink.sql.sink.cassandra.table;
 
-import com.dtstack.flink.sql.table.TargetTableInfo;
+import com.dtstack.flink.sql.table.AbstractTargetTableInfo;
 import com.google.common.base.Preconditions;
 
 /**
@@ -28,7 +28,7 @@
  *
  * @author xuqianjin
  */
-public class CassandraTableInfo extends TargetTableInfo {
+public class CassandraTableInfo extends AbstractTargetTableInfo {
 
     private static final String CURR_TYPE = "cassandra";