feat(integration/spark): extract table name on delta merges adding table location to urn & spark streaming lineage fixes #13602

acrylJonny · 2025-05-22T19:32:58Z

No description provided.

alwaysmeticulous · 2025-05-22T19:38:26Z

✅ Meticulous spotted visual differences in 3 of 1318 screens tested, but all differences have already been approved: view differences detected.

Meticulous evaluated ~9 hours of user flows against your PR.

_{Last updated for commit f42a9ef. This comment will update as new commits are pushed.}

codecov · 2025-05-22T19:54:58Z

Codecov Report

All modified and coverable lines are covered by tests ✅

❌ Unsupported file format

Upload processing failed due to unsupported file format. Please review the parser error message:

Error parsing JUnit XML in /home/runner/work/datahub/datahub/metadata-io/build/test-results/test/TEST-com.linkedin.metadata.graph.search.elasticsearch.SearchGraphServiceElasticSearchTest.xml at 117:1058
Caused by:

RuntimeError: Error converting computed name to ValidatedString
Caused by:
    string is too long</code></pre>


For more help, visit our troubleshooting guide.
:loudspeaker: Thoughts on this report? Let us know!

…-on-tables' of https://github.com/datahub-project/datahub into split-spark-merges-on-tables

treff7es · 2025-05-26T11:19:11Z

.../acryl-spark-lineage/src/main/java/datahub/spark/converter/SparkStreamingEventToDatahub.java

      Map<String, MetadataChangeProposalWrapper> schemaMap) {
    List<MetadataChangeProposalWrapper> mcps = new ArrayList<>();

+    String pipelineName = conf.getOpenLineageConf().getPipelineName();


Can we move this out to a separate method? This method became quite long

treff7es · 2025-05-26T11:20:22Z

...ge-converter/src/main/java/io/datahubproject/openlineage/converter/OpenLineageToDataHub.java

+    boolean isMergeIntoCommand = job.getName().contains("execute_merge_into_command_edge");
+    String tableName = null;
+
+    // If this is a MERGE INTO command and enhanced extraction is enabled, try to extract the target table name


Can we move this as well into a separate method to make this method a bit more digestable

treff7es · 2025-05-26T11:20:46Z

...ge-converter/src/main/java/io/datahubproject/openlineage/converter/OpenLineageToDataHub.java

+      // Method 1: Check for table name in the SQL facet (most reliable)
+      if (job.getFacets() != null && job.getFacets().getSql() != null) {
+        String sqlQuery = job.getFacets().getSql().getQuery();
+        if (sqlQuery != null && sqlQuery.toUpperCase().contains("MERGE INTO")) {


Can you move MERGE INTO to a constant?

treff7es · 2025-05-26T11:21:38Z

...ge-converter/src/main/java/io/datahubproject/openlineage/converter/OpenLineageToDataHub.java

+        }
+      }
+
+      // Method 3: Check for table identifiers in symlinks


I feel like these methods can be one helper method

treff7es · 2025-05-26T11:24:36Z

...ge-converter/src/main/java/io/datahubproject/openlineage/converter/OpenLineageToDataHub.java

+    if (isMergeIntoCommand && tableName != null && datahubConf.isEnhancedMergeIntoExtraction()) {
+      // Create modified job names that include the table name
+      String tablePart = tableName.replace(".", "_").replace(" ", "_").toLowerCase();
+      String enhancedJobName = job.getName() + "." + tablePart;


Does jobname contain the operation (I wonder if we should add merge_into to the job name to identify these jobs

treff7es · 2025-05-26T11:29:37Z

...ava/io/openlineage/spark/agent/vendor/delta/lifecycle/plan/DeltaMergeIntoCommandBuilder.java

+
+      // Try to get table from 'catalogTable' if it exists
+      try {
+        Method getCatalogTable = relation.getClass().getMethod("catalogTable");


Can't we identify the class, cast, and use the catalog table method?
I'm afraid of how to detect if Openlineage will rename this method to catalogTable2 or something.

Co-authored-by: treff7es <[email protected]>

treff7es · 2025-06-05T12:19:14Z

.../java/acryl-spark-lineage/src/main/java/datahub/spark/ContinuousExecutionLogInterceptor.java

+  }
+
+  /** Install a Logback appender using reflection. */
+  private static boolean installLogbackAppender(Object logger) {


These looks magical. I hope this works :)

treff7es · 2025-06-05T12:21:04Z

.../java/acryl-spark-lineage/src/main/java/datahub/spark/ContinuousExecutionLogInterceptor.java

+  /** Extract the query ID from a log message. */
+  private static String extractQueryId(String logMessage) {
+    // The query ID is typically in the format [queryId]
+    java.util.regex.Pattern pattern = java.util.regex.Pattern.compile("\\[([0-9a-f-]+)\\]");


Can you move this to static to make sure it is recompiled only once?

// Precompiled patterns for better performance private static final java.util.regex.Pattern BRACKET_QUERY_ID_PATTERN = java.util.regex.Pattern.compile("\\[([0-9a-f-]+)\\]"); private static final java.util.regex.Pattern EQUALS_QUERY_ID_PATTERN = java.util.regex.Pattern.compile("id = ([0-9a-f-]+)");

treff7es · 2025-06-05T12:21:18Z

.../java/acryl-spark-lineage/src/main/java/datahub/spark/ContinuousExecutionLogInterceptor.java

+    // Extract offset information for commit messages
+    if (logMessage.contains("Committing offset")) {
+      java.util.regex.Pattern offsetPattern =
+          java.util.regex.Pattern.compile("offset: (\\{[^\\}]+\\})");


treff7es · 2025-06-05T12:21:36Z

.../java/acryl-spark-lineage/src/main/java/datahub/spark/ContinuousExecutionLogInterceptor.java

+      metadata.put("logicalPlan", plan);
+
+      // Look for table references in the plan
+      java.util.regex.Pattern tablePattern =


treff7es · 2025-06-05T13:46:14Z

...ge-converter/src/main/java/io/datahubproject/openlineage/converter/OpenLineageToDataHub.java

+      }
+    }
+
+    // Method 3: Check for table identifiers in symlinks


Can you split this into multiple methods?

delta merge extraction

f42a9ef

github-actions bot added the ingestion PR or Issue related to the ingestion of metadata label May 22, 2025

github-actions bot deployed to datahub-project-web-react (Preview) May 22, 2025 19:37 View deployment

datahub-cyborg bot added the needs-review Label for PRs that need review from a maintainer. label May 22, 2025

vercel bot deployed to Preview May 22, 2025 19:50 View deployment

acrylJonny and others added 2 commits May 22, 2025 21:31

add parameter and docs

c548d9f

Merge branch 'master' into split-spark-merges-on-tables

99e9f22

vercel bot deployed to Preview May 22, 2025 21:01 View deployment

acrylJonny and others added 2 commits May 23, 2025 13:08

standardized job name

75f1262

Merge branch 'master' into split-spark-merges-on-tables

5205ec2

vercel bot deployed to Preview May 23, 2025 12:45 View deployment

acrylJonny and others added 5 commits May 23, 2025 20:44

fixes for spark streaming NPE for generating flow_urns

da44176

Merge branches 'split-spark-merges-on-tables' and 'split-spark-merges…

22a5486

…-on-tables' of https://github.com/datahub-project/datahub into split-spark-merges-on-tables

Merge branch 'master' into split-spark-merges-on-tables

98cb074

add backstop

0446c68

docs prettier

2cee79f

vercel bot deployed to Preview May 23, 2025 20:22 View deployment

treff7es reviewed May 26, 2025

View reviewed changes

datahub-cyborg bot added pending-submitter-response Issue/request has been reviewed but requires a response from the submitter and removed needs-review Label for PRs that need review from a maintainer. labels May 26, 2025

acrylJonny and others added 2 commits May 29, 2025 20:50

Openlineage 1 31 0 upgrade - Tamas' changes (#13658)

cfc0574

Co-authored-by: treff7es <[email protected]>

Merge branch 'master' into split-spark-merges-on-tables

7a4ef72

datahub-cyborg bot added needs-review Label for PRs that need review from a maintainer. and removed pending-submitter-response Issue/request has been reviewed but requires a response from the submitter labels May 29, 2025

vercel bot had a problem deploying to Preview May 29, 2025 20:22 Failure

acrylJonny added 3 commits May 31, 2025 23:35

spark streaming with passed config

2a9b747

streaming improvements - adding correlator for inputs and outputs

f59afd7

streaming improvements - adding correlator for inputs and outputs

eca5a35

vercel bot had a problem deploying to Preview June 1, 2025 21:27 Failure

job lineage

954e369

vercel bot had a problem deploying to Preview June 2, 2025 08:21 Failure

acrylJonny added 2 commits June 3, 2025 01:04

linting

75086ad

better capture of microbatchexecution and logs

58d8a80

vercel bot had a problem deploying to Preview June 3, 2025 09:18 Failure

acrylJonny changed the title ~~feat(integration/spark): extract table name on delta merges adding table location to urn~~ feat(integration/spark): extract table name on delta merges adding table location to urn & spark streaming lineage fixes Jun 4, 2025

treff7es approved these changes Jun 5, 2025

View reviewed changes

datahub-cyborg bot added pending-submitter-merge and removed needs-review Label for PRs that need review from a maintainer. labels Jun 5, 2025

acrylJonny closed this Jun 26, 2025

feat(integration/spark): extract table name on delta merges adding table location to urn & spark streaming lineage fixes #13602

feat(integration/spark): extract table name on delta merges adding table location to urn & spark streaming lineage fixes #13602

Uh oh!

Conversation

acrylJonny commented May 22, 2025

Uh oh!

alwaysmeticulous bot commented May 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov bot commented May 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

❌ Unsupported file format

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

alwaysmeticulous bot commented May 22, 2025 •

edited

Loading

codecov bot commented May 22, 2025 •

edited

Loading