API: Restore the type of the identity transform (#6242)

* API: Restore the type of the identity transform

This caused some regression for the Iceberg 1.1.0 release:

```
2022-11-21T12:05:46.6549795Z [ERROR] io.trino.plugin.iceberg.TestIcebergSystemTables.testManifestsTable  Time elapsed: 0.701 s  <<< FAILURE!
2022-11-21T12:05:46.6550853Z java.lang.AssertionError:
2022-11-21T12:05:46.6551986Z [Rows for query [SELECT added_data_files_count, existing_rows_count, added_rows_count, deleted_data_files_count, deleted_rows_count, partitions FROM test_schema."test_table$manifests"]]
2022-11-21T12:05:46.6553075Z Expecting:
2022-11-21T12:05:46.6553593Z   <(2, 0, 3, 0, 0, [[false, false, 18148, 18149]]), (2, 0, 3, 0, 0, [[false, false, 18147, 18148]])>
2022-11-21T12:05:46.6553980Z to contain exactly in any order:
2022-11-21T12:05:46.6554557Z   <[(2, 0, 3, 0, 0, [[false, false, 2019-09-08, 2019-09-09]]),
2022-11-21T12:05:46.6554992Z     (2, 0, 3, 0, 0, [[false, false, 2019-09-09, 2019-09-10]])]>
2022-11-21T12:05:46.6555273Z elements not found:
2022-11-21T12:05:46.6555804Z   <(2, 0, 3, 0, 0, [[false, false, 2019-09-08, 2019-09-09]]), (2, 0, 3, 0, 0, [[false, false, 2019-09-09, 2019-09-10]])>
2022-11-21T12:05:46.6556132Z and elements not expected:
2022-11-21T12:05:46.6556488Z   <(2, 0, 3, 0, 0, [[false, false, 18148, 18149]]), (2, 0, 3, 0, 0, [[false, false, 18147, 18148]])>
```

The system tables (manifests in this example above), would return
the days since epoch instead of a date.

* Restore equals

* Bind SortOrder as well

* Add todo

Author: Fokko

api/src/main/java/org/apache/iceberg/SortOrder.java

@@ -284,7 +284,7 @@ SortOrder buildUnchecked() {
 
     private Transform<?, ?> toTransform(BoundTerm<?> term) {
       if (term instanceof BoundReference) {
-        return Transforms.identity();
+        return Transforms.identity(term.type());
       } else if (term instanceof BoundTransform) {
         return ((BoundTransform<?, ?>) term).transform();
       } else {

api/src/main/java/org/apache/iceberg/UnboundSortOrder.java

@@ -23,6 +23,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Type;
 
 public class UnboundSortOrder {
   private static final UnboundSortOrder UNSORTED_ORDER =
@@ -40,7 +41,14 @@ public SortOrder bind(Schema schema) {
     SortOrder.Builder builder = SortOrder.builderFor(schema).withOrderId(orderId);
 
     for (UnboundSortField field : fields) {
-      builder.addSortField(field.transform, field.sourceId, field.direction, field.nullOrder);
+      Type sourceType = schema.findType(field.sourceId);
+      Transform<?, ?> transform;
+      if (sourceType != null) {
+        transform = Transforms.fromString(sourceType, field.transform.toString());
+      } else {
+        transform = field.transform;
+      }
+      builder.addSortField(transform, field.sourceId, field.direction, field.nullOrder);
     }
 
     return builder.build();

api/src/main/java/org/apache/iceberg/transforms/Identity.java

@@ -29,6 +29,18 @@
 class Identity<T> implements Transform<T, T> {
   private static final Identity<?> INSTANCE = new Identity<>();
 
+  private final Type type;
+
+  /**
+   * Instantiates a new Identity Transform
+   *
+   * @deprecated use {@link #get()} instead; will be removed in 2.0.0
+   */
+  @Deprecated
+  public static <I> Identity<I> get(Type type) {
+    return new Identity<>(type);
+  }
+
   @SuppressWarnings("unchecked")
   public static <I> Identity<I> get() {
     return (Identity<I>) INSTANCE;
@@ -48,14 +60,21 @@ public T apply(T t) {
     }
   }
 
-  private Identity() {}
+  private Identity() {
+    this(null);
+  }
+
+  private Identity(Type type) {
+    this.type = type;
+  }
 
   @Override
   public T apply(T value) {
     return value;
   }
 
   @Override
+  @SuppressWarnings("checkstyle:HiddenField")
   public SerializableFunction<T, T> bind(Type type) {
     Preconditions.checkArgument(canTransform(type), "Cannot bind to unsupported type: %s", type);
     return Apply.get();
@@ -66,6 +85,24 @@ public boolean canTransform(Type maybePrimitive) {
     return maybePrimitive.isPrimitiveType();
   }
 
+  /**
+   * Returns a human-readable String representation of a transformed value.
+   *
+   * <p>null values will return "null"
+   *
+   * @param value a transformed value
+   * @return a human-readable String representation of the value
+   * @deprecated use {@link #toHumanString(Type, Object)} instead; will be removed in 2.0.0
+   */
+  @Deprecated
+  @Override
+  public String toHumanString(T value) {
+    if (this.type != null) {
+      return toHumanString(this.type, value);
+    }
+    return Transform.super.toHumanString(value);
+  }
+
   @Override
   public Type getResultType(Type sourceType) {
     return sourceType;
@@ -105,6 +142,23 @@ public boolean isIdentity() {
     return true;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    // Can be removed with 2.0.0 deprecation of get(Type)
+    if (this == o) {
+      return true;
+    } else if (o instanceof Identity) {
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  public int hashCode() {
+    // Can be removed with 2.0.0 deprecation of get(Type)
+    return 0;
+  }
+
   @Override
   public String toString() {
     return "identity";

api/src/main/java/org/apache/iceberg/transforms/Transforms.java

@@ -81,7 +81,7 @@ private Transforms() {}
     }
 
     if (transform.equalsIgnoreCase("identity")) {
-      return Identity.get();
+      return Identity.get(type);
     }
 
     try {
@@ -111,7 +111,7 @@ private Transforms() {}
    */
   @Deprecated
   public static <T> Transform<T, T> identity(Type type) {
-    return Identity.get();
+    return Identity.get(type);
   }
 
   /**

api/src/test/java/org/apache/iceberg/transforms/TestIdentity.java

@@ -69,6 +69,18 @@ public void testDateHumanString() {
         "Should produce identical date", dateString, identity.toHumanString(date, dateLit.value()));
   }
 
+  @Test
+  public void testDateHumanStringDeprecated() {
+    Types.DateType date = Types.DateType.get();
+    Transform<Integer, Integer> identity = Transforms.identity(date);
+
+    String dateString = "2017-12-01";
+    Literal<Integer> dateLit = Literal.of(dateString).to(date);
+
+    Assert.assertEquals(
+        "Should produce identical date", dateString, identity.toHumanString(dateLit.value()));
+  }
+
   @Test
   public void testTimeHumanString() {
     Types.TimeType time = Types.TimeType.get();