feat: upsert mode in Kafka(source) - #1161
Conversation
GitHub Actions
Failure
GitHub Actions
Inactive
GitHub Actions
Inactive
GitHub Actions
Inactive
GitHub Actions
Inactive
GitHub Actions
Failure
GitHub Actions
Inactive
GitHub Actions
Inactive
GitHub Actions
Inactive
GitHub Actions
Inactive
There was a problem hiding this comment.
We should migrate Kafka types to the drivers/kafka having it in common code is not correct.
Please migrate it.
There was a problem hiding this comment.
should we also have to migrate pkg/kafka into drivers/kafka
| ChunkColumn string `json:"chunk_column,omitempty"` | ||
| PartitionRegex string `json:"partition_regex"` | ||
| StreamName string `json:"stream_name"` | ||
| AppendMode bool `json:"append_mode,omitempty"` |
There was a problem hiding this comment.
We should think of adding some enum here like Mode and have the upsert follow the enum naming
| FilterConfig *FilterConfig `json:"filter_config,omitempty"` | ||
| SelectedColumns *SelectedColumns `json:"selected_columns"` | ||
|
|
||
| DedupKeys []string `json:"dedup_keys,omitempty"` |
There was a problem hiding this comment.
We need to Validate presence of at least dedup key in case of upsert mode.
Also do you think dedup_columns might make more sense given all other configs use column by convention?
There was a problem hiding this comment.
validation in upsert.go
PRD mentions dedup_keys
| return false, err | ||
| } | ||
| } | ||
| } else { |
There was a problem hiding this comment.
Make a separate function that will resolve the olake_id from kafkaesque record, that will help us to write unit tests clearly for all cases.
2 participants
Description
Added source side Kafka Upsert support.
When
append_modeisfalseanddedup_keysare set on streams_olake_idderived from those keysdedup_keys = ["_kafka_key"]only): Upsert + Kafka tombstones deletesdedup_keys = selected fields): Upsert onlyType of change
How Has This Been Tested?
Documentation