diff --git a/.palantir/revapi.yml b/.palantir/revapi.yml index cd4afe6fdc8d..82e3477e9518 100644 --- a/.palantir/revapi.yml +++ b/.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.RewriteTablePathUtil::rewriteDataManifest(org.apache.iceberg.ManifestFile,\ + \ org.apache.iceberg.io.OutputFile, org.apache.iceberg.io.FileIO, int, java.util.Map, 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.RewriteTablePathUtil::rewriteDeleteManifest(org.apache.iceberg.ManifestFile,\ + \ org.apache.iceberg.io.OutputFile, org.apache.iceberg.io.FileIO, int, java.util.Map, 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, 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)" + 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)" + 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)" + 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)" + 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" diff --git a/api/src/main/java/org/apache/iceberg/PartitionField.java b/api/src/main/java/org/apache/iceberg/PartitionField.java index 3ed765a89834..119cc86586ac 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionField.java +++ b/api/src/main/java/org/apache/iceberg/PartitionField.java @@ -19,18 +19,28 @@ 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 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 sourceIds, int fieldId, String name, Transform transform) { + this.sourceIds = ImmutableList.copyOf(sourceIds); this.fieldId = fieldId; this.name = name; this.transform = transform; @@ -38,7 +48,21 @@ public class PartitionField implements Serializable { /** 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. + * + *

For single-argument transforms, this list contains one element. For multi-argument + * transforms, this list contains multiple source field ids. + */ + public List 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,7 +94,7 @@ 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()); @@ -78,6 +102,6 @@ public boolean equals(Object other) { @Override public int hashCode() { - return Objects.hashCode(sourceId, fieldId, name, transform); + return Objects.hashCode(sourceIds, fieldId, name, transform); } } diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 90d2dc259dd1..4b778593b060 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/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 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 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 identitySourceIds() { Set 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 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 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 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 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 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); + } } } } diff --git a/api/src/main/java/org/apache/iceberg/SortField.java b/api/src/main/java/org/apache/iceberg/SortField.java index d7f110a26e3f..676740a935ae 100644 --- a/api/src/main/java/org/apache/iceberg/SortField.java +++ b/api/src/main/java/org/apache/iceberg/SortField.java @@ -19,20 +19,30 @@ package org.apache.iceberg; import java.io.Serializable; +import java.util.List; import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.transforms.Transform; /** A field in a {@link SortOrder}. */ public class SortField implements Serializable { private final Transform transform; - private final int sourceId; + private final List sourceIds; private final SortDirection direction; private final NullOrder nullOrder; SortField(Transform transform, int sourceId, SortDirection direction, NullOrder nullOrder) { + this(transform, ImmutableList.of(sourceId), direction, nullOrder); + } + + SortField( + Transform transform, + List sourceIds, + SortDirection direction, + NullOrder nullOrder) { this.transform = transform; - this.sourceId = sourceId; + this.sourceIds = ImmutableList.copyOf(sourceIds); this.direction = direction; this.nullOrder = nullOrder; } @@ -51,7 +61,17 @@ public Transform transform() { /** Returns the field id of the source field in the {@link SortOrder sort order's} table schema */ public int sourceId() { - return sourceId; + return sourceIds.get(0); + } + + /** + * Returns the field ids of all source fields for this sort field. + * + *

For single-argument transforms, this list contains one element. For multi-argument + * transforms, this list contains multiple source field ids. + */ + public List sourceIds() { + return sourceIds; } /** Returns the sort direction */ @@ -73,7 +93,7 @@ public NullOrder nullOrder() { public boolean satisfies(SortField other) { if (Objects.equals(this, other)) { return true; - } else if (sourceId != other.sourceId + } else if (!sourceIds.equals(other.sourceIds) || direction != other.direction || nullOrder != other.nullOrder) { return false; @@ -84,7 +104,10 @@ public boolean satisfies(SortField other) { @Override public String toString() { - return transform + "(" + sourceId + ") " + direction + " " + nullOrder; + if (sourceIds.size() == 1) { + return transform + "(" + sourceIds.get(0) + ") " + direction + " " + nullOrder; + } + return transform + "(" + sourceIds + ") " + direction + " " + nullOrder; } @Override @@ -97,13 +120,13 @@ public boolean equals(Object other) { SortField that = (SortField) other; return transform.toString().equals(that.transform.toString()) - && sourceId == that.sourceId + && sourceIds.equals(that.sourceIds) && direction == that.direction && nullOrder == that.nullOrder; } @Override public int hashCode() { - return Objects.hash(transform, sourceId, direction, nullOrder); + return Objects.hash(transform, sourceIds, direction, nullOrder); } } diff --git a/api/src/main/java/org/apache/iceberg/SortOrder.java b/api/src/main/java/org/apache/iceberg/SortOrder.java index e98e05bd1e88..52127a26856d 100644 --- a/api/src/main/java/org/apache/iceberg/SortOrder.java +++ b/api/src/main/java/org/apache/iceberg/SortOrder.java @@ -126,8 +126,13 @@ public UnboundSortOrder toUnbound() { UnboundSortOrder.Builder builder = UnboundSortOrder.builder().withOrderId(orderId); for (SortField field : fields) { - builder.addSortField( - field.transform().toString(), field.sourceId(), field.direction(), field.nullOrder()); + if (field.sourceIds().size() > 1) { + builder.addSortField( + field.transform().toString(), field.sourceIds(), field.direction(), field.nullOrder()); + } else { + builder.addSortField( + field.transform().toString(), field.sourceId(), field.direction(), field.nullOrder()); + } } return builder.build(); @@ -260,6 +265,16 @@ Builder addSortField( return this; } + Builder addSortField( + Transform transform, + List sourceIds, + SortDirection direction, + NullOrder nullOrder) { + SortField sortField = new SortField(transform, sourceIds, direction, nullOrder); + fields.add(sortField); + return this; + } + public SortOrder build() { SortOrder sortOrder = buildUnchecked(); checkCompatibility(sortOrder, schema); @@ -297,18 +312,20 @@ SortOrder buildUnchecked() { public static void checkCompatibility(SortOrder sortOrder, Schema schema) { for (SortField field : sortOrder.fields) { - Type sourceType = schema.findType(field.sourceId()); - ValidationException.check( - sourceType != null, "Cannot find source column for sort field: %s", field); - ValidationException.check( - sourceType.isPrimitiveType(), - "Cannot sort by non-primitive source field: %s", - sourceType); - ValidationException.check( - field.transform().canTransform(sourceType), - "Invalid source type %s for transform: %s", - sourceType, - field.transform()); + for (int srcId : field.sourceIds()) { + Type sourceType = schema.findType(srcId); + ValidationException.check( + sourceType != null, "Cannot find source column for sort field: %s", field); + ValidationException.check( + sourceType.isPrimitiveType(), + "Cannot sort by non-primitive source field: %s", + sourceType); + ValidationException.check( + field.transform().canTransform(sourceType), + "Invalid source type %s for transform: %s", + sourceType, + field.transform()); + } } } } diff --git a/api/src/main/java/org/apache/iceberg/UnboundPartitionSpec.java b/api/src/main/java/org/apache/iceberg/UnboundPartitionSpec.java index 30b3cce35fc8..27e4ae3d0e47 100644 --- a/api/src/main/java/org/apache/iceberg/UnboundPartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/UnboundPartitionSpec.java @@ -19,6 +19,7 @@ package org.apache.iceberg; import java.util.List; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.transforms.Transform; import org.apache.iceberg.transforms.Transforms; @@ -58,17 +59,25 @@ private PartitionSpec.Builder copyToBuilder(Schema schema) { PartitionSpec.Builder builder = PartitionSpec.builderFor(schema).withSpecId(specId); for (UnboundPartitionField field : fields) { - Type fieldType = schema.findType(field.sourceId); + Type fieldType = schema.findType(field.sourceId()); Transform transform; if (fieldType != null) { - transform = Transforms.fromString(fieldType, field.transform.toString()); + transform = Transforms.fromString(fieldType, field.transformAsString()); } else { - transform = Transforms.fromString(field.transform.toString()); + transform = Transforms.fromString(field.transformAsString()); } - if (field.partitionId != null) { - builder.add(field.sourceId, field.partitionId, field.name, transform); + if (field.sourceIds().size() > 1) { + if (field.partitionId() != null) { + builder.add(field.sourceIds(), field.partitionId(), field.name(), transform); + } else { + builder.add(field.sourceIds(), field.name(), transform); + } } else { - builder.add(field.sourceId, field.name, transform); + if (field.partitionId() != null) { + builder.add(field.sourceId(), field.partitionId(), field.name(), transform); + } else { + builder.add(field.sourceId(), field.name(), transform); + } } } @@ -102,6 +111,17 @@ Builder addField(String transformAsString, int sourceId, String name) { return this; } + Builder addField( + String transformAsString, List sourceIds, int partitionId, String name) { + fields.add(new UnboundPartitionField(transformAsString, sourceIds, partitionId, name)); + return this; + } + + Builder addField(String transformAsString, List sourceIds, String name) { + fields.add(new UnboundPartitionField(transformAsString, sourceIds, null, name)); + return this; + } + UnboundPartitionSpec build() { return new UnboundPartitionSpec(specId, fields); } @@ -109,7 +129,7 @@ UnboundPartitionSpec build() { static class UnboundPartitionField { private final Transform transform; - private final int sourceId; + private final List sourceIds; private final Integer partitionId; private final String name; @@ -122,7 +142,11 @@ public String transformAsString() { } public int sourceId() { - return sourceId; + return sourceIds.get(0); + } + + public List sourceIds() { + return sourceIds; } public Integer partitionId() { @@ -135,8 +159,13 @@ public String name() { private UnboundPartitionField( String transformAsString, int sourceId, Integer partitionId, String name) { + this(transformAsString, ImmutableList.of(sourceId), partitionId, name); + } + + private UnboundPartitionField( + String transformAsString, List sourceIds, Integer partitionId, String name) { this.transform = Transforms.fromString(transformAsString); - this.sourceId = sourceId; + this.sourceIds = ImmutableList.copyOf(sourceIds); this.partitionId = partitionId; this.name = name; } diff --git a/api/src/main/java/org/apache/iceberg/UnboundSortOrder.java b/api/src/main/java/org/apache/iceberg/UnboundSortOrder.java index ce9f6b1d6b2c..57a1b95182a0 100644 --- a/api/src/main/java/org/apache/iceberg/UnboundSortOrder.java +++ b/api/src/main/java/org/apache/iceberg/UnboundSortOrder.java @@ -20,6 +20,7 @@ import java.util.Collections; import java.util.List; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.transforms.Transform; import org.apache.iceberg.transforms.Transforms; @@ -41,14 +42,18 @@ public SortOrder bind(Schema schema) { SortOrder.Builder builder = SortOrder.builderFor(schema).withOrderId(orderId); for (UnboundSortField field : fields) { - Type sourceType = schema.findType(field.sourceId); + 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); + if (field.sourceIds().size() > 1) { + builder.addSortField(transform, field.sourceIds(), field.direction, field.nullOrder); + } else { + builder.addSortField(transform, field.sourceId(), field.direction, field.nullOrder); + } } return builder.build(); @@ -58,7 +63,11 @@ SortOrder bindUnchecked(Schema schema) { SortOrder.Builder builder = SortOrder.builderFor(schema).withOrderId(orderId); for (UnboundSortField field : fields) { - builder.addSortField(field.transform, field.sourceId, field.direction, field.nullOrder); + if (field.sourceIds().size() > 1) { + builder.addSortField(field.transform, field.sourceIds(), field.direction, field.nullOrder); + } else { + builder.addSortField(field.transform, field.sourceId(), field.direction, field.nullOrder); + } } return builder.buildUnchecked(); @@ -103,6 +112,15 @@ Builder addSortField( return this; } + Builder addSortField( + String transformAsString, + List sourceIds, + SortDirection direction, + NullOrder nullOrder) { + fields.add(new UnboundSortField(transformAsString, sourceIds, direction, nullOrder)); + return this; + } + UnboundSortOrder build() { if (fields.isEmpty()) { if (orderId != null && orderId != 0) { @@ -123,14 +141,22 @@ UnboundSortOrder build() { static class UnboundSortField { private final Transform transform; - private final int sourceId; + private final List sourceIds; private final SortDirection direction; private final NullOrder nullOrder; private UnboundSortField( String transformAsString, int sourceId, SortDirection direction, NullOrder nullOrder) { + this(transformAsString, ImmutableList.of(sourceId), direction, nullOrder); + } + + private UnboundSortField( + String transformAsString, + List sourceIds, + SortDirection direction, + NullOrder nullOrder) { this.transform = Transforms.fromString(transformAsString); - this.sourceId = sourceId; + this.sourceIds = ImmutableList.copyOf(sourceIds); this.direction = direction; this.nullOrder = nullOrder; } @@ -140,7 +166,11 @@ public String transformAsString() { } public int sourceId() { - return sourceId; + return sourceIds.get(0); + } + + public List sourceIds() { + return sourceIds; } public SortDirection direction() { diff --git a/core/src/main/java/org/apache/iceberg/PartitionSpecParser.java b/core/src/main/java/org/apache/iceberg/PartitionSpecParser.java index 7becf0c62943..39202758db2d 100644 --- a/core/src/main/java/org/apache/iceberg/PartitionSpecParser.java +++ b/core/src/main/java/org/apache/iceberg/PartitionSpecParser.java @@ -24,17 +24,24 @@ import com.github.benmanes.caffeine.cache.Caffeine; import java.io.IOException; import java.util.Iterator; +import java.util.List; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.JsonUtil; import org.apache.iceberg.util.Pair; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class PartitionSpecParser { + private static final Logger LOG = LoggerFactory.getLogger(PartitionSpecParser.class); + private PartitionSpecParser() {} private static final String SPEC_ID = "spec-id"; private static final String FIELDS = "fields"; private static final String SOURCE_ID = "source-id"; + private static final String SOURCE_IDS = "source-ids"; private static final String FIELD_ID = "field-id"; private static final String TRANSFORM = "transform"; private static final String NAME = "name"; @@ -98,7 +105,16 @@ static void toJsonFields(UnboundPartitionSpec spec, JsonGenerator generator) thr generator.writeStartObject(); generator.writeStringField(NAME, field.name()); generator.writeStringField(TRANSFORM, field.transformAsString()); - generator.writeNumberField(SOURCE_ID, field.sourceId()); + if (field.sourceIds().size() == 1) { + generator.writeNumberField(SOURCE_ID, field.sourceId()); + } else { + generator.writeFieldName(SOURCE_IDS); + generator.writeStartArray(); + for (Integer value : field.sourceIds()) { + generator.writeNumber(value); + } + generator.writeEndArray(); + } generator.writeNumberField(FIELD_ID, field.partitionId()); generator.writeEndObject(); } @@ -132,15 +148,48 @@ private static void buildFromJsonFields(UnboundPartitionSpec.Builder builder, Js String name = JsonUtil.getString(NAME, element); String transform = JsonUtil.getString(TRANSFORM, element); - int sourceId = JsonUtil.getInt(SOURCE_ID, element); - // partition field ids are missing in old PartitionSpec, they always auto-increment from - // PARTITION_DATA_ID_START - if (element.has(FIELD_ID)) { - builder.addField(transform, sourceId, JsonUtil.getInt(FIELD_ID, element), name); - fieldIdCount++; + if (element.has(SOURCE_IDS)) { + JsonNode sourceIdsNode = element.get(SOURCE_IDS); + Preconditions.checkArgument( + sourceIdsNode.isArray() && sourceIdsNode.size() > 0, + "Cannot parse partition field, source-ids must be a non-empty array: %s", + element); + List sourceIds = Lists.newArrayListWithCapacity(sourceIdsNode.size()); + for (int i = 0; i < sourceIdsNode.size(); i++) { + sourceIds.add(sourceIdsNode.get(i).asInt()); + } + + if (element.has(SOURCE_ID)) { + int sourceId = JsonUtil.getInt(SOURCE_ID, element); + if (!sourceIds.get(0).equals(sourceId)) { + LOG.warn( + "Conflicting source-id ({}) and source-ids ({}) in partition field: {}", + sourceId, + sourceIds, + element); + } + } + + // partition field ids are missing in old PartitionSpec, they always auto-increment from + // PARTITION_DATA_ID_START + if (element.has(FIELD_ID)) { + builder.addField(transform, sourceIds, JsonUtil.getInt(FIELD_ID, element), name); + fieldIdCount++; + } else { + builder.addField(transform, sourceIds, name); + } } else { - builder.addField(transform, sourceId, name); + int sourceId = JsonUtil.getInt(SOURCE_ID, element); + + // partition field ids are missing in old PartitionSpec, they always auto-increment from + // PARTITION_DATA_ID_START + if (element.has(FIELD_ID)) { + builder.addField(transform, sourceId, JsonUtil.getInt(FIELD_ID, element), name); + fieldIdCount++; + } else { + builder.addField(transform, sourceId, name); + } } } diff --git a/core/src/main/java/org/apache/iceberg/SortOrderParser.java b/core/src/main/java/org/apache/iceberg/SortOrderParser.java index 31307cf9dc7f..0156a279e53c 100644 --- a/core/src/main/java/org/apache/iceberg/SortOrderParser.java +++ b/core/src/main/java/org/apache/iceberg/SortOrderParser.java @@ -25,8 +25,10 @@ import com.fasterxml.jackson.databind.JsonNode; import java.io.IOException; import java.util.Iterator; +import java.util.List; import java.util.Locale; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.util.JsonUtil; public class SortOrderParser { @@ -36,6 +38,7 @@ public class SortOrderParser { private static final String NULL_ORDER = "null-order"; private static final String TRANSFORM = "transform"; private static final String SOURCE_ID = "source-id"; + private static final String SOURCE_IDS = "source-ids"; private SortOrderParser() {} @@ -69,7 +72,7 @@ private static void toJsonFields(SortOrder sortOrder, JsonGenerator generator) for (SortField field : sortOrder.fields()) { generator.writeStartObject(); generator.writeStringField(TRANSFORM, field.transform().toString()); - generator.writeNumberField(SOURCE_ID, field.sourceId()); + writeSourceIds(field.sourceIds(), generator); generator.writeStringField(DIRECTION, toJson(field.direction())); generator.writeStringField(NULL_ORDER, toJson(field.nullOrder())); generator.writeEndObject(); @@ -100,7 +103,7 @@ private static void toJsonFields(UnboundSortOrder sortOrder, JsonGenerator gener for (UnboundSortOrder.UnboundSortField field : sortOrder.fields()) { generator.writeStartObject(); generator.writeStringField(TRANSFORM, field.transformAsString()); - generator.writeNumberField(SOURCE_ID, field.sourceId()); + writeSourceIds(field.sourceIds(), generator); generator.writeStringField(DIRECTION, toJson(field.direction())); generator.writeStringField(NULL_ORDER, toJson(field.nullOrder())); generator.writeEndObject(); @@ -108,6 +111,20 @@ private static void toJsonFields(UnboundSortOrder sortOrder, JsonGenerator gener generator.writeEndArray(); } + private static void writeSourceIds(List sourceIds, JsonGenerator generator) + throws IOException { + if (sourceIds.size() == 1) { + generator.writeNumberField(SOURCE_ID, sourceIds.get(0)); + } else { + generator.writeFieldName(SOURCE_IDS); + generator.writeStartArray(); + for (Integer value : sourceIds) { + generator.writeNumber(value); + } + generator.writeEndArray(); + } + } + public static SortOrder fromJson(Schema schema, String json) { return fromJson(json).bind(schema); } @@ -151,7 +168,6 @@ private static void buildFromJsonFields(UnboundSortOrder.Builder builder, JsonNo element.isObject(), "Cannot parse sort field, not an object: %s", element); String transform = JsonUtil.getString(TRANSFORM, element); - int sourceId = JsonUtil.getInt(SOURCE_ID, element); String directionAsString = JsonUtil.getString(DIRECTION, element); SortDirection direction = SortDirection.fromString(directionAsString); @@ -159,7 +175,21 @@ private static void buildFromJsonFields(UnboundSortOrder.Builder builder, JsonNo String nullOrderingAsString = JsonUtil.getString(NULL_ORDER, element); NullOrder nullOrder = toNullOrder(nullOrderingAsString); - builder.addSortField(transform, sourceId, direction, nullOrder); + if (element.has(SOURCE_IDS)) { + JsonNode sourceIdsNode = element.get(SOURCE_IDS); + Preconditions.checkArgument( + sourceIdsNode.isArray() && sourceIdsNode.size() > 0, + "Cannot parse sort field, source-ids must be a non-empty array: %s", + element); + List sourceIds = Lists.newArrayListWithCapacity(sourceIdsNode.size()); + for (int i = 0; i < sourceIdsNode.size(); i++) { + sourceIds.add(sourceIdsNode.get(i).asInt()); + } + builder.addSortField(transform, sourceIds, direction, nullOrder); + } else { + int sourceId = JsonUtil.getInt(SOURCE_ID, element); + builder.addSortField(transform, sourceId, direction, nullOrder); + } } } diff --git a/core/src/test/java/org/apache/iceberg/TestPartitionSpecParser.java b/core/src/test/java/org/apache/iceberg/TestPartitionSpecParser.java index e97d2f98b416..5c41f861cc47 100644 --- a/core/src/test/java/org/apache/iceberg/TestPartitionSpecParser.java +++ b/core/src/test/java/org/apache/iceberg/TestPartitionSpecParser.java @@ -19,9 +19,13 @@ package org.apache.iceberg; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.Arrays; +import java.util.Collections; import java.util.List; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.JsonUtil; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; @@ -69,6 +73,75 @@ public void testToJsonForV1Table() { assertThat(PartitionSpecParser.toJson(table.spec(), true)).isEqualTo(expected); } + @TestTemplate + public void testToJsonWithMultipleSourceIds() { + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField("bucket[8]", Arrays.asList(1, 2), 1001, "id_data") + .build(); + + String expected = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_data\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"field-id\" : 1001\n" + + " } ]\n" + + "}"; + assertThat(PartitionSpecParser.toJson(spec, true)).isEqualTo(expected); + } + + @TestTemplate + public void testToJsonWithMultipleFieldsAndMultipleSourceIds() { + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField("bucket[8]", Arrays.asList(1, 2), 1001, "id_data_bucket") + .addField("truncate[10]", Arrays.asList(3, 7, 10), 1002, "multi_truncate") + .build(); + + String expected = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_data_bucket\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"field-id\" : 1001\n" + + " }, {\n" + + " \"name\" : \"multi_truncate\",\n" + + " \"transform\" : \"truncate[10]\",\n" + + " \"source-ids\" : [ 3, 7, 10 ],\n" + + " \"field-id\" : 1002\n" + + " } ]\n" + + "}"; + assertThat(PartitionSpecParser.toJson(spec, true)).isEqualTo(expected); + } + + @TestTemplate + public void testToJsonWithSingleSourceId() { + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(0) + .addField("bucket[16]", Collections.singletonList(2), 1000, "data_bucket") + .build(); + + String expected = + "{\n" + + " \"spec-id\" : 0,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"data_bucket\",\n" + + " \"transform\" : \"bucket[16]\",\n" + + " \"source-id\" : 2,\n" + + " \"field-id\" : 1000\n" + + " } ]\n" + + "}"; + assertThat(PartitionSpecParser.toJson(spec, true)).isEqualTo(expected); + } + @TestTemplate public void testFromJsonWithFieldId() { String specString = @@ -119,6 +192,266 @@ public void testFromJsonWithoutFieldId() { assertThat(spec.fields().get(1).fieldId()).isEqualTo(1001); } + @TestTemplate + public void testFromJsonWithSourceIds() { + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_bucket\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-id\" : 1,\n" + + " \"source-ids\" : [ 1 ],\n" + + " \"field-id\" : 1001\n" + + " }, {\n" + + " \"name\" : \"data_bucket\",\n" + + " \"transform\" : \"bucket[16]\",\n" + + " \"source-id\" : 2,\n" + + " \"source-ids\" : [ 2 ],\n" + + " \"field-id\" : 1000\n" + + " } ]\n" + + "}"; + + PartitionSpec spec = PartitionSpecParser.fromJson(table.schema(), specString); + + assertThat(spec.fields()).hasSize(2); + assertThat(spec.fields().get(0).sourceId()).isEqualTo(1); + assertThat(spec.fields().get(0).fieldId()).isEqualTo(1001); + assertThat(spec.fields().get(1).sourceId()).isEqualTo(2); + assertThat(spec.fields().get(1).fieldId()).isEqualTo(1000); + } + + @TestTemplate + public void testFromJsonWithSourceIdsOnly() { + // source-ids without source-id for forward compatibility + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_bucket\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : [ 1 ],\n" + + " \"field-id\" : 1001\n" + + " } ]\n" + + "}"; + + PartitionSpec spec = PartitionSpecParser.fromJson(table.schema(), specString); + + assertThat(spec.fields()).hasSize(1); + assertThat(spec.fields().get(0).sourceId()).isEqualTo(1); + assertThat(spec.fields().get(0).fieldId()).isEqualTo(1001); + } + + @TestTemplate + public void testFromJsonWithMultipleSourceIds() { + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_data\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"field-id\" : 1001\n" + + " } ]\n" + + "}"; + + UnboundPartitionSpec unboundSpec = JsonUtil.parse(specString, PartitionSpecParser::fromJson); + + assertThat(unboundSpec.fields()).hasSize(1); + assertThat(unboundSpec.fields().get(0).sourceIds()).containsExactly(1, 2); + assertThat(unboundSpec.fields().get(0).sourceId()).isEqualTo(1); + assertThat(unboundSpec.fields().get(0).partitionId()).isEqualTo(1001); + + // round-trip should preserve source-ids + String json = PartitionSpecParser.toJson(unboundSpec, true); + assertThat(json).contains("\"source-ids\" : [ 1, 2 ]"); + assertThat(json).doesNotContain("\"source-id\""); + + UnboundPartitionSpec roundTripped = JsonUtil.parse(json, PartitionSpecParser::fromJson); + assertThat(roundTripped.fields().get(0).sourceIds()).containsExactly(1, 2); + } + + @TestTemplate + public void testFromJsonWithEmptySourceIds() { + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_bucket\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : [ ],\n" + + " \"field-id\" : 1001\n" + + " } ]\n" + + "}"; + + assertThatThrownBy(() -> PartitionSpecParser.fromJson(table.schema(), specString)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("source-ids must be a non-empty array"); + } + + @TestTemplate + public void testFromJsonWithSourceIdsNotAnArray() { + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_bucket\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : 1,\n" + + " \"field-id\" : 1001\n" + + " } ]\n" + + "}"; + + assertThatThrownBy(() -> PartitionSpecParser.fromJson(table.schema(), specString)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("source-ids must be a non-empty array"); + } + + @TestTemplate + public void testFromJsonWithMultipleSourceIdsWithoutFieldId() { + // source-ids without field-id should auto-increment field ids + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_data\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-ids\" : [ 1, 2 ]\n" + + " } ]\n" + + "}"; + + UnboundPartitionSpec unboundSpec = JsonUtil.parse(specString, PartitionSpecParser::fromJson); + + assertThat(unboundSpec.fields()).hasSize(1); + assertThat(unboundSpec.fields().get(0).sourceIds()).containsExactly(1, 2); + assertThat(unboundSpec.fields().get(0).partitionId()).isNull(); + } + + @TestTemplate + public void testFromJsonSourceIdsPrecedenceOverSourceId() { + // when both source-id and source-ids are present, source-ids takes precedence + String specString = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_data\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-id\" : 99,\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"field-id\" : 1001\n" + + " } ]\n" + + "}"; + + UnboundPartitionSpec unboundSpec = JsonUtil.parse(specString, PartitionSpecParser::fromJson); + + assertThat(unboundSpec.fields()).hasSize(1); + // source-ids should be used, not source-id + assertThat(unboundSpec.fields().get(0).sourceIds()).containsExactly(1, 2); + assertThat(unboundSpec.fields().get(0).sourceId()).isEqualTo(1); + } + + @TestTemplate + public void testToJsonCompactWithMultipleSourceIds() { + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField("bucket[8]", Arrays.asList(1, 2), 1001, "id_data") + .build(); + + String json = PartitionSpecParser.toJson(spec, false); + assertThat(json).contains("\"source-ids\":[1,2]"); + assertThat(json).doesNotContain("\"source-id\""); + } + + @TestTemplate + public void testToJsonMixedSingleAndMultipleSourceIds() { + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField("bucket[8]", Collections.singletonList(1), 1001, "id_bucket") + .addField("bucket[16]", Arrays.asList(1, 2), 1002, "id_data_bucket") + .build(); + + String expected = + "{\n" + + " \"spec-id\" : 1,\n" + + " \"fields\" : [ {\n" + + " \"name\" : \"id_bucket\",\n" + + " \"transform\" : \"bucket[8]\",\n" + + " \"source-id\" : 1,\n" + + " \"field-id\" : 1001\n" + + " }, {\n" + + " \"name\" : \"id_data_bucket\",\n" + + " \"transform\" : \"bucket[16]\",\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"field-id\" : 1002\n" + + " } ]\n" + + "}"; + assertThat(PartitionSpecParser.toJson(spec, true)).isEqualTo(expected); + } + + @TestTemplate + public void testRoundTripSingleSourceIdUsesSingularKey() { + // a single-element source-ids list should serialize as source-id (not source-ids) + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField("bucket[8]", Collections.singletonList(1), 1001, "id_bucket") + .build(); + + String json = PartitionSpecParser.toJson(spec, true); + assertThat(json).contains("\"source-id\" : 1"); + assertThat(json).doesNotContain("\"source-ids\""); + + UnboundPartitionSpec parsed = JsonUtil.parse(json, PartitionSpecParser::fromJson); + assertThat(parsed.fields().get(0).sourceIds()).containsExactly(1); + } + + @TestTemplate + public void testFromJsonFieldsWithMultipleSourceIds() { + // exercise the fromJsonFields path used by table metadata parsing + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "data", Types.StringType.get())); + + String fieldsJson = + "[{\"name\":\"id_data\",\"transform\":\"bucket[8]\"," + + "\"source-ids\":[1,2],\"field-id\":1001}]"; + + PartitionSpec spec = + JsonUtil.parse(fieldsJson, node -> PartitionSpecParser.fromJsonFields(schema, 1, node)); + assertThat(spec.specId()).isEqualTo(1); + assertThat(spec.fields()).hasSize(1); + assertThat(spec.fields().get(0).sourceIds()).containsExactly(1, 2); + assertThat(spec.fields().get(0).fieldId()).isEqualTo(1001); + } + + @TestTemplate + public void testBoundSpecRoundTripPreservesSourceIds() { + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "data", Types.StringType.get())); + + // build an unbound spec with multi-arg source-ids, bind it, then round-trip through JSON + UnboundPartitionSpec unboundSpec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField("bucket[8]", Arrays.asList(1, 2), 1001, "id_data_bucket") + .build(); + + PartitionSpec boundSpec = unboundSpec.bind(schema); + assertThat(boundSpec.fields().get(0).sourceIds()).containsExactly(1, 2); + + // round-trip through JSON via toUnbound + String json = PartitionSpecParser.toJson(boundSpec, true); + assertThat(json).contains("\"source-ids\" : [ 1, 2 ]"); + + UnboundPartitionSpec parsed = JsonUtil.parse(json, PartitionSpecParser::fromJson); + assertThat(parsed.fields().get(0).sourceIds()).containsExactly(1, 2); + } + @TestTemplate public void testTransforms() { for (PartitionSpec spec : PartitionSpecTestBase.SPECS) { @@ -126,6 +459,34 @@ public void testTransforms() { } } + @TestTemplate + public void testTransformRoundTripWithMultipleSourceIds() { + // test various transform types with multiple source-ids + String[] transforms = {"bucket[16]", "identity", "truncate[10]"}; + + for (String transform : transforms) { + UnboundPartitionSpec spec = + UnboundPartitionSpec.builder() + .withSpecId(1) + .addField(transform, Arrays.asList(1, 2, 7), 1001, "multi_field") + .build(); + + String json = PartitionSpecParser.toJson(spec, true); + assertThat(json).contains("\"source-ids\" : [ 1, 2, 7 ]"); + assertThat(json).doesNotContain("\"source-id\""); + assertThat(json).contains("\"transform\" : \"" + transform + "\""); + + UnboundPartitionSpec parsed = JsonUtil.parse(json, PartitionSpecParser::fromJson); + assertThat(parsed.specId()).isEqualTo(1); + assertThat(parsed.fields()).hasSize(1); + assertThat(parsed.fields().get(0).sourceIds()).containsExactly(1, 2, 7); + assertThat(parsed.fields().get(0).sourceId()).isEqualTo(1); + assertThat(parsed.fields().get(0).transformAsString()).isEqualTo(transform); + assertThat(parsed.fields().get(0).partitionId()).isEqualTo(1001); + assertThat(parsed.fields().get(0).name()).isEqualTo("multi_field"); + } + } + private static PartitionSpec roundTripJSON(PartitionSpec spec) { return PartitionSpecParser.fromJson( PartitionSpecTestBase.SCHEMA, PartitionSpecParser.toJson(spec)); diff --git a/core/src/test/java/org/apache/iceberg/TestSortOrderParser.java b/core/src/test/java/org/apache/iceberg/TestSortOrderParser.java index adcc130c2390..783e63e9f948 100644 --- a/core/src/test/java/org/apache/iceberg/TestSortOrderParser.java +++ b/core/src/test/java/org/apache/iceberg/TestSortOrderParser.java @@ -19,6 +19,8 @@ package org.apache.iceberg; import static org.apache.iceberg.NullOrder.NULLS_FIRST; +import static org.apache.iceberg.NullOrder.NULLS_LAST; +import static org.apache.iceberg.SortDirection.ASC; import static org.apache.iceberg.SortDirection.DESC; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -62,6 +64,113 @@ public void testUnknownTransforms() { assertThat(order.fields().get(0).nullOrder()).isEqualTo(NULLS_FIRST); } + @TestTemplate + public void testSourceIdsDeserialization() { + String jsonString = + "{\n" + + " \"order-id\" : 10,\n" + + " \"fields\" : [ {\n" + + " \"transform\" : \"bucket[16]\",\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"direction\" : \"asc\",\n" + + " \"null-order\" : \"nulls-last\"\n" + + " } ]\n" + + "}"; + + UnboundSortOrder order = SortOrderParser.fromJson(jsonString); + + assertThat(order.orderId()).isEqualTo(10); + assertThat(order.fields()).hasSize(1); + assertThat(order.fields().get(0).sourceIds()).containsExactly(1, 2); + assertThat(order.fields().get(0).sourceId()).isEqualTo(1); + assertThat(order.fields().get(0).direction()).isEqualTo(ASC); + assertThat(order.fields().get(0).nullOrder()).isEqualTo(NULLS_LAST); + } + + @TestTemplate + public void testSourceIdsSerialization() { + UnboundSortOrder order = + UnboundSortOrder.builder() + .withOrderId(10) + .addSortField("bucket[16]", Arrays.asList(1, 2), ASC, NULLS_LAST) + .build(); + + String json = SortOrderParser.toJson(order, true); + assertThat(json).contains("\"source-ids\" : [ 1, 2 ]"); + assertThat(json).doesNotContain("\"source-id\""); + } + + @TestTemplate + public void testSingleSourceIdSerialization() { + UnboundSortOrder order = + UnboundSortOrder.builder() + .withOrderId(10) + .addSortField("bucket[16]", 1, ASC, NULLS_LAST) + .build(); + + String json = SortOrderParser.toJson(order, true); + assertThat(json).contains("\"source-id\" : 1"); + assertThat(json).doesNotContain("\"source-ids\""); + } + + @TestTemplate + public void testSourceIdsRoundTrip() { + UnboundSortOrder order = + UnboundSortOrder.builder() + .withOrderId(10) + .addSortField("bucket[16]", Arrays.asList(1, 2), ASC, NULLS_LAST) + .addSortField("identity", 3, DESC, NULLS_FIRST) + .build(); + + String json = SortOrderParser.toJson(order, true); + UnboundSortOrder parsed = SortOrderParser.fromJson(json); + + assertThat(parsed.orderId()).isEqualTo(10); + assertThat(parsed.fields()).hasSize(2); + assertThat(parsed.fields().get(0).sourceIds()).containsExactly(1, 2); + assertThat(parsed.fields().get(0).direction()).isEqualTo(ASC); + assertThat(parsed.fields().get(1).sourceIds()).containsExactly(3); + assertThat(parsed.fields().get(1).direction()).isEqualTo(DESC); + } + + @TestTemplate + public void testSourceIdsPrecedenceOverSourceId() { + String jsonString = + "{\n" + + " \"order-id\" : 10,\n" + + " \"fields\" : [ {\n" + + " \"transform\" : \"bucket[16]\",\n" + + " \"source-id\" : 99,\n" + + " \"source-ids\" : [ 1, 2 ],\n" + + " \"direction\" : \"asc\",\n" + + " \"null-order\" : \"nulls-last\"\n" + + " } ]\n" + + "}"; + + UnboundSortOrder order = SortOrderParser.fromJson(jsonString); + + assertThat(order.fields()).hasSize(1); + assertThat(order.fields().get(0).sourceIds()).containsExactly(1, 2); + } + + @TestTemplate + public void testEmptySourceIdsArray() { + String jsonString = + "{\n" + + " \"order-id\" : 10,\n" + + " \"fields\" : [ {\n" + + " \"transform\" : \"bucket[16]\",\n" + + " \"source-ids\" : [ ],\n" + + " \"direction\" : \"asc\",\n" + + " \"null-order\" : \"nulls-last\"\n" + + " } ]\n" + + "}"; + + assertThatThrownBy(() -> SortOrderParser.fromJson(jsonString)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("source-ids must be a non-empty array"); + } + @TestTemplate public void invalidSortDirection() { String jsonString =