diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 27d0712ec..09e3e38d2 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -22,7 +22,7 @@ jobs:
     - name: Install Go
       uses: actions/setup-go@v6
       with:
-        go-version: stable
+        go-version-file: 'go.mod'
 
     - name: Install Task
       uses: ./.github/actions/setup-task
@@ -49,7 +49,7 @@ jobs:
     - name: Install Go
       uses: actions/setup-go@v6
       with:
-        go-version: stable
+        go-version-file: 'go.mod'
 
     - name: Set version env variables
       run: |
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e8f6934e5..cb7ddb1d3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,10 @@ All notable changes to this project will be documented in this file.
 
 - Input `websocket`: Added a `max_message_size` field that bounds the size of individual inbound messages via the underlying connection's read limit. (@prakhargarg105)
 
+### Fixed
+
+- Input `file`: The known file size is now passed to the scanner as a hint, allowing the `to_the_end` scanner to pre-allocate its read buffer instead of growing it repeatedly via `io.ReadAll`. This avoids the transient double-buffering that previously made peak memory roughly twice the file size when reading a whole file (measured ~59% fewer bytes allocated, and 36 allocations down to 2, for a 64 MiB file). The size is treated purely as a hint and never affects the bytes returned: sources of unknown size retain the exact `io.ReadAll` behaviour, pre-allocation from a hint is capped at 128 MiB to guard against wildly wrong sizes, a hint that overestimates the content does not pin the excess capacity for the message's lifetime, and the `decompress` scanner strips the hint before it reaches its child, as it describes the compressed stream. Deprecated string `codec` configs are unaffected. (@rockdatasrl001)
+
 ## 4.77.0 - 2026-07-30
 
 ### Added
diff --git a/internal/component/scanner/interface.go b/internal/component/scanner/interface.go
index 82c309c5b..8577875fa 100644
--- a/internal/component/scanner/interface.go
+++ b/internal/component/scanner/interface.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package scanner
 
@@ -24,6 +24,22 @@ type Scanner interface {
 // by codec implementations in order to determine the underlying data format.
 type SourceDetails struct {
 	Name string
+
+	// SizeHint is the total number of bytes of the source, when known, and is
+	// zero otherwise. A source measured as empty is therefore
+	// indistinguishable from one of unknown size, and implementations must
+	// treat zero as unknown. It is a hint only, provided so that
+	// implementations can pre-allocate buffers, and must never be relied upon
+	// for correctness as the underlying source may change between it being
+	// measured and read.
+	//
+	// A scanner that wraps a child scanner and materially transforms the
+	// length of the stream (such as decompress) must clear the hint before
+	// forwarding details, as it describes the stream the wrapper reads, not
+	// the one the child does. A wrapper that shortens the stream by only a
+	// bounded few bytes (such as skip_bom) may forward the hint unchanged, as
+	// it remains a tight upper bound.
+	SizeHint int64
 }
 
 // Creator is an interface implemented by all scanners, which allows components
diff --git a/internal/component/scanner/testutil/capture_scanner.go b/internal/component/scanner/testutil/capture_scanner.go
new file mode 100644
index 000000000..053dc7d7b
--- /dev/null
+++ b/internal/component/scanner/testutil/capture_scanner.go
@@ -0,0 +1,81 @@
+// Copyright 2026 Redpanda Data, Inc.
+
+package testutil
+
+import (
+	"context"
+	"io"
+	"slices"
+	"sync"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+)
+
+// CapturedDetails is a snapshot of the source details a capture scanner
+// observed at stream creation.
+type CapturedDetails struct {
+	Name     string
+	SizeHint int64
+}
+
+// MustRegisterDetailsCaptureScanner registers a scanner plugin under name that
+// records the source details each stream is created with, then reads the
+// stream to the end and delivers it as a single message. The returned function
+// reports every capture so far, oldest first, and is safe for concurrent use.
+// It is intended for tests asserting on the details an input or wrapping
+// scanner propagates.
+func MustRegisterDetailsCaptureScanner(name string) func() []CapturedDetails {
+	var mut sync.Mutex
+	var captured []CapturedDetails
+
+	service.MustRegisterBatchScannerCreator(name,
+		service.NewConfigSpec().Field(service.NewObjectField("").Default(map[string]any{})),
+		func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) {
+			return &captureScannerCreator{fn: func(details *service.ScannerSourceDetails) {
+				mut.Lock()
+				captured = append(captured, CapturedDetails{Name: details.Name(), SizeHint: details.SizeHint()})
+				mut.Unlock()
+			}}, nil
+		})
+
+	return func() []CapturedDetails {
+		mut.Lock()
+		defer mut.Unlock()
+		return slices.Clone(captured)
+	}
+}
+
+type captureScannerCreator struct {
+	fn func(*service.ScannerSourceDetails)
+}
+
+func (c *captureScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) {
+	c.fn(details)
+	return service.AutoAggregateBatchScannerAcks(&captureScanner{r: rdr}, aFn), nil
+}
+
+func (c *captureScannerCreator) Close(context.Context) error { return nil }
+
+type captureScanner struct {
+	r io.ReadCloser
+}
+
+func (c *captureScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) {
+	if c.r == nil {
+		return nil, io.EOF
+	}
+	b, err := io.ReadAll(c.r)
+	if err != nil {
+		return nil, err
+	}
+	_ = c.r.Close()
+	c.r = nil
+	return service.MessageBatch{service.NewMessage(b)}, nil
+}
+
+func (c *captureScanner) Close(ctx context.Context) error {
+	if c.r == nil {
+		return nil
+	}
+	return c.r.Close()
+}
diff --git a/internal/impl/io/input_file.go b/internal/impl/io/input_file.go
index ba906c7d1..297e759ad 100644
--- a/internal/impl/io/input_file.go
+++ b/internal/impl/io/input_file.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package io
 
@@ -163,6 +163,21 @@ func (f *fileConsumer) getReader(ctx context.Context) (scannerInfo, error) {
 	details := service.NewScannerSourceDetails()
 	details.SetName(nextPath)
 
+	var modTimeUTC time.Time
+	if fInfo, err := file.Stat(); err == nil {
+		modTimeUTC = fInfo.ModTime().UTC()
+
+		// Only regular files have a meaningful size to pre-allocate against, a
+		// FIFO, device or directory does not. The positivity check keeps the
+		// documented SizeHint contract (zero means unknown, never negative)
+		// even when a custom filesystem misreports a size.
+		if s := fInfo.Size(); s > 0 && fInfo.Mode().IsRegular() {
+			details.SetSizeHint(s)
+		}
+	} else {
+		f.log.Errorf("Failed to read metadata from file '%v': %v", nextPath, err)
+	}
+
 	scanner, err := f.scannerCtor.Create(file, func(ctx context.Context, err error) error {
 		if err == nil && f.delete {
 			return f.nm.FS().Remove(nextPath)
@@ -174,13 +189,6 @@ func (f *fileConsumer) getReader(ctx context.Context) (scannerInfo, error) {
 		return scannerInfo{}, err
 	}
 
-	var modTimeUTC time.Time
-	if fInfo, err := file.Stat(); err == nil {
-		modTimeUTC = fInfo.ModTime().UTC()
-	} else {
-		f.log.Errorf("Failed to read metadata from file '%v'", nextPath)
-	}
-
 	f.scannerInfo = &scannerInfo{
 		scanner:     scanner,
 		currentPath: nextPath,
diff --git a/internal/impl/io/input_file_size_test.go b/internal/impl/io/input_file_size_test.go
new file mode 100644
index 000000000..035c4ceae
--- /dev/null
+++ b/internal/impl/io/input_file_size_test.go
@@ -0,0 +1,65 @@
+// Copyright 2026 Redpanda Data, Inc.
+
+package io_test
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	scannertestutil "github.com/redpanda-data/benthos/v4/internal/component/scanner/testutil"
+	"github.com/redpanda-data/benthos/v4/internal/component/testutil"
+	"github.com/redpanda-data/benthos/v4/internal/manager/mock"
+	"github.com/redpanda-data/benthos/v4/internal/message"
+)
+
+// capturedSizes reports the source details observed by the capture_size_test
+// scanner, so that a test can assert on what the file input propagated.
+var capturedSizes = scannertestutil.MustRegisterDetailsCaptureScanner("capture_size_test")
+
+// TestFileInputPropagatesSize asserts that the file input reads Size() from the
+// Stat it already performs and passes it to the scanner, for regular files.
+func TestFileInputPropagatesSize(t *testing.T) {
+	tmpDir := t.TempDir()
+
+	const content = "hello world, this content has a known length"
+	path := filepath.Join(tmpDir, "sized.txt")
+	require.NoError(t, os.WriteFile(path, []byte(content), 0o644))
+
+	conf, err := testutil.InputFromYAML(fmt.Sprintf(`
+file:
+  paths: [ "%v/*.txt" ]
+  scanner:
+    capture_size_test: {}
+`, tmpDir))
+	require.NoError(t, err)
+
+	i, err := mock.NewManager().NewInput(conf)
+	require.NoError(t, err)
+
+	i.TriggerStartConsuming()
+
+	var tran message.Transaction
+	select {
+	case tran = <-i.TransactionChan():
+	case <-time.After(time.Second):
+		t.Fatal("timed out")
+	}
+
+	assert.Equal(t, content, string(tran.Payload.Get(0).AsBytes()))
+	require.NoError(t, tran.Ack(t.Context(), nil))
+
+	var got int64
+	for _, c := range capturedSizes() {
+		if filepath.Base(c.Name) == "sized.txt" {
+			got = c.SizeHint
+		}
+	}
+	assert.Equal(t, int64(len(content)), got,
+		"file input should propagate the size it already obtained from Stat")
+}
diff --git a/internal/impl/pure/scanner_decompress.go b/internal/impl/pure/scanner_decompress.go
index d651e4a36..f8cda4181 100644
--- a/internal/impl/pure/scanner_decompress.go
+++ b/internal/impl/pure/scanner_decompress.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package pure
 
@@ -63,6 +63,14 @@ func (c *decompressScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc
 	if !ok {
 		cRdr = io.NopCloser(dRdr)
 	}
+	// A size hint, if present, describes the compressed stream, but the child
+	// reads the decompressed one, whose length is unknown. Strip the hint on a
+	// copy rather than mislead the child (or mutate the caller's details).
+	if details.SizeHint() != 0 {
+		trimmed := *details
+		trimmed.SetSizeHint(0)
+		details = &trimmed
+	}
 	return c.child.Create(cRdr, aFn, details)
 }
 
diff --git a/internal/impl/pure/scanner_decompress_test.go b/internal/impl/pure/scanner_decompress_test.go
index b307bb399..0320151db 100644
--- a/internal/impl/pure/scanner_decompress_test.go
+++ b/internal/impl/pure/scanner_decompress_test.go
@@ -1,11 +1,15 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package pure_test
 
 import (
+	"bytes"
+	"context"
 	"encoding/hex"
+	"io"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/redpanda-data/benthos/v4/internal/component/scanner/testutil"
@@ -32,3 +36,53 @@ test:
 
 	testutil.ScannerTestSuite(t, rdr, nil, inputBytes, "hello", "world", "this", "is", "compressed")
 }
+
+// capturedHints reports the source details observed by the capture_hint_test
+// scanner, so a test can assert on what a wrapping scanner propagated.
+var capturedHints = testutil.MustRegisterDetailsCaptureScanner("capture_hint_test")
+
+// TestDecompressScannerStripsSizeHint asserts that a size hint, which
+// describes the compressed stream, is not forwarded to the child scanner
+// reading the decompressed one, while the rest of the details survive.
+func TestDecompressScannerStripsSizeHint(t *testing.T) {
+	confSpec := service.NewConfigSpec().Field(service.NewScannerField("test"))
+	pConf, err := confSpec.ParseYAML(`
+test:
+  decompress:
+    algorithm: gzip
+    into:
+      capture_hint_test: {}
+`, nil)
+	require.NoError(t, err)
+
+	rdr, err := pConf.FieldScanner("test")
+	require.NoError(t, err)
+
+	inputBytes, err := hex.DecodeString("1f8b080000096e8800ff001e00e1ff68656c6c6f58776f726c64587468697358697358636f6d7072657373656403009104d92d1e000000")
+	require.NoError(t, err)
+
+	details := service.NewScannerSourceDetails()
+	details.SetName("compressed.gz")
+	details.SetSizeHint(int64(len(inputBytes)))
+
+	strm, err := rdr.Create(io.NopCloser(bytes.NewReader(inputBytes)), func(ctx context.Context, err error) error {
+		return nil
+	}, details)
+	require.NoError(t, err)
+
+	m, _, err := strm.NextBatch(t.Context())
+	require.NoError(t, err)
+	require.Len(t, m, 1)
+
+	mBytes, err := m[0].AsBytes()
+	require.NoError(t, err)
+	assert.Equal(t, "helloXworldXthisXisXcompressed", string(mBytes))
+
+	captures := capturedHints()
+	require.NotEmpty(t, captures)
+	last := captures[len(captures)-1]
+	assert.Equal(t, "compressed.gz", last.Name, "name should survive the decompress layer")
+	assert.Zero(t, last.SizeHint, "compressed size hint should not reach the child scanner")
+
+	require.NoError(t, strm.Close(t.Context()))
+}
diff --git a/internal/impl/pure/scanner_skip_bom.go b/internal/impl/pure/scanner_skip_bom.go
index 05c1b8a91..1b6bb4d2a 100644
--- a/internal/impl/pure/scanner_skip_bom.go
+++ b/internal/impl/pure/scanner_skip_bom.go
@@ -45,6 +45,9 @@ type ssbScannerCreator struct {
 }
 
 func (c *ssbScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) {
+	// The size hint is deliberately forwarded unchanged: stripping a BOM
+	// shortens the stream by at most 4 bytes, so the hint remains a tight
+	// upper bound, which the SizeHint contract permits.
 	return c.child.Create(skipBOM(rdr), aFn, details)
 }
 
diff --git a/internal/impl/pure/scanner_to_the_end.go b/internal/impl/pure/scanner_to_the_end.go
index 0b76308d3..2cb06cab0 100644
--- a/internal/impl/pure/scanner_to_the_end.go
+++ b/internal/impl/pure/scanner_to_the_end.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package pure
 
@@ -37,7 +37,9 @@ func toTheEndScannerCreatorFromParsed(conf *service.ParsedConfig) (s *toTheEndSc
 type toTheEndScannerCreator struct{}
 
 func (l *toTheEndScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) {
-	return service.AutoAggregateBatchScannerAcks(&toTheEndScanner{r: rdr}, aFn), nil
+	// The size hint is used to pre-allocate the read buffer only, and is zero
+	// for sources of unknown length.
+	return service.AutoAggregateBatchScannerAcks(&toTheEndScanner{r: rdr, sizeHint: details.SizeHint()}, aFn), nil
 }
 
 func (l *toTheEndScannerCreator) Close(context.Context) error {
@@ -45,14 +47,100 @@ func (l *toTheEndScannerCreator) Close(context.Context) error {
 }
 
 type toTheEndScanner struct {
-	r io.ReadCloser
+	r        io.ReadCloser
+	sizeHint int64
+}
+
+const (
+	// defaultReadAllCap matches the initial buffer capacity io.ReadAll uses.
+	// It floors the pre-allocation so a hint that underestimates wildly (a
+	// file appended to after being measured) doesn't begin with
+	// pathologically small reads.
+	defaultReadAllCap = 512
+
+	// maxPreallocHint caps the capacity pre-allocated from a size hint. The
+	// hint originates from external data (typically a Stat) and can be wildly
+	// wrong: procfs files report multi-terabyte sizes, and a custom
+	// filesystem can return anything, including values for which make()
+	// panics outright (above the runtime's allocation limit, or overflowing
+	// hint+1). io.ReadAll bounded memory by growing incrementally, so
+	// pre-allocation must not turn a bogus hint into a memory spike that can
+	// take out a constrained deployment before a single byte is read. 128 MiB
+	// covers typical whole-stream messages while bounding the blast radius of
+	// a wrong hint; content beyond the clamp reads correctly, growing by
+	// doubling.
+	maxPreallocHint = 128 << 20
+
+	// maxReturnWaste is the largest gap between the returned buffer's
+	// capacity and its content that readAllHinted will leave in place. Beyond
+	// it the content is copied down, as callers hold the returned bytes for
+	// the lifetime of the message and would otherwise pin the excess. Below
+	// it a copy costs more than the memory it reclaims.
+	maxReturnWaste = 64 << 10
+)
+
+// hintedCap converts a positive size hint into the capacity to pre-allocate:
+// the hint floored at io.ReadAll's own starting size, capped at the
+// pre-allocation clamp, plus one. SetSizeHint is exported, so the hint is
+// untrusted input; the clamp keeps a bogus value from panicking make() or
+// spiking memory before a byte is read.
+//
+// The +1 is deliberate: the read loop can only Read while cap > len, so a
+// buffer sized to exactly the content would find len == cap on the final
+// iteration and grow once more, reintroducing the reallocation this avoids.
+func hintedCap(hint int64) int64 {
+	return max(min(hint, maxPreallocHint), defaultReadAllCap) + 1
+}
+
+// readAllHinted is io.ReadAll with a starting capacity hint.
+//
+// The bytes and error returned are identical to io.ReadAll for every hint
+// value; the hint only shapes allocation. A hint of zero or below means the
+// size is unknown, in which case io.ReadAll is used directly.
+//
+// An accurate hint reads the content into a single allocation returned as-is.
+// A hint that falls short (the source grew after being measured, or exceeds
+// the clamp) grows by doubling, keeping cumulative allocation linear in the
+// content size much like io.ReadAll's own chunking. A buffer returned with
+// meaningfully more capacity than content (the source shrank, or the
+// measurement was wrong) is copied down first, so a wrong hint can't pin
+// excess memory for the lifetime of the returned bytes.
+func readAllHinted(r io.Reader, hint int64) ([]byte, error) {
+	if hint <= 0 {
+		return io.ReadAll(r)
+	}
+	buf := make([]byte, 0, hintedCap(hint))
+	for {
+		if len(buf) == cap(buf) {
+			// The +1 serves the same purpose as in hintedCap: content ending
+			// exactly at a doubled capacity can observe EOF in the spare byte
+			// rather than forcing one more doubling.
+			grown := make([]byte, len(buf), 2*cap(buf)+1)
+			copy(grown, buf)
+			buf = grown
+		}
+		n, err := r.Read(buf[len(buf):cap(buf)])
+		buf = buf[:len(buf)+n]
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			}
+			// The error path skips the copy-down: the only caller discards
+			// the buffer when err != nil, and io.ReadAll's contract covers
+			// contents and error, not capacity.
+			if err == nil && cap(buf)-len(buf) > maxReturnWaste {
+				buf = append(make([]byte, 0, len(buf)), buf...)
+			}
+			return buf, err
+		}
+	}
 }
 
 func (t *toTheEndScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) {
 	if t.r == nil {
 		return nil, io.EOF
 	}
-	mBytes, err := io.ReadAll(t.r)
+	mBytes, err := readAllHinted(t.r, t.sizeHint)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/impl/pure/scanner_to_the_end_internal_test.go b/internal/impl/pure/scanner_to_the_end_internal_test.go
new file mode 100644
index 000000000..02abecc6d
--- /dev/null
+++ b/internal/impl/pure/scanner_to_the_end_internal_test.go
@@ -0,0 +1,281 @@
+// Copyright 2026 Redpanda Data, Inc.
+
+package pure
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"math"
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// shortReader wraps a reader and caps every Read at max bytes, emulating the
+// short reads returned by real files and sockets. A bytes.Reader satisfies the
+// whole request in a single call and so never exercises the refill path.
+type shortReader struct {
+	r   io.Reader
+	max int
+}
+
+func (s *shortReader) Read(p []byte) (int, error) {
+	if len(p) > s.max {
+		p = p[:s.max]
+	}
+	return s.r.Read(p)
+}
+
+func randomBytes(n int) []byte {
+	b := make([]byte, n)
+	// Fixed seed, so a failure reproduces exactly.
+	rnd := rand.New(rand.NewSource(1))
+	_, _ = rnd.Read(b)
+	return b
+}
+
+// TestReadAllHintedMatchesReadAll asserts that readAllHinted is a drop in
+// replacement for io.ReadAll for every hint value, correct or otherwise. The
+// hint is an optimisation and must never be a correctness input, as a file can
+// be appended to or truncated between Stat and read.
+func TestReadAllHintedMatchesReadAll(t *testing.T) {
+	sizes := []int{0, 1, 511, 512, 513, 4096, 1 << 20}
+
+	for _, size := range sizes {
+		content := randomBytes(size)
+
+		hints := []int64{
+			0,
+			1,
+			int64(size / 2),
+			int64(size - 1),
+			int64(size),
+			int64(size + 1),
+			int64(size * 2),
+			-5,
+		}
+
+		for _, hint := range hints {
+			for _, chunk := range []int{1, 7, 512, 1 << 16} {
+				t.Run(fmt.Sprintf("size=%v/hint=%v/chunk=%v", size, hint, chunk), func(t *testing.T) {
+					exp, err := io.ReadAll(&shortReader{r: bytes.NewReader(content), max: chunk})
+					require.NoError(t, err)
+
+					act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: chunk}, hint)
+					require.NoError(t, err)
+
+					assert.Equal(t, exp, act)
+					assert.Len(t, act, size)
+				})
+			}
+		}
+	}
+}
+
+// TestReadAllHintedPropagatesErrors asserts that a non-EOF error is returned
+// rather than being swallowed, matching io.ReadAll.
+func TestReadAllHintedPropagatesErrors(t *testing.T) {
+	expErr := fmt.Errorf("nope")
+
+	_, err := readAllHinted(io.MultiReader(
+		bytes.NewReader([]byte("partial")),
+		&errReader{err: expErr},
+	), 1024)
+	require.ErrorIs(t, err, expErr)
+}
+
+type errReader struct{ err error }
+
+func (e *errReader) Read([]byte) (int, error) { return 0, e.err }
+
+// TestReadAllHintedExactHintDoesNotRealloc asserts that an accurate hint at or
+// above the default floor results in exactly one allocation, by checking the
+// returned buffer still has the capacity it was created with. This is the test
+// that catches a regression of the +1 subtlety: sizing the buffer to exactly the
+// content length would find len == cap on the final iteration and grow once more.
+func TestReadAllHintedExactHintDoesNotRealloc(t *testing.T) {
+	// Sizes are >= defaultReadAllCap so the floor does not apply and the exact
+	// hint is honoured; the sub-floor case is covered by TestReadAllHintedFloor.
+	for _, size := range []int{defaultReadAllCap, 4096, 1 << 20} {
+		t.Run(fmt.Sprintf("size=%v", size), func(t *testing.T) {
+			content := randomBytes(size)
+
+			act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, int64(size))
+			require.NoError(t, err)
+
+			assert.Equal(t, content, act)
+			assert.Len(t, act, size)
+			assert.Equal(t, size+1, cap(act), "buffer was reallocated despite an exact hint")
+		})
+	}
+}
+
+// TestHintedCap asserts the hint is floored at the io.ReadAll default and
+// clamped at the pre-allocation cap, so a bogus value (up to and including
+// math.MaxInt64, where an unguarded +1 would overflow) can neither panic
+// make() nor spike memory. The clamping is unit tested in isolation here as
+// well as through readAllHinted below.
+func TestHintedCap(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		hint int64
+		exp  int64
+	}{
+		{"negative is floored", -5, defaultReadAllCap + 1},
+		{"zero is floored", 0, defaultReadAllCap + 1},
+		{"below default is floored", 100, defaultReadAllCap + 1},
+		{"exactly default", defaultReadAllCap, defaultReadAllCap + 1},
+		{"above default is honoured", 4096, 4097},
+		{"exactly the clamp", maxPreallocHint, maxPreallocHint + 1},
+		{"above the clamp", maxPreallocHint + 1, maxPreallocHint + 1},
+		{"max int64 does not overflow", math.MaxInt64, maxPreallocHint + 1},
+		{"max int64 minus one does not overflow", math.MaxInt64 - 1, maxPreallocHint + 1},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			got := hintedCap(test.hint)
+			assert.Equal(t, test.exp, got)
+			assert.Positive(t, got, "capacity must stay positive (no overflow)")
+		})
+	}
+}
+
+// TestReadAllHintedHugeHintDoesNotPanic asserts that an absurdly large hint —
+// a procfs file reporting terabytes, a custom filesystem returning garbage, or
+// a value for which make(0, hint+1) would panic outright — is clamped rather
+// than trusted. Previously io.ReadAll simply ignored such values by growing
+// incrementally, so this must never become a new failure mode.
+func TestReadAllHintedHugeHintDoesNotPanic(t *testing.T) {
+	content := randomBytes(4096)
+
+	for _, hint := range []int64{
+		maxPreallocHint + 1,
+		1 << 40,
+		1 << 62,
+		math.MaxInt64,
+	} {
+		t.Run(fmt.Sprintf("hint=%v", hint), func(t *testing.T) {
+			act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, hint)
+			require.NoError(t, err)
+			assert.Equal(t, content, act)
+			assert.LessOrEqual(t, cap(act), maxPreallocHint+1, "hint was not clamped")
+		})
+	}
+}
+
+// TestReadAllHintedOverHintReleasesExcess asserts that a hint which grossly
+// overestimates the content (a file truncated between Stat and read, or a
+// misreporting filesystem) does not leave the returned slice pinning the
+// whole pre-allocated array for the lifetime of the message.
+func TestReadAllHintedOverHintReleasesExcess(t *testing.T) {
+	const size = 4096
+	content := randomBytes(size)
+
+	act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, 1<<20)
+	require.NoError(t, err)
+	assert.Equal(t, content, act)
+	assert.Less(t, cap(act), size*2+1, "over-hinted buffer was not copied down")
+}
+
+// TestReadAllHintedFloor asserts that a small positive hint still allocates
+// the io.ReadAll-sized buffer rather than growing up from a tiny one, so a
+// stale hint for a source that has since grown keeps io.ReadAll's profile.
+// (A hint of zero or below delegates to io.ReadAll itself, covered by the
+// differential test above.)
+func TestReadAllHintedFloor(t *testing.T) {
+	const contentLen = 10 // < defaultReadAllCap, so it fits without any growth
+	for _, hint := range []int64{1, 100, defaultReadAllCap} {
+		t.Run(fmt.Sprintf("hint=%v", hint), func(t *testing.T) {
+			content := randomBytes(contentLen)
+
+			act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 4}, hint)
+			require.NoError(t, err)
+
+			assert.Equal(t, content, act)
+			assert.Equal(t, defaultReadAllCap+1, cap(act),
+				"sub-floor hint should allocate the default-sized buffer")
+		})
+	}
+}
+
+// TestReadAllHintedGrowsWhenHintTooSmall asserts the buffer still grows to fit
+// content larger than the hint, the case where a file is appended to between
+// Stat and read.
+func TestReadAllHintedGrowsWhenHintTooSmall(t *testing.T) {
+	const size = 1 << 20
+	content := randomBytes(size)
+
+	act, err := readAllHinted(&shortReader{r: bytes.NewReader(content), max: 512}, 1024)
+	require.NoError(t, err)
+	assert.Equal(t, content, act)
+}
+
+//------------------------------------------------------------------------------
+
+const benchSize = 64 * 1024 * 1024
+
+// repeatReader yields n bytes in fixed size short reads without allocating a
+// backing buffer of its own, so the benchmark measures only the read strategy.
+type repeatReader struct {
+	remaining int
+	chunk     int
+}
+
+func (c *repeatReader) Read(p []byte) (int, error) {
+	if c.remaining == 0 {
+		return 0, io.EOF
+	}
+	n := min(min(len(p), c.chunk), c.remaining)
+	for i := range p[:n] {
+		p[i] = 'x'
+	}
+	c.remaining -= n
+	return n, nil
+}
+
+func BenchmarkToTheEndReadAll(b *testing.B) {
+	b.ReportAllocs()
+	b.SetBytes(benchSize)
+	for b.Loop() {
+		buf, err := io.ReadAll(&repeatReader{remaining: benchSize, chunk: 32 * 1024})
+		if err != nil {
+			b.Fatal(err)
+		}
+		if len(buf) != benchSize {
+			b.Fatalf("unexpected length %v", len(buf))
+		}
+	}
+}
+
+func BenchmarkToTheEndReadAllHinted(b *testing.B) {
+	b.ReportAllocs()
+	b.SetBytes(benchSize)
+	for b.Loop() {
+		buf, err := readAllHinted(&repeatReader{remaining: benchSize, chunk: 32 * 1024}, benchSize)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if len(buf) != benchSize {
+			b.Fatalf("unexpected length %v", len(buf))
+		}
+	}
+}
+
+// BenchmarkToTheEndReadAllUnhinted guards the no-hint path (every source that
+// isn't a regular file), which must stay on par with io.ReadAll rather than
+// growing a buffer from scratch.
+func BenchmarkToTheEndReadAllUnhinted(b *testing.B) {
+	b.ReportAllocs()
+	b.SetBytes(benchSize)
+	for b.Loop() {
+		buf, err := readAllHinted(&repeatReader{remaining: benchSize, chunk: 32 * 1024}, 0)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if len(buf) != benchSize {
+			b.Fatalf("unexpected length %v", len(buf))
+		}
+	}
+}
diff --git a/internal/impl/pure/scanner_to_the_end_test.go b/internal/impl/pure/scanner_to_the_end_test.go
index d94d9cbf2..490be8f4a 100644
--- a/internal/impl/pure/scanner_to_the_end_test.go
+++ b/internal/impl/pure/scanner_to_the_end_test.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package pure_test
 
@@ -54,6 +54,89 @@ test:
 	assert.True(t, acked)
 }
 
+// TestToTheEndScannerSizeHints asserts that the content returned is unchanged
+// regardless of the size hint supplied, including hints that are absent, wrong
+// in either direction, or negative. The hint is an optimisation only.
+func TestToTheEndScannerSizeHints(t *testing.T) {
+	confSpec := service.NewConfigSpec().Field(service.NewScannerField("test"))
+	pConf, err := confSpec.ParseYAML(`
+test:
+  to_the_end: {}
+`, nil)
+	require.NoError(t, err)
+
+	rdr, err := pConf.FieldScanner("test")
+	require.NoError(t, err)
+
+	const content = `firstXsecondXthird`
+
+	for _, test := range []struct {
+		name    string
+		details func() *service.ScannerSourceDetails
+	}{
+		{
+			name:    "no details at all",
+			details: func() *service.ScannerSourceDetails { return nil },
+		},
+		{
+			name:    "details without a size",
+			details: func() *service.ScannerSourceDetails { return service.NewScannerSourceDetails() },
+		},
+		{
+			name: "exact size",
+			details: func() *service.ScannerSourceDetails {
+				d := service.NewScannerSourceDetails()
+				d.SetSizeHint(int64(len(content)))
+				return d
+			},
+		},
+		{
+			name: "size too small",
+			details: func() *service.ScannerSourceDetails {
+				d := service.NewScannerSourceDetails()
+				d.SetSizeHint(3)
+				return d
+			},
+		},
+		{
+			name: "size too large",
+			details: func() *service.ScannerSourceDetails {
+				d := service.NewScannerSourceDetails()
+				d.SetSizeHint(int64(len(content)) * 100)
+				return d
+			},
+		},
+		{
+			name: "negative size",
+			details: func() *service.ScannerSourceDetails {
+				d := service.NewScannerSourceDetails()
+				d.SetSizeHint(-5)
+				return d
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			buf := bytes.NewReader([]byte(content))
+			strm, err := rdr.Create(io.NopCloser(buf), func(ctx context.Context, err error) error {
+				return nil
+			}, test.details())
+			require.NoError(t, err)
+
+			m, _, err := strm.NextBatch(t.Context())
+			require.NoError(t, err)
+			require.Len(t, m, 1)
+
+			mBytes, err := m[0].AsBytes()
+			require.NoError(t, err)
+			assert.Equal(t, content, string(mBytes))
+
+			_, _, err = strm.NextBatch(t.Context())
+			require.Equal(t, io.EOF, err)
+			require.NoError(t, strm.Close(t.Context()))
+		})
+	}
+}
+
 func TestToTheEndScannerSuite(t *testing.T) {
 	confSpec := service.NewConfigSpec().Field(service.NewScannerField("test"))
 	pConf, err := confSpec.ParseYAML(`
diff --git a/public/service/codec/scanner.go b/public/service/codec/scanner.go
index ed7487b0e..7ad708268 100644
--- a/public/service/codec/scanner.go
+++ b/public/service/codec/scanner.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package codec
 
@@ -118,10 +118,11 @@ type codecRPublic struct {
 }
 
 func (r *codecRPublic) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (DeprecatedFallbackStream, error) {
-	sDetails := service.NewScannerSourceDetails()
-	sDetails.SetName(details.Name())
-
-	os, err := r.newCtor.Create(rdr, aFn, sDetails)
+	// The details are passed through wholesale rather than copied field by
+	// field: a copy silently drops any field this layer doesn't know about
+	// (which is how SizeHint was originally lost here), and the downstream
+	// Create is nil-safe.
+	os, err := r.newCtor.Create(rdr, aFn, details)
 	if err != nil {
 		return nil, err
 	}
diff --git a/public/service/scanner.go b/public/service/scanner.go
index c512b7d61..27c8d6d47 100644
--- a/public/service/scanner.go
+++ b/public/service/scanner.go
@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 
 package service
 
@@ -28,16 +28,50 @@ func NewScannerSourceDetails() *ScannerSourceDetails {
 }
 
 // SetName sets a filename (or other equivalent name of the source) to details.
+// Unlike the getters, calling this on a nil receiver panics.
 func (r *ScannerSourceDetails) SetName(name string) {
 	r.details.Name = name
 }
 
 // Name returns a filename (or other equivalent name of the source), or an
-// empty string if it has not been set.
+// empty string if it has not been set. It is safe to call on a nil receiver,
+// as details are optional.
 func (r *ScannerSourceDetails) Name() string {
+	if r == nil {
+		return ""
+	}
 	return r.details.Name
 }
 
+// SetSizeHint sets the total size in bytes of the source to details. This is a
+// hint only, used by scanner implementations in order to pre-allocate buffers,
+// and must never be relied upon for correctness as the underlying source may
+// change between it being measured and read. A zero size is indistinguishable
+// from the hint not being set at all, so implementations treat zero as
+// unknown.
+//
+// A scanner that wraps a child scanner and materially transforms the length
+// of the stream (such as decompress) must clear the hint before forwarding
+// details, as it describes the stream the wrapper reads, not the one the
+// child does. A wrapper that shortens the stream by only a bounded few bytes
+// (such as skip_bom) may forward the hint unchanged, as it remains a tight
+// upper bound.
+//
+// Unlike the getters, calling this on a nil receiver panics.
+func (r *ScannerSourceDetails) SetSizeHint(size int64) {
+	r.details.SizeHint = size
+}
+
+// SizeHint returns the total size in bytes of the source, or zero if it is
+// unknown or has not been set. It is safe to call on a nil receiver, as
+// details are optional.
+func (r *ScannerSourceDetails) SizeHint() int64 {
+	if r == nil {
+		return 0
+	}
+	return r.details.SizeHint
+}
+
 // BatchScannerCreator is an interface implemented by Benthos scanner plugins.
 // Calls to Create must create a new instantiation of BatchScanner that consumes
 // the provided io.ReadCloser, produces batches of messages (batches containing
diff --git a/public/service/scanner_test.go b/public/service/scanner_test.go
new file mode 100644
index 000000000..0fc1c9367
--- /dev/null
+++ b/public/service/scanner_test.go
@@ -0,0 +1,19 @@
+// Copyright 2026 Redpanda Data, Inc.
+
+package service
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestScannerSourceDetailsNilGetters asserts that the getters are safe to call
+// on a nil receiver, returning zero values. Details are optional throughout
+// the scanner APIs, so a nil pointer must behave like empty details.
+func TestScannerSourceDetailsNilGetters(t *testing.T) {
+	var details *ScannerSourceDetails
+
+	assert.Empty(t, details.Name())
+	assert.Zero(t, details.SizeHint())
+}