rename

Author: todd5167

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java

@@ -0,0 +1,228 @@
+/*
+ * 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.rdb.all;
+
+import com.dtstack.flink.sql.side.AllReqRow;
+import com.dtstack.flink.sql.side.SideInfo;
+import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
+import com.dtstack.flink.sql.side.rdb.util.SwitchUtil;
+import org.apache.calcite.sql.JoinType;
+import org.apache.commons.collections.CollectionUtils;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.flink.table.runtime.types.CRow;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * side operator with cache for all(period reload)
+ * Date: 2018/11/26
+ * Company: www.dtstack.com
+ *
+ * @author maqi
+ */
+
+public abstract class AbstractRdbAllReqRow extends AllReqRow {
+
+    private static final long serialVersionUID = 2098635140857937718L;
+
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractRdbAllReqRow.class);
+
+    private static final int CONN_RETRY_NUM = 3;
+
+    private AtomicReference<Map<String, List<Map<String, Object>>>> cacheRef = new AtomicReference<>();
+
+    public AbstractRdbAllReqRow(SideInfo sideInfo) {
+        super(sideInfo);
+    }
+
+    @Override
+    public Row fillData(Row input, Object sideInput) {
+        Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
+        Row row = new Row(sideInfo.getOutFieldInfoList().size());
+        for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
+            Object obj = input.getField(entry.getValue());
+            boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
+
+            //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
+            if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
+                obj = ((Timestamp) obj).getTime();
+            }
+
+            row.setField(entry.getKey(), obj);
+        }
+
+        for (Map.Entry<Integer, String> entry : sideInfo.getSideFieldNameIndex().entrySet()) {
+            if (cacheInfo == null) {
+                row.setField(entry.getKey(), null);
+            } else {
+                row.setField(entry.getKey(), cacheInfo.get(entry.getValue()));
+            }
+        }
+
+        return row;
+    }
+
+    @Override
+    protected void initCache() throws SQLException {
+        Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
+        cacheRef.set(newCache);
+        loadData(newCache);
+    }
+
+    @Override
+    protected void reloadCache() {
+        //reload cacheRef and replace to old cacheRef
+        Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
+        try {
+            loadData(newCache);
+        } catch (SQLException e) {
+            LOG.error("", e);
+        }
+
+        cacheRef.set(newCache);
+        LOG.info("----- rdb all cacheRef reload end:{}", Calendar.getInstance());
+    }
+
+
+    @Override
+    public void flatMap(CRow value, Collector<CRow> out) throws Exception {
+        List<Object> inputParams = Lists.newArrayList();
+        for (Integer conValIndex : sideInfo.getEqualValIndex()) {
+            Object equalObj = value.row().getField(conValIndex);
+            if (equalObj == null) {
+                if (sideInfo.getJoinType() == JoinType.LEFT) {
+                    Row row = fillData(value.row(), null);
+                    out.collect(new CRow(row, value.change()));
+                }
+                return;
+            }
+            inputParams.add(equalObj);
+        }
+
+        String key = buildKey(inputParams);
+        List<Map<String, Object>> cacheList = cacheRef.get().get(key);
+        if (CollectionUtils.isEmpty(cacheList)) {
+            if (sideInfo.getJoinType() == JoinType.LEFT) {
+                Row row = fillData(value.row(), null);
+                out.collect(new CRow(row, value.change()));
+            } else {
+                return;
+            }
+
+            return;
+        }
+
+        for (Map<String, Object> one : cacheList) {
+            out.collect(new CRow(fillData(value.row(), one), value.change()));
+        }
+    }
+
+    private String buildKey(List<Object> equalValList) {
+        StringBuilder sb = new StringBuilder("");
+        for (Object equalVal : equalValList) {
+            sb.append(equalVal).append("_");
+        }
+
+        return sb.toString();
+    }
+
+    private String buildKey(Map<String, Object> val, List<String> equalFieldList) {
+        StringBuilder sb = new StringBuilder("");
+        for (String equalField : equalFieldList) {
+            sb.append(val.get(equalField)).append("_");
+        }
+
+        return sb.toString();
+    }
+
+    public abstract Connection getConn(String dbUrl, String userName, String password);
+
+
+    private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQLException {
+        RdbSideTableInfo tableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
+        Connection connection = null;
+
+        try {
+            for (int i = 0; i < CONN_RETRY_NUM; i++) {
+                try {
+                    connection = getConn(tableInfo.getUrl(), tableInfo.getUserName(), tableInfo.getPassword());
+                    break;
+                } catch (Exception e) {
+                    if (i == CONN_RETRY_NUM - 1) {
+                        throw new RuntimeException("", e);
+                    }
+                    try {
+                        String connInfo = "url:" + tableInfo.getUrl() + ";userName:" + tableInfo.getUserName() + ",pwd:" + tableInfo.getPassword();
+                        LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
+                        Thread.sleep(5 * 1000);
+                    } catch (InterruptedException e1) {
+                        LOG.error("", e1);
+                    }
+                }
+
+            }
+
+            //load data from table
+            String sql = sideInfo.getSqlCondition();
+            Statement statement = connection.createStatement();
+            statement.setFetchSize(getFetchSize());
+            ResultSet resultSet = statement.executeQuery(sql);
+            String[] sideFieldNames = sideInfo.getSideSelectFields().split(",");
+            String[] fields = sideInfo.getSideTableInfo().getFieldTypes();
+            while (resultSet.next()) {
+                Map<String, Object> oneRow = Maps.newHashMap();
+                for (String fieldName : sideFieldNames) {
+                    Object object = resultSet.getObject(fieldName.trim());
+                    int fieldIndex = sideInfo.getSideTableInfo().getFieldList().indexOf(fieldName.trim());
+                    object = SwitchUtil.getTarget(object, fields[fieldIndex]);
+                    oneRow.put(fieldName.trim(), object);
+                }
+
+                String cacheKey = buildKey(oneRow, sideInfo.getEqualFieldList());
+                List<Map<String, Object>> list = tmpCache.computeIfAbsent(cacheKey, key -> Lists.newArrayList());
+                list.add(oneRow);
+            }
+        } catch (Exception e) {
+            LOG.error("", e);
+        } finally {
+            if (connection != null) {
+                connection.close();
+            }
+        }
+    }
+
+    public int getFetchSize() {
+        return 1000;
+    }
+
+}