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 e63018eb fix(arrow/ipc): propagate CLI stream errors (#986)
e63018eb is described below
commit e63018eb7266599e5e1af166111e79d0ab1e2cd0
Author: Minh Vu <[email protected]>
AuthorDate: Thu Jul 23 18:09:31 2026 +0200
fix(arrow/ipc): propagate CLI stream errors (#986)
### Rationale for this change
`arrow-cat` and `arrow-ls` stopped when `Reader.Next` returned false but
did not inspect `Reader.Err`. A record read failure could therefore be
reported as a successful end of stream. `arrow-cat` could exit
successfully after partial output, while `arrow-ls` could print a record
count for an incomplete stream.
### What changes are included in this PR?
Check `Reader.Err` after iteration in both commands and return it after
releasing the reader. `arrow-ls` now prints the record count only after
the stream completes successfully. Regression tests use a valid schema
followed by truncated record metadata.
### Are these changes tested?
Yes:
- `go test ./arrow/ipc/cmd/arrow-cat ./arrow/ipc/cmd/arrow-ls`
- `go test -race -count=1 ./arrow/ipc/cmd/arrow-cat
./arrow/ipc/cmd/arrow-ls`
### Are there any user-facing changes?
Both command-line tools now return an error for a failed record read
instead of reporting success. Valid stream output is unchanged.
---
arrow/ipc/cmd/arrow-cat/main.go | 4 ++++
arrow/ipc/cmd/arrow-cat/main_test.go | 28 ++++++++++++++++++++++++++++
arrow/ipc/cmd/arrow-ls/main.go | 7 +++++--
arrow/ipc/cmd/arrow-ls/main_test.go | 28 ++++++++++++++++++++++++++++
4 files changed, 65 insertions(+), 2 deletions(-)
diff --git a/arrow/ipc/cmd/arrow-cat/main.go b/arrow/ipc/cmd/arrow-cat/main.go
index ca6cf886..9c1fcca1 100644
--- a/arrow/ipc/cmd/arrow-cat/main.go
+++ b/arrow/ipc/cmd/arrow-cat/main.go
@@ -105,7 +105,11 @@ func processStream(w io.Writer, rin io.Reader) error {
fmt.Fprintf(w, " col[%d] %q: %v\n", i,
rec.ColumnName(i), col)
}
}
+ err = r.Err()
r.Release()
+ if err != nil {
+ return err
+ }
}
return nil
}
diff --git a/arrow/ipc/cmd/arrow-cat/main_test.go
b/arrow/ipc/cmd/arrow-cat/main_test.go
index 0e25ee64..78b3853c 100644
--- a/arrow/ipc/cmd/arrow-cat/main_test.go
+++ b/arrow/ipc/cmd/arrow-cat/main_test.go
@@ -223,6 +223,34 @@ record 3...
}
}
+func TestCatStreamReturnsRecordReadError(t *testing.T) {
+ rec := arrdata.Records["primitives"][0]
+
+ var schemaOnly bytes.Buffer
+ schemaWriter := ipc.NewWriter(&schemaOnly, ipc.WithSchema(rec.Schema()))
+ if err := schemaWriter.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ var full bytes.Buffer
+ writer := ipc.NewWriter(&full, ipc.WithSchema(rec.Schema()))
+ if err := writer.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ if err := writer.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ const eosSize = 8
+ schemaSize := schemaOnly.Len() - eosSize
+ // Keep the record framing and one metadata byte so the record read
fails
+ // with io.ErrUnexpectedEOF instead of being treated as a clean stream
end.
+ truncated := full.Bytes()[:schemaSize+eosSize+1]
+ if err := processStream(io.Discard, bytes.NewReader(truncated)); err ==
nil {
+ t.Fatal("processStream returned nil for a truncated record
message")
+ }
+}
+
func TestCatFile(t *testing.T) {
tempDir := t.TempDir()
diff --git a/arrow/ipc/cmd/arrow-ls/main.go b/arrow/ipc/cmd/arrow-ls/main.go
index d329d0f4..cc4f00b2 100644
--- a/arrow/ipc/cmd/arrow-ls/main.go
+++ b/arrow/ipc/cmd/arrow-ls/main.go
@@ -101,10 +101,13 @@ func processStream(w io.Writer, rin io.Reader) error {
for r.Next() {
nrecs++
}
- fmt.Fprintf(w, "records: %d\n", nrecs)
+ err = r.Err()
r.Release()
+ if err != nil {
+ return err
+ }
+ fmt.Fprintf(w, "records: %d\n", nrecs)
}
- return nil
}
func processFiles(w io.Writer, names []string) error {
diff --git a/arrow/ipc/cmd/arrow-ls/main_test.go
b/arrow/ipc/cmd/arrow-ls/main_test.go
index f90e4a80..cb1ab8fd 100644
--- a/arrow/ipc/cmd/arrow-ls/main_test.go
+++ b/arrow/ipc/cmd/arrow-ls/main_test.go
@@ -168,6 +168,34 @@ records: 3
}
}
+func TestLsStreamReturnsRecordReadError(t *testing.T) {
+ rec := arrdata.Records["primitives"][0]
+
+ var schemaOnly bytes.Buffer
+ schemaWriter := ipc.NewWriter(&schemaOnly, ipc.WithSchema(rec.Schema()))
+ if err := schemaWriter.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ var full bytes.Buffer
+ writer := ipc.NewWriter(&full, ipc.WithSchema(rec.Schema()))
+ if err := writer.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ if err := writer.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ const eosSize = 8
+ schemaSize := schemaOnly.Len() - eosSize
+ // Keep the record framing and one metadata byte so the record read
fails
+ // with io.ErrUnexpectedEOF instead of being treated as a clean stream
end.
+ truncated := full.Bytes()[:schemaSize+eosSize+1]
+ if err := processStream(io.Discard, bytes.NewReader(truncated)); err ==
nil {
+ t.Fatal("processStream returned nil for a truncated record
message")
+ }
+}
+
func TestLsFile(t *testing.T) {
tempDir := t.TempDir()
