Core: Implement source-ids to deal with multi arguments transforms

.palantir/revapi.yml

@@ -404,6 +404,119 @@ acceptedBreaks:
       old: "method org.apache.iceberg.orc.ORC.WriteBuilder org.apache.iceberg.orc.ORC.WriteBuilder::config(java.lang.String,\
         \ java.lang.String)"
       justification: "Removing deprecations for 1.2.0"
+  "1.10.0":
+    org.apache.iceberg:iceberg-api:
+    - code: "java.class.defaultSerializationChanged"
+      old: "class org.apache.iceberg.PartitionField"
+      new: "class org.apache.iceberg.PartitionField"
+      justification: "Adding source-ids field to support multi-arg transforms"
+    - code: "java.class.defaultSerializationChanged"
+      old: "class org.apache.iceberg.SortField"
+      new: "class org.apache.iceberg.SortField"
+      justification: "Adding source-ids field to support multi-arg transforms"
+    - code: "java.class.defaultSerializationChanged"
+      old: "class org.apache.iceberg.encryption.EncryptingFileIO"
+      new: "class org.apache.iceberg.encryption.EncryptingFileIO"
+      justification: "New method for Manifest List reading"
+    org.apache.iceberg:iceberg-core:
+    - code: "java.class.noLongerInheritsFromClass"
+      old: "class org.apache.iceberg.rest.auth.OAuth2Manager"
+      new: "class org.apache.iceberg.rest.auth.OAuth2Manager"
+      justification: "Removing deprecations for 1.11.0"
+    - code: "java.class.nowImplementsInterface"
+      old: "class org.apache.iceberg.rest.auth.OAuth2Manager"
+      new: "class org.apache.iceberg.rest.auth.OAuth2Manager"
+      justification: "Removing deprecations for 1.11.0"
+    - code: "java.class.removed"
+      old: "class org.apache.iceberg.PartitionStatsUtil"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.class.removed"
+      old: "class org.apache.iceberg.rest.auth.RefreshingAuthManager"
+      justification: "Removing deprecations for 1.11.0"
+    - code: "java.field.constantValueChanged"
+      old: "field org.apache.iceberg.rest.ResourcePaths.V1_TABLE_SCAN_PLAN"
+      new: "field org.apache.iceberg.rest.ResourcePaths.V1_TABLE_SCAN_PLAN"
+      justification: "Plan API is table scoped and path constant value should include\
+        \ namespace. No actual breakage because it never worked before with incorrect\
+        \ value."
+    - code: "java.field.constantValueChanged"
+      old: "field org.apache.iceberg.rest.ResourcePaths.V1_TABLE_SCAN_PLAN_SUBMIT"
+      new: "field org.apache.iceberg.rest.ResourcePaths.V1_TABLE_SCAN_PLAN_SUBMIT"
+      justification: "Plan API is table scoped and path constant value should include\
+        \ namespace. No actual breakage because it never worked before with incorrect\
+        \ value."
+    - code: "java.field.constantValueChanged"
+      old: "field org.apache.iceberg.rest.ResourcePaths.V1_TABLE_SCAN_PLAN_TASKS"
+      new: "field org.apache.iceberg.rest.ResourcePaths.V1_TABLE_SCAN_PLAN_TASKS"
+      justification: "Plan API is table scoped and path constant value should include\
+        \ namespace. No actual breakage because it never worked before with incorrect\
+        \ value."
+    - code: "java.method.removed"
+      old: "method java.lang.String org.apache.iceberg.RewriteTablePathUtil::stagingPath(java.lang.String,\
+        \ java.lang.String)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.RewriteTablePathUtil.RewriteResult<org.apache.iceberg.DataFile>\
+        \ org.apache.iceberg.RewriteTablePathUtil::rewriteDataManifest(org.apache.iceberg.ManifestFile,\
+        \ org.apache.iceberg.io.OutputFile, org.apache.iceberg.io.FileIO, int, java.util.Map<java.lang.Integer,\
+        \ org.apache.iceberg.PartitionSpec>, java.lang.String, java.lang.String) throws\
+        \ java.io.IOException"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.RewriteTablePathUtil.RewriteResult<org.apache.iceberg.DeleteFile>\
+        \ org.apache.iceberg.RewriteTablePathUtil::rewriteDeleteManifest(org.apache.iceberg.ManifestFile,\
+        \ org.apache.iceberg.io.OutputFile, org.apache.iceberg.io.FileIO, int, java.util.Map<java.lang.Integer,\
+        \ org.apache.iceberg.PartitionSpec>, java.lang.String, java.lang.String, java.lang.String)\
+        \ throws java.io.IOException"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.Schema org.apache.iceberg.PartitionStatsHandler::schema(org.apache.iceberg.types.Types.StructType)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.TableMetadata org.apache.iceberg.TableMetadataParser::read(org.apache.iceberg.io.FileIO,\
+        \ org.apache.iceberg.io.InputFile)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.encryption.EncryptionManager org.apache.iceberg.encryption.EncryptionUtil::createEncryptionManager(java.util.Map<java.lang.String,\
+        \ java.lang.String>, org.apache.iceberg.encryption.KeyManagementClient)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.rest.responses.OAuthTokenResponse org.apache.iceberg.rest.auth.OAuth2Util::exchangeToken(org.apache.iceberg.rest.RESTClient,\
+        \ java.util.Map<java.lang.String, java.lang.String>, java.lang.String, java.lang.String,\
+        \ java.lang.String, java.lang.String, java.lang.String)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.rest.responses.OAuthTokenResponse org.apache.iceberg.rest.auth.OAuth2Util::exchangeToken(org.apache.iceberg.rest.RESTClient,\
+        \ java.util.Map<java.lang.String, java.lang.String>, java.lang.String, java.lang.String,\
+        \ java.lang.String, java.lang.String, java.lang.String, java.lang.String)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.rest.responses.OAuthTokenResponse org.apache.iceberg.rest.auth.OAuth2Util::fetchToken(org.apache.iceberg.rest.RESTClient,\
+        \ java.util.Map<java.lang.String, java.lang.String>, java.lang.String, java.lang.String)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.rest.responses.OAuthTokenResponse org.apache.iceberg.rest.auth.OAuth2Util::fetchToken(org.apache.iceberg.rest.RESTClient,\
+        \ java.util.Map<java.lang.String, java.lang.String>, java.lang.String, java.lang.String,\
+        \ java.lang.String)"
+      justification: "Removing deprecated code for 1.11.0"
+    - code: "java.method.visibilityReduced"
+      old: "method void org.apache.iceberg.PartitionStats::appendStats(org.apache.iceberg.PartitionStats)"
+      new: "method void org.apache.iceberg.PartitionStats::appendStats(org.apache.iceberg.PartitionStats)"
+      justification: "Changing deprecated code"
+    - code: "java.method.visibilityReduced"
+      old: "method void org.apache.iceberg.PartitionStats::deletedEntry(org.apache.iceberg.Snapshot)"
+      new: "method void org.apache.iceberg.PartitionStats::deletedEntry(org.apache.iceberg.Snapshot)"
+      justification: "Changing deprecated code"
+    - code: "java.method.visibilityReduced"
+      old: "method void org.apache.iceberg.PartitionStats::liveEntry(org.apache.iceberg.ContentFile<?>,\
+        \ org.apache.iceberg.Snapshot)"
+      new: "method void org.apache.iceberg.PartitionStats::liveEntry(org.apache.iceberg.ContentFile<?>,\
+        \ org.apache.iceberg.Snapshot)"
+      justification: "Changing deprecated code"
+    org.apache.iceberg:iceberg-data:
+    - code: "java.class.removed"
+      old: "class org.apache.iceberg.data.PartitionStatsHandler"
+      justification: "Removing deprecated code for 1.11.0"
   "1.2.0":
     org.apache.iceberg:iceberg-api:
     - code: "java.field.constantValueChanged"
@@ -1365,6 +1478,14 @@ acceptedBreaks:
       justification: "Removing deprecations for 1.10.0"
   "1.10.0":
     org.apache.iceberg:iceberg-api:
+      - code: "java.class.defaultSerializationChanged"
+        old: "class org.apache.iceberg.PartitionField"
+        new: "class org.apache.iceberg.PartitionField"
+        justification: "Adding source-ids field to support multi-arg transforms"
+      - code: "java.class.defaultSerializationChanged"
+        old: "class org.apache.iceberg.SortField"
+        new: "class org.apache.iceberg.SortField"
+        justification: "Adding source-ids field to support multi-arg transforms"
       - code: "java.class.defaultSerializationChanged"
         old: "class org.apache.iceberg.encryption.EncryptingFileIO"
         new: "class org.apache.iceberg.encryption.EncryptingFileIO"

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

@@ -19,26 +19,50 @@
 package org.apache.iceberg;
 
 import java.io.Serializable;
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Objects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.transforms.Transform;
 
 /** Represents a single field in a {@link PartitionSpec}. */
 public class PartitionField implements Serializable {
-  private final int sourceId;
+  private static final Joiner COMMA = Joiner.on(", ");
+
+  private final List<Integer> sourceIds;
   private final int fieldId;
   private final String name;
   private final Transform<?, ?> transform;
 
   PartitionField(int sourceId, int fieldId, String name, Transform<?, ?> transform) {
-    this.sourceId = sourceId;
+    this(ImmutableList.of(sourceId), fieldId, name, transform);
+  }
+
+  PartitionField(List<Integer> sourceIds, int fieldId, String name, Transform<?, ?> transform) {
+    this.sourceIds = ImmutableList.copyOf(sourceIds);
     this.fieldId = fieldId;
     this.name = name;
     this.transform = transform;
   }
 
   /** Returns the field id of the source field in the {@link PartitionSpec spec's} table schema. */
   public int sourceId() {
-    return sourceId;
+    Preconditions.checkState(
+        sourceIds.size() == 1,
+        "Use sourceIds() for multi-argument transforms, found %s source ids",
+        sourceIds.size());
+    return sourceIds.get(0);
+  }
+
+  /**
+   * Returns the field ids of all source fields for this partition field.
+   *
+   * <p>For single-argument transforms, this list contains one element. For multi-argument
+   * transforms, this list contains multiple source field ids.
+   */
+  public List<Integer> sourceIds() {
+    return sourceIds;
   }
 
   /** Returns the partition field id across all the table metadata's partition specs. */
@@ -58,7 +82,7 @@ public String name() {
 
   @Override
   public String toString() {
-    return fieldId + ": " + name + ": " + transform + "(" + sourceId + ")";
+    return fieldId + ": " + name + ": " + transform + "(" + COMMA.join(sourceIds) + ")";
   }
 
   @Override
@@ -70,14 +94,14 @@ public boolean equals(Object other) {
     }
 
     PartitionField that = (PartitionField) other;
-    return sourceId == that.sourceId
+    return sourceIds.equals(that.sourceIds)
         && fieldId == that.fieldId
         && name.equals(that.name)
         && transform.toString().equals(that.transform.toString());
   }
 
   @Override
   public int hashCode() {
-    return Objects.hashCode(sourceId, fieldId, name, transform);
+    return Objects.hashCode(sourceIds, fieldId, name, transform);
   }
 }

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

@@ -106,7 +106,7 @@ public UnboundPartitionSpec toUnbound() {
 
     for (PartitionField field : fields) {
       builder.addField(
-          field.transform().toString(), field.sourceId(), field.fieldId(), field.name());
+          field.transform().toString(), field.sourceIds(), field.fieldId(), field.name());
     }
 
     return builder.build();
@@ -130,7 +130,7 @@ public StructType partitionType() {
           List<Types.NestedField> structFields = Lists.newArrayListWithExpectedSize(fields.length);
 
           for (PartitionField field : fields) {
-            Type sourceType = schema.findType(field.sourceId());
+            Type sourceType = schema.findType(field.sourceIds().get(0));
             Type resultType = field.transform().getResultType(sourceType);
 
             // When the source field has been dropped we cannot determine the type
@@ -183,7 +183,7 @@ public Class<?>[] javaClasses() {
             if (field.transform() instanceof UnknownTransform) {
               classes[i] = Object.class;
             } else {
-              Type sourceType = schema.findType(field.sourceId());
+              Type sourceType = schema.findType(field.sourceIds().get(0));
               if (null == sourceType) {
                 // When the source field has been dropped we cannot determine the type
                 sourceType = Types.UnknownType.get();
@@ -248,7 +248,7 @@ public boolean compatibleWith(PartitionSpec other) {
     for (int i = 0; i < fields.length; i += 1) {
       PartitionField thisField = fields[i];
       PartitionField thatField = other.fields[i];
-      if (thisField.sourceId() != thatField.sourceId()
+      if (!thisField.sourceIds().equals(thatField.sourceIds())
           || !thisField.transform().toString().equals(thatField.transform().toString())
           || !thisField.name().equals(thatField.name())) {
         return false;
@@ -297,7 +297,9 @@ private ListMultimap<Integer, PartitionField> lazyFieldsBySourceId() {
               Multimaps.newListMultimap(
                   Maps.newHashMap(), () -> Lists.newArrayListWithCapacity(fields.length));
           for (PartitionField field : fields) {
-            multiMap.put(field.sourceId(), field);
+            for (int sourceId : field.sourceIds()) {
+              multiMap.put(sourceId, field);
+            }
           }
           this.fieldsBySourceId = multiMap;
         }
@@ -316,7 +318,7 @@ public Set<Integer> identitySourceIds() {
     Set<Integer> sourceIds = Sets.newHashSet();
     for (PartitionField field : fields()) {
       if (field.transform().isIdentity()) {
-        sourceIds.add(field.sourceId());
+        sourceIds.addAll(field.sourceIds());
       }
     }
 
@@ -427,15 +429,17 @@ private void checkAndAddPartitionName(String name, Integer sourceColumnId) {
     }
 
     private void checkForRedundantPartitions(PartitionField field) {
-      Map.Entry<Integer, String> dedupKey =
-          new AbstractMap.SimpleEntry<>(field.sourceId(), field.transform().dedupName());
-      PartitionField partitionField = dedupFields.get(dedupKey);
-      Preconditions.checkArgument(
-          partitionField == null,
-          "Cannot add redundant partition: %s conflicts with %s",
-          partitionField,
-          field);
-      dedupFields.put(dedupKey, field);
+      for (int sourceId : field.sourceIds()) {
+        Map.Entry<Integer, String> dedupKey =
+            new AbstractMap.SimpleEntry<>(sourceId, field.transform().dedupName());
+        PartitionField partitionField = dedupFields.get(dedupKey);
+        Preconditions.checkArgument(
+            partitionField == null,
+            "Cannot add redundant partition: %s conflicts with %s",
+            partitionField,
+            field);
+        dedupFields.put(dedupKey, field);
+      }
     }
 
     public Builder caseSensitive(boolean sensitive) {
@@ -621,6 +625,17 @@ Builder add(int sourceId, int fieldId, String name, Transform<?, ?> transform) {
       return this;
     }
 
+    Builder add(List<Integer> sourceIds, int fieldId, String name, Transform<?, ?> transform) {
+      checkAndAddPartitionName(name, sourceIds.get(0));
+      fields.add(new PartitionField(sourceIds, fieldId, name, transform));
+      lastAssignedFieldId.getAndAccumulate(fieldId, Math::max);
+      return this;
+    }
+
+    Builder add(List<Integer> sourceIds, String name, Transform<?, ?> transform) {
+      return add(sourceIds, nextFieldId(), name, transform);
+    }
+
     public PartitionSpec build() {
       return build(false);
     }
@@ -643,37 +658,39 @@ static void checkCompatibility(PartitionSpec spec, Schema schema) {
   static void checkCompatibility(PartitionSpec spec, Schema schema, boolean allowMissingFields) {
     final Map<Integer, Integer> parents = TypeUtil.indexParents(schema.asStruct());
     for (PartitionField field : spec.fields) {
-      Type sourceType = schema.findType(field.sourceId());
       Transform<?, ?> transform = field.transform();
-      // In the case the underlying field is dropped, we cannot check if they are compatible
-      if (allowMissingFields && sourceType == null) {
-        continue;
-      }
-      // In the case of a Version 1 partition-spec field gets deleted,
-      // it is replaced with a void transform, see:
-      // https://iceberg.apache.org/spec/#partition-transforms
-      // We don't care about the source type since a VoidTransform is always compatible and skip the
-      // checks
-      if (!transform.equals(Transforms.alwaysNull())) {
-        ValidationException.check(
-            sourceType != null, "Cannot find source column for partition field: %s", field);
-        ValidationException.check(
-            sourceType.isPrimitiveType(),
-            "Cannot partition by non-primitive source field: %s",
-            sourceType);
-        ValidationException.check(
-            transform.canTransform(sourceType),
-            "Invalid source type %s for transform: %s",
-            sourceType,
-            transform);
-        // The only valid parent types for a PartitionField are StructTypes. This must be checked
-        // recursively.
-        Integer parentId = parents.get(field.sourceId());
-        while (parentId != null) {
-          Type parentType = schema.findType(parentId);
+      for (int srcId : field.sourceIds()) {
+        Type sourceType = schema.findType(srcId);
+        // In the case the underlying field is dropped, we cannot check if they are compatible
+        if (allowMissingFields && sourceType == null) {
+          continue;
+        }
+        // In the case of a Version 1 partition-spec field gets deleted,
+        // it is replaced with a void transform, see:
+        // https://iceberg.apache.org/spec/#partition-transforms
+        // We don't care about the source type since a VoidTransform is always compatible and skip
+        // the checks
+        if (!transform.equals(Transforms.alwaysNull())) {
           ValidationException.check(
-              parentType.isStructType(), "Invalid partition field parent: %s", parentType);
-          parentId = parents.get(parentId);
+              sourceType != null, "Cannot find source column for partition field: %s", field);
+          ValidationException.check(
+              sourceType.isPrimitiveType(),
+              "Cannot partition by non-primitive source field: %s",
+              sourceType);
+          ValidationException.check(
+              transform.canTransform(sourceType),
+              "Invalid source type %s for transform: %s",
+              sourceType,
+              transform);
+          // The only valid parent types for a PartitionField are StructTypes. This must be checked
+          // recursively.
+          Integer parentId = parents.get(srcId);
+          while (parentId != null) {
+            Type parentType = schema.findType(parentId);
+            ValidationException.check(
+                parentType.isStructType(), "Invalid partition field parent: %s", parentType);
+            parentId = parents.get(parentId);
+          }
         }
       }
     }