pvary

Thanks @fbocse for the review! > @pvary thank you for following up on this Actually @lcspinter is the one who is working on this 😄 > This basically "locks" the...

How would the single phase commit solution support the exact one semantics for the Iceberg Sink?

@mudit-97: If your job fails after the FlinkSink finishing the `snapshotState`, but before the `notifyCheckpointComplete`, and any of the other operators fail in the `snapshotState` then Flink could decide that...

@mudit-97: The Flink community did outstanding work to handle all of these corner cases. I do not think it is a good idea to throw away that work. What is...

> I thought Hive only lowercases the top level column names. Does it also lowercase the fields in structs? Maybe this is because we use the lowercase config when we...

Could we add a test too? Minimally for the new timestamp?

> @pvary There are already quite a few tests: https://github.com/apache/iceberg/blob/main/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReader.java > > This writes and then reads a Parquet files using the visitor. I've added another test, but it is...

Thanks @Fokko for the PR! Please do not forget the PR for the other Flink versions as well.

@Aireed: Maybe the issue would be better handled, if Flink would run the refresh before commiting a new snapshot. Is this issue still happens with newer Iceberg versions?

@maekchi, @Aireed: Which catalog are you using? The `SnapshotProducer` constructor uses `ops.current()` to refresh the base snapshot, like: https://github.com/apache/iceberg/blob/d6c8358ff26957c9234580addb03a0db1e441c4d/core/src/main/java/org/apache/iceberg/SnapshotProducer.java#L111 Which should take care of refreshing the current snapshot when the...