anew commented on code in PR #56686:
URL: https://github.com/apache/spark/pull/56686#discussion_r3553106516
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -342,11 +351,127 @@ object DatasetManager extends Logging {
)
}
- table.copy(
- normalizedPath = Option(
-
catalog.loadTable(identifier).properties().get(TableCatalog.PROP_LOCATION)
+ val catalogTableEntity = catalog.loadTable(identifier)
+ val tableWithMaterializationMetadata =
+ table.copy(
+ normalizedPath =
+
Option(catalogTableEntity.properties().get(TableCatalog.PROP_LOCATION))
)
+
+ (tableWithMaterializationMetadata, catalogTableEntity)
+ }
+
+ /**
+ * Validate that the AutoCDC target table is backed by a connector
implementing
+ * [[SupportsRowLevelOperations]], the DSv2 contract for the
MERGE/UPDATE/DELETE-with-rewrite
+ * operations the AutoCDC transformation relies on. Reuses the target handle
already loaded by
+ * [[materializeTable]], so it performs no additional catalog I/O. Only
AutoCDC auxiliary specs
+ * carry a MERGE-backed target; other auxiliary tables have no such
requirement.
+ *
+ * @param targetTable the target table graph entity, source of
the identifier and
+ * declared format used in the error message.
+ * @param targetTableCatalogEntity the target table's loaded DSv2 handle.
+ */
+ private def requireAutoCdcTargetSupportsRowLevelOps(
+ targetTable: Table,
+ targetTableCatalogEntity: V2Table): Unit = {
+ if (!targetTableCatalogEntity.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> targetTable.identifier.quotedString,
+ // Prefer the flow-declared format, falling back to the connector's
provider property.
+ "format" -> targetTable.format
+
.orElse(Option(targetTableCatalogEntity.properties.get(TableCatalog.PROP_PROVIDER)))
+ .getOrElse("<unknown>")
+ )
+ )
+ }
+ }
+
+ /**
+ * Materialize the auxiliary table according to the provided spec.
+ *
+ * @param auxiliaryTableSpec the spec describing the auxiliary table to
create/evolve.
+ * @param isFullRefresh whether the owning table is being fully refreshed.
+ * @param context the context for the pipeline update.
+ */
+ private def materializeAuxiliaryTable(
+ auxiliaryTableSpec: AuxiliaryTableSpec,
+ isFullRefresh: Boolean,
+ context: PipelineUpdateContext): Unit = {
+ // Get the DSv2 catalog handler and identifier for the aux table.
+ val (catalog, auxiliaryTableIdentifier) =
+ PipelinesCatalogUtils.resolveTableCatalog(context.spark,
auxiliaryTableSpec.identifier)
+
+ logInfo(
+ log"Materializing auxiliary table " +
+ log"${MDC(LogKeys.TABLE_NAME, auxiliaryTableSpec.identifier)}."
)
+
+ if (isFullRefresh) {
+ // Intentionally DROP and not TRUNCATE on full refresh. The auxiliary
table is an internal
Review Comment:
I learned:
DROP is intentional here (see the comment just above at the full-refresh
branch). The aux table's identity and its metadata -- SCD type and recorded
key
columns in table properties -- must not persist across a full refresh, since
a
full refresh is precisely how a user changes them. TRUNCATE would keep the
stale
properties and force extra reconciliation. It's an internal state table, so
history preservation / undo isn't a goal for it; the owning target table
still
uses TRUNCATE and retains its history.
--
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]
