[INLONG-4128][Manager] Abstracting the logic for creating Hive tables (#4129)

* [INLONG-4128][Manager] Abstracting the logic for creating Hive tables

* [INLONG-4128][Manager] Format the code

* [INLONG-4128][Manager] Select fields by the sink id

Author: healchow

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/query/hive/HiveColumnInfoBean.java inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/DataNodeType.java

@@ -15,10 +15,22 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.manager.common.pojo.query.hive;
+package org.apache.inlong.manager.common.enums;
 
-import org.apache.inlong.manager.common.pojo.query.ColumnInfoBean;
+import java.util.Locale;
 
-public class HiveColumnInfoBean extends ColumnInfoBean {
+public enum DataNodeType {
+
+    HIVE,
+    KAFKA,
+    ICEBERG,
+    CLICKHOUSE,
+    ES,
+    ;
+
+    @Override
+    public String toString() {
+        return this.name().toUpperCase(Locale.ROOT);
+    }
 
 }

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/query/ColumnQueryBean.java

@@ -15,15 +15,24 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.manager.common.pojo.query;
+package org.apache.inlong.manager.common.pojo.sink.hive;
 
 import lombok.Data;
 
+/**
+ * Hive column info
+ */
 @Data
-public class ColumnInfoBean {
+public class HiveColumnInfo {
+
+    private String name;
+    private String type;
+    private String desc;
+    private String format;
+
+    private String oldColumnName;
 
-    private String columnName;
-    private String columnType;
-    private String columnDesc;
+    private boolean isPartition;
+    private String exampleData;
 
-}
+}

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/query/ConnectionInfo.java

@@ -24,6 +24,7 @@
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
 import org.apache.inlong.manager.common.exceptions.BusinessException;
 
@@ -110,6 +111,9 @@ public static HiveSinkDTO getFromRequest(HiveSinkRequest request) {
                 .build();
     }
 
+    /**
+     * Get Hive sink info from JSON string
+     */
     public static HiveSinkDTO getFromJson(@NotNull String extParams) {
         try {
             OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
@@ -119,4 +123,33 @@ public static HiveSinkDTO getFromJson(@NotNull String extParams) {
         }
     }
 
+    /**
+     * Get Hive table info
+     */
+    public static HiveTableInfo getHiveTableInfo(HiveSinkDTO hiveInfo, List<HiveColumnInfo> columnList) {
+        HiveTableInfo tableInfo = new HiveTableInfo();
+        tableInfo.setDbName(hiveInfo.getDbName());
+        tableInfo.setTableName(hiveInfo.getTableName());
+
+        // Set partition fields
+        if (CollectionUtils.isNotEmpty(hiveInfo.getPartitionFieldList())) {
+            for (HivePartitionField field : hiveInfo.getPartitionFieldList()) {
+                HiveColumnInfo columnInfo = new HiveColumnInfo();
+                columnInfo.setName(field.getFieldName());
+                columnInfo.setPartition(true);
+                columnInfo.setType("string");
+                columnList.add(columnInfo);
+            }
+        }
+        tableInfo.setColumns(columnList);
+
+        // set terminated symbol
+        if (hiveInfo.getDataSeparator() != null) {
+            char ch = (char) Integer.parseInt(hiveInfo.getDataSeparator());
+            tableInfo.setFieldTerSymbol(String.valueOf(ch));
+        }
+
+        return tableInfo;
+    }
+
 }