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/10 22:39:12 UTC
[iceberg] branch master updated: Spark: Create shared assertions
for common task validation (#1280)
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 5e44e45 Spark: Create shared assertions for common task validation (#1280)
5e44e45 is described below
commit 5e44e45dd273aea89e93ffbcdbfc2c36e3a0e9cb
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Tue Aug 11 07:39:00 2020 +0900
Spark: Create shared assertions for common task validation (#1280)
---
.../java/org/apache/iceberg/TaskCheckHelper.java | 93 ++++++++++++++++++++++
.../apache/iceberg/TestDataFileSerialization.java | 32 +-------
.../apache/iceberg/TestScanTaskSerialization.java | 70 +---------------
3 files changed, 98 insertions(+), 97 deletions(-)
diff --git a/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java b/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java
new file mode 100644
index 0000000..9939664
--- /dev/null
+++ b/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java
@@ -0,0 +1,93 @@
+/*
+ * 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;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.Assert;
+
+public final class TaskCheckHelper {
+ private TaskCheckHelper() {
+ }
+
+ public static void assertEquals(BaseCombinedScanTask expected, BaseCombinedScanTask actual) {
+ List<FileScanTask> expectedTasks = getFileScanTasksInFilePathOrder(expected);
+ List<FileScanTask> actualTasks = getFileScanTasksInFilePathOrder(actual);
+
+ Assert.assertEquals("The number of file scan tasks should match",
+ expectedTasks.size(), actualTasks.size());
+
+ for (int i = 0; i < expectedTasks.size(); i++) {
+ FileScanTask expectedTask = expectedTasks.get(i);
+ FileScanTask actualTask = actualTasks.get(i);
+ assertEquals(expectedTask, actualTask);
+ }
+ }
+
+ public static void assertEquals(FileScanTask expected, FileScanTask actual) {
+ assertEquals(expected.file(), actual.file());
+
+ // PartitionSpec implements its own equals method
+ Assert.assertEquals("PartitionSpec doesn't match", expected.spec(), actual.spec());
+
+ Assert.assertEquals("starting position doesn't match", expected.start(), actual.start());
+
+ Assert.assertEquals("the number of bytes to scan doesn't match", expected.start(), actual.start());
+
+ // simplify comparison on residual expression via comparing toString
+ Assert.assertEquals("Residual expression doesn't match",
+ expected.residual().toString(), actual.residual().toString());
+ }
+
+ public static void assertEquals(DataFile expected, DataFile actual) {
+ Assert.assertEquals("Should match the serialized record path",
+ expected.path(), actual.path());
+ Assert.assertEquals("Should match the serialized record format",
+ expected.format(), actual.format());
+ Assert.assertEquals("Should match the serialized record partition",
+ expected.partition().get(0, Object.class), actual.partition().get(0, Object.class));
+ Assert.assertEquals("Should match the serialized record count",
+ expected.recordCount(), actual.recordCount());
+ Assert.assertEquals("Should match the serialized record size",
+ expected.fileSizeInBytes(), actual.fileSizeInBytes());
+ Assert.assertEquals("Should match the serialized record value counts",
+ expected.valueCounts(), actual.valueCounts());
+ Assert.assertEquals("Should match the serialized record null value counts",
+ expected.nullValueCounts(), actual.nullValueCounts());
+ Assert.assertEquals("Should match the serialized record lower bounds",
+ expected.lowerBounds(), actual.lowerBounds());
+ Assert.assertEquals("Should match the serialized record upper bounds",
+ expected.upperBounds(), actual.upperBounds());
+ Assert.assertEquals("Should match the serialized record key metadata",
+ expected.keyMetadata(), actual.keyMetadata());
+ Assert.assertEquals("Should match the serialized record offsets",
+ expected.splitOffsets(), actual.splitOffsets());
+ Assert.assertEquals("Should match the serialized record offsets",
+ expected.keyMetadata(), actual.keyMetadata());
+ }
+
+ private static List<FileScanTask> getFileScanTasksInFilePathOrder(BaseCombinedScanTask task) {
+ return task.files().stream()
+ // use file path + start position to differentiate the tasks
+ .sorted(Comparator.comparing(o -> o.file().path().toString() + "##" + o.start()))
+ .collect(Collectors.toList());
+ }
+}
diff --git a/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java b/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java
index d2406a7..2c06372 100644
--- a/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java
+++ b/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java
@@ -50,6 +50,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import static org.apache.iceberg.TaskCheckHelper.assertEquals;
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
@@ -107,7 +108,7 @@ public class TestDataFileSerialization {
for (int i = 0; i < 2; i += 1) {
Object obj = kryo.readClassAndObject(in);
Assert.assertTrue("Should be a DataFile", obj instanceof DataFile);
- checkDataFile(DATA_FILE, (DataFile) obj);
+ assertEquals(DATA_FILE, (DataFile) obj);
}
}
}
@@ -124,38 +125,11 @@ public class TestDataFileSerialization {
for (int i = 0; i < 2; i += 1) {
Object obj = in.readObject();
Assert.assertTrue("Should be a DataFile", obj instanceof DataFile);
- checkDataFile(DATA_FILE, (DataFile) obj);
+ assertEquals(DATA_FILE, (DataFile) obj);
}
}
}
- private void checkDataFile(DataFile expected, DataFile actual) {
- Assert.assertEquals("Should match the serialized record path",
- expected.path(), actual.path());
- Assert.assertEquals("Should match the serialized record format",
- expected.format(), actual.format());
- Assert.assertEquals("Should match the serialized record partition",
- expected.partition().get(0, Object.class), actual.partition().get(0, Object.class));
- Assert.assertEquals("Should match the serialized record count",
- expected.recordCount(), actual.recordCount());
- Assert.assertEquals("Should match the serialized record size",
- expected.fileSizeInBytes(), actual.fileSizeInBytes());
- Assert.assertEquals("Should match the serialized record value counts",
- expected.valueCounts(), actual.valueCounts());
- Assert.assertEquals("Should match the serialized record null value counts",
- expected.nullValueCounts(), actual.nullValueCounts());
- Assert.assertEquals("Should match the serialized record lower bounds",
- expected.lowerBounds(), actual.lowerBounds());
- Assert.assertEquals("Should match the serialized record upper bounds",
- expected.upperBounds(), actual.upperBounds());
- Assert.assertEquals("Should match the serialized record key metadata",
- expected.keyMetadata(), actual.keyMetadata());
- Assert.assertEquals("Should match the serialized record offsets",
- expected.splitOffsets(), actual.splitOffsets());
- Assert.assertEquals("Should match the serialized record offsets",
- expected.keyMetadata(), actual.keyMetadata());
- }
-
@Test
public void testParquetWriterSplitOffsets() throws IOException {
Iterable<InternalRow> records = RandomData.generateSpark(DATE_SCHEMA, 1, 33L);
diff --git a/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java b/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java
index f443f38..00b17ad 100644
--- a/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java
+++ b/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java
@@ -29,10 +29,8 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
-import java.util.Comparator;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.CloseableIterable;
@@ -88,7 +86,7 @@ public abstract class TestScanTaskSerialization extends SparkTestBase {
try (Input in = new Input(new FileInputStream(data))) {
Object obj = kryo.readClassAndObject(in);
Assert.assertTrue("Should be a BaseCombinedScanTask", obj instanceof BaseCombinedScanTask);
- checkBaseCombinedScanTask(scanTask, (BaseCombinedScanTask) obj);
+ TaskCheckHelper.assertEquals(scanTask, (BaseCombinedScanTask) obj);
}
}
@@ -104,7 +102,7 @@ public abstract class TestScanTaskSerialization extends SparkTestBase {
try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) {
Object obj = in.readObject();
Assert.assertTrue("Should be a BaseCombinedScanTask", obj instanceof BaseCombinedScanTask);
- checkBaseCombinedScanTask(scanTask, (BaseCombinedScanTask) obj);
+ TaskCheckHelper.assertEquals(scanTask, (BaseCombinedScanTask) obj);
}
}
@@ -131,70 +129,6 @@ public abstract class TestScanTaskSerialization extends SparkTestBase {
return new BaseCombinedScanTask(Lists.newArrayList(tasks));
}
- private void checkBaseCombinedScanTask(BaseCombinedScanTask expected, BaseCombinedScanTask actual) {
- List<FileScanTask> expectedTasks = getFileScanTasksInFilePathOrder(expected);
- List<FileScanTask> actualTasks = getFileScanTasksInFilePathOrder(actual);
-
- Assert.assertEquals("The number of file scan tasks should match",
- expectedTasks.size(), actualTasks.size());
-
- for (int i = 0; i < expectedTasks.size(); i++) {
- FileScanTask expectedTask = expectedTasks.get(i);
- FileScanTask actualTask = actualTasks.get(i);
- checkFileScanTask(expectedTask, actualTask);
- }
- }
-
- private List<FileScanTask> getFileScanTasksInFilePathOrder(BaseCombinedScanTask task) {
- return task.files().stream()
- // use file path + start position to differentiate the tasks
- .sorted(Comparator.comparing(o -> o.file().path().toString() + "##" + o.start()))
- .collect(Collectors.toList());
- }
-
- private void checkFileScanTask(FileScanTask expected, FileScanTask actual) {
- checkDataFile(expected.file(), actual.file());
-
- // PartitionSpec implements its own equals method
- Assert.assertEquals("PartitionSpec doesn't match", expected.spec(), actual.spec());
-
- Assert.assertEquals("starting position doesn't match", expected.start(), actual.start());
-
- Assert.assertEquals("the number of bytes to scan doesn't match", expected.start(), actual.start());
-
- // simplify comparison on residual expression via comparing toString
- Assert.assertEquals("Residual expression doesn't match",
- expected.residual().toString(), actual.residual().toString());
- }
-
- // TODO: this is a copy of TestDataFileSerialization.checkDataFile, deduplicate
- private void checkDataFile(DataFile expected, DataFile actual) {
- Assert.assertEquals("Should match the serialized record path",
- expected.path(), actual.path());
- Assert.assertEquals("Should match the serialized record format",
- expected.format(), actual.format());
- Assert.assertEquals("Should match the serialized record partition",
- expected.partition().get(0, Object.class), actual.partition().get(0, Object.class));
- Assert.assertEquals("Should match the serialized record count",
- expected.recordCount(), actual.recordCount());
- Assert.assertEquals("Should match the serialized record size",
- expected.fileSizeInBytes(), actual.fileSizeInBytes());
- Assert.assertEquals("Should match the serialized record value counts",
- expected.valueCounts(), actual.valueCounts());
- Assert.assertEquals("Should match the serialized record null value counts",
- expected.nullValueCounts(), actual.nullValueCounts());
- Assert.assertEquals("Should match the serialized record lower bounds",
- expected.lowerBounds(), actual.lowerBounds());
- Assert.assertEquals("Should match the serialized record upper bounds",
- expected.upperBounds(), actual.upperBounds());
- Assert.assertEquals("Should match the serialized record key metadata",
- expected.keyMetadata(), actual.keyMetadata());
- Assert.assertEquals("Should match the serialized record offsets",
- expected.splitOffsets(), actual.splitOffsets());
- Assert.assertEquals("Should match the serialized record offsets",
- expected.keyMetadata(), actual.keyMetadata());
- }
-
private void writeRecords(List<ThreeColumnRecord> records) {
Dataset<Row> df = spark.createDataFrame(records, ThreeColumnRecord.class);
writeDF(df);