Flink: Adjust the configuration precedence for the dynamic sink

[Guosmilesmile]

Now in DynamicWriter , the table configuration have higher priority and override the sink configuration. But generally, User-provided options should have precedence over table properties.

This aim is to adjust the configuration precedence for the dynamic sink.

[mxm]

Thanks for addressing this TODO @Guosmilesmile!

[mxm]

If we addressed this, we should remove the comment.

[mxm]

I wonder, should we use

iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkConfParser.java

Line 35 in cea1093

public class FlinkConfParser {

here?

[Guosmilesmile]

The Map<String, String> tableWriteProperties is passed multiple times to other classes within RowDataTaskWriterFactory. Changing it to use FlinkConfParser would involve significant modifications, especially since some places have special logic for tableWriteProperties.

Alternatively, could we add a new method Map<String, String> properties() in FlinkConfParser to handle priority uniformly?

Do you have any better suggestions?

[Guosmilesmile]

I made a version using FlinkConfParser. Please take a look and see if it’s appropriate.

[mxm]

Sorry, I think the old version was better. The Flink related configs are resolved earlier. At this point, we only have the general Iceberg write properties, for which it doesn't make sense to use the Flink conf parser.

[Guosmilesmile]

Thank you for your suggestion. I will revert the changes.

[pvary]

Is this a behavioral change? Do we need to document it? Is there a release already with this feature?

[pvary]

+1 so it could be merged to 1.10

Please create a test for this in follow up PR

[pvary]

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

Please in a follow-up PR create a test case which checks the config precedence.
Thanks,
Peter

[mxm]

Is this a behavioral change? Do we need to document it? Is there a release already with this feature?

Yes, but there is no official release of Dynamic Sink yet.

+1 for the test case.

[Guosmilesmile]

Thanks for the review ! I will prepare a new pr for the test case soon.

[b-rick]

This change results in unintuitive behaviour for parquet compression codec:

This is flink-write-conf:

FlinkWriteConf

  public String parquetCompressionCodec() {
    return confParser
        .stringConf()
        .option(FlinkWriteOptions.COMPRESSION_CODEC.key())
        .flinkConfig(FlinkWriteOptions.COMPRESSION_CODEC)
        .tableProperty(TableProperties.PARQUET_COMPRESSION)
        .defaultValue(TableProperties.PARQUET_COMPRESSION_DEFAULT)
        .parse();
  }

It defaults to TableProperties.PARQUET_COMPRESSION_DEFAULT (which is gzip, which seems wrong, but that is unrelated)

Now, in my table properties, I have set write.parquet-compression-codec to zstd explicitly.

in Dynamic Writer:

 Map<String, String> tableWriteProperties =
                          Maps.newHashMap(table.properties());
                      tableWriteProperties.putAll(commonWriteProperties);

So as a result of this change, my explicit setting of write.parquet-compression-codec is overriden by a default in FlinkWriteConf, which seems like bad behaviour. Is this intended?

[Guosmilesmile]

@b-rick Thank you very much for pointing this out. The configuration for parquet's default compression codec is somewhat special.

Since version 1.4.0, parquet's default compression codec changed from gzip to zstd by explicitly setting defaults in the table properties that apply only to new tables.

iceberg/core/src/main/java/org/apache/iceberg/TableMetadata.java

Lines 94 to 96 in 1bd8d5e

	persistedProperties.put(
	TableProperties.PARQUET_COMPRESSION,
	TableProperties.PARQUET_COMPRESSION_DEFAULT_SINCE_1_4_0);

Common sinks can obtain the table information in advance, but dynamic sinks can only get the table information at runtime. So, in common sinks, the table is passed to SinkUtil.writeProperties and the compression codec is obtained as zstd, but in DynamicIcebergSink, null is passed, so it defaults to gzip.

iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java

Lines 340 to 341 in 1bd8d5e

	Map<String, String> writeProperties =
	SinkUtil.writeProperties(flinkWriteConf.dataFileFormat(), flinkWriteConf, null);

In my opinion, instead of passing null to SinkUtil.writeProperties, we could pass a Map with the property write.parquet.compression-codec: zstd configured. This would solve the issue in this scenario. However, if new default table properties are added in the future, this place would also need to be updated accordingly, which is not very convenient.

@pvary @mxm Do you have any other suggestions?

[mxm]

I think we should be moving configuration code like

iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java

Line 339 in 1bd8d5e

FlinkWriteConf flinkWriteConf = new FlinkWriteConf(writeOptions, readableConfig);

into the runtime path, e.g. into the writer, where we can correctly resolve the write properties based on the user-provided properties, Flink config, table properties, and Iceberg-level defaults. It would be nice to use

iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java

Line 53 in 1bd8d5e

public FlinkWriteConf(

to resolve the write options.

[pvary]

I was playing around getting the config to the DynamicWriter, and found some serialization issues. If we can solve those then it would be nice the use the same mechanism to solve the configurations that in the normal sink