Function parse_source_export_details 

fn parse_source_export_details<'a>(a: &'a str) -> Result<Jsonb, EvalError>

Extracts source-export (source table) metadata from a catalog create_sql.

Returns, for a CREATE TABLE ... FROM SOURCE or a non-progress CREATE SUBSOURCE ... OF SOURCE ... statement:

{
  "source_id": "<parent source item id>",
  "external_reference": ["part1", "part2", ...],
  "envelope_type": <text | null>,
  "key_format": <text | null>,
  "value_format": <text | null>
}

envelope_type, key_format, and value_format are always null for a CREATE SUBSOURCE (the postgres/mysql/sql-server exports that use the old subsource syntax carry neither format nor envelope). They may also be null for a CREATE TABLE ... FROM SOURCE that omits FORMAT/ENVELOPE.

Returns jsonb null for progress subsources and for any statement that is not a source export. The caller distinguishes the four source-table views by joining source_id against mz_sources and filtering on the parent’s type, so this helper stays connection-type agnostic.

Errors if the statement fails to parse, references an unresolved item name, or is a non-progress subsource missing its OF SOURCE or EXTERNAL REFERENCE.

The key_format/value_format derivation mirrors the runtime DataSourceDesc::formats() that the removed pack_kafka_source_tables_update packer read. A bare FORMAT only carries a key when it resolves to an encoding that has one, which among bare formats is only Avro or Protobuf read from a Confluent Schema Registry whose purified seed carries a key schema. A KEY FORMAT … VALUE FORMAT … spec always carries both.