[Watermarking] Allow assigning watermark on stream start

[george-zubrienko]

There are a few edge cases that will benefit from having watermark assigned on start:

• Test setups, to avoid creating temporary tables with fake schema
• Watermark repairs or complete watermark reset due to source versioning changes

Hence, the ideal scenario would be to add WatermarkProvider[Watermark] trait that will be injected in StreamBootstrapper (can be combined with SchemaProvider), and conditionally based on a flag in sinkSettings apply epoch watermark value to target. This change should include further unification of schema and watermarking evaluation interfaces on Reader class levels.

[f-thiele]

For my understanding: the watermark assigned on start would indicate the earliest mark that arcane reads from the source I would venture?

• Does it just assume the start watermark to be set as configured and then read until HEAD of the source?
• Or is it just assigning that start-watermark to the output table (that may or may not have a conflicting watermark already) and then proceed with it's normal streaming operation?

I presume a difference would be in the situation where the stream fails, i.e. whether the watermark has been assigned to the table or is just living within the stream definition.

[george-zubrienko]

For my understanding: the watermark assigned on start would indicate the earliest mark that arcane reads from the source I would venture

In the proposition, a much more simple option: we assign epoch watermark, which is null time essentially, value depends on the source. Hence:

• Does it just assume the start watermark to be set as configured and then read until HEAD of the source - No, epoch is a constant defined in respective watermark implementations
• Or is it just assigning that start-watermark to the output table (that may or may not have a conflicting watermark already) and then proceed with it's normal streaming operation - Yes, this is the case. Any previous watermark, if exists, will be overriden.