This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 5dd4b01c fix(internal/utils): handle reader reset and short peeks
(#987)
5dd4b01c is described below
commit 5dd4b01c4e2b985289073aa8491a402718147ac4
Author: Minh Vu <[email protected]>
AuthorDate: Thu Jul 23 18:06:27 2026 +0200
fix(internal/utils): handle reader reset and short peeks (#987)
### Rationale for this change
The internal buffered reader can retain a deferred error when an
underlying `Read` returns data and an error together. `Reset` replaced
the underlying reader but left that error pending, so the next read
could return the old error without touching the new input.
`Peek` also returned the requested-length slice when input ended early,
exposing unused buffer bytes instead of only the bytes that were read.
### What changes are included in this PR?
Clear pending read errors in `Reset`, and clamp short `Peek` results to
the number of buffered bytes. Regression tests cover reuse after a
data-plus-error read and a partial peek at EOF.
### Are these changes tested?
Yes:
- `go test ./internal/utils`
- `go test -race -count=1 ./internal/utils`
### Are there any user-facing changes?
Buffered Parquet reads no longer carry an error across reader reuse or
expose unused bytes from a short peek. There is no public API change.
---
internal/utils/buf_reader.go | 3 +-
internal/utils/buf_reader_test.go | 68 +++++++++++++++++++++++++++++++++++++++
2 files changed, 70 insertions(+), 1 deletion(-)
diff --git a/internal/utils/buf_reader.go b/internal/utils/buf_reader.go
index c222c8bd..40a7af8d 100644
--- a/internal/utils/buf_reader.go
+++ b/internal/utils/buf_reader.go
@@ -136,6 +136,7 @@ func (b *bufferedReader) Reset(rd Reader) {
b.resetBuffer()
b.rd = rd
b.r, b.w = 0, 0
+ b.err = nil
}
func (b *bufferedReader) resetBuffer() {
@@ -222,7 +223,7 @@ func (b *bufferedReader) Peek(n int) ([]byte, error) {
b.fill() // b.w-b.r < len(b.buf) => buffer is not full
}
- return b.buf[b.r : b.r+n], b.readErr()
+ return b.buf[b.r : b.r+min(n, b.w-b.r)], b.readErr()
}
// Discard skips the next n bytes either by advancing the internal buffer
diff --git a/internal/utils/buf_reader_test.go
b/internal/utils/buf_reader_test.go
new file mode 100644
index 00000000..e22b80dc
--- /dev/null
+++ b/internal/utils/buf_reader_test.go
@@ -0,0 +1,68 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package utils
+
+import (
+ "bytes"
+ "errors"
+ "io"
+ "testing"
+)
+
+type readerReturningDataAndError struct {
+ *bytes.Reader
+ err error
+ returned bool
+}
+
+func (r *readerReturningDataAndError) Read(p []byte) (int, error) {
+ if r.returned {
+ return 0, r.err
+ }
+ r.returned = true
+ n, _ := r.Reader.Read(p)
+ return n, r.err
+}
+
+func TestBufferedReaderResetClearsPendingError(t *testing.T) {
+ r := NewBufferedReader(&readerReturningDataAndError{
+ Reader: bytes.NewReader([]byte("a")),
+ err: io.ErrUnexpectedEOF,
+ }, 2)
+
+ buf := make([]byte, 1)
+ if n, err := r.Read(buf); n != 1 || err != nil || string(buf) != "a" {
+ t.Fatalf("initial read = (%d, %v, %q), want (1, nil, %q)", n,
err, buf, "a")
+ }
+
+ r.Reset(bytes.NewReader([]byte("b")))
+ if n, err := r.Read(buf); n != 1 || err != nil || string(buf) != "b" {
+ t.Fatalf("read after reset = (%d, %v, %q), want (1, nil, %q)",
n, err, buf, "b")
+ }
+}
+
+func TestBufferedReaderPeekReturnsAvailableBytesOnError(t *testing.T) {
+ r := NewBufferedReader(bytes.NewReader([]byte("a")), 2)
+
+ got, err := r.Peek(2)
+ if !errors.Is(err, io.EOF) {
+ t.Fatalf("Peek error = %v, want EOF", err)
+ }
+ if string(got) != "a" {
+ t.Fatalf("Peek = %q, want %q", got, "a")
+ }
+}
