Could the DirectRunner just do an equality check whenever it does an
encode/decode? It sounds like it's already effectively performing
a CoderProperties.coderDecodeEncodeEqual for every element, just omitting
the equality check.
On Wed, Jun 2, 2021 at 12:04 PM Reuven Lax <re...@google.com> wrote:
> There is no bug in the Coder itself, so that wouldn't catch it. We could
> insert CoderProperties.coderDecodeEncodeEqual in a subsequent ParDo, but if
> the Direct runner already does an encode/decode before that ParDo, then
> that would have fixed the problem before we could see it.
>
> On Wed, Jun 2, 2021 at 11:53 AM Kenneth Knowles <k...@apache.org> wrote:
>
>> Would it be caught by CoderProperties?
>>
>> Kenn
>>
>> On Wed, Jun 2, 2021 at 8:16 AM Reuven Lax <re...@google.com> wrote:
>>
>>> I don't think this bug is schema specific - we created a Java object
>>> that is inconsistent with its encoded form, which could happen to any
>>> transform.
>>>
>>> This does seem to be a gap in DirectRunner testing though. It also makes
>>> it hard to test using PAssert, as I believe that puts everything in a side
>>> input, forcing an encoding/decoding.
>>>
>>> On Wed, Jun 2, 2021 at 8:12 AM Brian Hulette <bhule...@google.com>
>>> wrote:
>>>
>>>> +dev <d...@beam.apache.org>
>>>>
>>>> > I bet the DirectRunner is encoding and decoding in between, which
>>>> fixes the object.
>>>>
>>>> Do we need better testing of schema-aware (and potentially other
>>>> built-in) transforms in the face of fusion to root out issues like this?
>>>>
>>>> Brian
>>>>
>>>> On Wed, Jun 2, 2021 at 5:13 AM Matthew Ouyang <matthew.ouy...@gmail.com>
>>>> wrote:
>>>>
>>>>> I have some other work-related things I need to do this week, so I
>>>>> will likely report back on this over the weekend. Thank you for the
>>>>> explanation. It makes perfect sense now.
>>>>>
>>>>> On Tue, Jun 1, 2021 at 11:18 PM Reuven Lax <re...@google.com> wrote:
>>>>>
>>>>>> Some more context - the problem is that RenameFields outputs (in this
>>>>>> case) Java Row objects that are inconsistent with the actual schema.
>>>>>> For example if you have the following schema:
>>>>>>
>>>>>> Row {
>>>>>> field1: Row {
>>>>>> field2: string
>>>>>> }
>>>>>> }
>>>>>>
>>>>>> And rename field1.field2 -> renamed, you'll get the following schema
>>>>>>
>>>>>> Row {
>>>>>> field1: Row {
>>>>>> renamed: string
>>>>>> }
>>>>>> }
>>>>>>
>>>>>> However the Java object for the _nested_ row will return the old
>>>>>> schema if getSchema() is called on it. This is because we only update the
>>>>>> schema on the top-level row.
>>>>>>
>>>>>> I think this explains why your test works in the direct runner. If
>>>>>> the row ever goes through an encode/decode path, it will come back
>>>>>> correct.
>>>>>> The original incorrect Java objects are no longer around, and new
>>>>>> (consistent) objects are constructed from the raw data and the
>>>>>> PCollection
>>>>>> schema. Dataflow tends to fuse ParDos together, so the following ParDo
>>>>>> will
>>>>>> see the incorrect Row object. I bet the DirectRunner is encoding and
>>>>>> decoding in between, which fixes the object.
>>>>>>
>>>>>> You can validate this theory by forcing a shuffle after RenameFields
>>>>>> using Reshufflle. It should fix the issue If it does, let me know and
>>>>>> I'll
>>>>>> work on a fix to RenameFields.
>>>>>>
>>>>>> On Tue, Jun 1, 2021 at 7:39 PM Reuven Lax <re...@google.com> wrote:
>>>>>>
>>>>>>> Aha, yes this indeed another bug in the transform. The schema is set
>>>>>>> on the top-level Row but not on any nested rows.
>>>>>>>
>>>>>>> On Tue, Jun 1, 2021 at 6:37 PM Matthew Ouyang <
>>>>>>> matthew.ouy...@gmail.com> wrote:
>>>>>>>
>>>>>>>> Thank you everyone for your input. I believe it will be easiest to
>>>>>>>> respond to all feedback in a single message rather than messages per
>>>>>>>> person.
>>>>>>>>
>>>>>>>> - NeedsRunner - The tests are run eventually, so obviously all
>>>>>>>> good on my end. I was trying to run the smallest subset of test
>>>>>>>> cases
>>>>>>>> possible and didn't venture beyond `gradle test`.
>>>>>>>> - Stack Trace - There wasn't any unfortunately because no
>>>>>>>> exception thrown in the code. The Beam Row was translated into a BQ
>>>>>>>> TableRow and an insertion was attempted. The error "message" was
>>>>>>>> part of
>>>>>>>> the response JSON that came back as a result of a request against
>>>>>>>> the BQ
>>>>>>>> API.
>>>>>>>> - Desired Behaviour - (field0_1.field1_0, nestedStringField) ->
>>>>>>>> field0_1.nestedStringField is what I am looking for.
>>>>>>>> - Info Logging Findings (In Lieu of a Stack Trace)
>>>>>>>> - The Beam Schema was as expected with all renames applied.
>>>>>>>> - The example I provided was heavily stripped down in order
>>>>>>>> to isolate the problem. My work example which a bit impractical
>>>>>>>> because
>>>>>>>> it's part of some generic tooling has 4 levels of nesting and
>>>>>>>> also produces
>>>>>>>> the correct output too.
>>>>>>>> - BigQueryUtils.toTableRow(Row) returns the expected
>>>>>>>> TableRow in DirectRunner. In DataflowRunner however, only the
>>>>>>>> top-level
>>>>>>>> renames were reflected in the TableRow and all renames in the
>>>>>>>> nested fields
>>>>>>>> weren't.
>>>>>>>> - BigQueryUtils.toTableRow(Row) recurses on the Row values
>>>>>>>> and uses the Row.schema to get the field names. This makes
>>>>>>>> sense to me,
>>>>>>>> but if a value is actually a Row then its schema appears to be
>>>>>>>> inconsistent
>>>>>>>> with the top-level schema
>>>>>>>> - My Current Workaround - I forked RenameFields and replaced
>>>>>>>> the attachValues in expand method to be a "deep" rename. This is
>>>>>>>> obviously
>>>>>>>> inefficient and I will not be submitting a PR for that.
>>>>>>>> - JIRA ticket - https://issues.apache.org/jira/browse/BEAM-12442
>>>>>>>>
>>>>>>>>
>>>>>>>> On Tue, Jun 1, 2021 at 5:51 PM Reuven Lax <re...@google.com> wrote:
>>>>>>>>
>>>>>>>>> This transform is the same across all runners. A few comments on
>>>>>>>>> the test:
>>>>>>>>>
>>>>>>>>> - Using attachValues directly is error prone (per the comment on
>>>>>>>>> the method). I recommend using the withFieldValue builders instead.
>>>>>>>>> - I recommend capturing the RenameFields PCollection into a
>>>>>>>>> local variable of type PCollection<Row> and printing out the schema
>>>>>>>>> (which
>>>>>>>>> you can get using the PCollection.getSchema method) to ensure that the
>>>>>>>>> output schema looks like you expect.
>>>>>>>>> - RenameFields doesn't flatten. So renaming field0_1.field1_0 -
>>>>>>>>> > nestedStringField results in field0_1.nestedStringField; if you
>>>>>>>>> > wanted to
>>>>>>>>> flatten, then the better transform would be
>>>>>>>>> Select.fieldNameAs("field0_1.field1_0", nestedStringField).
>>>>>>>>>
>>>>>>>>> This all being said, eyeballing the implementation of RenameFields
>>>>>>>>> makes me think that it is buggy in the case where you specify a
>>>>>>>>> top-level
>>>>>>>>> field multiple times like you do. I think it is simply adding the
>>>>>>>>> top-level
>>>>>>>>> field into the output schema multiple times, and the second time is
>>>>>>>>> with
>>>>>>>>> the field0_1 base name; I have no idea why your test doesn't catch
>>>>>>>>> this in
>>>>>>>>> the DirectRunner, as it's equally broken there. Could you file a JIRA
>>>>>>>>> about
>>>>>>>>> this issue and assign it to me?
>>>>>>>>>
>>>>>>>>> Reuven
>>>>>>>>>
>>>>>>>>> On Tue, Jun 1, 2021 at 12:47 PM Kenneth Knowles <k...@apache.org>
>>>>>>>>> wrote:
>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> On Tue, Jun 1, 2021 at 12:42 PM Brian Hulette <
>>>>>>>>>> bhule...@google.com> wrote:
>>>>>>>>>>
>>>>>>>>>>> Hi Matthew,
>>>>>>>>>>>
>>>>>>>>>>> > The unit tests also seem to be disabled for this as well and
>>>>>>>>>>> so I don’t know if the PTransform behaves as expected.
>>>>>>>>>>>
>>>>>>>>>>> The exclusion for NeedsRunner tests is just a quirk in our
>>>>>>>>>>> testing framework. NeedsRunner indicates that a test suite can't be
>>>>>>>>>>> executed with the SDK alone, it needs a runner. So that exclusion
>>>>>>>>>>> just
>>>>>>>>>>> makes sure we don't run the test when we're verifying the SDK by
>>>>>>>>>>> itself in
>>>>>>>>>>> the :sdks:java:core:test task. The test is still run in other tasks
>>>>>>>>>>> where
>>>>>>>>>>> we have a runner, most notably in the Java PreCommit [1], where we
>>>>>>>>>>> run it
>>>>>>>>>>> as part of the :runners:direct-java:test task.
>>>>>>>>>>>
>>>>>>>>>>> That being said, we may only run these tests continuously with
>>>>>>>>>>> the DirectRunner, I'm not sure if we test them on all the runners
>>>>>>>>>>> like we
>>>>>>>>>>> do with ValidatesRunner tests.
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> That is correct. The tests are tests _of the transform_ so they
>>>>>>>>>> run only on the DirectRunner. They are not tests of the runner,
>>>>>>>>>> which is
>>>>>>>>>> only responsible for correctly implementing Beam's primitives. The
>>>>>>>>>> transform should not behave differently on different runners, except
>>>>>>>>>> for
>>>>>>>>>> fundamental differences in how they schedule work and checkpoint.
>>>>>>>>>>
>>>>>>>>>> Kenn
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>> > The error message I’m receiving, : Error while reading data,
>>>>>>>>>>> error message: JSON parsing error in row starting at position 0: No
>>>>>>>>>>> such
>>>>>>>>>>> field: nestedField.field1_0, suggests the BigQuery is trying to
>>>>>>>>>>> use the original name for the nested field and not the substitute
>>>>>>>>>>> name.
>>>>>>>>>>>
>>>>>>>>>>> Is there a stacktrace associated with this error? It would be
>>>>>>>>>>> helpful to see where the error is coming from.
>>>>>>>>>>>
>>>>>>>>>>> Brian
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> [1]
>>>>>>>>>>> https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/4101/testReport/org.apache.beam.sdk.schemas.transforms/RenameFieldsTest/
>>>>>>>>>>>
>>>>>>>>>>> On Mon, May 31, 2021 at 5:02 PM Matthew Ouyang <
>>>>>>>>>>> matthew.ouy...@gmail.com> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> I’m trying to use the RenameFields transform prior to inserting
>>>>>>>>>>>> into BigQuery on nested fields. Insertion into BigQuery is
>>>>>>>>>>>> successful with
>>>>>>>>>>>> DirectRunner, but DataflowRunner has an issue with renamed nested
>>>>>>>>>>>> fields
>>>>>>>>>>>> The error message I’m receiving, : Error while reading data,
>>>>>>>>>>>> error message: JSON parsing error in row starting at position 0:
>>>>>>>>>>>> No such
>>>>>>>>>>>> field: nestedField.field1_0, suggests the BigQuery is trying
>>>>>>>>>>>> to use the original name for the nested field and not the
>>>>>>>>>>>> substitute name.
>>>>>>>>>>>>
>>>>>>>>>>>> The code for RenameFields seems simple enough but does it
>>>>>>>>>>>> behave differently in different runners? Will a deep attachValues
>>>>>>>>>>>> be
>>>>>>>>>>>> necessary in order get the nested renames to work across all
>>>>>>>>>>>> runners? Is
>>>>>>>>>>>> there something wrong in my code?
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> https://github.com/apache/beam/blob/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/RenameFields.java#L186
>>>>>>>>>>>>
>>>>>>>>>>>> The unit tests also seem to be disabled for this as well and so
>>>>>>>>>>>> I don’t know if the PTransform behaves as expected.
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> https://github.com/apache/beam/blob/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/core/build.gradle#L67
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> https://github.com/apache/beam/blob/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/RenameFieldsTest.java
>>>>>>>>>>>>
>>>>>>>>>>>> package ca.loblaw.cerebro.PipelineControl;
>>>>>>>>>>>>>
>>>>>>>>>>>>> import com.google.api.services.bigquery.model.TableReference;
>>>>>>>>>>>>> import
>>>>>>>>>>>>> org.apache.beam.runners.dataflow.options.DataflowPipelineOptions
>>>>>>>>>>>>> ;
>>>>>>>>>>>>> import org.apache.beam.sdk.Pipeline;
>>>>>>>>>>>>> import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
>>>>>>>>>>>>> import org.apache.beam.sdk.options.PipelineOptionsFactory;
>>>>>>>>>>>>> import org.apache.beam.sdk.schemas.Schema;
>>>>>>>>>>>>> import org.apache.beam.sdk.schemas.transforms.RenameFields;
>>>>>>>>>>>>> import org.apache.beam.sdk.transforms.Create;
>>>>>>>>>>>>> import org.apache.beam.sdk.values.Row;
>>>>>>>>>>>>>
>>>>>>>>>>>>> import java.io.File;
>>>>>>>>>>>>> import java.util.Arrays;
>>>>>>>>>>>>> import java.util.HashSet;
>>>>>>>>>>>>> import java.util.stream.Collectors;
>>>>>>>>>>>>>
>>>>>>>>>>>>> import static java.util.Arrays.*asList*;
>>>>>>>>>>>>>
>>>>>>>>>>>>> public class BQRenameFields {
>>>>>>>>>>>>> public static void main(String[] args) {
>>>>>>>>>>>>> PipelineOptionsFactory.*register*(
>>>>>>>>>>>>> DataflowPipelineOptions.class);
>>>>>>>>>>>>> DataflowPipelineOptions options =
>>>>>>>>>>>>> PipelineOptionsFactory.*fromArgs*(args).as(
>>>>>>>>>>>>> DataflowPipelineOptions.class);
>>>>>>>>>>>>> options.setFilesToStage(
>>>>>>>>>>>>> Arrays.*stream*(System.*getProperty*(
>>>>>>>>>>>>> "java.class.path").
>>>>>>>>>>>>> split(File.*pathSeparator*)).
>>>>>>>>>>>>> map(entry -> (new
>>>>>>>>>>>>> File(entry)).toString()).collect(Collectors.*toList*()));
>>>>>>>>>>>>>
>>>>>>>>>>>>> Pipeline pipeline = Pipeline.*create*(options);
>>>>>>>>>>>>>
>>>>>>>>>>>>> Schema nestedSchema = Schema.*builder*().addField(
>>>>>>>>>>>>> Schema.Field.*nullable*("field1_0", Schema.FieldType.*STRING*
>>>>>>>>>>>>> )).build();
>>>>>>>>>>>>> Schema.Field field = Schema.Field.*nullable*(
>>>>>>>>>>>>> "field0_0", Schema.FieldType.*STRING*);
>>>>>>>>>>>>> Schema.Field nested = Schema.Field.*nullable*(
>>>>>>>>>>>>> "field0_1", Schema.FieldType.*row*(nestedSchema));
>>>>>>>>>>>>> Schema.Field runner = Schema.Field.*nullable*(
>>>>>>>>>>>>> "field0_2", Schema.FieldType.*STRING*);
>>>>>>>>>>>>> Schema rowSchema = Schema.*builder*()
>>>>>>>>>>>>> .addFields(field, nested, runner)
>>>>>>>>>>>>> .build();
>>>>>>>>>>>>> Row testRow = Row.*withSchema*(rowSchema
>>>>>>>>>>>>> ).attachValues("value0_0", Row.*withSchema*(nestedSchema
>>>>>>>>>>>>> ).attachValues("value1_0"), options.getRunner().toString());
>>>>>>>>>>>>> pipeline
>>>>>>>>>>>>> .apply(Create.*of*(testRow).withRowSchema(
>>>>>>>>>>>>> rowSchema))
>>>>>>>>>>>>> .apply(RenameFields.<Row>*create*()
>>>>>>>>>>>>> .rename("field0_0", "stringField")
>>>>>>>>>>>>> .rename("field0_1", "nestedField")
>>>>>>>>>>>>> .rename("field0_1.field1_0",
>>>>>>>>>>>>> "nestedStringField")
>>>>>>>>>>>>> .rename("field0_2", "runner"))
>>>>>>>>>>>>> .apply(BigQueryIO.<Row>*write*()
>>>>>>>>>>>>> .to(new TableReference().setProjectId(
>>>>>>>>>>>>> "lt-dia-lake-exp-raw").setDatasetId("prototypes").setTableId(
>>>>>>>>>>>>> "matto_renameFields"))
>>>>>>>>>>>>> .withCreateDisposition(BigQueryIO.
>>>>>>>>>>>>> Write.CreateDisposition.*CREATE_IF_NEEDED*)
>>>>>>>>>>>>> .withWriteDisposition(BigQueryIO.Write
>>>>>>>>>>>>> .WriteDisposition.*WRITE_APPEND*)
>>>>>>>>>>>>> .withSchemaUpdateOptions(new
>>>>>>>>>>>>> HashSet<>(*asList*(BigQueryIO.Write.SchemaUpdateOption.
>>>>>>>>>>>>> *ALLOW_FIELD_ADDITION*)))
>>>>>>>>>>>>> .useBeamSchema());
>>>>>>>>>>>>> pipeline.run();
>>>>>>>>>>>>> }
>>>>>>>>>>>>> }
>>>>>>>>>>>>>
>>>>>>>>>>>>
