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();
>>>>>>>> }
>>>>>>>> }
>>>>>>>>
>>>>>>>
