Flink: Watermark read options

[rodmeneses]

1. Adds read option to specify watermark-column for Flink Iceberg connector
2. Adds read option to specify watermark-column-time-unit for Flink Iceberg connector
   Now it is possible to pass the above parameters in the SQL statement:

select * from t /*+ OPTIONS('watermark-column'='t2','watermark-column-time-unit'='MILLISECONDS')*/

[mas-chen]

Is this intended? This would be a breaking change if a user did not provide a split assigner type before

[rodmeneses]

good catch.
I changed that to null, as we had an assertion before:

      if (splitAssignerFactory != null) {
        Preconditions.checkArgument(
            watermarkColumn == null,
            "Watermark column and SplitAssigner should not be set in the same source");
      }

however, after my work on this PR, it is safe to remove the above assertion, since we will override the factory with a OrderedSplitAssignerFactory when the watermarkColumn is specified

[mas-chen]

Makes sense. It would be good to update the docs to reveal how the default is determined

[rodmeneses]

I'm updating the doc entry for watermark-column:

Specifies the watermark column to use for watermark generation. If this option is present, the splitAssignerFactory will be overriden with OrderedSplitAssignerFactory.

wdyt? @mas-chen @pvary

[mas-chen]

If users have not opt-ed into the new feature, I would expect their SQL queries to still work. Can you add tests to ensure that?

After some thinking, I'm totally not sure about the default change. Couldn't it be possible to use the watermark column, without the ordered split assigner factory?

[rodmeneses]

Not according to the code.

   if (watermarkColumn != null) {
        // Column statistics is needed for watermark generation
        context = context.copyWithColumnStat(watermarkColumn);
        SplitWatermarkExtractor watermarkExtractor =
            new ColumnStatsWatermarkExtractor(icebergSchema, watermarkColumn, watermarkTimeUnit);
        emitter = SerializableRecordEmitter.emitterWithWatermark(watermarkExtractor);
        splitAssignerFactory =
            new OrderedSplitAssignerFactory(SplitComparators.watermark(watermarkExtractor));
      }

[rodmeneses]

cc: @pvary

[pvary]

If the splits are not ordered, then we will have fluctuating watermarks. We do not emit those, which are not in order, but beats the purpose of the whole watermark generation feature.
Imagine a situation where we reading time series data, and read the latest file first. Every other file will contain late data in this case, and might be dropped.

So while technically possible, I rather not allow the users to shoot themselves in the foot.

[rodmeneses]

If users have not opt-ed into the new feature, I would expect their SQL queries to still work. Can you add tests to ensure that?

Hi! I didn't get your suggestion about the unit test. Would you please rephrase? Thanks @mas-chen

[pvary]

This comment is not correct:

// File 2 - late timestamps, old longs

[pvary]

I do not get this comment.
Maybe something like this - feel free to reword if you feel so:

// Expected records if the splits are ordered
//     - ascending (watermark from t1) - records from the split with early timestamps then records from the split with late timestamps
//     - descending (watermark from t2) - records from the split with old longs then records from the split with new longs

[pvary]

Left some minor comments for comments 😄
Otherwise +1 LGTM

[pvary]

Thanks @rodmeneses for the PR, and @stevenzwu and @mas-chen for the review!