Proposal 1 would also interact poorly with SELECT * EXCEPT ... statements, which returns all columns except specific ones. Adding an unknown field does seem like a reasonable way to handle this. It probably needs to be something that is native to the Row type, so columns added to nested rows also work.
Andrew On Tue, Dec 8, 2020 at 9:50 AM Reuven Lax <re...@google.com> wrote: > There's a difference between a fully dynamic schema and simply being able > to forward "unknown" fields to the output. > > A fully-dynamic schema is not really necessary unless we also had dynamic > SQL statements. Since the existing SQL statements do not reference the new > fields by name, there's no reason to add them to the main schema. > > However, if you have a SELECT * FROM WHERE XXXX statement that does no > aggregation, there's fundamentally no reason we couldn't forward the > messages exactly. In theory we could forward the exact bytes that are in > the input PCollection, which would necessarily forward the new fields. In > practice I believe that we convert the input messages to Beam Row objects > in order to evaluate the WHERE clause, and then convert back to Avro to > output those messages. I believe this is where we "lose" the unknown > messages,but this is an implementation artifact - in theory we could output > the original bytes whenever we see a SELECT *. This is not truly a dynamic > schema, since you can't really do anything with these extra fields except > forward them to your output. > > I see two possible ways to address this. > > 1. As I mentioned above, in the case of a SELECT * we could output the > original bytes, and only use the Beam Row for evaluating the WHERE clause. > This might be very expensive though - we risk having to keep two copies of > every message around, one in the original Avro format and one in Row format. > > 2. The other way would be to do what protocol buffers do. We could add one > extra field to the inferred Beam schema to store new, unknown fields > (probably this would be a map-valued field). This extra field would simply > store the raw bytes of these unknown fields, and then when converting back > to Avro they would be added to the output message. This might also add some > overhead to the pipeline, so might be best to make this behavior opt in. > > Reuven > > On Tue, Dec 8, 2020 at 9:33 AM Brian Hulette <bhule...@google.com> wrote: > >> Reuven, could you clarify what you have in mind? I know multiple times >> we've discussed the possibility of adding update compatibility support to >> SchemaCoder, including support for certain schema changes (field >> additions/deletions) - I think the most recent discussion was here [1]. >> >> But it sounds like Talat is asking for something a little beyond that, >> effectively a dynamic schema. Is that something you think we can support? >> >> [1] >> https://lists.apache.org/thread.html/ref73a8c40e24e0b038b4e5b065cd502f4c5df2e5e15af6f7ea1cdaa7%40%3Cdev.beam.apache.org%3E >> >> On Tue, Dec 8, 2020 at 9:20 AM Reuven Lax <re...@google.com> wrote: >> >>> Thanks. It might be theoretically possible to do this (at least for the >>> case where existing fields do not change). Whether anyone currently has >>> available time to do this is a different question, but it's something that >>> can be looked into. >>> >>> On Mon, Dec 7, 2020 at 9:29 PM Talat Uyarer < >>> tuya...@paloaltonetworks.com> wrote: >>> >>>> Adding new fields is more common than modifying existing fields. But >>>> type change is also possible for existing fields, such as regular mandatory >>>> field(string,integer) to union(nullable field). No field deletion. >>>> >>>> On Mon, Dec 7, 2020 at 9:22 PM Reuven Lax <re...@google.com> wrote: >>>> >>>>> And when you say schema changes, are these new fields being added to >>>>> the schema? Or are you making changes to the existing fields? >>>>> >>>>> On Mon, Dec 7, 2020 at 9:02 PM Talat Uyarer < >>>>> tuya...@paloaltonetworks.com> wrote: >>>>> >>>>>> Hi, >>>>>> For sure let me explain a little bit about my pipeline. >>>>>> My Pipeline is actually simple >>>>>> Read Kafka -> Convert Avro Bytes to Beam Row(DoFn<KV<byte[], byte[]>, >>>>>> Row>) -> Apply Filter(SqlTransform.query(sql)) -> Convert back from >>>>>> Row to Avro (DoFn<Row, byte[]>)-> Write DB/GCS/GRPC etc >>>>>> >>>>>> On our jobs We have three type sqls >>>>>> - SELECT * FROM PCOLLECTION >>>>>> - SELECT * FROM PCOLLECTION <with Where Condition> >>>>>> - SQL Projection with or without Where clause SELECT col1, col2 FROM >>>>>> PCOLLECTION >>>>>> >>>>>> We know writerSchema for each message. While deserializing avro >>>>>> binary we use writer schema and reader schema on Convert Avro Bytes to >>>>>> Beam >>>>>> Row step. It always produces a reader schema's generic record and we >>>>>> convert that generic record to Row. >>>>>> While submitting DF job we use latest schema to generate beamSchema. >>>>>> >>>>>> In the current scenario When we have schema changes first we restart >>>>>> all 15k jobs with the latest updated schema then whenever we are done we >>>>>> turn on the latest schema for writers. Because of Avro's >>>>>> GrammerResolver[1] >>>>>> we read different versions of the schema and we always produce the latest >>>>>> schema's record. Without breaking our pipeline we are able to handle >>>>>> multiple versions of data in the same streaming pipeline. If we can >>>>>> generate SQL's java code when we get notified wirth latest schema we will >>>>>> handle all schema changes. The only remaining obstacle is Beam's SQL Java >>>>>> code. That's why I am looking for some solution. We dont need multiple >>>>>> versions of SQL. We only need to regenerate SQL schema with the latest >>>>>> schema on the fly. >>>>>> >>>>>> I hope I can explain it :) >>>>>> >>>>>> Thanks >>>>>> >>>>>> [1] >>>>>> https://avro.apache.org/docs/1.7.2/api/java/org/apache/avro/io/parsing/doc-files/parsing.html >>>>>> <https://urldefense.proofpoint.com/v2/url?u=https-3A__avro.apache.org_docs_1.7.2_api_java_org_apache_avro_io_parsing_doc-2Dfiles_parsing.html&d=DwMFaQ&c=V9IgWpI5PvzTw83UyHGVSoW3Uc1MFWe5J8PTfkrzVSo&r=BkW1L6EF7ergAVYDXCo-3Vwkpy6qjsWAz7_GD7pAR8g&m=0qahAe7vDisJq_hMYGY8F-Bp7-_5lOwOKzNoQ3r3-IQ&s=lwwIMsJO9nmj6_xZcSG_7qkBIaxOwyUXry4st1q70Rc&e=> >>>>>> >>>>>> On Mon, Dec 7, 2020 at 7:49 PM Reuven Lax <re...@google.com> wrote: >>>>>> >>>>>>> Can you explain the use case some more? Are you wanting to change >>>>>>> your SQL statement as well when the schema changes? If not, what are >>>>>>> those >>>>>>> new fields doing in the pipeline? What I mean is that your old SQL >>>>>>> statement clearly didn't reference those fields in a SELECT statement >>>>>>> since >>>>>>> they didn't exist, so what are you missing by not having them unless you >>>>>>> are also changing the SQL statement? >>>>>>> >>>>>>> Is this a case where you have a SELECT *, and just want to make sure >>>>>>> those fields are included? >>>>>>> >>>>>>> Reuven >>>>>>> >>>>>>> On Mon, Dec 7, 2020 at 6:31 PM Talat Uyarer < >>>>>>> tuya...@paloaltonetworks.com> wrote: >>>>>>> >>>>>>>> Hi Andrew, >>>>>>>> >>>>>>>> I assume SQL query is not going to change. Changing things is the >>>>>>>> Row schema by adding new columns or rename columns. if we keep a >>>>>>>> version >>>>>>>> information on somewhere for example a KV pair. Key is schema >>>>>>>> information, >>>>>>>> value is Row. Can not we generate SQL code ? Why I am asking We have >>>>>>>> 15k >>>>>>>> pipelines. When we have a schema change we restart a 15k DF job which >>>>>>>> is >>>>>>>> pain. I am looking for a possible way to avoid job restart. Dont you >>>>>>>> think >>>>>>>> it is not still doable ? >>>>>>>> >>>>>>>> Thanks >>>>>>>> >>>>>>>> >>>>>>>> On Mon, Dec 7, 2020 at 6:10 PM Andrew Pilloud <apill...@google.com> >>>>>>>> wrote: >>>>>>>> >>>>>>>>> Unfortunately we don't have a way to generate the SQL Java code on >>>>>>>>> the fly, even if we did, that wouldn't solve your problem. I believe >>>>>>>>> our >>>>>>>>> recommended practice is to run both the old and new pipeline for some >>>>>>>>> time, >>>>>>>>> then pick a window boundary to transition the output from the old >>>>>>>>> pipeline >>>>>>>>> to the new one. >>>>>>>>> >>>>>>>>> Beam doesn't handle changing the format of data sent between >>>>>>>>> intermediate steps in a running pipeline. Beam uses "coders" to >>>>>>>>> serialize >>>>>>>>> data between steps of the pipeline. The builtin coders (including the >>>>>>>>> Schema Row Coder used by SQL) have a fixed data format and don't >>>>>>>>> handle >>>>>>>>> schema evolution. They are optimized for performance at all costs. >>>>>>>>> >>>>>>>>> If you worked around this, the Beam model doesn't support changing >>>>>>>>> the structure of the pipeline graph. This would significantly limit >>>>>>>>> the >>>>>>>>> changes you can make. It would also require some changes to SQL to >>>>>>>>> try to >>>>>>>>> produce the same plan for an updated SQL query. >>>>>>>>> >>>>>>>>> Andrew >>>>>>>>> >>>>>>>>> On Mon, Dec 7, 2020 at 5:44 PM Talat Uyarer < >>>>>>>>> tuya...@paloaltonetworks.com> wrote: >>>>>>>>> >>>>>>>>>> Hi, >>>>>>>>>> >>>>>>>>>> We are using Beamsql on our pipeline. Our Data is written in Avro >>>>>>>>>> format. We generate our rows based on our Avro schema. Over time the >>>>>>>>>> schema >>>>>>>>>> is changing. I believe Beam SQL generates Java code based on what we >>>>>>>>>> define >>>>>>>>>> as BeamSchema while submitting the pipeline. Do you have any idea >>>>>>>>>> How can >>>>>>>>>> we handle schema changes with resubmitting our beam job. Is it >>>>>>>>>> possible to >>>>>>>>>> generate SQL java code on the fly ? >>>>>>>>>> >>>>>>>>>> Thanks >>>>>>>>>> >>>>>>>>>
