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 48a8340e fix(arrow/ipc): return compression errors from record encoder
(#1141)
48a8340e is described below
commit 48a8340efd5d1ca7038c5126ac9fdb4c709b444a
Author: Minh Vu <[email protected]>
AuthorDate: Fri Aug 14 19:30:29 2026 +0200
fix(arrow/ipc): return compression errors from record encoder (#1141)
### Rationale for this change
`recordEncoder.encode` ignored errors returned by body compression, so a
failed compressor could still result in record metadata being built and
returned as a successful encode.
### What changes are included in this PR?
Return compression errors from `recordEncoder.encode`, release the
temporary compressed buffer on failure, and add a regression test with a
failing compressor.
### Are these changes tested?
- `go test ./arrow/ipc`
### Are there any user-facing changes?
IPC record encoding now reports compressor failures to the caller.
---
arrow/ipc/ipc.go | 7 +++++++
arrow/ipc/writer.go | 4 +++-
arrow/ipc/writer_test.go | 33 ++++++++++++++++++++++++++++-----
3 files changed, 38 insertions(+), 6 deletions(-)
diff --git a/arrow/ipc/ipc.go b/arrow/ipc/ipc.go
index da9cf333..9a043d9b 100644
--- a/arrow/ipc/ipc.go
+++ b/arrow/ipc/ipc.go
@@ -75,6 +75,7 @@ type config struct {
minSpaceSavings float64
maxMetadataSize int64
maxBodySize int64
+ compressors []compressor
}
const (
@@ -86,6 +87,12 @@ const (
// and streams.
type Option func(*config)
+func withCompressors(compressors ...compressor) Option {
+ return func(cfg *config) {
+ cfg.compressors = compressors
+ }
+}
+
func newConfig(opts ...Option) *config {
cfg := &config{
alloc: memory.NewGoAllocator(),
diff --git a/arrow/ipc/writer.go b/arrow/ipc/writer.go
index 32edaf25..24340f44 100644
--- a/arrow/ipc/writer.go
+++ b/arrow/ipc/writer.go
@@ -1153,6 +1153,8 @@ func needTruncate(offset int64, buf *memory.Buffer,
minLength int64) bool {
// method after it is no longer needed.
func GetRecordBatchPayload(batch arrow.RecordBatch, opts ...Option) (Payload,
error) {
cfg := newConfig(opts...)
+ compressors := make([]compressor, cfg.compressNP)
+ copy(compressors, cfg.compressors)
var (
data = Payload{msg: MessageRecordBatch}
enc = newRecordEncoder(
@@ -1163,7 +1165,7 @@ func GetRecordBatchPayload(batch arrow.RecordBatch, opts
...Option) (Payload, er
cfg.codec,
cfg.compressNP,
cfg.minSpaceSavings,
- make([]compressor, cfg.compressNP),
+ compressors,
)
)
diff --git a/arrow/ipc/writer_test.go b/arrow/ipc/writer_test.go
index 957d812c..6d18f5f0 100644
--- a/arrow/ipc/writer_test.go
+++ b/arrow/ipc/writer_test.go
@@ -48,13 +48,19 @@ type failingPayloadWriter struct {
type shortWriteWriter struct{}
type failingCompressor struct {
- err error
+ err error
+ closeErr error
}
-func (failingCompressor) MaxCompressedLen(n int) int { return n }
-func (failingCompressor) Reset(io.Writer) {}
-func (f failingCompressor) Write([]byte) (int, error) { return 0, f.err }
-func (failingCompressor) Close() error { return nil }
+func (failingCompressor) MaxCompressedLen(n int) int { return n }
+func (failingCompressor) Reset(io.Writer) {}
+func (f failingCompressor) Write(p []byte) (int, error) {
+ if f.err != nil {
+ return 0, f.err
+ }
+ return len(p), nil
+}
+func (f failingCompressor) Close() error { return f.closeErr }
func (failingCompressor) Type() flatbuf.CompressionType {
return flatbuf.CompressionTypeZSTD
}
@@ -412,6 +418,23 @@ func TestRecordEncoderReturnsCompressionError(t
*testing.T) {
require.ErrorIs(t, encoder.Encode(&payload, record), want)
}
+func TestGetRecordBatchPayloadReturnsCompressionErrorOnClose(t *testing.T) {
+ want := errors.New("compression failed")
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ schema := arrow.NewSchema([]arrow.Field{{Name: "col", Type:
arrow.PrimitiveTypes.Int8}}, nil)
+ builder := array.NewRecordBuilder(mem, schema)
+ defer builder.Release()
+ builder.Field(0).(*array.Int8Builder).Append(1)
+ record := builder.NewRecordBatch()
+ defer record.Release()
+
+ _, err := GetRecordBatchPayload(record, WithAllocator(mem), WithZstd(),
+ withCompressors(failingCompressor{closeErr: want}))
+ require.ErrorIs(t, err, want)
+}
+
func TestWriteWithCompressionAndMinSavings(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
defer mem.AssertSize(t, 0)
