add iceberg for alink

Author: zhangjun0x01

…che/iceberg/flink/InputOutputFormat.java …che/flink/iceberg/InputOutputFormat.javaconnectors/connector-iceberg/iceberg-bridge/src/main/java/org/apache/iceberg/flink/InputOutputFormat.java renamed to connectors/connector-iceberg/iceberg-bridge/src/main/java/org/apache/flink/iceberg/InputOutputFormat.java connectors/connector-iceberg/iceberg-bridge/src/main/java/org/apache/iceberg/flink/InputOutputFormat.java renamed to connectors/connector-iceberg/iceberg-bridge/src/main/java/org/apache/flink/iceberg/InputOutputFormat.java

@@ -1,25 +1,26 @@
-package org.apache.iceberg.flink;
+package org.apache.flink.iceberg;
 
 import org.apache.flink.api.common.io.RichInputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.iceberg.source.FlinkInputFormat;
+import org.apache.flink.iceberg.source.FlinkSource;
+import org.apache.flink.iceberg.util.FlinkCompatibilityUtil;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.util.Preconditions;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
-import org.apache.iceberg.flink.source.FlinkInputFormat;
-import org.apache.iceberg.flink.source.FlinkSource;
-import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
 
 public class InputOutputFormat {
 
-  public static Tuple2<TypeInformation<RowData>, RichInputFormat<RowData, ?>> createInputFormat(
+  public static Tuple3<TableSchema, TypeInformation<RowData>, RichInputFormat<RowData, ?>> createInputFormat(
       StreamExecutionEnvironment execEnv, Catalog catalog, ObjectPath objectPath) {
 
     if (!(catalog instanceof FlinkCatalog)) {
@@ -38,13 +39,15 @@ public class InputOutputFormat {
       throw new UncheckedIOException(e);
     }
 
+    TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(icebergSchema));
+
     TypeInformation<RowData> typeInfo = FlinkCompatibilityUtil.toTypeInfo(FlinkSchemaUtil.convert(icebergSchema));
     FlinkInputFormat flinkInputFormat = FlinkSource.forRowData()
         .env(execEnv)
         .tableLoader(tableLoader)
         .table(table)
         .buildFormat();
-    return Tuple2.of(typeInfo, flinkInputFormat);
+    return Tuple3.of(tableSchema, typeInfo, flinkInputFormat);
   }
 
   private static TableLoader createTableLoader(FlinkCatalog catalog, ObjectPath objectPath) {