perf: optimize batch execution path for reduced allocations

Motivation
----------
The BATCH execution path (executeBatch / writeBatchFrame) in the gocql
driver performed several avoidable allocations per batch execution:
individual make() calls for each statement's queryValues slice, a
separate stmts map[string]string to track prepared-ID-to-statement
mappings, sequential preparation of all statements, and unbounded
recursive retries on RequestErrUnprepared errors.

Changes
-------
1. Concurrent statement preparation (Item 1):
   Split executeBatch into a retry loop (executeBatch) and single-attempt
   logic (executeBatchOnce). Unique statements are collected into a map
   and prepared concurrently via goroutines + sync.WaitGroup. The
   existing prepareStatement coalesces concurrent calls for the same
   statement via execIfMissing on the LRU cache, so this is safe.

2. Bulk-allocate queryValues (Item 2):
   Two-pass approach: first pass computes totalValues across all entries,
   second pass slices from a single []queryValues allocation. Replaces N
   individual make([]queryValues, colCount) calls with one.

3. Eliminate stmts map (Item 3):
   Added sourceStmt field to batchStatment struct (frame.go). On
   RequestErrUnprepared, a linear scan of req.statements replaces the
   map lookup. This is acceptable because it only runs on the error
   recovery path, not the hot path. Added unexported stmt field to
   RequestErrUnprepared (errors.go) to carry the source statement text
   from executeBatchOnce back to the retry loop.

4. Reserve(n int) API (Item 4):
   Added Batch.Reserve(n int) *Batch to session.go, which pre-allocates
   the Entries slice capacity. Named Reserve (not Size) because Size()
   already exists and returns len(b.Entries).

5. Bounded retry on RequestErrUnprepared (Item 9):
   Converted the unbounded recursive executeBatch call to an iterative
   loop with maxBatchPrepareRetries = 1. If re-preparation fails twice,
   the error is returned rather than retrying indefinitely.

Benchmark results (synthetic, 8 samples, not end-to-end latency)
----------------------------------------------------------------
These benchmarks measure allocation patterns only. They do NOT measure
real Cassandra/Scylla round-trip latency or throughput. The numbers
below reflect the cost of building batch data structures in isolation.

Reserve() — Batch.Query append:
  entries=10:   2407 B/op 35 allocs -> 1110 B/op 31 allocs  (-54% B, -11% allocs)
  entries=100: 21791 B/op 308 allocs -> 11720 B/op 301 allocs (-46% B, -2% allocs)

Bulk queryValues allocation — writeBatchFrame construction:
  entries=10:   3434 B/op 74 allocs  ->  2294 B/op 42 allocs  (-33% B, -43% allocs)
  entries=100: 32319 B/op 704 allocs -> 21668 B/op 402 allocs (-33% B, -43% allocs)

Files changed: conn.go, frame.go, errors.go, session.go
Files added: batch_bench_test.go

Author: mykaul

batch_bench_test.go

@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package gocql
+
+import (
+	"fmt"
+	"testing"
+)
+
+// BenchmarkBatchQueryAppend measures the cost of appending entries to a Batch
+// via the Query() method. This exercises slice growth and BatchEntry allocation.
+func BenchmarkBatchQueryAppend(b *testing.B) {
+	for _, size := range []int{10, 100} {
+		b.Run(fmt.Sprintf("entries=%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				batch := &Batch{
+					Type: LoggedBatch,
+				}
+				for j := 0; j < size; j++ {
+					batch.Query("INSERT INTO ks.tbl (pk, v) VALUES (?, ?)", j, fmt.Sprintf("val_%d", j))
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkBatchQueryAppendPreallocated measures the cost of appending entries
+// to a Batch with a pre-allocated Entries slice, to serve as comparison target
+// for the Reserve() optimization.
+func BenchmarkBatchQueryAppendPreallocated(b *testing.B) {
+	for _, size := range []int{10, 100} {
+		b.Run(fmt.Sprintf("entries=%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				batch := (&Batch{
+					Type: LoggedBatch,
+				}).Reserve(size)
+				for j := 0; j < size; j++ {
+					batch.Query("INSERT INTO ks.tbl (pk, v) VALUES (?, ?)", j, fmt.Sprintf("val_%d", j))
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkBatchBuildWriteFrame measures the cost of building a writeBatchFrame
+// from pre-populated batch statements with prepared IDs and queryValues.
+// This isolates the allocation patterns in executeBatch's frame-building logic.
+func BenchmarkBatchBuildWriteFrame(b *testing.B) {
+	for _, size := range []int{10, 100} {
+		b.Run(fmt.Sprintf("entries=%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+
+			// Pre-build mock column types and values
+			colCount := 2
+			typ := NativeType{proto: protoVersion4, typ: TypeInt}
+
+			for i := 0; i < b.N; i++ {
+				req := &writeBatchFrame{
+					typ:              LoggedBatch,
+					statements:       make([]batchStatment, size),
+					consistency:      Quorum,
+					defaultTimestamp: true,
+				}
+
+				stmts := make(map[string]string, size)
+
+				// Simulate the allocation pattern from executeBatch
+				for j := 0; j < size; j++ {
+					bs := &req.statements[j]
+					bs.preparedID = []byte(fmt.Sprintf("prepared_%d", j%5))
+					stmts[string(bs.preparedID)] = fmt.Sprintf("INSERT INTO ks.tbl (pk, v) VALUES (?, ?)")
+
+					bs.values = make([]queryValues, colCount)
+					for k := 0; k < colCount; k++ {
+						val, _ := Marshal(typ, j+k)
+						bs.values[k] = queryValues{value: val}
+					}
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkBatchBuildWriteFrameBulkAlloc measures the cost of building a
+// writeBatchFrame using a single bulk allocation for all queryValues and
+// sourceStmt on the struct (no separate map). This reflects the optimized
+// allocation pattern from Items 2 and 3.
+func BenchmarkBatchBuildWriteFrameBulkAlloc(b *testing.B) {
+	for _, size := range []int{10, 100} {
+		b.Run(fmt.Sprintf("entries=%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+
+			colCount := 2
+			typ := NativeType{proto: protoVersion4, typ: TypeInt}
+
+			for i := 0; i < b.N; i++ {
+				req := &writeBatchFrame{
+					typ:              LoggedBatch,
+					statements:       make([]batchStatment, size),
+					consistency:      Quorum,
+					defaultTimestamp: true,
+				}
+
+				// Bulk-allocate all queryValues in a single slice
+				allValues := make([]queryValues, size*colCount)
+
+				for j := 0; j < size; j++ {
+					bs := &req.statements[j]
+					bs.preparedID = []byte(fmt.Sprintf("prepared_%d", j%5))
+					bs.sourceStmt = "INSERT INTO ks.tbl (pk, v) VALUES (?, ?)"
+
+					bs.values = allValues[j*colCount : (j+1)*colCount]
+					for k := 0; k < colCount; k++ {
+						val, _ := Marshal(typ, j+k)
+						bs.values[k] = queryValues{value: val}
+					}
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkBatchWriteFrameSerialization measures the cost of serializing a
+// writeBatchFrame to bytes via the framer.
+func BenchmarkBatchWriteFrameSerialization(b *testing.B) {
+	for _, size := range []int{10, 100} {
+		b.Run(fmt.Sprintf("entries=%d", size), func(b *testing.B) {
+			b.ReportAllocs()
+
+			colCount := 2
+			typ := NativeType{proto: protoVersion4, typ: TypeInt}
+
+			// Pre-build the frame once
+			frame := &writeBatchFrame{
+				typ:              LoggedBatch,
+				statements:       make([]batchStatment, size),
+				consistency:      Quorum,
+				defaultTimestamp: true,
+			}
+
+			for j := 0; j < size; j++ {
+				bs := &frame.statements[j]
+				bs.preparedID = []byte(fmt.Sprintf("prepared_%d", j%5))
+				bs.values = make([]queryValues, colCount)
+				for k := 0; k < colCount; k++ {
+					val, _ := Marshal(typ, j+k)
+					bs.values[k] = queryValues{value: val}
+				}
+			}
+
+			b.ResetTimer()
+
+			for i := 0; i < b.N; i++ {
+				f := newFramer(nil, protoVersion4)
+				err := frame.buildFrame(f, 1)
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}