feat: only ingest key-ed value in additional header column

[tabVersion]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

syntax INCLUDE HEADER 'header_col' AS column_name and only header can specify the inner field part.
if alias not specified, the header with inner field will be named _rw_kafka_header_{inner field}

if an inner field name is specified, the column type becomes bytea instead of Array[Struct<Varchar, Bytea>]

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

support syntax specially for header

create table/source s (...) include header '<header col name>' [varchar/bytea] [as <alias>]

now we can specify desired keys in header and make it a column

---

here is an example: given a header [(key1, value1), (key2, value2)]

and the clause is include header 'key1', the column content is value1 in bytes
if the clause is include header 'key1' varchar the column content is value1 in varchar

about the default naming:

in prev impl, the default name is _rw_{connector name}_{additional column type}
In this pr we introduced inner field and type hint, so it can be _rw_{connector name}_{additional column type}_{<inner field name>}_{type hint}
to be specific, for include header 'header1' bytea, the name is _rw_kafka_header_header1_bytea and for include header 'header2' varchar the name is _rw_kafka_header_header2_varchar

[xxchan]

additional_column_type is included in 1.6, but not documented.

#14215 (comment)

Why do we need to deprecate this field here?

[xxchan]

Oh, for source created in 1.6, it will have AdditionalColumnType::NORMAL. So we cannot change the type for field 9

[xxchan]

Have we told any poc user to try include before? We might need to tell them to rebuild the sources later. Or maybe we just document this breaking change in the release note.

[tabVersion]

Yes, the code is in v1.6.0 but the feature but not considered as released. It is ok to ignore the non-normal columns.
And I don't want to make breaking changes to normal columns here so I choose to use a new field.

I want to make things flexible when handling additional columns. Just like this change, the prev enum is not sufficient with handling an extra inner field arg. I don't know what comes next, so I make all columns a message instead of an enum.

[fuyufjh]

The breaking change looks acceptable to me, although it seems not hard to make it backward compatible.

[xxchan]

Do we need to add a new ColumnDescVersion? TBH I'm not sure about why it's added, and it doesn't seem to be needed here. Ask just in case.

[tabVersion]

the field is introduced in #13707 to deal with DEFAULT_KEY_COLUMN_NAME change in future.
discussions are available here #13707 (comment)

[fuyufjh]

Materialize parse the value to UTF-8 string by default, unless specifying the data type BYTES

Ref. https://materialize.com/docs/sql/create-source/kafka/#syntax

I think we should follow the design, because in most cases the value will just be used as varchar instead of bytea, and converting bytea to varchar is verbose.

My idea:

INCLUDE HEADER key                  -- decode as UTF-8 string
INCLUDE HEADER key AS name          -- decode as UTF-8 string
INCLUDE HEADER key AS name VARCHAR  -- decode as UTF-8 string
INCLUDE HEADER key AS name BYTEA    -- output raw bytes

[fuyufjh]

The breaking change looks acceptable to me, although it seems not hard to make it backward compatible.

[tabVersion]

Materialize parse the value to UTF-8 string by default, unless specifying the data type BYTES

Ref. [materialize.com/docs/sql/create-source/kafka/#syntax](https://materialize.com/docs/sql/create-source/kafka/#syntax)

I think we should follow the design, because in most cases the value will just be used as varchar instead of bytea, and converting bytea to varchar is verbose.

My idea:

INCLUDE HEADER key                  -- decode as UTF-8 string
INCLUDE HEADER key AS name          -- decode as UTF-8 string
INCLUDE HEADER key AS name VARCHAR  -- decode as UTF-8 string
INCLUDE HEADER key AS name BYTEA    -- output raw bytes

The solution seems a little verbose to me, the original purpose for this new syntax is to handle the problem that users can have trouble finding the key they want in Array[Struct<varchar, bytea>]. And the solution works well solving it.
Users have no obstacle converting varchar from bytes in sql and the efficiency is good. I'd keep this external because it serves as a shortcut in parser, not sharing the logic that fills NULL when parsing failure. We won't want to be responsible for some unexpected behavior caused by users forgetting to specify varchar in the clause.

[fuyufjh]

The solution seems a little verbose to me, the original purpose for this new syntax is to handle the problem that users can have trouble finding the key they want in Array[Struct<varchar, bytea>]. And the solution works well solving it. Users have no obstacle converting varchar from bytes in sql and the efficiency is good. I'd keep this external because it serves as a shortcut in parser, not sharing the logic that fills NULL when parsing failure. We won't want to be responsible for some unexpected behavior caused by users forgetting to specify varchar in the clause.

As a connector, I would hope to complete every parsing work inside of it. I consider parsing a header value to string as part of this.

I can forecast that users will ask you 2 questions frequently without such an option.

1. How to convert bytea to varchar with PG functions? -- Please use encode()
2. How to create a table with a header value in varchar type? -- Please use generated columns.

Both are natural requirements but the solution is obscure, so you can't blame the users actually.

[tabVersion]

The solution seems a little verbose to me, the original purpose for this new syntax is to handle the problem that users can have trouble finding the key they want in Array[Struct<varchar, bytea>]. And the solution works well solving it. Users have no obstacle converting varchar from bytes in sql and the efficiency is good. I'd keep this external because it serves as a shortcut in parser, not sharing the logic that fills NULL when parsing failure. We won't want to be responsible for some unexpected behavior caused by users forgetting to specify varchar in the clause.

As a connector, I would hope to complete every parsing work inside of it. I consider parsing a header value to string as part of this.

I can forecast that users will ask you 2 questions frequently without such an option.

1. How to convert bytea to varchar with PG functions? -- Please use encode()
2. How to create a table with a header value in varchar type? -- Please use generated columns.

Both are natural requirements but the solution is obscure, so you can't blame the users actually.

Already implemented. Please review.

[fuyufjh]

LGTM!