Implement file processor by henrikschristensen · Pull Request #690 · warpstreamlabs/bento

henrikschristensen · 2026-02-09T13:36:21Z

resolves #649

internal/impl/io/processor_file.go

jem-davies

I see that you have altered it to draft after I started to take a look...

But as well as the comments - what about thinking about the shutdown procedure - i.e. reading from a big file and Bento is shutdown - will that happen gracefully -

What about the atomicity of the operations too?

henrikschristensen · 2026-02-21T16:49:29Z

@jem-davies should the read method use a scanner like the file input? I seems that scanner impl handles the prob with large files. Reading in chunks etc.

jem-davies · 2026-02-21T16:51:25Z

@jem-davies should the read method use a scanner like the file input? I seems that scanner impl handles the prob with large files. Reading in chunks etc.

Yeah that's what I was thinking 👍

henrikschristensen · 2026-02-23T17:04:42Z

For gracefull shutdown during large file reads i changed the read method to use a scanner.
For atomicity of write and move i implemented steps like using temporary intermediate files to make sure we dont alter data if we fail during the operation.

jem-davies · 2026-03-23T14:53:49Z

internal/impl/io/processor_file.go

+				Description("The file operation to perform.").
+				Examples("read", "write", "delete", "move", "rename", "stat"),


Suggested change

Description("The file operation to perform.").

Examples("read", "write", "delete", "move", "rename", "stat"),

Description("The file operation to perform."),

nit: remove these examples - the options are already displayed on the docs page.

jem-davies · 2026-03-23T14:54:53Z

internal/impl/io/processor_file.go

+				Description("The file operation to perform.").
+				Examples("read", "write", "delete", "move", "rename", "stat"),
+			service.NewInterpolatedStringField(fileProcessorFieldPath).
+				Description("The source file path. Supports interpolation for dynamic paths.").


Suggested change

Description("The source file path. Supports interpolation for dynamic paths.").

Description("The source file path.").

It is already mentioned automatically for NewInterpolatedStringFields.

jem-davies · 2026-03-23T14:55:08Z

internal/impl/io/processor_file.go

+					"/tmp/${! json(\"document.id\") }.txt",
+				).LintRule(`if this == "" { [ "'path' must be set to a non-empty string" ] }`),
+			service.NewInterpolatedStringField(fileProcessorFieldDest).
+				Description("The destination path for 'move' and 'rename' operations. Supports interpolation.").


Suggested change

Description("The destination path for 'move' and 'rename' operations. Supports interpolation.").

Description("The destination path for 'move' and 'rename' operations.").

jem-davies · 2026-03-23T14:56:27Z

internal/impl/io/processor_file.go

+	fileProcessorFieldDest      = "dest"
+	fileProcessorFieldScanner   = "scanner"
+)
+


nit: Add const for the operations options - and then reference them later

jem-davies · 2026-03-23T14:57:25Z

internal/impl/io/processor_file.go

+	fileProcessorFieldPath      = "path"
+	fileProcessorFieldDest      = "dest"


Suggested change

fileProcessorFieldPath = "path"

fileProcessorFieldDest = "dest"

fileProcessorFieldPath = "source_path"

fileProcessorFieldDest = "destination_path"

nit

jem-davies · 2026-03-23T15:23:40Z

internal/impl/io/processor_file.go

+		if err := ackFn(ctx, nil); err != nil {
+			p.log.Warnf("Failed to acknowledge scanner batch: %v", err)
+		}


Suggested change

if err := ackFn(ctx, nil); err != nil {

p.log.Warnf("Failed to acknowledge scanner batch: %v", err)

}

Replace with the defer file.Close()

will all scanners move on to NextBatch if we fail to ack the one we are processing?

jem-davies · 2026-03-23T15:26:29Z

internal/impl/io/processor_file.go

+}
+
+func (p *fileProcessor) processWrite(_ context.Context, msg *service.Message) (service.MessageBatch, error) {
+	path, err := p.conf.Path.TryString(msg)


Just noting again regarding path / dest or source_path / destination_path - here we are writing to path and not dest - which is fine. But worth documenting this in the operations section of the docs imo.

jem-davies · 2026-03-23T15:27:35Z

internal/impl/io/processor_file.go

+	}
+
+	// Use atomic write pattern: write to temp file, then rename
+	tempFile := path + ".tmp"


Would add some sort of random prefix to ".tmp" - so to make sure we don't get a collision.

jem-davies · 2026-03-23T15:29:16Z

internal/impl/io/processor_file.go

+
+// atomicCopyAndDelete performs an atomic copy from src to dest and then deletes src
+// This ensures that either the operation completes fully or leaves the source intact
+func (p *fileProcessor) atomicCopyAndDelete(_ context.Context, srcPath, destPath string, msg *service.Message) (service.MessageBatch, error) {


We have context in the func signature but we aren't going to use it, and it's an unexported func so unclear why.

jem-davies · 2026-03-23T15:45:57Z

internal/impl/io/processor_file.go

+	}
+	defer srcFile.Close()
+
+	content, err := io.ReadAll(srcFile)


This is potentially going to lead to OOM issues - thinking that this could be re-worked using a streaming approach to writing to the tempFile - and then we could also observe context cancellation.

henrikschristensen added 3 commits February 9, 2026 11:22

Merge branch 'warpstreamlabs:main' into main

afa09f6

Merge branch 'warpstreamlabs:main' into main

2667db9

Merge branch 'warpstreamlabs:main' into main

5c98bc9

henrikschristensen requested review from gregfurman and jem-davies as code owners February 9, 2026 13:36

henrikschristensen marked this pull request as draft February 18, 2026 16:34

jem-davies reviewed Feb 18, 2026

View reviewed changes

internal/impl/io/processor_file.go Outdated Show resolved Hide resolved

jem-davies reviewed Feb 18, 2026

View reviewed changes

internal/impl/io/processor_file.go Outdated Show resolved Hide resolved

jem-davies reviewed Feb 18, 2026

View reviewed changes

internal/impl/io/processor_file.go Show resolved Hide resolved

jem-davies reviewed Feb 18, 2026

View reviewed changes

internal/impl/io/processor_file.go Show resolved Hide resolved

jem-davies reviewed Feb 18, 2026

View reviewed changes

internal/impl/io/processor_file.go Show resolved Hide resolved

jem-davies reviewed Feb 18, 2026

View reviewed changes

henrikschristensen force-pushed the implement-file-processor branch 2 times, most recently from 2627bd5 to 53f70d5 Compare February 23, 2026 16:28

added file processor

6ec3a54

henrikschristensen force-pushed the implement-file-processor branch from 53f70d5 to 6ec3a54 Compare February 28, 2026 10:39

henrikschristensen marked this pull request as ready for review February 28, 2026 11:01

henrikschristensen requested a review from jem-davies March 2, 2026 18:03

jem-davies self-assigned this Mar 23, 2026

jem-davies reviewed Mar 23, 2026

View reviewed changes

jem-davies removed their assignment Mar 23, 2026

		Description("The file operation to perform.").
		Examples("read", "write", "delete", "move", "rename", "stat"),

	Description("The source file path. Supports interpolation for dynamic paths.").
	Description("The source file path.").

	Description("The destination path for 'move' and 'rename' operations. Supports interpolation.").
	Description("The destination path for 'move' and 'rename' operations.").

		fileProcessorFieldPath = "path"
		fileProcessorFieldDest = "dest"

	if err := ackFn(ctx, nil); err != nil {
	p.log.Warnf("Failed to acknowledge scanner batch: %v", err)
	}

Conversation

henrikschristensen commented Feb 9, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jem-davies left a comment

Choose a reason for hiding this comment

Uh oh!

henrikschristensen commented Feb 21, 2026

Uh oh!

jem-davies commented Feb 21, 2026

Uh oh!

henrikschristensen commented Feb 23, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants