Abacn commented on code in PR #36515:
URL: https://github.com/apache/beam/pull/36515#discussion_r2430465339
##########
sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/TransformProvider.java:
##########
@@ -127,7 +134,30 @@ default String getTransformUniqueID(RunnerApi.FunctionSpec
spec) {
ExternalTransforms.SchemaTransformPayload payload;
try {
payload =
ExternalTransforms.SchemaTransformPayload.parseFrom(spec.getPayload());
- return payload.getIdentifier();
+ if
(PTransformTranslation.MANAGED_TRANSFORM_URN.equals(payload.getIdentifier())) {
+ try {
+ // ManagedSchemaTransform includes a schema field
transform_identifier that includes the
+ // underlying schema
+ // transform ID so we special case that here.
+ Row configRow =
+
RowCoder.of(SchemaTranslation.schemaFromProto(payload.getConfigurationSchema()))
+ .decode(new
ByteArrayInputStream(payload.getConfigurationRow().toByteArray()));
Review Comment:
This contains multi level deref expression. Any risk that any field could be
null or empty? If so shall we skip to return `configRow.getValue` in that case?
##########
sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceOptions.java:
##########
@@ -65,6 +65,13 @@ public interface ExpansionServiceOptions extends
PipelineOptions {
void setUseAltsServer(boolean useAltsServer);
+ @Description(
Review Comment:
is this pipeline option going to be a temporary one or persistent in code
base?
--
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]
