Non-data fields - Metadata fields to Kafka Connector

[vaibhawvipul]

enabling users to query using the following syntax -

create table users (
    id TEXT,
    name TEXT,
    offset BIGINT GENERATED ALWAYS AS (metadata('offset_id')) STORED
) with (
    connector = 'kafka',
    ...
);

Demo -

[mwylde]

Great progress on the PR!

A few remaining issues.

We're now validating metadata names (although the key/value are switched), but we're missing two types of validations for the user:

• Data types, which if the user sets the wrong type they get a runtime error, for example if they declare

    offset_id INT GENERATED ALWAYS AS (metadata('offset_id')) STORED

and we're not validating that the metadata columns have only a metadata function in them, so this plans:

    offset_id BIGINT GENERATED ALWAYS AS (concat(metadata('offset_id'), '1')) STORED

Beyond that, I'd like to see some tests, both for the deserialization logic and for planning. It might also be worth modifying the kafka tests to use metadata fields.

[vaibhawvipul]

@mwylde can you please trigger the CI again? I am asking because my last commit ran successfully, the failed CI is on master merge.

[mwylde]

Thanks, this is a great contribution!