implement destination sinks

[rudolfix]

Background
Sinks are the reverse of the source. A Python function decorated with @dlt.sink should generate destination factory (#746 ) that may be passed to pipeline in destination argument. Sink will consume data that is sent to it during the load stage.
Sink will be fed data in configurable chunks and will keep state of what got consumed already so it will be "transactional" when size of the chunk == 1.
Sink state is separate from source state but initially will be handled together with it - and preserved in file. For "transactional" sinks there will be tons of state updates so general mechanism we use now (store state at destination) does not make sense. We may support bucket storage for load.py and load storage for fully resumable, transactional sinks.

Tasks
This is rather a big ticket, tasks below are just outline:

1. • implement sink destination (like any other) with a proper destination factory
2. • implement @dlt.sink decorator working similarly to @dlt.transformer. It returns configured destination factory
3. • sink decorator should accept a sink function, where first argument is TDataItems, second a table schema (and maybe something else), and optional parameters (same as parametrized transformer)
4. • sink decorator should allow to specify batch size, file format (json, parquet). batch_size=1 sends single rows, >1 sends lists of objects (or arrow tables if parquet), batch_size=0 sends the FileDict as data item so user can handle file processing as s/he wants
5. • sink should accept naming convention to be used (a lot of people will use direct)
6. • we should decorate function properly. we should allow for additional parameters which should become a SPEC of this destination and is used as such (ideally we would derive this spec from DestinationClientConfiguration (trivial change in with_config may be needed)
7. • expose current load package context via dlt.current.load_package which should at least contain load id and state dict (I think we have it)
8. • (optional) allow for 3 callbacks in decorator: on_init (called o schema update), on_completed (called on completed, mind the abort flag), on_garbage_collected (called when factory is garbage collected, that allows for some kind of cleanup)

Implementation
Basic implementation may be quite simple. Sink destination implements a copy job that opens a job file and then streams it into a sink function in chunks that were configured. After each chunk it emits sink state.

On restart the state must be retrieved and processed rows must be skipped.

State handling could be part of LoadStorage and we may make the LoadStorage comparible with fsspec (maybe in separate tickets - it depends a lot on "file move" to be fast and atomic)

[rudolfix]

optionally sinks could be chained after main destination consumes data and not just replace it. there's for sure a clever way of doing that without complicating user interface

[sh-rp]

Questions:

• Should we support awaitable sink functions?
• Should we add a flag to guarantee sink function execution on the main thread for each item, probably using multiprocessing queues?
• I don't get what the dltsource interaction should be, we should talk about it :)
• where should we store the pipeline and the schemas plus the execution state of the current loadpackage (if at all?)

Other ideas:

• Should we sent sentinel values as an extra function parameter for special occassions? For example at the start of the load and the end of the load (or start and end of a file)? This would be useful for creating and destroying database or queue connections.
• Should we allow the user to return certain sentinel values to indicate that this batch should be retried or the load should fail or something like that?
• Should we provide a context object (simple dictionary) that can carry over state between sink function calls? Would be useful to store open database or queue connections. It could also be one context object per thread for example, so we can enable multithreaded processing.

[sh-rp]

@rudolfix what do you mean with "filedict"? In my current pr i am simply passing the full filepath as a string into the sink function.