We could support EXPECT statements in proposal 2 as long as we restricted it to known fields.
We are getting into implementation details now. Making unknown fields just a normal column introduces a number of problems. ZetaSQL doesn't support Map type. All our IOs would need to explicitly deal with that special column. There would be a lack of consistency between the various types (Avro, Proto, Json) which should all support this. We might also want something even more invasive: everything is an unknown field unless it is referenced in the SQL query. All of these options are possible. I guess we need someone who has time to work on it to write a proposal. On Tue, Dec 8, 2020 at 10:03 AM Reuven Lax <re...@google.com> wrote: > I'm not sure that we could support EXCEPT statements, as that would > require introspecting the unknown fields (what if the EXCEPT statement > matches a field that later is added as an unknown field?). IMO this sort of > behavior only makes sense on true pass-through queries. Anything that > modifies the input record would be tricky to support. > > Nested rows would work for proposal 2. You would need to make sure that > the unknown-fields map is recursively added to all nested rows, and you > would do this when you infer a schema from the avro schema. > > On Tue, Dec 8, 2020 at 9:58 AM Andrew Pilloud <apill...@google.com> wrote: > >> 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 >>>>>>>>>>>> >>>>>>>>>>>
