Add flink data type convertor to convert data types between iceberg data type and flink data type.

api/src/main/java/org/apache/iceberg/types/TypeUtil.java

@@ -186,6 +186,37 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
     return false;
   }
 
+  /**
+   * Check whether we could write the iceberg table with the user-provided write schema.
+   *
+   * @param tableSchema      the table schema written in iceberg meta data.
+   * @param writeSchema      the user-provided write schema.
+   * @param checkNullability If true, not allow to write optional values to a required field.
+   * @param checkOrdering    If true, not allow input schema to have different ordering than table schema.
+   */
+  public static void validateWriteSchema(Schema tableSchema, Schema writeSchema,
+                                         Boolean checkNullability, Boolean checkOrdering) {
+    List<String> errors;
+    if (checkNullability) {
+      errors = CheckCompatibility.writeCompatibilityErrors(tableSchema, writeSchema, checkOrdering);
+    } else {
+      errors = CheckCompatibility.typeCompatibilityErrors(tableSchema, writeSchema, checkOrdering);
+    }
+
+    if (!errors.isEmpty()) {
+      StringBuilder sb = new StringBuilder();
+      sb.append("Cannot write incompatible dataset to table with schema:\n")
+          .append(tableSchema)
+          .append("\nwrite schema:")
+          .append(writeSchema)
+          .append("\nProblems:");
+      for (String error : errors) {
+        sb.append("\n* ").append(error);
+      }
+      throw new IllegalArgumentException(sb.toString());
+    }
+  }
+
   /**
    * Interface for passing a function that assigns column IDs.
    */

build.gradle

@@ -235,6 +235,35 @@ project(':iceberg-data') {
   }
 }
 
+project(':iceberg-flink') {
+  dependencies {
+    compile project(':iceberg-api')
+    compile project(':iceberg-common')
+    compile project(':iceberg-core')
+    compile project(':iceberg-data')
+    compile project(':iceberg-orc')
+    compile project(':iceberg-parquet')
+
+    compileOnly "org.apache.flink:flink-streaming-java_2.12"
+    compileOnly "org.apache.flink:flink-streaming-java_2.12::tests"
+    compileOnly "org.apache.flink:flink-table-api-java-bridge_2.12"
+    compileOnly "org.apache.flink:flink-table-planner-blink_2.12"
+    compileOnly "org.apache.flink:flink-table-planner_2.12"
+    compileOnly "org.apache.hadoop:hadoop-hdfs"
+    compileOnly "org.apache.hadoop:hadoop-common"
+    compileOnly("org.apache.hadoop:hadoop-minicluster") {
+      exclude group: 'org.apache.avro', module: 'avro'
+    }
+
+    testCompile "org.apache.flink:flink-core"
+    testCompile "org.apache.flink:flink-runtime_2.12"
+    testCompile "org.apache.flink:flink-test-utils-junit"
+    testCompile("org.apache.flink:flink-test-utils_2.12") {
+      exclude group: "org.apache.curator", module: 'curator-test'
+    }
+  }
+}
+
 project(':iceberg-hive') {
   dependencies {
     compile project(':iceberg-core')