feat(dlm): supports sharding using unique indexes (#1327)

* DLM supports sharding using unique indexes

* upgrade sdk to snapshot version.

* fix shard key validator

* upgrade sdk to 1.0.9

Author: guowl3

pom.xml

@@ -119,7 +119,7 @@
         <mina.version>2.1.6</mina.version>
 
         <!-- data-lifecycle-manager version -->
-        <data-lifecycle-manager.version>1.0.8</data-lifecycle-manager.version>
+        <data-lifecycle-manager.version>1.0.9</data-lifecycle-manager.version>
 
         <!-- plugin version -->
         <formatter-maven-plugin.version>2.11.0</formatter-maven-plugin.version>

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/DataSourceInfoBuilder.java

@@ -45,15 +45,15 @@ public static DataSourceInfo build(ConnectionConfig connectionConfig) {
                 dataSourceInfo.setIp(connectionConfig.getHost());
                 dataSourceInfo.setPort(connectionConfig.getPort());
                 dataSourceInfo.setFullUserName(connectionConfig.getUsername());
-                dataSourceInfo.setDbType(DataBaseType.MYSQL.name());
+                dataSourceInfo.setDatabaseType(DataBaseType.MYSQL);
                 break;
             }
             case OB_MYSQL: {
                 dataSourceInfo
                         .setObProxy(String.format("%s:%s", connectionConfig.getHost(), connectionConfig.getPort()));
                 dataSourceInfo
                         .setFullUserName(OBConsoleDataSourceFactory.getUsername(connectionConfig));
-                dataSourceInfo.setDbType(DataBaseType.OCEANBASEV10.name());
+                dataSourceInfo.setDatabaseType(DataBaseType.OCEANBASEV10);
                 dataSourceInfo.setSysUser(connectionConfig.getSysTenantUsername());
                 dataSourceInfo.setClusterName(connectionConfig.getClusterName());
                 if (StringUtils.isNotEmpty(connectionConfig.getSysTenantPassword())) {

server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/flowtask/AbstractDlmJobPreprocessor.java

@@ -78,7 +78,7 @@ public ScheduleEntity buildScheduleEntity(CreateFlowInstanceReq req) {
     public void checkTableAndCondition(ConnectionSession connectionSession, Database sourceDb,
             List<DataArchiveTableConfig> tables,
             List<OffsetConfig> variables) {
-        checkPrimaryKey(connectionSession, sourceDb.getName(), tables);
+        checkShardKey(connectionSession, sourceDb.getName(), tables);
         Map<DataArchiveTableConfig, String> sqlMap = getDataArchiveSqls(sourceDb, tables, variables);
         checkDataArchiveSql(connectionSession, sqlMap);
     }
@@ -89,14 +89,14 @@ public void checkDatasource(ConnectionConfig datasource) {
         }
     }
 
-    private void checkPrimaryKey(ConnectionSession connectionSession, String databaseName,
+    private void checkShardKey(ConnectionSession connectionSession, String databaseName,
             List<DataArchiveTableConfig> tables) {
         SyncJdbcExecutor syncJdbcExecutor = connectionSession.getSyncJdbcExecutor(
                 ConnectionSessionConstants.CONSOLE_DS_KEY);
         SqlBuilder sqlBuilder = new MySQLSqlBuilder();
-        sqlBuilder.append("select table_name from information_schema.COLUMNS where ");
-        sqlBuilder.append(String.format("table_schema='%s' ", databaseName));
-        sqlBuilder.append("and column_key = 'PRI' group by table_name;");
+        sqlBuilder.append(
+                "SELECT TABLE_NAME from INFORMATION_SCHEMA.STATISTICS where NON_UNIQUE = 0 AND NULLABLE != 'YES' ");
+        sqlBuilder.append(String.format("AND TABLE_SCHEMA='%s' GROUP BY TABLE_NAME", databaseName));
         HashSet<String> tableNames =
                 new HashSet<>(syncJdbcExecutor.query(sqlBuilder.toString(), (rs, num) -> rs.getString(1)));
         tables.forEach(tableConfig -> {