[INLONG-4099][Manager] Support create Iceberg resource (#4192)

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/IcebergPartition.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.enums;
+
+import org.apache.inlong.manager.common.util.Preconditions;
+
+import java.util.Locale;
+
+/**
+ * Iceberg partition type
+ */
+public enum IcebergPartition {
+    IDENTITY,
+    BUCKET,
+    TRUNCATE,
+    YEAR,
+    MONTH,
+    DAY,
+    HOUR,
+    ;
+
+    /**
+     * Get partition type from name
+     */
+    public static IcebergPartition forName(String name) {
+        Preconditions.checkNotNull(name, "IcebergPartition should not be null");
+        for (IcebergPartition value : values()) {
+            if (value.toString().equals(name) || value.toString().equals(name.toUpperCase(Locale.ROOT))) {
+                return value;
+            }
+        }
+        throw new IllegalArgumentException(String.format("Unsupported IcebergPartition : %s", name));
+    }
+
+    @Override
+    public String toString() {
+        return name();
+    }
+}

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/IcebergType.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.enums;
+
+import lombok.Getter;
+
+/**
+ * Iceberg data type
+ */
+public enum IcebergType {
+
+    BOOLEAN("boolean"),
+    INT("int"),
+    LONG("long"),
+    FLOAT("float"),
+    DOUBLE("double"),
+    DECIMAL("decimal(P,S)"),
+    DATE("date"),
+    TIME("time"),
+    TIMESTAMP("timestamp"),
+    TIMESTAMPTZ("timestamptz"),
+    STRING("string"),
+    UUID("uuid"),
+    FIXED("fixed(L)"),
+    BINARY("binary");
+
+    @Getter
+    private String type;
+
+    IcebergType(String type) {
+        this.type = type;
+    }
+
+    /**
+     * Get type from name
+     */
+    public static IcebergType forType(String type) {
+        for (IcebergType ibType : values()) {
+            if (ibType.getType().equalsIgnoreCase(type)) {
+                return ibType;
+            }
+        }
+        throw new IllegalArgumentException(String.format("invalid iceberg type = %s", type));
+    }
+}

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergColumnInfo.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.pojo.sink.iceberg;
+
+import lombok.Data;
+
+/**
+ * Iceberg column info
+ */
+@Data
+public class IcebergColumnInfo {
+
+    private String name;
+    private String type;
+    private String desc;
+    private boolean required;
+    private Integer length;
+    private String partitionStrategy;
+    private Integer precision;
+    private Integer scale;
+    private Integer bucketNum;
+    private Integer width;
+}

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkDTO.java

@@ -28,6 +28,7 @@
 import org.apache.inlong.manager.common.exceptions.BusinessException;
 
 import javax.validation.constraints.NotNull;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -102,4 +103,17 @@ public static IcebergSinkDTO getFromJson(@NotNull String extParams) {
         }
     }
 
+    /**
+     * Get Iceberg table info
+     */
+    public static IcebergTableInfo getIcebergTableInfo(IcebergSinkDTO icebergInfo, List<IcebergColumnInfo> columnList) {
+        IcebergTableInfo info = new IcebergTableInfo();
+        info.setDbName(icebergInfo.getDbName());
+        info.setTableName(icebergInfo.getTableName());
+        info.setFileFormat(icebergInfo.getFileFormat());
+        info.setTblProperties(icebergInfo.getProperties());
+        info.setColumns(columnList);
+        return info;
+    }
+
 }

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergTableInfo.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.pojo.sink.iceberg;
+
+import lombok.Data;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Iceberg table info
+ */
+@Data
+public class IcebergTableInfo {
+
+    private String dbName;
+    private String tableName;
+    private String tableDesc;
+    private String fileFormat;
+    private Map<String, Object> tblProperties;
+    private List<IcebergColumnInfo> columns;
+}

inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkFieldEntity.java

@@ -45,6 +45,8 @@ public class StreamSinkFieldEntity implements Serializable {
     private Integer fieldPrecision;
     private Integer fieldScale;
     private String partitionStrategy;
+    private Integer bucketNum;
+    private Integer width;
 
     private Integer isMetaField;
     private String fieldFormat;

inlong-manager/manager-service/pom.xml

@@ -167,6 +167,55 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-hive-metastore</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hive</groupId>
+            <artifactId>hive-metastore</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>servlet-api</artifactId>
+                    <groupId>javax.servlet</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>servlet-api</artifactId>
+                    <groupId>javax.servlet</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-mapreduce-client-core</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>hadoop-yarn-common</artifactId>
+                    <groupId>org.apache.hadoop</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
     </dependencies>
 
 </project>