vbhanuchander-lang commented on PR #28490:
URL: https://github.com/apache/flink/pull/28490#issuecomment-5526519234
@wilmerdooley this is the missing branch that FLINK-39724 needs, and it is
also a prerequisite for
your #28491 — `assign_timestamps_and_watermarks` calls `get_type()`, so a
CSV source cannot get
through that path until `_from_java_type` handles `InternalTypeInfo`. I left
the detail on that PR.
**One question I could not settle by reading, and I think it is the crux of
the review.**
`InternalTypeInfo` describes a stream of `RowData`, not `Row`. This change
makes `_from_java_type`
*report* `Types.ROW([...])` for such a stream, which fixes `get_type()`, but
the records on the wire
are still `RowData`. So does a downstream Python operator decode them
correctly once it is told they
are `Row`s — or does the exception get replaced by a wrong-decoding path?
What makes me ask is that the codebase already has a place where it meets an
`InternalTypeInfo`
stream from Python, and it converts the records rather than relabelling the
type.
`FileSink.BulkFormatBuilder._with_row_type` (`file_system.py:709`) does:
```python
def _check_if_row_data_type(ds) -> bool:
j_type_info = ds._j_data_stream.getType()
if not is_instance_of(j_type_info,
'org.apache.flink.table.runtime.typeutils.InternalTypeInfo'):
return False
return is_instance_of(j_type_info.toLogicalType(),
'org.apache.flink.table.types.logical.RowType')
```
and then applies a `RowRowTransformer`. That precedent suggests the
conversion is load-bearing, and
it would also explain the reporter's workaround: putting the stream through
an identity `map` with an
explicit `output_type` gives it a real PyFlink type instead of the internal
one.
If a Python operator does handle it, this is fine as written and worth a
comment saying why. If it
does not, then `get_type()` starts succeeding while records decode
incorrectly, which would be worse
than today's `TypeError` — and the fix would belong nearer
`RowRowTransformer` than
`_from_java_type`. Your test asserts the type conversion, which is the right
thing to assert for
this change, but it would pass under either answer, so it cannot distinguish
them.
To be clear about my own limits: there is no Flink build on this machine, so
I read `master` rather
than running anything. The `PythonCsvUtils.java:48` and `file_system.py:709`
references are from the
current tree; the question above is a question, not a claim.
--
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]
