Merge branch 'revert-ae75fae6' into 'v1.8.0_dev'

Revert "Merge branch 'v1.8.0_dev_cassandra' into 'v1.8.0_dev'"

This reverts merge request !155

See merge request !157

Author: todd5167

cassandra/cassandra-side/cassandra-all-side/pom.xml

@@ -38,7 +38,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-                                    <exclude>org.slf4j</exclude>
+
                                 </excludes>
                             </artifactSet>
                             <filters>

cassandra/cassandra-side/cassandra-async-side/pom.xml

@@ -54,7 +54,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-                                    <exclude>org.slf4j</exclude>
+
                                 </excludes>
                             </artifactSet>
                             <filters>

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

@@ -174,16 +174,9 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                 return;
             }
             inputParams.add(equalObj);
-            StringBuffer sqlTemp = stringBuffer.append(sideInfo.getEqualFieldList().get(i))
-                    .append(" = ");
-            if (equalObj instanceof String) {
-                sqlTemp.append("'" + equalObj + "'")
-                        .append(" and ");
-            } else {
-                sqlTemp.append(equalObj)
-                        .append(" and ");
-            }
-
+            stringBuffer.append(sideInfo.getEqualFieldList().get(i))
+                    .append(" = ").append("'" + equalObj + "'")
+                    .append(" and ");
         }
 
         String key = buildCacheKey(inputParams);
@@ -197,12 +190,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     dealMissKey(input, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
-                    List<Row> rowList = Lists.newArrayList();
-                    for (Object jsonArray : (List) val.getContent()) {
-                        Row row = fillData(input, jsonArray);
-                        rowList.add(row);
+
+                    for (Object rowArray : (List) val.getContent()) {
+                        Row row = fillData(input, rowArray);
+                        resultFuture.complete(Collections.singleton(row));
                     }
-                    resultFuture.complete(rowList);
+
                 } else {
                     throw new RuntimeException("not support cache obj type " + val.getType());
                 }
@@ -213,7 +206,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         //connect Cassandra
         connCassandraDB(cassandraSideTableInfo);
 
-        String sqlCondition = sideInfo.getSqlCondition() + " " + sqlWhere + "  ALLOW FILTERING ";
+        String sqlCondition = sideInfo.getSqlCondition() + " " + sqlWhere;
         System.out.println("sqlCondition:" + sqlCondition);
 
         ListenableFuture<ResultSet> resultSet = Futures.transformAsync(session,
@@ -238,15 +231,14 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                 cluster.closeAsync();
                 if (rows.size() > 0) {
                     List<com.datastax.driver.core.Row> cacheContent = Lists.newArrayList();
-                    List<Row> rowList = Lists.newArrayList();
                     for (com.datastax.driver.core.Row line : rows) {
                         Row row = fillData(input, line);
                         if (openCache()) {
                             cacheContent.add(line);
                         }
-                        rowList.add(row);
+                        resultFuture.complete(Collections.singleton(row));
                     }
-                    resultFuture.complete(rowList);
+
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }
@@ -293,6 +285,7 @@ public Row fillData(Row input, Object line) {
             }
         }
 
+        System.out.println("row:" + row.toString());
         return row;
     }
 

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

@@ -23,9 +23,6 @@
 import com.dtstack.flink.sql.table.TableInfo;
 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;
@@ -99,32 +96,4 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
 
     private static void dealSideSign(Matcher matcher, TableInfo tableInfo) {
     }
-
-    public Class dbTypeConvertToJavaType(String fieldType) {
-        switch (fieldType.toLowerCase()) {
-            case "bigint":
-                return Long.class;
-            case "int":
-            case "counter":
-                return Integer.class;
-
-            case "text":
-            case "inet":
-            case "varchar":
-            case "ascii":
-            case "timeuuid":
-                return String.class;
-
-            case "decimal":
-            case "float":
-                return Float.class;
-            case "double":
-                return Double.class;
-            case "timestamp":
-                return Timestamp.class;
-        }
-
-        throw new RuntimeException("不支持 " + fieldType + " 类型");
-
-    }
 }

cassandra/cassandra-sink/pom.xml

@@ -33,7 +33,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-                                    <exclude>org.slf4j</exclude>
+
                                 </excludes>
                             </artifactSet>
                             <filters>

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

@@ -48,9 +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.config.DirtyConfig;
 import com.dtstack.flink.sql.metric.MetricConstant;
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
 import org.apache.flink.api.common.io.RichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
@@ -75,7 +73,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends MetricOutputFormat {
+public class CassandraOutputFormat extends RichOutputFormat<Tuple2> {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);
@@ -96,9 +94,17 @@ public class CassandraOutputFormat extends MetricOutputFormat {
     protected String[] fieldNames;
     TypeInformation<?>[] fieldTypes;
 
+    private int batchInterval = 5000;
+
     private Cluster cluster;
     private Session session = null;
 
+    private int batchCount = 0;
+
+    private transient Counter outRecords;
+
+    private transient Meter outRecordsRate;
+
     public CassandraOutputFormat() {
     }
 
@@ -114,9 +120,7 @@ public void configure(Configuration parameters) {
      *                     I/O problem.
      */
     @Override
-    public void open(int taskNumber, int numTasks) {
-        initMetric();
-        initDirtyDataOutputStream();
+    public void open(int taskNumber, int numTasks) throws IOException {
         try {
             if (session == null) {
                 QueryOptions queryOptions = new QueryOptions();
@@ -179,6 +183,10 @@ public void open(int taskNumber, int numTasks) {
         }
     }
 
+    private void initMetric() {
+        outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(outRecords, 20));
+    }
 
     /**
      * Adds a record to the prepared statement.
@@ -218,7 +226,7 @@ private void insertWrite(Row row) {
                 resultSet.wasApplied();
             }
         } catch (Exception e) {
-            dealInsertError(row.toString(), e);
+            LOG.error("[upsert] is error:" + e.getMessage());
         }
     }
 
@@ -229,11 +237,7 @@ private String buildSql(Row row) {
             if (row.getField(index) == null) {
             } else {
                 fields.append(fieldNames[index] + ",");
-                if (row.getField(index) instanceof String) {
-                    values.append("'" + row.getField(index) + "'" + ",");
-                } else {
-                    values.append(row.getField(index) + ",");
-                }
+                values.append("'" + row.getField(index) + "'" + ",");
             }
         }
         fields.deleteCharAt(fields.length() - 1);
@@ -348,12 +352,6 @@ public CassandraFormatBuilder setPoolTimeoutMillis(Integer poolTimeoutMillis) {
             return this;
         }
 
-        public CassandraFormatBuilder setDirtyConfig(DirtyConfig dirtyConfig) {
-            format.dirtyConfig = dirtyConfig;
-            return this;
-        }
-
-
         /**
          * Finalizes the configuration and checks validity.
          *

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

@@ -20,7 +20,6 @@
 package com.dtstack.flink.sql.sink.cassandra;
 
 
-import com.dtstack.flink.sql.config.DirtyConfig;
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.cassandra.table.CassandraTableInfo;
 import com.dtstack.flink.sql.table.TargetTableInfo;
@@ -58,7 +57,6 @@ public class CassandraSink implements RetractStreamTableSink<Row>, IStreamSinkGe
     protected Integer readTimeoutMillis;
     protected Integer connectTimeoutMillis;
     protected Integer poolTimeoutMillis;
-    protected DirtyConfig dirtyConfig;
 
     public CassandraSink() {
         // TO DO NOTHING
@@ -79,7 +77,6 @@ public CassandraSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.readTimeoutMillis = cassandraTableInfo.getReadTimeoutMillis();
         this.connectTimeoutMillis = cassandraTableInfo.getConnectTimeoutMillis();
         this.poolTimeoutMillis = cassandraTableInfo.getPoolTimeoutMillis();
-        this.dirtyConfig = cassandraTableInfo.getDirtyConfig();
         return this;
     }
 
@@ -99,9 +96,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 .setConnectTimeoutMillis(this.connectTimeoutMillis)
                 .setPoolTimeoutMillis(this.poolTimeoutMillis)
                 .setFieldNames(this.fieldNames)
-                .setFieldTypes(this.fieldTypes)
-                .setDirtyConfig(dirtyConfig);
-
+                .setFieldTypes(this.fieldTypes);
 
         CassandraOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);