## What
Removing redundant references and duplicate information passed around using `WriteConfig` objects. No functional changes and resurrected all the information needed through `StreamConfig` and adapted changes accordingly.
This PR should be in a mergeable state with no functional changes after the ones down the stack are published.
## Review guide
* Removed references of `BigQueryWriteConfig` and reused already built `StreamConfig`
* Removing unnecessary `StagingOperations` interface and made concrete class, this will help for later adding a shim on this and refactoring without large changes
* Removed other unnecessary references of getting dynamic schema, `WriteDispostion` etc. Probably remnant of bigquery-denormalized bespoke connector.
## User Impact
<!--
* What is the end result perceived by the user?
* If there are negative side effects, please list them.
-->
## Can this PR be safely reverted and rolled back?
<!--
* If unsure, leave it blank.
-->
- [x] YES 💚
- [ ] NO ❌
## What
Adapting to CDK signature changes in `StreamConfig`
## Review guide
<!--
1. `x.py`
2. `y.py`
-->
## User Impact
<!--
* What is the end result perceived by the user?
* If there are negative side effects, please list them.
-->
## Can this PR be safely reverted and rolled back?
<!--
* If unsure, leave it blank.
-->
- [x] YES 💚
- [ ] NO ❌
as part of the move of destination-snowflake to the kotlin CDK, we tried improve concurrency by only `DELETE`ing from `_airbyte_destination_state` if it has some data to delete (by issuing an `IF EXISTS` in the same transaction.
Looks like it might be causing some stuck syncs, so we're reverting that "improvement"
not only bringing snowflake to the latest CDK but also:
1) Bringing the `SourceOperation` into production code from the test code. There's really no reason those improvements should stay out of production (and they're present in the source-snowflake)
2) adding `putTimestamp` into the `SourceOperation`, so that snowflake doesn't throw an exception at every call, which implies it also creates a new thread
3) make use of the newly added ability to filter orphan thread on shutdown. We filter all the threads created during calls to `SFStatement.close()`
4) don't always take a lock when deleting destinationStates. We now check if there's any states to delete by doing a `SELECT` (and not taking any table lock) before issuing the `DELETE` (the old behavior was causing test contention, and it's a bad idea in general)
5) only execute `airbyte_internal._airbyte_destination_state`
