Skip to content

Flink: Add TableCreator interface to set table properties/location on DynamicIcebergSink table creation - #14578

Merged
pvary merged 1 commit into
apache:mainfrom
jordepic:FLINK_INITIAL_TABLE_PROPERTIES
Nov 17, 2025
Merged

pvary merged 1 commit into
apache:mainfrom
jordepic:FLINK_INITIAL_TABLE_PROPERTIES

Conversation

@jordepic

Copy link
Copy Markdown
Contributor

The DynamicIcebergSink does not currently allow providing a set of reasonable default table properties at table creation. This change will apply these properties when a table is created (not updated). This helps for specifying things like iceberg version, merge-on-read vs. copy-on-write, table location, and others.

@jordepic
jordepic force-pushed the FLINK_INITIAL_TABLE_PROPERTIES branch from 06425db to 1ddd45d Compare November 12, 2025 23:01
Comment thread docs/docs/flink-writes.md Outdated
@pvary

pvary commented Nov 13, 2025

Copy link
Copy Markdown
Contributor

I’ve added a few comments on the PR, but the bigger question is deciding what functionality we actually want.
Here are the ideas I’ve heard so far:

  • Apply the same table properties to every table at creation (current PR).
  • Set table properties at creation based on the DynamicRecord for each table individually.
  • Define the table location during creation (per table).
  • Update table properties to ensure a specific set of properties is always present.
  • Allow V2 → V3 migration (convert DVs) — related only in that operators need up-to-date knowledge of table properties.

We need to draw a clear line: what should the Dynamic Sink support, and what should be handled by external operators outside the Dynamic Iceberg Sink?

To make this decision, I’d like input from actual users: @jordepic, @mxm, @Guosmilesmile, or anyone else interested. Maybe someone could even start a discussion on the dev list to reach a wider audience.

If some changes are handled by external operators, we may need a mechanism for those operators to notify the Dynamic Iceberg Sink to refresh its caches. One option could be using something like the Flink Orchestrator restart nonce (https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/job-management/#application-restarts-without-spec-change). If the cache detects a new nonce (greater than the current one), it could invalidate the current table cache and reload the data.

@jordepic

Copy link
Copy Markdown
Contributor Author

I think our requirements, for now, are twofold:

  1. Sensible default properties on table creation
  2. Configurable locations per table (also just on creation)
    I don't think that we'll ever need to dynamically update table properties though due to a dynamic record!

This change is just addressing number 1! I'm happy to expand it to do number 2 as well, though I'm less sure of the business logic that everybody needs there, hence why I excluded it for now.

@jordepic
jordepic force-pushed the FLINK_INITIAL_TABLE_PROPERTIES branch from 1ddd45d to 3805827 Compare November 13, 2025 17:33
@mxm

mxm commented Nov 14, 2025

Copy link
Copy Markdown
Contributor

Let's start with a narrow scope for this feature. We could always extend it to DynamicRecord, but I'm not convinced the feature should live in DynamicRecord because table properties aren't connected directly to the table data (that's why I kept it outside in #13883). The majority of the use cases will only require setting table properties on table creation. Concerning (2), we've had users request setting the table location as well, so I would suggest to add this feature alongside the table properties on table creation.

@pvary

pvary commented Nov 14, 2025

Copy link
Copy Markdown
Contributor

Do we need different properties for different tables, or we can use/set the same properties for every table?

@mxm

mxm commented Nov 14, 2025

Copy link
Copy Markdown
Contributor

I think users will need the table identifier.

@jordepic
jordepic force-pushed the FLINK_INITIAL_TABLE_PROPERTIES branch from 3805827 to a1aca93 Compare November 14, 2025 15:53
Comment thread docs/docs/flink-writes.md Outdated
@pvary

pvary commented Nov 14, 2025

Copy link
Copy Markdown
Contributor

Please rename the PR to match the new feature added

@jordepic
jordepic force-pushed the FLINK_INITIAL_TABLE_PROPERTIES branch from a1aca93 to 59e291a Compare November 14, 2025 16:06
@jordepic jordepic changed the title Flink: Set table properties on DynamicIcebergSink table creation Flink: Add TableCreator interface to set table properties/location on DynamicIcebergSink table creation Nov 14, 2025
@jordepic

Copy link
Copy Markdown
Contributor Author

Please rename the PR to match the new feature added

Done.

The DynamicIcebergSink does not currently allow providing
a set of reasonable default table properties at creation time.
This change will apply these properties when a table is created
(not updated). This helps for specifying things like iceberg
version, merge-on-read vs. copy-on-write, table location,
and others.
@jordepic
jordepic force-pushed the FLINK_INITIAL_TABLE_PROPERTIES branch from 59e291a to 274d118 Compare November 14, 2025 18:02

@pvary pvary left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM.
@mxm, @Guosmilesmile: Please review!

@Guosmilesmile Guosmilesmile left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM.

@mxm mxm left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM. Great improvement @jordepic!

@pvary
pvary merged commit 31ce47b into apache:main Nov 17, 2025
22 checks passed
@pvary

pvary commented Nov 17, 2025

Copy link
Copy Markdown
Contributor

Merged to main.
Thanks for the PR @jordepic, and @mxm and @Guosmilesmile for the reviews!

@jordepic: Please port the changes to the other Flink versions.
This command could help:

g diff HEAD^ flink/v2.1 |sed "s/v2.1/v2.0/g">/tmp/patch;g apply -3 -p1 /tmp/patch

Please tell us on the PR if the backport was clean, and if not, please highlight the manual changes required.

Thanks,
Peter

jordepic pushed a commit to jordepic/iceberg that referenced this pull request Nov 17, 2025
DynamicIcebergSink table creation (backport)

This is a *clean* backport of apache#14578 to Flink 2.0

The DynamicIcebergSink does not currently allow providing
a set of reasonable default table properties at creation time.
This change will apply these properties when a table is created
(not updated). This helps for specifying things like iceberg
version, merge-on-read vs. copy-on-write, table location,
and others.
jordepic pushed a commit to jordepic/iceberg that referenced this pull request Nov 17, 2025
DynamicIcebergSink table creation (backport)

This is a *clean* backport of apache#14578 to Flink 2.0

The DynamicIcebergSink does not currently allow providing
a set of reasonable default table properties at creation time.
This change will apply these properties when a table is created
(not updated). This helps for specifying things like iceberg
version, merge-on-read vs. copy-on-write, table location,
and others.
jordepic pushed a commit to jordepic/iceberg that referenced this pull request Nov 17, 2025
DynamicIcebergSink table creation (backport to 2.0 and 1.20)

This is a *clean* backport of apache#14578 to Flink 2.0

There are 2 unclean lines in TestDynamicTableUpdateOperator
where we add the TableCreator parameter.

The DynamicIcebergSink does not currently allow providing
a set of reasonable default table properties at creation time.
This change will apply these properties when a table is created
(not updated). This helps for specifying things like iceberg
version, merge-on-read vs. copy-on-write, table location,
and others.
jordepic pushed a commit to jordepic/iceberg that referenced this pull request Nov 17, 2025
DynamicIcebergSink table creation (backport to 2.0 and 1.20)

This is a *clean* backport of apache#14578 to Flink 2.0

There are 2 unclean lines in TestDynamicTableUpdateOperator
where we add the TableCreator parameter.

The DynamicIcebergSink does not currently allow providing
a set of reasonable default table properties at creation time.
This change will apply these properties when a table is created
(not updated). This helps for specifying things like iceberg
version, merge-on-read vs. copy-on-write, table location,
and others.
thomaschow pushed a commit to thomaschow/iceberg that referenced this pull request Jan 19, 2026
talatuyarer pushed a commit to talatuyarer/iceberg that referenced this pull request Apr 1, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants