[fix-36431][logminer]logminer支持解析toDate函数

Author: yanghuaiGit

flinkx-oraclelogminer/flinkx-oraclelogminer-reader/src/main/java/com/dtstack/flinkx/oraclelogminer/format/LogMinerConnection.java

@@ -124,7 +124,7 @@ public LogMinerConnection(LogMinerConfig logMinerConfig) {
     }
 
     public void connect() {
-        PreparedStatement preparedStatement = null;
+
         try {
             ClassUtil.forName(logMinerConfig.getDriverName(), getClass().getClassLoader());
 
@@ -133,19 +133,36 @@ public void connect() {
             oracleVersion = connection.getMetaData().getDatabaseMajorVersion();
             isOracle10 = oracleVersion == 10;
 
-            if(isOracle10){
-                //oracle10开启logMiner之前 需要设置会话级别的日期格式 否则sql语句会含有todate函数 而不是todate函数计算后的值
-                preparedStatement = connection.prepareStatement(SqlUtil.SQL_ALTER_DATE_FORMAT);
-                preparedStatement.execute();
-                preparedStatement = connection.prepareStatement(SqlUtil.NLS_TIMESTAMP_FORMAT);
-                preparedStatement.execute();
+            //修改session级别的 NLS_DATE_FORMAT 值为 "YYYY-MM-DD HH24:MI:SS"，否则在解析日志时 redolog出现 TO_DATE('18-APR-21', 'DD-MON-RR')
+            boolean isAlterDateFormat = false;
+            try {
+                try (PreparedStatement preparedStatement = connection.prepareStatement(SqlUtil.SQL_QUERY_NLS_DATE_FORMAT)) {
+                    try (ResultSet resultSet = preparedStatement.executeQuery(SqlUtil.SQL_QUERY_NLS_DATE_FORMAT)) {
+                        while (resultSet.next()) {
+                            String nlsDateFormat = resultSet.getString(1);
+                            isAlterDateFormat = !nlsDateFormat.equalsIgnoreCase("YYYY-MM-DD HH24:MI:SS");
+                            LOG.info("nlsDateFormat {}, isAlterDateFormat {}", nlsDateFormat, isAlterDateFormat);
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                LOG.info("query nlsDateFormat failed ,exception is {}", ExceptionUtil.getErrorMessage(e));
+            }
+
+            if (isAlterDateFormat) {
+                try (PreparedStatement preparedStatement = connection.prepareStatement(SqlUtil.SQL_ALTER_DATE_FORMAT)) {
+                    preparedStatement.execute();
+                }
+                try (PreparedStatement preparedStatement = connection.prepareStatement(SqlUtil.NLS_TIMESTAMP_FORMAT)) {
+                    preparedStatement.execute();
+                }
             }
 
            LOG.info("get connection successfully, url:{}, username:{}, Oracle version：{}", logMinerConfig.getJdbcUrl(), logMinerConfig.getUsername(), oracleVersion);
         } catch (Exception e){
             String message = String.format("get connection failed，url:[%s], username:[%s], e:%s", logMinerConfig.getJdbcUrl(), logMinerConfig.getUsername(), ExceptionUtil.getErrorMessage(e));
             LOG.error(message);
-            closeResources(null, preparedStatement, connection);
+            closeResources(null, null, connection);
             throw new RuntimeException(message, e);
         }
     }

flinkx-oraclelogminer/flinkx-oraclelogminer-reader/src/main/java/com/dtstack/flinkx/oraclelogminer/format/LogMinerListener.java

@@ -117,7 +117,7 @@ public void run() {
             try {
                 if (logMinerConnection.hasNext()) {
                     log = logMinerConnection.next();
-                    queue.put(logParser.parse(log, logMinerConnection.isOracle10));
+                    queue.put(logParser.parse(log));
                 } else {
                     logMinerConnection.closeStmt();
                     logMinerConnection.startOrUpdateLogMiner(positionManager.getPosition());

flinkx-oraclelogminer/flinkx-oraclelogminer-reader/src/main/java/com/dtstack/flinkx/oraclelogminer/format/LogParser.java

@@ -142,18 +142,16 @@ public void visit(final EqualsTo expr){
         });
     }
 
-    public QueueData parse(QueueData pair, boolean isOracle10) throws JSQLParserException {
+    public QueueData parse(QueueData pair) throws JSQLParserException {
         Map<String, Object> logData = pair.getData();
         String schema = MapUtils.getString(logData, "schema");
         String tableName = MapUtils.getString(logData, "tableName");
         String operation = MapUtils.getString(logData, "operation");
         String sqlLog = MapUtils.getString(logData, "sqlLog");
+        LOG.debug("before alter sqlLog, sqlLog = {}",sqlLog);
         String sqlRedo = sqlLog.replace("IS NULL", "= NULL");
-        //只有oracle10需要进行toDate toTimestamp转换
-        LOG.debug("before parse toDate/toTimestamp sqlRedo = {}",sqlRedo);
-        if (isOracle10) {
-            sqlRedo = parseToTimeStamp(parseToDate(sqlRedo));
-        }
+        sqlRedo = parseToTimeStamp(parseToDate(sqlRedo));
+
         Timestamp timestamp = (Timestamp)MapUtils.getObject(logData, "opTime");
 
         Map<String,Object> message = new LinkedHashMap<>();

flinkx-oraclelogminer/flinkx-oraclelogminer-reader/src/main/java/com/dtstack/flinkx/oraclelogminer/util/SqlUtil.java

@@ -285,6 +285,9 @@ public class SqlUtil {
 
     public final static String SQL_GET_LOG_FILE_START_POSITION_BY_TIME_10 = "select min(FIRST_CHANGE#) FIRST_CHANGE# from (select FIRST_CHANGE# from v$log where TO_DATE(?, 'YYYY-MM-DD HH24:MI:SS') > FIRST_TIME union select FIRST_CHANGE# from v$archived_log where TO_DATE(?, 'YYYY-MM-DD HH24:MI:SS') between FIRST_TIME and NEXT_TIME and standby_dest='NO' and name is not null)";
 
+    //查询会话级别 NLS_DATE_FORMAT 参数值
+    public final static String SQL_QUERY_NLS_DATE_FORMAT="select VALUE from nls_session_parameters where parameter = 'NLS_DATE_FORMAT'";
+
     //修改当前会话的date日期格式
     public final static String SQL_ALTER_DATE_FORMAT ="ALTER SESSION SET NLS_DATE_FORMAT = 'YYYY-MM-DD HH24:MI:SS'";