feat(v2): read path improvements

[aleks-p]

Addresses a few issues encountered in the read path for v2:

1. Simplifies the query plan representation. The current approach encodes the graph for the query plan as a int32 array which makes it difficult to reason about and maintain. We now have a straightforward node graph in place.
2. Replaces the static concurrency limit in the query backend with the gradient2 implementation from go-concurrency-limits. This adds some elasticity to the query backend under high load and should allow for fine tuning in the future. The limiter parameters are internal for now, but might be exposed in the future.
3. Updates the query backend client config to reduce aggressive retries. This gives large queries a better chance of succeeding by slowing them down. Further tuning will likely be needed here in the future.

[kolesnikovae]

Brilliant work, Aleks! Thank you so much for solving this!

[kolesnikovae]

👍🏻 I'll check other places

[kolesnikovae]

Nit: The official guide prescribes to name the options in a cumbersome way. I propose to name enums in the way we want (but keeping the UNSPECIFIED as 0 value) – READ and MERGE work perfectly here, so we use them. I would also add UNSPECIFIED.

Also, you likely know this, but just in case: there's a nice trick for "local" enums, that makes things a little less verbose. I'm planning to do that with ReportType and QueryType (or get rid of the type prefixes, at least).

message QueryNode {
  Type type = 1;
  repeated QueryNode children = 2;
  repeated metastore.v1.BlockMeta blocks = 3;
  enum Type {
    UNKNOWN = 0;
    MERGE = 1;
    READ = 2;
  }
}

[kolesnikovae]

There's an optimisation we can make later:

The breakdown is not entirely uniform, as is also the case in the source version. If we have maxReads = 20, and 101 blocks, the last node will have just 1 block.

I had this snippet (but I'm sure there's a better way):

func Split[S ~[]E, E any](slice S, batchSize int) []S {
	size := len(slice)
	if size == 0 || batchSize < 1 {
		return nil
	}
	n := int(math.Ceil(float64(size) / float64(batchSize)))
	base := size / n
	remainder := size % n
	batches := make([]S, 0, n)
	var offset int
	for i := 0; i < n; i++ {
		end := offset + base
		if i < remainder {
			// Distribute remaining elements among the first few batches.
			end++
		}
		batches = append(batches, slice[offset:end])
		offset = end
	}
	return batches
}

Example:

data   []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
batch  3

Batch 1: [1 2 3]
Batch 2: [4 5 6]
Batch 3: [7 8]
Batch 4: [9 10]

Probably the same for merge nodes, as far as I understand.