You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rahulsmahadev (via GitHub)" <gi...@apache.org> on 2026/09/04 21:00:53 UTC
[PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
rahulsmahadev opened a new pull request, #17957:
URL: https://github.com/apache/iceberg/pull/17957
Implements Spark 4.2's `SupportsSchemaEvolution` mix-in on Iceberg's `SparkTable`, so Iceberg participates in DSv2 automatic schema evolution during writes.
During automatic schema evolution Spark passes each candidate column change to `supportsColumnChange(TableChange.ColumnChange)` and skips the ones the source reports as unsupported. `SparkTable` now reports support consistent with what Iceberg's schema update actually applies:
- add column: only when nullable, without a default, and with a Spark type Iceberg can convert;
- update type: only when the new type is a primitive and an allowed Iceberg type promotion;
- update nullability: only when relaxing to nullable;
- delete column: rejected for identifier fields and for any field whose subtree contains an identifier field (matching `SchemaUpdate`), and for unknown fields;
- rename, comment, and position changes are supported.
`SparkTable` already advertises `TableCapability.AUTOMATIC_SCHEMA_EVOLUTION` (gated by the `write.spark.auto-schema-evolution.enabled` table property), which is what makes Spark consult this hook. A unit test is added in `TestSparkTable`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#issuecomment-5902758635
> Could we log a warning when this hook is invoked with `write.spark.accept-any-schema=true` (the deprecated one), deduplicated so it isn't emitted for every column?
>
> Suggested text:
>
> > Spark-native schema evolution is being used with `write.spark.accept-any-schema=true`, which skips Spark’s normal schema checks and casts. For native evolution, use `accept-any-schema=false`.
>
> I believe it would run both schema evolutions which is weird. We can also add a test for this case?
added
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#issuecomment-5962904077
Could we add native `INSERT WITH SCHEMA EVOLUTION` integration tests? The current end-to-end coverage is for MERGE; the existing DataFrameWriter tests exercise legacy `merge-schema` behavior.
With `write.spark.accept-any-schema=false`, cover:
1. Adding a column while casting a SMALLINT source into an existing INT column in the same write. Assert that the target stays INT, the new column is added, existing rows get null for it, and inserted values are correct.
2. Widening INT to BIGINT using a value above `Integer.MAX_VALUE`. Assert the resulting type and preservation of old and new values.
3. Writing `map<bigint,string>` into `map<int,string>` with keys that fit INT. Assert that the key type stays INT and the write succeeds with the expected values.
Please parameterize these for by-name and by-position INSERT. Also, the accept-any-schema test currently only calls the hook twice; an actual write with native evolution and legacy `merge-schema` enabled would verify their interaction. Assert the final schema and rows.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4170631465
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
Review Comment:
Optional cleanup: return early for required columns or columns with defaults, avoiding type conversion for additions that will be rejected anyway.
```suggestion
TableChange.AddColumn add = (TableChange.AddColumn) change;
if (!add.isNullable() || add.defaultValue() != null) {
return false;
}
Type type = tryConvert(add.dataType());
return type != null && isSupportedAtFormatVersion(type);
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4140182837
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,89 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return ((TableChange.UpdateColumnNullability) change).nullable();
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds.contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && newType.isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(field.type(), newType.asPrimitiveType());
Review Comment:
Good pointt, supportsTypeUpdate now also requires the converted Iceberg type to round-trip back to the requested Spark type
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "huan233usc (via GitHub)" <gi...@apache.org>.
huan233usc commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4040060010
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +169,60 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
Review Comment:
Spark 4.2 recursively generates schema changes for map keys.
For example, evolving a target `map<int, string>` from a source `map<long, string>` produces `updateColumnType(["m", "key"], LongType)`. T
his implementation returns `true`, but `SchemaUpdate.apply()` explicitly rejects(as Iceberg don't allow it) updates to map keys (`Cannot update map keys`), so the write fails during `alterTable` instead of letting Spark skip the unsupported evolution and attempt a cast.
The same false positive occurs for additions or type changes inside a struct-valued map key. Could we reject changes whose field path targets the map-key subtree and add regression coverage for 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4170027177
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,90 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return ((TableChange.UpdateColumnNullability) change).nullable();
Review Comment:
Please also reject making an identifier field, or a struct containing one, nullable. `SchemaUpdate.apply()` rejects both through `Schema.validateIdentifierField`. Spark 4.2 currently generates only additions and type updates, so this is a hook-contract edge case that could be covered with a unit test.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -125,6 +139,7 @@ private SparkTable(
Table table, Schema schema, Snapshot snapshot, String branch, TimeTravel timeTravel) {
super(table, schema);
this.schema = schema;
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
Review Comment:
Could we lazily compute and cache `mapKeyFieldIds` when it is first needed? Eager initialization traverses the schema and allocates indexes for every `SparkTable` load, including reads and tables with schema evolution disabled. The schema is pinned, so the cache needs no invalidation.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,90 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
Review Comment:
Please check the converted type against the table’s format version before returning true, including nested types. `VariantType` and `NullType` convert successfully, but v1/v2 tables reject their Iceberg types in `Schema.checkCompatibility` during commit. Tests for v2 rejection and v3 acceptance would cover this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4150909103
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -94,10 +108,12 @@ public class SparkTable extends BaseSparkTable
TableCapability.OVERWRITE_DYNAMIC);
private final Schema schema; // effective schema (not necessarily current table schema)
+ private final Set<Integer> mapKeyFieldIds;
private final Snapshot snapshot; // always set unless table is empty
private final String branch; // set if table is loaded for specific branch
private final TimeTravel timeTravel; // set if table is loaded for time travel
private final Set<TableCapability> capabilities;
+ private final AtomicBoolean acceptAnySchemaWarningLogged = new AtomicBoolean(false);
Review Comment:
this is a bit hacky and has some issue:
There are two scope limitations:
- Reusing the same wrapper suppresses subsequent warnings throughout its lifetime.
- Reloading creates a fresh flag. Spark reloads after schema evolution, so one query could warn again if it checks remaining candidate changes.
But yes its not so clean. Maybe we should just drop this idea, sorry. As long as we have test coverage that nothing bad happens.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -94,10 +108,12 @@ public class SparkTable extends BaseSparkTable
TableCapability.OVERWRITE_DYNAMIC);
private final Schema schema; // effective schema (not necessarily current table schema)
+ private final Set<Integer> mapKeyFieldIds;
private final Snapshot snapshot; // always set unless table is empty
private final String branch; // set if table is loaded for specific branch
private final TimeTravel timeTravel; // set if table is loaded for time travel
private final Set<TableCapability> capabilities;
+ private final AtomicBoolean acceptAnySchemaWarningLogged = new AtomicBoolean(false);
Review Comment:
this is a bit hacky and has some issue:
There are two scope limitations:
- Reusing the same wrapper suppresses subsequent warnings throughout its lifetime.
- Reloading creates a fresh flag. Spark reloads after schema evolution, so one query could warn again if it checks remaining candidate changes.
Maybe we should just drop this idea, sorry. As long as we have test coverage that nothing bad happens.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#issuecomment-5896987560
Could we log a warning when this hook is invoked with `write.spark.accept-any-schema=true`, deduplicated so it isn't emitted for every column?
Suggested text:
> Spark-native schema evolution is being used with `write.spark.accept-any-schema=true`, which skips Spark’s normal schema checks and casts. For native evolution, use `accept-any-schema=false`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4237386588
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
Review Comment:
Optional: Could we move the schema-evolution checks and lazy map-key index into a package-private `SparkSchemaEvolution` helper in this package, leaving `supportsColumnChange` as a delegate? This would group the rules and make them easier to unit-test directly. There’s precedent for stateful helpers such as `SparkMicroBatchPlanner` and `SparkConfParser`, though keeping these checks here is also reasonable.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4140178379
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +169,60 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
Review Comment:
Fixed, good point supportsColumnChange now rejects map-key changes up front
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4264602141
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -94,10 +108,12 @@ public class SparkTable extends BaseSparkTable
TableCapability.OVERWRITE_DYNAMIC);
private final Schema schema; // effective schema (not necessarily current table schema)
+ private final Set<Integer> mapKeyFieldIds;
private final Snapshot snapshot; // always set unless table is empty
private final String branch; // set if table is loaded for specific branch
private final TimeTravel timeTravel; // set if table is loaded for time travel
private final Set<TableCapability> capabilities;
+ private final AtomicBoolean acceptAnySchemaWarningLogged = new AtomicBoolean(false);
Review Comment:
Done
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,90 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
Review Comment:
Done
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,90 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return ((TableChange.UpdateColumnNullability) change).nullable();
Review Comment:
Done
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -125,6 +139,7 @@ private SparkTable(
Table table, Schema schema, Snapshot snapshot, String branch, TimeTravel timeTravel) {
super(table, schema);
this.schema = schema;
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
Review Comment:
Done
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4237378072
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ if (!add.isNullable() || add.defaultValue() != null) {
+ return false;
+ }
+
+ Type type = tryConvert(add.dataType());
+ return type != null && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
+ if (mapKeyFieldIds == null) {
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
+ }
+
+ return mapKeyFieldIds;
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds().contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && newType.isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(field.type(), newType.asPrimitiveType())
+ && SparkSchemaUtil.convert(newType).equals(update.newDataType());
+ }
+
+ private boolean supportsNullabilityUpdate(TableChange.UpdateColumnNullability update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ return update.nullable() && field != null && !containsIdentifierField(field);
+ }
+
+ private boolean supportsDeleteColumn(TableChange.DeleteColumn delete) {
+ Types.NestedField field = schema.findField(String.join(".", delete.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ // Iceberg rejects dropping an identifier field or a field whose subtree contains one
+ return !containsIdentifierField(field);
Review Comment:
Nit: Could we return false when deleting a list element or map value itself, and add unit coverage? For example, `items.element` and `m.value` currently return true, but `SchemaUpdate.apply()` rejects both. Spark 4.2 only generates additions and type updates today, so this is a hook-contract edge case.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4225891945
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
+ if (mapKeyFieldIds == null) {
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
+ }
+
+ return mapKeyFieldIds;
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds().contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && isSupportedAtFormatVersion(newType)
Review Comment:
ok, the format version check now only runs for additions.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
Review Comment:
done
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4170580134
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
+ if (mapKeyFieldIds == null) {
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
+ }
+
+ return mapKeyFieldIds;
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds().contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && isSupportedAtFormatVersion(newType)
Review Comment:
As an optional cleanup, consider keeping `isSupportedAtFormatVersion` for additions and removing it from type updates. With the current promotion rules (INT to LONG, FLOAT to DOUBLE, decimal precision widening, or unchanged types), an update cannot introduce a type requiring a newer format version when the existing schema is valid. The primitive-type, promotion, and Spark round-trip checks are sufficient here.
```suggestion:-1+3
return newType != null
&& newType.isPrimitiveType()
&& TypeUtil.isPromotionAllowed(field.type(), newType.asPrimitiveType())
&& SparkSchemaUtil.convert(newType).equals(update.newDataType());
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4170570690
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
Review Comment:
Could we make this accessor `synchronized`, or declare `mapKeyFieldIds` `volatile`, so the initialized set is safely published if the same `SparkTable` is accessed concurrently? With the plain field, another thread is not guaranteed to see all of the `HashSet` contents. `volatile` is sufficient if duplicate initialization is acceptable, since the computed set is never mutated afterward.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4137377788
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,89 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ return add.isNullable() && add.defaultValue() == null && canConvert(add.dataType());
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return ((TableChange.UpdateColumnNullability) change).nullable();
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds.contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && newType.isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(field.type(), newType.asPrimitiveType());
Review Comment:
Could we reject type updates when the converted Iceberg type does not map back to the requested Spark type, and add an INSERT or MERGE test? For an INT target with a SMALLINT/TINYINT source, ShortType/ByteType converts to Iceberg int, so `isPromotionAllowed(int, int)` returns true and `updateColumn` is a no-op. Spark reloads the table, sees the same pending change, and fails with `UNSUPPORTED_AUTO_SCHEMA_EVOLUTION_CHANGES` instead of casting the source to INT.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4150912257
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -94,10 +108,12 @@ public class SparkTable extends BaseSparkTable
TableCapability.OVERWRITE_DYNAMIC);
private final Schema schema; // effective schema (not necessarily current table schema)
+ private final Set<Integer> mapKeyFieldIds;
private final Snapshot snapshot; // always set unless table is empty
private final String branch; // set if table is loaded for specific branch
private final TimeTravel timeTravel; // set if table is loaded for time travel
private final Set<TableCapability> capabilities;
+ private final AtomicBoolean acceptAnySchemaWarningLogged = new AtomicBoolean(false);
Review Comment:
I'll look on spark side to log the warning there.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4170580134
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
+ if (mapKeyFieldIds == null) {
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
+ }
+
+ return mapKeyFieldIds;
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds().contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && isSupportedAtFormatVersion(newType)
Review Comment:
As an optional cleanup, consider keeping `isSupportedAtFormatVersion` for additions and removing it from type updates. With the current promotion rules (INT to LONG, FLOAT to DOUBLE, decimal precision widening, or unchanged types), an update cannot introduce a type requiring a newer format version when the existing schema is valid. The primitive-type, promotion, and Spark round-trip checks are sufficient here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4225890078
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = tryConvert(add.dataType());
+ return add.isNullable()
+ && add.defaultValue() == null
+ && type != null
+ && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
Review Comment:
made it volatile
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho merged PR #17957:
URL: https://github.com/apache/iceberg/pull/17957
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#issuecomment-6179978440
Merged, thanks @rahulsmahadev and everhone else for additional reviews!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution [iceberg]
Posted by "rahulsmahadev (via GitHub)" <gi...@apache.org>.
rahulsmahadev commented on code in PR #17957:
URL: https://github.com/apache/iceberg/pull/17957#discussion_r4264603927
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ if (!add.isNullable() || add.defaultValue() != null) {
+ return false;
+ }
+
+ Type type = tryConvert(add.dataType());
+ return type != null && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability) change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
+ if (mapKeyFieldIds == null) {
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
+ }
+
+ return mapKeyFieldIds;
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field : TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+ keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 : fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames, pathLength)));
+ return field != null && mapKeyFieldIds().contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && newType.isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(field.type(), newType.asPrimitiveType())
+ && SparkSchemaUtil.convert(newType).equals(update.newDataType());
+ }
+
+ private boolean supportsNullabilityUpdate(TableChange.UpdateColumnNullability update) {
+ Types.NestedField field = schema.findField(String.join(".", update.fieldNames()));
+ return update.nullable() && field != null && !containsIdentifierField(field);
+ }
+
+ private boolean supportsDeleteColumn(TableChange.DeleteColumn delete) {
+ Types.NestedField field = schema.findField(String.join(".", delete.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ // Iceberg rejects dropping an identifier field or a field whose subtree contains one
+ return !containsIdentifierField(field);
Review Comment:
Done
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -158,6 +173,118 @@ public Set<TableCapability> capabilities() {
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
Review Comment:
Done
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org