[BEAM-12384] Set output typeDescriptor explictly in Read.Bounded transform

[iemejia]

R: @boyuanzz @kennknowles

[boyuanzz]

Do we maintain the TypeDescriptor information before for Read? I was under impression that for most of cases we only set Coder for a output PCollection.

[iemejia]

You are right and I don't know why we don't pay more attention to this. Probably because coders seem to include the TypeDescriptor, any ideas @kennknowles ? is this redundant somehow?

In any case having this information seems important for the downstream transforms.

[boyuanzz]

It seems like the typeDescriptor can be inferred from Coder.getEncodedTypeDescriptor(). If we really want to populate this information in a consistent way, probably we can consider changing PCollection.getTypeDescriptor() to infer the typeDescriptor from Coder if the typeDescriptor is set.

[kennknowles]

Yes, the expected use of this method is to set the type descriptor but not the coder. This way, the coder registry still can choose the coder.

Setting both is redundant, in theory. Setting just the coder should suffice. Maybe some plumbing needed? It was not really expected to look at either one in this way.

Another angle to consider is that type descriptor is Java-specific, while coder is the portable "type" of the data. I don't know if that matters here.

[boyuanzz]

I'm thinking about changes like: #14870

[iemejia]

I like @boyuanzz fix because even in the presence of different Coders the TypeDescriptor is commonly preserved inside of the Coders. WDYT @kennknowles can you spot some particular issues about it?
I can rebase this PR targetting a generic implementation like the one on #14870 but I did not do it like that because I was not really familiar with the reasoning behind not relying on the coder typeDescriptor.

[kennknowles]

Yea makes lots of sense.

[boyuanzz]

Run Java PreCommit

[boyuanzz]

Run Java_Examples_Dataflow PreCommit

[boyuanzz]

Run Java_Examples_Dataflow_Java11 PreCommit

[boyuanzz]

Run Java PreCommit

[boyuanzz]

Run Java_Examples_Dataflow_Java11 PreCommit

[iemejia]

I cherry-picked @boyuanzz commit from the other PR notice however that I could not get the Type to be preserved after the two DoFns are applied.

In OutputSingleSource I was able to export correctly the type by overriding

    @Override
    public TypeDescriptor<SourceT> getOutputTypeDescriptor() {
      return (TypeDescriptor<SourceT>)
          new TypeDescriptor<Source<T>>(getClass()) {}.where(
              new TypeParameter<T>() {}, source.getOutputCoder().getEncodedTypeDescriptor());
    }

but in the second DoFn BoundedSourceAsSDFWrapperFn

I did not find a way to recover the real type of T when overwriting getOutputTypeDescriptor. Any suggestion for this? Otherwise I suppose the solution on Read is good enough.

[boyuanzz]

I cherry-picked @boyuanzz commit from the other PR notice however that I could not get the Type to be preserved after the two DoFns are applied.

In OutputSingleSource I was able to export correctly the type by overriding

    @Override
    public TypeDescriptor<SourceT> getOutputTypeDescriptor() {
      return (TypeDescriptor<SourceT>)
          new TypeDescriptor<Source<T>>(getClass()) {}.where(
              new TypeParameter<T>() {}, source.getOutputCoder().getEncodedTypeDescriptor());
    }

but in the second DoFn BoundedSourceAsSDFWrapperFn

I did not find a way to recover the real type of T when overwriting getOutputTypeDescriptor. Any suggestion for this? Otherwise I suppose the solution on Read is good enough.

I would say, let's unblock you first : ) Can you file a JIRA issue on this? I think we should find a ultimate solution there.

[boyuanzz]

Run Java PreCommit

[boyuanzz]

Run Java_Examples_Dataflow PreCommit

[boyuanzz]

Thanks!

[iemejia]

Thanks @boyuanzz I filled BEAM-12420 TypeDescriptor information gets lost when applying multiple DoFn on Composite Transform as a follow up.
I need to focus on finishing the Convert work so feel free to take it if you have some free cycles.