GH-37429: [C++] Add arrow::ipc::StreamDecoder::Reset()

[kou]

Rationale for this change

We can reuse the same StreamDecoder to read multiple streams with this.

What changes are included in this PR?

Add StreamDecoder::Reset().

Are these changes tested?

Yes.

Are there any user-facing changes?

Yes.

• Closes: [C++] Make public version of StreamDecoderImpl-like class #37429

[github-actions]

⚠️ GitHub issue #37429 has been automatically assigned in GitHub to PR creator.

[pitrou]

Honestly, I don't think the endless decoder is very useful (why would I want an endless decoder?). Perhaps we can simply remove it?

[kou]

OK. I'll remove this example.

See #37429 for a use case of the endless decoder. It reads multiple streams with one decoder instance.

FYI: We can implement the use case by using next_required_size() instead but we need to recreate decoder instances manually:

while (true) {
  auto buffer = get_data(decoder->next_required_size());
  if (!buffer) {
    break;
  }
  decoder->Consume(buffer);
  if (listener->status() == EOS) {
    decoder = create_decoder(listener);
  }
}

[pitrou]

Nit: std::move on a rvalue is probably a no-op.

[kou]

Oh, sorry. I'll remove it.

[MrLolthe1st]

@kou That's not all: after return from OnEOS() and from OnEOS() at StreamDecoder, we'll back in old MessageReader, but after Reset() call, we've destroyed it. Since old MessageReader hadn't been reseted, MessageReader still in EOS state and will ignore all payload after EOS message.

[MrLolthe1st]

You can checkout our temporary workaround here:
stream_decoder.cc ydb-platform/ydb@f288403#diff-89894a6bbc78b5f2aacc62338e7cbfbd6a7a841dd14bb44eed25fd5dd151dc3a
stream_decoder.h ydb-platform/ydb@f288403#diff-a442528f4921f6781d036840f10ce3faf49990cdbba0a88bffb7c8832592ec39
There some changes around MessageReader too. Also, dictionary_memo_ is now unique pointer: dictionary_memo_ = std::make_unique<DictionaryMemo>();

[kou]

Why do we need to care about MessageReader?
This implementation re-creates StreamDecoderImpl instead of reusing existing StreamDecoderImpl (including MessageReader).

[MrLolthe1st]

Why do we need to care about MessageReader? This implementation re-creates StreamDecoderImpl instead of reusing existing StreamDecoderImpl (including MessageReader).

Because when we got EOS in the middle of payload.
Call stack when we've got EOS looks like that:
Listener::OnEOS(..)
StreamDecoderImpl::OnEOS()
MessageDecoderImpl::OnEOS() <- that is old message decoder, that already is in EOS state. after exiting method above, we will return here.
...
MessageDecoderImpl::Consume(...)
MessageDecoder::Consume(...)
StreamDecoderImpl::Consume(...)

So, if we recreate StreamDecoderImpl in Listener::OnEOS(), after exiting Listener::OnEOS we will

1. Return to old StreamDecoderImpl, that isn't exists.
2. Than return to old MessageDocederImpl, that isn't exists. But, if we're somehow save old message decoder, it is in state EOS and will ignore all messages, that left in payload, because of EOS state.

[kou]

Ah, I understand.
I've fixed the case by splitting the given data. If StreamDecoderImpl::Consume() is finished for each state, the problem isn't happen.

[MrLolthe1st]

Looks like truth!

[kou]

I'll merge this in a few days if nobody objects it.

[kou]

No objection.
I'll merge this.

[conbench-apache-arrow]

After merging your PR, Conbench analyzed the 6 benchmarking runs that have been run so far on merge-commit cead3dd.

There were no benchmark performance regressions. 🎉

The full Conbench report has more details. It also includes information about 32 possible false positives for unstable benchmarks that are known to sometimes produce them.