pvary commented on code in PR #13608:
URL: https://github.com/apache/iceberg/pull/13608#discussion_r2228613329
##########
docs/docs/flink-writes.md:
##########
@@ -396,3 +396,137 @@ To use SinkV2 based implementation, replace `FlinkSink`
with `IcebergSink` in th
- The `RANGE` distribution mode is not yet available for the `IcebergSink`
- When using `IcebergSink` use `uidSuffix` instead of the `uidPrefix`
+
+
+## Dynamic Iceberg Flink Sink
+
+Dynamic Flink Iceberg Sink allows:
+
+1. **Writing to any number of tables**
+ A single sink can dynamically route records to multiple Iceberg tables.
+
+2. **Dynamic table creation and updates**
+ Tables are created and updated based on user-defined routing logic.
+
+3. **Dynamic schema and partition evolution**
+ Table schemas and partition specs update during streaming execution.
+
+All configurations are controlled through the `DynamicRecord` class,
eliminating the need for Flink job restarts when requirements change.
+
+```java
+
+ DynamicIcebergSink.forInput(dataStream)
+ .generator((inputRecord, out) -> out.collect(
+ new DynamicRecord(
+ TableIdentifier.of("db", "table"),
+ "branch",
+ SCHEMA,
+ (RowData) inputRecord,
+ PartitionSpec.unpartitioned(), DistributionMode.HASH,
2)))
+ .catalogLoader(CatalogLoader.hive("hive", new Configuration(),
Map.of()))
+ .writeParallelism(10)
+ .immediateTableUpdate(true)
+ .append();
+```
+
+### Configuration Example
+
+```java
+DynamicIcebergSink.Builder<RowData> builder =
DynamicIcebergSink.forInput(inputStream);
+
+// Set common properties
+builder
+ .set("write-format", "parquet");
+
+// Set Dynamic Sink specific options
+builder
+ .writeParallelism(4)
+ .uidPrefix("dynamic-sink")
+ .cacheMaxSize(500)
+ .cacheRefreshMs(5000);
+
+// Add generator and append sink
+builder.generator(new CustomRecordGenerator());
+builder.append();
+```
+
+### Dynamic Routing Configuration
+
+Dynamic table routing can be customized by implementing the
`DynamicRecordGenerator` interface:
+
+```java
+public class CustomRecordGenerator implements DynamicRecordGenerator<RowData> {
+ @Override
+ public DynamicRecord generate(RowData row) {
+ DynamicRecord record = new DynamicRecord();
+ // Set table name based on business logic
+ TableIdentifier tableIdentifier = TableIdentifier.of(database,
tableName);
+ record.setTableIdentifier(tableIdentifier);
+ record.setData(row);
+ // Set the maximum number of parallel writers for a given
table/branch/schema/spec
+ record.writeParallelism(2);
+ return record;
+ }
+}
+
+// Set custom record generator when building the sink
+DynamicIcebergSink.Builder<RowData> builder =
DynamicIcebergSink.forInput(inputStream);
+builder.generator(new CustomRecordGenerator());
+// ... other config ...
+builder.append();
+```
+The user should provide a converter which converts the input record to a
DynamicRecord.
+We need the following information (DynamicRecord) for every record:
+
+| Property | Description
|
+|---------------------|-------------------------------------------------------------------------------------------|
+| `TableIdentifier` | The target table to which the record will be written.
|
+| `Branch` | The target branch for writing the record (optional).
|
+| `Schema` | The schema of the record.
|
+| `Spec` | The expected partitioning specification for the
record. |
+| `RowData` | The actual row data to be written.
|
+| `DistributionMode` | The distribution mode for writing the record
(currently supports NONE or HASH). |
+| `Parallelism` | The maximum number of parallel writers for a given
table/branch/schema/spec (WriteTarget). |
+| `upsertMode` | Overrides this table's write.upsert.enabled
(optional). |
+| `equalityFields` | The equality fields for the table(optional).
|
+
+### Schema Update
+
+When the schema in the DynamicRecord is inconsistent with the schema currently
cached for the target table, a schema update will be triggered. This cache is
an LRU cache controlled by size and time.
+
+If an update is triggered and immediateTableUpdate is set to false, a
non-concurrent way of updating the table will be provided.
+
+We support:
+- Adding new columns
+- Widening the type of existing columsn
+- Reordering columns
+
+We don't support:
+- Dropping columns
+- Renaming columns
+
+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.
+
Review Comment:
The dynamic sink maintains an LRU cache for both table metadata and incoming
schemas, with eviction based on size and time constraints. When a DynamicRecord
contains a schema that is incompatible with the current table schema, a schema
update is triggered. This update can occur either immediately or via a
centralized executor, depending on the immediateTableUpdate configuration.
While centralized updates reduce load on the Catalog, they may introduce
backpressure on the sink.
Supported schema updates:
- Adding new columns
- Widening existing column types (e.g., Integer → Long, Float → Double)
Unsupported schema updates:
- Dropping columns
- Renaming columns
Dropping columns is avoided to prevent issues with late or out-of-order
data, as removed fields cannot be easily restored without data loss. Renaming
is unsupported because schema comparison is name-based, and renames would
require additional metadata or hints to resolve.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
