[Bug]: xlang JDBC (versions 2.65.0, 2.66.65)- No logical type registered for URN 'beam:logical_type:javasdk:v1'

[mataralhawiti]

What happened?

our pipeline uses JDBC Xlang to reads from data from MSSQL Server.
The pipeline runs without any issues on version 2.64.0, however, it fails when we upgrade to 2.65.0 or 2.66.0

Table/data

-- create table
CREATE TABLE Transactions (
    TRX_ID INT PRIMARY KEY,
    Sale_Date DATE
);

-- insert sample data
INSERT INTO Transactions (TRX_ID, Sale_Date) VALUES (1, '2025-07-01');
INSERT INTO Transactions (TRX_ID, Sale_Date) VALUES (2, '2025-07-15');

Xlan

from apache_beam.coders.row_coder import LogicalType
from apache_beam.typehints.schemas import MillisInstant
LogicalType.register_logical_type(MillisInstant)
    
# Start the pipeline
with beam.Pipeline(options=pipeline_options) as p:
p
| f"Read-jdbc" >> ReadFromJdbc(
    query="SELECT TRX_ID, Sale_Date FROM Transactions",
    table_name="Transactions",
    driver_class_name="com.microsoft.sqlserver.jdbc.SQLServerDriver",
    jdbc_url="jdbc:sqlserver://{0};databaseName={1}".format(
        credentials["server"],
        credentials["database"]
    ),
    username=credentials["username"],
    password=credentials["password"],
    classpath=[
        "gs://xxxxxxx/mssql-jdbc-12.6.2.jre11.jar"
    ],
)

our pipeline logs

Error processing instruction process_bundle-2230229969916398560-22. Original traceback is
Traceback (most recent call last):
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 560, in named_tuple_from_schema
field_py_type = self.typing_from_runner_api(field.type)
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 541, in typing_from_runner_api
return LogicalType.from_runner_api(
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 793, in from_runner_api
raise ValueError(
ValueError: No logical type registered for URN 'beam:logical_type:javasdk:v1'

File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 564, in named_tuple_from_schema
raise ValueError(
ValueError: Failed to decode schema due to an issue with Field proto:

name: "Sale_Date"
type {
logical_type {
urn: "beam:logical_type:javasdk:v1"
representation {
logical_type {
urn: "beam:logical_type:millis_instant:v1"
representation {
atomic_type: INT64
}
}
}
argument_type {
atomic_type: STRING
}
argument {
atomic_value {
string: ""
}
}
}
}

some issues might be related to this:
#35426
#35736
#35488

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

This should be fixed by #35426 (tests updated accordingly to verify this https://github.com/apache/beam/pull/35426/files#diff-f74abd8bee97790c1ea61a3be880191eda760a12b20df0206ffa67ab7d5a94c7). Would you mind testing with 2.67.0RC1 when it is out

[mataralhawiti]

thanks @Abacn

Note:

• just to clarify our pipeline works fine on 2.64.0 only when we use workaround (LogicalType.register_logical_type(MillisInstant))
• and failing on 2.65.0 and 2.66.0 either with our without workaround.

Test with new RC

I've tested with (2.67.0rc1) and I'm still facing the same issue. I tried with and without workaround mentioed above.

here's the new log from rc1:

File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 567, in named_tuple_from_schema
raise ValueError(
ValueError: Failed to decode schema due to an issue with Field proto:

name: "Sale_Date"
type {
logical_type {
urn: "beam:logical_type:javasdk_date:v1"
payload: "\202SNAPPY\000\000\000\000\001\000\000\000\001\000\000\002c\303\007\360@\254\355\000\005sr\000*org.apache.beam.sdk.io.jdbc.LogicalTypes$2M\352\236\036h\3034/\002\000\000xr\000?N9\000 schemas.l\t9Dtypes.PassThroughL\t\030\001Q\270\210\324\331\211\313P\033\263\002\000\004L\000\010argumentt\000\022Ljava/lang/Object;L\000\014a\r \001:\034t\000.Lorg/\t\266\000/\001\266\020/sdk/\r}\004/S\005\205\024$Field\0010\020;L\000\tf\021\rDq\000~\000\003L\000\nidentifier6s\000<String;xpt\000\000sr\0006n\346\000$AutoValue_\ts\000_\025sh9\304m\364S\243\227P\002\000\010L\000\025collectionEle\001\346\001\226\r\211\000\013-+\001\023\010t\0000\216\331\000\000L9E$;L\000\nmapKey\001@\rS\014\014map\005\227\035\024,\010metadatat\000\017)aXutil/Map;L\000\010nullablet\000\023\t\035%\030Boolean!?\010row\t\343\010t\000$\212\243\000\001T!\374\030Namet\000-\2122\000\000$\001\254\001/\020;xr\000,nu\001\t\2109\3360\013PLl[\357\3103\002\000\000xp\001\001\014sr\000\036AC\000.\001\342\004.C5|Ds$EmptyMapY6\024\205Z\334\347\320\0053\014sr\000\021\005/\020lang.\r\3648\315 r\200\325\234\372\356\002\000\001Z\000\005v!\344\034xp\000pr\000+\212\234\000\021\314\000\000\r\001\000\022e1\000\016\031f\014Enum\r\034\005\035(pt\000\006STRINGs!\304\020\007pppp\001\t\000\020\001\005\010\022p~\001\007H\023t\000\010DATETIMEt\000\004DATE"
representation {
logical_type {
urn: "beam:logical_type:millis_instant:v1"
representation {
atomic_type: INT64
}
}
}
argument_type {
atomic_type: STRING
}
argument {
atomic_value {
string: ""
}
}
}
}
id: 1
encoding_position: 1

[Abacn]

"sale_date" should resolve to "urn: "beam:logical_type:javasdk_date:v1" but it now becomes "beam:logical_type:javasdk:v1". Let me investigate further what has been changed. (looks like 2.67.0rc1 at least resolved to "beam:logical_type:javasdk_date:v1"

update - wasn't able to reproduce this on postgres. Date field was read correctly. Setting up a mssqlserver locally had some challenges

[shunping]

Yes, I also notice that the urn was beam:logical_type:javasdk:v1 on beam 2.65 and 2.66.

It seems that the following codepath is triggered in sqlserver:

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java

Lines 87 to 91 in ac2706a

	} else {
	// raw "javasdk" name should only be a last resort. Types defined in Beam should have their
	// own URN.
	return "beam:logical_type:javasdk:v1";
	}

However, on beam 2.67rc1, it becomes beam:logical_type:javasdk_date:v1.

[shunping]

@mataralhawiti, could you include more traceback for the 2.67rc1 case?

What are the other error messages before the following?

raise ValueError(
ValueError: Failed to decode schema due to an issue with Field proto:

[mataralhawiti]

@shunping here's the full error message - it's the first error message :

Error message from worker: Traceback (most recent call last):
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 563, in named_tuple_from_schema
field_py_type = self.typing_from_runner_api(field.type)
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 544, in typing_from_runner_api
return LogicalType.from_runner_api(
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 803, in from_runner_api
raise ValueError(
ValueError: No logical type registered for URN 'beam:logical_type:javasdk_date:v1'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 313, in _execute
response = task()
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 387, in
lambda: self.create_worker().do_instruction(request), request)
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 659, in do_instruction
return getattr(self, request_type)(
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 690, in process_bundle
bundle_processor = self.bundle_processor_cache.get(
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 511, in get
processor = bundle_processor.BundleProcessor(
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1133, in init
self.ops = self.create_execution_tree(self.process_bundle_descriptor)
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1190, in create_execution_tree
return collections.OrderedDict([(
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1193, in
get_operation(transform_id))) for transform_id in sorted(
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1038, in wrapper
result = cache[args] = func(*args)
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1167, in get_operation
transform_consumers = {
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1168, in
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1168, in
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1038, in wrapper
result = cache[args] = func(*args)
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1175, in get_operation
self.data_sampler.initialize_samplers(
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/data_sampler.py", line 310, in initialize_samplers
coder = coder_factory(coder_id)
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1519, in get_coder
return self.context.coders.get_by_id(coder_id)
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/pipeline_context.py", line 106, in get_by_id
self._id_to_obj[id] = self._obj_type.from_runner_api(
File "/usr/local/lib/python3.10/site-packages/apache_beam/coders/coders.py", line 370, in from_runner_api
[context.coders.get_by_id(c) for c in coder_proto.component_coder_ids],
File "/usr/local/lib/python3.10/site-packages/apache_beam/coders/coders.py", line 370, in
[context.coders.get_by_id(c) for c in coder_proto.component_coder_ids],
File "/usr/local/lib/python3.10/site-packages/apache_beam/runners/pipeline_context.py", line 106, in get_by_id
self._id_to_obj[id] = self._obj_type.from_runner_api(
File "/usr/local/lib/python3.10/site-packages/apache_beam/coders/coders.py", line 368, in from_runner_api
return constructor(
File "/usr/local/lib/python3.10/site-packages/apache_beam/coders/row_coder.py", line 107, in from_runner_api_parameter
return RowCoder(schema)
File "/usr/local/lib/python3.10/site-packages/apache_beam/coders/row_coder.py", line 66, in init
self._type_hint = named_tuple_from_schema(self.schema)
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 606, in named_tuple_from_schema
schema_registry=schema_registry).named_tuple_from_schema(schema)
File "/usr/local/lib/python3.10/site-packages/apache_beam/typehints/schemas.py", line 567, in named_tuple_from_schema
raise ValueError(
ValueError: Failed to decode schema due to an issue with Field proto:

name: "Sale_Date"
type {
logical_type {
urn: "beam:logical_type:javasdk_date:v1"
payload: "\202SNAPPY\000\000\000\000\001\000\000\000\001\000\000\002c\303\007\360@\254\355\000\005sr\000*org.apache.beam.sdk.io.jdbc.LogicalTypes$2M\352\236\036h\3034/\002\000\000xr\000?N9\000 schemas.l\t9Dtypes.PassThroughL\t\030\001Q\270\210\324\331\211\313P\033\263\002\000\004L\000\010argumentt\000\022Ljava/lang/Object;L\000\014a\r \001:\034t\000.Lorg/\t\266\000/\001\266\020/sdk/\r}\004/S\005\205\024$Field\0010\020;L\000\tf\021\rDq\000~\000\003L\000\nidentifier6s\000<String;xpt\000\000sr\0006n\346\000$AutoValue_\ts\000_\025sh9\304m\364S\243\227P\002\000\010L\000\025collectionEle\001\346\001\226\r\211\000\013-+\001\023\010t\0000\216\331\000\000L9E$;L\000\nmapKey\001@\rS\014\014map\005\227\035\024,\010metadatat\000\017)aXutil/Map;L\000\010nullablet\000\023\t\035%\030Boolean!?\010row\t\343\010t\000$\212\243\000\001T!\374\030Namet\000-\2122\000\000$\001\254\001/\020;xr\000,nu\001\t\2109\3360\013PLl[\357\3103\002\000\000xp\001\001\014sr\000\036AC\000.\001\342\004.C5|Ds$EmptyMapY6\024\205Z\334\347\320\0053\014sr\000\021\005/\020lang.\r\3648\315 r\200\325\234\372\356\002\000\001Z\000\005v!\344\034xp\000pr\000+\212\234\000\021\314\000\000\r\001\000\022e1\000\016\031f\014Enum\r\034\005\035(pt\000\006STRINGs!\304\020\007pppp\001\t\000\020\001\005\010\022p~\001\007H\023t\000\010DATETIMEt\000\004DATE"
representation {
logical_type {
urn: "beam:logical_type:millis_instant:v1"
representation {
atomic_type: INT64
}
}
}
argument_type {
atomic_type: STRING
}
argument {
atomic_value {
string: ""
}
}
}
}
id: 1
encoding_position: 1

[shunping]

Given this error is ValueError: No logical type registered for URN 'beam:logical_type:javasdk_date:v1' (same as #33442), I think putting the following before defining the pipeline, i.e. with beam.Pipeline(options=pipeline_options) as p: should work.

from apache_beam.io.jdbc import JdbcDateType  # pylint: disable=unused-import

Please give it a try on 2.67rc1.