You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "JiaqiWang18 (via GitHub)" <gi...@apache.org> on 2025/07/16 01:27:07 UTC
[PR] [WIP][SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
JiaqiWang18 opened a new pull request, #51507:
URL: https://github.com/apache/spark/pull/51507
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213862264
##########
python/pyspark/pipelines/cli.py:
##########
@@ -283,6 +341,11 @@ def run(spec_path: Path) -> None:
else:
spec_path = find_pipeline_spec(Path.cwd())
- run(spec_path=spec_path)
+ run(
+ spec_path=spec_path,
+ full_refresh=flatten_table_lists(args.full_refresh),
Review Comment:
Very good point, `extend` creates a 1D list directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213864934
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
Review Comment:
extracted a [createTableFilters](https://github.com/apache/spark/pull/51507/commits/1693ac546225c8a6be1d96eb5e64fcf03f77a344#diff-44e47ef13083c7fae6bd89d2774a8141eec6e76874424aaf4d9dc93f59362210R315) function
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
+ // Extract refresh parameters from protobuf command
+ val fullRefreshTables = cmd.getFullRefreshList.asScala.toSeq
+ val fullRefreshAll = cmd.getFullRefreshAll
+ val refreshTables = cmd.getRefreshList.asScala.toSeq
+
+ // Convert table names to fully qualified TableIdentifier objects
+ def parseTableNames(tableNames: Seq[String]): Set[TableIdentifier] = {
+ tableNames.map { name =>
+ GraphIdentifierManager
+ .parseAndQualifyTableIdentifier(
+ rawTableIdentifier =
+ GraphIdentifierManager.parseTableIdentifier(name, sessionHolder.session),
+ currentCatalog = Some(graphElementRegistry.defaultCatalog),
+ currentDatabase = Some(graphElementRegistry.defaultDatabase))
+ .identifier
+ }.toSet
+ }
+
+ if (fullRefreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to refresh when full refresh all is set to true.")
+ }
+
+ if (refreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to full refresh when full refresh all is set to true.")
+ }
+ val refreshTableNames = parseTableNames(refreshTables)
+ val fullRefreshTableNames = parseTableNames(fullRefreshTables)
+
+ if (refreshTables.nonEmpty && fullRefreshTables.nonEmpty) {
+ // check if there is an intersection between the subset
+ val intersection = refreshTableNames.intersect(fullRefreshTableNames)
+ if (intersection.nonEmpty) {
+ throw new IllegalArgumentException(
+ "Datasets specified for refresh and full refresh cannot overlap: " +
+ s"${intersection.mkString(", ")}")
+ }
+ }
+
+ val fullRefreshTablesFilter: TableFilter = if (fullRefreshAll) {
+ AllTables
+ } else if (fullRefreshTables.nonEmpty) {
+ SomeTables(fullRefreshTableNames)
+ } else {
+ NoTables
+ }
+
+ val refreshTablesFilter: TableFilter =
+ if (refreshTables.nonEmpty) {
+ SomeTables(refreshTableNames)
+ } else if (fullRefreshTablesFilter != NoTables) {
+ NoTables
+ } else {
+ AllTables
+ }
Review Comment:
extracted a [createTableFilters](https://github.com/apache/spark/pull/51507/commits/1693ac546225c8a6be1d96eb5e64fcf03f77a344#diff-44e47ef13083c7fae6bd89d2774a8141eec6e76874424aaf4d9dc93f59362210R315) function
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "sryza (via GitHub)" <gi...@apache.org>.
sryza commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213971064
##########
python/pyspark/pipelines/cli.py:
##########
@@ -242,20 +264,56 @@ def run(spec_path: Path) -> None:
register_definitions(spec_path, registry, spec)
log_with_curr_timestamp("Starting run...")
- result_iter = start_run(spark, dataflow_graph_id)
+ result_iter = start_run(
+ spark,
+ dataflow_graph_id,
+ full_refresh=full_refresh,
+ full_refresh_all=full_refresh_all,
+ refresh=refresh,
+ )
try:
handle_pipeline_events(result_iter)
finally:
spark.stop()
+def parse_table_list(value: str) -> List[str]:
+ """Parse a comma-separated list of table names, handling whitespace."""
+ return [table.strip() for table in value.split(",") if table.strip()]
+
+
+def flatten_table_lists(table_lists: Optional[List[List[str]]]) -> Optional[List[str]]:
+ """Flatten a list of lists of table names into a single list."""
+ if not table_lists:
+ return None
+ result = []
+ for table_list in table_lists:
+ result.extend(table_list)
+ return result if result else None
+
+
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Pipeline CLI")
subparsers = parser.add_subparsers(dest="command", required=True)
# "run" subcommand
run_parser = subparsers.add_parser("run", help="Run a pipeline.")
run_parser.add_argument("--spec", help="Path to the pipeline spec.")
+ run_parser.add_argument(
+ "--full-refresh",
+ type=parse_table_list,
+ action="append",
+ help="List of datasets to reset and recompute (comma-separated).",
Review Comment:
Will `extend` split using commas?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211732544
##########
python/pyspark/pipelines/cli.py:
##########
@@ -283,6 +341,11 @@ def run(spec_path: Path) -> None:
else:
spec_path = find_pipeline_spec(Path.cwd())
- run(spec_path=spec_path)
+ run(
+ spec_path=spec_path,
+ full_refresh=flatten_table_lists(args.full_refresh),
Review Comment:
If we were to mark this argument field as `extend` rather than `append`, would we still need to do any manual flattening?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213863285
##########
sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto:
##########
@@ -116,6 +116,15 @@ message PipelineCommand {
message StartRun {
// The graph to start.
optional string dataflow_graph_id = 1;
+
+ // List of tables to reset and recompute.
+ repeated string full_refresh = 2;
Review Comment:
renamed in one commit with proto generation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213861341
##########
python/pyspark/pipelines/cli.py:
##########
@@ -242,20 +264,56 @@ def run(spec_path: Path) -> None:
register_definitions(spec_path, registry, spec)
log_with_curr_timestamp("Starting run...")
- result_iter = start_run(spark, dataflow_graph_id)
+ result_iter = start_run(
+ spark,
+ dataflow_graph_id,
+ full_refresh=full_refresh,
+ full_refresh_all=full_refresh_all,
+ refresh=refresh,
+ )
try:
handle_pipeline_events(result_iter)
finally:
spark.stop()
+def parse_table_list(value: str) -> List[str]:
+ """Parse a comma-separated list of table names, handling whitespace."""
+ return [table.strip() for table in value.split(",") if table.strip()]
+
+
+def flatten_table_lists(table_lists: Optional[List[List[str]]]) -> Optional[List[str]]:
+ """Flatten a list of lists of table names into a single list."""
+ if not table_lists:
+ return None
+ result = []
+ for table_list in table_lists:
+ result.extend(table_list)
+ return result if result else None
Review Comment:
Removed this by using the `extend` option in arg parser to avoid creating nested list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213867860
##########
python/pyspark/pipelines/tests/test_cli.py:
##########
@@ -359,6 +360,139 @@ def test_python_import_current_directory(self):
)
+@unittest.skipIf(
+ not should_test_connect or not have_yaml,
+ connect_requirement_message or yaml_requirement_message,
+)
+class CLIValidationTests(unittest.TestCase):
Review Comment:
yeah I think they can be combined into one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "sryza (via GitHub)" <gi...@apache.org>.
sryza commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211609713
##########
python/pyspark/pipelines/cli.py:
##########
@@ -217,8 +217,30 @@ def change_dir(path: Path) -> Generator[None, None, None]:
os.chdir(prev)
-def run(spec_path: Path) -> None:
- """Run the pipeline defined with the given spec."""
+def run(
+ spec_path: Path,
+ full_refresh: Optional[Sequence[str]] = None,
+ full_refresh_all: bool = False,
+ refresh: Optional[Sequence[str]] = None,
+) -> None:
+ """Run the pipeline defined with the given spec.
+
+ :param spec_path: Path to the pipeline specification file.
+ :param full_refresh: List of datasets to reset and recompute.
+ :param full_refresh_all: Perform a full graph reset and recompute.
+ :param refresh: List of datasets to update.
+ """
+ # Validate conflicting arguments
+ if full_refresh_all:
+ if full_refresh:
Review Comment:
Should these be consolidated into `if full_refresh or refresh`?
##########
python/pyspark/pipelines/spark_connect_pipeline.py:
##########
@@ -65,12 +65,26 @@ def handle_pipeline_events(iter: Iterator[Dict[str, Any]]) -> None:
log_with_provided_timestamp(event.message, dt)
-def start_run(spark: SparkSession, dataflow_graph_id: str) -> Iterator[Dict[str, Any]]:
+def start_run(
+ spark: SparkSession,
+ dataflow_graph_id: str,
+ full_refresh: Optional[Sequence[str]] = None,
Review Comment:
Same comment as above about default args.
##########
python/pyspark/pipelines/cli.py:
##########
@@ -217,8 +217,30 @@ def change_dir(path: Path) -> Generator[None, None, None]:
os.chdir(prev)
-def run(spec_path: Path) -> None:
- """Run the pipeline defined with the given spec."""
+def run(
+ spec_path: Path,
+ full_refresh: Optional[Sequence[str]] = None,
Review Comment:
This never gets invoked without these parameters specified, so no need to have default values for them.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineRefreshFunctionalSuite.scala:
##########
@@ -0,0 +1,367 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.pipelines
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.connect.proto.{DatasetType, PipelineCommand, PipelineEvent}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.connect.service.{SessionKey, SparkConnectService}
+import org.apache.spark.sql.pipelines.utils.{EventVerificationTestHelpers, TestPipelineUpdateContextMixin}
+
+/**
+ * Comprehensive test suite that validates pipeline refresh functionality by running actual
+ * pipelines with different refresh parameters and validating the results.
+ */
+class PipelineRefreshFunctionalSuite
+ extends SparkDeclarativePipelinesServerTest
+ with TestPipelineUpdateContextMixin
+ with EventVerificationTestHelpers {
+
+ private val externalSourceTable = TableIdentifier(
+ catalog = Some("spark_catalog"),
+ database = Some("default"),
+ table = "source_data")
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ // Create source table to simulate streaming updates
+ spark.sql(s"CREATE TABLE $externalSourceTable AS SELECT * FROM RANGE(1, 2)")
+ }
+
+ override def afterEach(): Unit = {
+ super.afterEach()
+ // Clean up the source table after each test
+ spark.sql(s"DROP TABLE IF EXISTS $externalSourceTable")
+ }
+
+ private def createTestPipeline(graphId: String): TestPipelineDefinition = {
+ new TestPipelineDefinition(graphId) {
+ // Create tables that depend on the mv
+ createTable(
+ name = "a",
+ datasetType = DatasetType.TABLE,
+ sql = Some(s"SELECT id FROM STREAM $externalSourceTable"))
+ createTable(
+ name = "b",
+ datasetType = DatasetType.TABLE,
+ sql = Some(s"SELECT id FROM STREAM $externalSourceTable"))
+ createTable(
+ name = "mv",
+ datasetType = DatasetType.MATERIALIZED_VIEW,
+ sql = Some(s"SELECT id FROM a"))
+ }
+ }
+
+ /**
+ * Helper method to run refresh tests with common setup and verification logic. This reduces
+ * code duplication across the refresh test cases.
+ */
+ private def runRefreshTest(
+ refreshConfigBuilder: String => Option[PipelineCommand.StartRun] = _ => None,
+ expectedContentAfterRefresh: Map[String, Set[Map[String, Any]]],
+ eventValidation: Option[ArrayBuffer[PipelineEvent] => Unit] = None): Unit = {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ // First run to populate tables
+ startPipelineAndWaitForCompletion(graphId)
+
+ // Verify initial data - all tests expect the same initial state
+ verifyMultipleTableContent(
+ tableNames =
+ Set("spark_catalog.default.a", "spark_catalog.default.b", "spark_catalog.default.mv"),
+ columnsToVerify = Map(
+ "spark_catalog.default.a" -> Seq("id"),
+ "spark_catalog.default.b" -> Seq("id"),
+ "spark_catalog.default.mv" -> Seq("id")),
+ expectedContent = Map(
+ "spark_catalog.default.a" -> Set(Map("id" -> 1)),
+ "spark_catalog.default.b" -> Set(Map("id" -> 1)),
+ "spark_catalog.default.mv" -> Set(Map("id" -> 1))))
+
+ // Clear cached pipeline execution before starting new run
+ SparkConnectService.sessionManager
+ .getIsolatedSessionIfPresent(SessionKey(defaultUserId, defaultSessionId))
+ .foreach(_.removeAllPipelineExecutions())
+
+ // Replace source data to simulate a streaming update
+ spark.sql(
+ "INSERT OVERWRITE TABLE spark_catalog.default.source_data " +
+ "SELECT * FROM VALUES (2), (3) AS t(id)")
+
+ // Run with specified refresh configuration
+ val capturedEvents = refreshConfigBuilder(graphId) match {
+ case Some(startRun) => startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ case None => startPipelineAndWaitForCompletion(graphId)
+ }
+
+ // Additional validation if provided
+ eventValidation.foreach(_(capturedEvents))
+
+ // Verify final content
+ verifyMultipleTableContent(
+ tableNames =
+ Set("spark_catalog.default.a", "spark_catalog.default.b", "spark_catalog.default.mv"),
+ columnsToVerify = Map(
+ "spark_catalog.default.a" -> Seq("id"),
+ "spark_catalog.default.b" -> Seq("id"),
+ "spark_catalog.default.mv" -> Seq("id")),
+ expectedContent = expectedContentAfterRefresh)
+ }
+ }
+
+ test("pipeline runs selective full_refresh") {
+ runRefreshTest(
+ refreshConfigBuilder = { graphId =>
+ Some(
+ PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .addAllFullRefresh(List("a").asJava)
+ .build())
+ },
+ expectedContentAfterRefresh = Map(
+ "spark_catalog.default.a" -> Set(
+ Map("id" -> 2), // a is fully refreshed and only contains the new values
+ Map("id" -> 3)),
+ "spark_catalog.default.b" -> Set(
+ Map("id" -> 1) // b is not refreshed, so it retains the old value
+ ),
+ "spark_catalog.default.mv" -> Set(
+ Map("id" -> 1) // mv is not refreshed, so it retains the old value
+ )),
+ eventValidation = Some { capturedEvents =>
+ // assert that table_b is excluded
+ assert(
+ capturedEvents.exists(
+ _.getMessage.contains(s"Flow \'spark_catalog.default.b\' is EXCLUDED.")))
+ // assert that table_a ran to completion
+ assert(
+ capturedEvents.exists(
+ _.getMessage.contains(s"Flow spark_catalog.default.a has COMPLETED.")))
+ // assert that mv is excluded
+ assert(
+ capturedEvents.exists(
+ _.getMessage.contains(s"Flow \'spark_catalog.default.mv\' is EXCLUDED.")))
+ // Verify completion event
+ assert(capturedEvents.exists(_.getMessage.contains("Run is COMPLETED")))
+ })
+ }
+
+ test("pipeline runs selective full_refresh and selective refresh") {
+ runRefreshTest(
+ refreshConfigBuilder = { graphId =>
+ Some(
+ PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .addAllFullRefresh(Seq("a", "mv").asJava)
+ .addRefresh("b")
+ .build())
+ },
+ expectedContentAfterRefresh = Map(
+ "spark_catalog.default.a" -> Set(
+ Map("id" -> 2), // a is fully refreshed and only contains the new values
+ Map("id" -> 3)),
+ "spark_catalog.default.b" -> Set(
+ Map("id" -> 1), // b is refreshed, so it retains the old value and adds the new ones
+ Map("id" -> 2),
+ Map("id" -> 3)),
+ "spark_catalog.default.mv" -> Set(
+ Map("id" -> 2), // mv is fully refreshed and only contains the new values
+ Map("id" -> 3))))
+ }
+
+ test("pipeline runs refresh by default") {
+ runRefreshTest(expectedContentAfterRefresh =
+ Map(
+ "spark_catalog.default.a" -> Set(
+ Map(
+ "id" -> 1
+ ), // a is refreshed by default, retains the old value and adds the new ones
+ Map("id" -> 2),
+ Map("id" -> 3)),
+ "spark_catalog.default.b" -> Set(
+ Map(
+ "id" -> 1
+ ), // b is refreshed by default, retains the old value and adds the new ones
+ Map("id" -> 2),
+ Map("id" -> 3)),
+ "spark_catalog.default.mv" -> Set(
+ Map("id" -> 1),
+ Map("id" -> 2), // mv is refreshed from table a, retains all values
+ Map("id" -> 3))))
+ }
+
+ test("pipeline runs full refresh all") {
+ runRefreshTest(
+ refreshConfigBuilder = { graphId =>
+ Some(
+ PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .setFullRefreshAll(true)
+ .build())
+ },
+ // full refresh all causes all tables to lose the initial value
+ // and only contain the new values after the source data is updated
+ expectedContentAfterRefresh = Map(
+ "spark_catalog.default.a" -> Set(Map("id" -> 2), Map("id" -> 3)),
+ "spark_catalog.default.b" -> Set(Map("id" -> 2), Map("id" -> 3)),
+ "spark_catalog.default.mv" -> Set(Map("id" -> 2), Map("id" -> 3))))
+ }
+
+ test("validation: cannot specify subset refresh when full_refresh_all is true") {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ val startRun = PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .setFullRefreshAll(true)
+ .addRefresh("a")
+ .build()
+
+ val exception = intercept[IllegalArgumentException] {
+ startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ }
+ assert(
+ exception.getMessage.contains(
+ "Cannot specify a subset to full refresh when full refresh all is set to true"))
+ }
+ }
+
+ test("validation: cannot specify subset full_refresh when full_refresh_all is true") {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ val startRun = PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .setFullRefreshAll(true)
+ .addFullRefresh("a")
+ .build()
+
+ val exception = intercept[IllegalArgumentException] {
+ startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ }
+ assert(
+ exception.getMessage.contains(
+ "Cannot specify a subset to refresh when full refresh all is set to true"))
+ }
+ }
+
+ test("validation: refresh and full_refresh cannot overlap") {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ val startRun = PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .addRefresh("a")
+ .addFullRefresh("a")
+ .build()
+
+ val exception = intercept[IllegalArgumentException] {
+ startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ }
+ assert(
+ exception.getMessage.contains(
+ "Datasets specified for refresh and full refresh cannot overlap"))
+ assert(exception.getMessage.contains("a"))
+ }
+ }
+
+ test("validation: multiple overlapping tables in refresh and full_refresh not allowed") {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ val startRun = PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .addRefresh("a")
+ .addRefresh("b")
+ .addFullRefresh("a")
+ .build()
+
+ val exception = intercept[IllegalArgumentException] {
+ startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ }
+ assert(
+ exception.getMessage.contains(
+ "Datasets specified for refresh and full refresh cannot overlap"))
+ assert(exception.getMessage.contains("a"))
+ }
+ }
+
+ test("validation: fully qualified table names in validation") {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ val startRun = PipelineCommand.StartRun
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .addRefresh("spark_catalog.default.a")
+ .addFullRefresh("a") // This should be treated as the same table
+ .build()
+
+ val exception = intercept[IllegalArgumentException] {
+ startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ }
+ assert(
+ exception.getMessage.contains(
+ "Datasets specified for refresh and full refresh cannot overlap"))
+ }
+ }
+
+ private def verifyMultipleTableContent(
Review Comment:
Are we handling each table independently? If so, would it make sense to expose a function that operates on a single table and `foreach` over it where it's invoked?
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerTest.scala:
##########
@@ -125,15 +128,25 @@ class SparkDeclarativePipelinesServerTest extends SparkConnectServerTest {
def createPlanner(): SparkConnectPlanner =
new SparkConnectPlanner(SparkConnectTestUtils.createDummySessionHolder(spark))
- def startPipelineAndWaitForCompletion(graphId: String): Unit = {
+ def startPipelineAndWaitForCompletion(
+ graphId: String,
+ customStartRunCommand: Option[PipelineCommand.StartRun] = None)
Review Comment:
Supporting these two arguments at the same time allows an undefined situation where the custom `StartRun` command has a different ID than the value supplied for the `graphId` argument.
Could we instead factor it like this:
```scala
def startPipelineAndWaitForCompletion(startRunCommand: PipelineCommand.StartRun): Unit = ...
def startPipelineAndWaitForCompletion(graphId: String): Unit = {
startPipelineAndWaitForCompletion(PipelineCommand.StartRun.newBuilder().setDataflowGraphId(graphId).build())
}
```
##########
python/pyspark/pipelines/tests/test_cli.py:
##########
@@ -359,6 +360,139 @@ def test_python_import_current_directory(self):
)
+@unittest.skipIf(
+ not should_test_connect or not have_yaml,
+ connect_requirement_message or yaml_requirement_message,
+)
+class CLIValidationTests(unittest.TestCase):
Review Comment:
Is there a meaningful difference between the kinds of tests that are included in this class and the kinds of tests that included in the other class in this file?
##########
python/pyspark/pipelines/cli.py:
##########
@@ -28,7 +28,7 @@
import yaml
from dataclasses import dataclass
from pathlib import Path
-from typing import Any, Generator, Mapping, Optional, Sequence
+from typing import Any, Generator, Mapping, Optional, Sequence, List
Review Comment:
Out of alphabetical order: you may need to run `dev/reformat-python` to format this.
##########
python/pyspark/pipelines/cli.py:
##########
@@ -217,8 +217,30 @@ def change_dir(path: Path) -> Generator[None, None, None]:
os.chdir(prev)
-def run(spec_path: Path) -> None:
- """Run the pipeline defined with the given spec."""
+def run(
Review Comment:
If we expect it to vary across run for the same pipeline, it should be a CLI arg. If we expect it to be static for a pipeline, it should live in the spec. I would expect selections to vary across runs.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineRefreshFunctionalSuite.scala:
##########
@@ -0,0 +1,367 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.pipelines
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.connect.proto.{DatasetType, PipelineCommand, PipelineEvent}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.connect.service.{SessionKey, SparkConnectService}
+import org.apache.spark.sql.pipelines.utils.{EventVerificationTestHelpers, TestPipelineUpdateContextMixin}
+
+/**
+ * Comprehensive test suite that validates pipeline refresh functionality by running actual
+ * pipelines with different refresh parameters and validating the results.
+ */
+class PipelineRefreshFunctionalSuite
+ extends SparkDeclarativePipelinesServerTest
+ with TestPipelineUpdateContextMixin
+ with EventVerificationTestHelpers {
+
+ private val externalSourceTable = TableIdentifier(
+ catalog = Some("spark_catalog"),
+ database = Some("default"),
+ table = "source_data")
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ // Create source table to simulate streaming updates
+ spark.sql(s"CREATE TABLE $externalSourceTable AS SELECT * FROM RANGE(1, 2)")
+ }
+
+ override def afterEach(): Unit = {
+ super.afterEach()
+ // Clean up the source table after each test
+ spark.sql(s"DROP TABLE IF EXISTS $externalSourceTable")
+ }
+
+ private def createTestPipeline(graphId: String): TestPipelineDefinition = {
+ new TestPipelineDefinition(graphId) {
+ // Create tables that depend on the mv
+ createTable(
+ name = "a",
+ datasetType = DatasetType.TABLE,
+ sql = Some(s"SELECT id FROM STREAM $externalSourceTable"))
+ createTable(
+ name = "b",
+ datasetType = DatasetType.TABLE,
+ sql = Some(s"SELECT id FROM STREAM $externalSourceTable"))
+ createTable(
+ name = "mv",
+ datasetType = DatasetType.MATERIALIZED_VIEW,
+ sql = Some(s"SELECT id FROM a"))
+ }
+ }
+
+ /**
+ * Helper method to run refresh tests with common setup and verification logic. This reduces
+ * code duplication across the refresh test cases.
+ */
+ private def runRefreshTest(
+ refreshConfigBuilder: String => Option[PipelineCommand.StartRun] = _ => None,
+ expectedContentAfterRefresh: Map[String, Set[Map[String, Any]]],
+ eventValidation: Option[ArrayBuffer[PipelineEvent] => Unit] = None): Unit = {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ val pipeline = createTestPipeline(graphId)
+ registerPipelineDatasets(pipeline)
+
+ // First run to populate tables
+ startPipelineAndWaitForCompletion(graphId)
+
+ // Verify initial data - all tests expect the same initial state
+ verifyMultipleTableContent(
+ tableNames =
+ Set("spark_catalog.default.a", "spark_catalog.default.b", "spark_catalog.default.mv"),
+ columnsToVerify = Map(
+ "spark_catalog.default.a" -> Seq("id"),
+ "spark_catalog.default.b" -> Seq("id"),
+ "spark_catalog.default.mv" -> Seq("id")),
+ expectedContent = Map(
+ "spark_catalog.default.a" -> Set(Map("id" -> 1)),
+ "spark_catalog.default.b" -> Set(Map("id" -> 1)),
+ "spark_catalog.default.mv" -> Set(Map("id" -> 1))))
+
+ // Clear cached pipeline execution before starting new run
+ SparkConnectService.sessionManager
+ .getIsolatedSessionIfPresent(SessionKey(defaultUserId, defaultSessionId))
+ .foreach(_.removeAllPipelineExecutions())
+
+ // Replace source data to simulate a streaming update
+ spark.sql(
+ "INSERT OVERWRITE TABLE spark_catalog.default.source_data " +
+ "SELECT * FROM VALUES (2), (3) AS t(id)")
+
+ // Run with specified refresh configuration
+ val capturedEvents = refreshConfigBuilder(graphId) match {
+ case Some(startRun) => startPipelineAndWaitForCompletion(graphId, Some(startRun))
+ case None => startPipelineAndWaitForCompletion(graphId)
+ }
+
+ // Additional validation if provided
+ eventValidation.foreach(_(capturedEvents))
+
+ // Verify final content
+ verifyMultipleTableContent(
Review Comment:
Would it make sense to use `QueryTest.checkAnswer` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on PR #51507:
URL: https://github.com/apache/spark/pull/51507#issuecomment-3079699490
@AnishMahto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211483342
##########
python/pyspark/pipelines/cli.py:
##########
@@ -283,6 +341,11 @@ def run(spec_path: Path) -> None:
else:
spec_path = find_pipeline_spec(Path.cwd())
- run(spec_path=spec_path)
+ run(
+ spec_path=spec_path,
+ full_refresh=flatten_table_lists(args.full_refresh),
Review Comment:
Why do we need to flatten `args.full_refresh` and `args.refresh`? I thought we defined their types with the `parse_table_list` function, which returns `List[str]`
##########
python/pyspark/pipelines/cli.py:
##########
@@ -242,20 +264,56 @@ def run(spec_path: Path) -> None:
register_definitions(spec_path, registry, spec)
log_with_curr_timestamp("Starting run...")
- result_iter = start_run(spark, dataflow_graph_id)
+ result_iter = start_run(
+ spark,
+ dataflow_graph_id,
+ full_refresh=full_refresh,
+ full_refresh_all=full_refresh_all,
+ refresh=refresh,
+ )
try:
handle_pipeline_events(result_iter)
finally:
spark.stop()
+def parse_table_list(value: str) -> List[str]:
+ """Parse a comma-separated list of table names, handling whitespace."""
+ return [table.strip() for table in value.split(",") if table.strip()]
+
+
+def flatten_table_lists(table_lists: Optional[List[List[str]]]) -> Optional[List[str]]:
+ """Flatten a list of lists of table names into a single list."""
+ if not table_lists:
+ return None
+ result = []
+ for table_list in table_lists:
+ result.extend(table_list)
+ return result if result else None
+
+
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Pipeline CLI")
subparsers = parser.add_subparsers(dest="command", required=True)
# "run" subcommand
run_parser = subparsers.add_parser("run", help="Run a pipeline.")
run_parser.add_argument("--spec", help="Path to the pipeline spec.")
+ run_parser.add_argument(
+ "--full-refresh",
+ type=parse_table_list,
+ action="append",
+ help="List of datasets to reset and recompute (comma-separated).",
Review Comment:
Here and below, should we document default behavior if this arg is not specified at all?
##########
python/pyspark/pipelines/cli.py:
##########
@@ -242,20 +264,56 @@ def run(spec_path: Path) -> None:
register_definitions(spec_path, registry, spec)
log_with_curr_timestamp("Starting run...")
- result_iter = start_run(spark, dataflow_graph_id)
+ result_iter = start_run(
+ spark,
+ dataflow_graph_id,
+ full_refresh=full_refresh,
+ full_refresh_all=full_refresh_all,
+ refresh=refresh,
+ )
try:
handle_pipeline_events(result_iter)
finally:
spark.stop()
+def parse_table_list(value: str) -> List[str]:
+ """Parse a comma-separated list of table names, handling whitespace."""
+ return [table.strip() for table in value.split(",") if table.strip()]
+
+
+def flatten_table_lists(table_lists: Optional[List[List[str]]]) -> Optional[List[str]]:
+ """Flatten a list of lists of table names into a single list."""
+ if not table_lists:
+ return None
+ result = []
+ for table_list in table_lists:
+ result.extend(table_list)
+ return result if result else None
Review Comment:
If result is an empty list, do we still want to return None? Or should we just return the empty list? What is the implication of either here
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
Review Comment:
Can we extract all this added logic to deduce the full refresh and regular refresh table filters into its own function? And then as part of the scala docs, map the expected filter results depending on what combination of full refresh and partial refresh is selected
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
+ // Extract refresh parameters from protobuf command
+ val fullRefreshTables = cmd.getFullRefreshList.asScala.toSeq
+ val fullRefreshAll = cmd.getFullRefreshAll
+ val refreshTables = cmd.getRefreshList.asScala.toSeq
+
+ // Convert table names to fully qualified TableIdentifier objects
+ def parseTableNames(tableNames: Seq[String]): Set[TableIdentifier] = {
+ tableNames.map { name =>
+ GraphIdentifierManager
+ .parseAndQualifyTableIdentifier(
+ rawTableIdentifier =
+ GraphIdentifierManager.parseTableIdentifier(name, sessionHolder.session),
+ currentCatalog = Some(graphElementRegistry.defaultCatalog),
+ currentDatabase = Some(graphElementRegistry.defaultDatabase))
+ .identifier
+ }.toSet
+ }
+
+ if (fullRefreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to refresh when full refresh all is set to true.")
+ }
+
+ if (refreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to full refresh when full refresh all is set to true.")
+ }
+ val refreshTableNames = parseTableNames(refreshTables)
+ val fullRefreshTableNames = parseTableNames(fullRefreshTables)
+
+ if (refreshTables.nonEmpty && fullRefreshTables.nonEmpty) {
+ // check if there is an intersection between the subset
+ val intersection = refreshTableNames.intersect(fullRefreshTableNames)
+ if (intersection.nonEmpty) {
+ throw new IllegalArgumentException(
+ "Datasets specified for refresh and full refresh cannot overlap: " +
+ s"${intersection.mkString(", ")}")
+ }
+ }
+
+ val fullRefreshTablesFilter: TableFilter = if (fullRefreshAll) {
+ AllTables
+ } else if (fullRefreshTables.nonEmpty) {
+ SomeTables(fullRefreshTableNames)
+ } else {
+ NoTables
+ }
+
+ val refreshTablesFilter: TableFilter =
+ if (refreshTables.nonEmpty) {
+ SomeTables(refreshTableNames)
+ } else if (fullRefreshTablesFilter != NoTables) {
+ NoTables
+ } else {
+ AllTables
+ }
Review Comment:
just an optional nit, but as a code reader it's difficult for me to reason about the combinations of `fullRefreshTables` and `refreshTables` when reading them as sequential but related validation here.
My suggestion would be to restructure this as a match statement, that explicitly handles each combination. Ex.
```
(fullRefreshTables, refreshTableNames) match {
case (Nil, Nil) => ...
case (fullRefreshTables, Nil) => ...
case ...
}
```
##########
sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto:
##########
@@ -116,6 +116,15 @@ message PipelineCommand {
message StartRun {
// The graph to start.
optional string dataflow_graph_id = 1;
+
+ // List of tables to reset and recompute.
+ repeated string full_refresh = 2;
+
+ // Perform a full graph reset and recompute.
+ optional bool full_refresh_all = 3;
+
+ // List of tables to update.
+ repeated string refresh = 4;
Review Comment:
```suggestion
repeated string refresh_selection = 4;
```
##########
sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto:
##########
@@ -116,6 +116,15 @@ message PipelineCommand {
message StartRun {
// The graph to start.
optional string dataflow_graph_id = 1;
+
+ // List of tables to reset and recompute.
+ repeated string full_refresh = 2;
Review Comment:
```suggestion
repeated string full_refresh_selection = 2;
```
##########
python/pyspark/pipelines/cli.py:
##########
@@ -217,8 +217,30 @@ def change_dir(path: Path) -> Generator[None, None, None]:
os.chdir(prev)
-def run(spec_path: Path) -> None:
- """Run the pipeline defined with the given spec."""
+def run(
+ spec_path: Path,
+ full_refresh: Optional[Sequence[str]] = None,
+ full_refresh_all: bool = False,
+ refresh: Optional[Sequence[str]] = None,
+) -> None:
+ """Run the pipeline defined with the given spec.
+
+ :param spec_path: Path to the pipeline specification file.
+ :param full_refresh: List of datasets to reset and recompute.
+ :param full_refresh_all: Perform a full graph reset and recompute.
+ :param refresh: List of datasets to update.
+ """
+ # Validate conflicting arguments
+ if full_refresh_all:
+ if full_refresh:
+ raise PySparkException(
+ errorClass="CONFLICTING_PIPELINE_REFRESH_OPTIONS", messageParameters={}
Review Comment:
Thoughts on having sub error classes for mismatched combinations? Or maybe just pass along which two configs are conflicting as a message parameter?
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
+ // Extract refresh parameters from protobuf command
+ val fullRefreshTables = cmd.getFullRefreshList.asScala.toSeq
+ val fullRefreshAll = cmd.getFullRefreshAll
+ val refreshTables = cmd.getRefreshList.asScala.toSeq
+
+ // Convert table names to fully qualified TableIdentifier objects
+ def parseTableNames(tableNames: Seq[String]): Set[TableIdentifier] = {
+ tableNames.map { name =>
+ GraphIdentifierManager
+ .parseAndQualifyTableIdentifier(
+ rawTableIdentifier =
+ GraphIdentifierManager.parseTableIdentifier(name, sessionHolder.session),
+ currentCatalog = Some(graphElementRegistry.defaultCatalog),
+ currentDatabase = Some(graphElementRegistry.defaultDatabase))
+ .identifier
+ }.toSet
+ }
+
+ if (fullRefreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to refresh when full refresh all is set to true.")
+ }
+
+ if (refreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to full refresh when full refresh all is set to true.")
+ }
Review Comment:
Are the exception messages flipped here?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineUpdateContextImpl.scala:
##########
@@ -24,10 +24,17 @@ import org.apache.spark.sql.pipelines.logging.{FlowProgressEventLogger, Pipeline
* An implementation of the PipelineUpdateContext trait used in production.
* @param unresolvedGraph The graph (unresolved) to be executed in this update.
* @param eventCallback A callback function to be called when an event is added to the event buffer.
+ * @param refreshTables Filter for which tables should be refreshed when performing this update.
+ * @param fullRefreshTables Filter for which tables should be full refreshed
+ * when performing this update.
+ * @param resetCheckpointFlows Filter for which flows should be reset.
*/
class PipelineUpdateContextImpl(
override val unresolvedGraph: DataflowGraph,
- override val eventCallback: PipelineEvent => Unit
+ override val eventCallback: PipelineEvent => Unit,
+ override val refreshTables: TableFilter = AllTables,
+ override val fullRefreshTables: TableFilter = NoTables,
+ override val resetCheckpointFlows: FlowFilter = NoFlows
Review Comment:
Also is it possible this argument always being `NoFlows` now can logically conflict with the `fullRefreshTables` argument, which is no longer always `NoTables`?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineUpdateContextImpl.scala:
##########
@@ -24,10 +24,17 @@ import org.apache.spark.sql.pipelines.logging.{FlowProgressEventLogger, Pipeline
* An implementation of the PipelineUpdateContext trait used in production.
* @param unresolvedGraph The graph (unresolved) to be executed in this update.
* @param eventCallback A callback function to be called when an event is added to the event buffer.
+ * @param refreshTables Filter for which tables should be refreshed when performing this update.
+ * @param fullRefreshTables Filter for which tables should be full refreshed
+ * when performing this update.
+ * @param resetCheckpointFlows Filter for which flows should be reset.
*/
class PipelineUpdateContextImpl(
override val unresolvedGraph: DataflowGraph,
- override val eventCallback: PipelineEvent => Unit
+ override val eventCallback: PipelineEvent => Unit,
+ override val refreshTables: TableFilter = AllTables,
+ override val fullRefreshTables: TableFilter = NoTables,
+ override val resetCheckpointFlows: FlowFilter = NoFlows
Review Comment:
Why are we moving `resetCheckpointFlows` into the constructor arg for `PipelineUpdateContextImpl`?
Not saying I disagree with this decision, but since the argument doesn't seem to be used by any callers in this PR I'm curious when `resetCheckpointFlows` will not be the default value of `NoFlows`.
##########
python/pyspark/pipelines/spark_connect_pipeline.py:
##########
@@ -65,12 +65,26 @@ def handle_pipeline_events(iter: Iterator[Dict[str, Any]]) -> None:
log_with_provided_timestamp(event.message, dt)
-def start_run(spark: SparkSession, dataflow_graph_id: str) -> Iterator[Dict[str, Any]]:
+def start_run(
+ spark: SparkSession,
+ dataflow_graph_id: str,
+ full_refresh: Optional[Sequence[str]] = None,
+ full_refresh_all: bool = False,
+ refresh: Optional[Sequence[str]] = None,
+) -> Iterator[Dict[str, Any]]:
"""Start a run of the dataflow graph in the Spark Connect server.
:param dataflow_graph_id: The ID of the dataflow graph to start.
+ :param full_refresh: List of datasets to reset and recompute.
+ :param full_refresh_all: Perform a full graph reset and recompute.
+ :param refresh: List of datasets to update.
"""
- inner_command = pb2.PipelineCommand.StartRun(dataflow_graph_id=dataflow_graph_id)
+ inner_command = pb2.PipelineCommand.StartRun(
+ dataflow_graph_id=dataflow_graph_id,
+ full_refresh=full_refresh or [],
Review Comment:
I'm a little confused by this. If `full_refresh` is None, we return an empty list. But in `flatten_table_lists`, which we use to construct this `full_refresh` list, we return None if the flattened list is empty. Can we simplify the logic here or there?
##########
python/pyspark/pipelines/cli.py:
##########
@@ -217,8 +217,30 @@ def change_dir(path: Path) -> Generator[None, None, None]:
os.chdir(prev)
-def run(spec_path: Path) -> None:
- """Run the pipeline defined with the given spec."""
+def run(
Review Comment:
High level question: did we consider putting refresh selection options in the pipeline spec, rather than as a CLI arg?
More generally, what's the philosophy for whether a configuration should be accepted as a CLI arg vs a pipeline spec field?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213864934
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
Review Comment:
extracted a `createTableFilters` function
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213866316
##########
python/pyspark/pipelines/cli.py:
##########
@@ -28,7 +28,7 @@
import yaml
from dataclasses import dataclass
from pathlib import Path
-from typing import Any, Generator, Mapping, Optional, Sequence
+from typing import Any, Generator, Mapping, Optional, Sequence, List
Review Comment:
actually it didn't reformat this but I manually reordered it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211691116
##########
python/pyspark/pipelines/cli.py:
##########
@@ -283,6 +341,11 @@ def run(spec_path: Path) -> None:
else:
spec_path = find_pipeline_spec(Path.cwd())
- run(spec_path=spec_path)
+ run(
+ spec_path=spec_path,
+ full_refresh=flatten_table_lists(args.full_refresh),
Review Comment:
This is for the case if user provide the same args multiple times.
Ex: (`--full_refresh: "a,b"` `--full_refresh: "c,d"`). Then we will receive a nested list `[["a","b"],["c"]]`. Need to perform a flattening to transform it into a 1D list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211719750
##########
python/pyspark/pipelines/cli.py:
##########
@@ -283,6 +341,11 @@ def run(spec_path: Path) -> None:
else:
spec_path = find_pipeline_spec(Path.cwd())
- run(spec_path=spec_path)
+ run(
+ spec_path=spec_path,
+ full_refresh=flatten_table_lists(args.full_refresh),
Review Comment:
Ah got it, makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "sryza (via GitHub)" <gi...@apache.org>.
sryza commented on PR #51507:
URL: https://github.com/apache/spark/pull/51507#issuecomment-3085268665
Merged to master
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213865684
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -224,6 +225,64 @@ private[connect] object PipelinesHandler extends Logging {
sessionHolder: SessionHolder): Unit = {
val dataflowGraphId = cmd.getDataflowGraphId
val graphElementRegistry = DataflowGraphRegistry.getDataflowGraphOrThrow(dataflowGraphId)
+
+ // Extract refresh parameters from protobuf command
+ val fullRefreshTables = cmd.getFullRefreshList.asScala.toSeq
+ val fullRefreshAll = cmd.getFullRefreshAll
+ val refreshTables = cmd.getRefreshList.asScala.toSeq
+
+ // Convert table names to fully qualified TableIdentifier objects
+ def parseTableNames(tableNames: Seq[String]): Set[TableIdentifier] = {
+ tableNames.map { name =>
+ GraphIdentifierManager
+ .parseAndQualifyTableIdentifier(
+ rawTableIdentifier =
+ GraphIdentifierManager.parseTableIdentifier(name, sessionHolder.session),
+ currentCatalog = Some(graphElementRegistry.defaultCatalog),
+ currentDatabase = Some(graphElementRegistry.defaultDatabase))
+ .identifier
+ }.toSet
+ }
+
+ if (fullRefreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to refresh when full refresh all is set to true.")
+ }
+
+ if (refreshTables.nonEmpty && fullRefreshAll) {
+ throw new IllegalArgumentException(
+ "Cannot specify a subset to full refresh when full refresh all is set to true.")
+ }
+ val refreshTableNames = parseTableNames(refreshTables)
+ val fullRefreshTableNames = parseTableNames(fullRefreshTables)
+
+ if (refreshTables.nonEmpty && fullRefreshTables.nonEmpty) {
+ // check if there is an intersection between the subset
+ val intersection = refreshTableNames.intersect(fullRefreshTableNames)
+ if (intersection.nonEmpty) {
+ throw new IllegalArgumentException(
+ "Datasets specified for refresh and full refresh cannot overlap: " +
+ s"${intersection.mkString(", ")}")
+ }
+ }
+
+ val fullRefreshTablesFilter: TableFilter = if (fullRefreshAll) {
+ AllTables
+ } else if (fullRefreshTables.nonEmpty) {
+ SomeTables(fullRefreshTableNames)
+ } else {
+ NoTables
+ }
+
+ val refreshTablesFilter: TableFilter =
+ if (refreshTables.nonEmpty) {
+ SomeTables(refreshTableNames)
+ } else if (fullRefreshTablesFilter != NoTables) {
+ NoTables
+ } else {
+ AllTables
+ }
Review Comment:
Replaced with match statement in the new `createTableFilters` function
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2213860432
##########
python/pyspark/pipelines/cli.py:
##########
@@ -217,8 +217,30 @@ def change_dir(path: Path) -> Generator[None, None, None]:
os.chdir(prev)
-def run(spec_path: Path) -> None:
- """Run the pipeline defined with the given spec."""
+def run(
+ spec_path: Path,
+ full_refresh: Optional[Sequence[str]] = None,
+ full_refresh_all: bool = False,
+ refresh: Optional[Sequence[str]] = None,
+) -> None:
+ """Run the pipeline defined with the given spec.
+
+ :param spec_path: Path to the pipeline specification file.
+ :param full_refresh: List of datasets to reset and recompute.
+ :param full_refresh_all: Perform a full graph reset and recompute.
+ :param refresh: List of datasets to update.
+ """
+ # Validate conflicting arguments
+ if full_refresh_all:
+ if full_refresh:
+ raise PySparkException(
+ errorClass="CONFLICTING_PIPELINE_REFRESH_OPTIONS", messageParameters={}
Review Comment:
Added [logic](https://github.com/apache/spark/pull/51507/commits/1693ac546225c8a6be1d96eb5e64fcf03f77a344#diff-273b1e23d9cb13f05f8167735c9571b549972f8b5270ad2aba3bc4a92e8b9766R179) to pass along the conflicting option
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "sryza (via GitHub)" <gi...@apache.org>.
sryza closed pull request #51507: [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options
URL: https://github.com/apache/spark/pull/51507
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on PR #51507:
URL: https://github.com/apache/spark/pull/51507#issuecomment-3079855544
@sryza
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211741560
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineUpdateContextImpl.scala:
##########
@@ -24,10 +24,17 @@ import org.apache.spark.sql.pipelines.logging.{FlowProgressEventLogger, Pipeline
* An implementation of the PipelineUpdateContext trait used in production.
* @param unresolvedGraph The graph (unresolved) to be executed in this update.
* @param eventCallback A callback function to be called when an event is added to the event buffer.
+ * @param refreshTables Filter for which tables should be refreshed when performing this update.
+ * @param fullRefreshTables Filter for which tables should be full refreshed
+ * when performing this update.
+ * @param resetCheckpointFlows Filter for which flows should be reset.
*/
class PipelineUpdateContextImpl(
override val unresolvedGraph: DataflowGraph,
- override val eventCallback: PipelineEvent => Unit
+ override val eventCallback: PipelineEvent => Unit,
+ override val refreshTables: TableFilter = AllTables,
+ override val fullRefreshTables: TableFilter = NoTables,
+ override val resetCheckpointFlows: FlowFilter = NoFlows
Review Comment:
I see, so we do a union of this argument and the deduced flow refreshes from `refreshTables` and `fullRefreshTables`.
Why do we need `resetCheckpointFlows`? Is it possible it was added before solely for testing, and its redundant now because we're propagating `refreshTables` and `fullRefreshTables`? Or is there any non-test use case for this field? If not, we should remove it and refactor current references
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK- 52810][SDP][SQL] Spark Pipelines CLI Selection Options [spark]
Posted by "JiaqiWang18 (via GitHub)" <gi...@apache.org>.
JiaqiWang18 commented on code in PR #51507:
URL: https://github.com/apache/spark/pull/51507#discussion_r2211686949
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineUpdateContextImpl.scala:
##########
@@ -24,10 +24,17 @@ import org.apache.spark.sql.pipelines.logging.{FlowProgressEventLogger, Pipeline
* An implementation of the PipelineUpdateContext trait used in production.
* @param unresolvedGraph The graph (unresolved) to be executed in this update.
* @param eventCallback A callback function to be called when an event is added to the event buffer.
+ * @param refreshTables Filter for which tables should be refreshed when performing this update.
+ * @param fullRefreshTables Filter for which tables should be full refreshed
+ * when performing this update.
+ * @param resetCheckpointFlows Filter for which flows should be reset.
*/
class PipelineUpdateContextImpl(
override val unresolvedGraph: DataflowGraph,
- override val eventCallback: PipelineEvent => Unit
+ override val eventCallback: PipelineEvent => Unit,
+ override val refreshTables: TableFilter = AllTables,
+ override val fullRefreshTables: TableFilter = NoTables,
+ override val resetCheckpointFlows: FlowFilter = NoFlows
Review Comment:
`resetCheckpointFlows` was accidentally moved into the constructor. I will move it back.
It being set to `NoTables` won't conflict with the `fullRefreshTables` argument because [this logic](https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineUpdateContext.scala#L41) will include the flows for the full refresh tables directly so we don't need to set it manually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org