[Java IO] Add ArrowFlight IO connector by bvolpato · Pull Request #37904 · apache/beam

bvolpato · 2026-03-22T04:17:59Z

Summary

Add a new I/O connector for Apache Arrow Flight, enabling high-performance data transfer over gRPC using the Arrow columnar format. Arrow Flight avoids serialization/deserialization overhead and can achieve 10-50x faster data transfer compared to JDBC/ODBC.

Fixes #20116

Changes

ArrowFlightIO.Read — BoundedSource-based read transform that fetches Arrow Flight schemas via getInfo, splits work by FlightEndpoint, and converts record batches to Beam Row using the existing ArrowConversion extension.
ArrowFlightIO.Write — DoFn-based write transform that buffers Beam Rows into configurable batches, converts them to Arrow vectors, and streams them to a Flight server via doPut.
Endpoint-level split parallelism — each FlightEndpoint returned by the server becomes a separate sub-source for parallel reads.
Bearer token authentication — optional token support via withToken().
Build integration — registered flight-core and flight-sql artifacts in BeamModulePlugin.groovy; added the :sdks:java:io:arrow-flight module to settings.gradle.kts.
Documentation — added ArrowFlightIO to CHANGES.md (2.73.0) and the built-in connectors table in connectors.md.

Testing

Tests use a real in-process FlightServer with a TestFlightProducer:

testRead() — starts a Flight server serving two rows, reads them with ArrowFlightIO.read(), and asserts the output via PAssert.
testWrite() — writes two rows with ArrowFlightIO.write() and verifies the producer received exactly 2 records.

Both tests run the full pipeline end-to-end using TestPipeline and the DirectRunner.

Did this cause any problems?

Rollback/revert this commit and redeploy the service.

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.

gemini-code-assist · 2026-03-22T04:18:21Z

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a new Java I/O connector for Apache Arrow Flight, significantly enhancing data transfer capabilities within Beam pipelines. The connector leverages Arrow Flight's high-performance RPC framework to facilitate efficient reading from and writing to Arrow Flight servers, minimizing serialization overhead and enabling faster data movement. This integration expands Beam's ecosystem by providing a robust solution for interacting with various Arrow Flight-compatible data systems.

Highlights

ArrowFlightIO.Read: Implemented a BoundedSource-based read transform that fetches Arrow Flight schemas, splits work by FlightEndpoint, and converts record batches to Beam Rows using the existing ArrowConversion extension.
ArrowFlightIO.Write: Developed a DoFn-based write transform that buffers Beam Rows into configurable batches, converts them to Arrow vectors, and streams them to a Flight server via doPut.
Endpoint-level Split Parallelism: Enabled each FlightEndpoint returned by the server to become a separate sub-source for parallel reads, enhancing data ingestion efficiency.
Bearer Token Authentication: Added optional support for bearer token authentication via withToken() for secure connections.
Build Integration: Registered flight-core and flight-sql artifacts in BeamModulePlugin.groovy and added the new :sdks:java:io:arrow-flight module to settings.gradle.kts.
Documentation: Updated CHANGES.md and the built-in connectors table in connectors.md to reflect the addition of the ArrowFlightIO connector.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

github-actions · 2026-03-22T23:10:56Z

Assigning reviewers:

R: @chamikaramj for label java.
R: @damccorm for label build.
R: @shunping for label website.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

Abacn

Thanks for the new IO! Had an initial comment to kick off

Add a new IO connector for Apache Arrow Flight, enabling high-performance data transfer over gRPC using the Arrow columnar format. Includes read (BoundedSource) and write (DoFn with doPut) support with endpoint-level split parallelism and bearer token authentication. Fixes apache#20116

- Register :sdks:java:io:arrow-flight in javaioPreCommit task (build.gradle.kts) - Add --add-opens JVM arg for Arrow native memory on JDK 17+ - Make host() nullable at AutoValue level to fix factory method NPE - Eagerly materialize rows from Arrow buffers to prevent stale access - Move root.setRowCount() after vector population for correct ordering - Use AtomicInteger for thread-safe write record counting in tests

bvolpato · 2026-04-12T00:11:44Z

Run Java_Amazon-Web-Services2_IO_Direct PreCommit

github-actions · 2026-04-19T12:25:12Z

Reminder, please take a look at this pr: @chamikaramj @damccorm @shunping

Abacn · 2026-04-20T20:33:08Z

+          break;
+        default:
+          throw new IllegalArgumentException(
+              String.format(


Can we add a TODO here about supporting aggregation and logical types? Currently supported types are pretty limited

Abacn · 2026-04-20T20:42:00Z

+}
+
+test {
+  jvmArgs '--add-opens=java.base/java.nio=ALL-UNNAMED'


I understsnd add-opens are needed:

Caused by: java.lang.ExceptionInInitializerError: Exception java.lang.RuntimeException: Failed to initialize MemoryUtil. You must start Java with `--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED` (See https://arrow.apache.org/docs/java/install.html) [in thread "direct-runner-worker"] at org.apache.arrow.memory.util.MemoryUtil.<clinit>(MemoryUtil.java:140)

this means using ArrowFlightIO one needs to set --JdkAddOpenModules=java.base/java.nio=ALL-UNNAMED pipeline options in Java17+. We should note this here and in ArrowFlightIO Javadoc

Also, Java8 runtime doesn't recognize '--add-opens' pipeline options. However we'll soon remove Java8 support so it's fine here.

Abacn · 2026-04-20T20:53:01Z

+      Read spec = source.spec;
+
+      String hostName = checkNotNull(spec.host(), "host");
+      if (source.endpoint != null) {


A potential issue arises when a source (with endPoint=null) first get read (from endpoint[0]) then split is called on getSource(), it will produce a list of sources for each available endpoint, because source.endpoint=null here and source.split() will unconditionally split. Would this result in any issue like duplicate records get read?

Note that source are usually split before read, with a few exceptions (e.g. BoundedReadFromUnbounded fallback scenario)

Abacn · 2026-04-20T21:10:25Z

+          return ArrowType.Binary.INSTANCE;
+        case DATETIME:
+          return new ArrowType.Timestamp(TimeUnit.MILLISECOND, "UTC");
+        default:


same here, add TODO for aggregation and logical types

Abacn · 2026-04-20T21:12:59Z

+    }
+
+    @Override
+    public PDone expand(PCollection<Row> input) {


PDone is discouraged. Please add a TODO about implementing dead letter queue (preferred) or WriteWithResult

Abacn · 2026-04-20T21:14:05Z

+      return new Field(beamField.getName(), fieldType, Collections.emptyList());
+    }
+
+    private ArrowType beamTypeToArrowType(Schema.FieldType beamType) {


Consider move Beam schema->Arrow Schema to ArrowConversion as a public API (the inversion of toBeamSchema)

github-actions · 2026-04-23T12:37:06Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @Abacn for label build.
R: @kennknowles for label website.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

# Conflicts: # CHANGES.md

bvolpato · 2026-05-05T09:05:07Z

[review-prs-verification]
Head: 3e4fe51
Ready to merge: no

Files reviewed:

CHANGES.md
build.gradle.kts
buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
sdks/java/io/amazon-web-services2/src/test/java/org/apache/beam/sdk/io/aws2/sqs/SqsIOWriteBatchesTest.java
sdks/java/io/arrow-flight/build.gradle
sdks/java/io/arrow-flight/src/main/java/org/apache/beam/sdk/io/arrowflight/ArrowFlightIO.java
sdks/java/io/arrow-flight/src/main/java/org/apache/beam/sdk/io/arrowflight/package-info.java
sdks/java/io/arrow-flight/src/test/java/org/apache/beam/sdk/io/arrowflight/ArrowFlightIOTest.java
settings.gradle.kts
website/www/site/content/en/documentation/io/connectors.md

Tests/checks run and results:

JAVA_HOME=$HOME/.sdkman/candidates/java/21.0.6-tem ./gradlew --no-daemon --console=plain :sdks:java:io:arrow-flight:spotlessCheck :sdks:java:io:arrow-flight:test :sdks:java:io:amazon-web-services2:test --tests org.apache.beam.sdk.io.aws2.sqs.SqsIOWriteBatchesTest ✅ before rebase
Same targeted Gradle command ✅ after rebase/merge conflict resolution
GitHub checks snapshot as of 2026-05-05T09:02:32Z: 24 success, 1 neutral, 1 in progress, 0 failures
Remaining in-progress check: beam_PreCommit_Java_GCP_IO_Direct (Run Java_GCP_IO_Direct PreCommit)
gh pr checks --required reports no required checks for this branch

Fixes committed/pushed:

Merged origin/master into bvolpato/arrow-flight-io
Resolved CHANGES.md conflict by keeping both ArrowFlightIO and upstream IcebergIO release-note entries
Pushed merge commit 3e4fe5100645397449b062ceebd3aeaa218162d6 to bvolpato/arrow-flight-io

Final CI status:

Not final yet: one optional Beam precommit check is still running and has not exposed logs while in progress.
PR is now mergeable/clean from GitHub mergeability perspective, but status is unstable while that check runs.

Merge/close result:

Not merged or closed from this pass because final optional CI is still pending.

bvolpato · 2026-05-05T09:36:13Z

[review-prs-verification]
Ready to merge: yes

Verified after resolving the master merge conflict and pushing 3e4fe5100645397449b062ceebd3aeaa218162d6:

./gradlew --no-daemon --console=plain :sdks:java:io:arrow-flight:spotlessCheck :sdks:java:io:arrow-flight:test :sdks:java:io:amazon-web-services2:test --tests org.apache.beam.sdk.io.aws2.sqs.SqsIOWriteBatchesTest
GitHub checks on PR head: 25 success, 1 neutral, 0 pending, 0 failing.

Notes:

Conflict was only in CHANGES.md; kept ArrowFlight entry plus upstream Iceberg entries.

github-actions Bot added java build website io labels Mar 22, 2026

bvolpato force-pushed the bvolpato/arrow-flight-io branch 7 times, most recently from b384457 to 1b3518d Compare March 22, 2026 21:20

bvolpato marked this pull request as ready for review March 22, 2026 21:43

github-actions Bot added the Next Action: Reviewers label Mar 22, 2026

Abacn reviewed Mar 26, 2026

View reviewed changes

Comment thread settings.gradle.kts

bvolpato force-pushed the bvolpato/arrow-flight-io branch from e1e26fd to d0551f0 Compare April 2, 2026 13:09

bvolpato added 2 commits April 5, 2026 22:22

bvolpato force-pushed the bvolpato/arrow-flight-io branch from d0551f0 to 7252108 Compare April 6, 2026 02:22

[Java IO] Harden ArrowFlightIO writes

8f4a816

[Java IO] Stabilize SqsIO timeout batch tests

3af9937

github-actions Bot added aws and removed aws labels Apr 12, 2026

github-actions Bot added the slow-review label Apr 19, 2026

Abacn reviewed Apr 20, 2026

View reviewed changes

github-actions Bot removed the slow-review label Apr 23, 2026

github-actions Bot added the reassigned-reviewers label Apr 23, 2026

Abacn added Next Action: Author and removed Next Action: Reviewers labels Apr 23, 2026

Merge remote-tracking branch 'origin/master' into codex-review-pr-37904

3e4fe51

# Conflicts: # CHANGES.md

github-actions Bot added aws Next Action: Reviewers and removed aws Next Action: Author labels May 5, 2026

Conversation

bvolpato commented Mar 22, 2026

Summary

Changes

Testing

Did this cause any problems?

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist Bot commented Mar 22, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

github-actions Bot commented Mar 22, 2026

Uh oh!

Abacn left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

bvolpato commented Apr 12, 2026

Uh oh!

github-actions Bot commented Apr 19, 2026

Uh oh!

Abacn Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

Abacn Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Apr 23, 2026

Uh oh!

bvolpato commented May 5, 2026

Uh oh!

bvolpato commented May 5, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants