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.schemas.utils;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNull;
  
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
  import org.apache.beam.sdk.schemas.Schema;
  import org.apache.beam.sdk.schemas.Schema.FieldType;
  import org.apache.beam.sdk.values.Row;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
  import org.junit.Test;
  import org.junit.runner.RunWith;
  import org.junit.runners.Parameterized;
  import org.junit.runners.Parameterized.Parameters;
  
  /** Tests for {@link SelectHelpers}. */
  @RunWith(Parameterized.class)
  public class SelectHelpersTest {
    @Parameterized.Parameter public boolean useOptimizedSelect;
  
    @Parameters
    public static Collection<Object[]> data() {
      return Arrays.asList(new Object[][] {{false}, {true}});
    }
  
    static final Schema FLAT_SCHEMA =
        Schema.builder()
            .addStringField("field1")
            .addInt32Field("field2")
            .addDoubleField("field3")
            .addStringField("field_extra")
            .build();
    static final Row FLAT_ROW =
        Row.withSchema(FLAT_SCHEMA).addValues("first", 42, 3.14, "extra").build();
  
    static final Schema NESTED_SCHEMA =
        Schema.builder().addRowField("nested", FLAT_SCHEMA).addStringField("foo").build();
    static final Row NESTED_ROW = Row.withSchema(NESTED_SCHEMA).addValues(FLAT_ROW, "").build();
  
    static final Schema DOUBLE_NESTED_SCHEMA =
        Schema.builder().addRowField("nested2", NESTED_SCHEMA).build();
    static final Row DOUBLE_NESTED_ROW =
        Row.withSchema(DOUBLE_NESTED_SCHEMA).addValue(NESTED_ROW).build();
  
    static final Schema ARRAY_SCHEMA =
        Schema.builder()
            .addArrayField("primitiveArray", FieldType.INT32)
            .addArrayField("rowArray", FieldType.row(FLAT_SCHEMA))
            .addArrayField("arrayOfRowArray", FieldType.array(FieldType.row(FLAT_SCHEMA)))
            .addArrayField("nestedRowArray", FieldType.row(NESTED_SCHEMA))
            .build();
    static final Row ARRAY_ROW =
        Row.withSchema(ARRAY_SCHEMA)
            .addArray(1, 2)
            .addArray(FLAT_ROW, FLAT_ROW)
            .addArray(ImmutableList.of(FLAT_ROW), ImmutableList.of(FLAT_ROW))
            .addArray(NESTED_ROW, NESTED_ROW)
            .build();
  
    static final Schema ITERABLE_SCHEMA =
        Schema.builder()
            .addIterableField("primitiveIter", FieldType.INT32)
            .addIterableField("rowIter", FieldType.row(FLAT_SCHEMA))
            .addIterableField("iterOfRowIter", FieldType.iterable(FieldType.row(FLAT_SCHEMA)))
            .addIterableField("nestedRowIter", FieldType.row(NESTED_SCHEMA))
            .build();
    static final Row ITERABLE_ROW =
        Row.withSchema(ARRAY_SCHEMA)
            .addIterable(ImmutableList.of(1, 2))
            .addIterable(ImmutableList.of(FLAT_ROW, FLAT_ROW))
            .addIterable(ImmutableList.of(ImmutableList.of(FLAT_ROW), ImmutableList.of(FLAT_ROW)))
            .addIterable(ImmutableList.of(NESTED_ROW, NESTED_ROW))
            .build();
    static final Schema MAP_SCHEMA =
        Schema.builder().addMapField("map", FieldType.INT32, FieldType.row(FLAT_SCHEMA)).build();
    static final Row MAP_ROW =
        Row.withSchema(MAP_SCHEMA).addValue(ImmutableMap.of(1, FLAT_ROW)).build();
  
    static final Schema MAP_ARRAY_SCHEMA =
        Schema.builder()
            .addMapField("map", FieldType.INT32, FieldType.array(FieldType.row(FLAT_SCHEMA)))
            .build();
    static final Row MAP_ARRAY_ROW =
        Row.withSchema(MAP_ARRAY_SCHEMA)
            .addValue(ImmutableMap.of(1, ImmutableList.of(FLAT_ROW)))
            .build();
  
    static final Schema MAP_ITERABLE_SCHEMA =
        Schema.builder()
            .addMapField("map", FieldType.INT32, FieldType.iterable(FieldType.row(FLAT_SCHEMA)))
            .build();
    static final Row MAP_ITERABLE_ROW =
        Row.withSchema(MAP_ITERABLE_SCHEMA)
            .addValue(ImmutableMap.of(1, ImmutableList.of(FLAT_ROW)))
            .build();
  
    Row selectRow(Schema inputScema, FieldAccessDescriptor fieldAccessDescriptor, Row row) {
      RowSelector rowSelector =
          useOptimizedSelect
              ? SelectHelpers.getRowSelectorOptimized(inputScema, fieldAccessDescriptor)
              : SelectHelpers.getRowSelector(inputScema, fieldAccessDescriptor);
      return rowSelector.select(row);
    }
  
    @Test
    public void testSelectAll() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("*").resolve(FLAT_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(FLAT_SCHEMA, fieldAccessDescriptor);
      assertEquals(FLAT_SCHEMA, outputSchema);
  
      Row row = selectRow(FLAT_SCHEMA, fieldAccessDescriptor, FLAT_ROW);
      assertEquals(FLAT_ROW, row);
    }
  
    @Test
    public void testsSimpleSelectSingle() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("field1").resolve(FLAT_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(FLAT_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema = Schema.builder().addStringField("field1").build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(FLAT_SCHEMA, fieldAccessDescriptor, FLAT_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValue("first").build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testsSimpleSelectSingleWithUnderscore() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("field_extra").resolve(FLAT_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(FLAT_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema = Schema.builder().addStringField("field_extra").build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(FLAT_SCHEMA, fieldAccessDescriptor, FLAT_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValue("extra").build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testsSimpleSelectMultiple() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("field1", "field3").resolve(FLAT_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(FLAT_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addStringField("field1").addDoubleField("field3").build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(FLAT_SCHEMA, fieldAccessDescriptor, FLAT_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValues("first", 3.14).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectedNested() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("nested").resolve(NESTED_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(NESTED_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema = Schema.builder().addRowField("nested", FLAT_SCHEMA).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(NESTED_SCHEMA, fieldAccessDescriptor, NESTED_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValue(FLAT_ROW).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectedNestedSingle() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("nested.field1").resolve(NESTED_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(NESTED_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema = Schema.builder().addStringField("field1").build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(NESTED_SCHEMA, fieldAccessDescriptor, NESTED_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValue("first").build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectedNestedWildcard() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("nested.*").resolve(NESTED_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(NESTED_SCHEMA, fieldAccessDescriptor);
      assertEquals(FLAT_SCHEMA, outputSchema);
  
      Row row = selectRow(NESTED_SCHEMA, fieldAccessDescriptor, NESTED_ROW);
      assertEquals(FLAT_ROW, row);
    }
  
    @Test
    public void testSelectDoubleNested() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("nested2.nested.field1").resolve(DOUBLE_NESTED_SCHEMA);
      Schema outputSchema =
          SelectHelpers.getOutputSchema(DOUBLE_NESTED_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema = Schema.builder().addStringField("field1").build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(DOUBLE_NESTED_SCHEMA, fieldAccessDescriptor, DOUBLE_NESTED_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValue("first").build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectArrayOfPrimitive() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("primitiveArray").resolve(ARRAY_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ARRAY_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addArrayField("primitiveArray", FieldType.INT32).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ARRAY_SCHEMA, fieldAccessDescriptor, ARRAY_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addArray(1, 2).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectIterableOfPrimitive() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("primitiveIter").resolve(ITERABLE_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ITERABLE_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addIterableField("primitiveIter", FieldType.INT32).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ITERABLE_SCHEMA, fieldAccessDescriptor, ITERABLE_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addIterable(ImmutableList.of(1, 2)).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectArrayOfRow() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("rowArray").resolve(ARRAY_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ARRAY_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addArrayField("rowArray", FieldType.row(FLAT_SCHEMA)).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ARRAY_SCHEMA, fieldAccessDescriptor, ARRAY_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addArray(FLAT_ROW, FLAT_ROW).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectIterableOfRow() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("rowIter").resolve(ITERABLE_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ITERABLE_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addIterableField("rowIter", FieldType.row(FLAT_SCHEMA)).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ITERABLE_SCHEMA, fieldAccessDescriptor, ITERABLE_ROW);
      Row expectedRow =
          Row.withSchema(expectedSchema).addIterable(ImmutableList.of(FLAT_ROW, FLAT_ROW)).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectArrayOfRowPartial() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("rowArray[].field1").resolve(ARRAY_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ARRAY_SCHEMA, fieldAccessDescriptor);
  
      Schema expectedSchema = Schema.builder().addArrayField("field1", FieldType.STRING).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ARRAY_SCHEMA, fieldAccessDescriptor, ARRAY_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addArray("first", "first").build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectIterableOfRowPartial() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("rowIter[].field1").resolve(ITERABLE_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ITERABLE_SCHEMA, fieldAccessDescriptor);
  
      Schema expectedSchema = Schema.builder().addIterableField("field1", FieldType.STRING).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ITERABLE_SCHEMA, fieldAccessDescriptor, ITERABLE_ROW);
      Row expectedRow =
          Row.withSchema(expectedSchema).addIterable(ImmutableList.of("first", "first")).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectArrayOfRowArray() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("arrayOfRowArray[][].field1").resolve(ARRAY_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ARRAY_SCHEMA, fieldAccessDescriptor);
  
      Schema expectedSchema =
          Schema.builder().addArrayField("field1", FieldType.array(FieldType.STRING)).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ARRAY_SCHEMA, fieldAccessDescriptor, ARRAY_ROW);
  
      Row expectedRow =
          Row.withSchema(expectedSchema)
              .addArray(ImmutableList.of("first"), ImmutableList.of("first"))
              .build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectIterableOfRowIterable() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("iterOfRowIter[][].field1").resolve(ITERABLE_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ITERABLE_SCHEMA, fieldAccessDescriptor);
  
      Schema expectedSchema =
          Schema.builder().addIterableField("field1", FieldType.iterable(FieldType.STRING)).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ITERABLE_SCHEMA, fieldAccessDescriptor, ITERABLE_ROW);
  
      Row expectedRow =
          Row.withSchema(expectedSchema)
              .addIterable(ImmutableList.of(ImmutableList.of("first"), ImmutableList.of("first")))
              .build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectArrayOfNestedRow() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("nestedRowArray[].nested.field1")
              .resolve(ARRAY_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ARRAY_SCHEMA, fieldAccessDescriptor);
  
      Schema.builder().addStringField("field1").build();
      Schema expectedSchema = Schema.builder().addArrayField("field1", FieldType.STRING).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ARRAY_SCHEMA, fieldAccessDescriptor, ARRAY_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addArray("first", "first").build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectIterableOfNestedRow() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("nestedRowIter[].nested.field1")
              .resolve(ITERABLE_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(ITERABLE_SCHEMA, fieldAccessDescriptor);
  
      Schema.builder().addStringField("field1").build();
      Schema expectedSchema = Schema.builder().addIterableField("field1", FieldType.STRING).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(ITERABLE_SCHEMA, fieldAccessDescriptor, ITERABLE_ROW);
      Row expectedRow =
          Row.withSchema(expectedSchema).addIterable(ImmutableList.of("first", "first")).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectMapOfRowSelectSingle() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("map{}.field1").resolve(MAP_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(MAP_SCHEMA, fieldAccessDescriptor);
  
      Schema.builder().addStringField("field1").build();
      Schema expectedSchema =
          Schema.builder().addMapField("field1", FieldType.INT32, FieldType.STRING).build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(MAP_SCHEMA, fieldAccessDescriptor, MAP_ROW);
      Row expectedRow = Row.withSchema(expectedSchema).addValue(ImmutableMap.of(1, "first")).build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectMapOfRowSelectAll() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("map{}.*").resolve(MAP_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(MAP_SCHEMA, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder()
              .addMapField("field1", FieldType.INT32, FieldType.STRING)
              .addMapField("field2", FieldType.INT32, FieldType.INT32)
              .addMapField("field3", FieldType.INT32, FieldType.DOUBLE)
              .addMapField("field_extra", FieldType.INT32, FieldType.STRING)
              .build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(MAP_SCHEMA, fieldAccessDescriptor, MAP_ROW);
      Row expectedRow =
          Row.withSchema(expectedSchema)
              .addValue(ImmutableMap.of(1, FLAT_ROW.getValue(0)))
              .addValue(ImmutableMap.of(1, FLAT_ROW.getValue(1)))
              .addValue(ImmutableMap.of(1, FLAT_ROW.getValue(2)))
              .addValue(ImmutableMap.of(1, FLAT_ROW.getValue(3)))
              .build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectMapOfArray() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("map.field1").resolve(MAP_ARRAY_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(MAP_ARRAY_SCHEMA, fieldAccessDescriptor);
  
      Schema expectedSchema =
          Schema.builder()
              .addMapField("field1", FieldType.INT32, FieldType.array(FieldType.STRING))
              .build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(MAP_ARRAY_SCHEMA, fieldAccessDescriptor, MAP_ARRAY_ROW);
  
      Row expectedRow =
          Row.withSchema(expectedSchema)
              .addValue(ImmutableMap.of(1, ImmutableList.of("first")))
              .build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectMapOfIterable() {
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("map.field1").resolve(MAP_ITERABLE_SCHEMA);
      Schema outputSchema = SelectHelpers.getOutputSchema(MAP_ITERABLE_SCHEMA, fieldAccessDescriptor);
  
      Schema expectedSchema =
          Schema.builder()
              .addMapField("field1", FieldType.INT32, FieldType.iterable(FieldType.STRING))
              .build();
      assertEquals(expectedSchema, outputSchema);
  
      Row row = selectRow(MAP_ITERABLE_SCHEMA, fieldAccessDescriptor, MAP_ITERABLE_ROW);
  
      Row expectedRow =
          Row.withSchema(expectedSchema)
              .addValue(ImmutableMap.of(1, ImmutableList.of("first")))
              .build();
      assertEquals(expectedRow, row);
    }
  
    @Test
    public void testSelectFieldOfRecord() {
      Schema f1 = Schema.builder().addInt64Field("f0").build();
      Schema f2 = Schema.builder().addRowField("f1", f1).build();
      Schema f3 = Schema.builder().addRowField("f2", f2).build();
  
      Row r1 = Row.withSchema(f1).addValue(42L).build(); // {"f0": 42}
      Row r2 = Row.withSchema(f2).addValue(r1).build(); // {"f1": {"f0": 42}}
      Row r3 = Row.withSchema(f3).addValue(r2).build(); // {"f2": {"f1": {"f0": 42}}}
  
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("f2.f1").resolve(f3);
  
      Schema outputSchema = SelectHelpers.getOutputSchema(f3, fieldAccessDescriptor);
  
      Row out = selectRow(f3, fieldAccessDescriptor, r3);
  
      assertEquals(f2, outputSchema);
      assertEquals(r2, out);
    }
  
    @Test
    public void testSelectFieldOfRecordOrRecord() {
      Schema f1 = Schema.builder().addInt64Field("f0").build();
      Schema f2 = Schema.builder().addRowField("f1", f1).build();
      Schema f3 = Schema.builder().addRowField("f2", f2).build();
      Schema f4 = Schema.builder().addRowField("f3", f3).build();
  
      Row r1 = Row.withSchema(f1).addValue(42L).build(); // {"f0": 42}
      Row r2 = Row.withSchema(f2).addValue(r1).build(); // {"f1": {"f0": 42}}
      Row r3 = Row.withSchema(f3).addValue(r2).build(); // {"f2": {"f1": {"f0": 42}}}
      Row r4 = Row.withSchema(f4).addValue(r3).build(); // {"f3": {"f2": {"f1": {"f0": 42}}}}
  
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("f3.f2").resolve(f4);
  
      Schema outputSchema = SelectHelpers.getOutputSchema(f4, fieldAccessDescriptor);
  
      Row out = selectRow(f4, fieldAccessDescriptor, r4);
  
      assertEquals(f3, outputSchema);
      assertEquals(r3, out);
    }
  
    @Test
    public void testArrayRowArray() {
      Schema f1 = Schema.builder().addStringField("f0").build();
      Schema f2 = Schema.builder().addArrayField("f1", FieldType.row(f1)).build();
      Schema f3 = Schema.builder().addRowField("f2", f2).build();
      Schema f4 = Schema.builder().addArrayField("f3", FieldType.row(f3)).build();
  
      Row r1 = Row.withSchema(f1).addValue("first").build();
      Row r2 = Row.withSchema(f2).addArray(r1, r1).build();
      Row r3 = Row.withSchema(f3).addValue(r2).build();
      Row r4 = Row.withSchema(f4).addArray(r3, r3).build();
  
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("f3.f2.f1.f0").resolve(f4);
      Schema outputSchema = SelectHelpers.getOutputSchema(f4, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addArrayField("f0", FieldType.array(FieldType.STRING)).build();
      assertEquals(expectedSchema, outputSchema);
      Row out = selectRow(f4, fieldAccessDescriptor, r4);
      Row expected =
          Row.withSchema(outputSchema)
              .addArray(Lists.newArrayList("first", "first"), Lists.newArrayList("first", "first"))
              .build();
      assertEquals(expected, out);
    }
  
    @Test
    public void testIterableRowIterable() {
      Schema f1 = Schema.builder().addStringField("f0").build();
      Schema f2 = Schema.builder().addIterableField("f1", FieldType.row(f1)).build();
      Schema f3 = Schema.builder().addRowField("f2", f2).build();
      Schema f4 = Schema.builder().addIterableField("f3", FieldType.row(f3)).build();
  
      Row r1 = Row.withSchema(f1).addValue("first").build();
      Row r2 = Row.withSchema(f2).addIterable(ImmutableList.of(r1, r1)).build();
      Row r3 = Row.withSchema(f3).addValue(r2).build();
      Row r4 = Row.withSchema(f4).addIterable(ImmutableList.of(r3, r3)).build();
  
      FieldAccessDescriptor fieldAccessDescriptor =
          FieldAccessDescriptor.withFieldNames("f3.f2.f1.f0").resolve(f4);
      Schema outputSchema = SelectHelpers.getOutputSchema(f4, fieldAccessDescriptor);
      Schema expectedSchema =
          Schema.builder().addIterableField("f0", FieldType.iterable(FieldType.STRING)).build();
      assertEquals(expectedSchema, outputSchema);
      Row out = selectRow(f4, fieldAccessDescriptor, r4);
      Row expected =
          Row.withSchema(outputSchema)
              .addIterable(
                  ImmutableList.of(
                      ImmutableList.of("first", "first"), ImmutableList.of("first", "first")))
              .build();
      assertEquals(expected, out);
    }
  
    static final Schema NESTED_NULLABLE_SCHEMA =
        Schema.builder()
            .addNullableField("nested", FieldType.row(FLAT_SCHEMA))
            .addNullableField("nestedArray", FieldType.array(FieldType.row(FLAT_SCHEMA)))
            .addNullableField(
                "nestedMap", FieldType.map(FieldType.STRING, FieldType.row(FLAT_SCHEMA)))
            .build();
    static final Row NESTED_NULLABLE_ROW = Row.nullRow(NESTED_NULLABLE_SCHEMA);
  
    @Test
    public void testNullableSchema() {
      FieldAccessDescriptor fieldAccessDescriptor1 =
          FieldAccessDescriptor.withFieldNames("nested.field1").resolve(NESTED_NULLABLE_SCHEMA);
      Schema schema1 = SelectHelpers.getOutputSchema(NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor1);
      Schema expectedSchema1 = Schema.builder().addNullableField("field1", FieldType.STRING).build();
      assertEquals(expectedSchema1, schema1);
  
      FieldAccessDescriptor fieldAccessDescriptor2 =
          FieldAccessDescriptor.withFieldNames("nested.*").resolve(NESTED_NULLABLE_SCHEMA);
      Schema schema2 = SelectHelpers.getOutputSchema(NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor2);
      Schema expectedSchema2 =
          Schema.builder()
              .addNullableField("field1", FieldType.STRING)
              .addNullableField("field2", FieldType.INT32)
              .addNullableField("field3", FieldType.DOUBLE)
              .addNullableField("field_extra", FieldType.STRING)
              .build();
      assertEquals(expectedSchema2, schema2);
    }
  
    @Test
    public void testNullableSchemaArray() {
      FieldAccessDescriptor fieldAccessDescriptor1 =
          FieldAccessDescriptor.withFieldNames("nestedArray.field1").resolve(NESTED_NULLABLE_SCHEMA);
      Schema schema1 = SelectHelpers.getOutputSchema(NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor1);
      Schema expectedSchema1 =
          Schema.builder().addNullableField("field1", FieldType.array(FieldType.STRING)).build();
      assertEquals(expectedSchema1, schema1);
  
      FieldAccessDescriptor fieldAccessDescriptor2 =
          FieldAccessDescriptor.withFieldNames("nestedArray.*").resolve(NESTED_NULLABLE_SCHEMA);
      Schema schema2 = SelectHelpers.getOutputSchema(NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor2);
      Schema expectedSchema2 =
          Schema.builder()
              .addNullableField("field1", FieldType.array(FieldType.STRING))
              .addNullableField("field2", FieldType.array(FieldType.INT32))
              .addNullableField("field3", FieldType.array(FieldType.DOUBLE))
              .addNullableField("field_extra", FieldType.array(FieldType.STRING))
              .build();
      assertEquals(expectedSchema2, schema2);
    }
  
    @Test
    public void testNullableSchemaMap() {
      FieldAccessDescriptor fieldAccessDescriptor1 =
          FieldAccessDescriptor.withFieldNames("nestedMap.field1").resolve(NESTED_NULLABLE_SCHEMA);
      Schema schema1 = SelectHelpers.getOutputSchema(NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor1);
      Schema expectedSchema1 =
          Schema.builder()
              .addNullableField("field1", FieldType.map(FieldType.STRING, FieldType.STRING))
              .build();
      assertEquals(expectedSchema1, schema1);
  
      FieldAccessDescriptor fieldAccessDescriptor2 =
          FieldAccessDescriptor.withFieldNames("nestedMap.*").resolve(NESTED_NULLABLE_SCHEMA);
      Schema schema2 = SelectHelpers.getOutputSchema(NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor2);
      Schema expectedSchema2 =
          Schema.builder()
              .addNullableField("field1", FieldType.map(FieldType.STRING, FieldType.STRING))
              .addNullableField("field2", FieldType.map(FieldType.STRING, FieldType.INT32))
              .addNullableField("field3", FieldType.map(FieldType.STRING, FieldType.DOUBLE))
              .addNullableField("field_extra", FieldType.map(FieldType.STRING, FieldType.STRING))
              .build();
      assertEquals(expectedSchema2, schema2);
    }
  
    @Test
    public void testSelectNullableNestedRow() {
      FieldAccessDescriptor fieldAccessDescriptor1 =
          FieldAccessDescriptor.withFieldNames("nested.field1").resolve(NESTED_NULLABLE_SCHEMA);
      Row out1 =
          selectRow(
              NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor1, Row.nullRow(NESTED_NULLABLE_SCHEMA));
      assertNull(out1.getValue(0));
  
      FieldAccessDescriptor fieldAccessDescriptor2 =
          FieldAccessDescriptor.withFieldNames("nested.*").resolve(NESTED_NULLABLE_SCHEMA);
      Row out2 =
          selectRow(
              NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor2, Row.nullRow(NESTED_NULLABLE_SCHEMA));
      assertEquals(Collections.nCopies(4, null), out2.getValues());
    }
  
    @Test
    public void testSelectNullableNestedRowArray() {
      FieldAccessDescriptor fieldAccessDescriptor1 =
          FieldAccessDescriptor.withFieldNames("nestedArray.field1").resolve(NESTED_NULLABLE_SCHEMA);
      Row out1 =
          selectRow(
              NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor1, Row.nullRow(NESTED_NULLABLE_SCHEMA));
      assertNull(out1.getValue(0));
  
      FieldAccessDescriptor fieldAccessDescriptor2 =
          FieldAccessDescriptor.withFieldNames("nestedArray.*").resolve(NESTED_NULLABLE_SCHEMA);
      Row out2 =
          selectRow(
              NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor2, Row.nullRow(NESTED_NULLABLE_SCHEMA));
      assertEquals(Collections.nCopies(4, null), out2.getValues());
    }
  
    @Test
    public void testSelectNullableNestedRowMap() {
      FieldAccessDescriptor fieldAccessDescriptor1 =
          FieldAccessDescriptor.withFieldNames("nestedMap.field1").resolve(NESTED_NULLABLE_SCHEMA);
      Row out1 =
          selectRow(
              NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor1, Row.nullRow(NESTED_NULLABLE_SCHEMA));
      assertNull(out1.getValue(0));
  
      FieldAccessDescriptor fieldAccessDescriptor2 =
          FieldAccessDescriptor.withFieldNames("nestedMap.*").resolve(NESTED_NULLABLE_SCHEMA);
      Row out2 =
          selectRow(
              NESTED_NULLABLE_SCHEMA, fieldAccessDescriptor2, Row.nullRow(NESTED_NULLABLE_SCHEMA));
      assertEquals(Collections.nCopies(4, null), out2.getValues());
    }
  }
  