lightning: unclosed Parquet metadata readers can exhaust S3 HTTP connections and stall source loading

## Bug Report

### 1. Minimal reproduce step (Required)

TiDB Lightning can stop making progress during Parquet source loading when S3 HTTP connections are exhausted. `ReadParquetFileRowCountByFile` opens a reader but does not close it on either normal return or a Parquet reader construction error.

A standalone differential reproducer demonstrates the connection leak without concurrent sampling:

1. Use the repository fixture `pkg/lightning/mydump/examples/test.parquet` from commit `202b7f47286a1109b5c957401d34c9358d130ae0`.
2. Repeatedly execute the same `Open -> file.NewParquetReader -> MetaData().NumRows` ownership path, using the affected Arrow version, the affected AWS SDK HTTP transport, and a local HTTP Range endpoint modeling the relevant S3 reader behavior.
3. Compare the original path with a variant that only adds `defer r.Close()` immediately after successful `Open`.

| HTTP connections per host | Files requested | Original: files completed | With Close: files completed |
|---|---:|---:|---:|
| 4 | 12 | 4; next Open blocks | 12 |
| 2048 (SDK default) | 2052 | 2048; next Open blocks | 2052 |

The original default-limit run leaves 2048 response bodies unclosed. The corrected run closes all 6156 bodies opened across the 2052 metadata reads, with zero unclosed bodies. A local goroutine dump confirms the blocked request is waiting in `net/http.(*Transport).getConn`. The test deadline converts the wait into a bounded failure.

**Reproduction scope:** this is a standalone mechanism reproducer, not a full Lightning/S3 integration test. It uses the actual Arrow dependency and AWS SDK transport, but models the relevant S3 Read/Seek/Close behavior against `httptest`; it does not use real S3, SDK GetObject middleware, TLS, or customer data. Execution is sequential, so concurrent column sampling is not required to trigger the reproduced leak. The malformed-file construction-error path has not been dynamically tested.

Self-contained reproducer and commands

Create a new directory and save the following files as `go.mod` and `main.go`. Obtain the public test fixture and run:

```bash
curl -fsSL https://raw.githubusercontent.com/pingcap/tidb/202b7f47286a1109b5c957401d34c9358d130ae0/pkg/lightning/mydump/examples/test.parquet -o test.parquet
go run -mod=mod .
```

`go.mod`:

```go
module lightning-reader-repro

go 1.25.0

require (
github.com/apache/arrow-go/v18 v18.0.0
github.com/aws/aws-sdk-go-v2 v1.41.5
)

require (
github.com/andybalholm/brotli v1.1.1 // indirect
github.com/apache/thrift v0.21.0 // indirect
github.com/aws/smithy-go v1.24.2 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/goccy/go-json v0.10.4 // indirect
github.com/golang/snappy v0.0.4 // indirect
github.com/google/flatbuffers v24.3.25+incompatible // indirect
github.com/klauspost/compress v1.17.11 // indirect
github.com/klauspost/cpuid/v2 v2.2.9 // indirect
github.com/pierrec/lz4/v4 v4.1.21 // indirect
github.com/zeebo/xxh3 v1.0.2 // indirect
golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
golang.org/x/sync v0.16.0 // indirect
golang.org/x/sys v0.34.0 // indirect
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
)

replace github.com/apache/arrow-go/v18 => github.com/joechenrh/arrow-go/v18 v18.0.0-20250911101656-62c34c9a3b82
```

`main.go`:

```go
package main

import (
"bytes"
"context"
"fmt"
"github.com/apache/arrow-go/v18/parquet/file"
awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
"io"
"net/http"
"net/http/httptest"
"os"
"runtime"
"runtime/pprof"
"sync/atomic"
"time"
)

// Local S3-style GET/Range endpoint; actual production-version AWS transport.
type objectStore struct {
client *http.Client
url string
data []byte
opened, closed atomic.Int64
}
type countedBody struct {
io.ReadCloser
store *objectStore
closed bool
}

func (b *countedBody) Close() error {
if !b.closed {
b.closed = true
b.store.closed.Add(1)
}
return b.ReadCloser.Close()
}
func (s *objectStore) get(ctx context.Context, off int64) (io.ReadCloser, error) {
req, _ := http.NewRequestWithContext(ctx, "GET", s.url, nil)
if off > 0 {
req.Header.Set("Range", fmt.Sprintf("bytes=%d-", off))
}
resp, err := s.client.Do(req)
if err != nil {
return nil, err
}
s.opened.Add(1)
return &countedBody{ReadCloser: resp.Body, store: s}, nil
}
func (s *objectStore) Open(ctx context.Context) (*objectReader, error) {
b, err := s.get(ctx, 0)
if err != nil {
return nil, err
}
return &objectReader{store: s, ctx: ctx, body: b}, nil
}

// Relevant production S3 seek semantics, excluding retries/prefetch.
type objectReader struct {
store *objectStore
ctx context.Context
body io.ReadCloser
pos int64
}

func (r *objectReader) Read(p []byte) (int, error) {
remain := int64(len(r.store.data)) - r.pos
if remain == 0 {
return 0, io.EOF
}
if int64(len(p)) > remain {
p = p[:remain]
}
n, e := r.body.Read(p)
r.pos += int64(n)
return n, e
}
func (r *objectReader) Close() error { return r.body.Close() }
func (r *objectReader) Seek(off int64, whence int) (int64, error) {
switch whence {
case io.SeekCurrent:
off += r.pos
case io.SeekEnd:
off += int64(len(r.store.data))
}
if off == r.pos {
return off, nil
}
if off >= int64(len(r.store.data)) {
r.body.Close()
r.body = io.NopCloser(bytes.NewReader(nil))
r.pos = int64(len(r.store.data))
return r.pos, nil
}
if off > r.pos && off-r.pos <= 64*1024 {
_, err := io.CopyN(io.Discard, r, off-r.pos)
return r.pos, err
}
if err := r.body.Close(); err != nil {
return 0, err
}
b, err := r.store.get(r.ctx, off)
if err != nil {
return 0, err
}
r.body = b
r.pos = off
return off, nil
}

// Production parquetFileWrapper ReadAt/Seek logic, omitting errors.Trace.
type readSeekCloser interface {
io.Reader
io.Seeker
io.Closer
}
type parquetFileWrapper struct {
readSeekCloser
lastOff int64
skipBuf []byte
}

func (pf *parquetFileWrapper) readNBytes(p []byte) (int, error) {
n, err := io.ReadFull(pf, p)
if err != nil && err != io.EOF {
return 0, err
}
if n != len(p) {
return n, fmt.Errorf("error reading %d bytes, only read %d bytes", len(p), n)
}
return n, nil
}
func (pf *parquetFileWrapper) ReadAt(p []byte, off int64) (int, error) {
gap := int(off - pf.lastOff)
if gap < 0 || gap > cap(pf.skipBuf) {
if _, err := pf.Seek(off, io.SeekStart); err != nil {
return 0, err
}
} else {
pf.skipBuf = pf.skipBuf[:gap]
if read, err := pf.readNBytes(pf.skipBuf); err != nil {
return read, err
}
}
read, err := pf.readNBytes(p)
if err != nil {
return read, err
}
pf.lastOff = off + int64(read)
return len(p), nil
}
func (pf *parquetFileWrapper) Seek(off int64, w int) (int64, error) {
v, e := pf.readSeekCloser.Seek(off, w)
pf.lastOff = v
return v, e
}

// Same ownership path as ReadParquetFileRowCountByFile; fix adds one defer.
func readRowCount(ctx context.Context, s *objectStore, fixed bool) (int64, error) {
r, err := s.Open(ctx)
if err != nil {
return 0, err
}
if fixed {
defer r.Close()
}
reader, err := file.NewParquetReader(&parquetFileWrapper{readSeekCloser: r})
if err != nil {
return 0, err
}
return reader.MetaData().NumRows, nil
}
func run(label string, data []byte, limit, files int, fixed bool) bool {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.ServeContent(w, r, "test.parquet", time.Time{}, bytes.NewReader(data))
}))
defer srv.Close()
tr := awshttp.NewBuildableClient().GetTransport()
tr.MaxConnsPerHost = limit
defer tr.CloseIdleConnections()
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
s := &objectStore{client: &http.Client{Transport: tr}, url: srv.URL, data: data}
completed := 0
var lastErr error
for i := 0; i < files; i++ {
if !fixed && i == limit {
time.AfterFunc(300*time.Millisecond, func() {
f, e := os.Create(fmt.Sprintf("blocked-%d.goroutines.txt", limit))
if e == nil {
defer f.Close()
pprof.Lookup("goroutine").WriteTo(f, 2)
}
})
}
n, e := readRowCount(ctx, s, fixed)
if e != nil {
lastErr = e
break
}
if n <= 0 {
panic("fixture has no rows")
}
completed++
}
fmt.Printf("%s limit=%d requested=%d completed=%d opened=%d closed=%d unclosed=%d error=%v\n", label, limit, files, completed, s.opened.Load(), s.closed.Load(), s.opened.Load()-s.closed.Load(), lastErr)
if fixed {
return completed == files && lastErr == nil && s.opened.Load() == s.closed.Load()
}
return completed == limit && lastErr != nil
}
func main() {
data, e := os.ReadFile("test.parquet")
if e != nil {
panic(e)
}
fmt.Printf("go=%s AWS-default-MaxConnsPerHost=%d\n", runtime.Version(), awshttp.DefaultHTTPTransportMaxConnsPerHost)
a := run("original", data, 4, 12, false)
b := run("with-close", data, 4, 12, true)
c := run("original-default-limit", data, awshttp.DefaultHTTPTransportMaxConnsPerHost, 2052, false)
d := run("with-close-default-limit", data, awshttp.DefaultHTTPTransportMaxConnsPerHost, 2052, true)
if !a || !b || !c || !d {
os.Exit(1)
}
fmt.Println("PASS: original exhausts pool; adding Close completes all files and closes every opened response")
}
```

The program exits with status 0 only when the original variant exhausts both tested pool limits and the Close variant completes both workloads without unclosed response bodies. It also writes local blocked-goroutine dumps.

### 2. What did you expect to see? (Required)

Parquet metadata readers should release their underlying storage readers on every return path. Source loading should not permanently consume one HTTP connection per scanned file or wait indefinitely for connections leaked by completed metadata reads.

### 3. What did you see instead (Required)

Observed on affected Lightning instances:

- Source loading remains at `load data source start`, while the progress HTTP endpoint remains responsive.
- File descriptors plateau around 2057 and goroutines around 4182; CPU is nearly idle. The process file-descriptor limit is much higher, so this is not exhaustion of the OS descriptor limit.
- Normal row-reading and import counters remain zero, while some Parquet total-row metadata has already been collected.
- Runtime stacks show S3 GetObject requests waiting in the same HTTP transport's `getConn` for 39 minutes. Both per-file row-count reads and Parquet sampling are blocked.

Sanitized representative call chains (caller to callee):

```text
mdLoaderSetup.constructFileInfo
-> ReadParquetFileRowCountByFile
-> S3Storage.Open (or parquetFileWrapper.ReadAt -> s3ObjectReader.Seek)
-> S3Storage.open -> S3.GetObject
-> net/http.(*Transport).getConn

mdLoaderSetup.constructFileInfo
-> SampleStatisticsFromParquet
-> NewParquetParser -> parquetFileWrapper.Open
OR ParquetParser.readSingleRow -> parquetFileWrapper.ReadAt -> s3ObjectReader.Seek
-> S3Storage.open -> S3.GetObject
-> net/http.(*Transport).getConn
```

#### Analysis

- [ReadParquetFileRowCountByFile](https://github.com/pingcap/tidb/blob/202b7f47286a1109b5c957401d34c9358d130ae0/pkg/lightning/mydump/parquet_parser.go#L517-L533) returns the metadata row count without closing the reader.
- Arrow seeks to the footer and then reads the metadata preceding it. The final S3 range response can still contain trailing footer bytes when metadata parsing returns. With neither EOF consumption nor Close, the connection remains occupied. [Arrow metadata reader](https://github.com/joechenrh/arrow-go/blob/62c34c9a3b82/parquet/file/file_reader.go).
- The affected AWS SDK defaults to [MaxConnsPerHost = 2048](https://github.com/aws/aws-sdk-go-v2/blob/v1.41.5/aws/transport/http/client.go). The standalone original/Close comparison reproduces exhaustion at this exact limit.

The missing Close and its connection-exhaustion mechanism are reproduced. Runtime observations strongly support this mechanism contributing to the incident, but waiting stacks do not identify every existing connection owner. Concurrent per-column sampling can also hold connections; a Close fix should be validated with that workload before claiming it resolves every source-loading stall.

Scope: the loader calls this helper for Parquet. Uncompressed SQL/CSV size estimation does not open files; compressed SQL/CSV sampling closes its reader. This report does not claim a full audit of other reader paths.

Related: #56104 discusses expensive Parquet pre-scanning. This report specifically covers unclosed readers and HTTP connection-pool exhaustion rather than scan latency alone.

### 4. What is your TiDB version? (Required)

```text
TiDB Lightning: v8.5.7
Git commit: 202b7f47286a1109b5c957401d34c9358d130ae0
Runtime reported by Lightning: go1.25.10
AWS SDK for Go v2: v1.41.5
Arrow replacement: github.com/joechenrh/arrow-go/v18 v18.0.0-20250911101656-62c34c9a3b82
```

The standalone reproducer ran with Go 1.25.1. Other TiDB versions have not been verified.