I have a postgres table with schema
row_id INT not null
person_id INT not null
case_id INT not null
seq_num INT
code VARCHAR(10)
The test pipeline to read the data using ReadFromJDBC from Apache Beam Python sdk (2.34.0), Python 3.8.x
types = [
('row_id', int), ('person_id', int), ('case_id', int), ('seq_num', int), ('code', str)
]
TableRow = typing.NamedTuple('TableRow', types)
with TestPipeline() as p:
coders.registry.register_coder(TableRow, coders.RowCoder)
p.not_use_test_runner_api = True
result = (
p
| 'Read from jdbc' >> ReadFromJdbc(
table_name=table_name,
query='SELECT * FROM table_name LIMIT 100',
driver_class_name='org.postgresql.Driver',
jdbc_url=postgres_jdbc_url,
username=username,
password=password,
))
is giving me the following error
ValueError: Failed to decode schema due to an issue with Field proto:
name: "code"
type {
nullable: true
logical_type {
urn: "beam:logical_type:javasdk:v1"
payload: "\202SNAPPY\000\000\000\000\001\000\000\000\001\000\000\002\233\250\010\360U\254\355\000\005sr\000=org.apache.beam.sdk.io.jdbc.LogicalTypes$VariableLengthString\r<\273\'6u\341\257\002\000\001I\000\tmax\t\035\014xr\0008\242X\000\020JdbcL\031i\270\246\376\361\367\203_\313a\002\000\004L\000\010argumentt\000\022Ljava/lang/Object;L\000\014a\r \001\243\034t\000.Lorg/\t\316\000/\001\3164/sdk/schemas/S\005\010\024$Field\0010\034;L\000\010base\001\014Dq\000~\000\003L\000\nidentifier6r\000\t\357\030;xpsr\000\021\001\211\000.\001\211<.Integer\022\342\240\244\367\201\2078%\007$\005valuexr\000\020\031(hNumber\206\254\225\035\013\224\340\213\002\000\000xp\000\000\000\nsr\0006N(\001\r\262\024.AutoV\001N\000_\t\274\004_F\021\274h9\304m\364S\243\227P\002\000\010L\000\025collectionEle!/\035\323\004\013l9\\\010t\0000\216\"\001\000L\0312$;L\000\nmapKey\035S\014\014map\005\227\035\024,\010metadatat\000\017)\252\034util/Map!g(nullablet\000\023\t\035!>8/Boolean;L\000\trow\t\343\010t\000$\212\243\000\001T(typeNamet\000-\2122\000\000$\001\254\001/\020;xr\000,nu\001\t\210Y\'\034\345\260\035\252D\000\370H%\266\001\001\014sr\000\036%\333\001\342\004.C5|Ds$EmptyMapY6\024\205Z\334\347\320\005362\002\r\364,\315 r\200\325\234\372\356\002\000\001ZQ2\030p\000p~r\000+\212\234\000\021\314\000\000\r\001\030\022\000\000xr\000\016\005\225!Z\020.Enum\r\034\005\035$pt\000\005INT32sA\351\000\t\001\306\001\t\000\022\001\005\010\024p~\001\007\\\025t\000\006STRINGt\000\007VARCHAR\000\000\000\n"
representation {
atomic_type: STRING
}
argument_type {
atomic_type: INT32
}
argument {
atomic_value {
int32: 10
}
}
}
}
id: 4
encoding_position: 4
I am having trouble understanding what should I do to fix it. It seems like Apache Beam doesn't know how to decode VARCHAR(10) into string. What is the correct schema to use? Or what else should I do?