Flink: Dynamic Sink: Add support for dropping columns

[mxm]

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.

[pvary]

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?

[mxm]

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.

[pvary]

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?

[mxm]

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.

[Guosmilesmile]

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.

[pvary]

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

[mxm]

Rebased to fix merge conflicts in CI after #14406.

[pvary]

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

[pvary]

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.

[pvary]

Merged to main.
Thanks for the new feature @mxm and @Guosmilesmile for the review!