dekaf: Implement DeletionMode to allow representing deletions as a Kafka header instead of a tombstone
#1738
+158
−57
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
jshearer
force-pushed
the
jshearer/dekaf_deletions_in_header
branch
7 times, most recently
from
1ab8273
to
cb3fd03
Compare
jshearer
force-pushed
the
jshearer/dekaf_deletions_in_header
branch
from
e49688a
to
cb3fd03
Compare
Merged
…Kafka header instead of a tombstone
jshearer
force-pushed
the
jshearer/dekaf_deletions_in_header
branch
2 times, most recently
from
972aecd
to
b391c01
Compare
…_deleted` for deletions Also needed to add this field to the generated Avro schemas
jshearer
force-pushed
the
jshearer/dekaf_deletions_in_header
branch
from
b391c01
to
e337f39
Compare
crates/dekaf/src/read.rs
Outdated
| if matches!(self.deletes, DeletionMode::CDC) { | ||
| let mut heap_node = HeapNode::from_node(root.get(), &self.alloc); | ||
| let foo = DELETION_INDICATOR_PTR | ||
| .create_heap_node(&mut heap_node, &self.alloc) |
There was a problem hiding this comment.
there's a memory leak here -- you're never resetting the bump allocator.
I would recommend not having it as a field of self and instead creating one on the stack. Use it, and then call reset() on it immediately after each avro::encode so that it's only retaining a single HeapNode.
There was a problem hiding this comment.
I was going to ask you to take a look at this as I wasn't sure I was doing it correctly. That makes sense to me, thanks 👍
jshearer
commented
Oct 31, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This adds support for customizable deletion handling:
DeleteMode::Default, which is what you get if you don't specify a deletion mode, operates the same was as before where deletes are mapped into Kafka records with a key, and null value.DeleteMode::Headerchanges this behavior to include an_is_deletedheader on every returned Record. Non-deletes get0, deletes get1. In addition, the full deletion tombstone stored in the Flow collection is returned for the Record's value instead of null.I tested this out in Tinybird and it looks like this works as expected: setting a username of
{"deletions": "header"}causes Flow deletion tombstones (op: d) to cause the corresponding row to get deleted in Tinybird.Question: Should we just go all the way and name this
DeleteMode::Tinybird? Or does someone have a better suggestion thanHeader?Also included is a small change to
Readto make data preview UIs work reliably again. I realized they stopped working consistently after merging #1733, and this fixes that.This change is