You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "mxm (via GitHub)" <gi...@apache.org> on 2025/12/01 13:53:58 UTC

[PR] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

mxm opened a new pull request, #14728:
URL: https://github.com/apache/iceberg/pull/14728

   This commit adds support for strict 1:1 schema synchronization by allowing columns to be dropped from the table when they are not present in the input schema. This is controlled via a new dropUnusedColumns parameter in DynamicIcebergSink.
   
   The default behavior (dropUnusedColumns=false) remains unchanged.


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3601229063

   >Before I go into the details of the PR, could you please help me understand what to expect in the interim period, when the sink receives records with and without the dropped column?
   
   New records will write with the new schema with any columns not part of the input schema dropped. Old records will continue to write with the old schema, which still exists. If there are any previously unseen schemas which include removed columns, those columns will be re-added as new columns. This is a catch which users will have to accept. That's why the feature is opt-in and disabled by default.


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2592654983


##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -99,20 +104,76 @@ public void testMakeTopLevelPrimitivesOptional() {
     assertThat(existingSchema.columns().stream().allMatch(Types.NestedField::isRequired)).isTrue();
 
     UpdateSchema updateApi = loadUpdateApi(existingSchema);
-    EvolveSchemaVisitor.visit(updateApi, existingSchema, new Schema());
+    EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, new Schema(), PRESERVE_COLUMNS);
     Schema newSchema = updateApi.apply();
     assertThat(newSchema.asStruct().fields()).hasSize(14);
     assertThat(newSchema.columns().stream().allMatch(Types.NestedField::isOptional)).isTrue();
   }
 
+  @Test
+  public void testDropUnusedColumns() {

Review Comment:
   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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2589023112


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   Yes, I agree.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3616678386

   Please update the documentation, and highlight the caveats, like what happens when the column is added back with the same name, and what happens when multiple schema changes happen simultaneously


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2598026436


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -36,30 +39,39 @@
  *   <li>Adding new columns
  *   <li>Widening the type of existing columsn
  *   <li>Reordering columns
+ *   <li>Dropping columns (when dropUnusedColumns is enabled)
  * </ul>
  *
  * We don't support:
  *
  * <ul>
- *   <li>Dropping columns
  *   <li>Renaming columns
  * </ul>
  *
- * The reason is that dropping columns would create issues with late / out of order data. Once we
- * drop fields, we wouldn't be able to easily add them back later without losing the associated
- * data. Renaming columns is not supported because we compare schemas by name, which doesn't allow
- * for renaming without additional hints.
+ * By default, columns that exist in the table but not in the input schema are made optional to
+ * avoid issues with late/out-of-order data. When dropUnusedColumns is enabled, these columns are
+ * dropped instead for 1:1 schema syncing.

Review Comment:
   Maybe something like this:
   > By default, any columns present in the table but absent from the input schema are marked as optional to prevent issues caused by late or out-of-order data. If dropUnusedColumns is enabled, these columns are removed instead to ensure a strict one-to-one schema alignment.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3601206474

   Before I go into the details of the PR, could you please help me understand what to expect in the interim period, when the sink receives records with and without the dropped column?


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588746794


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   Here both blocks:
   
   ```java
       for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
         if (tableField.isRequired() && struct.field(tableField.name()) == null) {
           // If a field from the table schema does not exist in the input schema, then we won't visit
           // it and check for required/optional compatibility. The only choice is to make the table
           // field optional.
           return Result.SCHEMA_UPDATE_NEEDED;
         }
       }
   
       if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
         if (dropUnusedColumns
             && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
           // We need to drop fields
           return Result.SCHEMA_UPDATE_NEEDED;
         }
         return Result.DATA_CONVERSION_NEEDED;
       }
   ```
   
   This can probably be simplified to:
   
   ```java
   
       for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
         if (struct.field(tableField.name()) == null && (tableField.isRequired() || dropUnusedColumns)) {
           // If a field from the table schema does not exist in the input schema, then we won't visit
           // it. The only choice is to make the table field optional or drop it.
           return Result.SCHEMA_UPDATE_NEEDED;
         }
       }
   
       if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
         return Result.DATA_CONVERSION_NEEDED;
       }
   
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: 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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588787885


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   We need the second solution, as there could be situations where one column is dropped, but one other column is added. We still need to detect this situation and update the schema.
   
   



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2598026436


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -36,30 +39,39 @@
  *   <li>Adding new columns
  *   <li>Widening the type of existing columsn
  *   <li>Reordering columns
+ *   <li>Dropping columns (when dropUnusedColumns is enabled)
  * </ul>
  *
  * We don't support:
  *
  * <ul>
- *   <li>Dropping columns
  *   <li>Renaming columns
  * </ul>
  *
- * The reason is that dropping columns would create issues with late / out of order data. Once we
- * drop fields, we wouldn't be able to easily add them back later without losing the associated
- * data. Renaming columns is not supported because we compare schemas by name, which doesn't allow
- * for renaming without additional hints.
+ * By default, columns that exist in the table but not in the input schema are made optional to
+ * avoid issues with late/out-of-order data. When dropUnusedColumns is enabled, these columns are
+ * dropped instead for 1:1 schema syncing.

Review Comment:
   Maybe something like this:
   > By default, any columns present in the table but absent from the input schema are made as optional to prevent issues caused by late or out-of-order data. If dropUnusedColumns is enabled, these columns are removed instead to ensure a strict one-to-one schema alignment.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2586056806


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   >  We don't change the RowData in this branch (which would be DATA_CONVERSION_NEEDED).
   
   I think this is what I don't understand. If we receive an R5 with S3 (C1 column dropped), then we drop the column and we don't need to change the RowData, as it already should not contain the C1 column.
   
   What do I miss?



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2584507940


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   What you describe will happen. Once we have created S2 via R2 and receive R3, we will iterate through all the table schemas and we will discover the old schema which matches S1. We will continue to write to F1.
   
   The branch here will be triggered when receiving a record with a new schema S3 (column dropped). `SCHEMA_UPDATE_NEEDED` will then trigger the schema update logic which does the column dropping. We don't change the RowData in this branch (which would be `DATA_CONVERSION_NEEDED`).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: 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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3601904822

   IMHO this is fine if the user opts in. We deliberately chose not to allow dropping columns because of this race condition; it's important that this remains the default setting. I agree that we should add more documentation around the semantics of removing columns. 


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588800380


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   Not sure if your comment points to the current version of my comment. I edited it 2 minutes after posting, around 8 minutes before your reply.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588317371


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   When we receive R5 with S3, we drop the column and write the associated RowData using the resulting schema. When we receive R6 with S3, we won't drop a column but write directly with that table schema which has the column dropped.
   
   >Why do we want to change the row?
   
   I think the misunderstanding stems from this question. The code here is not about changing RowData. It's about triggering to change the table schema.
   



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2584466119


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -103,11 +121,16 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
       after = columnName;
     }
 
-    // Ensure that unused fields are made optional
     for (Types.NestedField existingField : partnerStruct.fields()) {
       if (struct.field(existingField.name()) == null) {
-        if (existingField.isRequired()) {
-          this.api.makeColumnOptional(this.existingSchema.findColumnName(existingField.fieldId()));
+        String columnName = this.existingSchema.findColumnName(existingField.fieldId());
+        if (dropUnusedColumns) {
+          LOG.info("{}: Dropping column: {}", identifier.name(), columnName);

Review Comment:
   It logs for every record that triggers a schema change, but probably better to change this to DEBUG.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588746794


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   Here both blocks:
   
   ```java
       for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
         if (tableField.isRequired() && struct.field(tableField.name()) == null) {
           // If a field from the table schema does not exist in the input schema, then we won't visit
           // it and check for required/optional compatibility. The only choice is to make the table
           // field optional.
           return Result.SCHEMA_UPDATE_NEEDED;
         }
       }
   
       if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
         if (dropUnusedColumns
             && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
           // We need to drop fields
           return Result.SCHEMA_UPDATE_NEEDED;
         }
         return Result.DATA_CONVERSION_NEEDED;
       }
   ```
   
   This can probably be simplified to:
   
   ```java
   
       for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
         if (struct.field(tableField.name()) == null && (tableField.isRequired() || dropUnusedColumns)) {
           // If a field from the table schema does not exist in the input schema, then we won't visit
           // it. The only choice is to make the table field optional or drop it.
           return Result.SCHEMA_UPDATE_NEEDED;
         }
       }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: 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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3601724329

   Essentially, there’s a race condition between adding and dropping columns. For example, if a user does the following:
   1. Creates a new schema S1
   2. Sends a record R1 using S1
   3. Creates a new schema S2
   4. Sends a record R2 using S2
   
   If these actions occur within a short time frame and the streams are skewed, the table could end up with either:
   - Schema S2, if R2 arrives later
   - Schema S1, if R1 arrives later
   
   Afterward, querying the table with the β€œold” schema becomes difficult.
   
   Additionally, users cannot revert the table to any previously created schema using DynamicSink. This behavior is consistent with the current implementation, but with column-dropping support, users might expect this capability.
   
   @Guosmilesmile: Would these restrictions impact your use cases?
   


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2581519091


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;
+      }

Review Comment:
   nit: newline



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588432507


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   The original decision was `DATA_CONVERSION_NEEDED` needed. That is what confused me.
   
   But, `struct.fields().size() < tableSchemaType.asStructType().fields().size()` should be always `true` in this case. Otherwise we would always stop in the previous loop.
   



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3602358172

   1. For my scenario, because there are many downstream consumers, the impact of removing fields is often unknown and hard to assess, so removals are rare β€” fields are basically deprecated instead.  
   2. Since this feature is optional, it's acceptable to me; whether to use it should be left to users, but the limitations must be clearly documented. This is my personal opinion.


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2581536573


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -77,7 +77,11 @@ String branch(TableIdentifier identifier, String branch) {
   }
 
   ResolvedSchemaInfo schema(TableIdentifier identifier, Schema input) {
-    return schema(identifier, input, true);
+    return schema(identifier, input, true, false);

Review Comment:
   we usually add a comment to the boolean parameters if we can't avoid them.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2592498474


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -36,30 +39,39 @@
  *   <li>Adding new columns
  *   <li>Widening the type of existing columsn
  *   <li>Reordering columns
+ *   <li>Dropping columns (when dropUnusedColumns is enabled)
  * </ul>
  *
  * We don't support:
  *
  * <ul>
- *   <li>Dropping columns
  *   <li>Renaming columns
  * </ul>

Review Comment:
   Nit: remove the list, as it is not needed anymore πŸ˜„ 



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #14728:
URL: https://github.com/apache/iceberg/pull/14728


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2592496794


##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -99,20 +104,76 @@ public void testMakeTopLevelPrimitivesOptional() {
     assertThat(existingSchema.columns().stream().allMatch(Types.NestedField::isRequired)).isTrue();
 
     UpdateSchema updateApi = loadUpdateApi(existingSchema);
-    EvolveSchemaVisitor.visit(updateApi, existingSchema, new Schema());
+    EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, new Schema(), PRESERVE_COLUMNS);
     Schema newSchema = updateApi.apply();
     assertThat(newSchema.asStruct().fields()).hasSize(14);
     assertThat(newSchema.columns().stream().allMatch(Types.NestedField::isOptional)).isTrue();
   }
 
+  @Test
+  public void testDropUnusedColumns() {

Review Comment:
   Could we have a test when the dropped column is inside a struct, array or map?
   Just to ensure that we traverse the hierarchy correctly



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3616520267

   Rebased to fix merge conflicts in CI after #14406.


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#issuecomment-3602429696

   Cool! Seems like an ok feature.
   Let's move forward with the review then.


-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2592237853


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -118,13 +120,16 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
 
   private TableMetadataCache.ResolvedSchemaInfo findOrCreateSchema(
       TableIdentifier identifier, Schema schema) {
-    TableMetadataCache.ResolvedSchemaInfo fromCache = cache.schema(identifier, schema);
-    if (fromCache.compareResult() != CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED) {
+    TableMetadataCache.ResolvedSchemaInfo fromCache =
+        cache.schema(identifier, schema, dropUnusedColumns);
+    if (!dropUnusedColumns
+        && fromCache.compareResult() != CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED) {

Review Comment:
   Could you please help me why is this `dropUnusedColumns` needed here?
   If I understand correctly the schema comparison only changed to return the `SCHEMA_UPDATE_NEEDED` more often (even if we need to drop a column)



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2592655694


##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -99,20 +104,76 @@ public void testMakeTopLevelPrimitivesOptional() {
     assertThat(existingSchema.columns().stream().allMatch(Types.NestedField::isRequired)).isTrue();
 
     UpdateSchema updateApi = loadUpdateApi(existingSchema);
-    EvolveSchemaVisitor.visit(updateApi, existingSchema, new Schema());
+    EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, new Schema(), PRESERVE_COLUMNS);
     Schema newSchema = updateApi.apply();
     assertThat(newSchema.asStruct().fields()).hasSize(14);
     assertThat(newSchema.columns().stream().allMatch(Types.NestedField::isOptional)).isTrue();
   }
 
+  @Test
+  public void testDropUnusedColumns() {

Review Comment:
   in 4bbc394.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2592755838


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -36,30 +39,39 @@
  *   <li>Adding new columns
  *   <li>Widening the type of existing columsn
  *   <li>Reordering columns
+ *   <li>Dropping columns (when dropUnusedColumns is enabled)
  * </ul>
  *
  * We don't support:
  *
  * <ul>
- *   <li>Dropping columns
  *   <li>Renaming columns
  * </ul>

Review Comment:
   If you don't mind, I keep this list with one element πŸ˜„ 



##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -36,30 +39,39 @@
  *   <li>Adding new columns
  *   <li>Widening the type of existing columsn
  *   <li>Reordering columns
+ *   <li>Dropping columns (when dropUnusedColumns is enabled)
  * </ul>
  *
  * We don't support:
  *
  * <ul>
- *   <li>Dropping columns
  *   <li>Renaming columns
  * </ul>

Review Comment:
   If you don't mind, I'll keep this list with one element πŸ˜„ 



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2588732641


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   The previous loop is there to detect required fields not present in the input schema. If the to-be-dropped field is not required, then the previous loop won't lead to returning from the method.



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2581596634


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -97,6 +101,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
     }
 
     if (struct.fields().size() != tableSchemaType.asStructType().fields().size()) {
+      if (dropUnusedColumns
+          && struct.fields().size() < tableSchemaType.asStructType().fields().size()) {
+        // We need to drop fields
+        return Result.SCHEMA_UPDATE_NEEDED;

Review Comment:
   Why do we need the `SCHEMA_UPDATE_NEEDED` for `dropUnusedColumns`?
   
   IIUC the use-case is that previously we had a schema which contained all of the columns, but we received an new schema without the column. My understanding was that in this case we add a new file with the old schema.
   
   So as an example:
   - Table has S1 schema
   - R1 arrives with S1 schema - R1 is written to F1
   - R2 arrives with S2 schema (added new column) - R2 is written to F2
   - R3 arrives with S1 schema - R3 is written to F1
   - R4 arrives with S2 schema - R4 is written to F2
   
   Why do we want to change the row?



-- 
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] Flink: Dynamic Sink: Add support for dropping columns [iceberg]

Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14728:
URL: https://github.com/apache/iceberg/pull/14728#discussion_r2581531958


##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -103,11 +121,16 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
       after = columnName;
     }
 
-    // Ensure that unused fields are made optional
     for (Types.NestedField existingField : partnerStruct.fields()) {
       if (struct.field(existingField.name()) == null) {
-        if (existingField.isRequired()) {
-          this.api.makeColumnOptional(this.existingSchema.findColumnName(existingField.fieldId()));
+        String columnName = this.existingSchema.findColumnName(existingField.fieldId());
+        if (dropUnusedColumns) {
+          LOG.info("{}: Dropping column: {}", identifier.name(), columnName);

Review Comment:
   We will log this for every single record?



-- 
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