[SPARK-52283][SQL] Declarative Pipelines DataflowGraph creation and resolution

[aakash-db]

What changes were proposed in this pull request?

This PR introduces the DataflowGraph, a container for Declarative Pipelines datasets and flows, as described in the Declarative Pipelines SPIP. It also adds functionality for

• Constructing a graph by registering a set of graph elements in succession (GraphRegistrationContext)
• "Resolving" a graph, which means resolving each of the flows within a graph. Resolving a flow means:
  • Validating that its plan can be successfully analyzed
  • Determining the schema of the data it will produce
  • Determining what upstream datasets within the graph it depends on

It also introduces various secondary changes:

• Changes to SparkBuild to support declarative pipelines.
• Updates to the pom.xml for the module.
• New error conditions

Why are the changes needed?

In order to implement Declarative Pipelines.

Does this PR introduce any user-facing change?

No changes to existing behavior.

How was this patch tested?

New test suites:

• ConnectValidPipelineSuite – test cases where the graph can be successfully resolved
• ConnectInvalidPipelineSuite – test cases where the graph fails to be resolved

Was this patch authored or co-authored using generative AI tooling?

No

[jonmio]

flushing some comments

[jonmio]

What is the purpose of this conf and do we really need it?

[jonmio]

Nit: Document return. I'm especially curious why this is a Seq and when processNode would return more than one element

[aakash-db]

Right now, it's mostly just for flexibility - in case one node maps to several in the future.

[cloud-fan]

TableIdentifier only supports 3-level namespace. Shall we use Seq[String] to better support DS v2, which can have an arbitrary level of namespace?

[cloud-fan]

Seq[String] is a bit hard to use here. We can switch to the DS v2 API after we create an encapsulation class to match TableIdentifier.

[cloud-fan]

I find it hard to understand this comment. We are resolving a flow, and the input of a flow can be other flows? Why it means the upstream table is view?

[aakash-db]

Because for views, we don't have a ViewInput object - we instead map the view's downstream directly to the view's upstream. Thus if a flow's upstream is a flow, the node it is writing to is a view. If a flow's upstream was a table, then there would be a TableInput object here.

[sryza]

@aakash-db

I'm still seeing some test failures:

• https://github.com/aakash-db/spark/actions/runs/15329080067/job/43131215330#step:12:68 – seems like maybe duplicate dependencies in the pom.xml?
• https://github.com/aakash-db/spark/actions/runs/15329080067/job/43131215320#step:14:5070 – seems like some docstrings have broken references?
• https://github.com/aakash-db/spark/actions/runs/15329080067/job/43131111744#step:7:3213 – OOM in the Docker integration tests. I've seen this on other PRs – is this generally flaky @cloud-fan ?

[cloud-fan]

yea the docker test is generally flaky and we can ignore

[cloud-fan]

If tables are both input and output, shall we just have a single name to table map, instead of a output map and a tableInput method?

[cloud-fan]

Actually there is already a table map here, why do we need the output map which is identical to the table map but just give different error message for duplication?

[sryza]

Hmm yeah it looks like we do not need a separate tableInput and we should take this out. I suspect this was a relic of an older version of a code where there was a subclass of DataflowGraph that overrode tableInput.

output is a little bit forward-looking. We will eventually allow DataflowGraphs to have sinks (described in the SPIP but not yet implemented here), and output will include both sinks and tables. However we can leave it out for now and add it back when we add sinks.

[aakash-db]

Removed tableInput.

[aakash-db]

Hm, I don't really understand the test failures:

Error:  Failed to execute goal org.apache.maven.plugins:maven-enforcer-plugin:3.5.0:enforce (enforce-no-duplicate-dependencies) on project spark-pipelines_2.13: 
Error:  Rule 0: org.apache.maven.enforcer.rules.BanDuplicatePomDependencyVersions failed with message:
Error:  Found 1 duplicate dependency declaration in this project:
Error:   - dependencies.dependency[org.apache.spark:spark-core_${scala.binary.version}:jar] (2 times)
Error:  
Error:  -> [Help 1]
Error:  
Error:  To see the full stack trace of the errors, re-run Maven with the -e switch.
Error:  Re-run Maven using the -X switch to enable full debug logging.
Error:  
Error:  For more information about the errors and possible solutions, please read the following articles:
Error:  [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException
Error:  
Error:  After correcting the problems, you can resume the build with the command
Error:    mvn <args> -rf :spark-pipelines_2.13

We do not include spark_core twice. We have one dep for production and one for tests (i.e test-jar). Is that wrong?

[error] /__w/spark/spark/sql/pipelines/target/java/org/apache/spark/sql/pipelines/graph/Input.java:7:1:  error: reference not found
[error]    * Returns a {@link DataFrame} that is a result of loading data from this {@link Input}.
[error]                       ^
[error] /__w/spark/spark/sql/pipelines/target/java/org/apache/spark/sql/pipelines/graph/Input.java:9:1:  error: reference not found
[error]    * @return Streaming or batch {@link DataFrame} of this Input's data.
[error]                                        ^Generating 

We definitely import Dataframe into this file.

[jonmio]

Should this a be a function or lazy val?

[sryza]

lazy val makes sense to me – it gets used in a few places, and we'd be redundantly recomputing it if it were a function

[jonmio]

How can this happen? Can we document that?

[jonmio]

not really obvious what this is for

[aakash-db]

I think it's useful in tests to assert on the event log.

[sryza]

We definitely import Dataframe into this file.

Hmm you are right. Not sure what's going on here. This is in auto-generated Java code that doesn't appear to have DataFrame imported. Can we just take out the square brackets for now to get past this?

Java code – Click to expand

package org.apache.spark.sql.pipelines.graph;
/**
 * Holds the {@link DataFrame} returned by a {@link FlowFunction} along with the inputs used to
 * construct it.
 * param:  usedBatchInputs the identifiers of the complete inputs read by the flow
 * param:  usedStreamingInputs the identifiers of the incremental inputs read by the flow
 * param:  usedExternalInputs the identifiers of the external inputs read by the flow
 * param:  dataFrame the {@link DataFrame} expression executed by the flow if the flow can be resolved
 */
public  class FlowFunctionResult implements scala.Product, java.io.Serializable {
  static public abstract  R apply (T1 v1, T2 v2, T3 v3, T4 v4, T5 v5, T6 v6, T7 v7)  ;
  static public  java.lang.String toString ()  { throw new RuntimeException(); }
  public  scala.collection.immutable.Set<org.apache.spark.sql.catalyst.TableIdentifier> requestedInputs ()  { throw new RuntimeException(); }
  public  scala.collection.immutable.Set<org.apache.spark.sql.pipelines.graph.ResolvedInput> usedBatchInputs ()  { throw new RuntimeException(); }
  public  scala.collection.immutable.Set<org.apache.spark.sql.pipelines.graph.ResolvedInput> usedStreamingInputs ()  { throw new RuntimeException(); }
  public  scala.collection.immutable.Set<java.lang.String> usedExternalInputs ()  { throw new RuntimeException(); }
  public  scala.util.Try<org.apache.spark.sql.classic.Dataset<org.apache.spark.sql.Row>> dataFrame ()  { throw new RuntimeException(); }
  public  scala.collection.immutable.Map<java.lang.String, java.lang.String> sqlConf ()  { throw new RuntimeException(); }
  public  scala.collection.immutable.Seq<org.apache.spark.sql.pipelines.AnalysisWarning> analysisWarnings ()  { throw new RuntimeException(); }
  // not preceding
  public   FlowFunctionResult (scala.collection.immutable.Set<org.apache.spark.sql.catalyst.TableIdentifier> requestedInputs, scala.collection.immutable.Set<org.apache.spark.sql.pipelines.graph.ResolvedInput> usedBatchInputs, scala.collection.immutable.Set<org.apache.spark.sql.pipelines.graph.ResolvedInput> usedStreamingInputs, scala.collection.immutable.Set<java.lang.String> usedExternalInputs, scala.util.Try<org.apache.spark.sql.classic.Dataset<org.apache.spark.sql.Row>> dataFrame, scala.collection.immutable.Map<java.lang.String, java.lang.String> sqlConf, scala.collection.immutable.Seq<org.apache.spark.sql.pipelines.AnalysisWarning> analysisWarnings)  { throw new RuntimeException(); }
  /**
   * Returns the names of all of the {@link Input}s used when resolving this {@link Flow}. If the
   * flow failed to resolve, we return the all the datasets that were requested when evaluating the
   * flow.
   * @return (undocumented)
   */
  public  scala.collection.immutable.Set<org.apache.spark.sql.catalyst.TableIdentifier> inputs ()  { throw new RuntimeException(); }
  /** Names of {@link Input}s read completely by this {@link Flow}. */
  public  scala.collection.immutable.Set<org.apache.spark.sql.pipelines.graph.ResolvedInput> batchInputs ()  { throw new RuntimeException(); }
  /** Names of {@link Input}s read incrementally by this {@link Flow}. */
  public  scala.collection.immutable.Set<org.apache.spark.sql.pipelines.graph.ResolvedInput> streamingInputs ()  { throw new RuntimeException(); }
  /** Returns errors that occurred when attempting to analyze this {@link Flow}. */
  public  scala.collection.immutable.Seq<java.lang.Throwable> failure ()  { throw new RuntimeException(); }
  /** Whether this {@link Flow} is successfully analyzed. */
  public final  boolean resolved ()  { throw new RuntimeException(); }
}

[sryza]

We do not include spark_core twice. We have one dep for production and one for tests (i.e test-jar). Is that wrong?

Are we consistent with what other modules that depend on spark_core do? (e.g. SQL or streaming)

[aakash-db]

Are we consistent with what other modules that depend on spark_core do? (e.g. SQL or streaming)

Yes, AFAICT. sql/hive/pom.xml or sql/core/pom.xml or streaming/pom.xml or many others do the same thing.