Hey everyone, Summary:
There is an issue with the Dataflow runner and the “Update” capability while using the beam native Row type, which I imagine also blocks the snapshots feature (as the docs say the snapshots have the same restrictions as the Update feature) but I have no experience there. Currently when reading from KafkaIO with the valueCoder set as a SchemaCoder: ``` KafkaIO.Read<ByteArray, Row>() .withTopic(topic) .withKeyDeserializer(ByteArrayDeserializer::class.java) .withValueDeserializerAndCoder([Deserializer<Row>], SchemaCoder.of(outputSchema)) ``` Updates fail consistently with the error: ``` The original job has not been aborted., The Coder or type for step ReadInputTopic/Read(KafkaUnboundedSource)/DataflowRunner.StreamingUnboundedRead.ReadWithIds has changed ``` There is an open issue about this, https://issues.apache.org/jira/browse/BEAM-9502 <https://issues.apache.org/jira/browse/BEAM-9502> but I have not seen it discussed in the mailing list so I wanted to start it. Investigation so far: This failing on Beam 2.20 and below makes sense, as before the code path that called equals on this Coder first checked that the schema’s were equal (This part has not changed): https://github.com/apache/beam/blob/release-2.25.0/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaCoder.java#L194 <https://github.com/apache/beam/blob/release-2.25.0/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaCoder.java#L194>. Then this called the equals on the schema here, which if UUIDs were different caused false to be returned: https://github.com/apache/beam/blob/release-2.20.0/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java#L272 <https://github.com/apache/beam/blob/release-2.20.0/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java#L272> This means that as the above issue suggests, the UUID being random meant that no two SchemaCoders were ever the same, causing the equals to return false. In [BEAM-4076] this was changed (PR link: https://github.com/apache/beam/pull/11041/files <https://github.com/apache/beam/pull/11041/files>, direct link: https://github.com/reuvenlax/incubator-beam/blob/37b1fbaa7ea1b64303eaacae9e670797a92b1a50/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java#L272 <https://github.com/reuvenlax/incubator-beam/blob/37b1fbaa7ea1b64303eaacae9e670797a92b1a50/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java#L272> ) So now, it appears that if the UUIDs don’t match, it still checks the fieldIndices, fields and fieldOptions. Reasonably, this should mean that we don’t get the incompatibility error, but is potentially slower than would be ideal. But we are still getting the same error on Beam 2.24. I added a way to deterministically generate a UUID which we do on the schema we use in `.withValueDeserializerAndCoder([Deserializer<Row>], SchemaCoder.of(outputSchema))`. When we do not set the Schema UUID it fails with the same error every time so it is difficult to log the SchemaCoder. When we do set the schema UUID deterministically, it fails sometimes and other times succeeds. When it does succeed, logging the coder, the fields are the same in the same order with the same UUID on the schema. When I launch a new job, the logging prints the same coder as in iteration 1, stumping me on why the update failed at all. I need some advice here. Follow up discussion: I am curious if it has been determined what makes a Schema the same as another schema. From what I have seen in the codebase, it changes. In this PR: https://github.com/apache/beam/pull/13049 <https://github.com/apache/beam/pull/13049> I have generated a Schema UUID based on the the field names, typeNames, and nullability. Let me know if this is the right direction. I think this UUID is worth making deterministic even if it doesn’t solve the Dataflow update issues, so that comparison is faster. There was a previous draft PR but I got busy and didn’t finish: https://github.com/apache/beam/pull/11447#issuecomment-615442205, which had the reply: ``` I would like to talk about the use case a bit more. I think this is probably to do with update support on the Dataflow runner, in which case this may not be the right solution (and may not be a sufficient change). ``` Follow up questions that may have overlap: This is the only issue related to Dataflow updates with the Row type I could see, but are there more internally, and if so, can they be made public so that we can work on them? I know Row type is “Experimental” but I have only seen discussions that DATETIME should be removed before removing that tag. What other issues and blockers are required for this to happen, if there are any? Is this the totally wrong approach/wrong place for this discussion? Should SchemaOptions and field descriptions be included to create a UUID? I don’t think they should, so that descriptions and options can be updated (just my naive opinion, to be changed). Perhaps there are greater plans for schemas. Can this (or something similar) merge as a stop gap? This is my first message to the mailing list, so I apologize if any info is missing. Thanks, Cameron