Flink: Dynamic Iceberg Sink: Add sink / core processing logic / benchmarking

[mxm]

This completes the breakdown of #12424. Most importantly, it adds the following missing components:

DynamicIcebergSink

The actual Flink sink which ties together the previously merged components. The sink will be instantiated by the user like this:

DynamicIcebergSink
  .forInput(inputStream)
  .withConverter(new MyConverter())
  .catalogLoader(..)
  .append();  

An example for a converter implementation would be:

class MyConverter implements DynamicRecordConverter<MyInput> {

  final RuntimeParams params;

  MyConverter(RuntimeParams params) {
    this.params = params;
  }

  @Override
  public void convert(MyInput inputRecord, Collector<DynamicRecord> out) {

    DynamicRecord record = new DynamicRecord(
        inputRecord.getTable(),
        inputRecord.getBranch(),
        inputRecord.buildSchema(),
        inputRecord.convertData(),
        PartitionSpec.unpartitioned(),
        params.distributionMode(),
        params.writeParallelism());
        
    out.collect(record);
  }
}

DynamicRecordProcessor

The core processing logic which takes user-provided input to write to Iceberg tables and make the necessary changes to the Iceberg metadata.

Benchmarks

• DynamicRecordSerializerDeserializerBenchmark
• TestDynamicIcebergSinkPerf

Previous PRs

The following previous PRs have been merged already:

1. Flink: Add DynamicRecord / DynamicRecordInternal / DynamicRecordInternalSerializer #12996
2. Flink: Dynamic Iceberg Sink: Add table update code for schema comparison and evolution  #13032
3. Flink: Dynamic Iceberg Sink: Add dynamic writer and committer #13080
4. Flink: Dynamic Iceberg Sink: Add HashKeyGenerator / RowDataEvolver / TableUpdateOperator #13277

[pvary]

Maybe some tests with state restore?

[pvary]

Also, can we have a check were there is concurrent insert to one of the tables we are writing to?

[pvary]

Do we have a test where there is a concurrent commit to one of the tables by a different sink, or application?

[mxm]

I've added a test to fail before / after commit, and one for concurrent commits: 3ef4d87.

Do we have a test where there is a concurrent commit to one of the tables by a different sink, or application?

Not yet.

[pvary]

Merged to main.
Thanks for the PR @mxm!

[mxm]

Thanks for reviewing / merging @pvary!