/*
 * 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.extensions.protobuf;

import static org.apache.beam.sdk.extensions.protobuf.ProtoSchemaTranslator.SCHEMA_OPTION_META_NUMBER;
import static org.apache.beam.sdk.extensions.protobuf.ProtoSchemaTranslator.SCHEMA_OPTION_META_TYPE_NAME;
import static org.apache.beam.sdk.extensions.protobuf.ProtoSchemaTranslator.getFieldNumber;

import com.google.protobuf.BoolValue;
import com.google.protobuf.ByteString;
import com.google.protobuf.BytesValue;
import com.google.protobuf.DoubleValue;
import com.google.protobuf.Duration;
import com.google.protobuf.FloatValue;
import com.google.protobuf.Int32Value;
import com.google.protobuf.Int64Value;
import com.google.protobuf.StringValue;
import com.google.protobuf.Timestamp;
import com.google.protobuf.UInt32Value;
import com.google.protobuf.UInt64Value;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.beam.sdk.extensions.protobuf.Proto2SchemaMessages.OptionalNested;
import org.apache.beam.sdk.extensions.protobuf.Proto2SchemaMessages.OptionalPrimitive;
import org.apache.beam.sdk.extensions.protobuf.Proto2SchemaMessages.RequiredNested;
import org.apache.beam.sdk.extensions.protobuf.Proto2SchemaMessages.RequiredPrimitive;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.MapPrimitive;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.Nested;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.OneOf;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.OuterOneOf;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.Primitive;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.RepeatPrimitive;
import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.WktMessage;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.Fixed32;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.Fixed64;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.SFixed32;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.SFixed64;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.SInt32;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.SInt64;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.UInt32;
import org.apache.beam.sdk.extensions.protobuf.ProtoSchemaLogicalTypes.UInt64;
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.logicaltypes.NanosDuration;
import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
import org.apache.beam.sdk.schemas.logicaltypes.OneOfType;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;

class TestProtoSchemas {

  static Field withFieldNumber(String name, FieldType fieldType, int fieldNumber) {
    return Field.of(name, fieldType)
        .withOptions(
            Schema.Options.builder()
                .setOption(SCHEMA_OPTION_META_NUMBER, FieldType.INT32, fieldNumber));
  }

  static Schema.Options withTypeName(String typeName) {
    return Schema.Options.builder()
        .setOption(SCHEMA_OPTION_META_TYPE_NAME, FieldType.STRING, typeName)
        .build();
  }

  // The schema we expect from the Primitive proto.
  static final Schema PRIMITIVE_SCHEMA =
      Schema.builder()
          .addField(withFieldNumber("primitive_double", FieldType.DOUBLE, 1))
          .addField(withFieldNumber("primitive_float", FieldType.FLOAT, 2))
          .addField(withFieldNumber("primitive_int32", FieldType.INT32, 3))
          .addField(withFieldNumber("primitive_int64", FieldType.INT64, 4))
          .addField(withFieldNumber("primitive_uint32", FieldType.logicalType(new UInt32()), 5))
          .addField(withFieldNumber("primitive_uint64", FieldType.logicalType(new UInt64()), 6))
          .addField(withFieldNumber("primitive_sint32", FieldType.logicalType(new SInt32()), 7))
          .addField(withFieldNumber("primitive_sint64", FieldType.logicalType(new SInt64()), 8))
          .addField(withFieldNumber("primitive_fixed32", FieldType.logicalType(new Fixed32()), 9))
          .addField(withFieldNumber("primitive_fixed64", FieldType.logicalType(new Fixed64()), 10))
          .addField(
              withFieldNumber("primitive_sfixed32", FieldType.logicalType(new SFixed32()), 11))
          .addField(
              withFieldNumber("primitive_sfixed64", FieldType.logicalType(new SFixed64()), 12))
          .addField(withFieldNumber("primitive_bool", FieldType.BOOLEAN, 13))
          .addField(withFieldNumber("primitive_string", FieldType.STRING, 14))
          .addField(withFieldNumber("primitive_bytes", FieldType.BYTES, 15))
          .setOptions(
              Schema.Options.builder()
                  .setOption(
                      SCHEMA_OPTION_META_TYPE_NAME,
                      FieldType.STRING,
                      "proto3_schema_messages.Primitive"))
          .build();

  static final Schema OPTIONAL_PRIMITIVE_SCHEMA =
      Schema.builder()
          .addField(withFieldNumber("primitive_int32", FieldType.INT32, 1))
          .addField(withFieldNumber("primitive_bool", FieldType.BOOLEAN, 2))
          .addField(withFieldNumber("primitive_string", FieldType.STRING, 3))
          .addField(withFieldNumber("primitive_bytes", FieldType.BYTES, 4))
          .setOptions(
              Schema.Options.builder()
                  .setOption(
                      SCHEMA_OPTION_META_TYPE_NAME,
                      FieldType.STRING,
                      "proto2_schema_messages.OptionalPrimitive"))
          .build();

  static final Schema REQUIRED_PRIMITIVE_SCHEMA =
      Schema.builder()
          .addField(withFieldNumber("primitive_int32", FieldType.INT32, 1))
          .addField(withFieldNumber("primitive_bool", FieldType.BOOLEAN, 2))
          .addField(withFieldNumber("primitive_string", FieldType.STRING, 3))
          .addField(withFieldNumber("primitive_bytes", FieldType.BYTES, 4))
          .setOptions(
              Schema.Options.builder()
                  .setOption(
                      SCHEMA_OPTION_META_TYPE_NAME,
                      FieldType.STRING,
                      "proto2_schema_messages.RequiredPrimitive"))
          .build();

  // A sample instance of the  row.
  private static final byte[] BYTE_ARRAY = new byte[] {1, 2, 3, 4};
  static final Row PRIMITIVE_ROW =
      Row.withSchema(PRIMITIVE_SCHEMA)
          .addValues(
              1.1, 2.2F, 32, 64L, 33, 65L, 123, 124L, 30, 62L, 31, 63L, true, "horsey", BYTE_ARRAY)
          .build();

  // A sample instance of the proto.
  static final Primitive PRIMITIVE_PROTO =
      Primitive.newBuilder()
          .setPrimitiveDouble(1.1)
          .setPrimitiveFloat(2.2F)
          .setPrimitiveInt32(32)
          .setPrimitiveInt64(64)
          .setPrimitiveUint32(33)
          .setPrimitiveUint64(65)
          .setPrimitiveSint32(123)
          .setPrimitiveSint64(124)
          .setPrimitiveFixed32(30)
          .setPrimitiveFixed64(62)
          .setPrimitiveSfixed32(31)
          .setPrimitiveSfixed64(63)
          .setPrimitiveBool(true)
          .setPrimitiveString("horsey")
          .setPrimitiveBytes(ByteString.copyFrom(BYTE_ARRAY))
          .build();

  // A sample instance of the  row.
  static final Row OPTIONAL_PRIMITIVE_ROW =
      Row.withSchema(OPTIONAL_PRIMITIVE_SCHEMA).addValues(32, true, "horsey", BYTE_ARRAY).build();

  // A sample instance of the proto.
  static final OptionalPrimitive OPTIONAL_PRIMITIVE_PROTO =
      OptionalPrimitive.newBuilder()
          .setPrimitiveInt32(32)
          .setPrimitiveBool(true)
          .setPrimitiveString("horsey")
          .setPrimitiveBytes(ByteString.copyFrom(BYTE_ARRAY))
          .build();

  // A sample instance of the  row.
  static final Row REQUIRED_PRIMITIVE_ROW =
      Row.withSchema(REQUIRED_PRIMITIVE_SCHEMA).addValues(32, true, "horsey", BYTE_ARRAY).build();

  // A sample instance of the proto.
  static final RequiredPrimitive REQUIRED_PRIMITIVE_PROTO =
      RequiredPrimitive.newBuilder()
          .setPrimitiveInt32(32)
          .setPrimitiveBool(true)
          .setPrimitiveString("horsey")
          .setPrimitiveBytes(ByteString.copyFrom(BYTE_ARRAY))
          .build();

  // The schema for the RepeatedPrimitive proto.
  static final Schema REPEATED_SCHEMA =
      Schema.builder()
          .addField(withFieldNumber("repeated_double", FieldType.array(FieldType.DOUBLE), 1))
          .addField(withFieldNumber("repeated_float", FieldType.array(FieldType.FLOAT), 2))
          .addField(withFieldNumber("repeated_int32", FieldType.array(FieldType.INT32), 3))
          .addField(withFieldNumber("repeated_int64", FieldType.array(FieldType.INT64), 4))
          .addField(
              withFieldNumber(
                  "repeated_uint32", FieldType.array(FieldType.logicalType(new UInt32())), 5))
          .addField(
              withFieldNumber(
                  "repeated_uint64", FieldType.array(FieldType.logicalType(new UInt64())), 6))
          .addField(
              withFieldNumber(
                  "repeated_sint32", FieldType.array(FieldType.logicalType(new SInt32())), 7))
          .addField(
              withFieldNumber(
                  "repeated_sint64", FieldType.array(FieldType.logicalType(new SInt64())), 8))
          .addField(
              withFieldNumber(
                  "repeated_fixed32", FieldType.array(FieldType.logicalType(new Fixed32())), 9))
          .addField(
              withFieldNumber(
                  "repeated_fixed64", FieldType.array(FieldType.logicalType(new Fixed64())), 10))
          .addField(
              withFieldNumber(
                  "repeated_sfixed32", FieldType.array(FieldType.logicalType(new SFixed32())), 11))
          .addField(
              withFieldNumber(
                  "repeated_sfixed64", FieldType.array(FieldType.logicalType(new SFixed64())), 12))
          .addField(withFieldNumber("repeated_bool", FieldType.array(FieldType.BOOLEAN), 13))
          .addField(withFieldNumber("repeated_string", FieldType.array(FieldType.STRING), 14))
          .addField(withFieldNumber("repeated_bytes", FieldType.array(FieldType.BYTES), 15))
          .setOptions(withTypeName("proto3_schema_messages.RepeatPrimitive"))
          .build();

  // A sample instance of the row.
  static final Row REPEATED_ROW =
      Row.withSchema(REPEATED_SCHEMA)
          .addArray(1.1, 1.1)
          .addArray(2.2F, 2.2F)
          .addArray(32, 32)
          .addArray(64L, 64L)
          .addArray(33, 33)
          .addArray(65L, 65L)
          .addArray(123, 123)
          .addArray(124L, 124L)
          .addArray(30, 30)
          .addArray(62L, 62L)
          .addArray(31, 31)
          .addArray(63L, 63L)
          .addArray(true, true)
          .addArray("horsey", "horsey")
          .addArray(BYTE_ARRAY, BYTE_ARRAY)
          .build();

  // A sample instance of the proto.
  static final RepeatPrimitive REPEATED_PROTO =
      RepeatPrimitive.newBuilder()
          .addAllRepeatedDouble(ImmutableList.of(1.1, 1.1))
          .addAllRepeatedFloat(ImmutableList.of(2.2F, 2.2F))
          .addAllRepeatedInt32(ImmutableList.of(32, 32))
          .addAllRepeatedInt64(ImmutableList.of(64L, 64L))
          .addAllRepeatedUint32(ImmutableList.of(33, 33))
          .addAllRepeatedUint64(ImmutableList.of(65L, 65L))
          .addAllRepeatedSint32(ImmutableList.of(123, 123))
          .addAllRepeatedSint64(ImmutableList.of(124L, 124L))
          .addAllRepeatedFixed32(ImmutableList.of(30, 30))
          .addAllRepeatedFixed64(ImmutableList.of(62L, 62L))
          .addAllRepeatedSfixed32(ImmutableList.of(31, 31))
          .addAllRepeatedSfixed64(ImmutableList.of(63L, 63L))
          .addAllRepeatedBool(ImmutableList.of(true, true))
          .addAllRepeatedString(ImmutableList.of("horsey", "horsey"))
          .addAllRepeatedBytes(
              ImmutableList.of(ByteString.copyFrom(BYTE_ARRAY), ByteString.copyFrom(BYTE_ARRAY)))
          .build();

  static final Row NULL_REPEATED_ROW =
      Row.withSchema(REPEATED_SCHEMA)
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .addArray()
          .build();

  // A sample instance of the proto.
  static final RepeatPrimitive NULL_REPEATED_PROTO = RepeatPrimitive.newBuilder().build();

  // The schema for the MapPrimitive proto.
  static final Schema MAP_PRIMITIVE_SCHEMA =
      Schema.builder()
          .addField(
              withFieldNumber(
                  "string_string_map", FieldType.map(FieldType.STRING, FieldType.STRING), 1))
          .addField(
              withFieldNumber(
                  "string_int_map", FieldType.map(FieldType.STRING, FieldType.INT32), 2))
          .addField(
              withFieldNumber(
                  "int_string_map", FieldType.map(FieldType.INT32, FieldType.STRING), 3))
          .addField(
              withFieldNumber(
                  "string_bytes_map", FieldType.map(FieldType.STRING, FieldType.BYTES), 4))
          .setOptions(withTypeName("proto3_schema_messages.MapPrimitive"))
          .build();

  // A sample instance of the row.
  static final Row MAP_PRIMITIVE_ROW =
      Row.withSchema(MAP_PRIMITIVE_SCHEMA)
          .addValue(ImmutableMap.of("k1", "v1", "k2", "v2"))
          .addValue(ImmutableMap.of("k1", 1, "k2", 2))
          .addValue(ImmutableMap.of(1, "v1", 2, "v2"))
          .addValue(ImmutableMap.of("k1", BYTE_ARRAY, "k2", BYTE_ARRAY))
          .build();

  // A sample instance of the proto.
  static final MapPrimitive MAP_PRIMITIVE_PROTO =
      MapPrimitive.newBuilder()
          .putAllStringStringMap(ImmutableMap.of("k1", "v1", "k2", "v2"))
          .putAllStringIntMap(ImmutableMap.of("k1", 1, "k2", 2))
          .putAllIntStringMap(ImmutableMap.of(1, "v1", 2, "v2"))
          .putAllStringBytesMap(
              ImmutableMap.of(
                  "k1", ByteString.copyFrom(BYTE_ARRAY), "k2", ByteString.copyFrom(BYTE_ARRAY)))
          .build();

  // A sample instance of the row.
  static final Row NULL_MAP_PRIMITIVE_ROW =
      Row.withSchema(MAP_PRIMITIVE_SCHEMA)
          .addValue(ImmutableMap.of())
          .addValue(ImmutableMap.of())
          .addValue(ImmutableMap.of())
          .addValue(ImmutableMap.of())
          .build();

  // A sample instance of the proto.
  static final MapPrimitive NULL_MAP_PRIMITIVE_PROTO = MapPrimitive.newBuilder().build();

  // The schema for the Nested proto.
  static final Schema NESTED_SCHEMA =
      Schema.builder()
          .addField(
              withFieldNumber("nested", FieldType.row(PRIMITIVE_SCHEMA).withNullable(true), 1))
          .addField(
              withFieldNumber("nested_list", FieldType.array(FieldType.row(PRIMITIVE_SCHEMA)), 2))
          .addField(
              withFieldNumber(
                  "nested_map",
                  FieldType.map(FieldType.STRING, FieldType.row(PRIMITIVE_SCHEMA)),
                  3))
          .setOptions(withTypeName("proto3_schema_messages.Nested"))
          .build();

  // A sample instance of the row.
  static final Row NESTED_ROW =
      Row.withSchema(NESTED_SCHEMA)
          .addValue(PRIMITIVE_ROW)
          .addArray(ImmutableList.of(PRIMITIVE_ROW, PRIMITIVE_ROW))
          .addValue(ImmutableMap.of("k1", PRIMITIVE_ROW, "k2", PRIMITIVE_ROW))
          .build();

  // A sample instance of the proto.
  static final Nested NESTED_PROTO =
      Nested.newBuilder()
          .setNested(PRIMITIVE_PROTO)
          .addAllNestedList(ImmutableList.of(PRIMITIVE_PROTO, PRIMITIVE_PROTO))
          .putAllNestedMap(ImmutableMap.of("k1", PRIMITIVE_PROTO, "k2", PRIMITIVE_PROTO))
          .build();

  // The schema for the OneOf proto.
  private static final List<Field> ONEOF_FIELDS =
      ImmutableList.of(
          withFieldNumber("oneof_int32", FieldType.INT32, 2),
          withFieldNumber("oneof_bool", FieldType.BOOLEAN, 3),
          withFieldNumber("oneof_string", FieldType.STRING, 4),
          withFieldNumber("oneof_primitive", FieldType.row(PRIMITIVE_SCHEMA), 5));
  private static final Map<String, Integer> ONE_OF_ENUM_MAP =
      ONEOF_FIELDS.stream().collect(Collectors.toMap(Field::getName, f -> getFieldNumber(f)));
  static final OneOfType ONE_OF_TYPE = OneOfType.create(ONEOF_FIELDS, ONE_OF_ENUM_MAP);
  static final Schema ONEOF_SCHEMA =
      Schema.builder()
          .addField("special_oneof", FieldType.logicalType(ONE_OF_TYPE))
          .addField(withFieldNumber("place1", FieldType.STRING, 1))
          .addField(withFieldNumber("place2", FieldType.INT32, 6))
          .setOptions(withTypeName("proto3_schema_messages.OneOf"))
          .build();

  // Sample row instances for each OneOf case.
  static final Row ONEOF_ROW_INT32 =
      Row.withSchema(ONEOF_SCHEMA)
          .addValues(ONE_OF_TYPE.createValue("oneof_int32", 1), "foo", 0)
          .build();
  static final Row ONEOF_ROW_BOOL =
      Row.withSchema(ONEOF_SCHEMA)
          .addValues(ONE_OF_TYPE.createValue("oneof_bool", true), "foo", 0)
          .build();
  static final Row ONEOF_ROW_STRING =
      Row.withSchema(ONEOF_SCHEMA)
          .addValues(ONE_OF_TYPE.createValue("oneof_string", "foo"), "foo", 0)
          .build();
  static final Row ONEOF_ROW_PRIMITIVE =
      Row.withSchema(ONEOF_SCHEMA)
          .addValues(ONE_OF_TYPE.createValue("oneof_primitive", PRIMITIVE_ROW), "foo", 0)
          .build();

  // Sample proto instances for each oneof case.
  static final OneOf ONEOF_PROTO_INT32 =
      OneOf.newBuilder().setOneofInt32(1).setPlace1("foo").setPlace2(0).build();
  static final OneOf ONEOF_PROTO_BOOL =
      OneOf.newBuilder().setOneofBool(true).setPlace1("foo").setPlace2(0).build();
  static final OneOf ONEOF_PROTO_STRING =
      OneOf.newBuilder().setOneofString("foo").setPlace1("foo").setPlace2(0).build();
  static final OneOf ONEOF_PROTO_PRIMITIVE =
      OneOf.newBuilder().setOneofPrimitive(PRIMITIVE_PROTO).setPlace1("foo").setPlace2(0).build();

  // The schema for the OuterOneOf proto.
  private static final List<Field> OUTER_ONEOF_FIELDS =
      ImmutableList.of(
          withFieldNumber("oneof_oneof", FieldType.row(ONEOF_SCHEMA), 1),
          withFieldNumber("oneof_int32", FieldType.INT32, 2));
  private static final Map<String, Integer> OUTER_ONE_OF_ENUM_MAP =
      OUTER_ONEOF_FIELDS.stream().collect(Collectors.toMap(Field::getName, f -> getFieldNumber(f)));
  static final OneOfType OUTER_ONEOF_TYPE =
      OneOfType.create(OUTER_ONEOF_FIELDS, OUTER_ONE_OF_ENUM_MAP);
  static final Schema OUTER_ONEOF_SCHEMA =
      Schema.builder()
          .addField("outer_oneof", FieldType.logicalType(OUTER_ONEOF_TYPE))
          .setOptions(withTypeName("proto3_schema_messages.OuterOneOf"))
          .build();

  // A sample instance of the Row.
  static final Row OUTER_ONEOF_ROW =
      Row.withSchema(OUTER_ONEOF_SCHEMA)
          .addValues(OUTER_ONEOF_TYPE.createValue("oneof_oneof", ONEOF_ROW_PRIMITIVE))
          .build();

  // A sample instance of the proto.
  static final OuterOneOf OUTER_ONEOF_PROTO =
      OuterOneOf.newBuilder().setOneofOneof(ONEOF_PROTO_PRIMITIVE).build();

  static final Schema WKT_MESSAGE_SCHEMA =
      Schema.builder()
          .addField(withFieldNumber("double", FieldType.DOUBLE, 1).withNullable(true))
          .addField(withFieldNumber("float", FieldType.FLOAT, 2).withNullable(true))
          .addField(withFieldNumber("int32", FieldType.INT32, 3).withNullable(true))
          .addField(withFieldNumber("int64", FieldType.INT64, 4).withNullable(true))
          .addField(
              withFieldNumber("uint32", FieldType.logicalType(new UInt32()), 5).withNullable(true))
          .addField(
              withFieldNumber("uint64", FieldType.logicalType(new UInt64()), 6).withNullable(true))
          .addField(withFieldNumber("bool", FieldType.BOOLEAN, 13).withNullable(true))
          .addField(withFieldNumber("string", FieldType.STRING, 14).withNullable(true))
          .addField(withFieldNumber("bytes", FieldType.BYTES, 15).withNullable(true))
          .addField(
              withFieldNumber("timestamp", FieldType.logicalType(new NanosInstant()), 16)
                  .withNullable(true))
          .addField(
              withFieldNumber("duration", FieldType.logicalType(new NanosDuration()), 17)
                  .withNullable(true))
          .setOptions(withTypeName("proto3_schema_messages.WktMessage"))
          .build();

  static final Schema WKT_MESSAGE_SHUFFLED_SCHEMA =
      Schema.builder()
          .addField(withFieldNumber("int64", FieldType.INT64, 4).withNullable(true))
          .addField(withFieldNumber("int32", FieldType.INT32, 3).withNullable(true))
          .addField(withFieldNumber("double", FieldType.DOUBLE, 1).withNullable(true))
          .addField(
              withFieldNumber("uint32", FieldType.logicalType(new UInt32()), 5).withNullable(true))
          .addField(withFieldNumber("float", FieldType.FLOAT, 2).withNullable(true))
          .addField(
              withFieldNumber("uint64", FieldType.logicalType(new UInt64()), 6).withNullable(true))
          .addField(withFieldNumber("bytes", FieldType.BYTES, 15).withNullable(true))
          .addField(withFieldNumber("string", FieldType.STRING, 14).withNullable(true))
          .addField(withFieldNumber("bool", FieldType.BOOLEAN, 13).withNullable(true))
          .addField(
              withFieldNumber("timestamp", FieldType.logicalType(new NanosInstant()), 16)
                  .withNullable(true))
          .addField(
              withFieldNumber("duration", FieldType.logicalType(new NanosDuration()), 17)
                  .withNullable(true))
          .setOptions(withTypeName("proto3_schema_messages.WktMessage"))
          .build();

  // A sample instance of the  row.
  static final Instant JAVA_NOW = Instant.now();
  static final Timestamp PROTO_NOW =
      Timestamp.newBuilder()
          .setSeconds(JAVA_NOW.getEpochSecond())
          .setNanos(JAVA_NOW.getNano())
          .build();
  static final java.time.Duration JAVA_DURATION =
      java.time.Duration.ofSeconds(JAVA_NOW.getEpochSecond(), JAVA_NOW.getNano());
  static final Duration PROTO_DURATION =
      Duration.newBuilder()
          .setSeconds(JAVA_NOW.getEpochSecond())
          .setNanos(JAVA_NOW.getNano())
          .build();
  static final Row WKT_MESSAGE_ROW =
      Row.withSchema(WKT_MESSAGE_SCHEMA)
          .addValues(
              1.1, 2.2F, 32, 64L, 33, 65L, true, "horsey", BYTE_ARRAY, JAVA_NOW, JAVA_DURATION)
          .build();
  static final Row WKT_MESSAGE_SHUFFLED_ROW =
      Row.withSchema(WKT_MESSAGE_SHUFFLED_SCHEMA)
          .addValues(
              64L, 32, 1.1, 33, 2.2F, 65L, BYTE_ARRAY, "horsey", true, JAVA_NOW, JAVA_DURATION)
          .build();

  // A sample instance of the proto.
  static final WktMessage WKT_MESSAGE_PROTO =
      WktMessage.newBuilder()
          .setDouble(DoubleValue.of(1.1))
          .setFloat(FloatValue.of(2.2F))
          .setInt32(Int32Value.of(32))
          .setInt64(Int64Value.of(64))
          .setUint32(UInt32Value.of(33))
          .setUint64(UInt64Value.of(65))
          .setBool(BoolValue.of(true))
          .setString(StringValue.of("horsey"))
          .setBytes(BytesValue.of(ByteString.copyFrom(BYTE_ARRAY)))
          .setTimestamp(PROTO_NOW)
          .setDuration(PROTO_DURATION)
          .build();

  // The schema for the OptionalNested proto.
  static final Schema OPTIONAL_NESTED_SCHEMA =
      Schema.builder()
          .addField(
              withFieldNumber("nested", FieldType.row(OPTIONAL_PRIMITIVE_SCHEMA), 1)
                  .withNullable(true))
          .setOptions(withTypeName("proto2_schema_messages.OptionalNested"))
          .build();

  // A sample instance of the proto.
  static final OptionalNested OPTIONAL_NESTED =
      OptionalNested.newBuilder().setNested(OPTIONAL_PRIMITIVE_PROTO).build();

  // The schema for the Required Nested proto.
  static final Schema REQUIRED_NESTED_SCHEMA =
      Schema.builder()
          .addField(
              withFieldNumber("nested", FieldType.row(REQUIRED_PRIMITIVE_SCHEMA), 1)
                  .withNullable(false))
          .setOptions(withTypeName("proto2_schema_messages.RequiredNested"))
          .build();

  // A sample instance of the proto.
  static final RequiredNested REQUIRED_NESTED =
      RequiredNested.newBuilder().setNested(REQUIRED_PRIMITIVE_PROTO).build();
}
