private void queryAndFillData(Map<String, List<Map<String, Object>>> tmpCache, Connection connection) throws SQLException {
        //load data from table
        String sql = sideInfo.getSqlCondition();
        Statement statement = connection.createStatement();
        statement.setFetchSize(getFetchSize());
        ResultSet resultSet = statement.executeQuery(sql);

        String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields(), ",");
        String[] sideFieldTypes = sideInfo.getSideTableInfo().getFieldTypes();
        String[] fields = sideInfo.getSideTableInfo().getFields();
        Map<String, String> sideFieldNamesAndTypes = Maps.newHashMap();
        for (int i = 0; i < fields.length; i++) {
            sideFieldNamesAndTypes.put(fields[i], sideFieldTypes[i]);
        }

        while (resultSet.next()) {
            Map<String, Object> oneRow = Maps.newHashMap();
            for (String fieldName : sideFieldNames) {
                Object object = resultSet.getObject(fieldName.trim());
                object = SwitchUtil.getTarget(object, sideFieldNamesAndTypes.get(fieldName));
                oneRow.put(fieldName.trim(), object);
            }

            String cacheKey = sideInfo.getEqualFieldList().stream()
                    .map(oneRow::get)
                    .map(String::valueOf)
                    .collect(Collectors.joining("_"));

            tmpCache.computeIfAbsent(cacheKey, key -> Lists.newArrayList())
                    .add(oneRow);
        }
        JdbcConnectionUtil.closeConnectionResource(resultSet, statement, connection, false);
    }
covert flink time attribute.Type information for indicating event or processing time.
However, it behaves like a regular SQL timestamp but is serialized as Long.

@param entry
@param obj
@return
rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java