Introduce schema evolution via the -S flag #164

rtyler · 2024-01-07T20:53:37Z

This set of changes implements more nuanced handling of Delta vs. message schema with the introduction of the -S command line flag which enables schema evolutions.

In order for schema evolution to work, there is a necessary performance hit. kafka-delta-ingest must determine the schema of every message that is read from Kafka, infer its schema, and if necessary add nullable columns to the Delta table. This is related to similar work in delta-rs for the RecordBatchWriter but deviates because of the mechanism by which RecordBatches and schema is handled in kafka-delta-ingest.

Sponsored-by: Raft LLC

NOTE: This pull request builds on #162

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

mightyshazam

The transform usage won't work as implemented without a change, but everything else is nice to have.

mightyshazam · 2024-07-08T20:02:32Z

src/dead_letters.rs

                        self.transformer
-                            .transform(&mut v, None as Option<&BorrowedMessage>)?;
+                            // TODO: this can't be right, shouldn't this function takje DeserializedMessage
+                            .transform(&mut v.clone().into(), None as Option<&BorrowedMessage>)?;


I don't think this does what is intended. The transform function takes a mutable reference to the clone of v instead of a mutable reference to v itself. The function modifies and consumes the clone, but it only exists in the scope of the input to transform, so it should be dropped once the method finishes. That means Ok(v) is returning the original instead of the transformed message. Implementing Into<Value> for DeserializedMessage and changing the code to

let mut dm = DeserializedMessage::from(v); self.transformer // TODO: this can't be right, shouldn't this function takje DeserializedMessage .transform(&mut dm, None as Option<&BorrowedMessage>)?; Ok(dm.into())

should do the trick. Also the Into implementation will allow for the removal of all of instances of to_owned() that my last commit added.

The transform code might be some bungled merge conflict resolution, I'll be able to look more into this shortly

mightyshazam · 2024-07-08T20:06:08Z

src/transforms.rs

@@ -348,13 +349,13 @@ impl Transformer {
    /// The optional `kafka_message` must be provided to include well known Kafka properties in the value.
    pub(crate) fn transform<M>(
        &self,
-        value: &mut Value,
+        value: &mut DeserializedMessage,
        kafka_message: Option<&M>,
    ) -> Result<(), TransformError>


Another option, with respect to my earlier comment about changing DeserializedMessage, is to have transform return Result<DeserializedMessage, TransformError. Then it can take value: DeserializedMessage. That will allow us to mutate and return what we mutated, but that is a bigger code change.

src/transforms.rs

mightyshazam · 2024-07-08T20:07:23Z

src/coercions.rs

@@ -442,7 +447,7 @@ mod tests {
        ];

        for i in 0..messages.len() {
-            assert_eq!(messages[i], expected[i]);
+            assert_eq!(messages[i].message().to_owned(), expected[i]);


.into() works here if implemented, saving us a clone

mightyshazam · 2024-07-08T20:08:30Z

src/writer.rs

+            }
+        }
+
+        let values = values.into_iter().map(|v| v.message().to_owned()).collect();


.into() works here if implemented, saving us a clone

rtyler force-pushed the schema-evolution-from-outside-in branch 3 times, most recently from e4a1ade to c1bcd37 Compare January 8, 2024 05:51

rtyler changed the title ~~Implementing stricter schema conformance and evolution~~ Introduce schema evolution via the -S flag Jan 8, 2024

rtyler force-pushed the schema-evolution-from-outside-in branch from c1bcd37 to 820421f Compare January 8, 2024 22:53

rtyler added 11 commits January 9, 2024 10:02

Begin introducing schema conformance testing ahead of more substantia…

93e3d73

…l refactor The intention here is to enable more consistent schema handling within the writers Sponsored-by: Raft LLC

Put avro dependenciies behind the avro feature flag

f8eb561

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

Remove unused dependencies

4ed2c8a

Identified by `cargo +nightly udeps`

Remove avro from the default features and feature gate its cpde

72aff33

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

Relocate DataWriter in the writer.rs file for easier readability

071a6a2

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

Bring a -S option for schema evolution and bring that setting through

baf9696

Ensure all features are building and running during GitHub actions

72c2f0b

Disable scheam inference when schema evolution is disabled

50e81da

This will ensure the non-evolution case stands relatively speedy! Sponsored-by: Raft LLC

rtyler mentioned this pull request Jan 9, 2024

Put avro support behind an optional feature flag #163

Open

rtyler added this to the kafka-delta-ingest 0.2 release milestone Jan 9, 2024

rtyler force-pushed the schema-evolution-from-outside-in branch from 820421f to 50e81da Compare January 9, 2024 19:13

Merge remote-tracking branch 'upstream/main' into schema-evolution-fr…

1983bb0

…om-outside-in

rtyler marked this pull request as ready for review July 8, 2024 17:37

rtyler enabled auto-merge July 8, 2024 17:38

rtyler requested a review from mightyshazam July 8, 2024 17:38

mightyshazam and others added 2 commits July 8, 2024 15:47

removed some cloning of deserialized message

144513a

Merge branch 'main' into schema-evolution-from-outside-in

706163c

mightyshazam requested changes Jul 8, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce schema evolution via the -S flag #164

Introduce schema evolution via the -S flag #164

rtyler commented Jan 7, 2024 •

edited

Loading

mightyshazam left a comment

mightyshazam Jul 8, 2024

rtyler Jul 8, 2024

mightyshazam Jul 8, 2024

mightyshazam Jul 8, 2024

mightyshazam Jul 8, 2024

Introduce schema evolution via the -S flag #164

Are you sure you want to change the base?

Introduce schema evolution via the -S flag #164

Conversation

rtyler commented Jan 7, 2024 • edited Loading

mightyshazam left a comment

Choose a reason for hiding this comment

mightyshazam Jul 8, 2024

Choose a reason for hiding this comment

rtyler Jul 8, 2024

Choose a reason for hiding this comment

mightyshazam Jul 8, 2024

Choose a reason for hiding this comment

mightyshazam Jul 8, 2024

Choose a reason for hiding this comment

mightyshazam Jul 8, 2024

Choose a reason for hiding this comment

rtyler commented Jan 7, 2024 •

edited

Loading