AnishMahto commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3307072436
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -303,6 +303,25 @@ 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
across AutoCDC and
+ // therefore cannot collide with a user-managed table.
+
+ // Intentionally DROP and not TRUNCATE for two reasons; First, the
auxiliary table may
+ // contain table properties that represent stateful information (ex. SCD
key count) that
+ // should not be carried forward on a full refresh. Second, the
auxiliary table is an
+ // internal table and not part of the dataflow graph. That means it does
not go through
+ // schema evolution like other tables and hence on a full refresh, we
should explicitly
+ // drop the existing auxiliary table schema so it can be recomputed.
+
+ val auxiliaryTableId = AutoCdcAuxiliaryTable.identifier(table.identifier)
+ context.spark.sql(s"DROP TABLE IF EXISTS
${auxiliaryTableId.quotedString}")
Review Comment:
Yeah this is a pretty fair concern, it's something I've been thinking about
too.
The AutoCDC flow is the first type of flow to introduce the concept of an
internal state table.
Streaming queries/flows have internal state, but that's managed in the form
of checkpoint directories, not catalog table entities. And while its an
intentional decision to make the AutoCDC auxiliary tables real catalog tables
(ex. to inherit catalog based permission model and other functionality), it
also means the pipeline needs to _manage_ those internal tables in a similar
fashion to actual destination tables.
I'm not too worried about compounding behavior with SCD2, but I agree
there's probably a better data model here that we should eventually refactor
to. Either an `onFullRefresh` hook like you mentioned, or maybe just add
something like `internalTableIdentifiers: Seq[TableIdentifier]` to the flow
interface, and then have `DatasetManager` do the same full refresh and schema
evolution logic on the internal tables as it does for actual destination tables.
But in either case, these are pure refactorings and don't affect user
observed behavior at all. Let's revisit this in the future as SCD2 lands, so
that we don't end up prematurely choosing a refactor path that doesn't actually
fit well.
--
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]
