[Fix-670] [metadata] Fix Oracle column nullable error (#671)

* [Feature-654][web] Add task info tab

* [Feature-666][client] Capture column type conversion exception details in CDCSOURCE

* [Feature-668][web] Add task manager info

* [Fix-670] [metadata] Fix Oracle column nullable  error
Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/AbstractJdbcDriver.java

 package com.dlink.metadata.driver;
 
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.alibaba.druid.sql.SQLUtils;
 import com.alibaba.druid.sql.ast.SQLStatement;
 import com.dlink.assertion.Asserts;
@@ -11,11 +27,6 @@ import com.dlink.model.Schema;
 import com.dlink.model.Table;
 import com.dlink.result.SqlExplainResult;
 import com.dlink.utils.LogUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.sql.*;
-import java.util.*;
 
 /**
  * AbstractJdbcDriver
@@ -204,7 +215,7 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
                     field.setComment(columnComment);
                 }
                 if (columnList.contains(dbQuery.isNullable())) {
-                    field.setNullable(Asserts.isEqualsIgnoreCase(results.getString(dbQuery.isNullable()), "YES"));
+                    field.setNullable(Asserts.isEqualsIgnoreCase(results.getString(dbQuery.isNullable()), dbQuery.nullableValue()));
                 }
                 if (columnList.contains(dbQuery.characterSet())) {
                     field.setCharacterSet(results.getString(dbQuery.characterSet()));
@@ -239,13 +250,13 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
     public List<Column> listColumnsSortByPK(String schemaName, String tableName) {
         List<Column> columnList = listColumns(schemaName, tableName);
         List<Column> columnListSortByPK = new ArrayList<>();
-        for(Column column: columnList){
-            if(column.isKeyFlag()){
+        for (Column column : columnList) {
+            if (column.isKeyFlag()) {
                 columnListSortByPK.add(column);
             }
         }
-        for(Column column: columnList){
-            if(!column.isKeyFlag()){
+        for (Column column : columnList) {
+            if (!column.isKeyFlag()) {
                 columnListSortByPK.add(column);
             }
         }

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/query/AbstractDBQuery.java

@@ -137,4 +137,9 @@ public abstract class AbstractDBQuery implements IDBQuery {
     public String isPK() {
         return "PRI";
     }
+
+    @Override
+    public String nullableValue() {
+        return "YES";
+    }
 }

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/query/IDBQuery.java

@@ -151,4 +151,9 @@ public interface IDBQuery {
      * @return 主键值
      */
     String isPK();
+
+    /**
+     * @return 允许为空的值
+     */
+    String nullableValue();
 }

dlink-metadata/dlink-metadata-oracle/src/main/java/com/dlink/metadata/query/OracleQuery.java

@@ -21,21 +21,21 @@ public class OracleQuery extends AbstractDBQuery {
     @Override
     public String columnsSql(String schemaName, String tableName) {
         return "SELECT A.COLUMN_NAME, CASE WHEN A.DATA_TYPE='NUMBER' THEN "
-                + "(CASE WHEN A.DATA_PRECISION IS NULL THEN A.DATA_TYPE "
-                + "WHEN NVL(A.DATA_SCALE, 0) > 0 THEN A.DATA_TYPE||'('||A.DATA_PRECISION||','||A.DATA_SCALE||')' "
-                + "ELSE A.DATA_TYPE||'('||A.DATA_PRECISION||')' END) "
-                + "ELSE A.DATA_TYPE END DATA_TYPE,A.DATA_PRECISION NUMERIC_PRECISION,A.DATA_SCALE NUMERIC_SCALE,"
-                + " B.COMMENTS,A.NULLABLE,DECODE((select count(1) from all_constraints pc,all_cons_columns pcc"
-                + "  where pcc.column_name = A.column_name"
-                + "  and pcc.constraint_name = pc.constraint_name"
-                + "  and pc.constraint_type ='P'"
-                + "  and pcc.owner = upper(A.OWNER)"
-                + "  and pcc.table_name = upper(A.TABLE_NAME)),0,'','PRI') KEY "
-                + "FROM ALL_TAB_COLUMNS A "
-                + " INNER JOIN ALL_COL_COMMENTS B ON A.TABLE_NAME = B.TABLE_NAME AND A.COLUMN_NAME = B.COLUMN_NAME AND B.OWNER = '" + schemaName + "'"
-                + " LEFT JOIN ALL_CONSTRAINTS D ON D.TABLE_NAME = A.TABLE_NAME AND D.CONSTRAINT_TYPE = 'P' AND D.OWNER = '" + schemaName + "'"
-                + " LEFT JOIN ALL_CONS_COLUMNS C ON C.CONSTRAINT_NAME = D.CONSTRAINT_NAME AND C.COLUMN_NAME=A.COLUMN_NAME AND C.OWNER = '" + schemaName + "'"
-                + "WHERE A.OWNER = '" + schemaName + "' AND A.TABLE_NAME = '" + tableName + "' ORDER BY A.COLUMN_ID ";
+            + "(CASE WHEN A.DATA_PRECISION IS NULL THEN A.DATA_TYPE "
+            + "WHEN NVL(A.DATA_SCALE, 0) > 0 THEN A.DATA_TYPE||'('||A.DATA_PRECISION||','||A.DATA_SCALE||')' "
+            + "ELSE A.DATA_TYPE||'('||A.DATA_PRECISION||')' END) "
+            + "ELSE A.DATA_TYPE END DATA_TYPE,A.DATA_PRECISION NUMERIC_PRECISION,A.DATA_SCALE NUMERIC_SCALE,"
+            + " B.COMMENTS,A.NULLABLE,DECODE((select count(1) from all_constraints pc,all_cons_columns pcc"
+            + "  where pcc.column_name = A.column_name"
+            + "  and pcc.constraint_name = pc.constraint_name"
+            + "  and pc.constraint_type ='P'"
+            + "  and pcc.owner = upper(A.OWNER)"
+            + "  and pcc.table_name = upper(A.TABLE_NAME)),0,'','PRI') KEY "
+            + "FROM ALL_TAB_COLUMNS A "
+            + " INNER JOIN ALL_COL_COMMENTS B ON A.TABLE_NAME = B.TABLE_NAME AND A.COLUMN_NAME = B.COLUMN_NAME AND B.OWNER = '" + schemaName + "'"
+            + " LEFT JOIN ALL_CONSTRAINTS D ON D.TABLE_NAME = A.TABLE_NAME AND D.CONSTRAINT_TYPE = 'P' AND D.OWNER = '" + schemaName + "'"
+            + " LEFT JOIN ALL_CONS_COLUMNS C ON C.CONSTRAINT_NAME = D.CONSTRAINT_NAME AND C.COLUMN_NAME=A.COLUMN_NAME AND C.OWNER = '" + schemaName + "'"
+            + "WHERE A.OWNER = '" + schemaName + "' AND A.TABLE_NAME = '" + tableName + "' ORDER BY A.COLUMN_ID ";
     }
 
     @Override
@@ -82,4 +82,9 @@ public class OracleQuery extends AbstractDBQuery {
     public String isNullable() {
         return "NULLABLE";
     }
+
+    @Override
+    public String nullableValue() {
+        return "Y";
+    }
 }