You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2022/06/27 23:44:00 UTC
[iceberg] branch master updated: Core: Fix partition clustering to produce table sort order (#5131)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 4ce3900e3 Core: Fix partition clustering to produce table sort order (#5131)
4ce3900e3 is described below
commit 4ce3900e34a1513a53db95725938053b2b90851e
Author: Ryan Blue <bl...@apache.org>
AuthorDate: Mon Jun 27 16:43:54 2022 -0700
Core: Fix partition clustering to produce table sort order (#5131)
---
.../org/apache/iceberg/util/SortOrderUtil.java | 75 ++++--
.../org/apache/iceberg/util/TestSortOrderUtil.java | 257 +++++++++++++++++++--
2 files changed, 302 insertions(+), 30 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java b/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
index 118e9bcd2..74b8e2c36 100644
--- a/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
+++ b/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
@@ -19,8 +19,8 @@
package org.apache.iceberg.util;
-import java.util.Collection;
import java.util.Collections;
+import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
@@ -31,9 +31,9 @@ import org.apache.iceberg.SortField;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
-import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.transforms.SortOrderVisitor;
+import org.apache.iceberg.transforms.Transform;
public class SortOrderUtil {
@@ -49,23 +49,50 @@ public class SortOrderUtil {
return buildSortOrder(table.schema(), table.spec(), sortOrder);
}
+ /**
+ * Build a final sort order that satisfies the clustering required by the partition spec.
+ * <p>
+ * The incoming sort order may or may not satisfy the clustering needed by the partition spec. This modifies the sort
+ * order so that it clusters by partition and still produces the same order within each partition.
+ *
+ * @param schema a schema
+ * @param spec a partition spec
+ * @param sortOrder a sort order
+ * @return the sort order with additional sort fields to satisfy the clustering required by the spec
+ */
public static SortOrder buildSortOrder(Schema schema, PartitionSpec spec, SortOrder sortOrder) {
if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
return SortOrder.unsorted();
}
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
+ // make a map of the partition fields that need to be included in the clustering produced by the sort order
+ Map<Pair<Transform<?, ?>, Integer>, PartitionField> requiredClusteringFields = requiredClusteringFields(spec);
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(Expressions.transform(sourceName, field.transform()));
+ // remove any partition fields that are clustered by the sort order by iterating over a prefix in the sort order.
+ // this will stop when a non-partition field is found, or when the sort field only satisfies the partition field.
+ for (SortField sortField : sortOrder.fields()) {
+ Pair<Transform<?, ?>, Integer> sourceAndTransform = Pair.of(sortField.transform(), sortField.sourceId());
+ if (requiredClusteringFields.containsKey(sourceAndTransform)) {
+ requiredClusteringFields.remove(sourceAndTransform);
+ continue; // keep processing the prefix
+ }
+
+ // if the field satisfies the order of any partition fields, also remove them before stopping
+ // use a set to avoid concurrent modification
+ for (PartitionField field : spec.fields()) {
+ if (sortField.sourceId() == field.sourceId() && sortField.transform().satisfiesOrderOf(field.transform())) {
+ requiredClusteringFields.remove(Pair.of(field.transform(), field.sourceId()));
+ }
}
+
+ break;
+ }
+
+ // build a sort prefix of partition fields that are not already in the sort order's prefix
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
+ for (PartitionField field : requiredClusteringFields.values()) {
+ String sourceName = schema.findColumnName(field.sourceId());
+ builder.asc(Expressions.transform(sourceName, field.transform()));
}
// add the configured sort to the partition spec prefix sort
@@ -74,6 +101,28 @@ public class SortOrderUtil {
return builder.build();
}
+ private static Map<Pair<Transform<?, ?>, Integer>, PartitionField> requiredClusteringFields(PartitionSpec spec) {
+ Map<Pair<Transform<?, ?>, Integer>, PartitionField> requiredClusteringFields = Maps.newLinkedHashMap();
+ for (PartitionField partField : spec.fields()) {
+ if (!partField.transform().toString().equals("void")) {
+ requiredClusteringFields.put(Pair.of(partField.transform(), partField.sourceId()), partField);
+ }
+ }
+
+ // remove any partition fields that are satisfied by another partition field, like days(ts) and hours(ts)
+ for (PartitionField partField : spec.fields()) {
+ for (PartitionField field : spec.fields()) {
+ if (!partField.equals(field) &&
+ partField.sourceId() == field.sourceId() &&
+ partField.transform().satisfiesOrderOf(field.transform())) {
+ requiredClusteringFields.remove(Pair.of(field.transform(), field.sourceId()));
+ }
+ }
+ }
+
+ return requiredClusteringFields;
+ }
+
public static Set<String> orderPreservingSortedColumns(SortOrder sortOrder) {
if (sortOrder == null) {
return Collections.emptySet();
diff --git a/core/src/test/java/org/apache/iceberg/util/TestSortOrderUtil.java b/core/src/test/java/org/apache/iceberg/util/TestSortOrderUtil.java
index 08471b50d..090b5b8aa 100644
--- a/core/src/test/java/org/apache/iceberg/util/TestSortOrderUtil.java
+++ b/core/src/test/java/org/apache/iceberg/util/TestSortOrderUtil.java
@@ -25,6 +25,7 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.TestTables;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.types.Types;
import org.junit.After;
import org.junit.Assert;
@@ -32,37 +33,25 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
import static org.apache.iceberg.NullOrder.NULLS_LAST;
import static org.apache.iceberg.SortDirection.ASC;
import static org.apache.iceberg.types.Types.NestedField.required;
-@RunWith(Parameterized.class)
public class TestSortOrderUtil {
// column ids will be reassigned during table creation
private static final Schema SCHEMA = new Schema(
required(10, "id", Types.IntegerType.get()),
- required(11, "data", Types.StringType.get())
+ required(11, "data", Types.StringType.get()),
+ required(12, "ts", Types.TimestampType.withZone()),
+ required(13, "category", Types.StringType.get())
);
@Rule
public TemporaryFolder temp = new TemporaryFolder();
private File tableDir = null;
- @Parameterized.Parameters(name = "formatVersion = {0}")
- public static Object[] parameters() {
- return new Object[] { 1, 2 };
- }
-
- private final int formatVersion;
-
- public TestSortOrderUtil(int formatVersion) {
- this.formatVersion = formatVersion;
- }
-
@Before
public void setupTableDir() throws IOException {
this.tableDir = temp.newFolder();
@@ -74,13 +63,32 @@ public class TestSortOrderUtil {
}
@Test
- public void testEmptySpecs() {
+ public void testEmptySpecsV1() {
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("id", NULLS_LAST)
+ .build();
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, order, 1);
+
+ // pass PartitionSpec.unpartitioned() on purpose as it has an empty schema
+ SortOrder actualOrder = SortOrderUtil.buildSortOrder(table.schema(), spec, table.sortOrder());
+
+ Assert.assertEquals("Order ID must be fresh", 1, actualOrder.orderId());
+ Assert.assertEquals("Order must have 1 field", 1, actualOrder.fields().size());
+ Assert.assertEquals("Field id must be fresh", 1, actualOrder.fields().get(0).sourceId());
+ Assert.assertEquals("Direction must match", ASC, actualOrder.fields().get(0).direction());
+ Assert.assertEquals("Null order must match", NULLS_LAST, actualOrder.fields().get(0).nullOrder());
+ }
+
+ @Test
+ public void testEmptySpecsV2() {
PartitionSpec spec = PartitionSpec.unpartitioned();
SortOrder order = SortOrder.builderFor(SCHEMA)
.withOrderId(1)
.asc("id", NULLS_LAST)
.build();
- TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, order, formatVersion);
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, order, 2);
// pass PartitionSpec.unpartitioned() on purpose as it has an empty schema
SortOrder actualOrder = SortOrderUtil.buildSortOrder(table.schema(), spec, table.sortOrder());
@@ -91,4 +99,219 @@ public class TestSortOrderUtil {
Assert.assertEquals("Direction must match", ASC, actualOrder.fields().get(0).direction());
Assert.assertEquals("Null order must match", NULLS_LAST, actualOrder.fields().get(0).nullOrder());
}
+
+ @Test
+ public void testSortOrderClusteringNoPartitionFields() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .day("ts")
+ .identity("category")
+ .build();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc(Expressions.day("ts"))
+ .asc("category")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringAllPartitionFields() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .day("ts")
+ .identity("category")
+ .build();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc(Expressions.day("ts"))
+ .asc("category")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should leave the order unchanged",
+ order, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringAllPartitionFieldsReordered() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("category")
+ .day("ts")
+ .build();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc(Expressions.day("ts"))
+ .asc("category")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should leave the order unchanged",
+ order, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringSomePartitionFields() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("category")
+ .day("ts")
+ .build();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("category")
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc(Expressions.day("ts"))
+ .asc("category")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringSatisfiedPartitionLast() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("category")
+ .day("ts")
+ .build();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("category")
+ .asc("ts") // satisfies the ordering of days(ts)
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("category")
+ .asc("ts")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringSatisfiedPartitionFirst() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .day("ts")
+ .identity("category")
+ .build();
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("ts") // satisfies the ordering of days(ts)
+ .asc("category")
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("category") // prefix is added, the rest of the sort order stays the same
+ .asc("ts")
+ .asc("category")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringSatisfiedPartitionFields() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .day("ts")
+ .identity("category")
+ .build();
+
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("ts") // satisfies the ordering of days(ts)
+ .asc("category")
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(SCHEMA)
+ .withOrderId(1)
+ .asc("category") // prefix is added, the rest of the sort order stays the same
+ .asc("ts")
+ .asc("category")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(SCHEMA, spec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringWithRedundantPartitionFields() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .day("ts")
+ .identity("category")
+ .build();
+
+ // Specs with redundant time fields can't be constructed directly and have to use UpdatePartitionSpec
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, SortOrder.unsorted(), 2);
+ table.updateSpec().addField(Expressions.hour("ts")).commit();
+ PartitionSpec updatedSpec = table.spec();
+
+ SortOrder order = SortOrder.builderFor(table.schema())
+ .withOrderId(1)
+ .asc("category")
+ .asc("ts") // satisfies the ordering of days(ts) and hours(ts)
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(table.schema())
+ .withOrderId(1)
+ .asc("category")
+ .asc("ts")
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(table.schema(), updatedSpec, order));
+ }
+
+ @Test
+ public void testSortOrderClusteringWithRedundantPartitionFieldsMissing() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .day("ts")
+ .identity("category")
+ .build();
+
+ // Specs with redundant time fields can't be constructed directly and have to use UpdatePartitionSpec
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, SortOrder.unsorted(), 1);
+ table.updateSpec()
+ .removeField("ts_day") // introduce a void transform
+ .addField(Expressions.hour("ts"))
+ .commit();
+ PartitionSpec updatedSpec = table.spec();
+
+ SortOrder order = SortOrder.builderFor(table.schema())
+ .withOrderId(1)
+ .desc("id")
+ .build();
+
+ SortOrder expected = SortOrder.builderFor(table.schema())
+ .withOrderId(1)
+ .asc("category")
+ .asc(Expressions.hour("ts"))
+ .desc("id")
+ .build();
+
+ Assert.assertEquals("Should add spec fields as prefix",
+ expected, SortOrderUtil.buildSortOrder(table.schema(), updatedSpec, order));
+ }
}