cloud-fan commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3308075122
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala:
##########
@@ -73,16 +74,35 @@ class FlowPlanner(
trigger = triggerFor(sf),
checkpointPath = flowMetadata.latestCheckpointLocation
)
- case _ =>
- throw new UnsupportedOperationException(
- s"Unsupported destination type: ${output.getClass.getSimpleName}
for " +
- s"streaming flow ${sf.identifier}
(${flow.destinationIdentifier})"
+ case _ => unsupportedDestinationType(sf, output)
+ }
+ case acmf: AutoCdcMergeFlow if acmf.changeArgs.storedAsScdType ==
ScdType.Type1 =>
Review Comment:
This guarded case (Type1 only) plus the file-level catch-all means that for
any other SCD type — SCD2 today, hypothetically SCD3 tomorrow — the planner
surfaces `Unable to plan flow of type AutoCdcMergeFlow`, which is misleading:
the flow type *is* supported, only the SCD variant isn't. The reason this is
fine today is that `AutoCdcMergeFlow.schema` (a `val`) eagerly throws
`AUTOCDC_SCD2_NOT_SUPPORTED` at flow construction, so SCD2 never reaches the
planner. That coupling is invisible from `FlowPlanner` and would silently break
if `schema` ever stops being eagerly evaluated or a new SCD type is added
without an eager guard.
Making the SCD dispatch explicit at the planner — e.g. a nested match on
`acmf.changeArgs.storedAsScdType` inside the `AutoCdcMergeFlow` arm with an
explicit `case ScdType.Type2 => throw AUTOCDC_SCD2_NOT_SUPPORTED` — keeps
planner-level errors useful regardless of how the flow class evolves.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -303,6 +303,23 @@ object DatasetManager extends Logging {
context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
}
+ if (isFullRefresh) {
+ // On full refresh, drop the AutoCDC auxiliary state associated with
this table (if any) so
+ // that stale delete-tracking data and table properties are not carried
forward into the new
+ // table generation. We unconditionally issue the DROP for every
fully-refreshed target; for
+ // non-AutoCDC tables this is a no-op because
[[AutoCdcAuxiliaryTable.identifier]] derives
+ // its name from [[AutoCdcReservedNames.prefix]], which is reserved by
AutoCDC and
+ // therefore cannot collide with a user-managed table.
Review Comment:
**Reserved prefix is asserted, not enforced.** This comment ('reserved by
AutoCDC and therefore cannot collide with a user-managed table') is what makes
the unconditional `DROP IF EXISTS` below safe — and what makes
`createAuxiliaryTableIfNotExists` safe against silently adopting a stale user
table. The PR hoisted `AutoCdcReservedNames.prefix` and uses it to reject
*column* names via
`AutoCdcMergeFlow.requireReservedPrefixAbsentInSourceColumns`, but the
symmetric reservation for *table* names is not enforced anywhere. A user who
happens to have created `${catalog}.${db}.__spark_autocdc_aux_state_<target>`
would either be silently dropped on full refresh or silently adopted as an aux
table.
Either (a) soften the wording ("… by convention …") and acknowledge that
AutoCDC trusts user tables not to use this prefix, or (b) add a
graph-registration-time check that rejects any user-registered table name
starting with `AutoCdcReservedNames.prefix`. This concern compounds with the
layer-mixing thread on this same block.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala:
##########
@@ -0,0 +1,241 @@
+/*
+ * 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.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ ColumnSelection,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests covering AutoCDC's full-refresh semantics: full refresh must wipe
both the
+ * target rows and the AutoCDC auxiliary table for the refreshed targets, and
must leave
+ * non-refreshed targets untouched in selective-refresh mode.
+ */
+class AutoCdcScd1FullRefreshSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ test("full refresh wipes target rows and the auxiliary table for the
refreshed flow") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ // Run #1: populate target + auxiliary table.
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "alice", 5L))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream1.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ runPipeline(ctx1)
+ assert(
+ spark.catalog.tableExists(auxTableNameFor("target")),
+ "Auxiliary table should exist after first run"
+ )
+
+ // Run #2 (full refresh): auxiliary table should be dropped by
DatasetManager, target
+ // truncated. The new run brings only id=2 at seq=1.
+ val stream2 = MemoryStream[(Int, String, Long)]
+ stream2.addData((2, "bob", 1L))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ val updateCtx = TestPipelineUpdateContext(
+ spark,
+ ctx2.toDataflowGraph,
+ storageRoot,
+ fullRefreshTables = AllTables
+ )
+ updateCtx.pipelineExecution.runPipeline()
+ updateCtx.pipelineExecution.awaitCompletion()
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(2, "bob", 1L, cdcMeta(None, Some(1L))))
+ )
+ }
+
+ test("after a full refresh, an event with a sequence below the previous
run's " +
+ "watermark now lands") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ // Run #1: delete at seq=10 sets a high watermark in the auxiliary table.
+ val stream1 = MemoryStream[(Int, String, Long, Boolean)]
+ stream1.addData((1, "alice", 10L, true))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream1.toDF().toDF("id", "name", "version",
"is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ ))
+ ))
+ }
+ runPipeline(ctx1)
+
+ // Run #2 (full refresh): auxiliary table is dropped, watermark reset.
seq=5 should
+ // now land.
+ val stream2 = MemoryStream[(Int, String, Long, Boolean)]
+ stream2.addData((1, "fresh", 5L, false))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("id", "name", "version",
"is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ ))
+ ))
+ }
+ val updateCtx = TestPipelineUpdateContext(
+ spark,
+ ctx2.toDataflowGraph,
+ storageRoot,
+ fullRefreshTables = AllTables
+ )
+ updateCtx.pipelineExecution.runPipeline()
+ updateCtx.pipelineExecution.awaitCompletion()
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "fresh", 5L, cdcMeta(None, Some(5L))))
+ )
+ }
+
+ test("selective full refresh wipes only the requested target's auxiliary
state") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_a " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_b " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+
+ // Run #1: populate both targets at seq=10.
+ val streamA1 = MemoryStream[(Int, Long)]
+ val streamB1 = MemoryStream[(Int, Long)]
+ streamA1.addData((1, 10L))
+ streamB1.addData((1, 10L))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("t_a", catalog = Some(catalog), database = Some(namespace))
+ registerTable("t_b", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "flow_a",
+ target = "t_a",
+ query = dfFlowFunc(streamA1.toDF().toDF("id", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ registerFlow(autoCdcFlow(
+ name = "flow_b",
+ target = "t_b",
+ query = dfFlowFunc(streamB1.toDF().toDF("id", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ runPipeline(ctx1)
+
+ // Run #2: full refresh ONLY on t_a; t_b's auxiliary state must persist.
+ val streamA2 = MemoryStream[(Int, Long)]
+ val streamB2 = MemoryStream[(Int, Long)]
Review Comment:
This test creates a fresh `streamB2` for run #2, but `t_b` is **not**
full-refreshed (`fullRefreshTables = SomeTables(Set(t_a))` on L224-226), so its
streaming checkpoint should resume from `streamB1`. The peer suites use a
single `MemoryStream` reused across runs with an explicit comment — see
`AutoCdcScd1AuxiliaryTableDurabilitySuite.scala:51`, `:99`, `:209` ("Single
MemoryStream reused across both runs so the streaming checkpoint can resume
cleanly") and `AutoCdcScd1SchemaEvolutionSuite.scala:54`, `:207`.
The assertion at L237-238 (seq=5 dropped) could pass either because the aux
watermark suppressed it (the intent) or because the streaming source didn't
resume against the new `MemoryStream` and seq=5 was never processed at all.
Suggest reusing a single `streamB` across both runs (with `streamA1`/`streamA2`
left distinct since `t_a` is full-refreshed and its checkpoint is reset).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
