feat: Implement batch sender.

[marianogappa]

Problem

The DFS scheduler implements concurrent resolution of resources, but only if resources are sent in batches to the top-level res channel. Thus, the same plugin syncs a lot quicker simply by batching resources before sending them through the channel.

Solution this PR implements

// BatchSender is a helper struct that batches items and sends them in batches of batchSize or after batchTimeout.
//
// - If item is already a slice, it will be sent directly
// - Otherwise, it will be added to the current batch
// - If the current batch has reached the batch size, it will be sent immediately
// - Otherwise, a timer will be started to send the current batch after the batch timeout
type BatchSender struct {

Currently with a batch timeout of 100ms and batch size of 100, it manages a 5x improvement on a sync that reproduces the issue brought forward by the community (https://github.com/jeromewir/cq-source-concurrency-childtable-example):

The PR tries to touch the least amount of scheduler code, and stays away from tricky language constructs as much as possible.

[marianogappa]

😱 Looks like the solution doesn't contemplate that the res channel might get closed on the last batch. Potentially cannot use timer.AfterFunc since I cannot .Wait on it 🤔

Anyway, it shows promise but we might not need it if we choose to move to shuffle-queue. I can look into the special case, though.

[marianogappa]

I think this can be mitigated with a .Close() method. It will stop the timer if set, and flush the last batch immediately.

[erezrokah]

Looks good 👍

1. Can we add some tests that cover the new code (if we don't have them already via the scheduler tests)? I mean sending both single items and slices to the result channel
2. I think batchSize should default to 5 which is our singleResourceMaxConcurrency. There's not point in sending more than 5 items at a time as they will block on

   plugin-sdk/scheduler/scheduler_dfs.go

   Line 151 in 32855ea

   resourceSemVal, _ := s.scheduler.singleTableConcurrency.LoadOrStore(resourceConcurrencyKey, semaphore.NewWeighted(s.scheduler.singleResourceMaxConcurrency))

   I think

[erezrokah]

We decided to add e2e test using our source and destination test plugins that will cover the changes in this PR

[marianogappa]

@erezrokah passing E2E: cloudquery/cloudquery#19875

[marianogappa]

Will merge this based on it passing E2E tests in CI

[marianogappa]

The testing I did initially was flawed (upgraded SDK on CLI rather than on source plugin), but since then I did a separate test that passed as well:
cloudquery/cloudquery#19908