zeroshade commented on issue #1325: URL: https://github.com/apache/arrow-go/issues/1325#issuecomment-5764896287
> Was this behavior change intended to ship in a minor release (v18.7 →
v18.8)? It's source-breaking for the reuse pattern, which reads unusual for a
minor bump.
Honestly, the use case you're describing was simply not a use case I was
aware of people doing since it was never intended and just accidentally worked.
Interestingly, this was asymmetric. The pattern you were using only ever
worked through `Read()` and wouldn't have worked if you were using `Next()`.
Essentially what happened wasn't that we intentionally broke "reuse the reader"
but rather closed an inconsistency. Basically, the guard was deliberate but
the effect on your particular usage pattern wasn't intended or even noticed.
That said, "worked in the last release, now silently returns zero rows with a
nil error" is a bad result regardless and something we should fix.
> If reuse is not supported, what's the recommended way to decode this "many
small streams, schema known once" pattern? Right now our options seem to be a
fresh Reader per blob plus re-injecting a serialized schema message in front of
every schema-less blob, or reaching into unexported state. Is there a better
approach we're missing?
The approach that works today, on both v18.7.x and v18.8.x, with no
unexported state: splice the blobs at the *message* level with a custom
`MessageReader` and
hand it to `ipc.NewReaderFromMessageReader`. That's what that constructor
exists for (it's how Flight feeds the reader). The reader never observes an EOF
between
blobs, so `done` never gets set:
```go
// concatMessageReader splices a sequence of independent IPC blobs into one
// logical message stream. Per-blob EOF is swallowed and the next blob
pulled
// in transparently; io.EOF surfaces only once every blob is drained.
type concatMessageReader struct {
next func() (io.Reader, error) // return io.EOF when no blobs remain
cur ipc.MessageReader
mem memory.Allocator
refCount atomic.Int64
}
func (c *concatMessageReader) Retain() { c.refCount.Add(1) }
func (c *concatMessageReader) Release() {
if c.refCount.Add(-1) == 0 && c.cur != nil {
c.cur.Release()
c.cur = nil
}
}
func (c *concatMessageReader) Message() (*ipc.Message, error) {
for {
if c.cur == nil {
r, err := c.next()
if err != nil {
return nil, err // io.EOF here means genuinely finished
}
c.cur = ipc.NewMessageReader(r, ipc.WithAllocator(c.mem))
}
msg, err := c.cur.Message()
if err == nil {
return msg, nil
}
if !errors.Is(err, io.EOF) {
// io.ErrUnexpectedEOF lands here: a truncated message is
// corruption, not a blob boundary. Do not swallow it.
return nil, err
}
// clean end of this blob; roll to the next one
c.cur.Release()
c.cur = nil
}
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
