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

xloya · 2021-09-17T06:22:52Z

1. What this MR does:
When the equality delete set in DeleteFilter saves a piece of data read from an equal delete parquet file in memory, it will call the get() method of the GenericRecord class for data comparison.If the table schema definition of the current position of GenericRecord is DATE/TIMESTAMP/TIME type, it will be converted to LocalDate/LocalDateTime/LocalTime type in memory. When the get() method of GenericRecord is called, the java class Integer/Long/Long defined in the Types enum of the DATE/TIMESTAMP/TIME type is passed in. Because the type data of LocalDate/LocalDateTime/LocalTime is not an instance of Integer/Long/Long type, an IllegalStateException with Not an intance of xxx is thrown.

The pr first converts the data to the InternalRecordWrapper before saving the data in the equality delete set, and calls its get() method during comparison. If it is a type that needs to be converted such as DATE/TIMESTAMP/TIME, use the transforms parameter to convert, otherwise directly Output raw data type.
2. Which issue this PR fixes:
Fix issue #3119.

@rdblue @chenjunjiedada @openinx Could you help to take a look at this PR? Thanks in advance!

…hrow IllegalStateException

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

rdblue · 2021-09-19T16:48:02Z

Running CI

…imestamp/etc-data

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java

rdblue · 2021-09-20T16:19:36Z

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

+
+    Iterables.addAll(actualSet, actual.stream()
+            .map(record -> new InternalRecordWrapper(table2.schema().asStruct()).wrap(record))
+            .collect(Collectors.toList()));


Why are you using InternalRecordWrapper here and for the expected set? I think that StructLikeSet should work without it, right? If not, then we need to update the set that is produced by rowSet to do the wrapping since future tests may not know to wrap.

@xloya, can you briefly describe what you did to address this?

@rdblue OK. The original rowSetWithoutIds()method of DeleteReadTests and the original rowSet() method of TestGenericReaderDeletes returns both the GenericRecord class's data. In TestFlinkInputFormatReaderDeletes the rowSet() method returns RowDataWrapper class's data, and in TestSparkReaderDeletes the rowSet() method returns SparkStructLike class's data.Calling the Assert.assertEquals() method will compare the two StructLikeSets. If it is DATE/TIMESTAMP type of the data of GenericRecord, it will compare(GenericRecord, GenericRecord) and throw an exception of Not a instance of Integer/Long. If actual data class is RowDataWrapper or SparkStructLike, it will compare(GenericRecord, RowDataWrapper/SparkStructLike) and their class mismatch. so they both need to be converted to the InternalRecordWrapper class for comparison

I think either the conversion at the assertion or the conversion at rowSetWithoutIds() and rowSet() is fine, but it must be converted.

rdblue · 2021-09-20T16:22:34Z

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

+    );
+
+    DeleteFile eqDeletes = FileHelpers.writeDeleteFile(
+            table2, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes, deleteRowSchema);


I don't see how this delete file would be applied. The partition you're writing it to is Row.of(0), which ends up being 1970-01-01. That should prevent the delete file from being found when planning, so none of the deletes would be applied.

@xloya, what was the resolution to this? Were the deletes not applied? Was the test incorrect and not working?

@rdblue Previously, the data file was written using Row.of(0), which means that all data is written to the 1970-01-01 partition. So all the equality delete data before is also written to Row.of(0). Now it is changed that the data is written to the corresponding date time partition, and then the equality delete data is also written to the corresponding partition.
If keep the data file written to the partition as 1970-01-01, but the partition where the equality delete data is actually written is changed to 2021-09-01, it will not work.
Found spec on official website:
http://iceberg.apache.org/spec/
Like data files, delete files are tracked by partition. In general, a delete file must be applied to older data files with the same partition; see Scan Planning for details. Column metrics can be used to determine whether a delete file’s rows overlap the contents of a data file or a scan range.

Makes sense. Thanks!

rdblue · 2021-09-20T20:03:39Z

Thanks, @xloya! The implementation looks correct to me, but there are a few things to fix in the tests.

xloya · 2021-09-21T04:22:01Z

Thanks @rdblue for your particular review! I will fix them according to your suggestion these days

…imestamp/etc-data

…f github.com:xloya/iceberg into fix-equality-delete-set-save-date/timestamp/etc-data

rdblue · 2021-09-22T18:12:01Z

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

    return set;
  }

+  private StructLikeSet rowSetWithoutIds(Set<Integer> idSet, Table iTable, List<Record> recordList) {


Nit: the name iTable is odd. Why not just table?

I originally used table, but code style check would report a hidden field error.Maybe there is a conflict with a property called table. iTable means iceberg table.

rdblue · 2021-09-22T18:14:47Z

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

+        .map(record -> new InternalRecordWrapper(iTable.schema().asStruct()).wrap(record))
+        .forEach(set::add);
+    return set;
+  }


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.

ok i'll rewrite the origin rowSetWithoutIds() method

rdblue · 2021-09-26T23:19:23Z

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

  }

-  private StructLikeSet rowSetWithoutIds(int... idsToRemove) {
+  private StructLikeSet rowSetWithoutIds(Table iTable, List<Record> recordList, int... idsToRemove) {


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.

A helpful suggestion, it works, thanks!

rdblue · 2021-09-28T16:35:12Z

Looks good to me. Running CI.

rdblue · 2021-09-28T18:05:16Z

Thanks, @xloya! Merged.

xloya · 2021-09-29T01:27:34Z

Thanks for your time, @rdblue !

Co-authored-by: xiaojiebao <[email protected]>

…IllegalStateException apache#3135(merged)

Fix equality delete set save DATE/TIMESTAMP and other types of data t…

6469812

…hrow IllegalStateException

github-actions bot added data flink spark labels Sep 17, 2021

jiasheng55 reviewed Sep 18, 2021

View reviewed changes

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java Show resolved Hide resolved

xiaojiebao added 3 commits September 18, 2021 16:21

add origin comment

4dcfe1a

fix code style

2dcdc84

adjust comment position

69d3d76

xiaojiebao and others added 3 commits September 20, 2021 18:37

fix unit-test

fc9759c

Merge branch 'apache:master' into fix-equality-delete-set-save-date/t…

e615d51

…imestamp/etc-data

ut drop table first

9436bf0