Spline oct 29414 qa - #921
Conversation
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
…irectory Co-authored-by: Cursor <cursoragent@cursor.com>
- Safely extract writeOptions using Try().getOrElse - Add wildcard fallback to capture unknown V2 WriteCommands like ReplaceData - Add safe null checking for extracting location path Co-authored-by: Cursor <cursoragent@cursor.com>
Catch and log Spline listener, dispatch, and cleanup failures so unexpected lineage issues do not fail Spark job execution. Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Fix NoSuchFieldException on ReplaceData in Spark V2 WriteCommands
- DataSourceV2Plugin: replace unsafe .head with .headOption + fallback to tableSpec.location/provider for Iceberg CTAS where the table does not exist yet at lineage capture time (OCT-29414) - HDFSLineageDispatcher: retry atomic rename up to 3 times with 1s delay to tolerate transient HDFS pipeline failures from bad datanodes Co-authored-by: Cursor <cursoragent@cursor.com>
⛔ Snyk checks have failed. 12 issues have been found so far.
💻 Catch issues earlier using the plugins for VS Code, JetBrains IDEs, Visual Studio, and Eclipse. |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
WalkthroughThe PR switches Maven modules to Scala 2.11, adds resilient error handling across lineage dispatch and Spark callbacks, introduces atomic HDFS lineage persistence, hardens DataSource V2 extraction, and adds a Java example for file-based lineage output. ChangesScala 2.11 build configuration
Lineage dispatch resilience
Atomic HDFS persistence
DataSource V2 extraction
Java example
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant SplineQueryExecutionListener
participant SplineAgent
participant CompositeLineageDispatcher
participant HDFSLineageDispatcher
participant HadoopFileSystem
SplineQueryExecutionListener->>SplineAgent: delegate query callback
SplineAgent->>CompositeLineageDispatcher: dispatch lineage
CompositeLineageDispatcher->>HDFSLineageDispatcher: send plan or event
HDFSLineageDispatcher->>HadoopFileSystem: write temporary lineage file
HadoopFileSystem-->>HDFSLineageDispatcher: rename file after retries
CompositeLineageDispatcher-->>SplineAgent: continue or rethrow collected failure
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (2)
core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala (1)
196-204: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDead code flagged by static analysis.
Static analysis reports
persistToHadoopFsas unused; no callers are visible in the provided context either. Remove it, or if it's kept intentionally for external callers/compatibility, document why.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala` around lines 196 - 204, Remove the unused private persistToHadoopFs method from HDFSLineageDispatcher, since private compatibility code cannot serve external callers and has no visible call sites.Source: Linters/SAST tools
examples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.java (1)
80-80: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDead code:
printLineageFilesis never called.The only call site is commented out. Either wire it in after the write job to demonstrate the HDFS output, or remove the method.
Also applies to: 84-97
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.java` at line 80, Address the unused printLineageFiles method by either invoking it after the write job to demonstrate the HDFS output, or removing the method and its related code. Keep the example consistent so no dead, unreachable demonstration logic remains.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala`:
- Around line 103-110: Update withErrorHandling so error reporting does not
depend solely on session.sparkContext: safely attempt to obtain the application
name and ID, and fall back to a generic stage-specific error message when Spark
metadata is unavailable. Always pass the original exception to logError,
including in the fallback path.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala`:
- Around line 47-60: Update the dispatch logic around delegatees and
failOnErrors so the failOnErrors=true path invokes dispatchers sequentially and
immediately propagates the first NonFatal failure, preventing later delegatees
from running. Preserve the existing tolerant behavior and failure collection
when failOnErrors is false, including warning logs for recovered dispatcher
errors.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`:
- Around line 70-91: Update HDFSLineageDispatcher so the configured filename
field from Configuration is used when constructing the output file name in
send(event), rather than always hardcoding the lineage_ prefix. Preserve the
plan ID suffix and JSON extension, and ensure the existing filename
configuration meaningfully controls the generated name.
- Line 72: Update HDFSLineageDispatcher cleanupOldRunFolders so dispatching an
event does not unconditionally delete every run folder except the current
appId-derived folder. Make cleanup opt-in via configuration and/or replace the
cross-run deletion with conservative age-based retention, ensuring concurrently
running or recently completed runs remain intact; preserve normal lineage writes
and current-run handling.
- Around line 65-68: Replace the Windows-specific fallback in the lineageBaseDir
configuration lookup with an OS-agnostic default suitable for all supported
deployments, while preserving the configured
spark.spline.lineageDispatcher.hdfs.directory value when provided. Update only
the default argument to SparkContext.get.
- Around line 153-194: Update writeFileAtomically and its renameWithRetry flow
to handle an existing finalPath, since fs.rename does not overwrite targets.
Before or during the move, explicitly replace the existing destination or use an
overwrite-capable filesystem operation while preserving atomic publication and
retry behavior, ensuring the lineage record is not lost when finalPath already
exists.
In
`@core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala`:
- Around line 121-130: Update the locationOpt and provider extraction in
DataSourceV2Plugin so extractValue reads the location and provider values
directly from tableSpec without accessing the Option wrapper’s "x" field or
casting the unwrapped String back to Option[String]. Preserve the existing
Try(...).toOption.flatten behavior so populated values remain available to the
fallback source ID.
---
Nitpick comments:
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`:
- Around line 196-204: Remove the unused private persistToHadoopFs method from
HDFSLineageDispatcher, since private compatibility code cannot serve external
callers and has no visible call sites.
In
`@examples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.java`:
- Line 80: Address the unused printLineageFiles method by either invoking it
after the write job to demonstrate the HDFS output, or removing the method and
its related code. Keep the example consistent so no dead, unreachable
demonstration logic remains.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 304977cb-4133-4257-992f-287d212a05d8
📒 Files selected for processing (17)
bundle-2.2/pom.xmlbundle-2.3/pom.xmlbundle-2.4/pom.xmlcommons/pom.xmlcore/pom.xmlcore/src/main/scala/za/co/absa/spline/agent/SplineAgent.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/FallbackLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/HttpLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/dispatcher/KafkaLineageDispatcher.scalacore/src/main/scala/za/co/absa/spline/harvester/listener/SplineQueryExecutionListener.scalacore/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scalaexamples/pom.xmlexamples/src/main/java/za/co/absa/spline/example/batch/JavaExampleJob_Files.javaintegration-tests/pom.xmlpom.xml
| private def withErrorHandling(stage: String = "lineage processing")(body: => Unit): Unit = { | ||
| try body | ||
| catch { | ||
| case NonFatal(e) => | ||
| val ctx = session.sparkContext | ||
| logError(s"Unexpected error occurred during lineage processing for application: ${ctx.appName} #${ctx.applicationId}", e) | ||
| logError(s"Unexpected error occurred during $stage for application: ${ctx.appName} #${ctx.applicationId}", e) | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Search for other unprotected `session.sparkContext` accesses in catch blocks.
ast-grep run --pattern 'catch { $$$ case $_ => $$$ session.sparkContext $$$ }' --lang scalaRepository: AbsaOSS/spline-spark-agent
Length of output: 164
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target file and inspect the relevant section.
ast-grep outline core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala --view expanded
echo '--- lines around withErrorHandling ---'
sed -n '80,130p' core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala
echo '--- all session.sparkContext references in the file ---'
rg -n 'session\.sparkContext|sparkContext' core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala
echo '--- all withErrorHandling references in the repository ---'
rg -n 'withErrorHandling\(' core/src/main/scalaRepository: AbsaOSS/spline-spark-agent
Length of output: 2282
🌐 Web query:
SparkContext applicationId IllegalStateException after stop sparkContext access docs
💡 Result:
The IllegalStateException: Cannot call methods on a stopped SparkContext occurs because Spark requires that you do not access the SparkContext, or attempt to call methods on it (such as accessing its applicationId), after the stop method has been invoked [1][2][3]. The SparkContext class contains an internal state check, assertNotStopped, which is triggered by most public methods in the context [1][2]. If the context has been stopped, this check throws the IllegalStateException [2][3]. This error is commonly encountered when: 1. Code Execution Order: Your application logic, such as a read or write operation or an access to the applicationId, is accidentally triggered after a call to spark.stop or sc.stop [4]. 2. Premature Shutdown: In scripts or notebooks, a stop command may be positioned where it executes before the intended tasks are completed (e.g., due to incorrect indentation or control flow) [4]. 3. Re-initialization Issues: Spark allows only one active SparkContext per JVM [3][5]. If you stop a context and attempt to perform operations without correctly re-initializing or ensuring the previous instance is fully cleaned up, you may encounter this error when accessing stale references [6][5]. To resolve or prevent this error: - Check for unintended calls to stop: Ensure that your SparkContext is stopped only after all processing is complete [4]. - Verify Context State: Before calling methods that rely on the SparkContext, you can use the isStopped method to check its status [1][5]. - Manage Context Lifecycle: In interactive environments like notebooks or when running multiple jobs in the same JVM, ensure you are not holding onto a reference to a SparkContext that has already been shut down [6]. If you are encountering this when trying to access the applicationId, confirm that the SparkContext instance you are querying is still active and has not been closed by a shutdown hook or an explicit stop call [2][4].
Citations:
- 1: https://issues.apache.org/jira/browse/SPARK-10426
- 2: https://github.com/apache/spark/blob/v2.3.4/core/src/main/scala/org/apache/spark/SparkContext.scala
- 3: https://github.com/apache/spark/blob/b6de0c98c70960a97b07615b0b08fbd8f900fbe7/core/src/main/scala/org/apache/spark/SparkContext.scala
- 4: https://stackoverflow.com/questions/62576237/cannot-call-methods-on-a-stopped-sparkcontext-this-stopped-sparkcontext-was-cre
- 5: https://apache.github.io/spark/api/java/org/apache/spark/SparkContext.html
- 6: https://issues.apache.org/jira/browse/SPARK-17261
Guard the error logging path session.sparkContext can throw after Spark shutdown, so this catch block can fail while reporting the original error. Add a fallback log message when app metadata is unavailable.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/src/main/scala/za/co/absa/spline/agent/SplineAgent.scala` around lines
103 - 110, Update withErrorHandling so error reporting does not depend solely on
session.sparkContext: safely attempt to obtain the application name and ID, and
fall back to a generic stage-specific error message when Spark metadata is
unavailable. Always pass the original exception to logError, including in the
fallback path.
| val failures = delegatees.flatMap { disp => | ||
| try { | ||
| call(disp) | ||
| None | ||
| } catch { | ||
| case NonFatal(e) => | ||
| logWarning(s"Proceeding after an error occurred in an underlying dispatcher: ${disp.getClass.getName}", e) | ||
| Some(e) | ||
| } | ||
| } | ||
|
|
||
| private def withErrorHandling(call: LineageDispatcher => Unit): LineageDispatcher => Unit = { disp => | ||
| try call(disp) | ||
| catch { | ||
| case NonFatal(e) => | ||
| if (failOnErrors) throw e | ||
| else logWarning(s"Proceeding after an error occurred in an underlying dispatcher: ${disp.getClass.getName}", e) | ||
| if (failOnErrors) { | ||
| failures.headOption.foreach(e => throw e) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Check that `CompositeLineageDispatcherSpec` still expects no further interactions on failure.
rg 'verifyNoMoreInteractions' core/src/test/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcherSpec.scala -B 3Repository: AbsaOSS/spline-spark-agent
Length of output: 742
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the implementation and the relevant spec sections.
sed -n '1,140p' core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala
printf '\n---- SPEC ----\n'
sed -n '1,220p' core/src/test/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcherSpec.scalaRepository: AbsaOSS/spline-spark-agent
Length of output: 5808
Restore fail-fast dispatching when failOnErrors is enabled. delegatees.flatMap still evaluates every dispatcher, so later delegates run after the first failure and mockDispatcher2 can be touched even though the spec expects no further interactions. Split the tolerant and fail-fast paths so the first error aborts dispatching.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/CompositeLineageDispatcher.scala`
around lines 47 - 60, Update the dispatch logic around delegatees and
failOnErrors so the failOnErrors=true path invokes dispatchers sequentially and
immediately propagates the first NonFatal failure, preventing later delegatees
from running. Preserve the existing tolerant behavior and failure collection
when failOnErrors is false, including warning logs for recovered dispatcher
errors.
| val lineageBaseDir = sparkContext.getConf.get( | ||
| "spark.spline.lineageDispatcher.hdfs.directory", | ||
| "file:///C:/tmp" // Default to local storage on Windows | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Hardcoded Windows-only default path in cross-platform library code.
Defaulting lineageBaseDir to "file:///C:/tmp" means any deployment (typically Linux Spark clusters) that forgets to set spark.spline.lineageDispatcher.hdfs.directory gets a nonsensical path. Since dispatch errors are swallowed by the caller (withErrorHandling in SplineAgent.scala), this fails silently — lineage is just never written, with no obvious signal why.
Proposed fix: use an OS-agnostic default
- val lineageBaseDir = sparkContext.getConf.get(
- "spark.spline.lineageDispatcher.hdfs.directory",
- "file:///C:/tmp" // Default to local storage on Windows
- )
+ val lineageBaseDir = sparkContext.getConf.get(
+ "spark.spline.lineageDispatcher.hdfs.directory",
+ s"file://${System.getProperty("java.io.tmpdir")}/spline-lineage"
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| val lineageBaseDir = sparkContext.getConf.get( | |
| "spark.spline.lineageDispatcher.hdfs.directory", | |
| "file:///C:/tmp" // Default to local storage on Windows | |
| ) | |
| val lineageBaseDir = sparkContext.getConf.get( | |
| "spark.spline.lineageDispatcher.hdfs.directory", | |
| s"file://${System.getProperty("java.io.tmpdir")}/spline-lineage" | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
around lines 65 - 68, Replace the Windows-specific fallback in the
lineageBaseDir configuration lookup with an OS-agnostic default suitable for all
supported deployments, while preserving the configured
spark.spline.lineageDispatcher.hdfs.directory value when provided. Update only
the default argument to SparkContext.get.
| val executionPlanID = this._lastSeenPlan.id.getOrElse("unknown_plan") | ||
| val executionPlanName = this._lastSeenPlan.name.replaceAll("[^a-zA-Z0-9_\\-]", "_") | ||
| val runID = event.extra.get("appId").map(_.toString).getOrElse("unknown_runID") | ||
|
|
||
| val appDir = s"$lineageBaseDir/$executionPlanName" | ||
| val runDir = s"$appDir/$runID" | ||
|
|
||
| // Best-effort cleanup of previous runs | ||
| cleanupOldRunFolders(appDir, runID) | ||
|
|
||
| val fileName = s"lineage_${executionPlanID}.json" | ||
|
|
||
| // Build JSON payload | ||
| val planWithEvent = Map( | ||
| "executionPlan" -> this._lastSeenPlan, | ||
| "executionEvent" -> event | ||
| ) | ||
|
|
||
| import HarvesterJsonSerDe.impl._ | ||
| persistToHadoopFs(planWithEvent.toJson, path) | ||
| val content = planWithEvent.toJson | ||
|
|
||
| // Atomically materialize the run folder & file | ||
| persistRunFolderAtomically(runDir, fileName, content) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Configured fileName is now dead — output naming is fully hardcoded.
The class still accepts/requires a filename field via its Configuration constructor (unchanged lines 39-47), but send(event) never references this.filename; the file is always named lineage_<planId>.json. Any user setting fileName (defaulted to _LINEAGE in spline.default.yaml) has zero effect, which is misleading for anyone tuning that config.
Either wire this.filename into the computed name (e.g. as a prefix) or drop the now-unused parameter/config key to avoid a misleading public contract.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
around lines 70 - 91, Update HDFSLineageDispatcher so the configured filename
field from Configuration is used when constructing the output file name in
send(event), rather than always hardcoding the lineage_ prefix. Preserve the
plan ID suffix and JSON extension, and ensure the existing filename
configuration meaningfully controls the generated name.
|
|
||
| val executionPlanID = this._lastSeenPlan.id.getOrElse("unknown_plan") | ||
| val executionPlanName = this._lastSeenPlan.name.replaceAll("[^a-zA-Z0-9_\\-]", "_") | ||
| val runID = event.extra.get("appId").map(_.toString).getOrElse("unknown_runID") |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift
Unconditional cross-run cleanup can delete another concurrently-running job's lineage data.
cleanupOldRunFolders recursively deletes every subdirectory of appDir that isn't the current run's appId-derived folder — every single time an event is dispatched. appDir is keyed only by the sanitized execution-plan name, not by anything guaranteeing exclusivity. Two overlapping executions of the same job (retry, back-to-back scheduled runs, blue/green rollout) will each see the other's run folder as "old" and delete it — potentially while the other run is actively writing into it. This turns what should be a persistent lineage audit trail into something that can lose or corrupt a peer application's output.
Consider making this cleanup opt-in (config flag) and/or scoping deletion more conservatively (e.g. age-based retention rather than "everything but mine").
Also applies to: 97-124
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala`
at line 72, Update HDFSLineageDispatcher cleanupOldRunFolders so dispatching an
event does not unconditionally delete every run folder except the current
appId-derived folder. Make cleanup opt-in via configuration and/or replace the
cross-run deletion with conservative age-based retention, ensuring concurrently
running or recently completed runs remain intact; preserve normal lineage writes
and current-run handling.
| val locationOpt = Try { | ||
| val tableSpec = extractValue[AnyRef](ctc, "tableSpec") | ||
| val locationOption = extractValue[AnyRef](tableSpec, "location") | ||
| extractValue[Option[String]](locationOption, "x") | ||
| }.toOption.flatten | ||
| val provider = Try { | ||
| val tableSpec = extractValue[AnyRef](ctc, "tableSpec") | ||
| val providerOption = extractValue[AnyRef](tableSpec, "provider") | ||
| extractValue[Option[String]](providerOption, "x") | ||
| }.toOption.flatten |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target file first
ast-grep outline core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala --view expanded || true
echo
echo "---- relevant lines ----"
sed -n '90,150p' core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala
echo
echo "---- extractValue definition search ----"
rg -n "def extractValue|extractValue\\[" core/src/main/scala -SRepository: AbsaOSS/spline-spark-agent
Length of output: 18865
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the implementation of extractValue and any related reflection helpers.
rg -n "def extractValue|object .*Reflection|class .*Reflection|method.*extractValue" core/src/main/scala -S
echo
echo "---- likely helper files ----"
fd -a ".*Reflection.*|.*Utils.*|.*Helper.*" core/src/main/scala | sed -n '1,80p'Repository: AbsaOSS/spline-spark-agent
Length of output: 164
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# If TableSpec is available in the local codebase or referenced docs, inspect its shape.
rg -n "TableSpec|tableSpec" core/src/main/scala -SRepository: AbsaOSS/spline-spark-agent
Length of output: 971
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== file slice =="
sed -n '110,140p' core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala
echo
echo "== extractValue implementation(s) =="
rg -n "def extractValue|extractValue\\[" core/src/main/scala -S
echo
echo "== search for field name x usage in reflection helpers =="
rg -n '"x"|\\bx\\b' core/src/main/scala -SRepository: AbsaOSS/spline-spark-agent
Length of output: 18300
🌐 Web query:
Spark DataSourceV2 TableSpec location provider Option[String] x field Some reflection
💡 Result:
In Apache Spark's DataSourceV2 API, TableSpec is an internal class used during logical planning to represent table creation parameters, including user-specified details like the table location and data source provider [1][2]. Within the Spark catalyst codebase, TableSpec (or its base traits like TableSpecBase) captures these attributes to bridge SQL DDL statements (like CREATE TABLE) with Spark's V2 command execution [1][2][3]. Key technical points regarding your query: 1. Location and Provider: The location (path to table data) and provider (the data source type, e.g., 'parquet', 'iceberg') are defined as fields within the TableSpec structure [1][2]. These are typically represented as Option[String], reflecting that a user might not specify them in every DDL statement (e.g., Spark may default a location if one is not provided) [4][1]. 2. Reflection and Field Access: Spark's internal command processing often relies on pattern matching or case class extraction to work with these structures [5][6]. Because TableSpec is an internal logical plan component, interaction with its fields often involves Scala's case class pattern matching (e.g., case class TableSpec(...) {... }) or extractors [5][1]. While standard Scala pattern matching is the preferred and idiomatic way to handle these fields in Spark's Catalyst engine, reflection may be encountered if working with older or highly dynamic internal APIs, though it is not the standard interface for interacting with DataSourceV2 metadata [5][6]. 3. Development Context: As Spark's DataSourceV2 framework evolves, the structure of these internal classes (like TableSpec or DataSourceV2Relation) is subject to change [6]. Recent contributions have aimed to simplify how tables are extracted from relations, often replacing complex manual field access with dedicated extractors to reduce boilerplate code [6]. If you are developing a custom data source, you should primarily interact with the TableProvider, Table, and Scan/Write APIs documented in the official Spark Data Source V2 documentation rather than relying on internal Catalyst classes like TableSpec, which are subject to breaking changes across minor versions [7][8][9].
Citations:
- 1: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
- 2: https://fossies.org/linux/misc/spark-4.1.2.tgz/spark-4.1.2/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala?M=626
- 3: [SPARK-36902][SQL] Migrate CreateTableAsSelectStatement to v2 command apache/spark#34667
- 4: https://spark.apache.org/docs/latest/sql-ref-syntax-ddl-create-table-datasource.html
- 5: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
- 6: [SPARK-53720][SQL] Simplify extracting Table from DataSourceV2Relation apache/spark#52460
- 7: https://spark.apache.org/docs/preview/sql-data-sources-v2.html
- 8: https://spark.apache.org/docs/4.2.0-preview4/sql-data-sources-v2.html
- 9: https://spark.incubator.apache.org/docs/4.2.0-preview4/sql-data-sources-v2.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== ReflectionUtils locations =="
fd -a "ReflectionUtils.scala" core/src/main/scala || true
rg -n "object ReflectionUtils|class ReflectionUtils|type ExtractFieldValueFn|def extractValue" core/src/main/scala -S
echo
echo "== likely helper file slice(s) =="
for f in $(fd -a "ReflectionUtils.scala" core/src/main/scala); do
echo "--- $f ---"
wc -l "$f"
sed -n '1,220p' "$f"
doneRepository: AbsaOSS/spline-spark-agent
Length of output: 377
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Probe the runtime shape of Scala Option/Something-like extraction via source-only reasoning.
# First inspect where Some defines its payload field name, then infer what field access would return.
rg -n "class Some|case class Some|final case class Some" /usr/share/scala /usr/lib/scala 2>/dev/null | head -n 20 || trueRepository: AbsaOSS/spline-spark-agent
Length of output: 164
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== DataSourceV2Plugin imports and class header =="
sed -n '1,90p' core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala
echo
echo "== ObjectStructureDumper slice =="
sed -n '1,140p' core/src/main/scala/za/co/absa/spline/harvester/logging/ObjectStructureDumper.scalaRepository: AbsaOSS/spline-spark-agent
Length of output: 8822
Extract location and provider directly from tableSpec. extractValue[Option[String]](..., "x") is unwrapping Some.x as a String and then casting it back to Option[String], so populated values are dropped and the fallback source ID loses location/provider information.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/main/scala/za/co/absa/spline/harvester/plugin/embedded/DataSourceV2Plugin.scala`
around lines 121 - 130, Update the locationOpt and provider extraction in
DataSourceV2Plugin so extractValue reads the location and provider values
directly from tableSpec without accessing the Option wrapper’s "x" field or
casting the unwrapped String back to Option[String]. Preserve the existing
Try(...).toOption.flatten behavior so populated values remain available to the
fallback source ID.
Delete an existing lineage JSON file before the atomic rename so repeated attempts for the same Spark application can refresh the payload instead of failing on an existing destination. Co-authored-by: Cursor <cursoragent@cursor.com>
|
4 participants
Summary by CodeRabbit
New Features
Bug Fixes
file://locations.