perf(table): Decrease memory pressure by releasing records

[abhirathod95]

The fanout function currently defer's record.Release(). This isn't a big deal if the iterable it reads from contains a small amount of data. However, if it's a long lived streaming channel or contains a lot of data, the fanout function will hold on to **all ** the records until it processes all of them. This dramatically increases memory use, especially in a streaming use case.

This PR extracts out the code to process each record into it's own function, and calls defer record.Release() there. No functional modifications were made. This will release each record after successfully processing the data.

[laskoviymishka]

Fair point. The defer scoping issue is real and obvious from reading the code, but the PR claims a memory improvement without any measurement. A benchmark would prove the claim and prevent regression.

Something like:

func BenchmarkFanoutMemory(b *testing.B) {
    // Feed N large record batches through the fanout writer
    // Report b.ReportAllocs() and track peak memory via runtime.MemStats
}

Doesn't need to be fancy, even feeding 100 batches through the fanout and showing that peak HeapInuse stays flat (new) vs grows linearly (old) would be enough.

The code change is obviously correct from reading it, but a benchmark makes the improvement concrete and guards against regression.

[zeroshade]

I agree with @laskoviymishka, let's add a benchmark or something to confirm and then this is good for me.

[abhirathod95]

Yep, that makes sense. Let me add a benchmark

[abhirathod95]

Added BenchmarkFanoutMemory in table/fanout_memory_bench_test.go to measure the improvement.

The benchmark uses a gated RecordReader that feeds 100 batches (50K rows each) one at a time through a single fanout writer, measuring HeapInuse between sends via runtime.GC() + runtime.ReadMemStats().
This directly observes whether previous batches are retained (old) or released promptly (new).

  benchstat (old vs new, n=6):

                  │   OLD (pre-fix)    │            NEW (with fix)            │
                  │ peak-heap-delta    │ peak-heap-delta  vs base             │
  FanoutMemory-14       2.111Gi ± 6%        1.593Gi ± 6%  -24.53% (p=0.002)

                  │     sec/op         │   sec/op         vs base             │
  FanoutMemory-14       1.555 ± 1%         1.550 ± 1%     ~ (p=0.699)

                  │     B/op           │     B/op         vs base             │
  FanoutMemory-14      19.48Gi ± 1%       19.59Gi ± 1%    ~ (p=0.485)

                  │   allocs/op        │   allocs/op      vs base             │
  FanoutMemory-14      36.31M ± 0%        36.31M ± 0%     ~ (p=0.240)

~25% reduction in peak live heap (p=0.002), no change in wall-clock time or total allocations

[zeroshade]

Nice, looks great! Thanks!

[laskoviymishka]