feat: Pass starting replication value from Tap to Target
kind/Featurevaluestream/SDK
### Feature scope
Taps (catalog, state, tests, etc.)
### Description
I would like Targets to have access to the starting_replication_value that is set in a Stream when incremental. The primary use case is to be able to have the target perform different functions depending on that value. I believe this to be a relatively common flow used in the world of extract-load processes so it would be beneficial to have as a core feature.
One example: a SQL based Target being able to see that the starting_replication_value is None (i.e. first time run or full-refresh) and truncating the target table. The existing solutions of `hard_delete` does not meet this need because:
1. If messy data was synced and the source was cleaned up, `hard_delete` would not cleanout those dirty records in the target; and
2. single record deletions are prohibitively expensive in OLAP databases.
## Potential solution
Update this method to write a state message immediately after writing the starting replication value:
https://github.com/meltano/sdk/blob/bad876d539da483ada658685878c2b94e003d54d/singer_sdk/streams/_state.py#L121-L169
```
def write_starting_replication_value(
...
write_starting_replication_value(state, value)
self._is_state_flushed = False
self._write_state_message()
```
The target `_process_state_message()` could then read it in by parsing it out of the STATE message:
```
{
"type": "STATE",
"value": {
"bookmarks": {
"EXAMPLE-STREAM": {
"replication_key": "last_update_date",
"replication_key_value": "2024-09-26T09:04:54 AD",
"starting_replication_value": "2024-09-26T09:04:54 AD"
}
}
}
}
```
EDIT: Whoops, looks like I submitted this from my company Github account and not my private one @iamlorax ! We are one and the same 😄
0 条评论