Flink: Add table update code for schema comparison and evolution

This adds the classes around schema / spec comparison and evolution. A breakdown
of the classes follows:

 # CompareSchemasVisitor

Compares the user-provided schema against the current table schema.

 # EvolveSchemaVisitor

Computes the changes required to the table schema to be compatible with the
user-provided schema.

 # ParititonSpecEvolution

Code for checking compatibility with the user-provided PartitionSpec and
computing a set of changes to rewrite the PartitionSpec.

 # TableDataCache

Cache which holds all relevant metadata of a table like its name, branch,
schema, partition spec. Also holds a cache of past comparison results for a
given table's schema and the user-provided input schema.

 # Table Updater

Core logic to compare and create/update a table given a user-provided input
schema.

Broken out of apache#12424, depends on apache#12996.

Author: mxm

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java

@@ -0,0 +1,254 @@
+/*
+ * 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.iceberg.flink.sink.dynamic;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.schema.SchemaWithPartnerVisitor;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+/** Visitor class which compares two schemas and decides whether they are compatible. */
+public class CompareSchemasVisitor
+    extends SchemaWithPartnerVisitor<Integer, CompareSchemasVisitor.Result> {
+
+  private final Schema tableSchema;
+
+  private CompareSchemasVisitor(Schema tableSchema) {
+    this.tableSchema = tableSchema;
+  }
+
+  public static Result visit(Schema dataSchema, Schema tableSchema) {
+    return visit(dataSchema, tableSchema, true);
+  }
+
+  public static Result visit(Schema dataSchema, Schema tableSchema, boolean caseSensitive) {
+    return visit(
+        dataSchema,
+        -1,
+        new CompareSchemasVisitor(tableSchema),
+        new PartnerIdByNameAccessors(tableSchema, caseSensitive));
+  }
+
+  @Override
+  public Result schema(Schema dataSchema, Integer tableSchemaId, Result downstream) {
+    if (tableSchemaId == null) {
+      return Result.INCOMPATIBLE;
+    }
+
+    return downstream;
+  }
+
+  @Override
+  public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result> fields) {
+    if (tableSchemaId == null) {
+      return Result.INCOMPATIBLE;
+    }
+
+    Result result = fields.stream().reduce(Result::merge).orElse(Result.INCOMPATIBLE);
+
+    if (result == Result.INCOMPATIBLE) {
+      return Result.INCOMPATIBLE;
+    }
+
+    Type tableSchemaType =
+        tableSchemaId == -1 ? tableSchema.asStruct() : tableSchema.findField(tableSchemaId).type();
+    if (!tableSchemaType.isStructType()) {
+      return Result.INCOMPATIBLE;
+    }
+
+    if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      return Result.CONVERSION_NEEDED;
+    }
+
+    for (int i = 0; i < struct.fields().size(); ++i) {
+      if (!struct
+          .fields()
+          .get(i)
+          .name()
+          .equals(tableSchemaType.asStructType().fields().get(i).name())) {
+        return Result.CONVERSION_NEEDED;
+      }
+    }
+
+    return result;
+  }
+
+  @Override
+  public Result field(Types.NestedField field, Integer tableSchemaId, Result typeResult) {
+    if (tableSchemaId == null) {
+      return Result.INCOMPATIBLE;
+    }
+
+    if (typeResult != Result.SAME) {
+      return typeResult;
+    }
+
+    if (tableSchema.findField(tableSchemaId).isRequired() && field.isOptional()) {
+      return Result.INCOMPATIBLE;
+    } else {
+      return Result.SAME;
+    }
+  }
+
+  @Override
+  public Result list(Types.ListType list, Integer tableSchemaId, Result elementsResult) {
+    if (tableSchemaId == null) {
+      return Result.INCOMPATIBLE;
+    }
+
+    return elementsResult;
+  }
+
+  @Override
+  public Result map(
+      Types.MapType map, Integer tableSchemaId, Result keyResult, Result valueResult) {
+    if (tableSchemaId == null) {
+      return Result.INCOMPATIBLE;
+    }
+
+    return keyResult.merge(valueResult);
+  }
+
+  @Override
+  @SuppressWarnings("checkstyle:CyclomaticComplexity")
+  public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
+    if (tableSchemaId == null) {
+      return Result.INCOMPATIBLE;
+    }
+
+    Type tableSchemaType = tableSchema.findField(tableSchemaId).type();
+    if (!tableSchemaType.isPrimitiveType()) {
+      return Result.INCOMPATIBLE;
+    }
+
+    Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
+    if (primitive.equals(tableSchemaPrimitiveType)) {
+      return Result.SAME;
+    } else if (primitive.equals(Types.IntegerType.get())
+        && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
+      return Result.CONVERSION_NEEDED;
+    } else if (primitive.equals(Types.FloatType.get())
+        && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+      return Result.CONVERSION_NEEDED;
+    } else if (primitive.equals(Types.DateType.get())
+        && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
+      return Result.CONVERSION_NEEDED;
+    } else if (primitive.typeId() == Type.TypeID.DECIMAL
+        && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
+      Types.DecimalType dataType = (Types.DecimalType) primitive;
+      Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
+      return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
+          ? Result.CONVERSION_NEEDED
+          : Result.INCOMPATIBLE;
+    } else {
+      return Result.INCOMPATIBLE;
+    }
+  }
+
+  static class PartnerIdByNameAccessors implements PartnerAccessors<Integer> {
+    private final Schema tableSchema;
+    private boolean caseSensitive = true;
+
+    PartnerIdByNameAccessors(Schema tableSchema) {
+      this.tableSchema = tableSchema;
+    }
+
+    private PartnerIdByNameAccessors(Schema tableSchema, boolean caseSensitive) {
+      this(tableSchema);
+      this.caseSensitive = caseSensitive;
+    }
+
+    @Override
+    public Integer fieldPartner(Integer tableSchemaFieldId, int fieldId, String name) {
+      Types.StructType struct;
+      if (tableSchemaFieldId == -1) {
+        struct = tableSchema.asStruct();
+      } else {
+        struct = tableSchema.findField(tableSchemaFieldId).type().asStructType();
+      }
+
+      Types.NestedField field =
+          caseSensitive ? struct.field(name) : struct.caseInsensitiveField(name);
+      if (field != null) {
+        return field.fieldId();
+      }
+
+      return null;
+    }
+
+    @Override
+    public Integer mapKeyPartner(Integer tableSchemaMapId) {
+      Types.NestedField mapField = tableSchema.findField(tableSchemaMapId);
+      if (mapField != null) {
+        return mapField.type().asMapType().fields().get(0).fieldId();
+      }
+
+      return null;
+    }
+
+    @Override
+    public Integer mapValuePartner(Integer tableSchemaMapId) {
+      Types.NestedField mapField = tableSchema.findField(tableSchemaMapId);
+      if (mapField != null) {
+        return mapField.type().asMapType().fields().get(1).fieldId();
+      }
+
+      return null;
+    }
+
+    @Override
+    public Integer listElementPartner(Integer tableSchemaListId) {
+      Types.NestedField listField = tableSchema.findField(tableSchemaListId);
+      if (listField != null) {
+        return listField.type().asListType().fields().get(0).fieldId();
+      }
+
+      return null;
+    }
+  }
+
+  public enum Result {
+    SAME(0),
+    CONVERSION_NEEDED(1),
+    INCOMPATIBLE(2);
+
+    private static final Map<Integer, Result> BY_ID = Maps.newHashMap();
+
+    static {
+      for (Result e : Result.values()) {
+        if (BY_ID.put(e.id, e) != null) {
+          throw new IllegalArgumentException("Duplicate id: " + e.id);
+        }
+      }
+    }
+
+    private final int id;
+
+    Result(int id) {
+      this.id = id;
+    }
+
+    private Result merge(Result other) {
+      return BY_ID.get(Math.max(this.id, other.id));
+    }
+  }
+}