Core: read delete files in parallel #3120

Reo-LEI · 2021-09-15T03:18:19Z

This PR is trying to close #3118.

I move ParallelIterable to api module and use it impl parallel concat api in CloseableIterable. Now we could use the parallel concat to replace the serial concat and read delete files in parallel when DeleteFilter construct the equalitySet and positionSet.

Reo-LEI · 2021-09-15T03:19:54Z

@rdblue @aokolnychyi @kbendick @openinx @stevenzwu Could you take a look of this? :)

core/src/main/java/org/apache/iceberg/util/ThreadPools.java

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

stevenzwu · 2021-09-15T18:14:07Z

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

      case AVRO:
        return Avro.read(input)
            .project(deleteSchema)
-            .reuseContainers()


is this related?

I think this is needed because records are now placed on a queue inside the parallel iterator. Reusing the record instance instead of copying would cause a problem.

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

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

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

core/src/main/java/org/apache/iceberg/util/ThreadPools.java

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

…ad pool to run.

Reo-LEI · 2021-09-22T12:04:56Z

I have been make the parallelize in configurable and let it run in a separate thread pool.
Could you help review this PR again? @rdblue @stevenzwu @kbendick @jackye1995

rdblue · 2021-09-27T22:52:06Z

Thanks, @Reo-LEI. I'll take another look.

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

rdblue · 2021-09-27T23:57:30Z

core/src/main/java/org/apache/iceberg/AllDataFilesTable.java

-        Iterables.transform(snapshots, snapshot -> (Iterable<ManifestFile>) () -> snapshot.dataManifests().iterator()),
-        ThreadPools.getWorkerPool())) {
+    try (CloseableIterable<ManifestFile> iterable = CloseableIterable.combine(
+        Iterables.transform(snapshots, snapshot -> CloseableIterable.withNoopClose(snapshot.dataManifests())),


I don't think that this should add a noop close here. Can you avoid adding this in all the updated calls to create a parallel iterable?

I think I can't avoid adding this, becasue combine will call concat when workerPool is null, and concat receive Iterable<CloseableIterable<E>> but not Iterable<Iterable<E>>. So I need to add a noop close to wrap it and pass to combine.

This does need to be fixed. ParallelIterable was constructed using Iterable<? extends Iterable<T>>. There should be a version of CloseableIterable.combine that accepts the same type. That may mean updating CloseableIterable.concat to accept the same ? extends Iterable<T> in addition to strictly a CloseableIterable<T>. But that should be okay, since you can update to close the iterable if it is closeable.

rdblue · 2021-09-27T23:58:33Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

-    try (CloseableIterable<StructLike> deletes = eqDeletes) {
+  public static <T extends StructLike> StructLikeSet toEqualitySet(CloseableIterable<T> eqDeletes,
+                                                                   Types.StructType eqType) {
+    try (CloseableIterable<T> deletes = eqDeletes) {


Why was this change needed?

To avoid to transform Record to StructLike as your comment #3120 (comment).

Seems reasonable.

rdblue · 2021-09-27T23:59:19Z

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

+  private ExecutorService readDeletesService;
+
+  protected DeleteFilter(FileScanTask task, Schema tableSchema, Schema requestedSchema,
+                         Map<String, String> tableProperties) {


This class should not take table properties. I think it should take an executor service instead.

I think DeleteFilter should make sure the reader will read delete files in parallel by itself, if we config it, but not pass a executor service by the caller. So I create the read service internally, and read the config from system properties.

rdblue · 2021-09-27T23:59:47Z

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

+            new ThreadFactoryBuilder()
+                .setNameFormat("Read-delete-Service-%d")
+                .build()));
+  }


When is this executor service cleaned up?

I think we chould create a custom executor and set corePoolSize = 0, maximumPoolSize = poolSzie, workQueue = new LinkedBlockingQueue<Runnable>() and let executor shutdown automatically. Otherwise, we need to call shutdown in filter or add close metod and call by DeleterFilter caller.

I make this executor service as static. Now, different filter task can share the same executor service, and we don't need to consider when we should shutdown and cleaned up the executor service.

rdblue · 2021-09-28T00:00:59Z

core/src/main/java/org/apache/iceberg/TableProperties.java

  public static final long SPLIT_OPEN_FILE_COST_DEFAULT = 4 * 1024 * 1024; // 4MB

+  public static final String READ_DELETE_FILES_WORKER_POOL_SIZE = "read.deletes.num-threads";
+  public static final int READ_DELETE_FILES_WORKER_POOL_SIZE_DEFAULT = 1;  // read delete files in serial.


I don't think this makes much sense as a table property. Table properties are for table configuration, but this is an engine concern. Many engines handle parallelism internally so this wouldn't be appropriate. I think that Flink should expose a setting and manage the thread pool for using a delete file reader pool.

I admit that this configuration may not be appropriate as a table property. But I think this optimization should apply to all engines, not only flink. Such as I will sync mysql cdc data to iceberg by flink and rewrite and merge the delete files into data files by spark.

Maybe we chould config this through SystemProperties like we config iceberg.worker.num-threads(https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/SystemProperties.java#L35). And stop propagate the 'tableProperties'.

I move this config to SystemProperties. So we can stop propagate the 'tableProperties' and handle the parallelism internally.

Is it not possible to have Flink pass in the executor service it chooses to use?

…-file-in-parallel

…m/Reo-LEI/iceberg into core-read-delete-file-in-parallel

Reo-LEI · 2021-11-01T11:26:58Z

I adressed some comment and leave some comment. @rdblue Could you take another looks of this PR? 😄

rdblue · 2021-11-03T16:20:13Z

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

+  private final int workerPoolSize;

+  /**
+   * @deprecated please use {@link CloseableIterable#combine(Iterable, ExecutorService, int)} instead.


You can omit "please" from documentation so that docs are direct and as short as possible.

Also, can you add "will be removed in 0.14.0"? We like to keep track of when things can be removed.

rdblue · 2021-11-03T16:20:40Z

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

      this.workerPool = workerPool;
-      // submit 2 tasks per worker at a time
-      this.taskFutures = new Future[2 * ThreadPools.WORKER_THREAD_POOL_SIZE];
+      // submit 2 tasks per worker at a time.


Can you remove the non-functional change on this line? We don't want unnecessary changes to cause commit conflicts.

rdblue · 2021-11-03T16:22:17Z

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

+
+  ParallelIterable(Iterable<? extends Iterable<T>> iterables,
+                          ExecutorService workerPool,
+                          int workerPoolSize) {


Can you fix the indentation here?

rdblue · 2021-11-03T16:23:40Z

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java

+ * Run iterables in parallel.
+ * @deprecated please use {@link CloseableIterable#combine(Iterable, ExecutorService, int)} instead.
+ */
+@Deprecated


This class is not deprecated. It is still public and will not be removed. This should be created using CloseableIterable#combine rather than directly using the constructor. Can you remove this deprecation?

I see the module here also changed. I don't think we are allowed to move public class to another module?

Good catch, @jackye1995!

rdblue · 2021-11-03T16:30:49Z

core/src/main/java/org/apache/iceberg/SystemProperties.java

+   */
+  public static final String READ_DELETE_FILES_WORKER_POOL_SIZE = "iceberg.worker.read-deletes-num-threads";
+
+  public static boolean getBoolean(String systemProperty, boolean defaultValue) {


As I said elsewhere, I don't think that this feature should be controlled through a system property. This should be a Flin-specific property for now and we can introduce a similar config for Spark later. But since this violates Spark's threading model on executors, we don't want to make this global.

rdblue · 2021-11-03T16:33:07Z

Does this comment need to be addressed for ORC? https://github.com/apache/iceberg/pull/3120/files#diff-a6641d31cdfd66835b3447bef04be87786849126b07761e47b852837f67a988aR241

rdblue · 2021-11-03T16:35:54Z

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

+      SystemProperties.READ_DELETE_FILES_WORKER_POOL_SIZE, READ_DELETES_WORKER_POOL_SIZE_DEFAULT);
+  private static final ExecutorService READ_DELETES_SERVICE = READ_DELETES_WORKER_POOL_SIZE <= 1 ? null :
+      MoreExecutors.getExitingExecutorService((ThreadPoolExecutor) Executors.newFixedThreadPool(
+          READ_DELETES_WORKER_POOL_SIZE, new ThreadFactoryBuilder().setNameFormat("Read-delete-Service-%d").build()));


It looks like the changes other than the ones in this file are to make it easier to use ParallelIterable. Can we separate those changes from these for Flink and DeleteFilter? A separate PR for the CloseableIterable changes would make it easier to review this when there are more Flink changes to handle the executor service and pass it into DeleteFilter.

Reo-LEI · 2021-11-08T06:05:09Z

@rdblue @stevenzwu @kbendick @jackye1995 Thanks every one for review! As Ryan's comment, current implementation will violates Spark's threading model. And since I start work on #3323, I think we should improve the read performance by rewrite data files and delete files in time, maybe this PR is not necessary. I will do some test after I finish #3323 and see how much benefit this optimization can bring, and then consider whether we still need this optimization. Thanks every one's work again!

github-actions · 2024-07-22T00:14:00Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the [email protected] list. Thank you for your contributions.

github-actions · 2024-07-30T00:13:10Z

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

Reo-LEI added 2 commits September 13, 2021 19:34

Core: read delete file in parallel.

5521a34

Fix checkstyle.

5c4f33d

github-actions bot added API core data labels Sep 15, 2021

stevenzwu reviewed Sep 15, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/util/ThreadPools.java Outdated Show resolved Hide resolved

stevenzwu reviewed Sep 15, 2021

View reviewed changes

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

stevenzwu reviewed Sep 15, 2021

View reviewed changes

rdblue reviewed Sep 15, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 15, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 15, 2021

View reviewed changes

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

Reo-LEI and others added 2 commits September 16, 2021 16:03

Merge branch 'apache:master' into core-read-delete-file-in-parallel

288da5e

Addressing comments.

57f858a

kbendick reviewed Sep 17, 2021

View reviewed changes

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

rdblue reviewed Sep 19, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 19, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 19, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 19, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java Show resolved Hide resolved

rdblue reviewed Sep 19, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/util/ThreadPools.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 19, 2021

View reviewed changes

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

rdblue reviewed Sep 19, 2021

View reviewed changes

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

Make read deletes parallelize in configurable and use a separate thre…

9203a2a

…ad pool to run.

github-actions bot added flink MR spark labels Sep 21, 2021

Reo-LEI and others added 3 commits September 21, 2021 17:24

Fix unittest.

7382b5e

Fix unittest.

b17a1fd

Merge branch 'apache:master' into core-read-delete-file-in-parallel

944d6ac

rdblue reviewed Sep 27, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java Show resolved Hide resolved

rdblue reviewed Sep 27, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java Show resolved Hide resolved

rdblue reviewed Sep 27, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/io/ParallelIterable.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 27, 2021

View reviewed changes

rdblue reviewed Sep 28, 2021

View reviewed changes

Reo-LEI and others added 6 commits September 30, 2021 19:17

Addressing the comments.

5da4cf1

Merge remote-tracking branch 'community/master' into core-read-delete…

f5692a2

…-file-in-parallel

Config read deletes service by system property.

f53b20e

Merge branch 'apache:master' into core-read-delete-file-in-parallel

0c812ef

Fix checkstyle.

9b186c6

Merge branch 'core-read-delete-file-in-parallel' of https://github.co…

9a3c259

…m/Reo-LEI/iceberg into core-read-delete-file-in-parallel

Reo-LEI and others added 2 commits November 1, 2021 21:02

Fix checkstyle.

004511f

Merge branch 'apache:master' into core-read-delete-file-in-parallel

40dfec8

rdblue reviewed Nov 3, 2021

View reviewed changes

github-actions bot added the stale label Jul 22, 2024

github-actions bot closed this Jul 30, 2024

Core: read delete files in parallel #3120

Core: read delete files in parallel #3120

Uh oh!

Conversation

Reo-LEI commented Sep 15, 2021

Uh oh!

Reo-LEI commented Sep 15, 2021

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reo-LEI commented Sep 22, 2021

Uh oh!

rdblue commented Sep 27, 2021

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reo-LEI Sep 30, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reo-LEI commented Nov 1, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reo-LEI Sep 30, 2021 •

edited

Loading