content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.beam.sdk.transforms;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertTrue;
  
  import com.google.auto.value.AutoValue;
  import java.io.Serializable;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.List;
  import java.util.Map;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  import org.apache.beam.sdk.coders.VarIntCoder;
  import org.apache.beam.sdk.schemas.AutoValueSchema;
  import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
  import org.apache.beam.sdk.schemas.NoSuchSchemaException;
  import org.apache.beam.sdk.schemas.Schema;
  import org.apache.beam.sdk.schemas.Schema.Field;
  import org.apache.beam.sdk.schemas.Schema.FieldType;
  import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
  import org.apache.beam.sdk.state.BagState;
  import org.apache.beam.sdk.state.CombiningState;
  import org.apache.beam.sdk.state.MapState;
  import org.apache.beam.sdk.state.SetState;
  import org.apache.beam.sdk.state.StateSpec;
  import org.apache.beam.sdk.state.StateSpecs;
  import org.apache.beam.sdk.testing.NeedsRunner;
  import org.apache.beam.sdk.testing.PAssert;
  import org.apache.beam.sdk.testing.TestPipeline;
  import org.apache.beam.sdk.testing.UsesMapState;
  import org.apache.beam.sdk.testing.UsesSchema;
  import org.apache.beam.sdk.testing.UsesSetState;
  import org.apache.beam.sdk.testing.UsesStatefulParDo;
  import org.apache.beam.sdk.testing.ValidatesRunner;
  import org.apache.beam.sdk.values.KV;
  import org.apache.beam.sdk.values.PCollection;
  import org.apache.beam.sdk.values.PCollectionTuple;
  import org.apache.beam.sdk.values.Row;
  import org.apache.beam.sdk.values.TupleTag;
  import org.apache.beam.sdk.values.TupleTagList;
  import org.apache.beam.sdk.values.TypeDescriptor;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.experimental.categories.Category;
  import org.junit.rules.ExpectedException;
  import org.junit.rules.Timeout;
  import org.junit.runner.RunWith;
  import org.junit.runners.JUnit4;
  
  /** Test {@link Schema} support. */
  @RunWith(JUnit4.class)
  @Category(UsesSchema.class)
  // TODO(https://github.com/apache/beam/issues/21230): Remove when new version of errorprone is
  // released (2.11.0)
  @SuppressWarnings("unused")
  public class ParDoSchemaTest implements Serializable {
    @Rule public final transient TestPipeline pipeline = TestPipeline.create();
    @Rule public transient ExpectedException thrown = ExpectedException.none();
    @Rule public transient Timeout globalTimeout = Timeout.seconds(1200);
  
    static class MyPojo implements Serializable {
      MyPojo(String stringField, Integer integerField) {
        this.stringField = stringField;
        this.integerField = integerField;
      }
  
      String stringField;
      Integer integerField;
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testSimpleSchemaPipeline() {
      List<MyPojo> pojoList =
          Lists.newArrayList(new MyPojo("a", 1), new MyPojo("b", 2), new MyPojo("c", 3));
  
      Schema schema =
          Schema.builder().addStringField("string_field").addInt32Field("integer_field").build();
  
      PCollection<String> output =
          pipeline
              .apply(
                  Create.of(pojoList)
                      .withSchema(
                          schema,
                          TypeDescriptor.of(MyPojo.class),
                          o ->
                              Row.withSchema(schema).addValues(o.stringField, o.integerField).build(),
                          r -> new MyPojo(r.getString("string_field"), r.getInt32("integer_field"))))
              .apply(
                  ParDo.of(
                      new DoFn<MyPojo, String>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<String> r) {
                          r.output(row.getString(0) + ":" + row.getInt32(1));
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testReadAndWrite() {
      List<MyPojo> pojoList =
          Lists.newArrayList(new MyPojo("a", 1), new MyPojo("b", 2), new MyPojo("c", 3));
  
      Schema schema1 =
          Schema.builder().addStringField("string_field").addInt32Field("integer_field").build();
  
      Schema schema2 =
          Schema.builder().addStringField("string2_field").addInt32Field("integer2_field").build();
  
      PCollection<String> output =
          pipeline
              .apply(
                  Create.of(pojoList)
                      .withSchema(
                          schema1,
                          TypeDescriptor.of(MyPojo.class),
                          o ->
                              Row.withSchema(schema1)
                                  .addValues(o.stringField, o.integerField)
                                  .build(),
                          r -> new MyPojo(r.getString("string_field"), r.getInt32("integer_field"))))
              .apply(
                  "first",
                  ParDo.of(
                      new DoFn<MyPojo, MyPojo>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<Row> r) {
                          r.output(
                              Row.withSchema(schema2)
                                  .addValues(row.getString(0), row.getInt32(1))
                                  .build());
                        }
                      }))
              .setSchema(
                  schema2,
                  TypeDescriptor.of(MyPojo.class),
                  o -> Row.withSchema(schema2).addValues(o.stringField, o.integerField).build(),
                  r -> new MyPojo(r.getString("string2_field"), r.getInt32("integer2_field")))
              .apply(
                  "second",
                  ParDo.of(
                      new DoFn<MyPojo, String>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<String> r) {
                          r.output(row.getString(0) + ":" + row.getInt32(1));
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testReadAndWriteMultiOutput() {
      List<MyPojo> pojoList =
          Lists.newArrayList(new MyPojo("a", 1), new MyPojo("b", 2), new MyPojo("c", 3));
  
      Schema schema1 =
          Schema.builder().addStringField("string_field").addInt32Field("integer_field").build();
  
      Schema schema2 =
          Schema.builder().addStringField("string2_field").addInt32Field("integer2_field").build();
  
      Schema schema3 =
          Schema.builder().addStringField("string3_field").addInt32Field("integer3_field").build();
  
      TupleTag<MyPojo> firstOutput = new TupleTag<>("first");
      TupleTag<MyPojo> secondOutput = new TupleTag<>("second");
  
      PCollectionTuple tuple =
          pipeline
              .apply(
                  Create.of(pojoList)
                      .withSchema(
                          schema1,
                          TypeDescriptor.of(MyPojo.class),
                          o ->
                              Row.withSchema(schema1)
                                  .addValues(o.stringField, o.integerField)
                                  .build(),
                          r -> new MyPojo(r.getString("string_field"), r.getInt32("integer_field"))))
              .apply(
                  "first",
                  ParDo.of(
                          new DoFn<MyPojo, MyPojo>() {
                            @ProcessElement
                            public void process(@Element Row row, MultiOutputReceiver r) {
                              r.getRowReceiver(firstOutput)
                                  .output(
                                      Row.withSchema(schema2)
                                          .addValues(row.getString(0), row.getInt32(1))
                                          .build());
                              r.getRowReceiver(secondOutput)
                                  .output(
                                      Row.withSchema(schema3)
                                          .addValues(row.getString(0), row.getInt32(1))
                                          .build());
                            }
                          })
                      .withOutputTags(firstOutput, TupleTagList.of(secondOutput)));
      tuple
          .get(firstOutput)
          .setSchema(
              schema2,
              TypeDescriptor.of(MyPojo.class),
              o -> Row.withSchema(schema2).addValues(o.stringField, o.integerField).build(),
              r -> new MyPojo(r.getString("string2_field"), r.getInt32("integer2_field")));
      tuple
          .get(secondOutput)
          .setSchema(
              schema3,
              TypeDescriptor.of(MyPojo.class),
              o -> Row.withSchema(schema3).addValues(o.stringField, o.integerField).build(),
              r -> new MyPojo(r.getString("string3_field"), r.getInt32("integer3_field")));
  
      PCollection<String> output1 =
          tuple
              .get(firstOutput)
              .apply(
                  "second",
                  ParDo.of(
                      new DoFn<MyPojo, String>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<String> r) {
                          r.output(
                              row.getString("string2_field") + ":" + row.getInt32("integer2_field"));
                        }
                      }));
  
      PCollection<String> output2 =
          tuple
              .get(secondOutput)
              .apply(
                  "third",
                  ParDo.of(
                      new DoFn<MyPojo, String>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<String> r) {
                          r.output(
                              row.getString("string3_field") + ":" + row.getInt32("integer3_field"));
                        }
                      }));
  
      PAssert.that(output1).containsInAnyOrder("a:1", "b:2", "c:3");
      PAssert.that(output2).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testReadAndWriteWithSchemaRegistry() {
      Schema schema =
          Schema.builder().addStringField("string_field").addInt32Field("integer_field").build();
  
      pipeline
          .getSchemaRegistry()
          .registerSchemaForClass(
              MyPojo.class,
              schema,
              o -> Row.withSchema(schema).addValues(o.stringField, o.integerField).build(),
              r -> new MyPojo(r.getString("string_field"), r.getInt32("integer_field")));
  
      List<MyPojo> pojoList =
          Lists.newArrayList(new MyPojo("a", 1), new MyPojo("b", 2), new MyPojo("c", 3));
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(pojoList))
              .apply(
                  "first",
                  ParDo.of(
                      new DoFn<MyPojo, MyPojo>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<Row> r) {
                          r.output(
                              Row.withSchema(schema)
                                  .addValues(row.getString(0), row.getInt32(1))
                                  .build());
                        }
                      }))
              .apply(
                  "second",
                  ParDo.of(
                      new DoFn<MyPojo, String>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<String> r) {
                          r.output(row.getString(0) + ":" + row.getInt32(1));
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testFieldAccessSchemaPipeline() {
      List<MyPojo> pojoList =
          Lists.newArrayList(new MyPojo("a", 1), new MyPojo("b", 2), new MyPojo("c", 3));
  
      Schema schema =
          Schema.builder().addStringField("string_field").addInt32Field("integer_field").build();
  
      PCollection<String> output =
          pipeline
              .apply(
                  Create.of(pojoList)
                      .withSchema(
                          schema,
                          TypeDescriptor.of(MyPojo.class),
                          o ->
                              Row.withSchema(schema).addValues(o.stringField, o.integerField).build(),
                          r -> new MyPojo(r.getString("string_field"), r.getInt32("integer_field"))))
              .apply(
                  ParDo.of(
                      new DoFn<MyPojo, String>() {
                        @FieldAccess("foo")
                        final FieldAccessDescriptor fieldAccess =
                            FieldAccessDescriptor.withAllFields();
  
                        @ProcessElement
                        public void process(@FieldAccess("foo") Row row, OutputReceiver<String> r) {
                          r.output(row.getString(0) + ":" + row.getInt32(1));
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @Test
    @Category(NeedsRunner.class)
    public void testNoSchema() {
      thrown.expect(IllegalArgumentException.class);
      pipeline
          .apply(Create.of("a", "b", "c"))
          .apply(
              ParDo.of(
                  new DoFn<String, Void>() {
                    @ProcessElement
                    public void process(@Element Row row) {}
                  }));
      pipeline.run();
    }
  
    @Test
    @Category(NeedsRunner.class)
    public void testUnmatchedSchema() {
      List<MyPojo> pojoList =
          Lists.newArrayList(new MyPojo("a", 1), new MyPojo("b", 2), new MyPojo("c", 3));
  
      Schema schema =
          Schema.builder().addStringField("string_field").addInt32Field("integer_field").build();
  
      thrown.expect(IllegalArgumentException.class);
      pipeline
          .apply(
              Create.of(pojoList)
                  .withSchema(
                      schema,
                      TypeDescriptor.of(MyPojo.class),
                      o -> Row.withSchema(schema).addValues(o.stringField, o.integerField).build(),
                      r -> new MyPojo(r.getString("string_field"), r.getInt32("integer_field"))))
          .apply(
              ParDo.of(
                  new DoFn<MyPojo, Void>() {
                    @FieldAccess("a")
                    FieldAccessDescriptor fieldAccess = FieldAccessDescriptor.withFieldNames("baad");
  
                    @ProcessElement
                    public void process(@FieldAccess("a") Row row) {}
                  }));
      pipeline.run();
    }
  
    /** POJO used for testing. */
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class Inferred {
      abstract String getStringField();
  
      abstract Integer getIntegerField();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testInferredSchemaPipeline() {
      List<Inferred> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_Inferred("a", 1),
              new AutoValue_ParDoSchemaTest_Inferred("b", 2),
              new AutoValue_ParDoSchemaTest_Inferred("c", 3));
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(pojoList))
              .apply(
                  ParDo.of(
                      new DoFn<Inferred, String>() {
                        @ProcessElement
                        public void process(@Element Row row, OutputReceiver<String> r) {
                          r.output(row.getString("stringField") + ":" + row.getInt32("integerField"));
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testSchemasPassedThrough() {
      List<Inferred> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_Inferred("a", 1),
              new AutoValue_ParDoSchemaTest_Inferred("b", 2),
              new AutoValue_ParDoSchemaTest_Inferred("c", 3));
  
      PCollection<Inferred> out = pipeline.apply(Create.of(pojoList)).apply(Filter.by(e -> true));
      assertTrue(out.hasSchema());
  
      pipeline.run();
    }
  
    /** Pojo used for testing. */
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class Inferred2 {
      abstract Integer getIntegerField();
  
      abstract String getStringField();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testSchemaConversionPipeline() {
      List<Inferred> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_Inferred("a", 1),
              new AutoValue_ParDoSchemaTest_Inferred("b", 2),
              new AutoValue_ParDoSchemaTest_Inferred("c", 3));
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(pojoList))
              .apply(
                  ParDo.of(
                      new DoFn<Inferred, String>() {
                        @ProcessElement
                        public void process(@Element Inferred2 pojo, OutputReceiver<String> r) {
                          r.output(pojo.getStringField() + ":" + pojo.getIntegerField());
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class Nested {
      abstract int getField1();
  
      abstract Inferred getInner();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testNestedSchema() {
      List<Nested> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_Nested(1, new AutoValue_ParDoSchemaTest_Inferred("a", 1)),
              new AutoValue_ParDoSchemaTest_Nested(2, new AutoValue_ParDoSchemaTest_Inferred("b", 2)),
              new AutoValue_ParDoSchemaTest_Nested(
                  3, new AutoValue_ParDoSchemaTest_Inferred("c", 3)));
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(pojoList))
              .apply(WithKeys.of("foo"))
              .apply(Reshuffle.of())
              .apply(Values.create())
              .apply(
                  ParDo.of(
                      new DoFn<Nested, String>() {
                        @ProcessElement
                        public void process(@Element Nested nested, OutputReceiver<String> r) {
                          r.output(
                              nested.getInner().getStringField()
                                  + ":"
                                  + nested.getInner().getIntegerField());
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1", "b:2", "c:3");
      pipeline.run();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class ForExtraction {
      abstract Integer getIntegerField();
  
      abstract String getStringField();
  
      abstract List<Integer> getInts();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testSchemaFieldSelectionUnboxing() {
      List<ForExtraction> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_ForExtraction(1, "a", Lists.newArrayList(1, 2)),
              new AutoValue_ParDoSchemaTest_ForExtraction(2, "b", Lists.newArrayList(2, 3)),
              new AutoValue_ParDoSchemaTest_ForExtraction(3, "c", Lists.newArrayList(3, 4)));
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(pojoList))
              .apply(
                  ParDo.of(
                      new DoFn<ForExtraction, String>() {
                        // Read the list twice as two equivalent types to ensure that Beam properly
                        // converts.
                        @ProcessElement
                        public void process(
                            @FieldAccess("stringField") String stringField,
                            @FieldAccess("integerField") Integer integerField,
                            @FieldAccess("ints") Integer[] intArray,
                            @FieldAccess("ints") List<Integer> intList,
                            OutputReceiver<String> r) {
  
                          r.output(
                              stringField
                                  + ":"
                                  + integerField
                                  + ":"
                                  + Arrays.toString(intArray)
                                  + ":"
                                  + intList.toString());
                        }
                      }));
      PAssert.that(output)
          .containsInAnyOrder("a:1:[1, 2]:[1, 2]", "b:2:[2, 3]:[2, 3]", "c:3:[3, 4]:[3, 4]");
      pipeline.run();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testSchemaFieldDescriptorSelectionUnboxing() {
      List<ForExtraction> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_ForExtraction(1, "a", Lists.newArrayList(1, 2)),
              new AutoValue_ParDoSchemaTest_ForExtraction(2, "b", Lists.newArrayList(2, 3)),
              new AutoValue_ParDoSchemaTest_ForExtraction(3, "c", Lists.newArrayList(3, 4)));
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(pojoList))
              .apply(
                  ParDo.of(
                      new DoFn<ForExtraction, String>() {
                        @FieldAccess("stringSelector")
                        final FieldAccessDescriptor stringSelector =
                            FieldAccessDescriptor.withFieldNames("stringField");
  
                        @FieldAccess("intSelector")
                        final FieldAccessDescriptor intSelector =
                            FieldAccessDescriptor.withFieldNames("integerField");
  
                        @FieldAccess("intsSelector")
                        final FieldAccessDescriptor intsSelector =
                            FieldAccessDescriptor.withFieldNames("ints");
  
                        @ProcessElement
                        public void process(
                            @FieldAccess("stringSelector") String stringField,
                            @FieldAccess("intSelector") int integerField,
                            @FieldAccess("intsSelector") int[] intArray,
                            OutputReceiver<String> r) {
                          r.output(
                              stringField + ":" + integerField + ":" + Arrays.toString(intArray));
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1:[1, 2]", "b:2:[2, 3]", "c:3:[3, 4]");
      pipeline.run();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class NestedForExtraction {
      abstract ForExtraction getInner();
    }
  
    @Test
    @Category(ValidatesRunner.class)
    public void testSchemaFieldSelectionNested() {
      List<ForExtraction> pojoList =
          Lists.newArrayList(
              new AutoValue_ParDoSchemaTest_ForExtraction(1, "a", Lists.newArrayList(1, 2)),
              new AutoValue_ParDoSchemaTest_ForExtraction(2, "b", Lists.newArrayList(2, 3)),
              new AutoValue_ParDoSchemaTest_ForExtraction(3, "c", Lists.newArrayList(3, 4)));
      List<NestedForExtraction> outerList =
          pojoList.stream()
              .map(AutoValue_ParDoSchemaTest_NestedForExtraction::new)
              .collect(Collectors.toList());
  
      PCollection<String> output =
          pipeline
              .apply(Create.of(outerList))
              .apply(
                  ParDo.of(
                      new DoFn<NestedForExtraction, String>() {
  
                        @ProcessElement
                        public void process(
                            @FieldAccess("inner.*") ForExtraction extracted,
                            @FieldAccess("inner") ForExtraction extracted1,
                            @FieldAccess("inner.stringField") String stringField,
                            @FieldAccess("inner.integerField") int integerField,
                            @FieldAccess("inner.ints") List<Integer> intArray,
                            OutputReceiver<String> r) {
                          assertEquals(extracted, extracted1);
                          assertEquals(stringField, extracted.getStringField());
                          assertEquals(integerField, (int) extracted.getIntegerField());
                          assertEquals(intArray, extracted.getInts());
                          r.output(
                              extracted.getStringField()
                                  + ":"
                                  + extracted.getIntegerField()
                                  + ":"
                                  + extracted.getInts().toString());
                        }
                      }));
      PAssert.that(output).containsInAnyOrder("a:1:[1, 2]", "b:2:[2, 3]", "c:3:[3, 4]");
      pipeline.run();
    }
  
    @Test
    @Category({NeedsRunner.class, UsesStatefulParDo.class})
    public void testRowBagState() {
      final String stateId = "foo";
  
      Schema type =
          Stream.of(Schema.Field.of("f_string", FieldType.STRING)).collect(Schema.toSchema());
      Schema outputType = Schema.of(Field.of("values", FieldType.array(FieldType.row(type))));
  
      DoFn<KV<String, Row>, Row> fn =
          new DoFn<KV<String, Row>, Row>() {
  
            @StateId(stateId)
            private final StateSpec<BagState<Row>> bufferState = StateSpecs.rowBag(type);
  
            @ProcessElement
            public void processElement(
                @Element KV<String, Row> element,
                @StateId(stateId) BagState<Row> state,
                OutputReceiver<Row> o) {
              state.add(element.getValue());
              Iterable<Row> currentValue = state.read();
              if (Iterables.size(currentValue) >= 4) {
                List<Row> sorted = Lists.newArrayList(currentValue);
                Collections.sort(sorted, Comparator.comparing(r -> r.getString(0)));
                o.output(Row.withSchema(outputType).addArray(sorted).build());
              }
            }
          };
  
      PCollection<Row> output =
          pipeline
              .apply(
                  Create.of(
                      KV.of("hello", Row.withSchema(type).addValue("a").build()),
                      KV.of("hello", Row.withSchema(type).addValue("b").build()),
                      KV.of("hello", Row.withSchema(type).addValue("c").build()),
                      KV.of("hello", Row.withSchema(type).addValue("d").build())))
              .apply(ParDo.of(fn))
              .setRowSchema(outputType);
      PAssert.that(output)
          .containsInAnyOrder(
              Row.withSchema(outputType)
                  .addArray(
                      Lists.newArrayList(
                          Row.withSchema(type).addValue("a").build(),
                          Row.withSchema(type).addValue("b").build(),
                          Row.withSchema(type).addValue("c").build(),
                          Row.withSchema(type).addValue("d").build()))
                  .build());
  
      pipeline.run();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class TestStateSchemaValue {
      abstract String getName();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class TestStateSchemaValues {
      abstract List<TestStateSchemaValue> getValues();
    }
  
    @Test
    @Category({NeedsRunner.class, UsesStatefulParDo.class})
    public void tesBagStateSchemaInference() throws NoSuchSchemaException {
      final String stateId = "foo";
  
      DoFn<KV<String, TestStateSchemaValue>, TestStateSchemaValues> fn =
          new DoFn<KV<String, TestStateSchemaValue>, TestStateSchemaValues>() {
  
            // This should infer the schema.
  
            @StateId(stateId)
            private final StateSpec<BagState<TestStateSchemaValue>> bufferState = StateSpecs.bag();
  
            @ProcessElement
            public void processElement(
                @Element KV<String, TestStateSchemaValue> element,
                @StateId(stateId) BagState<TestStateSchemaValue> state,
                OutputReceiver<TestStateSchemaValues> o) {
              state.add(element.getValue());
              Iterable<TestStateSchemaValue> currentValue = state.read();
              if (Iterables.size(currentValue) >= 4) {
                List<TestStateSchemaValue> sorted = Lists.newArrayList(currentValue);
                Collections.sort(sorted, Comparator.comparing(TestStateSchemaValue::getName));
                o.output(new AutoValue_ParDoSchemaTest_TestStateSchemaValues(sorted));
              }
            }
          };
  
      PCollection<TestStateSchemaValue> input =
          pipeline.apply(
              Create.of(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("a"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("b"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("c"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("d")));
  
      PCollection<TestStateSchemaValues> output =
          input.apply(WithKeys.of("hello")).apply(ParDo.of(fn));
  
      PAssert.that(output)
          .containsInAnyOrder(
              new AutoValue_ParDoSchemaTest_TestStateSchemaValues(
                  Lists.newArrayList(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("a"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("b"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("c"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("d"))));
  
      pipeline.run();
    }
  
    @Test
    @Category({NeedsRunner.class, UsesStatefulParDo.class, UsesSetState.class})
    public void testSetStateSchemaInference() throws NoSuchSchemaException {
      final String stateId = "foo";
  
      DoFn<KV<String, TestStateSchemaValue>, TestStateSchemaValues> fn =
          new DoFn<KV<String, TestStateSchemaValue>, TestStateSchemaValues>() {
  
            // This should infer the schema.
  
            @StateId(stateId)
            private final StateSpec<SetState<TestStateSchemaValue>> bufferState = StateSpecs.set();
  
            @ProcessElement
            public void processElement(
                @Element KV<String, TestStateSchemaValue> element,
                @StateId(stateId) SetState<TestStateSchemaValue> state,
                OutputReceiver<TestStateSchemaValues> o) {
              state.add(element.getValue());
              Iterable<TestStateSchemaValue> currentValue = state.read();
              if (Iterables.size(currentValue) >= 4) {
                List<TestStateSchemaValue> sorted = Lists.newArrayList(currentValue);
                Collections.sort(sorted, Comparator.comparing(TestStateSchemaValue::getName));
                o.output(new AutoValue_ParDoSchemaTest_TestStateSchemaValues(sorted));
              }
            }
          };
  
      PCollection<TestStateSchemaValue> input =
          pipeline.apply(
              Create.of(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("a"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("b"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("c"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("d")));
  
      PCollection<TestStateSchemaValues> output =
          input.apply(WithKeys.of("hello")).apply(ParDo.of(fn));
  
      PAssert.that(output)
          .containsInAnyOrder(
              new AutoValue_ParDoSchemaTest_TestStateSchemaValues(
                  Lists.newArrayList(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("a"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("b"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("c"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("d"))));
  
      pipeline.run();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class TestStateSchemaValue2 {
      abstract Integer getInteger();
    }
  
    @DefaultSchema(AutoValueSchema.class)
    @AutoValue
    abstract static class TestStateSchemaMapEntry {
      abstract TestStateSchemaValue getKey();
  
      abstract TestStateSchemaValue2 getValue();
    }
  
    @Test
    @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesMapState.class})
    public void testMapStateSchemaInference() throws NoSuchSchemaException {
      final String stateId = "foo";
      final String countStateId = "count";
  
      DoFn<KV<String, TestStateSchemaMapEntry>, TestStateSchemaMapEntry> fn =
          new DoFn<KV<String, TestStateSchemaMapEntry>, TestStateSchemaMapEntry>() {
  
            @StateId(stateId)
            private final StateSpec<MapState<TestStateSchemaValue, TestStateSchemaValue2>> mapState =
                StateSpecs.map();
  
            @StateId(countStateId)
            private final StateSpec<CombiningState<Integer, int[], Integer>> countState =
                StateSpecs.combiningFromInputInternal(VarIntCoder.of(), Sum.ofIntegers());
  
            @ProcessElement
            public void processElement(
                @Element KV<String, TestStateSchemaMapEntry> element,
                @StateId(stateId) MapState<TestStateSchemaValue, TestStateSchemaValue2> state,
                @StateId(countStateId) CombiningState<Integer, int[], Integer> count,
                OutputReceiver<TestStateSchemaMapEntry> o) {
              TestStateSchemaMapEntry value = element.getValue();
              state.put(value.getKey(), value.getValue());
              count.add(1);
              if (count.read() >= 4) {
                Iterable<Map.Entry<TestStateSchemaValue, TestStateSchemaValue2>> iterate =
                    state.entries().read();
                for (Map.Entry<TestStateSchemaValue, TestStateSchemaValue2> entry : iterate) {
                  o.output(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                          entry.getKey(), entry.getValue()));
                }
              }
            }
          };
  
      PCollection<TestStateSchemaMapEntry> input =
          pipeline.apply(
              Create.of(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("a"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(1)),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("b"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(2)),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("c"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(3)),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue("d"),
                      new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(4))));
  
      PCollection<TestStateSchemaMapEntry> output =
          input.apply(WithKeys.of("hello")).apply(ParDo.of(fn));
  
      PAssert.that(output)
          .containsInAnyOrder(
              new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("a"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(1)),
              new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("b"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(2)),
              new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("c"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(3)),
              new AutoValue_ParDoSchemaTest_TestStateSchemaMapEntry(
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue("d"),
                  new AutoValue_ParDoSchemaTest_TestStateSchemaValue2(4)));
      pipeline.run();
    }
  }
  