Re-add iceberg bounded source; test splitting #30805

kennknowles · 2024-03-29T19:36:35Z

Currently we have failing in the exhaustive splitting, but I would also like feedback early.

This is exactly #30797 plus one commit, so if you click on just the last commit you should be able to see just the read/source diff

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

kennknowles · 2024-03-29T19:36:41Z

R: @chamikaramj

github-actions · 2024-03-29T19:37:48Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

kennknowles · 2024-03-29T19:54:30Z

The failure is in the basic reading of data - none is read. I'll try to grok things and see about that. Still interested in commentary on the approaches here.

chamikaramj

Thanks!

chamikaramj · 2024-04-01T18:16:55Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/IcebergIO.java

+    };
+  }
+
+  public static class Write<ElementT, DestinationT>


Noting that this also has Write stuff but I'm ignoring those for this review.

Thanks for noticing. I will revert any added write stuff in this PR.

chamikaramj · 2024-04-01T18:18:34Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/IcebergIO.java

+import org.apache.iceberg.data.Record;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+public class IcebergIO {


Seems like we are missing the Read transform ?

Will do soon. I realize all the tests are a vanilla Read.from(IcebergBoundedSource)

chamikaramj · 2024-04-01T18:20:51Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/IcebergBoundedSource.java

+import org.apache.iceberg.io.CloseableIterable;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+public class IcebergBoundedSource extends BoundedSource<Row> {


Ideally we should use SDF but I think it would suffice to add a TODO to convert this to an SDF in the future.

chamikaramj · 2024-04-01T23:59:31Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/IcebergBoundedSource.java

+        TableScan tableScan = table().newScan();
+
+        if (desiredBundleSizeBytes > 0) {
+          tableScan = tableScan.option(TableProperties.SPLIT_SIZE, "" + desiredBundleSizeBytes);


String.valueOf(desiredBundleSizeBytes)

chamikaramj · 2024-04-02T00:20:00Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/IcebergBoundedSource.java

+        }
+        break;
+      case BATCH:
+        // TODO: Add batch scan


We should fail here to prevent data loss.

chamikaramj · 2024-04-02T00:49:23Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/CombinedScanReader.java

+    InputFilesDecryptor decryptor =
+        checkStateNotNull(this.decryptor, "decryptor null in adance() - did you call start()?");
+
+    // This is a lie, but the most expedient way to work with IcebergIO's


Update/remove comment ?

Actually I added this comment to explain why the type was fake. I reworded it.

chamikaramj · 2024-04-02T00:54:26Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/CombinedScanReader.java

+      @Nullable FileScanTask fileTask = null;
+      while (!files.isEmpty() && fileTask == null) {
+        fileTask = files.remove();
+        if (fileTask.isDataTask()) {


Why are we skipping data tasks ? Was this supposed to be if (!fileTask.isDataTask()) ?

Seems like DataTasks contain actual data: https://iceberg.apache.org/javadoc/0.11.0/org/apache/iceberg/DataTask.html

I think this was obsolete code. The issue is that ScanTask is a oneof and this is checking a oneof that it is not.

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/CombinedScanReader.java

chamikaramj · 2024-04-02T00:59:07Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/CombinedScanReader.java

+  private Row convert(Record record) {
+    Row.Builder b = Row.withSchema(schema);
+    for (int i = 0; i < schema.getFieldCount(); i++) {
+      // TODO: A lot obviously


What does this mean ?

Haha it means that this was a fake implementation. I added conversions.

codecov-commenter · 2024-04-09T21:19:20Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 71.47%. Comparing base (b018c25) to head (21e0a47).
Report is 19 commits behind head on master.

❗ Current head 21e0a47 differs from pull request most recent head a169e6a. Consider uploading reports for the commit a169e6a to get more accurate results

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #30805      +/-   ##
============================================
+ Coverage     70.96%   71.47%   +0.51%     
============================================
  Files          1257      710     -547     
  Lines        140931   104815   -36116     
  Branches       4307        0    -4307     
============================================
- Hits         100007    74915   -25092     
+ Misses        37444    28268    -9176     
+ Partials       3480     1632    -1848

Flag	Coverage Δ
java	`?`

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

kennknowles · 2024-04-10T01:11:46Z

Looks like there is something broke in the GHA workflow FYI just so you don't wait for that to go green. I pushed some possible fixes up for that too but I'm not sure if in this case the workflow comes from master or the PR.

chamikaramj

Thanks.

chamikaramj · 2024-04-10T15:46:49Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/IcebergBoundedSource.java

+      case BATCH:
+        throw new UnsupportedOperationException("BATCH scan not supported");
+    }
+    return splits;


What will happen if the sources produced at line 104 above get re-split by the runner ?

If such sources cannot be re-split, we should have a trivial case where we just return the original source to prevent data loss.

chamikaramj · 2024-04-10T15:55:49Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/CombinedScanReader.java

+  }
+
+  @Override
+  public BoundedSource<Row> getCurrentSource() {


Is it possible to also implement getFractionConsumed to support progress reporting in a meaningful way ?

I think it will be very useful for autoscaling when using this with Dataflow.

This can be in a future PR since we want to get this in by release cut.

chamikaramj · 2024-04-10T16:05:03Z

sdks/java/io/iceberg/src/test/java/org/apache/beam/io/iceberg/IcebergIOReadTest.java

+
+    testPipeline.run();
+  }
+


Let's also add a test for splitting using SourceTestUtils.assertSourcesEqualReferenceSource.

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/SchemaAndRowConversions.java

chamikaramj · 2024-04-12T01:04:20Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/ScanSource.java

+        }
+        break;
+      case BATCH:
+        throw new UnsupportedOperationException("BATCH scan not supported");


Also fail for the default path prevent dataloss (or just return the original source if that can be read directly).

chamikaramj · 2024-04-12T01:09:15Z

LGTM other than handling the re-splitting case above.

kennknowles

Added a test for double-splitting. To make splitting behavior clearer, I factored into two different kinds of sources.

kennknowles · 2024-04-12T14:02:05Z

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/ScanSource.java

+        }
+        break;
+      case BATCH:
+        throw new UnsupportedOperationException("BATCH scan not supported");


chamikaramj

Thanks. LGTM.

github-actions bot added java build io labels Mar 29, 2024

chamikaramj self-requested a review April 1, 2024 17:48

chamikaramj reviewed Apr 2, 2024

View reviewed changes

kennknowles force-pushed the iceberg-read branch from 21e0a47 to 177aea2 Compare April 9, 2024 21:18

github-actions bot removed the build label Apr 9, 2024

kennknowles force-pushed the iceberg-read branch from 177aea2 to e26cacd Compare April 9, 2024 23:11

kennknowles marked this pull request as ready for review April 10, 2024 01:03

github-actions bot added the build label Apr 10, 2024

kennknowles force-pushed the iceberg-read branch 2 times, most recently from b53e022 to 0b5e391 Compare April 10, 2024 01:10

kennknowles force-pushed the iceberg-read branch from 0b5e391 to 37b23f7 Compare April 10, 2024 16:02

chamikaramj reviewed Apr 10, 2024

View reviewed changes

kennknowles force-pushed the iceberg-read branch 4 times, most recently from 19037fb to a5b995a Compare April 11, 2024 18:05

github-actions bot removed the build label Apr 11, 2024

kennknowles added 2 commits April 11, 2024 14:35

Fix iceberg dependencies

631f6e7

Refactor iceberg/beam conversions for clarity

36f3228

chamikaramj reviewed Apr 12, 2024

View reviewed changes

Re-add Iceberg bounded source; test splitting

a169e6a

kennknowles force-pushed the iceberg-read branch from a5b995a to a169e6a Compare April 12, 2024 15:22

kennknowles commented Apr 12, 2024

View reviewed changes

chamikaramj approved these changes Apr 12, 2024

View reviewed changes

kennknowles merged commit 96dc16a into apache:master Apr 12, 2024
23 checks passed

kennknowles deleted the iceberg-read branch April 13, 2024 00:56

ahmedabu98 mentioned this pull request Apr 15, 2024

Managed Transform protos & translation; Iceberg SchemaTransforms & translation #30910

Merged

Re-add iceberg bounded source; test splitting #30805

Re-add iceberg bounded source; test splitting #30805

Conversation

kennknowles commented Mar 29, 2024

GitHub Actions Tests Status (on master branch)

kennknowles commented Mar 29, 2024

github-actions bot commented Mar 29, 2024

kennknowles commented Mar 29, 2024

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Apr 9, 2024 • edited

Codecov Report

kennknowles commented Apr 10, 2024

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj commented Apr 12, 2024

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

codecov-commenter commented Apr 9, 2024 •

edited