API, Core, Spark:Add file groups failure in rewrite result #7361

waltczhang · 2023-04-17T09:45:34Z

Background:
A table with schema (int id, timestamp ts), When configured 'partial-progress.enabled = true' in rewrite job , the job will return success but in fact rewrite failed, so in order to solve this kind of problem, rewrite result should return group failure information.

aokolnychyi · 2023-04-18T05:28:00Z

@szehon-ho and I should be able to review this at some point

nastra · 2023-04-18T08:37:42Z

.palantir/revapi.yml

    - code: "java.field.removedWithConstant"
      old: "field org.apache.iceberg.TableProperties.HMS_TABLE_OWNER"
      justification: "Removing deprecations for 1.3.0"
+    - code: "java.method.numberOfParametersChanged"


we typically don't want to introduce breaking API changes that haven't gone through a deprecation cycle first. See https://iceberg.apache.org/contribute/#minor-version-deprecations-required for some details.

Thanks nastra, It is not reasonable to modify here, I will fix it.

@waltczhang can you please revert all changes to the revapi.yml? Those shouldn't be necessary anymore once my other comments are applied

nastra · 2023-04-18T08:41:27Z

core/src/main/java/org/apache/iceberg/actions/RewriteFileGroup.java

  private Set<DataFile> addedFiles = Collections.emptySet();

-  public RewriteFileGroup(FileGroupInfo info, List<FileScanTask> fileScanTasks) {
+  public RewriteFileGroup(FileGroupInfo info, List<FileScanTask> fileScanTasks, boolean isFailed) {


do we really need to pass a boolean here or could we derive that info maybe in a different way?

It is better to set the default, thanks nastra.

szehon-ho

Hi, I think its a good improvement suggestion, I had some implementation comments

szehon-ho · 2023-04-21T21:54:33Z

api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java

      return rewriteResults().stream().mapToLong(FileGroupRewriteResult::rewrittenBytesCount).sum();
    }
+
+    @Value.Default


Instead of re-using the FileGroupRewriteResult, would it make more sense to have a FailedFileGroupRewriteResult, with just failure?

I dont think the bytesCount/fileCount make too much sense on failure, and we can also put the exception there? cc @aokolnychyi @RussellSpitzer for thoughts

I think it is ok that have a FailedFileGroupRewriteResult. What about the following data structure?
FailedFileGroupRewriteResult {
FileGroupInfo info();
boolean isFailedGroup()
}

Not sure why we would need a "boolean" wouldn't it be this class because it failed?

I think we should probably have result have two new interfaces, something like

List<FileGroupFailureResult> rewriteFailures(); default long failedGroupCount() { return rewriteFailures().size(); } default long failedDataFilesRewrites() { return rewriteFailures().stream().mapToInt(FailedFileGroup::filescount).sum() }

Then we won't mix failures and successes in the same list

Thanks @RussellSpitzer reply,
Yes, "boolean" is redundant, They are great advice, I wiil fix them according to your advice.

szehon-ho · 2023-04-21T21:59:22Z

...k/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java

        .onFailure(
-            (fileGroup, exception) ->
-                LOG.error("Failure during rewrite group {}", fileGroup.info(), exception))
+            (fileGroup, exception) -> {


I'm wondering if we can't avoid touching the commitService. It doesnt seem like its really the responsibility of the commit service to handle the failed ones. Can't this method just keep a list internally? ie,

List<Failure> failures; ... onFailure( (fileGroup, exception) -> { LOG.error("...") failures.addFailure(fileGroup, exception) }

Yes, it is ok that keep a list internally, According to the comments above , it should be like this :

List< FailedFileGroupRewriteResult > failures; ... onFailure( (fileGroup, exception) -> { LOG.error("...") failures.addFailure(fileGroup, exception) } ... return ImmutableRewriteDataFiles.Result.builder() .rewriteResults(rewriteResults) .failedRewriteREsults(failures) .build();

RussellSpitzer · 2023-04-25T19:39:26Z

core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java

      commitReadyCommitGroups();
    }

+    /**


This method confuses me a bit,

It's called failedRewrite but sounds like its only for a very specific case of failure

It will never actually succeed since if it is called when service is closed it will fail the precondition and throw an exception

Thanks RussellSpitzer,
It has been modified here, using szehon-ho's suggestion above.

RussellSpitzer · 2023-04-25T19:42:17Z

core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java

          "Cannot get results from a service which has not been closed");
-      return committedRewrites;
+      results.addAll(committedRewrites);
+      results.addAll(failedRewrites);


Not sure we should mix these in the same list? Maybe we should split this into just two separate lists (or methods)?

Results() ++
Failures() ?

Thanks RussellSpitzer,
It has been modified here.

nastra

We can solve this without breaking the API. I've added a few comments on how to achieve that

nastra · 2023-05-10T08:38:46Z

.palantir/revapi.yml

    - code: "java.field.removedWithConstant"
      old: "field org.apache.iceberg.TableProperties.HMS_TABLE_OWNER"
      justification: "Removing deprecations for 1.3.0"
+    - code: "java.method.numberOfParametersChanged"


@waltczhang can you please revert all changes to the revapi.yml? Those shouldn't be necessary anymore once my other comments are applied

nastra · 2023-05-10T08:47:22Z

api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java

  interface Result {
    List<FileGroupRewriteResult> rewriteResults();

+    List<FileGroupFailureResult> rewriteFailures();


to avoid introducing API-breaking changes, what you'd rather want to do is

@Value.Default default List<FileGroupFailureResult> rewriteFailures() { return ImmutableList.of(); }

Thanks for your tips, I know how to do and modify.

api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java

nastra · 2023-05-10T08:51:21Z

core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesResult.java

  private final List<FileGroupRewriteResult> rewriteResults;

-  public BaseRewriteDataFilesResult(List<FileGroupRewriteResult> rewriteResults) {
+  private final List<FileGroupFailureResult> rewriteFailures;


this class is deprecated and not used anymore, so no need to add changes to it. Not modifying this class also ensures that no API-breaking changes are introduced (as indicated by RevAPI)

nastra · 2023-05-10T08:56:14Z

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java

    RewriteDataFiles.Result result = spyRewrite.execute();

    Assert.assertEquals("Should have 7 fileGroups", result.rewriteResults().size(), 7);
+    Assert.assertEquals("Should have 3 failed fileGroups", result.rewriteFailures().size(), 3);


I think it's better to change this to

assertThat(result.rewriteResults()).hasSize(7); assertThat(result.rewriteFailures()).hasSize(3); assertThat(result.failedDataFilesCount()).isEqualTo(6);

the advantage here is that the content of result.rewriteResults() / result.rewriteFailures() will be shown if the assertion ever fails, which makes debugging a lot easier.

Could you please do the same changes further below in L801ff

Thank you for your patience in explaining, I got it.

nastra · 2023-05-10T15:10:52Z

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java

    RewriteDataFiles.Result result = spyRewrite.execute();

    Assert.assertEquals("Should have 7 fileGroups", result.rewriteResults().size(), 7);
+    assertThat(result.rewriteFailures()).hasSize(3);


nit: could you also update the line above please?

nastra · 2023-05-10T15:11:00Z

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java

    RewriteDataFiles.Result result = spyRewrite.execute();

    Assert.assertEquals("Should have 7 fileGroups", result.rewriteResults().size(), 7);
+    assertThat(result.rewriteFailures()).hasSize(3);


nit: could you also update the line above please?

nastra

I've mainly been focusing on the API pieces and those LGTM, thanks @waltczhang

RussellSpitzer · 2023-05-10T15:26:42Z

...k/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java

        commitManager.service(groupsPerCommit);
    commitService.start();

+    List<FileGroupFailureResult> rewriteFailures = Lists.newArrayList();


Race Condition here, need to use a concurrent structure.

The code below adds to this arraylist in a concurrent manner, if multiple groups fail at the same time we will end up call "add" simultaneously which will have some unexpected behavior.

Thanks RussellSpitzer for pointing out this problem.

RussellSpitzer

Overall looking pretty good now
Just need to finish cleaning up

The race condition
Eduard's comments on the tests

Let me know when that's all done and we can go to merge

RussellSpitzer · 2023-05-11T21:35:58Z

...k/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java

        commitManager.service(groupsPerCommit);
    commitService.start();

+    List<FileGroupFailureResult> rewriteFailures = Lists.newCopyOnWriteArrayList();


In general this would probably be an expensive data-structure for this operation since we never traverse. I don't think this is ever going to be that huge, but let's use something a bit more efficient.

I think something like
ConcurrentLinkedQueue would be fine here, and then we could just create a list out of it when we are done

Just to note, I don't think we actually would have performance issues with copy on write array list, it just seems a bit wasteful to me in this context.

After reading the introduction of newCopyOnWriteArrayList and ConcurrentLinkedQueue, it is true that the latter is more efficient, thank you.

… ConcurrentLinkedQueue

RussellSpitzer · 2023-05-12T14:35:39Z

LGTM, will merge after tests pass

waltczhang · 2023-05-16T02:06:34Z

I noticed that the unit tests have all passed. Could you please merge them when you have time? Thanks.
@RussellSpitzer @nastra @aokolnychyi @szehon-ho

nastra · 2023-05-16T05:27:10Z

Merging, since we have 2 approving reviews and CI passed. Thanks @waltczhang.

github-actions bot added API core spark labels Apr 17, 2023

waltczhang changed the title ~~Add failed group info in rewrite result~~ API, Core, Spark:Add failed group info in rewrite result Apr 17, 2023

nastra reviewed Apr 18, 2023

View reviewed changes

waltczhang closed this Apr 19, 2023

waltczhang reopened this Apr 19, 2023

szehon-ho reviewed Apr 21, 2023

View reviewed changes

RussellSpitzer reviewed Apr 25, 2023

View reviewed changes

Add file groups failure in rewrite result

945a548

waltczhang force-pushed the add_failed_group_info_in_rewrite_result branch from 8ee0aba to 945a548 Compare May 6, 2023 03:35

waltczhang changed the title ~~API, Core, Spark:Add failed group info in rewrite result~~ API, Core, Spark:Add file groups failure in rewrite result May 6, 2023

Add import FileGroupFailureResult in BaseRewriteDataFilesResult

d75b98c

nastra requested changes May 10, 2023

View reviewed changes

waltczhang added 2 commits May 10, 2023 21:52

Delete api info in revapi.yml

1f31e5b

Delete api info in revapi.yml

6d59eff

nastra reviewed May 10, 2023

View reviewed changes

nastra approved these changes May 10, 2023

View reviewed changes

RussellSpitzer reviewed May 10, 2023

View reviewed changes

RussellSpitzer approved these changes May 10, 2023

View reviewed changes

Change rewriteFailures to concurrent list

a577e01

RussellSpitzer reviewed May 11, 2023

View reviewed changes

Change rewriteFailures's data-structure of newCopyOnWriteArrayList to…

d7c3749

… ConcurrentLinkedQueue

nastra merged commit 42cf157 into apache:master May 16, 2023

szehon-ho mentioned this pull request May 23, 2023

Spark 3.3: Harmonize RewriteDataFilesSparkAction #7676

Merged

API, Core, Spark:Add file groups failure in rewrite result #7361

API, Core, Spark:Add file groups failure in rewrite result #7361

Uh oh!

Conversation

waltczhang commented Apr 17, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aokolnychyi commented Apr 18, 2023

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!

szehon-ho left a comment

Choose a reason for hiding this comment

Uh oh!

szehon-ho Apr 21, 2023 • 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!

RussellSpitzer Apr 25, 2023 • 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!

waltczhang Apr 23, 2023 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nastra left a comment

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!

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!

nastra left a comment

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer May 10, 2023 • 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!

RussellSpitzer left a comment

waltczhang commented Apr 17, 2023 •

edited

Loading

szehon-ho Apr 21, 2023 •

edited

Loading

RussellSpitzer Apr 25, 2023 •

edited

Loading

waltczhang Apr 23, 2023 •

edited

Loading

RussellSpitzer May 10, 2023 •

edited

Loading