Skip to content

Data: Fix equality delete set save DATE/TIMESTAMP type of data throw IllegalStateException #3135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 20 commits into from
Sep 28, 2021
Merged

Data: Fix equality delete set save DATE/TIMESTAMP type of data throw IllegalStateException #3135

Show file tree
Hide file tree
Changes from 19 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
6469812
Fix equality delete set save DATE/TIMESTAMP and other types of data t…
Sep 17, 2021
4dcfe1a
add origin comment
Sep 18, 2021
2dcdc84
fix code style
Sep 18, 2021
69d3d76
adjust comment position
Sep 18, 2021
fc9759c
fix unit-test
Sep 20, 2021
e615d51
Merge branch 'apache:master' into fix-equality-delete-set-save-date/t…
xloya Sep 20, 2021
9436bf0
ut drop table first
Sep 20, 2021
d66edfb
fix DeleteFilter code style
Sep 22, 2021
190fbfa
delete duplicate ut
Sep 22, 2021
fc613f7
fix ut
Sep 22, 2021
3c4af3c
delete unrelated ut
Sep 22, 2021
c146f19
delete redundant code
Sep 22, 2021
3463021
Merge branch 'apache:master' into fix-equality-delete-set-save-date/t…
xloya Sep 22, 2021
507f49b
add enter space
Sep 22, 2021
75530dc
Merge branch 'fix-equality-delete-set-save-date/timestamp/etc-data' o…
Sep 22, 2021
e8ab7b0
delete useless import
Sep 22, 2021
f6114eb
remove ut table1 alias
Sep 22, 2021
16bb7bd
fix ut method scope
Sep 22, 2021
f6f0804
rewrite rowSetWithoutIds()
Sep 23, 2021
84ac227
make rowSetWithoutIds static
Sep 27, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions data/src/main/java/org/apache/iceberg/data/DeleteFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,9 +134,14 @@ private List<Predicate<T>> applyEqDeletes() {

Iterable<CloseableIterable<Record>> deleteRecords = Iterables.transform(deletes,
delete -> openDeletes(delete, deleteSchema));

// copy the delete records because they will be held in a set
CloseableIterable<Record> records = CloseableIterable.transform(
CloseableIterable.concat(deleteRecords), Record::copy);

StructLikeSet deleteSet = Deletes.toEqualitySet(
// copy the delete records because they will be held in a set
CloseableIterable.transform(CloseableIterable.concat(deleteRecords), Record::copy),
CloseableIterable.transform(
records, record -> new InternalRecordWrapper(deleteSchema.asStruct()).wrap(record)),
deleteSchema.asStruct());

Predicate<T> isInDeleteSet = record -> deleteSet.contains(projectRow.wrap(asStructLike(record)));
Expand Down
113 changes: 102 additions & 11 deletions data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.apache.iceberg.data;

import java.io.IOException;
import java.time.LocalDate;
import java.util.List;
import java.util.Set;
import org.apache.iceberg.DataFile;
Expand All @@ -34,6 +35,7 @@
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ArrayUtil;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.DateTimeUtil;
import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.StructLikeSet;
import org.apache.iceberg.util.StructProjection;
Expand All @@ -51,17 +53,30 @@ public abstract class DeleteReadTests {
Types.NestedField.required(2, "data", Types.StringType.get())
);

public static final Schema DATE_SCHEMA = new Schema(
Types.NestedField.required(1, "dt", Types.DateType.get()),
Types.NestedField.required(2, "data", Types.StringType.get()),
Types.NestedField.required(3, "id", Types.IntegerType.get())
);

// Partition spec used to create tables
public static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA)
.bucket("data", 16)
.build();

public static final PartitionSpec DATE_SPEC = PartitionSpec.builderFor(DATE_SCHEMA)
.day("dt")
.build();

@Rule
public TemporaryFolder temp = new TemporaryFolder();

private String tableName = null;
protected String tableName = null;
protected String dateTableName = null;
protected Table table = null;
protected Table dateTable = null;
private List<Record> records = null;
private List<Record> dateRecords = null;
private DataFile dataFile = null;

@Before
Expand Down Expand Up @@ -90,6 +105,46 @@ public void writeTestDataFile() throws IOException {
@After
public void cleanup() throws IOException {
dropTable("test");
dropTable("test2");
}

private void initDateTable() throws IOException {
dropTable("test2");
this.dateTableName = "test2";
this.dateTable = createTable(dateTableName, DATE_SCHEMA, DATE_SPEC);

GenericRecord record = GenericRecord.create(dateTable.schema());

this.dateRecords = Lists.newArrayList(
record.copy("dt", LocalDate.parse("2021-09-01"), "data", "a", "id", 1),
record.copy("dt", LocalDate.parse("2021-09-02"), "data", "b", "id", 2),
record.copy("dt", LocalDate.parse("2021-09-03"), "data", "c", "id", 3),
record.copy("dt", LocalDate.parse("2021-09-04"), "data", "d", "id", 4),
record.copy("dt", LocalDate.parse("2021-09-05"), "data", "e", "id", 5));

DataFile dataFile1 = FileHelpers.writeDataFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-01"))), dateRecords.subList(0, 1));
DataFile dataFile2 = FileHelpers.writeDataFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-02"))), dateRecords.subList(1, 2));
DataFile dataFile3 = FileHelpers.writeDataFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-03"))), dateRecords.subList(2, 3));
DataFile dataFile4 = FileHelpers.writeDataFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-04"))), dateRecords.subList(3, 4));
DataFile dataFile5 = FileHelpers.writeDataFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-05"))), dateRecords.subList(4, 5));

dateTable.newAppend()
.appendFile(dataFile1)
.appendFile(dataFile2)
.appendFile(dataFile3)
.appendFile(dataFile4)
.appendFile(dataFile5)
.commit();
}

protected abstract Table createTable(String name, Schema schema, PartitionSpec spec) throws IOException;
Expand Down Expand Up @@ -119,12 +174,47 @@ public void testEqualityDeletes() throws IOException {
.addDeletes(eqDeletes)
.commit();

StructLikeSet expected = rowSetWithoutIds(29, 89, 122);
StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 122);
StructLikeSet actual = rowSet(tableName, table, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
}

@Test
public void testEqualityDateDeletes() throws IOException {
initDateTable();

Schema deleteRowSchema = dateTable.schema().select("*");
Record dataDelete = GenericRecord.create(deleteRowSchema);
List<Record> dataDeletes = Lists.newArrayList(
dataDelete.copy("dt", LocalDate.parse("2021-09-01"), "data", "a", "id", 1),
dataDelete.copy("dt", LocalDate.parse("2021-09-02"), "data", "b", "id", 2),
dataDelete.copy("dt", LocalDate.parse("2021-09-03"), "data", "c", "id", 3)
);

DeleteFile eqDeletes1 = FileHelpers.writeDeleteFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-01"))), dataDeletes.subList(0, 1), deleteRowSchema);
DeleteFile eqDeletes2 = FileHelpers.writeDeleteFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-02"))), dataDeletes.subList(1, 2), deleteRowSchema);
DeleteFile eqDeletes3 = FileHelpers.writeDeleteFile(
dateTable, Files.localOutput(temp.newFile()),
Row.of(DateTimeUtil.daysFromDate(LocalDate.parse("2021-09-03"))), dataDeletes.subList(2, 3), deleteRowSchema);

dateTable.newRowDelta()
.addDeletes(eqDeletes1)
.addDeletes(eqDeletes2)
.addDeletes(eqDeletes3)
.commit();

StructLikeSet expected = rowSetWithoutIds(dateTable, dateRecords, 1, 2, 3);

StructLikeSet actual = rowSet(dateTableName, dateTable, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
}

@Test
public void testEqualityDeletesWithRequiredEqColumn() throws IOException {
Schema deleteRowSchema = table.schema().select("data");
Expand All @@ -142,7 +232,7 @@ public void testEqualityDeletesWithRequiredEqColumn() throws IOException {
.addDeletes(eqDeletes)
.commit();

StructLikeSet expected = selectColumns(rowSetWithoutIds(29, 89, 122), "id");
StructLikeSet expected = selectColumns(rowSetWithoutIds(table, records, 29, 89, 122), "id");
StructLikeSet actual = rowSet(tableName, table, "id");

if (expectPruned()) {
Expand Down Expand Up @@ -180,7 +270,7 @@ public void testEqualityDeletesSpanningMultipleDataFiles() throws IOException {
.addDeletes(eqDeletes)
.commit();

StructLikeSet expected = rowSetWithoutIds(29, 89, 122, 144);
StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 122, 144);
StructLikeSet actual = rowSet(tableName, table, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
Expand All @@ -202,7 +292,7 @@ public void testPositionDeletes() throws IOException {
.validateDataFilesExist(posDeletes.second())
.commit();

StructLikeSet expected = rowSetWithoutIds(29, 89, 122);
StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 122);
StructLikeSet actual = rowSet(tableName, table, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
Expand Down Expand Up @@ -235,7 +325,7 @@ public void testMixedPositionAndEqualityDeletes() throws IOException {
.validateDataFilesExist(posDeletes.second())
.commit();

StructLikeSet expected = rowSetWithoutIds(29, 89, 121, 122);
StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 121, 122);
StructLikeSet actual = rowSet(tableName, table, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
Expand Down Expand Up @@ -269,7 +359,7 @@ public void testMultipleEqualityDeleteSchemas() throws IOException {
.addDeletes(idEqDeletes)
.commit();

StructLikeSet expected = rowSetWithoutIds(29, 89, 121, 122);
StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 121, 122);
StructLikeSet actual = rowSet(tableName, table, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
Expand Down Expand Up @@ -306,7 +396,7 @@ public void testEqualityDeleteByNull() throws IOException {
.addDeletes(eqDeletes)
.commit();

StructLikeSet expected = rowSetWithoutIds(131);
StructLikeSet expected = rowSetWithoutIds(table, records, 131);
StructLikeSet actual = rowSet(tableName, table, "*");

Assert.assertEquals("Table should contain expected rows", expected, actual);
Expand All @@ -321,11 +411,12 @@ private StructLikeSet selectColumns(StructLikeSet rows, String... columns) {
return set;
}

private StructLikeSet rowSetWithoutIds(int... idsToRemove) {
private StructLikeSet rowSetWithoutIds(Table iTable, List<Record> recordList, int... idsToRemove) {
Copy link
Contributor

@rdblue rdblue Sep 26, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since you're passing the table in, can you make this a static method? You may also be able to use the name table instead of iTable if it is static.

Copy link
Contributor Author

@xloya xloya Sep 27, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A helpful suggestion, it works, thanks!

Set<Integer> deletedIds = Sets.newHashSet(ArrayUtil.toIntList(idsToRemove));
StructLikeSet set = StructLikeSet.create(table.schema().asStruct());
records.stream()
StructLikeSet set = StructLikeSet.create(iTable.schema().asStruct());
recordList.stream()
.filter(row -> !deletedIds.contains(row.getField("id")))
.map(record -> new InternalRecordWrapper(iTable.schema().asStruct()).wrap(record))
.forEach(set::add);
return set;
}
Copy link
Contributor

@rdblue rdblue Sep 22, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the other rowSetWithoutIds method should be rewritten to call this one since it is more generic. That will also avoid the wrapper problem in the future.

Copy link
Contributor Author

@xloya xloya Sep 23, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok i'll rewrite the origin rowSetWithoutIds() method

Expand Down
4 changes: 3 additions & 1 deletion data/src/test/java/org/apache/iceberg/data/TestGenericReaderDeletes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.iceberg.Table;
import org.apache.iceberg.TestTables;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.util.StructLikeSet;
import org.junit.Assert;

Expand All @@ -48,7 +49,8 @@ protected void dropTable(String name) {
public StructLikeSet rowSet(String name, Table table, String... columns) throws IOException {
StructLikeSet set = StructLikeSet.create(table.schema().asStruct());
try (CloseableIterable<Record> reader = IcebergGenerics.read(table).select(columns).build()) {
reader.forEach(set::add);
Iterables.addAll(set, CloseableIterable.transform(
reader, record -> new InternalRecordWrapper(table.schema().asStruct()).wrap(record)));
}
return set;
}
Expand Down
2 changes: 2 additions & 0 deletions mr/src/test/java/org/apache/iceberg/mr/TestInputFormatReaderDeletes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableOperations;
import org.apache.iceberg.data.DeleteReadTests;
import org.apache.iceberg.data.InternalRecordWrapper;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.util.StructLikeSet;
import org.junit.Assert;
Expand Down Expand Up @@ -104,6 +105,7 @@ public StructLikeSet rowSet(String name, Table table, String... columns) {
.filter(recordFactory -> recordFactory.name().equals(inputFormat))
.map(recordFactory -> recordFactory.create(builder.project(projected).conf()).getRecords())
.flatMap(List::stream)
.map(record -> new InternalRecordWrapper(projected.asStruct()).wrap(record))
.collect(Collectors.toList())
);

Expand Down