[Enhancement] Expose Kafka/Pulsar message metadata via an INCLUDE METADATA clause in Routine Load (backport #73840)

[mergify]

Why I'm doing:

Routine Load consumers already receive each Kafka/Pulsar message's metadata — topic, partition,
offset, timestamp, key, headers — but there was no way to land any of it in table columns; only the
message payload could be loaded. Users who need provenance (which topic/partition/offset a row came
from), event-time processing, or routing on a header value had no access to it.

What I'm doing:

Add an INCLUDE METADATA (...) clause to CREATE/ALTER ROUTINE LOAD that binds per-message source
metadata to source columns:

CREATE ROUTINE LOAD db.job ON tbl
INCLUDE METADATA ( KEY AS msg_key, PARTITION AS msg_part, TIMESTAMP_MS AS msg_ts, HEADERS AS msg_hdr ),
COLUMNS ( id, name,
          event_time = from_unixtime(msg_ts / 1000),
          trace_id   = element_at(msg_hdr, 'trace-id') )
FROM KAFKA ( ... );

Each <KEY> AS <alias> binds a metadata key to a hidden source column named <alias> that COLUMNS
expressions reference; the BE scanner fills it from the message by slot id, so a payload field with
the same name is never shadowed. The valid keys are source-dependent and inferred from the FROM
type:

• Kafka: TOPIC, KEY, PARTITION, OFFSET, TIMESTAMP_MS, HEADERS.
• Pulsar: TOPIC, KEY, PARTITION, MESSAGE_ID, PUBLISH_TIME_MS, EVENT_TIME_MS, PROPERTIES.

HEADERS/PROPERTIES is a MAP<VARCHAR, VARCHAR> (duplicate keys last-wins; raw bytes, no UTF-8
validation); a single value is read with element_at(alias, 'name'). Available for format = json
(Kafka and Pulsar) and format = avro (Kafka only — Pulsar maps non-JSON to CSV); not supported for
CSV, because the consumer accumulates several messages into one pipe buffer (the JSON/Avro path uses
one buffer per message), so a CSV row cannot be tied back to its own message's metadata. An
unsupported key for the source raises an explicit error that lists the source's supported keys.
Validated at both CREATE and ALTER.

Unit tests (FE clause validation/binding + persistence round-trip, BE metadata fill + Pulsar topic
parsing) and user documentation (en/zh/ja) added.

Fixes #73841

Compatibility & upgrade:

This change is additive; existing jobs are unaffected and no migration is required.

• Wire protocol: new optional thrift types/fields — TStreamSourceMetaKind,
  TRoutineLoadMetaColumn, TBrokerScanRangeParams.stream_source_meta_columns (field 35), and
  need_source_metadata / need_message_key / need_message_headers on TKafkaLoadInfo (8/6/7) and
  TPulsarLoadInfo (9/7/8). Nothing is required, so mixed-version FE and BE interoperate during a
  rolling upgrade.
• Existing jobs: a job with no INCLUDE METADATA clause takes the existing path — no hidden
  column, need_source_metadata unset, the consumer attaches only the partition/offset it always did
  (feeding malformed-record error logs), and the scanner reads every field from the payload. Output is
  unchanged; no job needs to be recreated.
• Persistence / FE replay: the clause is part of the CREATE ROUTINE LOAD SQL persisted in the
  job's origin statement, so SHOW CREATE ROUTINE LOAD renders it and FE replay rebuilds the job from
  that SQL.
• Upgrade order: standard rolling upgrade. A BE that does not read stream_source_meta_columns
  leaves those columns empty rather than failing, so upgrade BE before creating metadata jobs.
  Downgrade is safe: a downgraded BE ignores the optional fields and the persisted job SQL stays intact.

What type of PR is this:

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Does this PR entail a change in behavior?

• Yes, this PR will result in a change in behavior.
• No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

• Interface/UI changes: syntax, type conversion, expression evaluation, display information
• Parameter changes: default values, similar parameters but with different default values
• Policy changes: use new policy to replace old one, functionality automatically enabled
• Feature removed
• Miscellaneous: upgrade & downgrade compatibility, etc.

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr needs user documentation (for new or modified features or behaviors)
  • I have added documentation for my new feature or new function
  • This pr needs auto generate documentation
• This is a backport pr

Bugfix cherry-pick branch check:

• I have checked the version labels which the pr will be auto-backported to the target branch
  • 4.1
  • 4.0
  • 3.5

    ---

    This is an automatic backport of pull request [Enhancement] Expose Kafka/Pulsar message metadata via an INCLUDE METADATA clause in Routine Load #73840 done by Mergify.

[mergify]

Cherry-pick of 0796ea6 has failed:

On branch mergify/bp/branch-4.1/pr-73840
Your branch is up to date with 'origin/branch-4.1'.

You are currently cherry-picking commit 0796ea6077.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	new file:   be/src/exec/file_scanner/stream_source_meta.cpp
	new file:   be/src/exec/file_scanner/stream_source_meta.h
	modified:   be/src/runtime/routine_load/kafka_consumer_pipe.h
	modified:   be/src/runtime/stream_load/stream_load_context.h
	modified:   be/src/util/byte_buffer.cpp
	modified:   be/src/util/byte_buffer.h
	modified:   be/test/exec/file_scanner/avro_scanner_test.cpp
	new file:   be/test/exec/file_scanner/stream_source_meta_test.cpp
	modified:   be/test/runtime/kafka_consumer_pipe_test.cpp
	modified:   be/test/runtime/stream_load_pipe_test.cpp
	modified:   be/test/util/byte_buffer_test.cpp
	modified:   docs/en/loading/RoutineLoad.md
	modified:   docs/ja/loading/RoutineLoad.md
	modified:   docs/zh/loading/RoutineLoad.md
	modified:   fe/fe-core/src/main/java/com/starrocks/load/RoutineLoadDesc.java
	modified:   fe/fe-core/src/main/java/com/starrocks/load/routineload/KafkaTaskInfo.java
	modified:   fe/fe-core/src/main/java/com/starrocks/load/routineload/PulsarTaskInfo.java
	modified:   fe/fe-core/src/main/java/com/starrocks/load/routineload/RoutineLoadJob.java
	new file:   fe/fe-core/src/main/java/com/starrocks/load/routineload/RoutineLoadMetadata.java
	modified:   fe/fe-core/src/main/java/com/starrocks/load/streamload/StreamLoadInfo.java
	modified:   fe/fe-core/src/main/java/com/starrocks/planner/StreamLoadScanNode.java
	modified:   fe/fe-core/src/main/java/com/starrocks/qe/ShowExecutor.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/analyzer/CreateRoutineLoadAnalyzer.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/ast/CreateRoutineLoadStmt.java
	new file:   fe/fe-core/src/main/java/com/starrocks/sql/ast/ImportMetadataStmt.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java
	modified:   fe/fe-core/src/test/java/com/starrocks/load/RoutineLoadDescTest.java
	new file:   fe/fe-core/src/test/java/com/starrocks/load/StreamMetaFunctionTest.java
	modified:   fe/fe-core/src/test/java/com/starrocks/load/routineload/RoutineLoadJobEditLogTest.java
	modified:   fe/fe-grammar/src/main/antlr/com/starrocks/grammar/StarRocksLex.g4
	modified:   gensrc/thrift/BackendService.thrift
	modified:   gensrc/thrift/PlanNodes.thrift

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   be/src/exec/CMakeLists.txt
	both modified:   be/src/exec/file_scanner/avro_scanner.cpp
	both modified:   be/src/exec/file_scanner/avro_scanner.h
	both modified:   be/src/exec/file_scanner/json_scanner.cpp
	both modified:   be/src/exec/file_scanner/json_scanner.h
	both modified:   be/src/runtime/routine_load/data_consumer_group.cpp
	both modified:   be/src/runtime/routine_load/data_consumer_group.h
	both modified:   be/test/CMakeLists.txt
	both modified:   be/test/exec/file_scanner/json_scanner_test.cpp
	both modified:   be/test/runtime/routine_load/data_consumer_test.cpp
	both modified:   fe/fe-core/src/main/java/com/starrocks/load/Load.java
	both modified:   fe/fe-core/src/test/java/com/starrocks/load/LoadTest.java
	both modified:   fe/fe-core/src/test/java/com/starrocks/planner/StreamLoadScanNodeTest.java
	both modified:   fe/fe-grammar/src/main/antlr/com/starrocks/grammar/StarRocks.g4

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

[mergify]

@mergify[bot]: Backport conflict, please reslove the conflict and resubmit the pr