MR: apply row-level delete files when reading by chenjunjiedada · Pull Request #1497 · apache/iceberg

chenjunjiedada · 2020-09-23T13:39:47Z

This applies row-level delete files when reading for IcebergInputFormat. This also includes changes from #985.

This also refactors the deletes read unit tests to a separated base test class to avoid duplication.

rdblue · 2020-09-23T23:16:58Z


    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);


Can you import this class directly to avoid so many changes in this file?

rdblue · 2020-09-23T23:17:10Z

        .forEach(set::add);
    return set;
  }
+


Nit: unnecessary whitespace change

rdblue · 2020-09-23T23:19:39Z

          checkResiduals(task);
        }
-        splits.add(new IcebergSplit(conf, task));
+        splits.add(new IcebergSplit(conf, task, table.io(), table.encryption()));


While I would like to get the encryption manager and io changes in, I don't think that they should be mixed into this commit. Was it necessary to do this for some reason?

The GenericDeleteFilter constructor needs FileIO as parameter.

In that case, this can pass the FileIO somehow, or we can work on getting the other PR done before this one. But I don't think we should mix the two features together.

For now, this could create a new HadoopFileIO and use that instead. That would be the easiest path forward.

@holdenk just added a FileIO instance to this class, so you can use that instead of mixing the two PRs together. Thanks @holdenk!

Thank you @holdenk @rdblue. I updated this.

rdblue · 2020-09-23T23:22:47Z

+  public static final Schema SCHEMA = new Schema(
+      required(1, "id", Types.IntegerType.get()),
+      required(2, "data", Types.StringType.get())
+  );


Why not put the schema and spec in the parent class, DeletesReadTest? The data it generates is for this schema.

rdblue · 2020-09-23T23:25:20Z

  }

+  protected void generateTestData() throws IOException {
+    this.records = new ArrayList<>();


We prefer using Lists.newArrayList()

rdblue · 2020-09-23T23:25:40Z

+  public void writeTestDataFile() throws IOException {
+    File tableDir = temp.newFolder();
+    tableDir.delete();
+    this.table = TestTables.create(tableDir, "test", SCHEMA, SPEC, 2);


I think a better way to break down the class would be to have an abstract Table createTable(String name, Schema, Spec) method. Then the table and dataFile fields don't need to be shared. I also don't think that there is a need to make records public either.

Make sense to me. Updated.

rdblue · 2020-09-23T23:27:12Z

      return iterable;
    }

+    @SuppressWarnings("unchecked")


Why is this needed?

deletes.filter(...) needs this.

Okay, makes sense.

rdblue · 2020-09-23T23:27:56Z

+        case GENERIC:
+          DeleteFilter deletes = new GenericDeleteFilter(io, currentTask, tableSchema, readSchema);
+          Schema requiredSchema = deletes.requiredSchema();
+          iter = deletes.filter(openTask(currentTask, requiredSchema));


Why not return deletes.filter(...) here? That would remove the need for iter and break.

rdblue · 2020-09-23T23:29:25Z

+      }
+    }
+
+    return parameters;


Is there a simpler way to configure this? Normally, we build these using literals instead of a block of code.

Yes, I just updated it.

rdblue · 2020-09-23T23:35:08Z

-    return rowSet(table, "*");
-  }
-
-  private static StructLikeSet rowSet(Table table, String... columns) {


This method is what reads the rows from the table using Spark. Deleting this method and using the one in DeletesReadTest makes this test suite use the exact same read path as the generics -- IcebergGenerics.

You can probably make this method abstract and implement it in both classes to get around this. You'll also need to implement a read using the input format or Hive runner to test the Hive code.

Sorry, I missed this. I just added this back and also use input format to read records.

rdsr

LGTM!. The changes in MR to incorporate deletes are minimal and clean!

rdblue · 2020-09-29T16:47:25Z

  }

-  private abstract static class TestInputFormat<T> {
+  public abstract static class TestInputFormat<T> {


Why was this changed to public?

This is because the testInputFormat is type of TestIcebergInputFormats.TestInputFormat.Factory<Record> which needs to access the TestInputFormat

I tried to minimize the changes to access modifiers in TestIcebergInputFormat, the last place that needs the modifier of TestInputFormat to be public is getRecords method.

rdblue · 2020-10-01T16:39:38Z

    records.add(record.copy("id", 122, "data", "g"));

    this.dataFile = FileHelpers.writeDataFile(table, Files.localOutput(temp.newFile()), Row.of(0), records);
-


Nit: whitespace-only change.

rdblue · 2020-10-01T16:40:06Z


  @Before
-  public void writeTestDataFile() throws IOException {
+  public void prepareData() throws IOException {


Nit: unnecessary method renames cause more changed lines than needed.

rdblue · 2020-10-01T16:40:42Z

-  private List<Record> records = null;
-  private DataFile dataFile = null;
+  private DataFile dataFile;
+  private List<Record> records;


Changing the order of these two lines and dropping the default also causes unnecessary changes.

rdblue · 2020-10-01T16:47:59Z

+      .bucket("data", 16)
+      .build();
+
+  protected final String testTableName = "test";


I think this should be private. If it is needed by subclases, it should be passed into methods, not shared. I think this is only used by Spark, so it should be easy to fix.

rdblue · 2020-10-02T18:15:38Z

@chenjunjiedada, I went ahead and fixed the remaining issues and opened a PR against your branch. Could you please take a look?

rdblue · 2020-10-05T17:02:15Z

Merged. Thanks, @chenjunjiedada!

probot-autolabeler Bot added data MR spark labels Sep 23, 2020

rdblue requested review from rdsr and shardulm94 September 23, 2020 23:15

rdblue reviewed Sep 23, 2020

View reviewed changes

chenjunjiedada force-pushed the apply-row-level-deletes-when-reading branch from 8faeac6 to 84310d9 Compare September 24, 2020 03:30

rdsr approved these changes Sep 26, 2020

View reviewed changes

This was referenced Sep 27, 2020

Flink: apply row-level deletes when reading data #1516

Closed

Flink: apply row-level delete when reading #1517

Merged