Ingest decode tail: CSV/JSON file sources and the remaining CDC follow-ups

[jordepic]

What

The remaining tail of the native decode-to-Arrow ingest work. The core shipped: file sources (Parquet + ORC) read through DataFusion's file scan, and the streaming decode operator decodes Kafka bytes → Arrow for JSON, Confluent/bare Avro, CSV, protobuf, and all four CDC JSON envelopes, with the CSV/JSON decode envelopes matched to Flink's converters exactly (divergences/21).

Remaining items

• CSV/JSON file sources — the lower-priority file formats, in the existing native file-source mold (path + projection in, Arrow out). Text formats need the schema supplied and their decode semantics parity-gated against Flink's options; the native Flink-exact CSV/JSON decode kernels already exist from the streaming path. (Avro OCF was dropped: arrow-avro can't read Flink's top-level-union writer schema.)
• Nested Maxwell/Canal CDC schemas — the UPDATE_BEFORE key-presence scan currently routes flat scalar schemas only; nested needs path-aware presence (Flink's findValue recursive search).
• Canal database.include/table.include regex filters (currently gated to fallback).
• schema-include ({schema, payload} wrapper) and CDC metadata columns (ts_ms, source.*) — decode the wrapper / project metadata, then drop those fallback gates.
• debezium-avro-confluent — same envelope shape as debezium-json, Avro decoder inside (arrow-avro against a ROW<before, after, op> reader schema).
• Upsert/PK CDC tables — a ChangelogNormalize lands above the scan (the native normalizer exists); verify/handle the wrapper.
• Decode-path watermarks — the decode operator declines watermarked tables (it can't see split lifecycle, so it can't regenerate a pushed-down watermark the way the native Kafka source does). Matters only if CSV/raw/CDC event-time tables show up; the options are split-lifecycle visibility inside the operator or folding the decode into a source reader.

Each item lifts a specific fallback gate in docs/coverage-and-fallbacks.md; update that doc in the same commit as any of these.