SqlTransform only allows one registered RowCoder schema
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 196
Description
The current workflow for using the SqlTransform is:
```
Row = typing.NamedTuple("Row", [("col1", int), ("col2", str)])
beam.coders.registry.register_coder(Row,
beam.coders.RowCoder)
with self.create_pipeline() as p:
output = (
p
| 'Create' >> beam.Create([Row(x, str(x)) for x in range(5)])
| 'Sql' >> SqlTransform(
"""SELECT col1, col2 || '*' || col2 as col2,
power(col1, 2)
as col3
FROM PCOLLECTION))
```
This works fine, but when multiple row schemas are registered like this:
```
Row = typing.NamedTuple("Row", [("col1", int), ("col2", str)])
beam.coders.registry.register_coder(Row,
beam.coders.RowCoder)
with self.create_pipeline() as p:
output = (
p
| 'Create' >> beam.Create([Row(x, str(x)) for x in range(5)])
| 'Sql' >> SqlTransform(
"""SELECT col1, col2 || '*' || col2 as col2,
power(col1, 2)
as col3
FROM PCOLLECTION))
output2 = (
p
| 'Create2'
>> beam.Create([Row2(x, str(x)) for x in range(5)])
| 'Sql2' >> SqlTransform(
"""SELECT col1, col2 || '*' || col2 as col2,
power(col1, 2) as col3
FROM PCOLLECTION
"""))
```
This yields:
```
RuntimeError: Re-used coder id: ref_Coder_RowCoder_1
```
Source: https://github.com/apache/beam/blob/a8f390704925d3a371b007ccbfcfc28a48b312d1/sdks/python/apache_beam/transforms/external.py#L419
Imported from Jira [BEAM-10389](https://issues.apache.org/jira/browse/BEAM-10389). Original Jira may contain additional context.
Reported by: mxm.
Contributor guide
Assessment
This issue has not been assessed yet.