You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "SanJSp (via GitHub)" <gi...@apache.org> on 2026/04/23 11:17:16 UTC
[PR] [SPARK-55668][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
SanJSp opened a new pull request, #55507:
URL: https://github.com/apache/spark/pull/55507
### What changes were proposed in this pull request?
This is **PR 1 of a split** of #55426 (see the [split suggestion](https://github.com/apache/spark/pull/55426#issuecomment-4292375876) for the full plan). Can merge in any order.
Validates the CDC metadata columns, row identity and row versioning returned by a `Changelog` connector at relation construction time, and introduces a dedicated error class to report the failure at analysis time rather than later at execution time with a less helpful error.
- `ChangelogTable.validateSchema`: fail-fast checks that the connector schema contains the required metadata columns (`_change_type`, `_commit_version`, `_commit_timestamp`), and that — when the connector advertises a capability requiring it — `rowId()` and `rowVersion()` are declared and the row version column is a non-nullable top-level column. Invoked from the `ChangelogTable` constructor.
- New error class `INVALID_CHANGELOG_SCHEMA` with sub-classes:
- `MISSING_COLUMN`, `INVALID_COLUMN_TYPE`
- `MISSING_ROW_ID`, `MISSING_ROW_VERSION`, `NESTED_ROW_VERSION`, `NULLABLE_ROW_VERSION`
- Matching `QueryCompilationErrors` helpers for each sub-class.
- Tests: `ChangelogResolutionSuite` schema-validation cases using a `TestChangelog` fixture.
### Why are the changes needed?
Gives connector implementors a clear analysis-time error message for misshapen CDC schemas instead of an opaque execution-time failure. Background on the original PR and its [dev list thread](https://lists.apache.org/thread/dhxx6pohs7fvqc3knzhtoj4tbcgrwxts).
### Does this PR introduce _any_ user-facing change?
Yes, for connector implementors. A connector that returns an invalid changelog schema (or advertises a capability that requires row identity/row versioning without declaring them) now fails at analysis time with `INVALID_CHANGELOG_SCHEMA.*` instead of at execution time.
### How was this patch tested?
Added schema-validation cases to `ChangelogResolutionSuite` covering: missing metadata column (each of `_change_type`, `_commit_version`, `_commit_timestamp`), wrong data type, connector-defined `_commit_version` type accepted, row-identity-required capabilities without rowId/rowVersion, nested rowVersion, nullable rowVersion, and valid schemas with data columns pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.7
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang closed pull request #55507: [SPARK-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class
URL: https://github.com/apache/spark/pull/55507
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3132688322
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
Review Comment:
Duplicate column names in the connector schema are silently dropped (last write wins in `toMap`). A connector with a bug that emits two `_change_type` columns wouldn't trip the validator — it would surface later as an attribute-resolution ambiguity. Worth rejecting duplicates here. (Not a blocker.)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
Review Comment:
Consider scoping to `private[v2]`: `ChangelogTable` is documented as NOT a connector API surface, and `validateSchema` is only called from the primary constructor. Keeping it `public` invites external callers to bypass or double-invoke the validation.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
+
+ val needsRowVersion = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert()
+ if (needsRowVersion && rowVersionRef.isEmpty) {
+ throw QueryCompilationErrors.changelogMissingRowVersionError(cl.name)
+ }
+
+ // Schema constraints on rowVersion: must be a top-level non-nullable column.
+ // Nullable rowVersions break carry-over detection (NULL = NULL is unknown, so a
+ // delete+insert pair would be misclassified as a real update).
+ rowVersionRef.foreach { ref =>
+ val fieldNames = ref.fieldNames()
+ if (fieldNames.length != 1) {
+ throw QueryCompilationErrors.changelogNestedRowVersionError(
+ cl.name, fieldNames.mkString("."))
+ }
+ val columnName = fieldNames(0)
+ val col = byName.getOrElse(columnName,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, columnName))
Review Comment:
When `rowVersion()` points to a column that isn't in `columns()`, we throw `MISSING_COLUMN` with the referenced column name — the user sees "Required column `<x>` is missing", as if a required metadata column were missing. The real cause is "`rowVersion` references a column not in the schema." Consider a dedicated sub-class (e.g. `ROW_VERSION_COLUMN_NOT_FOUND`) or at least a reworded message. This path also has no test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
Review Comment:
`rowId` columns are not checked for non-nullability, even though (a) the `Changelog.rowId()` Javadoc requires "Each referenced column must be non-nullable", and (b) the peer row-level-operations path validates this via `RewriteRowLevelCommand.resolveRowIdAttrs` with `NULLABLE_ROW_ID_ATTRIBUTES`. Consider adding a parallel `NULLABLE_ROW_ID` sub-class (or at least stating explicitly that `rowId` column validation is deferred to a later PR). As written, `rowVersion` gets nullability + top-level-ness but `rowId` gets presence only.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
+
+ val needsRowVersion = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert()
+ if (needsRowVersion && rowVersionRef.isEmpty) {
+ throw QueryCompilationErrors.changelogMissingRowVersionError(cl.name)
+ }
+
+ // Schema constraints on rowVersion: must be a top-level non-nullable column.
+ // Nullable rowVersions break carry-over detection (NULL = NULL is unknown, so a
+ // delete+insert pair would be misclassified as a real update).
+ rowVersionRef.foreach { ref =>
+ val fieldNames = ref.fieldNames()
+ if (fieldNames.length != 1) {
Review Comment:
The top-level requirement is new — the `Changelog.rowVersion()` Javadoc only says "non-nullable". A connector that reads the contract and returns a nested `NamedReference` will fail with `NESTED_ROW_VERSION` but get no hint from the API docs. Please either (a) update the `Changelog.rowVersion()` Javadoc to state that the reference must be a top-level column of `columns()`, or (b) remove this check and accept nested references. Same applies (by extension) to `rowId` if top-level-ness is also intended there.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -35,6 +38,11 @@ case class ChangelogTable(
changelog: Changelog,
changelogInfo: ChangelogInfo) extends Table with SupportsRead {
+ // Validate the connector returned a schema with the required CDC metadata columns
Review Comment:
Minor grammar — "Validate **that** the connector …".
```suggestion
// Validate that the connector returned a schema with the required CDC metadata columns
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala:
##########
@@ -203,4 +206,159 @@ class ChangelogResolutionSuite extends QueryTest with SharedSparkSession {
assert(range.startingVersion() == "1")
assert(range.endingVersion().get() == "5")
}
+
+ // ===========================================================================
+ // Generic changelog schema validation
+ // ===========================================================================
+
+ private def stubInfo(): ChangelogInfo = new ChangelogInfo(
+ new ChangelogRange.VersionRange("1", java.util.Optional.of("2"), true, true),
+ ChangelogInfo.DeduplicationMode.DROP_CARRYOVERS,
+ false)
+
+ private def cl(name: String, cols: (String, org.apache.spark.sql.types.DataType)*)
+ : TestChangelog = {
+ new TestChangelog(name, cols.map { case (n, t) => Column.create(n, t) }.toArray)
+ }
+
+ private def missing(columnName: String): Map[String, String] =
+ Map("changelogName" -> "bad_cl", "columnName" -> columnName)
+
+ private def wrongType(columnName: String, expected: String, actual: String)
+ : Map[String, String] = Map(
+ "changelogName" -> "bad_cl",
+ "columnName" -> columnName,
+ "expectedType" -> expected,
+ "actualType" -> actual)
+
+ // Valid metadata tuples; tests swap one of these out to create broken schemas.
+ private val validChangeType = "_change_type" -> StringType
+ private val validVersion = "_commit_version" -> LongType
+ private val validTimestamp = "_commit_timestamp" -> TimestampType
+
+ test("ChangelogTable - missing _change_type column throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(cl("bad_cl", validVersion, validTimestamp), stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.MISSING_COLUMN",
+ parameters = missing("_change_type"))
+ }
+
+ test("ChangelogTable - missing _commit_version column throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(cl("bad_cl", validChangeType, validTimestamp), stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.MISSING_COLUMN",
+ parameters = missing("_commit_version"))
+ }
+
+ test("ChangelogTable - missing _commit_timestamp column throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(cl("bad_cl", validChangeType, validVersion), stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.MISSING_COLUMN",
+ parameters = missing("_commit_timestamp"))
+ }
+
+ test("ChangelogTable - wrong _change_type data type throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(
+ cl("bad_cl", "_change_type" -> IntegerType, validVersion, validTimestamp),
+ stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.INVALID_COLUMN_TYPE",
+ parameters = wrongType("_change_type", "STRING", "INT"))
+ }
+
+ test("ChangelogTable - wrong _commit_timestamp data type throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(
+ cl("bad_cl", validChangeType, validVersion, "_commit_timestamp" -> LongType),
+ stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.INVALID_COLUMN_TYPE",
+ parameters = wrongType("_commit_timestamp", "TIMESTAMP", "BIGINT"))
+ }
+
+ test("ChangelogTable - _commit_version type is connector-defined (any type accepted)") {
+ Seq(IntegerType, LongType, StringType).foreach { versionType =>
+ ChangelogTable(
+ cl("any_cl", validChangeType, "_commit_version" -> versionType, validTimestamp),
+ stubInfo())
+ }
+ }
+
+ test("ChangelogTable - valid schema with data columns passes") {
+ ChangelogTable(
+ cl("good_cl", "id" -> LongType, "name" -> StringType,
+ validChangeType, validVersion, validTimestamp),
+ stubInfo())
+ }
+
+ test("ChangelogTable - nullable rowVersion column fails") {
+ val cl = new TestChangelog(
+ "bad_cl",
+ Array(
+ Column.create("id", LongType, false),
+ Column.create("_change_type", StringType),
+ Column.create("_commit_version", LongType),
+ Column.create("_commit_timestamp", TimestampType),
+ Column.create("row_commit_version", LongType)),
+ carryoverRows = true,
+ rowIdRefs = Array(FieldReference.column("id")),
+ rowVersionRef = Some(FieldReference.column("row_commit_version")))
+ checkError(
+ intercept[AnalysisException] { ChangelogTable(cl, stubInfo()) },
+ condition = "INVALID_CHANGELOG_SCHEMA.NULLABLE_ROW_VERSION",
+ parameters = Map(
+ "changelogName" -> "bad_cl",
+ "columnName" -> "row_commit_version"))
+ }
+
+ test("ChangelogTable - non-nullable rowVersion column passes") {
+ val cl = new TestChangelog(
+ "good_cl",
+ Array(
+ Column.create("id", LongType, false),
+ Column.create("_change_type", StringType),
+ Column.create("_commit_version", LongType),
+ Column.create("_commit_timestamp", TimestampType),
+ Column.create("row_commit_version", LongType, false)),
+ carryoverRows = true,
+ rowIdRefs = Array(FieldReference.column("id")),
+ rowVersionRef = Some(FieldReference.column("row_commit_version")))
+ ChangelogTable(cl, stubInfo())
+ }
Review Comment:
Three sub-classes introduced in this PR have no test case: `MISSING_ROW_ID`, `MISSING_ROW_VERSION`, `NESTED_ROW_VERSION`. Capability triggers other than `containsCarryoverRows=true` are also untested (no case exercising `representsUpdateAsDeleteAndInsert=true` or `containsIntermediateChanges=true`, and no case with a `rowVersion` referencing a column that is absent from `columns()`). The `TestChangelog` fixture already supports these — a few small cases would close the coverage gap the PR description claims is already covered.
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "SanJSp (via GitHub)" <gi...@apache.org>.
SanJSp commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3137935393
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
Review Comment:
For now I'd defer to a later PR. Is that fine?
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "SanJSp (via GitHub)" <gi...@apache.org>.
SanJSp commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3137751813
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
Review Comment:
Done, using your option 2 from the [NULL-safety thread on #55426](https://github.com/apache/spark/pull/55426#discussion_r3120231839). Added `count(rowVersion)` to the carry-over Window as a third aggregate alongside `min` and `max` (no extra Window operator, no additional shuffle). The filter now requires `_rv_cnt = 2 AND _min_rv = _max_rv`. A NULL rowVersion on either side fails the count check and the pair falls through as raw delete+insert instead of being silently dropped. Nesting-agnostic. Implementation and regression test (`"NULL rowVersion on one side is NOT silently dropped as carry-over"`) in #55508.
On the rowId asymmetry: rowId nullability is not schema-checked. An analogous silent-drop path exists (multiple NULL-rowId rows collapse into one Window partition via SQL NULL-group semantics), but the trigger surface is narrower than for rowVersion and a `count()=2`-style runtime guard does not port cleanly.
A top-level-only schema check would cover `id` but miss, for example, Delta's nested `_metadata.row_id`. This asymmetric coverage feels worse than no coverage at all.
We can
- either do a full schema walk through metadata columns covering both top-level and nested (how deep do we go, I think all the way, right?),
- or leave it unenforced and trust the Javadoc contract.
Currently, it's implemented for the latter. Open to the recursive column check but could need some input here. What do you think @gengliangwang?
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3132688322
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
Review Comment:
Duplicate column names in the connector schema are silently dropped (last write wins in `toMap`). A connector with a bug that emits two `_change_type` columns wouldn't trip the validator — it would surface later as an attribute-resolution ambiguity. Worth rejecting duplicates here. (Not a blocker.)
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3151952552
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -33,7 +35,13 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
*/
case class ChangelogTable(
changelog: Changelog,
- changelogInfo: ChangelogInfo) extends Table with SupportsRead {
+ changelogInfo: ChangelogInfo,
+ resolved: Boolean = false) extends Table with SupportsRead {
Review Comment:
Please have a follow-up to mark resolved as true after post processing.
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "SanJSp (via GitHub)" <gi...@apache.org>.
SanJSp commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3137360750
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3282,6 +3282,44 @@
},
"sqlState" : "42K03"
},
+ "INVALID_CHANGELOG_SCHEMA" : {
+ "message" : [
+ "The Change Data Capture (CDC) schema returned by connector <changelogName> is invalid."
+ ],
+ "subClass" : {
+ "INVALID_COLUMN_TYPE" : {
+ "message" : [
+ "Column `<columnName>` has type <actualType>, expected <expectedType>."
+ ]
+ },
+ "MISSING_COLUMN" : {
+ "message" : [
+ "Required column `<columnName>` is missing."
+ ]
+ },
+ "MISSING_ROW_ID" : {
+ "message" : [
+ "Connector advertises one or more post-processing properties (`containsCarryoverRows`, `representsUpdateAsDeleteAndInsert`, `containsIntermediateChanges`) that require row identity, but `Changelog.rowId()` returned an empty array. Either set all three to `false`, or return at least one row-id `NamedReference`."
Review Comment:
Done.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3282,6 +3282,44 @@
},
"sqlState" : "42K03"
},
+ "INVALID_CHANGELOG_SCHEMA" : {
+ "message" : [
+ "The Change Data Capture (CDC) schema returned by connector <changelogName> is invalid."
+ ],
+ "subClass" : {
+ "INVALID_COLUMN_TYPE" : {
+ "message" : [
+ "Column `<columnName>` has type <actualType>, expected <expectedType>."
+ ]
+ },
+ "MISSING_COLUMN" : {
+ "message" : [
+ "Required column `<columnName>` is missing."
+ ]
+ },
+ "MISSING_ROW_ID" : {
+ "message" : [
+ "Connector advertises one or more post-processing properties (`containsCarryoverRows`, `representsUpdateAsDeleteAndInsert`, `containsIntermediateChanges`) that require row identity, but `Changelog.rowId()` returned an empty array. Either set all three to `false`, or return at least one row-id `NamedReference`."
+ ]
+ },
+ "MISSING_ROW_VERSION" : {
+ "message" : [
+ "Connector advertises `containsCarryoverRows` or `representsUpdateAsDeleteAndInsert` is `true`, but `Changelog.rowVersion()` is not implemented. Override `rowVersion()` to return a `NamedReference` pointing to a non-nullable column in `Changelog.columns()`."
Review Comment:
The entry is gone entirely. After surfacing `UnsupportedOperationException` directly (per your `ChangelogTable.scala:81` comment), `rowVersion()` has no legitimate empty state distinct from UOE, so `MISSING_ROW_VERSION` became unreachable.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
Review Comment:
Done. Side-effect: `MISSING_ROW_VERSION` became unreachable and was removed along with its helper and subclass.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -35,6 +38,11 @@ case class ChangelogTable(
changelog: Changelog,
changelogInfo: ChangelogInfo) extends Table with SupportsRead {
+ // Validate the connector returned a schema with the required CDC metadata columns
Review Comment:
Done.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
Review Comment:
Done.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
+
+ val needsRowVersion = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert()
+ if (needsRowVersion && rowVersionRef.isEmpty) {
+ throw QueryCompilationErrors.changelogMissingRowVersionError(cl.name)
+ }
+
+ // Schema constraints on rowVersion: must be a top-level non-nullable column.
+ // Nullable rowVersions break carry-over detection (NULL = NULL is unknown, so a
+ // delete+insert pair would be misclassified as a real update).
+ rowVersionRef.foreach { ref =>
+ val fieldNames = ref.fieldNames()
+ if (fieldNames.length != 1) {
Review Comment:
Done (option b). Removed the top-level-only restriction. `NESTED_ROW_VERSION` error class, helper, and subclass are gone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
+
+ val needsRowVersion = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert()
+ if (needsRowVersion && rowVersionRef.isEmpty) {
+ throw QueryCompilationErrors.changelogMissingRowVersionError(cl.name)
+ }
+
+ // Schema constraints on rowVersion: must be a top-level non-nullable column.
+ // Nullable rowVersions break carry-over detection (NULL = NULL is unknown, so a
+ // delete+insert pair would be misclassified as a real update).
+ rowVersionRef.foreach { ref =>
+ val fieldNames = ref.fieldNames()
+ if (fieldNames.length != 1) {
+ throw QueryCompilationErrors.changelogNestedRowVersionError(
+ cl.name, fieldNames.mkString("."))
+ }
+ val columnName = fieldNames(0)
+ val col = byName.getOrElse(columnName,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, columnName))
Review Comment:
Obsolete after removing the nested restriction: the `byName` lookup for `rowVersion` is gone. A stale reference hits resolution at plan-build time via `V2ExpressionUtils.resolveRef`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala:
##########
@@ -203,4 +206,159 @@ class ChangelogResolutionSuite extends QueryTest with SharedSparkSession {
assert(range.startingVersion() == "1")
assert(range.endingVersion().get() == "5")
}
+
+ // ===========================================================================
+ // Generic changelog schema validation
+ // ===========================================================================
+
+ private def stubInfo(): ChangelogInfo = new ChangelogInfo(
+ new ChangelogRange.VersionRange("1", java.util.Optional.of("2"), true, true),
+ ChangelogInfo.DeduplicationMode.DROP_CARRYOVERS,
+ false)
+
+ private def cl(name: String, cols: (String, org.apache.spark.sql.types.DataType)*)
+ : TestChangelog = {
+ new TestChangelog(name, cols.map { case (n, t) => Column.create(n, t) }.toArray)
+ }
+
+ private def missing(columnName: String): Map[String, String] =
+ Map("changelogName" -> "bad_cl", "columnName" -> columnName)
+
+ private def wrongType(columnName: String, expected: String, actual: String)
+ : Map[String, String] = Map(
+ "changelogName" -> "bad_cl",
+ "columnName" -> columnName,
+ "expectedType" -> expected,
+ "actualType" -> actual)
+
+ // Valid metadata tuples; tests swap one of these out to create broken schemas.
+ private val validChangeType = "_change_type" -> StringType
+ private val validVersion = "_commit_version" -> LongType
+ private val validTimestamp = "_commit_timestamp" -> TimestampType
+
+ test("ChangelogTable - missing _change_type column throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(cl("bad_cl", validVersion, validTimestamp), stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.MISSING_COLUMN",
+ parameters = missing("_change_type"))
+ }
+
+ test("ChangelogTable - missing _commit_version column throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(cl("bad_cl", validChangeType, validTimestamp), stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.MISSING_COLUMN",
+ parameters = missing("_commit_version"))
+ }
+
+ test("ChangelogTable - missing _commit_timestamp column throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(cl("bad_cl", validChangeType, validVersion), stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.MISSING_COLUMN",
+ parameters = missing("_commit_timestamp"))
+ }
+
+ test("ChangelogTable - wrong _change_type data type throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(
+ cl("bad_cl", "_change_type" -> IntegerType, validVersion, validTimestamp),
+ stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.INVALID_COLUMN_TYPE",
+ parameters = wrongType("_change_type", "STRING", "INT"))
+ }
+
+ test("ChangelogTable - wrong _commit_timestamp data type throws") {
+ checkError(
+ intercept[AnalysisException] {
+ ChangelogTable(
+ cl("bad_cl", validChangeType, validVersion, "_commit_timestamp" -> LongType),
+ stubInfo())
+ },
+ condition = "INVALID_CHANGELOG_SCHEMA.INVALID_COLUMN_TYPE",
+ parameters = wrongType("_commit_timestamp", "TIMESTAMP", "BIGINT"))
+ }
+
+ test("ChangelogTable - _commit_version type is connector-defined (any type accepted)") {
+ Seq(IntegerType, LongType, StringType).foreach { versionType =>
+ ChangelogTable(
+ cl("any_cl", validChangeType, "_commit_version" -> versionType, validTimestamp),
+ stubInfo())
+ }
+ }
+
+ test("ChangelogTable - valid schema with data columns passes") {
+ ChangelogTable(
+ cl("good_cl", "id" -> LongType, "name" -> StringType,
+ validChangeType, validVersion, validTimestamp),
+ stubInfo())
+ }
+
+ test("ChangelogTable - nullable rowVersion column fails") {
+ val cl = new TestChangelog(
+ "bad_cl",
+ Array(
+ Column.create("id", LongType, false),
+ Column.create("_change_type", StringType),
+ Column.create("_commit_version", LongType),
+ Column.create("_commit_timestamp", TimestampType),
+ Column.create("row_commit_version", LongType)),
+ carryoverRows = true,
+ rowIdRefs = Array(FieldReference.column("id")),
+ rowVersionRef = Some(FieldReference.column("row_commit_version")))
+ checkError(
+ intercept[AnalysisException] { ChangelogTable(cl, stubInfo()) },
+ condition = "INVALID_CHANGELOG_SCHEMA.NULLABLE_ROW_VERSION",
+ parameters = Map(
+ "changelogName" -> "bad_cl",
+ "columnName" -> "row_commit_version"))
+ }
+
+ test("ChangelogTable - non-nullable rowVersion column passes") {
+ val cl = new TestChangelog(
+ "good_cl",
+ Array(
+ Column.create("id", LongType, false),
+ Column.create("_change_type", StringType),
+ Column.create("_commit_version", LongType),
+ Column.create("_commit_timestamp", TimestampType),
+ Column.create("row_commit_version", LongType, false)),
+ carryoverRows = true,
+ rowIdRefs = Array(FieldReference.column("id")),
+ rowVersionRef = Some(FieldReference.column("row_commit_version")))
+ ChangelogTable(cl, stubInfo())
+ }
Review Comment:
Done. Added five tests covering `MISSING_ROW_ID` via `representsUpdateAsDeleteAndInsert` / `containsIntermediateChanges`, `UnsupportedOperationException` surfacing for both `rowId()` and `rowVersion()`, and a positive nested-refs case (Delta-style `_metadata`). Extended `TestChangelog` with `updateAsDeleteInsert`, `intermediateChanges`, `rowIdSupported` flags.
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3140648242
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
+
+ // Capability-driven presence checks: a connector that advertises a capability which
+ // requires row identity or row versioning must actually expose those references.
+ // Otherwise post-processing would crash with an UnsupportedOperationException at
+ // runtime instead of producing a clean AnalysisException here.
+ val needsRowId = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert() ||
+ cl.containsIntermediateChanges()
+ if (needsRowId && (rowIds == null || rowIds.isEmpty)) {
+ throw QueryCompilationErrors.changelogMissingRowIdError(cl.name)
+ }
+
+ val needsRowVersion = cl.containsCarryoverRows() ||
+ cl.representsUpdateAsDeleteAndInsert()
+ if (needsRowVersion && rowVersionRef.isEmpty) {
+ throw QueryCompilationErrors.changelogMissingRowVersionError(cl.name)
+ }
+
+ // Schema constraints on rowVersion: must be a top-level non-nullable column.
+ // Nullable rowVersions break carry-over detection (NULL = NULL is unknown, so a
+ // delete+insert pair would be misclassified as a real update).
+ rowVersionRef.foreach { ref =>
+ val fieldNames = ref.fieldNames()
+ if (fieldNames.length != 1) {
Review Comment:
There is no new commit since the review yesterday.
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3155467329
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -33,7 +35,13 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
*/
case class ChangelogTable(
changelog: Changelog,
- changelogInfo: ChangelogInfo) extends Table with SupportsRead {
+ changelogInfo: ChangelogInfo,
+ resolved: Boolean = false) extends Table with SupportsRead {
Review Comment:
Yes, when I reviewed this one, https://github.com/apache/spark/pull/55508 is not merged yet.
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "SanJSp (via GitHub)" <gi...@apache.org>.
SanJSp commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3152982804
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -33,7 +35,13 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
*/
case class ChangelogTable(
changelog: Changelog,
- changelogInfo: ChangelogInfo) extends Table with SupportsRead {
+ changelogInfo: ChangelogInfo,
+ resolved: Boolean = false) extends Table with SupportsRead {
Review Comment:
I'm not sure I understand 100%, aren't we already doing this? The problem is that we want to mark a node as processed to avoid reprocessing it, right? So [here](https://github.com/apache/spark/blob/697d021f3cfdb48adb4e5efbaa748bdfbc39e179/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala#L71) and [here](https://github.com/apache/spark/blob/697d021f3cfdb48adb4e5efbaa748bdfbc39e179/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala#L93) we create copies of the node and set their field `resolved = true`. Does this address your concern?
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #55507:
URL: https://github.com/apache/spark/pull/55507#issuecomment-4340818337
CI passed in https://github.com/gengliangwang/spark/runs/73505350455 (PR https://github.com/apache/spark/pull/55567).
Merging to master
--
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-55951][SQL] Add ChangelogTable schema validation and INVALID_CHANGELOG_SCHEMA error class [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on code in PR #55507:
URL: https://github.com/apache/spark/pull/55507#discussion_r3131119662
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3282,6 +3282,44 @@
},
"sqlState" : "42K03"
},
+ "INVALID_CHANGELOG_SCHEMA" : {
+ "message" : [
+ "The Change Data Capture (CDC) schema returned by connector <changelogName> is invalid."
+ ],
+ "subClass" : {
+ "INVALID_COLUMN_TYPE" : {
+ "message" : [
+ "Column `<columnName>` has type <actualType>, expected <expectedType>."
+ ]
+ },
+ "MISSING_COLUMN" : {
+ "message" : [
+ "Required column `<columnName>` is missing."
+ ]
+ },
+ "MISSING_ROW_ID" : {
+ "message" : [
+ "Connector advertises one or more post-processing properties (`containsCarryoverRows`, `representsUpdateAsDeleteAndInsert`, `containsIntermediateChanges`) that require row identity, but `Changelog.rowId()` returned an empty array. Either set all three to `false`, or return at least one row-id `NamedReference`."
Review Comment:
The last part is directed to connector developers and not actionable for actual users, remove it
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3282,6 +3282,44 @@
},
"sqlState" : "42K03"
},
+ "INVALID_CHANGELOG_SCHEMA" : {
+ "message" : [
+ "The Change Data Capture (CDC) schema returned by connector <changelogName> is invalid."
+ ],
+ "subClass" : {
+ "INVALID_COLUMN_TYPE" : {
+ "message" : [
+ "Column `<columnName>` has type <actualType>, expected <expectedType>."
+ ]
+ },
+ "MISSING_COLUMN" : {
+ "message" : [
+ "Required column `<columnName>` is missing."
+ ]
+ },
+ "MISSING_ROW_ID" : {
+ "message" : [
+ "Connector advertises one or more post-processing properties (`containsCarryoverRows`, `representsUpdateAsDeleteAndInsert`, `containsIntermediateChanges`) that require row identity, but `Changelog.rowId()` returned an empty array. Either set all three to `false`, or return at least one row-id `NamedReference`."
+ ]
+ },
+ "MISSING_ROW_VERSION" : {
+ "message" : [
+ "Connector advertises `containsCarryoverRows` or `representsUpdateAsDeleteAndInsert` is `true`, but `Changelog.rowVersion()` is not implemented. Override `rowVersion()` to return a `NamedReference` pointing to a non-nullable column in `Changelog.columns()`."
Review Comment:
Sane here, remove the last part.
Also grammar: 'Connector advertises ... ~is `true`~'
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala:
##########
@@ -45,3 +53,65 @@ case class ChangelogTable(
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ, MICRO_BATCH_READ)
}
+
+object ChangelogTable {
+
+ def validateSchema(cl: Changelog): Unit = {
+ val byName = cl.columns.map(c => c.name -> c).toMap
+ def check(name: String, expected: DataType*): Unit = {
+ val col = byName.getOrElse(name,
+ throw QueryCompilationErrors.changelogMissingColumnError(cl.name, name))
+ if (expected.nonEmpty && col.dataType != expected.head) {
+ throw QueryCompilationErrors.changelogInvalidColumnTypeError(
+ cl.name, name, expected.head.sql, col.dataType.sql)
+ }
+ }
+ check("_change_type", StringType)
+ check("_commit_version") // connector-defined, any type accepted
+ check("_commit_timestamp", TimestampType)
+
+ // `rowId()` / `rowVersion()` default to throwing UnsupportedOperationException for
+ // connectors that haven't opted in. Translate that into "not declared" so we can
+ // reason about it as Option/empty-array below.
+ val rowIds: Array[NamedReference] = try cl.rowId() catch {
+ case _: UnsupportedOperationException => Array.empty
+ }
+ val rowVersionRef: Option[NamedReference] = try Some(cl.rowVersion()) catch {
+ case _: UnsupportedOperationException => None
+ }
Review Comment:
I'm on the fence here, I think I'd rather not catch `UnsupportedOperationException` and surface that error directly.
This is an error from connector developers that didn't implement the right methods even tough the capabilities they reported require it. So it doesn't really need to have a user-facing error message, surfacing `UnsupportedOperationException` would actually make it a bit more obvious that there's an issue with the connector itself.
The important point is that we only call these methods here if the capabilities require it
--
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