chenjunjiedada commented on a change in pull request #1497:
URL: https://github.com/apache/iceberg/pull/1497#discussion_r494015472
##########
File path:
mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
##########
@@ -248,6 +258,26 @@ public void close() throws IOException {
return iterable;
}
+ @SuppressWarnings("unchecked")
Review comment:
`deletes.filter(...)` needs this.
##########
File path:
spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java
##########
@@ -93,212 +102,4 @@ public void createTable() throws IOException {
public void dropTable() {
catalog.dropTable(TableIdentifier.of("default", "table"));
}
-
- @Test
- public void testEqualityDeletes() throws IOException {
- Schema deleteRowSchema = table.schema().select("data");
- Record dataDelete = GenericRecord.create(deleteRowSchema);
- List<Record> dataDeletes = Lists.newArrayList(
- dataDelete.copy("data", "a"), // id = 29
- dataDelete.copy("data", "d"), // id = 89
- dataDelete.copy("data", "g") // id = 122
- );
-
- DeleteFile eqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes,
deleteRowSchema);
-
- table.newRowDelta()
- .addDeletes(eqDeletes)
- .commit();
-
- StructLikeSet expected = rowSetWithoutIds(29, 89, 122);
- StructLikeSet actual = rowSet(table);
-
- Assert.assertEquals("Table should contain expected rows", expected,
actual);
- }
-
- @Test
- public void testEqualityDeletesWithRequiredEqColumn() throws IOException {
- Schema deleteRowSchema = table.schema().select("data");
- Record dataDelete = GenericRecord.create(deleteRowSchema);
- List<Record> dataDeletes = Lists.newArrayList(
- dataDelete.copy("data", "a"), // id = 29
- dataDelete.copy("data", "d"), // id = 89
- dataDelete.copy("data", "g") // id = 122
- );
-
- DeleteFile eqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes,
deleteRowSchema);
-
- table.newRowDelta()
- .addDeletes(eqDeletes)
- .commit();
-
- StructLikeSet expected = selectColumns(rowSetWithoutIds(29, 89, 122),
"id");
- StructLikeSet actual = rowSet(table, "id"); // data is added by the reader
to apply the eq deletes
-
- Assert.assertEquals("Table should contain expected rows", expected,
actual);
- }
-
- @Test
- public void testPositionDeletes() throws IOException {
- List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(
- Pair.of(dataFile.path(), 0L), // id = 29
- Pair.of(dataFile.path(), 3L), // id = 89
- Pair.of(dataFile.path(), 6L) // id = 122
- );
-
- DeleteFile posDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), deletes);
-
- table.newRowDelta()
- .addDeletes(posDeletes)
- .commit();
-
- StructLikeSet expected = rowSetWithoutIds(29, 89, 122);
- StructLikeSet actual = rowSet(table);
-
- Assert.assertEquals("Table should contain expected rows", expected,
actual);
- }
-
- @Test
- public void testMixedPositionAndEqualityDeletes() throws IOException {
- Schema dataSchema = table.schema().select("data");
- Record dataDelete = GenericRecord.create(dataSchema);
- List<Record> dataDeletes = Lists.newArrayList(
- dataDelete.copy("data", "a"), // id = 29
- dataDelete.copy("data", "d"), // id = 89
- dataDelete.copy("data", "g") // id = 122
- );
-
- DeleteFile eqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes,
dataSchema);
-
- List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(
- Pair.of(dataFile.path(), 3L), // id = 89
- Pair.of(dataFile.path(), 5L) // id = 121
- );
-
- DeleteFile posDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), deletes);
-
- table.newRowDelta()
- .addDeletes(eqDeletes)
- .addDeletes(posDeletes)
- .commit();
-
- StructLikeSet expected = rowSetWithoutIds(29, 89, 121, 122);
- StructLikeSet actual = rowSet(table);
-
- Assert.assertEquals("Table should contain expected rows", expected,
actual);
- }
-
- @Test
- public void testMultipleEqualityDeleteSchemas() throws IOException {
- Schema dataSchema = table.schema().select("data");
- Record dataDelete = GenericRecord.create(dataSchema);
- List<Record> dataDeletes = Lists.newArrayList(
- dataDelete.copy("data", "a"), // id = 29
- dataDelete.copy("data", "d"), // id = 89
- dataDelete.copy("data", "g") // id = 122
- );
-
- DeleteFile dataEqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes,
dataSchema);
-
- Schema idSchema = table.schema().select("id");
- Record idDelete = GenericRecord.create(idSchema);
- List<Record> idDeletes = Lists.newArrayList(
- idDelete.copy("id", 121), // id = 121
- idDelete.copy("id", 29) // id = 29
- );
-
- DeleteFile idEqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), idDeletes,
idSchema);
-
- table.newRowDelta()
- .addDeletes(dataEqDeletes)
- .addDeletes(idEqDeletes)
- .commit();
-
- StructLikeSet expected = rowSetWithoutIds(29, 89, 121, 122);
- StructLikeSet actual = rowSet(table);
-
- Assert.assertEquals("Table should contain expected rows", expected,
actual);
- }
-
- @Test
- public void testEqualityDeleteByNull() throws IOException {
- // data is required in the test table; make it optional for this test
- table.updateSchema()
- .makeColumnOptional("data")
- .commit();
-
- // add a new data file with a record where data is null
- Record record = GenericRecord.create(table.schema());
- DataFile dataFileWithNull = FileHelpers.writeDataFile(
- table, Files.localOutput(temp.newFile()), Row.of(0),
- Lists.newArrayList(record.copy("id", 131, "data", null)));
-
- table.newAppend()
- .appendFile(dataFileWithNull)
- .commit();
-
- // delete where data is null
- Schema dataSchema = table.schema().select("data");
- Record dataDelete = GenericRecord.create(dataSchema);
- List<Record> dataDeletes = Lists.newArrayList(
- dataDelete.copy("data", null) // id = 131
- );
-
- DeleteFile eqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes,
dataSchema);
-
- table.newRowDelta()
- .addDeletes(eqDeletes)
- .commit();
-
- StructLikeSet expected = rowSetWithoutIds(131);
- StructLikeSet actual = rowSet(table);
-
- Assert.assertEquals("Table should contain expected rows", expected,
actual);
- }
-
- private static StructLikeSet rowSet(Table table) {
- return rowSet(table, "*");
- }
-
- private static StructLikeSet rowSet(Table table, String... columns) {
Review comment:
Sorry, I missed this. I just added this back and also use input format
to read records.
##########
File path: data/src/test/java/org/apache/iceberg/data/DeletesReadTest.java
##########
@@ -148,15 +144,15 @@ public void testMixedPositionAndEqualityDeletes() throws
IOException {
);
DeleteFile eqDeletes = FileHelpers.writeDeleteFile(
- table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes,
dataSchema);
+ table, Files.localOutput(temp.newFile()), TestHelpers.Row.of(0),
dataDeletes, dataSchema);
Review comment:
Done.
##########
File path:
data/src/test/java/org/apache/iceberg/data/GenericReaderDeletesTest.java
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.data;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
+import org.junit.Before;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+public class GenericReaderDeletesTest extends DeletesReadTest {
+ // Schema passed to create tables
+ public static final Schema SCHEMA = new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ required(2, "data", Types.StringType.get())
+ );
Review comment:
Updated.
##########
File path: data/src/test/java/org/apache/iceberg/data/DeletesReadTest.java
##########
@@ -92,6 +72,22 @@ public void testEqualityDeletes() throws IOException {
Assert.assertEquals("Table should contain expected rows", expected,
actual);
}
+ protected void generateTestData() throws IOException {
+ this.records = new ArrayList<>();
Review comment:
Done.
##########
File path:
data/src/test/java/org/apache/iceberg/data/GenericReaderDeletesTest.java
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.data;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
+import org.junit.Before;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+public class GenericReaderDeletesTest extends DeletesReadTest {
+ // Schema passed to create tables
+ public static final Schema SCHEMA = new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ required(2, "data", Types.StringType.get())
+ );
+
+ // Partition spec used to create tables
+ static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA)
+ .bucket("data", 16)
+ .build();
+
+ @Before
+ public void writeTestDataFile() throws IOException {
+ File tableDir = temp.newFolder();
+ tableDir.delete();
+ this.table = TestTables.create(tableDir, "test", SCHEMA, SPEC, 2);
Review comment:
Make sense to me. Updated.
##########
File path:
mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
##########
@@ -248,6 +258,26 @@ public void close() throws IOException {
return iterable;
}
+ @SuppressWarnings("unchecked")
+ private CloseableIterable<T> open(FileScanTask currentTask, Schema
readSchema) {
+ CloseableIterable<T> iter;
+ switch (inMemoryDataModel) {
+ case PIG:
+ case HIVE:
+ // TODO implement value readers for Pig and Hive
+ throw new UnsupportedOperationException("Avro support not yet
supported for Pig and Hive");
+ case GENERIC:
+ DeleteFilter deletes = new GenericDeleteFilter(io, currentTask,
tableSchema, readSchema);
+ Schema requiredSchema = deletes.requiredSchema();
+ iter = deletes.filter(openTask(currentTask, requiredSchema));
Review comment:
Updated.
##########
File path: mr/src/test/java/org/apache/iceberg/mr/TestMrReadDeletes.java
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.mr;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Locale;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.data.DeletesReadTest;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.types.Types;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+@RunWith(Parameterized.class)
+public class TestMrReadDeletes extends DeletesReadTest {
+ // Schema passed to create tables
+ public static final Schema SCHEMA = new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ required(2, "data", Types.StringType.get())
+ );
+
+ // Partition spec used to create tables
+ static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA)
+ .bucket("data", 16)
+ .build();
+
+ // parametrized variables
+ private final TestIcebergInputFormats.TestInputFormat.Factory<Record>
testInputFormat;
+ private final FileFormat fileFormat;
+
+ @Parameterized.Parameters
+ public static Object[][] parameters() {
+ Object[][] parameters = new
Object[TestIcebergInputFormats.TESTED_INPUT_FORMATS.size() *
+ TestIcebergInputFormats.TESTED_FILE_FORMATS.size()][2];
+
+ int idx = 0;
+
+ for (TestIcebergInputFormats.TestInputFormat.Factory<Record> inputFormat :
+ TestIcebergInputFormats.TESTED_INPUT_FORMATS) {
+ for (String fileFormat : TestIcebergInputFormats.TESTED_FILE_FORMATS) {
+ parameters[idx++] = new Object[] {inputFormat, fileFormat};
+ }
+ }
+
+ return parameters;
Review comment:
Yes, I just updated it.
##########
File path: data/src/test/java/org/apache/iceberg/data/DeletesReadTest.java
##########
@@ -269,4 +265,5 @@ private StructLikeSet rowSetWithoutIds(int... idsToRemove) {
.forEach(set::add);
return set;
}
+
Review comment:
Removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
