proposals for the connector specification

Hi team Airbyte!

**tl;dr**: We (team [Estuary](https://github.com/estuary/flow)) want to work with you to evolve the Airbyte spec into a community standard for building data connectors.

Things we especially like about your improvements to the Singer spec are the schema-driven configuration discovery workflow and the use of Docker for packaging and discovery of connectors (rather than PyPi).

The protocol is adaptable to continuous / low-latency streaming contexts, which is a particular focus for Flow.
We've worked to integrate the "source" protocol into Flow, and have also worked up some [OSS source connectors](github.com/estuary/connectors) for various technical systems (S3 & GCS with pattern matching, incremental bucket scans, CSV, and compression handling ; Kinesis ; Kafka soon).

In the process we've implemented and experimented with several extensions to the spec we wanted to talk through and propose for inclusion. I've bundled these in this Github issue, but they're largely separable and can be discussed/evolved/adopted independently.

## Proposals

1. **Allow a connector to produce multiple states over its lifetime.**

A connector invocation is permitted to produce many STATE messages over its lifetime.
Each STATE reflects a checkpoint that reflects the RECORDS which preceded it.

2. **Add optional `tail` to ConfiguredCatalog**

`tail` is an boolean which tells the connector that it may run indefinitely, producing new RECORDs and STATEs as new data becomes available.

If false, the connector should produce available RECORDS and then exit, as is the current behavior.
`tail` is optional (assumed to be `false` if missing), and connectors are not required to do anything with it.

A Kafka connector might use `tail` to determine whether it should consume indefinitely, producing new RECORDs as they arrive, or should read through offsets determined at startup and then exit.

3. **Output STATEs are reduced using [RFC 7368 Merge PATCH](https://datatracker.ietf.org/doc/html/rfc7386).**

As currently specified, each STATE produced across connector invocations is a complete replacement of a prior STATE.

Consider an AWS Kinesis source which is capturing from many underlying Kinesis shards, where the connector STATE encapsulates separate read offsets for each shard.

Today, when any one shard has data, the connector emits its RECORDs and then must re-state the STATE of _all_ shards. Under this proposal, it could instead write a STATE that reflects just the new shard data. The client of the connector then knows to PATCH in that delta STATE in order to update the complete STATE.

One option is to say that _all_ STATEs are PATCH merges, which changes the assumed behavior of a bunch of existing connectors. This may be backwards-compatible in practice, given how JSON PATCH works and the stable of existing Airbyte connectors, but we're not certain of that.

Another option might be to add a `patch` bool property to the STATE message that toggles this behavior.

4. **Add optional `range` to ConfiguredCatalog**

`range` is a pair of {begin, end} inclusive uint32 integers which tell each connector invocation what it's responsible for. `range` facilitates connector parallelism: each invocation can consult its `range` to determine if it should do a unit of work, or that one if its peers will.

The exact usage of `range` is up to the connector.

- A GCS / S3 connector might hash each file name into a uint32.
- A Kafka connector might hash each partition.
- Our Kinesis connector determines the overlap between the Kinesis shard's range and that of the connector.

Examples of ranges:

- `{"begin":"00000000","end":"ffffffff"}` : connector has only one invocation, covering the full range
- `{"begin":"00000000","end":"7fffffff"}`, `{"begin":"800000","end":"ffffffff"}` : connector has two invocations, each with 1/2 of the range.

If not specified, `range` is assumed to be `{"begin":"00000000","end":"ffffffff"}`.

A particular rationale for `range` (over, say, worker count and index) is that it allows for dynamic scale-out without breaking existing STATEs. A `range` and its prior STATE can be subdivided into new ranges, each starting from the common parent STATE.

5. **Add optional `projections` to ConfiguredStream**

Projections is a map of {field name: [JSON-pointer](https://datatracker.ietf.org/doc/html/rfc6901)}.

Many sources of data are tabular in nature (SQL tables, CSVs), while connectors work with JSON documents.
An reasonable projection used to map between these models is map table columns into properties of a root document object.
But it's not the only one!

`projections` instruct the connector of an alternative location where a given field should be inserted into a RECORD document.

To give a grounded use case, Citi Bike provides (somewhat messy CSV) of their [system data](https://s3.amazonaws.com/tripdata/index.html).
Our S3 connector is able to use [projections](https://github.com/estuary/flow/blob/master/examples/citi-bike/rides.flow.yaml) to map CSV columns into more natural document locations in the RECORDs it produces.

6. **Adopting community-oriented naming for the protocol**

It will be challenging to build multi-company momentum behind a specification named for one company.

While pretty blah, would you be open to something like the "Open Connector Specification" (or have better ideas?).