API: Remove source type from Transform

.palantir/revapi.yml

@@ -18,6 +18,10 @@ acceptedBreaks:
     - code: "java.method.removed"
       old: "method org.apache.iceberg.RowDelta org.apache.iceberg.RowDelta::validateNoConflictingAppends(org.apache.iceberg.expressions.Expression)"
       justification: "Deprecations for 1.0 release"
+    - code: "java.class.defaultSerializationChanged"
+      old: "class org.apache.iceberg.PartitionKey"
+      new: "class org.apache.iceberg.PartitionKey"
+      justification: "Serialization across versions is not supported"
   release-base-0.13.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"

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

@@ -22,8 +22,9 @@
 import java.lang.reflect.Array;
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Function;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.util.SerializableFunction;
 
 /**
  * A struct of partition values.
@@ -36,7 +37,7 @@ public class PartitionKey implements StructLike, Serializable {
   private final PartitionSpec spec;
   private final int size;
   private final Object[] partitionTuple;
-  private final Transform[] transforms;
+  private final SerializableFunction[] transforms;
   private final Accessor<StructLike>[] accessors;
 
   @SuppressWarnings("unchecked")
@@ -46,7 +47,7 @@ public PartitionKey(PartitionSpec spec, Schema inputSchema) {
     List<PartitionField> fields = spec.fields();
     this.size = fields.size();
     this.partitionTuple = new Object[size];
-    this.transforms = new Transform[size];
+    this.transforms = new SerializableFunction[size];
     this.accessors = (Accessor<StructLike>[]) Array.newInstance(Accessor.class, size);
 
     Schema schema = spec.schema();
@@ -57,7 +58,7 @@ public PartitionKey(PartitionSpec spec, Schema inputSchema) {
           accessor != null,
           "Cannot build accessor for field: " + schema.findField(field.sourceId()));
       this.accessors[i] = accessor;
-      this.transforms[i] = field.transform();
+      this.transforms[i] = field.transform().bind(accessor.type());
     }
   }
 
@@ -101,7 +102,7 @@ public String toPath() {
   @SuppressWarnings("unchecked")
   public void partition(StructLike row) {
     for (int i = 0; i < partitionTuple.length; i += 1) {
-      Transform<Object, Object> transform = transforms[i];
+      Function<Object, Object> transform = transforms[i];
       partitionTuple[i] = transform.apply(accessors[i].get(row));
     }
   }

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

@@ -59,6 +59,7 @@ public class PartitionSpec implements Serializable {
   private final PartitionField[] fields;
   private transient volatile ListMultimap<Integer, PartitionField> fieldsBySourceId = null;
   private transient volatile Class<?>[] lazyJavaClasses = null;
+  private transient volatile StructType lazyPartitionType = null;
   private transient volatile List<PartitionField> fieldList = null;
   private final int lastAssignedFieldId;
 
@@ -123,16 +124,23 @@ public List<PartitionField> getFieldsBySourceId(int fieldId) {
 
   /** Returns a {@link StructType} for partition data defined by this spec. */
   public StructType partitionType() {
-    List<Types.NestedField> structFields = Lists.newArrayListWithExpectedSize(fields.length);
+    if (lazyPartitionType == null) {
+      synchronized (this) {
+        if (lazyPartitionType == null) {
+          List<Types.NestedField> structFields = Lists.newArrayListWithExpectedSize(fields.length);
 
-    for (int i = 0; i < fields.length; i += 1) {
-      PartitionField field = fields[i];
-      Type sourceType = schema.findType(field.sourceId());
-      Type resultType = field.transform().getResultType(sourceType);
-      structFields.add(Types.NestedField.optional(field.fieldId(), field.name(), resultType));
+          for (PartitionField field : fields) {
+            Type sourceType = schema.findType(field.sourceId());
+            Type resultType = field.transform().getResultType(sourceType);
+            structFields.add(Types.NestedField.optional(field.fieldId(), field.name(), resultType));
+          }
+
+          this.lazyPartitionType = Types.StructType.of(structFields);
+        }
+      }
     }
 
-    return Types.StructType.of(structFields);
+    return lazyPartitionType;
   }
 
   public Class<?>[] javaClasses() {
@@ -175,9 +183,11 @@ private String escape(String string) {
   public String partitionToPath(StructLike data) {
     StringBuilder sb = new StringBuilder();
     Class<?>[] javaClasses = javaClasses();
+    List<Types.NestedField> outputFields = partitionType().fields();
     for (int i = 0; i < javaClasses.length; i += 1) {
       PartitionField field = fields[i];
-      String valueString = field.transform().toHumanString(get(data, i, javaClasses[i]));
+      Type type = outputFields.get(i).type();
+      String valueString = field.transform().toHumanString(type, get(data, i, javaClasses[i]));
 
       if (i > 0) {
         sb.append("/");
@@ -412,10 +422,7 @@ Builder identity(String sourceName, String targetName) {
       checkAndAddPartitionName(targetName, sourceColumn.fieldId());
       PartitionField field =
           new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.identity(sourceColumn.type()));
+              sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.identity());
       checkForRedundantPartitions(field);
       fields.add(field);
       return this;
@@ -429,11 +436,7 @@ public Builder year(String sourceName, String targetName) {
       checkAndAddPartitionName(targetName);
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
       PartitionField field =
-          new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.year(sourceColumn.type()));
+          new PartitionField(sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.year());
       checkForRedundantPartitions(field);
       fields.add(field);
       return this;
@@ -447,11 +450,7 @@ public Builder month(String sourceName, String targetName) {
       checkAndAddPartitionName(targetName);
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
       PartitionField field =
-          new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.month(sourceColumn.type()));
+          new PartitionField(sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.month());
       checkForRedundantPartitions(field);
       fields.add(field);
       return this;
@@ -465,11 +464,7 @@ public Builder day(String sourceName, String targetName) {
       checkAndAddPartitionName(targetName);
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
       PartitionField field =
-          new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.day(sourceColumn.type()));
+          new PartitionField(sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.day());
       checkForRedundantPartitions(field);
       fields.add(field);
       return this;
@@ -483,11 +478,7 @@ public Builder hour(String sourceName, String targetName) {
       checkAndAddPartitionName(targetName);
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
       PartitionField field =
-          new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.hour(sourceColumn.type()));
+          new PartitionField(sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.hour());
       checkForRedundantPartitions(field);
       fields.add(field);
       return this;
@@ -502,10 +493,7 @@ public Builder bucket(String sourceName, int numBuckets, String targetName) {
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
       fields.add(
           new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.bucket(sourceColumn.type(), numBuckets)));
+              sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.bucket(numBuckets)));
       return this;
     }
 
@@ -518,10 +506,7 @@ public Builder truncate(String sourceName, int width, String targetName) {
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
       fields.add(
           new PartitionField(
-              sourceColumn.fieldId(),
-              nextFieldId(),
-              targetName,
-              Transforms.truncate(sourceColumn.type(), width)));
+              sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.truncate(width)));
       return this;
     }
 
@@ -545,16 +530,10 @@ public Builder alwaysNull(String sourceName) {
 
     // add a partition field with an auto-increment partition field id starting from
     // PARTITION_DATA_ID_START
-    Builder add(int sourceId, String name, String transform) {
+    Builder add(int sourceId, String name, Transform<?, ?> transform) {
       return add(sourceId, nextFieldId(), name, transform);
     }
 
-    Builder add(int sourceId, int fieldId, String name, String transform) {
-      Types.NestedField column = schema.findField(sourceId);
-      Preconditions.checkNotNull(column, "Cannot find source column: %s", sourceId);
-      return add(sourceId, fieldId, name, Transforms.fromString(column.type(), transform));
-    }
-
     Builder add(int sourceId, int fieldId, String name, Transform<?, ?> transform) {
       checkAndAddPartitionName(name, sourceId);
       fields.add(new PartitionField(sourceId, fieldId, name, transform));

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

@@ -36,7 +36,6 @@
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.Types;
 
 /** A sort order that defines how data and delete files should be ordered in a table. */
 public class SortOrder implements Serializable {
@@ -253,14 +252,6 @@ private Builder addSortField(Term term, SortDirection direction, NullOrder nullO
       return this;
     }
 
-    Builder addSortField(
-        String transformAsString, int sourceId, SortDirection direction, NullOrder nullOrder) {
-      Types.NestedField column = schema.findField(sourceId);
-      ValidationException.check(column != null, "Cannot find source column: %s", sourceId);
-      Transform<?, ?> transform = Transforms.fromString(column.type(), transformAsString);
-      return addSortField(transform, sourceId, direction, nullOrder);
-    }
-
     Builder addSortField(
         Transform<?, ?> transform, int sourceId, SortDirection direction, NullOrder nullOrder) {
       SortField sortField = new SortField(transform, sourceId, direction, nullOrder);
@@ -293,7 +284,7 @@ SortOrder buildUnchecked() {
 
     private Transform<?, ?> toTransform(BoundTerm<?> term) {
       if (term instanceof BoundReference) {
-        return Transforms.identity(term.type());
+        return Transforms.identity();
       } else if (term instanceof BoundTransform) {
         return ((BoundTransform<?, ?>) term).transform();
       } else {

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

@@ -20,6 +20,8 @@
 
 import java.util.List;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.transforms.Transforms;
 
 public class UnboundPartitionSpec {
 
@@ -52,9 +54,9 @@ private PartitionSpec.Builder copyToBuilder(Schema schema) {
 
     for (UnboundPartitionField field : fields) {
       if (field.partitionId != null) {
-        builder.add(field.sourceId, field.partitionId, field.name, field.transformAsString);
+        builder.add(field.sourceId, field.partitionId, field.name, field.transform);
       } else {
-        builder.add(field.sourceId, field.name, field.transformAsString);
+        builder.add(field.sourceId, field.name, field.transform);
       }
     }
 
@@ -94,13 +96,17 @@ UnboundPartitionSpec build() {
   }
 
   static class UnboundPartitionField {
-    private final String transformAsString;
+    private final Transform<?, ?> transform;
     private final int sourceId;
     private final Integer partitionId;
     private final String name;
 
+    public Transform<?, ?> transform() {
+      return transform;
+    }
+
     public String transformAsString() {
-      return transformAsString;
+      return transform.toString();
     }
 
     public int sourceId() {
@@ -117,7 +123,7 @@ public String name() {
 
     private UnboundPartitionField(
         String transformAsString, int sourceId, Integer partitionId, String name) {
-      this.transformAsString = transformAsString;
+      this.transform = Transforms.fromString(transformAsString);
       this.sourceId = sourceId;
       this.partitionId = partitionId;
       this.name = name;

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

@@ -21,6 +21,8 @@
 import java.util.Collections;
 import java.util.List;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.transforms.Transforms;
 
 public class UnboundSortOrder {
   private static final UnboundSortOrder UNSORTED_ORDER =
@@ -38,8 +40,7 @@ public SortOrder bind(Schema schema) {
     SortOrder.Builder builder = SortOrder.builderFor(schema).withOrderId(orderId);
 
     for (UnboundSortField field : fields) {
-      builder.addSortField(
-          field.transformAsString, field.sourceId, field.direction, field.nullOrder);
+      builder.addSortField(field.transform, field.sourceId, field.direction, field.nullOrder);
     }
 
     return builder.build();
@@ -49,8 +50,7 @@ SortOrder bindUnchecked(Schema schema) {
     SortOrder.Builder builder = SortOrder.builderFor(schema).withOrderId(orderId);
 
     for (UnboundSortField field : fields) {
-      builder.addSortField(
-          field.transformAsString, field.sourceId, field.direction, field.nullOrder);
+      builder.addSortField(field.transform, field.sourceId, field.direction, field.nullOrder);
     }
 
     return builder.buildUnchecked();
@@ -114,21 +114,21 @@ UnboundSortOrder build() {
   }
 
   static class UnboundSortField {
-    private final String transformAsString;
+    private final Transform<?, ?> transform;
     private final int sourceId;
     private final SortDirection direction;
     private final NullOrder nullOrder;
 
     private UnboundSortField(
         String transformAsString, int sourceId, SortDirection direction, NullOrder nullOrder) {
-      this.transformAsString = transformAsString;
+      this.transform = Transforms.fromString(transformAsString);
       this.sourceId = sourceId;
       this.direction = direction;
       this.nullOrder = nullOrder;
     }
 
     public String transformAsString() {
-      return transformAsString;
+      return transform.toString();
     }
 
     public int sourceId() {

api/src/main/java/org/apache/iceberg/expressions/BoundTransform.java

@@ -21,6 +21,7 @@
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.types.Type;
+import org.apache.iceberg.util.SerializableFunction;
 
 /**
  * A transform expression.
@@ -31,15 +32,17 @@
 public class BoundTransform<S, T> implements BoundTerm<T> {
   private final BoundReference<S> ref;
   private final Transform<S, T> transform;
+  private final SerializableFunction<S, T> func;
 
   BoundTransform(BoundReference<S> ref, Transform<S, T> transform) {
     this.ref = ref;
     this.transform = transform;
+    this.func = transform.bind(ref.type());
   }
 
   @Override
   public T eval(StructLike struct) {
-    return transform.apply(ref.eval(struct));
+    return func.apply(ref.eval(struct));
   }
 
   @Override

api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java

@@ -18,17 +18,17 @@
  */
 package org.apache.iceberg.expressions;
 
+import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Types;
 
 /** Expression utility methods. */
 public class ExpressionUtil {
-  private static final Transform<CharSequence, Integer> HASH_FUNC =
-      Transforms.bucket(Types.StringType.get(), Integer.MAX_VALUE);
+  private static final Function<Object, Integer> HASH_FUNC =
+      Transforms.bucket(Integer.MAX_VALUE).bind(Types.StringType.get());
   private static final Pattern DATE = Pattern.compile("\\d\\d\\d\\d-\\d\\d-\\d\\d");
   private static final Pattern TIME = Pattern.compile("\\d\\d:\\d\\d(:\\d\\d(.\\d{1,6})?)?");
   private static final Pattern TIMESTAMP =