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 e1faf3da fix(arrow/ipc): reject non-dictionary initial messages (#1054)
e1faf3da is described below
commit e1faf3dacc626308b35ed514d928b68fe54d25bb
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 20:08:27 2026 +0200
fix(arrow/ipc): reject non-dictionary initial messages (#1054)
### Rationale for this change
When an IPC schema declares dictionaries, the reader expects dictionary
batches before records. It detected an unexpected message but continued
into `readDictionary`, which could replace the useful protocol error and
consume the wrong message.
### What changes are included in this PR?
Stop immediately, mark the reader done, and preserve the
expected-dictionary error.
### Are these changes tested?
Yes. A regression test provides a dictionary schema followed directly by
a record batch.
`go test ./arrow/ipc -run
TestReaderRejectsRecordBatchBeforeInitialDictionary`
### Are there any user-facing changes?
Malformed streams now fail deterministically with the specific
dictionary-ordering error.
---
arrow/ipc/reader.go | 4 ++++
arrow/ipc/writer_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 48 insertions(+)
diff --git a/arrow/ipc/reader.go b/arrow/ipc/reader.go
index df9e5000..dc035564 100644
--- a/arrow/ipc/reader.go
+++ b/arrow/ipc/reader.go
@@ -224,6 +224,7 @@ func (r *Reader) getInitialDicts() bool {
if msg.Type() != MessageDictionaryBatch {
r.err = fmt.Errorf("arrow/ipc: IPC stream did not have
the expected (%d) dictionaries at the start of the stream", numDicts)
+ return false
}
if _, err := readDictionary(&r.memo, msg.meta, msg.body,
r.swapEndianness, r.mem); err != nil {
r.done = true
@@ -241,6 +242,9 @@ func (r *Reader) next() bool {
r.err = utils.FormatRecoveredError("arrow/ipc: unknown
error while reading", pErr)
}
}()
+ if r.err != nil || r.done {
+ return false
+ }
if r.schema == nil {
if err := r.readSchema(r.expectedSchema); err != nil {
r.err = fmt.Errorf("arrow/ipc: could not read schema
from stream: %w", err)
diff --git a/arrow/ipc/writer_test.go b/arrow/ipc/writer_test.go
index decef53a..6de7ee0d 100644
--- a/arrow/ipc/writer_test.go
+++ b/arrow/ipc/writer_test.go
@@ -359,6 +359,50 @@ func TestWritePayload(t *testing.T) {
require.True(t, msg.Type() == MessageRecordBatch)
}
+func TestReaderRejectsRecordBatchBeforeInitialDictionary(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int8,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ schema := arrow.NewSchema([]arrow.Field{{Name: "dict", Type:
dictType}}, nil)
+ bldr := array.NewBuilder(mem, dictType)
+ defer bldr.Release()
+ require.NoError(t, bldr.UnmarshalJSON([]byte(`["value"]`)))
+ arr := bldr.NewArray()
+ defer arr.Release()
+ rec := array.NewRecordBatch(schema, []arrow.Array{arr}, 1)
+ defer rec.Release()
+
+ var stream bytes.Buffer
+ schemaPayload := GetSchemaPayload(schema, mem)
+ defer schemaPayload.Release()
+ _, err := schemaPayload.WritePayload(&stream)
+ require.NoError(t, err)
+ recordPayload, err := GetRecordBatchPayload(rec, WithAllocator(mem))
+ require.NoError(t, err)
+ defer recordPayload.Release()
+ _, err = recordPayload.WritePayload(&stream)
+ require.NoError(t, err)
+ streamBytes := append([]byte(nil), stream.Bytes()...)
+
+ rdr, err := NewReader(bytes.NewReader(streamBytes), WithAllocator(mem))
+ require.NoError(t, err)
+ defer rdr.Release()
+ require.False(t, rdr.Next())
+ require.EqualError(t, rdr.Err(), "arrow/ipc: IPC stream did not have
the expected (1) dictionaries at the start of the stream")
+
+ rdr, err = NewReader(bytes.NewReader(streamBytes), WithAllocator(mem))
+ require.NoError(t, err)
+ defer rdr.Release()
+ _, err = rdr.Read()
+ require.EqualError(t, err, "arrow/ipc: IPC stream did not have the
expected (1) dictionaries at the start of the stream")
+ _, err = rdr.Read()
+ require.EqualError(t, err, "arrow/ipc: IPC stream did not have the
expected (1) dictionaries at the start of the stream")
+}
+
// TestVariadicCountsNotAccumulatedAcrossEncode verifies that variadicCounts
// does not accumulate across encode calls separated by reset(). Without this,
// each batch's variadic counts would include counts from previous batches,
