fix(scanner): pre-size the to_the_end read buffer from a source size …

.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: |

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

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

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()
+}

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,

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")
+}

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)
 }
 

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()))
+}

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)
 }