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 2020/08/12 18:12:20 UTC
[iceberg] branch master updated: Core: Add equality field IDs to
DeleteFile (#1318)
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 3e8b750 Core: Add equality field IDs to DeleteFile (#1318)
3e8b750 is described below
commit 3e8b75081f79dee420872f05aac83e6e90f7e3d8
Author: Ryan Blue <bl...@apache.org>
AuthorDate: Wed Aug 12 11:12:09 2020 -0700
Core: Add equality field IDs to DeleteFile (#1318)
---
.../main/java/org/apache/iceberg/ContentFile.java | 12 +++++
api/src/main/java/org/apache/iceberg/DataFile.java | 13 ++++-
.../src/main/java/org/apache/iceberg/BaseFile.java | 32 +++++++----
.../main/java/org/apache/iceberg/FileMetadata.java | 7 ++-
.../java/org/apache/iceberg/GenericDataFile.java | 2 +-
.../java/org/apache/iceberg/GenericDeleteFile.java | 4 +-
.../main/java/org/apache/iceberg/V2Metadata.java | 10 +++-
.../java/org/apache/iceberg/util/ArrayUtil.java | 62 ++++++++++++++++++++++
.../apache/iceberg/TestManifestWriterVersions.java | 10 +++-
.../java/org/apache/iceberg/TestMergeAppend.java | 4 +-
10 files changed, 136 insertions(+), 20 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/ContentFile.java b/api/src/main/java/org/apache/iceberg/ContentFile.java
index 11d4a4d..625efff 100644
--- a/api/src/main/java/org/apache/iceberg/ContentFile.java
+++ b/api/src/main/java/org/apache/iceberg/ContentFile.java
@@ -102,6 +102,18 @@ public interface ContentFile<F> {
*/
List<Long> splitOffsets();
+ /**
+ * Returns the set of field IDs used for equality comparison, in equality delete files.
+ * <p>
+ * An equality delete file may contain additional data fields that are not used by equality
+ * comparison. The subset of columns in a delete file to be used in equality comparison are
+ * tracked by ID. Extra columns can be used to reconstruct changes and metrics from extra
+ * columns are used during job planning.
+ *
+ * @return IDs of the fields used in equality comparison with the records in this delete file
+ */
+ List<Integer> equalityFieldIds();
+
/**
* Copies this file. Manifest readers can reuse file instances; use
diff --git a/api/src/main/java/org/apache/iceberg/DataFile.java b/api/src/main/java/org/apache/iceberg/DataFile.java
index 747057b..1b80926 100644
--- a/api/src/main/java/org/apache/iceberg/DataFile.java
+++ b/api/src/main/java/org/apache/iceberg/DataFile.java
@@ -19,6 +19,7 @@
package org.apache.iceberg;
+import java.util.List;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.BinaryType;
import org.apache.iceberg.types.Types.IntegerType;
@@ -56,10 +57,12 @@ public interface DataFile extends ContentFile<DataFile> {
Types.NestedField KEY_METADATA = optional(131, "key_metadata", BinaryType.get(), "Encryption key metadata blob");
Types.NestedField SPLIT_OFFSETS = optional(132, "split_offsets", ListType.ofRequired(133, LongType.get()),
"Splittable offsets");
+ Types.NestedField EQUALITY_IDS = optional(135, "equality_ids", ListType.ofRequired(136, IntegerType.get()),
+ "Equality comparison field IDs");
int PARTITION_ID = 102;
String PARTITION_NAME = "partition";
String PARTITION_DOC = "Partition data tuple, schema based on the partition spec";
- // NEXT ID TO ASSIGN: 135
+ // NEXT ID TO ASSIGN: 137
static StructType getType(StructType partitionType) {
// IDs start at 100 to leave room for changes to ManifestEntry
@@ -76,7 +79,8 @@ public interface DataFile extends ContentFile<DataFile> {
LOWER_BOUNDS,
UPPER_BOUNDS,
KEY_METADATA,
- SPLIT_OFFSETS
+ SPLIT_OFFSETS,
+ EQUALITY_IDS
);
}
@@ -87,4 +91,9 @@ public interface DataFile extends ContentFile<DataFile> {
default FileContent content() {
return FileContent.DATA;
}
+
+ @Override
+ default List<Integer> equalityFieldIds() {
+ return null;
+ }
}
diff --git a/core/src/main/java/org/apache/iceberg/BaseFile.java b/core/src/main/java/org/apache/iceberg/BaseFile.java
index db8320e..24860bd 100644
--- a/core/src/main/java/org/apache/iceberg/BaseFile.java
+++ b/core/src/main/java/org/apache/iceberg/BaseFile.java
@@ -31,10 +31,10 @@ import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.specific.SpecificData;
import org.apache.iceberg.avro.AvroSchemaUtil;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ArrayUtil;
import org.apache.iceberg.util.ByteBuffers;
/**
@@ -67,7 +67,8 @@ abstract class BaseFile<F>
private Map<Integer, Long> nullValueCounts = null;
private Map<Integer, ByteBuffer> lowerBounds = null;
private Map<Integer, ByteBuffer> upperBounds = null;
- private Long[] splitOffsets = null;
+ private long[] splitOffsets = null;
+ private int[] equalityIds = null;
private byte[] keyMetadata = null;
// cached schema
@@ -113,7 +114,7 @@ abstract class BaseFile<F>
PartitionData partition, long fileSizeInBytes, long recordCount,
Map<Integer, Long> columnSizes, Map<Integer, Long> valueCounts, Map<Integer, Long> nullValueCounts,
Map<Integer, ByteBuffer> lowerBounds, Map<Integer, ByteBuffer> upperBounds, List<Long> splitOffsets,
- ByteBuffer keyMetadata) {
+ int[] equalityFieldIds, ByteBuffer keyMetadata) {
this.partitionSpecId = specId;
this.content = content;
this.filePath = filePath;
@@ -136,7 +137,8 @@ abstract class BaseFile<F>
this.nullValueCounts = nullValueCounts;
this.lowerBounds = SerializableByteBufferMap.wrap(lowerBounds);
this.upperBounds = SerializableByteBufferMap.wrap(upperBounds);
- this.splitOffsets = splitOffsets == null ? null : splitOffsets.toArray(new Long[0]);
+ this.splitOffsets = ArrayUtil.toLongArray(splitOffsets);
+ this.equalityIds = equalityFieldIds;
this.keyMetadata = ByteBuffers.toByteArray(keyMetadata);
}
@@ -247,7 +249,10 @@ abstract class BaseFile<F>
this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer) value);
return;
case 12:
- this.splitOffsets = value != null ? ((List<Long>) value).toArray(new Long[0]) : null;
+ this.splitOffsets = ArrayUtil.toLongArray((List<Long>) value);
+ return;
+ case 13:
+ this.equalityIds = ArrayUtil.toIntArray((List<Integer>) value);
return;
default:
// ignore the object, it must be from a newer version of the format
@@ -290,9 +295,11 @@ abstract class BaseFile<F>
case 10:
return upperBounds;
case 11:
- return keyMetadata != null ? ByteBuffer.wrap(keyMetadata) : null;
+ return keyMetadata();
case 12:
- return splitOffsets != null ? Lists.newArrayList(splitOffsets) : null;
+ return splitOffsets();
+ case 13:
+ return equalityFieldIds();
default:
throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
}
@@ -370,7 +377,12 @@ abstract class BaseFile<F>
@Override
public List<Long> splitOffsets() {
- return splitOffsets != null ? Lists.newArrayList(splitOffsets) : null;
+ return ArrayUtil.toLongList(splitOffsets);
+ }
+
+ @Override
+ public List<Integer> equalityFieldIds() {
+ return ArrayUtil.toIntList(equalityIds);
}
private static <K, V> Map<K, V> copy(Map<K, V> map) {
@@ -397,7 +409,9 @@ abstract class BaseFile<F>
.add("lower_bounds", lowerBounds)
.add("upper_bounds", upperBounds)
.add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
- .add("split_offsets", splitOffsets == null ? "null" : splitOffsets)
+ .add("split_offsets", splitOffsets == null ? "null" : splitOffsets())
+ .add("equality_ids", equalityIds == null ? "null" : equalityFieldIds())
.toString();
}
+
}
diff --git a/core/src/main/java/org/apache/iceberg/FileMetadata.java b/core/src/main/java/org/apache/iceberg/FileMetadata.java
index 7ff31e0..688cd8f 100644
--- a/core/src/main/java/org/apache/iceberg/FileMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/FileMetadata.java
@@ -43,6 +43,7 @@ class FileMetadata {
private final boolean isPartitioned;
private final int specId;
private FileContent content = null;
+ private int[] equalityFieldIds = null;
private PartitionData partitionData;
private String filePath = null;
private FileFormat format = null;
@@ -101,11 +102,13 @@ class FileMetadata {
public Builder ofPositionDeletes() {
this.content = FileContent.POSITION_DELETES;
+ this.equalityFieldIds = null;
return this;
}
- public Builder ofEqualityDeletes() {
+ public Builder ofEqualityDeletes(int... fieldIds) {
this.content = FileContent.EQUALITY_DELETES;
+ this.equalityFieldIds = fieldIds;
return this;
}
@@ -204,7 +207,7 @@ class FileMetadata {
specId, content, filePath, format, isPartitioned ? DataFiles.copy(spec, partitionData) : null,
fileSizeInBytes, new Metrics(
recordCount, columnSizes, valueCounts, nullValueCounts, lowerBounds, upperBounds),
- keyMetadata);
+ equalityFieldIds, keyMetadata);
}
}
}
diff --git a/core/src/main/java/org/apache/iceberg/GenericDataFile.java b/core/src/main/java/org/apache/iceberg/GenericDataFile.java
index 1075c63..92f662e 100644
--- a/core/src/main/java/org/apache/iceberg/GenericDataFile.java
+++ b/core/src/main/java/org/apache/iceberg/GenericDataFile.java
@@ -39,7 +39,7 @@ class GenericDataFile extends BaseFile<DataFile> implements DataFile {
ByteBuffer keyMetadata, List<Long> splitOffsets) {
super(specId, FileContent.DATA, filePath, format, partition, fileSizeInBytes, metrics.recordCount(),
metrics.columnSizes(), metrics.valueCounts(), metrics.nullValueCounts(),
- metrics.lowerBounds(), metrics.upperBounds(), splitOffsets, keyMetadata);
+ metrics.lowerBounds(), metrics.upperBounds(), splitOffsets, null, keyMetadata);
}
/**
diff --git a/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java b/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
index 0ad76a9..73a8df7 100644
--- a/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
+++ b/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
@@ -35,10 +35,10 @@ class GenericDeleteFile extends BaseFile<DeleteFile> implements DeleteFile {
}
GenericDeleteFile(int specId, FileContent content, String filePath, FileFormat format, PartitionData partition,
- long fileSizeInBytes, Metrics metrics, ByteBuffer keyMetadata) {
+ long fileSizeInBytes, Metrics metrics, int[] equalityFieldIds, ByteBuffer keyMetadata) {
super(specId, content, filePath, format, partition, fileSizeInBytes, metrics.recordCount(),
metrics.columnSizes(), metrics.valueCounts(), metrics.nullValueCounts(),
- metrics.lowerBounds(), metrics.upperBounds(), null, keyMetadata);
+ metrics.lowerBounds(), metrics.upperBounds(), null, equalityFieldIds, keyMetadata);
}
/**
diff --git a/core/src/main/java/org/apache/iceberg/V2Metadata.java b/core/src/main/java/org/apache/iceberg/V2Metadata.java
index c499950..94ff1da 100644
--- a/core/src/main/java/org/apache/iceberg/V2Metadata.java
+++ b/core/src/main/java/org/apache/iceberg/V2Metadata.java
@@ -253,7 +253,8 @@ class V2Metadata {
DataFile.LOWER_BOUNDS,
DataFile.UPPER_BOUNDS,
DataFile.KEY_METADATA,
- DataFile.SPLIT_OFFSETS
+ DataFile.SPLIT_OFFSETS,
+ DataFile.EQUALITY_IDS
);
}
@@ -403,6 +404,8 @@ class V2Metadata {
return wrapped.keyMetadata();
case 12:
return wrapped.splitOffsets();
+ case 13:
+ return wrapped.equalityFieldIds();
}
throw new IllegalArgumentException("Unknown field ordinal: " + pos);
}
@@ -483,6 +486,11 @@ class V2Metadata {
}
@Override
+ public List<Integer> equalityFieldIds() {
+ return wrapped.equalityFieldIds();
+ }
+
+ @Override
public F copy() {
throw new UnsupportedOperationException("Cannot copy IndexedDataFile wrapper");
}
diff --git a/core/src/main/java/org/apache/iceberg/util/ArrayUtil.java b/core/src/main/java/org/apache/iceberg/util/ArrayUtil.java
new file mode 100644
index 0000000..685f193
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/util/ArrayUtil.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.util;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.LongStream;
+
+public class ArrayUtil {
+ private ArrayUtil() {
+ }
+
+ public static List<Integer> toIntList(int[] ints) {
+ if (ints != null) {
+ return IntStream.of(ints).boxed().collect(Collectors.toList());
+ } else {
+ return null;
+ }
+ }
+
+ public static int[] toIntArray(List<Integer> ints) {
+ if (ints != null) {
+ return ints.stream().mapToInt(v -> v).toArray();
+ } else {
+ return null;
+ }
+ }
+
+ public static List<Long> toLongList(long[] longs) {
+ if (longs != null) {
+ return LongStream.of(longs).boxed().collect(Collectors.toList());
+ } else {
+ return null;
+ }
+ }
+
+ public static long[] toLongArray(List<Long> longs) {
+ if (longs != null) {
+ return longs.stream().mapToLong(v -> v).toArray();
+ } else {
+ return null;
+ }
+ }
+}
diff --git a/core/src/test/java/org/apache/iceberg/TestManifestWriterVersions.java b/core/src/test/java/org/apache/iceberg/TestManifestWriterVersions.java
index 1621539..c8ede07 100644
--- a/core/src/test/java/org/apache/iceberg/TestManifestWriterVersions.java
+++ b/core/src/test/java/org/apache/iceberg/TestManifestWriterVersions.java
@@ -70,8 +70,11 @@ public class TestManifestWriterVersions {
private static final DataFile DATA_FILE = new GenericDataFile(
0, PATH, FORMAT, PARTITION, 150972L, METRICS, null, OFFSETS);
+ private static final List<Integer> EQUALITY_IDS = ImmutableList.of(1);
+ private static final int[] EQUALITY_ID_ARR = new int[] { 1 };
+
private static final DeleteFile DELETE_FILE = new GenericDeleteFile(
- 0, FileContent.EQUALITY_DELETES, PATH, FORMAT, PARTITION, 22905L, METRICS, null);
+ 0, FileContent.EQUALITY_DELETES, PATH, FORMAT, PARTITION, 22905L, METRICS, EQUALITY_ID_ARR, null);
@Rule
public TemporaryFolder temp = new TemporaryFolder();
@@ -195,6 +198,11 @@ public class TestManifestWriterVersions {
Assert.assertEquals("Null value counts", METRICS.nullValueCounts(), dataFile.nullValueCounts());
Assert.assertEquals("Lower bounds", METRICS.lowerBounds(), dataFile.lowerBounds());
Assert.assertEquals("Upper bounds", METRICS.upperBounds(), dataFile.upperBounds());
+ if (dataFile.content() == FileContent.EQUALITY_DELETES) {
+ Assert.assertEquals(EQUALITY_IDS, dataFile.equalityFieldIds());
+ } else {
+ Assert.assertNull(dataFile.equalityFieldIds());
+ }
}
void checkManifest(ManifestFile manifest, long expectedSequenceNumber) {
diff --git a/core/src/test/java/org/apache/iceberg/TestMergeAppend.java b/core/src/test/java/org/apache/iceberg/TestMergeAppend.java
index dc47802..4fce937 100644
--- a/core/src/test/java/org/apache/iceberg/TestMergeAppend.java
+++ b/core/src/test/java/org/apache/iceberg/TestMergeAppend.java
@@ -241,8 +241,8 @@ public class TestMergeAppend extends TableTestBase {
public void testManifestMergeMinCount() throws IOException {
Assert.assertEquals("Table should start empty", 0, listManifestFiles().size());
table.updateProperties().set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "2")
- // each manifest file is 4554 bytes, so 10000 bytes limit will give us 2 bins with 3 manifest/data files.
- .set(TableProperties.MANIFEST_TARGET_SIZE_BYTES, "10000")
+ // each manifest file is 5227 bytes, so 12000 bytes limit will give us 2 bins with 3 manifest/data files.
+ .set(TableProperties.MANIFEST_TARGET_SIZE_BYTES, "12000")
.commit();
TableMetadata base = readMetadata();