ahmedabu98 commented on code in PR #40868:
URL: https://github.com/apache/beam/pull/40868#discussion_r4405127749
##########
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java:
##########
@@ -241,6 +242,94 @@ static Schema buildPublicBeamSchema(Schema baseSchema,
@Nullable List<String> me
return builder.build();
}
+ /**
+ * Returns whether two Delta schemas (or, recursively, data types) describe
the same data.
+ *
+ * <p>{@link ReadChanges} uses this to verify that the table schema did not
change across the
+ * range of table versions that is read, so that all emitted change records
conform to a single
+ * schema.
+ *
+ * <p>Two schemas describe the same data if their fields have the same names
(in the same order),
+ * data types, nullability, and column mapping (see {@code
hasSameColumnMapping()}), compared
+ * recursively for nested types. These properties determine both the Beam
schema of the emitted
+ * rows (see {@code ReadRows.convertToBeamSchema()}) and how the rows are
read from data files.
+ *
+ * <p>All other field metadata is ignored, since changing it does not change
the data. For
+ * example, column comments, column default values, or identity column
high-water marks (which
+ * writers update whenever they insert rows into the table) may differ
between the two schemas.
+ */
+ static boolean hasSameDataSchema(DataType left, DataType right) {
Review Comment:
Can we create a separate util class for this?
##########
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateCDCReadTasksDoFn.java:
##########
@@ -108,18 +108,30 @@ public void processElement(@Element String tablePath,
OutputReceiver<DeltaCDCRea
resolvedStartVersion, resolvedEndVersion));
}
- // 2. Load snapshot at resolvedEndVersion to get the scanStateRow
- // We use endVersion's schema because it represents the latest schema in
the
- // read range
- // which handles schema evolution (older files will just lack new columns).
- Snapshot endSnapshot = table.getSnapshotAsOfVersion(engine,
resolvedEndVersion);
+ // 2. Load snapshots at resolvedStartVersion and resolvedEndVersion and
verify that the data
+ // schema has not changed across the CDC range. Metadata-only schema
changes (for example,
+ // changes to column comments) are allowed. See
DeltaIO.hasSameDataSchema() for details.
+ Snapshot startSnapshot = table.getSnapshotAsOfVersion(engine,
resolvedStartVersion);
+ Snapshot endSnapshot =
+ resolvedStartVersion == resolvedEndVersion
+ ? startSnapshot
+ : table.getSnapshotAsOfVersion(engine, resolvedEndVersion);
+ io.delta.kernel.types.StructType startSchema = startSnapshot.getSchema();
+ io.delta.kernel.types.StructType endSchema = endSnapshot.getSchema();
+ if (startSchema == null || endSchema == null) {
+ throw new IllegalStateException("Table schema is null.");
+ }
+ if (!DeltaIO.hasSameDataSchema(startSchema, endSchema)) {
Review Comment:
Do we need to check for schema changes within the range? e.g. if a column
was added then dropped, this comparison wouldn't catch that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
