Report source lineage from HadoopFormatIO

[shnapz]

This PR adds source lineage reporting to HadoopFormatIO for file-based sources.

Background

While FileBasedSource already reports source lineage metrics, HadoopFormatIO did not. HadoopFormatIO is a more generic IO transform that supports various data sources including databases, NoSQL stores, and file systems through Hadoop's InputFormat/OutputFormat interfaces.

Changes

• Added lineage reporting to HadoopFormatIO: Implements source lineage reporting for file-based splits (when InputSplit is an instance of FileSplit)
• Refactored shared logic: Extracted common lineage reporting logic into FileSystems.reportSourceLineage(List<ResourceId>) to reduce code duplication
• Updated FileBasedSource: Refactored to use the new shared utility method

Implementation Details

The implementation reports lineage only for file-based sources detected via FileSplit instances. Non-file sources (e.g., database reads via DBInputFormat) are intentionally excluded as they don't represent file resources.

---

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]

Summary of Changes

Hello @shnapz, 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 enhances Apache Beam's data lineage capabilities by extending source lineage reporting to HadoopFormatIO for file-based inputs. Previously, only FileBasedSource provided this functionality. The change introduces a centralized utility method for reporting lineage from multiple ResourceIds, which is then adopted by both HadoopFormatIO and FileBasedSource to ensure consistent and efficient lineage tracking across file-based data sources. This improvement helps users better understand the origins of their data within Beam pipelines.

Highlights

• Source Lineage Reporting for HadoopFormatIO: Implemented source lineage reporting for file-based sources within HadoopFormatIO, specifically for InputSplit instances that are FileSplit.
• Refactored Lineage Logic: Extracted common lineage reporting logic into a new shared utility method FileSystems.reportSourceLineage(List<ResourceId>) to reduce code duplication and centralize the logic for handling multiple resources.
• Updated FileBasedSource: Refactored FileBasedSource to utilize the newly introduced shared FileSystems.reportSourceLineage(List<ResourceId>) method, simplifying its internal lineage reporting.

🧠 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.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. 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. ↩

[shnapz]

/gemini review

[gemini-code-assist]

Code Review

This pull request adds source lineage reporting to HadoopFormatIO for file-based sources, which is a valuable addition. The implementation correctly identifies file-based splits and extracts their paths for lineage tracking.

The refactoring of the common lineage reporting logic into a shared utility method FileSystems.reportSourceLineage(List<ResourceId>) is a great improvement, reducing code duplication and centralizing the logic.

I have a couple of suggestions to improve maintainability and readability:

• In FileSystems.java, I've suggested introducing a constant for the magic number 100 to make the code clearer.
• In HadoopFormatIO.java, I've proposed refactoring the new reportSourceLineage method to use Java Streams for a more concise and modern implementation.

Overall, this is a well-structured and useful contribution.

[github-actions]

Assigning reviewers:

R: @ahmedabu98 for label java.

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

[github-actions]

Reminder, please take a look at this pr: @ahmedabu98

[github-actions]

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: @Abacn for label java.

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)

[damccorm]

R: @ahmedabu98

Could you please take a look?

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[Abacn]

This change alone won't be sufficient to make lineage report effective, as HadoopFileSystem doesn't override the no-op reportLineage method in base FileSystem:

https://github.com/apache/beam/blob/master/sdks/java/io/hadoop-file-system/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystem.java

Also it would be nice to unit test the change to make sure its effectiveness and does not break existing use cases

[shnapz]

@Abacn I have added lineage reporting to HadoopFileSystem as well

[shnapz]

btw CI checks are OK, the only thing that failed is:
testCreateManyTables[0] (org.apache.beam.sdk.io.gcp.bigquery.StorageApiSinkCreateIfNeededIT) failed
which is a flaky integration test

[Abacn]

Just make this public and add "internal use only...." comments. Not necessarily to move code around

[shnapz]

Is the concern whether to commit to a new public API? I took a second look and wanted to propose an alternative before reverting the move:
I've marked the method @internal with a "for internal use only by Beam-provided file-based connectors; not a stable public API" note in the javadoc. FileSystems already uses this same pattern for setDefaultPipelineOptions and registerFileSystemsOnce, so there's precedent for @internal public static helpers living here

The reason I prefer FileSystems over making it public on FileBasedSource: HadoopFormatIO isn't a FileBasedSource (it's a generic InputFormat wrapper that only sometimes produces file splits) so reaching into FileBasedSource as a utility namespace feels off. FileSystems is already the shared home for lineage reporting, it has three other reportSourceLineage overloads and this logic is applicable to any file-producing connector, not just subclasses of FileBasedSource.

Iam happy to revert to the FileBasedSource approach if you still prefer it. Just wanted tofloat this as an option first

[Abacn]

A generic guide is specific component change ( HadoopIO/HadoopFormat) should have minimum touch on core (FileSystems) code.

It sounds weird because in current approach HadoopFormat lineage report uses internal method belongs to FileBasedSource. Moving it to FileSystems public API further diverge the encapsulations.

Another approach is to implement HadoopFormat's reportLineage with existing FileSystems.reportSourceLineage or it's own methods

[Abacn]

This duplicates base classes' default impl. We only need to override reportLineage(HadoopResourceId resourceId, Lineage lineage, LineageLevel level)

[shnapz]

Sure, done!

[damccorm]

@Abacn would you mind taking another look?

[Abacn]

waiting on author

[Abacn]

latest comment was in #37265 (comment)