Skip to content

Commit efb589e

Browse files
PorridgeSwimPorridgeSwim
committed
Support non-additive schema evolution in v2
# Conflicts: # spark/v2/src/main/java/io/delta/spark/internal/v2/read/SparkMicroBatchStream.java
1 parent 9a48144 commit efb589e

16 files changed

Lines changed: 1019 additions & 194 deletions

File tree

‎spark-unified/src/main/scala/io/delta/internal/ApplyV2Streaming.scala‎

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ import scala.jdk.OptionConverters._
2121

2222
import io.delta.spark.internal.v2.catalog.SparkTable
2323
import io.delta.spark.internal.v2.utils.ScalaUtils
24-
import org.apache.spark.sql.delta.DeltaV2Mode
24+
import org.apache.spark.sql.delta.{DeltaOptions, DeltaV2Mode}
2525
import org.apache.spark.sql.delta.sources.DeltaSourceUtils
2626

2727
import org.apache.spark.sql.SparkSession
@@ -71,6 +71,16 @@ class ApplyV2Streaming(
7171
deltaV2Mode.isStreamingReadsEnabled(s.dataSource.catalogTable.toJava)
7272
}
7373

74+
/** True when schema-tracking is set on extraOptions but not yet on the SparkTable. */
75+
private def needsSchemaTrackingRebuild(
76+
table: SparkTable, extraOptions: CaseInsensitiveStringMap): Boolean = {
77+
val tableOptions = new CaseInsensitiveStringMap(table.getOptions)
78+
(extraOptions.containsKey(DeltaOptions.SCHEMA_TRACKING_LOCATION) ||
79+
extraOptions.containsKey(DeltaOptions.SCHEMA_TRACKING_LOCATION_ALIAS)) &&
80+
!tableOptions.containsKey(DeltaOptions.SCHEMA_TRACKING_LOCATION) &&
81+
!tableOptions.containsKey(DeltaOptions.SCHEMA_TRACKING_LOCATION_ALIAS)
82+
}
83+
7484
override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperators {
7585
case s: StreamingRelation if shouldApplyV2Streaming(s) =>
7686
// catalogTable is guaranteed to be defined because shouldApplyV2Streaming checks it
@@ -102,5 +112,21 @@ class ApplyV2Streaming(
102112
identifier = Some(ident),
103113
// Keep this None to force the V2 path; we don't want to fall back to V1 here.
104114
v1Relation = None)
115+
116+
// For catalog-loaded relations (readStream.table("foo")), TableCatalog.loadTable has no
117+
// read-options channel, so schemaTrackingLocation arrives only on extraOptions and
118+
// SparkTable.schema() can't see it. Rebuild with the merged options so the schema-log
119+
// lookup fires, then re-derive output. Idempotent via needsSchemaTrackingRebuild.
120+
case s @ StreamingRelationV2(_, _, table: SparkTable, extraOptions, _, _, _, _)
121+
if needsSchemaTrackingRebuild(table, extraOptions) =>
122+
val merged = new java.util.HashMap[String, String]()
123+
merged.putAll(table.getOptions)
124+
merged.putAll(extraOptions.asCaseSensitiveMap())
125+
val rebuilt = if (table.getCatalogTable.isPresent) {
126+
new SparkTable(table.getIdentifier, table.getCatalogTable.get, merged)
127+
} else {
128+
new SparkTable(table.getIdentifier, table.getTablePath.toString, merged)
129+
}
130+
s.copy(table = rebuilt, output = toAttributes(rebuilt.schema))
105131
}
106132
}

‎spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala‎

Lines changed: 22 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -70,12 +70,10 @@ trait DeltaV2SourceSchemaEvolutionSuiteBase extends V2ForceTest {
7070
}
7171
}
7272

73-
// TODO(#5319): Move tests to shouldPassTests as V2 schema tracking log support is implemented.
74-
override protected def shouldPassTests: Set[String] = Set.empty[String]
75-
76-
// All tests from StreamingSchemaEvolutionSuiteBase.
77-
// Override in CDC suites to add CDC-specific tests.
78-
override protected def shouldFailTests: Set[String] = Set(
73+
// All tests from StreamingSchemaEvolutionSuiteBase pass in V2 non-CDC mode with the
74+
// schema tracking log implementation, except for tests that rely on consecutive
75+
// metadata-change merging (which the V2 merger is a no-op for).
76+
override protected def shouldPassTests: Set[String] = Set(
7977
// ========== Schema location validation ==========
8078
"schema location not under checkpoint",
8179
"schema location same as checkpoint",
@@ -144,24 +142,26 @@ class DeltaV2SourceSchemaEvolutionIdColumnMappingSuite
144142

145143
// CDC suites
146144

147-
class DeltaV2SourceSchemaEvolutionCDCNameColumnMappingSuite
148-
extends DeltaSourceSchemaEvolutionCDCNameColumnMappingSuite
149-
with DeltaV2SourceSchemaEvolutionSuiteBase {
145+
// CDC schema evolution is not in scope for V2 (see the V2 schema-evolution design doc
146+
// "Open Items / Future Work"). Flip the base's passing tests back into shouldFailTests so
147+
// all schema-evolution tests fail under CDC mode, and add the CDC-specific tests.
148+
trait DeltaV2SourceSchemaEvolutionCDCSuiteBase extends DeltaV2SourceSchemaEvolutionSuiteBase {
149+
self: StreamingSchemaEvolutionSuiteBase =>
150150

151-
override protected def shouldFailTests: Set[String] = super.shouldFailTests ++ Set(
152-
// Additional tests from CDCStreamingSchemaEvolutionSuiteBase
153-
"CDC streaming with schema evolution",
154-
"protocol and configuration evolution"
155-
)
151+
override protected def shouldPassTests: Set[String] = Set.empty[String]
152+
153+
override protected def shouldFailTests: Set[String] =
154+
super.shouldPassTests ++ super.shouldFailTests ++ Set(
155+
// Additional tests from CDCStreamingSchemaEvolutionSuiteBase
156+
"CDC streaming with schema evolution",
157+
"protocol and configuration evolution"
158+
)
156159
}
157160

161+
class DeltaV2SourceSchemaEvolutionCDCNameColumnMappingSuite
162+
extends DeltaSourceSchemaEvolutionCDCNameColumnMappingSuite
163+
with DeltaV2SourceSchemaEvolutionCDCSuiteBase
164+
158165
class DeltaV2SourceSchemaEvolutionCDCIdColumnMappingSuite
159166
extends DeltaSourceSchemaEvolutionCDCIdColumnMappingSuite
160-
with DeltaV2SourceSchemaEvolutionSuiteBase {
161-
162-
override protected def shouldFailTests: Set[String] = super.shouldFailTests ++ Set(
163-
// Additional tests from CDCStreamingSchemaEvolutionSuiteBase
164-
"CDC streaming with schema evolution",
165-
"protocol and configuration evolution"
166-
)
167-
}
167+
with DeltaV2SourceSchemaEvolutionCDCSuiteBase

‎spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSuite.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ class DeltaV2SourceSuite extends DeltaSourceSuite with V2ForceTest {
6767
"rename column: should fail with non-additive schema change error",
6868
"rename column: should throw schema change error with unsafe flag enabled",
6969
"type widening: should fail with non-additive schema change error when enable schema tracking",
70+
"handling nullability schema changes", // Uses .table() directly
7071

7172
// === Read options ===
7273
"excludeRegex works and doesn't mess up offsets across restarts - parquet version",
@@ -167,7 +168,6 @@ class DeltaV2SourceSuite extends DeltaSourceSuite with V2ForceTest {
167168
"disallow user specified schema", // Uses .schema() directly
168169
"make sure that the delta sources works fine", // Uses .delta() directly
169170
"self union a Delta table should pass the catalog table assert", // Uses .table() directly
170-
"handling nullability schema changes", // Uses .table() directly
171171
"allow user specified schema if consistent: v1 source", // Uses DataSource directly
172172
// Calls deltaSource.createSource() directly
173173
"createSource should create source with empty or matching table schema provided"

‎spark-unified/src/test/scala/org/apache/spark/sql/delta/test/columnmapping/RemoveColumnMappingStreamingReadV2Suite.scala‎

Lines changed: 6 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -32,9 +32,8 @@ class RemoveColumnMappingStreamingReadV2Suite
3232

3333
override protected def executeDml(sqlText: String): Unit = executeInV1Mode(sqlText)
3434

35-
// Tests that run without schema tracking. These exercise non-additive column-mapping schema
36-
// change detection, which is supported on the V2 connector.
3735
override protected def shouldPassTests: Set[String] = Set(
36+
// Tests that run without schema tracking.
3837
"Upgrade, StartStreamRead, Downgrade, FailNonAdditiveChange",
3938
"Upgrade, Downgrade, StartStreamRead, Success",
4039
"StartStreamRead, Upgrade, Rename, Downgrade, FailNonAdditiveChange",
@@ -50,18 +49,12 @@ class RemoveColumnMappingStreamingReadV2Suite
5049
"Upgrade, Drop, StartStreamRead, Downgrade, FailNonAdditiveChange",
5150
"Upgrade, Drop, StartStreamRead, Downgrade, Upgrade, FailNonAdditiveChange",
5251
"Upgrade, Rename, Downgrade, StartStreamRead, Success",
53-
"Upgrade, Drop, Downgrade, StartStreamRead, Success"
54-
)
55-
56-
// Tests that run with schema tracking enabled. The schema tracking log is not yet supported
57-
// on the V2 connector.
58-
override protected def shouldFailTests: Set[String] = Set(
59-
// TODO(#5319): the three tests are not supported in v2 yet due to the gap of columnMapping
60-
// check util.
61-
"StartStreamRead, Upgrade, Downgrade, SuccessAndFailSchemaTracking",
52+
"Upgrade, Drop, Downgrade, StartStreamRead, Success",
6253
"Upgrade, Rename, Downgrade, StartStreamRead, Upgrade, SuccessAndFailSchemaTracking",
6354
"Upgrade, Drop, Downgrade, StartStreamRead, Upgrade, SuccessAndFailSchemaTracking",
64-
// TODO(#5319): Move these to shouldPassTests as V2 schema tracking log support is implemented.
55+
"StartStreamRead, Upgrade, Downgrade, SuccessAndFailSchemaTracking",
56+
57+
// Tests that run with schema tracking enabled.
6558
"StartStreamRead, Upgrade, Downgrade, SuccessAndFailSchemaTracking with schema tracking",
6659
"Upgrade, StartStreamRead, Downgrade, FailNonAdditiveChange with schema tracking",
6760
"Upgrade, Downgrade, StartStreamRead, Success with schema tracking",
@@ -88,6 +81,5 @@ class RemoveColumnMappingStreamingReadV2Suite
8881
"Upgrade, Rename, Downgrade, StartStreamRead, Upgrade, SuccessAndFailSchemaTracking" +
8982
" with schema tracking",
9083
"Upgrade, Drop, Downgrade, StartStreamRead, Upgrade, SuccessAndFailSchemaTracking" +
91-
" with schema tracking"
92-
)
84+
" with schema tracking")
9385
}

‎spark-unified/src/test/scala/org/apache/spark/sql/delta/test/typewidening/TypeWideningStreamingV2SourceSuite.scala‎

Lines changed: 24 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,10 @@ import org.apache.spark.sql.delta.typewidening.{
2525

2626
/**
2727
* Base trait for V2 type widening streaming source tests.
28-
* Provides common shouldFail logic shared by both suites.
28+
*
29+
* The base lists every test from `TypeWideningStreamingSourceTests` as passing — V2 supports
30+
* type-widening streaming reads. Subclasses move specific tests back to `shouldFailTests` when
31+
* there is a concrete V2 limitation (e.g., partition-column schema bug, missing event logging).
2932
*/
3033
trait TypeWideningStreamingV2SourceSuiteBase extends V2ForceTest {
3134
self: TypeWideningStreamingSourceTestMixin =>
@@ -34,20 +37,13 @@ trait TypeWideningStreamingV2SourceSuiteBase extends V2ForceTest {
3437

3538
override protected def executeDml(sqlText: String): Unit = executeInV1Mode(sqlText)
3639

37-
// TODO(#5319): Move tests to shouldPassTests as V2 schema tracking log support is implemented.
38-
override protected def shouldPassTests: Set[String] = Set.empty[String]
39-
40-
// Tests from TypeWideningStreamingSourceTests, shared by both suites.
41-
// Override in subclasses to add suite-specific tests.
42-
override protected def shouldFailTests: Set[String] = Set(
40+
override protected def shouldPassTests: Set[String] = Set(
4341
"type change - filter",
4442
"type change - projection",
45-
"type change - projection partition column",
4643
"type change - widen unused scala udf field",
4744
"type change - widen scala udf argument",
4845
"type change - widen aggregation grouping key",
4946
"type change - widen aggregation expression",
50-
"type change - widen aggregation expression partition column",
5147
"type change - widen aggregation expression after projection",
5248
"type change - widen limit",
5349
"type change - widen distinct",
@@ -59,45 +55,37 @@ trait TypeWideningStreamingV2SourceSuiteBase extends V2ForceTest {
5955
"arbitrary type changes are not supported",
6056
"type change in delta source writing to a delta sink"
6157
)
62-
}
63-
64-
class TypeWideningStreamingV2SourceSuite
65-
extends TypeWideningStreamingSourceSuite
66-
with TypeWideningStreamingV2SourceSuiteBase {
67-
68-
// All tests pass without schema tracking enabled, except where noted in shouldFailTests.
69-
override protected def shouldPassTests: Set[String] =
70-
super.shouldFailTests -- shouldFailTests
7158

59+
// Failures that affect both the schema-tracking and non-schema-tracking suites.
7260
override protected def shouldFailTests: Set[String] = Set(
7361
// Delta log event is not supported in V2, so event-logging tests are not meaningful.
7462
"schema changed event is logged for type widening",
7563
"schema changed event is not logged when there are no schema changes",
7664
// TODO(#5319): Partition column schema has a bug in V2 causing these to fail.
7765
"type change - projection partition column",
78-
"type change - widen aggregation expression partition column",
79-
// TODO(#5319): V2 lacks the implementation of
80-
// validateAndInitMetadataLogForPlannedBatchesDuringStreamStart, so the
81-
// 2nd testStream restart does not throw on the incompatible type change.
82-
"widening type change then restore back",
83-
"narrowing type changes are not supported",
84-
"arbitrary type changes are not supported"
66+
"type change - widen aggregation expression partition column"
8567
)
8668
}
8769

70+
class TypeWideningStreamingV2SourceSuite
71+
extends TypeWideningStreamingSourceSuite
72+
with TypeWideningStreamingV2SourceSuiteBase
73+
8874
class TypeWideningStreamingV2SourceSchemaTrackingSuite
8975
extends TypeWideningStreamingSourceSchemaTrackingSuite
9076
with TypeWideningStreamingV2SourceSuiteBase {
9177

92-
override protected def shouldFailTests: Set[String] = super.shouldFailTests ++ Set(
93-
// Additional tests from TypeWideningStreamingSourceSchemaTrackingTests
94-
"type change first without schemaTrackingLocation and unblock using schemaTrackingLocation",
95-
"unblocking stream with sql conf after type change - unblock all",
96-
"unblocking stream with sql conf after type change - unblock stream",
97-
"unblocking stream with sql conf after type change - unblock version",
98-
"unblocking stream with reader option after type change - unblock stream",
99-
"unblocking stream with reader option after type change - unblock version",
100-
"overwrite schema with type change and dropped column",
101-
"disable schema tracking log using internal conf"
102-
)
78+
// Schema-tracking-specific tests from TypeWideningStreamingSourceSchemaTrackingTests, on top of
79+
// the base type-widening tests inherited from the trait, minus tests with known V2 issues.
80+
override protected def shouldPassTests: Set[String] =
81+
super.shouldPassTests ++ Set(
82+
"type change first without schemaTrackingLocation and unblock using schemaTrackingLocation",
83+
"unblocking stream with sql conf after type change - unblock all",
84+
"unblocking stream with sql conf after type change - unblock stream",
85+
"unblocking stream with sql conf after type change - unblock version",
86+
"unblocking stream with reader option after type change - unblock stream",
87+
"unblocking stream with reader option after type change - unblock version",
88+
"overwrite schema with type change and dropped column",
89+
"disable schema tracking log using internal conf"
90+
)
10391
}

‎spark/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala‎

Lines changed: 39 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
5656
import org.apache.spark.sql.catalyst.plans.logical.CloneTableStatement
5757
import org.apache.spark.sql.catalyst.plans.logical.RestoreTableStatement
5858
import org.apache.spark.sql.catalyst.rules.Rule
59-
import org.apache.spark.sql.catalyst.streaming.WriteToStream
59+
import org.apache.spark.sql.catalyst.streaming.{StreamingRelationV2, WriteToStream}
6060
import org.apache.spark.sql.catalyst.trees.TreeNodeTag
6161
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttribute
6262
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
@@ -1530,9 +1530,9 @@ class DeltaAnalysis(session: SparkSession)
15301530
private def verifyDeltaSourceSchemaLocation(
15311531
inputQuery: LogicalPlan,
15321532
checkpointLocation: String): Unit = {
1533-
// Maps StreamingRelation to schema location, similar to how MicroBatchExecution converts
1534-
// StreamingRelation to StreamingExecutionRelation.
1535-
val schemaLocationMap = mutable.Map[StreamingRelation, String]()
1533+
// Maps each Delta streaming source (V1 or V2) to its schema tracking location, similar to how
1534+
// MicroBatchExecution converts StreamingRelation to StreamingExecutionRelation.
1535+
val schemaLocationMap = mutable.Map[LogicalPlan, String]()
15361536
val allowSchemaLocationOutsideOfCheckpoint = session.sessionState.conf.getConf(
15371537
DeltaSQLConf.DELTA_STREAMING_ALLOW_SCHEMA_LOCATION_OUTSIDE_CHECKPOINT_LOCATION)
15381538
inputQuery.foreach {
@@ -1558,6 +1558,27 @@ class DeltaAnalysis(session: SparkSession)
15581558
// Save schema location for this streaming relation
15591559
schemaLocationMap.put(streamingRelation, schemaTrackingLocation.stripSuffix("/"))
15601560
}
1561+
case streamingRelationV2 @ StreamingRelationV2(_, _, table, extraOptions, _, _, _, _) =>
1562+
val opts = extraOptions.asCaseSensitiveMap.asScala.toMap
1563+
DeltaDataSource.extractSchemaTrackingLocationConfig(session, opts)
1564+
.foreach { rootSchemaTrackingLocation =>
1565+
// The analysis-time tableId only needs to be unique per source; it is decoupled from
1566+
// the runtime tableId (V1 uses the Delta UUID, V2 uses Kernel's snapshot id).
1567+
// `table.name` is path-aware ("delta.`/path`" for path-based, qualified name for
1568+
// catalog-based) and is sufficient to differentiate sources for the conflict check.
1569+
val tableId = table.name.replace(":", "").replace("/", "_")
1570+
val sourceIdOpt = opts.get(DeltaOptions.STREAMING_SOURCE_TRACKING_ID)
1571+
val schemaTrackingLocation =
1572+
DeltaSourceMetadataTrackingLog.fullMetadataTrackingLocation(
1573+
rootSchemaTrackingLocation, tableId, sourceIdOpt)
1574+
if (!allowSchemaLocationOutsideOfCheckpoint) {
1575+
assertSchemaTrackingLocationUnderCheckpoint(
1576+
checkpointLocation,
1577+
schemaTrackingLocation
1578+
)
1579+
}
1580+
schemaLocationMap.put(streamingRelationV2, schemaTrackingLocation.stripSuffix("/"))
1581+
}
15611582
case _ =>
15621583
}
15631584

@@ -1567,14 +1588,21 @@ class DeltaAnalysis(session: SparkSession)
15671588
.groupBy { rel => schemaLocationMap(rel) }
15681589
.find(_._2.size > 1)
15691590
conflictSchemaOpt.foreach { case (schemaLocation, relations) =>
1570-
val ds = relations.head.dataSource
15711591
// Pick one source that has conflict to make it more actionable for the user
1572-
val oneTableWithConflict = ds.catalogTable
1573-
.map(_.identifier.toString)
1574-
.getOrElse {
1575-
// `path` must exist
1576-
CaseInsensitiveMap(ds.options).get("path").get
1577-
}
1592+
val oneTableWithConflict = relations.head match {
1593+
case streamingRelation: StreamingRelation =>
1594+
val ds = streamingRelation.dataSource
1595+
ds.catalogTable
1596+
.map(_.identifier.toString)
1597+
.getOrElse {
1598+
// `path` must exist
1599+
CaseInsensitiveMap(ds.options).get("path").get
1600+
}
1601+
case streamingRelationV2: StreamingRelationV2 =>
1602+
streamingRelationV2.identifier
1603+
.map(_.toString)
1604+
.getOrElse(streamingRelationV2.table.name)
1605+
}
15781606
throw DeltaErrors.sourcesWithConflictingSchemaTrackingLocation(
15791607
schemaLocation, oneTableWithConflict)
15801608
}

0 commit comments

Comments
 (0)