I stumbled into this while pilfering code from kafka-delta-ingest for another
project and discovered that the code in `write_values` which does
`record_batch.schema() != arrow_schema` doesn't do what we think it does.

Basically if `Decoder` "works" the schema it's going to return is just the
schema passed into it. It has no bearing on whether the JSON has the same
schema. Don't ask me why.

Using the reader's `infer_json_schema_*` functions can provide a Schema that is
useful for comparison:

        let mut value_iter = json_buffer.iter().map(|j| Ok(j.to_owned()));
        let json_schema = infer_json_schema_from_iterator(value_iter.clone()).expect("Failed to infer!");
        let decoder = Decoder::new(Arc::new(json_schema), options);
        if let Some(batch) = decoder.next_batch(&mut value_iter).expect("Failed to create RecordBatch") {
            assert_eq!(batch.schema(), arrow_schema_ref, "Schemas don't match!");
        }

What's even more interesting, is that after a certain number of fields are
removed, the Decoder no longer pretends it can Decode the JSON. I am baffled as
to why.

…l refactor

The intention here is to enable more consistent schema handling within
the writers

Sponsored-by: Raft LLC

…e writers

The DeserializedMessage carries optional inferred schema information
along with the message itself. This is useful for understanding whether
schema evolution hould happen "later" in the message processing
pipeline.

The downside of this behavior is that there will be performance impact
as arrow_json does schema inference.

Sponsored-by: Raft LLC

Turning avro off drops about 50 crates from the default build, so useful
for development, but the code would need to be cleaned up to remove this
from the default features list

See #163

Identified by `cargo +nightly udeps`

This change is a little wrapped up in the introduction of
DeserializedMessage but the trade-off for development targeting S3 is
that I am linking 382 crates every cycle as opposed to 451.

Fixes #163

I don't know why the impl was way down there 😄

This commit introduces some interplay between the IngestProcessor and
DataWriter, the latter of which needs to keep track of whether or not it
has a changed schema.

What should be done with that changed schema must necessarily live in
IngestProcessor since that will perform the Delta transaction commits at
the tail end of batch processing.

There is some potential mismatches between the schema in storage and
what the DataWriter has, so this change tries to run the runloop again
if the current schema and the evolved schema are incompatible

Closes #131

Sponsored-by: Raft LLC

This will ensure the non-evolution case stands relatively speedy!

Sponsored-by: Raft LLC

…om-outside-in