You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "AnishMahto (via GitHub)" <gi...@apache.org> on 2026/06/22 23:49:37 UTC
[PR] [SPARK-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
AnishMahto opened a new pull request, #56684:
URL: https://github.com/apache/spark/pull/56684
<!--
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?
Step 1/2 for managing auxiliary table lifecycle in `DatasetManager`. Pure refactor that extracts catalog table creation and evolution logic into their own helpers so that auxiliary tables can reuse them.
### Why are the changes needed?
Refactor before implementing auxiliary table full refresh, creation, and schema evolution in `DatasetManager`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8.
--
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-57625][SDP] Manage Auxiliary Table Lifecycle in DatasetManager 1/2 [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #56684: [SPARK-57625][SDP] Manage Auxiliary Table Lifecycle in DatasetManager 1/2
URL: https://github.com/apache/spark/pull/56684
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3465641543
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,80 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[V2Table] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTable(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema
+ * and properties, skipping the catalog `alterTable` entirely when nothing actually changes.
+ * Partitioning/clustering cannot change in place, so no transforms are accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTable(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: V2Table,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+
+ val existingProperties = existingTable.properties()
+ val propertyChanges = properties.collect {
Review Comment:
This is actually a really good question, had to spend some time thinking about it:
1. No, AFAICT we do not actually handle property removal whatsoever. If a user drops properties specified in the `dp.table` decorator, it will not actually drop the property from the catalog table. The way that its currently implemented, properties can only be added and not removed
2. Agree this is not changed by this PR, but this is definitely a bug
3. Fixing this is non-trivial. Originally I wanted to simply diff the existing table properties and the declared table properties for the current pipeline run. But I realized there's a number of table properties that are automatically set by the catalog and various API, even if the user has not explicitly declared them using the `dp.table` decorator (ex. `clusteringColumns `). This means that even if we identify table properties set on the existing catalog table entry but not declared in the decorator for the current pipeline run, we can't actually differentiate between whether that property was previously explicitly set by the user OR if the catalog set it internally.
Fixing this bug is difficult, I think a true fix might require persisting the previously declared table properties somewhere. What I'm going to do for this PR is create a Jira issue and link it here as a TODO, but I'm not going to attempt to make that fix right now.
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3463651411
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
+ * accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTableInCatalog(
Review Comment:
I don't feel strongly, but used `evolveTable` instead because it felt more natural to what is happening conceptually - we are reconciling the existing table schema with newly declared schema, hence "evolve".
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3470301673
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
+ * accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: CatalogTable,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+ val setProperties = properties.map { case (k, v) => TableChange.setProperty(k, v) }
+ catalog.alterTable(tableIdentifier, (columnChanges ++ setProperties).toArray: _*)
Review Comment:
I ended up adding unit tests using a stubbed catalog.
--
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-57625][SDP] Manage Auxiliary Table Lifecycle in DatasetManager 1/2 [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #56684:
URL: https://github.com/apache/spark/pull/56684#issuecomment-4859156243
btw, renamed the pr as the initial one was somehow not accepted by the merge script
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3463646724
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
+ * accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: CatalogTable,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+ val setProperties = properties.map { case (k, v) => TableChange.setProperty(k, v) }
+ catalog.alterTable(tableIdentifier, (columnChanges ++ setProperties).toArray: _*)
Review Comment:
Yeah good point. I meant for this to be a pure refactor, but you're right that no-op alters will happen if nothing is really changing between pipeline runs.
Added a diff check to only issue an alter when needed.
Existing schema evolution/table property tests should test the op-path here. Not sure if we can easily test the no-op path, maybe by stubbing `catalog.alterTable` somehow.
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3463646724
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
+ * accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: CatalogTable,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+ val setProperties = properties.map { case (k, v) => TableChange.setProperty(k, v) }
+ catalog.alterTable(tableIdentifier, (columnChanges ++ setProperties).toArray: _*)
Review Comment:
Yeah good point. I meant for this to be a pure refactor, but you're right that no-op alters will happen if nothing is really changing between pipeline runs.
Added a diff check to only issue an alter when needed.
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3507912229
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala:
##########
@@ -1083,4 +1091,156 @@ abstract class MaterializeTablesSuite extends BaseCoreExecutionTest {
}
assert(ex.cause.isInstanceOf[AnalysisException])
}
+
+ // =============== Table evolution in catalog tests ===============
+
+ private val recordingCatalogName = "recording_cat"
+ private val recordingNamespace = "rec_ns"
+
+ /**
+ * Registers [[RecordingInMemoryTableCatalog]] under `recordingCatalogName`, resets its shared
+ * alter buffer and any cached catalog state, creates `recordingNamespace`, runs `body`, then
+ * tears the registration back down.
+ */
+ private def withRecordingCatalog(body: => Unit): Unit = {
+ spark.conf.set(
+ s"spark.sql.catalog.$recordingCatalogName",
+ classOf[RecordingInMemoryTableCatalog].getName
+ )
+ RecordingInMemoryTableCatalog.reset()
+ try {
+ spark.sql(s"CREATE NAMESPACE IF NOT EXISTS $recordingCatalogName.$recordingNamespace")
+ body
+ } finally {
+ RecordingInMemoryTableCatalog.reset()
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$recordingCatalogName")
+ }
+ }
+
+ /**
+ * Materializes a single streaming table under the recording catalog/namespace with the given
+ * schema and properties.
+ */
+ private def materializeStreamingTable(
+ name: String,
+ schema: StructType,
+ properties: Map[String, String]): Unit = {
+ // All nulls dummy row, compatible with any schema type
+ val row = Row.fromSeq(Seq.fill(schema.length)(null))
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(row)), schema)
+ materializeGraph(
+ new TestGraphRegistrationContext(spark) {
+ registerTable(
+ name,
+ query = Option(dfFlowFunc(df)),
+ specifiedSchema = Option(schema),
+ properties = properties,
+ catalog = Option(recordingCatalogName),
+ database = Option(recordingNamespace)
+ )
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+ }
+
+ private def loadTableFromRecordingCatalog(name: String): V2Table = {
+ val catalog = spark.sessionState.catalogManager
+ .catalog(recordingCatalogName)
+ .asInstanceOf[TableCatalog]
+ catalog.loadTable(Identifier.of(Array(recordingNamespace), name))
+ }
+
+ test("re-materializing an unchanged table does not issue an alterTable") {
+ withRecordingCatalog {
+ val schema = new StructType().add("id", IntegerType).add("value", StringType)
+ val props = Map("p.a" -> "1", "p.b" -> "2")
+ // Creating the table issues no alter, and re-materializing the unchanged table is a no-op,
+ // so no alter is ever recorded.
+ materializeStreamingTable("t", schema, props)
+ assert(RecordingInMemoryTableCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable("t", schema, props)
+ assert(RecordingInMemoryTableCatalog.recordedAlters.isEmpty)
+ }
+ }
+
+ test("re-materializing with changed/new properties issues an alterTable that sets them") {
+ withRecordingCatalog {
+ val schema = new StructType().add("id", IntegerType).add("value", StringType)
+ // Creating the table issues no alter; re-materializing with changed/added properties issues
+ // exactly one alter that sets them.
+ materializeStreamingTable("t", schema, Map("p.a" -> "1"))
+ assert(RecordingInMemoryTableCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable("t", schema, Map("p.a" -> "2", "p.new" -> "n"))
+ assert(RecordingInMemoryTableCatalog.recordedAlters.size == 1)
+
+ val changes = RecordingInMemoryTableCatalog.recordedAlters.flatten
+ assert(changes.forall(_.isInstanceOf[TableChange.SetProperty]))
+ val set = changes.collect {
+ case s: TableChange.SetProperty => s.property() -> s.value()
+ }.toMap
+ assert(set == Map("p.a" -> "2", "p.new" -> "n"))
+
+ val table = loadTableFromRecordingCatalog("t")
+ assert(table.properties().get("p.a") == "2")
+ assert(table.properties().get("p.new") == "n")
+ }
+ }
+
+ test("re-materializing with an added column issues an alterTable") {
+ withRecordingCatalog {
+ // Creating the table issues no alter; re-materializing with an added column issues exactly
+ // one alter that adds it.
+ materializeStreamingTable("t", new StructType().add("id", IntegerType), Map("p.a" -> "1"))
+ assert(RecordingInMemoryTableCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable(
+ "t",
+ new StructType().add("id", IntegerType).add("value", StringType),
+ Map("p.a" -> "1")
+ )
+ assert(RecordingInMemoryTableCatalog.recordedAlters.size == 1)
+
+ val changes = RecordingInMemoryTableCatalog.recordedAlters.flatten
+ assert(changes.exists(_.isInstanceOf[TableChange.AddColumn]))
+
+ assert(
+ loadTableFromRecordingCatalog("t").columns() sameElements
+ CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("value", StringType)
+ )
+ )
+ }
+ }
+
+ test("re-materializing with a dropped property neither removes it nor issues an alterTable") {
+ withRecordingCatalog {
+ val schema = new StructType().add("id", IntegerType)
+ // This test locks in the current buggy behavior where dropped properties do not materialize
+ // against the catalog table entity. See SPARK-57670.
+ materializeStreamingTable("t", schema, Map("p.keep" -> "v", "p.stale" -> "old"))
+ assert(RecordingInMemoryTableCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable("t", schema, Map("p.keep" -> "v"))
+ assert(RecordingInMemoryTableCatalog.recordedAlters.isEmpty)
+
+ assert(loadTableFromRecordingCatalog("t").properties().get("p.stale") == "old")
+ }
+ }
+}
+
+/**
+ * An [[InMemoryTableCatalog]] that records every `alterTable` invocation (into the shared companion
+ * buffer) while still applying it, so tests can assert whether materialization issued an alter or
+ * skipped it as a no-op.
+ */
+class RecordingInMemoryTableCatalog extends InMemoryTableCatalog {
+ override def alterTable(ident: Identifier, changes: TableChange*): V2Table = {
+ RecordingInMemoryTableCatalog.recordedAlters += changes.toSeq
+ super.alterTable(ident, changes: _*)
+ }
+}
+
+object RecordingInMemoryTableCatalog {
+ val recordedAlters: mutable.ArrayBuffer[Seq[TableChange]] =
+ mutable.ArrayBuffer.empty
+ def reset(): Unit = recordedAlters.clear()
}
Review Comment:
`recordedAlters` is global companion-object state, which only stays safe as long as nothing that uses this class ever runs concurrently in the same JVM. Since `CatalogManager` caches plugins by name (`catalogs.getOrElseUpdate(name, Catalogs.load(name, conf))`), every access within a `withRecordingCatalog` block resolves to the *same* instance, so the buffer can be per-instance and read back off the cached catalog. That removes the global state (and the need for `reset()`) entirely:
```suggestion
/**
* An [[InMemoryTableCatalog]] that records every `alterTable` invocation (into a per-instance
* buffer) while still applying it, so tests can assert whether materialization issued an alter or
* skipped it as a no-op. `CatalogManager` caches catalog plugins by name, so the same instance
* backs every access within a `withRecordingCatalog` block and reads go through that cached instance.
*/
class RecordingInMemoryTableCatalog extends InMemoryTableCatalog {
val recordedAlters: mutable.ArrayBuffer[Seq[TableChange]] = mutable.ArrayBuffer.empty
override def alterTable(ident: Identifier, changes: TableChange*): V2Table = {
recordedAlters += changes.toSeq
super.alterTable(ident, changes: _*)
}
}
```
This needs a few matching changes on the test side:
- Add an accessor next to `loadTableFromRecordingCatalog`:
```scala
private def recordingCatalog: RecordingInMemoryTableCatalog =
spark.sessionState.catalogManager
.catalog(recordingCatalogName)
.asInstanceOf[RecordingInMemoryTableCatalog]
```
- Replace the `RecordingInMemoryTableCatalog.recordedAlters` references in the assertions with `recordingCatalog.recordedAlters`.
- Drop both `RecordingInMemoryTableCatalog.reset()` calls in `withRecordingCatalog` — the `catalogManager.reset()` in the `finally` already clears the plugin cache, so the next block constructs a fresh instance with an empty buffer — and update the "resets its shared alter buffer" line in that method's scaladoc.
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3462034415
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
+ * accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTableInCatalog(
Review Comment:
nit: alterTable
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -29,16 +29,18 @@ import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{
CatalogV2Util,
Identifier,
+ Table => CatalogTable,
Review Comment:
CatalogTable is heavily used in many places, can we use V2Table? It looks a bit ugly but there are already some V2Table... in the code
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
Review Comment:
nit: I think 'InCatalog' is a bit unnecessary and long
--
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-57625][SDP] Manage Auxiliary Table Lifecycle in DatasetManager 1/2 [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #56684:
URL: https://github.com/apache/spark/pull/56684#issuecomment-4859150596
Merged to master/4.x, thanks @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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3462708588
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
+ * accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: CatalogTable,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+ val setProperties = properties.map { case (k, v) => TableChange.setProperty(k, v) }
+ catalog.alterTable(tableIdentifier, (columnChanges ++ setProperties).toArray: _*)
Review Comment:
Does `evolveTableInCatalog` call `alterTable` unconditionally, even when `columnChanges ++ setProperties` is empty? Is that effectively a no-op call?
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3462702429
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,70 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[CatalogTable] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTableInCatalog(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and
+ * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are
Review Comment:
Seems like a scalastyle line-length violation (>100).
--
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-57625][SDP][1/2] Manage Auxiliary Table Lifecycle in DatasetManager [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #56684:
URL: https://github.com/apache/spark/pull/56684#discussion_r3463712085
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -351,6 +349,80 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[V2Table] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table with.
+ */
+ private def createTable(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema
+ * and properties, skipping the catalog `alterTable` entirely when nothing actually changes.
+ * Partitioning/clustering cannot change in place, so no transforms are accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as computed in the current
+ * execution (the user-specified or inferred schema). This is the
+ * "incoming" side and may differ from `existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the table properties to (re)set on evolve.
+ * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and
+ * desired schemas (additive evolution) rather than the desired
+ * schema as-is.
+ */
+ private def evolveTable(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: V2Table,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+
+ val existingProperties = existingTable.properties()
+ val propertyChanges = properties.collect {
Review Comment:
had a question, do we handle property removal? its not changed by this pr, but just curious
--
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