Support for RowBinaryWithNames

[avx-wlauer]

Use case

As a pipeline author, I need the ability for my code to reliably write to a clickhouse table without having to provide values for all the columns AND without having to know about all the columns in the table. My schema is regularly updated with new columns independent of the deploy of a new pipeline version (schema updates are often required to be applied prior to code that actually populates the data). As long as the new columns have reasonable defaults, the pipeline should be able to continue to function, writing data without error, without being updated.

Describe the solution you'd like

Support writing a subset of the columns in a table using a RowBinary-based format.

For performance reasons, I'd like to use a RowBinary-based format instead of converting data to JSON. Unfortunately, I don't have complete control over when new columns are added to my schema, so its possible that new columns will be added sometime while my pipeline is currently running. The default RowBinary format requires that ALL columns are included in the data, or misaligment and other errors occur. By allowing the use of RowBinaryWithNames format, Clickhouse should be able to continue to map data generate by the pipeline to the correct columns even after the addition of new columns to a table's schema.

Describe the alternatives you've considered

It's possible to make this work with other formats, including JSON, which implicitly declare the column mapping, but I don't want to deal with the performance penalty of converting to/from JSON. This can probably also be managed with AVRO or Parquet formats, but the RowBinary and its associated derivatives are the native formats supported by ClickHouse and should have the best performance.

Additional context

[avx-wlauer]

Thinking about this, I think the preferred solution is likely something similar to the pattern used in the Sink v2 API. I'd suggest adding a new interface, SupportsHeader that can be implemented by a POJOConvertor. When its type to submit a batch, iif a call to ClickHouseFormat.hasHeader() returns true, then the pojo converter is checked to see if it implements SupportsHeader. If it does, then SupportsHeader.generateHeader() is called and the output prepended to the batch being sent to clickhouse. If hasHeader() is true but the SupportsHeader interface is NOT implemented by the POJOConverter, then an error is thrown.

Ideally, the presence of a header is checked early, during setup, so that errors happen before having to process any elements. Also, the size of the header needs to be computed and added in the batch size computations to avoid any overflows.

[avx-wlauer]

Raising PR #108 as a proposed solution to this problem. Currently, it only supports Flink 2.x, but I'll look at adding 1.7 support as well.