Lookup joins

[mwylde]

This PR introduces a new type of table, connector, and join, to support use cases where we wish to enrich a stream by performing ad-hoc queries on another system (typically a DB or cache). We call these lookup joins, as we are joining a stream to another table via a query mechanism, as well as an initial Redis lookup connector.

It looks like this:

CREATE TABLE events (
    event_id TEXT,
    timestamp TIMESTAMP,
    customer_id TEXT,
    event_type TEXT
) WITH (
    connector = 'kafka',
    topic = 'events',
    type = 'source',
    format = 'json',
    bootstrap_servers = 'broker:9092'
);

CREATE TEMPORARY TABLE customers (
    customer_id TEXT GENERATED ALWAYS AS (metadata('key')) STORED, 
    name TEXT,
    plan TEXT
) with (
    connector = 'redis',
    format = 'raw_string',
    address = 'redis://localhost:6379',
    format = 'json',
    'lookup.cache.max_bytes' = '1000000'
    'lookup.cache.ttl' = '5 second'
);

SELECT  e.event_id,  e.timestamp,  e.customer_id,  e.event_type, c.customer_name, c.plan
FROM  events e
LEFT JOIN customers c
ON concat('customer.', e.customer_id) = c.customer_id
WHERE c.plan = 'Premium';

Here, we create a "temporary table"—implying that it's not fully materialized—backed by a Redis cache, with the Redis key specified via the metadata syntax introduced in 0.13. We then perform a join (in this case a left join, although inner is also supported) against it, causing a lookup into Redis for the specified key.

Lookup joins also optionally include a cache, which can be configured by the user with a TTL and/or a max size in bytes.

This PR also includes a refactor of the deserialization system to reduce the number of independent codepaths that handle metadata field deserialization and combines logic between JSON and non-JSON paths. This also fixed several corner cases where additional fields were not being injected properly in certain formats.

This addresses #820