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

import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.schemas.Factory;
import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
import org.apache.beam.sdk.schemas.FieldValueGetter;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.Schema.TypeName;
import org.apache.beam.sdk.schemas.SchemaUtils;
import org.apache.beam.sdk.values.RowUtils.CapturingRowCases;
import org.apache.beam.sdk.values.RowUtils.FieldOverride;
import org.apache.beam.sdk.values.RowUtils.FieldOverrides;
import org.apache.beam.sdk.values.RowUtils.RowFieldMatcher;
import org.apache.beam.sdk.values.RowUtils.RowPosition;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.DateTime;
import org.joda.time.ReadableDateTime;
import org.joda.time.ReadableInstant;

/**
 * {@link Row} is an immutable tuple-like schema to represent one element in a {@link PCollection}.
 * The fields are described with a {@link Schema}.
 *
 * <p>{@link Schema} contains the names and types for each field.
 *
 * <p>There are several ways to build a new Row object. To build a row from scratch using a schema
 * object, {@link Row#withSchema} can be used. Schema fields can be specified by name, and nested
 * fields can be specified using the field selection syntax. For example:
 *
 * <pre>{@code
 * Row row = Row.withSchema(schema)
 *              .withFieldValue("userId", "user1)
 *              .withFieldValue("location.city", "seattle")
 *              .withFieldValue("location.state", "wa")
 *              .build();
 * }</pre>
 *
 * <p>The {@link Row#fromRow} builder can be used to base a row off of another row. The builder can
 * be used to specify values for specific fields, and all the remaining values will be taken from
 * the original row. For example, the following produces a row identical to the above row except for
 * the location.city field.
 *
 * <pre>{@code
 * Row modifiedRow =
 *     Row.fromRow(row)
 *        .withFieldValue("location.city", "tacoma")
 *        .build();
 * }</pre>
 */
@SuppressWarnings({
  "nullness", // TODO(https://github.com/apache/beam/issues/20497)
  "rawtypes"
})
public abstract class Row implements Serializable {
  private final Schema schema;

  Row(Schema schema) {
    this.schema = schema;
  }

  // Abstract methods to be implemented by subclasses that handle object access.

  /** Get value by field index, {@link ClassCastException} is thrown if schema doesn't match. */
  @SuppressWarnings("TypeParameterUnusedInFormals")
  public abstract <T extends @Nullable Object> T getValue(int fieldIdx);

  /** Return the size of data fields. */
  public abstract int getFieldCount();

  /** Return the list of raw unmodified data values to enable 0-copy code. */
  public abstract List<@Nullable Object> getValues();

  /** Return a list of data values. Any LogicalType values are returned as base values. * */
  public List<Object> getBaseValues() {
    return IntStream.range(0, getFieldCount())
        .mapToObj(i -> getBaseValue(i))
        .collect(Collectors.toList());
  }

  /** Get value by field name, {@link ClassCastException} is thrown if type doesn't match. */
  @SuppressWarnings("TypeParameterUnusedInFormals")
  public <T extends @Nullable Object> T getValue(String fieldName) {
    return getValue(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#BYTE} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Byte getByte(String fieldName) {
    return getByte(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#BYTES} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public byte @Nullable [] getBytes(String fieldName) {
    return getBytes(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#INT16} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Short getInt16(String fieldName) {
    return getInt16(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#INT32} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Integer getInt32(String fieldName) {
    return getInt32(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#INT64} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Long getInt64(String fieldName) {
    return getInt64(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#DECIMAL} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable BigDecimal getDecimal(String fieldName) {
    return getDecimal(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#FLOAT} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Float getFloat(String fieldName) {
    return getFloat(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#DOUBLE} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Double getDouble(String fieldName) {
    return getDouble(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#STRING} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable String getString(String fieldName) {
    return getString(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#DATETIME} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable ReadableDateTime getDateTime(String fieldName) {
    return getDateTime(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#BOOLEAN} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Boolean getBoolean(String fieldName) {
    return getBoolean(getSchema().indexOf(fieldName));
  }

  /**
   * Get an array value by field name, {@link IllegalStateException} is thrown if schema doesn't
   * match.
   */
  public <T> @Nullable Collection<T> getArray(String fieldName) {
    return getArray(getSchema().indexOf(fieldName));
  }

  /**
   * Get an iterable value by field name, {@link IllegalStateException} is thrown if schema doesn't
   * match.
   */
  public <T> @Nullable Iterable<T> getIterable(String fieldName) {
    return getIterable(getSchema().indexOf(fieldName));
  }

  /**
   * Get a MAP value by field name, {@link IllegalStateException} is thrown if schema doesn't match.
   */
  public <T1, T2> @Nullable Map<T1, T2> getMap(String fieldName) {
    return getMap(getSchema().indexOf(fieldName));
  }

  /**
   * Returns the Logical Type input type for this field. {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public <T extends @Nullable Object> T getLogicalTypeValue(String fieldName, Class<T> clazz) {
    return getLogicalTypeValue(getSchema().indexOf(fieldName), clazz);
  }

  /**
   * Returns the base type for this field. If this is a logical type, we convert to the base value.
   * Otherwise the field itself is returned.
   */
  public <T extends @Nullable Object> T getBaseValue(String fieldName, Class<T> clazz) {
    return getBaseValue(getSchema().indexOf(fieldName), clazz);
  }

  /**
   * Returns the base type for this field. If this is a logical type, we convert to the base value.
   * Otherwise the field itself is returned.
   */
  public @Nullable Object getBaseValue(String fieldName) {
    return getBaseValue(fieldName, Object.class);
  }

  /**
   * Get a {@link TypeName#ROW} value by field name, {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Row getRow(String fieldName) {
    return getRow(getSchema().indexOf(fieldName));
  }

  /**
   * Get a {@link TypeName#BYTE} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Byte getByte(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#BYTES} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public byte @Nullable [] getBytes(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#INT16} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Short getInt16(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#INT32} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Integer getInt32(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#FLOAT} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Float getFloat(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#DOUBLE} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Double getDouble(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#INT64} value by field index, {@link ClassCastException} is thrown if
   * schema doesn't match.
   */
  public @Nullable Long getInt64(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link String} value by field index, {@link ClassCastException} is thrown if schema
   * doesn't match.
   */
  public @Nullable String getString(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link TypeName#DATETIME} value by field index, {@link IllegalStateException} is thrown
   * if schema doesn't match.
   */
  public @Nullable ReadableDateTime getDateTime(int idx) {
    ReadableInstant instant = getValue(idx);
    return instant == null ? null : new DateTime(instant).withZone(instant.getZone());
  }

  /**
   * Get a {@link BigDecimal} value by field index, {@link ClassCastException} is thrown if schema
   * doesn't match.
   */
  public @Nullable BigDecimal getDecimal(int idx) {
    return getValue(idx);
  }

  /**
   * Get a {@link Boolean} value by field index, {@link ClassCastException} is thrown if schema
   * doesn't match.
   */
  public @Nullable Boolean getBoolean(int idx) {
    return getValue(idx);
  }

  /**
   * Get an array value by field index, {@link IllegalStateException} is thrown if schema doesn't
   * match.
   */
  public <T> @Nullable Collection<T> getArray(int idx) {
    return getValue(idx);
  }

  /**
   * Get an iterable value by field index, {@link IllegalStateException} is thrown if schema doesn't
   * match.
   */
  public <T> @Nullable Iterable<T> getIterable(int idx) {
    return getValue(idx);
  }

  /**
   * Get a MAP value by field index, {@link IllegalStateException} is thrown if schema doesn't
   * match.
   */
  public <T1, T2> @Nullable Map<T1, T2> getMap(int idx) {
    return getValue(idx);
  }

  /**
   * Returns the Logical Type input type for this field. {@link IllegalStateException} is thrown if
   * schema doesn't match.
   */
  public <T extends @Nullable Object> T getLogicalTypeValue(int idx, Class<T> clazz) {
    return (T) getValue(idx);
  }

  /**
   * Returns the base type for this field. If this is a logical type, we convert to the base value.
   * Otherwise the field itself is returned.
   */
  public <T extends @Nullable Object> T getBaseValue(int idx, Class<T> clazz) {
    Object value = getValue(idx);
    FieldType fieldType = getSchema().getField(idx).getType();
    if (fieldType.getTypeName().isLogicalType() && value != null) {
      while (fieldType.getTypeName().isLogicalType()) {
        Schema.LogicalType<Object, T> logicalType =
            (Schema.LogicalType<Object, T>) fieldType.getLogicalType();
        value = logicalType.toBaseType(value);
        fieldType = fieldType.getLogicalType().getBaseType();
      }
    }
    return (T) value;
  }

  /**
   * Returns the base type for this field. If this is a logical type, we convert to the base value.
   * Otherwise the field itself is returned.
   */
  public @Nullable Object getBaseValue(int idx) {
    return getBaseValue(idx, Object.class);
  }

  /**
   * Get a {@link Row} value by field index, {@link IllegalStateException} is thrown if schema
   * doesn't match.
   */
  public @Nullable Row getRow(int idx) {
    return getValue(idx);
  }

  /** Return {@link Schema} which describes the fields. */
  public Schema getSchema() {
    return schema;
  }

  @Override
  public boolean equals(@Nullable Object o) {
    if (this == o) {
      return true;
    }
    if (!(o instanceof Row)) {
      return false;
    }
    Row other = (Row) o;

    if (!Objects.equals(getSchema(), other.getSchema())) {
      return false;
    }

    for (int i = 0; i < getFieldCount(); i++) {
      if (!Equals.deepEquals(getValue(i), other.getValue(i), getSchema().getField(i).getType())) {
        return false;
      }
    }

    return true;
  }

  @Override
  public int hashCode() {
    int h = 1;
    for (int i = 0; i < getFieldCount(); i++) {
      h = 31 * h + Equals.deepHashCode(getValue(i), getSchema().getField(i).getType());
    }

    return h;
  }

  public static class Equals {
    public static boolean deepEquals(Object a, Object b, Schema.FieldType fieldType) {
      if (a == null || b == null) {
        return a == b;
      } else if (fieldType.getTypeName() == TypeName.LOGICAL_TYPE) {
        Schema.LogicalType<Object, Object> logicalType =
            (Schema.LogicalType<Object, Object>) fieldType.getLogicalType();
        return deepEquals(
            SchemaUtils.toLogicalBaseType(logicalType, a),
            SchemaUtils.toLogicalBaseType(logicalType, b),
            logicalType.getBaseType());
      } else if (fieldType.getTypeName() == Schema.TypeName.BYTES) {
        return Arrays.equals((byte[]) a, (byte[]) b);
      } else if (fieldType.getTypeName() == TypeName.ARRAY) {
        return deepEqualsForCollection(
            (Collection<Object>) a, (Collection<Object>) b, fieldType.getCollectionElementType());
      } else if (fieldType.getTypeName() == TypeName.ITERABLE) {
        return deepEqualsForIterable(
            (Iterable<Object>) a, (Iterable<Object>) b, fieldType.getCollectionElementType());
      } else if (fieldType.getTypeName() == Schema.TypeName.MAP) {
        return deepEqualsForMap(
            (Map<Object, Object>) a, (Map<Object, Object>) b, fieldType.getMapValueType());
      } else {
        return Objects.equals(a, b);
      }
    }

    public static int deepHashCode(Object a, Schema.FieldType fieldType) {
      if (a == null) {
        return 0;
      } else if (fieldType.getTypeName() == TypeName.LOGICAL_TYPE) {
        return deepHashCode(a, fieldType.getLogicalType().getBaseType());
      } else if (fieldType.getTypeName() == Schema.TypeName.BYTES) {
        return Arrays.hashCode((byte[]) a);
      } else if (fieldType.getTypeName().isCollectionType()) {
        return deepHashCodeForIterable((Iterable<Object>) a, fieldType.getCollectionElementType());
      } else if (fieldType.getTypeName() == Schema.TypeName.MAP) {
        return deepHashCodeForMap(
            (Map<Object, Object>) a, fieldType.getMapKeyType(), fieldType.getMapValueType());
      } else {
        return Objects.hashCode(a);
      }
    }

    static <K, V> boolean deepEqualsForMap(Map<K, V> a, Map<K, V> b, Schema.FieldType valueType) {
      if (a == b) {
        return true;
      }

      if (a.size() != b.size()) {
        return false;
      }

      for (Map.Entry<K, V> e : a.entrySet()) {
        K key = e.getKey();
        V value = e.getValue();
        V otherValue = b.get(key);

        if (value == null) {
          if (otherValue != null || !b.containsKey(key)) {
            return false;
          }
        } else {
          if (!deepEquals(value, otherValue, valueType)) {
            return false;
          }
        }
      }

      return true;
    }

    static int deepHashCodeForMap(
        Map<Object, Object> a, Schema.FieldType keyType, Schema.FieldType valueType) {
      int h = 0;

      for (Map.Entry<Object, Object> e : a.entrySet()) {
        Object key = e.getKey();
        Object value = e.getValue();

        h += deepHashCode(key, keyType) ^ deepHashCode(value, valueType);
      }

      return h;
    }

    static boolean deepEqualsForCollection(
        Collection<Object> a, Collection<Object> b, Schema.FieldType elementType) {
      if (a == b) {
        return true;
      }

      if (a.size() != b.size()) {
        return false;
      }

      return deepEqualsForIterable(a, b, elementType);
    }

    static boolean deepEqualsForIterable(
        Iterable<Object> a, Iterable<Object> b, Schema.FieldType elementType) {
      if (a == b) {
        return true;
      }
      Iterator<Object> bIter = b.iterator();
      for (Object currentA : a) {
        if (!bIter.hasNext()) {
          return false;
        }
        if (!deepEquals(currentA, bIter.next(), elementType)) {
          return false;
        }
      }
      return !bIter.hasNext();
    }

    static int deepHashCodeForIterable(Iterable<Object> a, Schema.FieldType elementType) {
      int h = 1;
      for (Object o : a) {
        h = 31 * h + deepHashCode(o, elementType);
      }

      return h;
    }
  }

  @Override
  public String toString() {
    return SchemaUtils.toPrettyString(this);
  }

  /** Convert Row to String. */
  public String toString(boolean includeFieldNames) {
    StringBuilder builder = new StringBuilder();
    builder.append("Row: ");
    builder.append(System.lineSeparator());
    for (int i = 0; i < getSchema().getFieldCount(); ++i) {
      Schema.Field field = getSchema().getField(i);
      if (includeFieldNames) {
        builder.append(field.getName() + ":");
      }
      builder.append(toString(field.getType(), getValue(i), includeFieldNames));
      builder.append(System.lineSeparator());
    }
    return builder.toString();
  }

  private String toString(Schema.FieldType fieldType, Object value, boolean includeFieldNames) {
    if (value == null) {
      return "<null>";
    }
    StringBuilder builder = new StringBuilder();
    switch (fieldType.getTypeName()) {
      case ARRAY:
      case ITERABLE:
        builder.append("[");
        for (Object element : (Iterable<?>) value) {
          builder.append(
              toString(fieldType.getCollectionElementType(), element, includeFieldNames));
          builder.append(", ");
        }
        builder.append("]");
        break;
      case MAP:
        builder.append("{");
        for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
          builder.append("(");
          builder.append(toString(fieldType.getMapKeyType(), entry.getKey(), includeFieldNames));
          builder.append(", ");
          builder.append(
              toString(fieldType.getMapValueType(), entry.getValue(), includeFieldNames));
          builder.append("), ");
        }
        builder.append("}");
        break;
      case BYTES:
        builder.append(Arrays.toString((byte[]) value));
        break;
      case ROW:
        builder.append(((Row) value).toString(includeFieldNames));
        break;
      default:
        builder.append(value);
    }
    return builder.toString();
  }

  /**
   * Creates a row builder with specified {@link #getSchema()}. {@link Builder#build()} will throw
   * an {@link IllegalArgumentException} if number of fields in {@link #getSchema()} does not match
   * the number of fields specified. If any of the arguments don't match the expected types for the
   * schema fields, {@link Builder#build()} will throw a {@link ClassCastException}.
   */
  public static Builder withSchema(Schema schema) {
    return new Builder(schema);
  }

  /**
   * Creates a row builder based on the specified row. Field values in the new row can be explicitly
   * set using {@link FieldValueBuilder#withFieldValue}. Any values not so overridden will be the
   * same as the values in the original row.
   */
  public static FieldValueBuilder fromRow(Row row) {
    return new FieldValueBuilder(row.getSchema(), row);
  }

  /** Builder for {@link Row} that bases a row on another row. */
  public static class FieldValueBuilder {
    private final Schema schema;
    private final @Nullable Row sourceRow;
    private final FieldOverrides fieldOverrides;

    private FieldValueBuilder(Schema schema, @Nullable Row sourceRow) {
      this.schema = schema;
      this.sourceRow = sourceRow;
      this.fieldOverrides = new FieldOverrides(schema);
    }

    public Schema getSchema() {
      return schema;
    }

    /**
     * Set a field value using the field name. Nested values can be set using the field selection
     * syntax.
     */
    public FieldValueBuilder withFieldValue(String fieldName, Object value) {
      return withFieldValue(FieldAccessDescriptor.withFieldNames(fieldName), value);
    }

    /** Set a field value using the field id. */
    public FieldValueBuilder withFieldValue(Integer fieldId, Object value) {
      return withFieldValue(FieldAccessDescriptor.withFieldIds(fieldId), value);
    }

    /** Set a field value using a FieldAccessDescriptor. */
    public FieldValueBuilder withFieldValue(
        FieldAccessDescriptor fieldAccessDescriptor, Object value) {
      FieldAccessDescriptor fieldAccess = fieldAccessDescriptor.resolve(getSchema());
      checkArgument(fieldAccess.referencesSingleField(), "");
      fieldOverrides.addOverride(fieldAccess, new FieldOverride(value));
      return this;
    }

    /**
     * Sets field values using the field names. Nested values can be set using the field selection
     * syntax.
     */
    public FieldValueBuilder withFieldValues(Map<String, Object> values) {
      values.entrySet().stream()
          .forEach(
              e ->
                  fieldOverrides.addOverride(
                      FieldAccessDescriptor.withFieldNames(e.getKey()).resolve(getSchema()),
                      new FieldOverride(e.getValue())));
      return this;
    }

    /**
     * Sets field values using the FieldAccessDescriptors. Nested values can be set using the field
     * selection syntax.
     */
    public FieldValueBuilder withFieldAccessDescriptors(Map<FieldAccessDescriptor, Object> values) {
      values.entrySet().stream()
          .forEach(e -> fieldOverrides.addOverride(e.getKey(), new FieldOverride(e.getValue())));
      return this;
    }

    public Row build() {
      Row row =
          (Row)
              new RowFieldMatcher()
                  .match(
                      new CapturingRowCases(getSchema(), this.fieldOverrides),
                      FieldType.row(getSchema()),
                      new RowPosition(FieldAccessDescriptor.create()),
                      sourceRow);
      return row;
    }
  }

  /** Builder for {@link Row}. */
  public static class Builder {
    private List<Object> values = Lists.newArrayList();
    private final Schema schema;

    Builder(Schema schema) {
      this.schema = schema;
    }

    /** Return the schema for the row being built. */
    public Schema getSchema() {
      return schema;
    }

    /**
     * Set a field value using the field name. Nested values can be set using the field selection
     * syntax.
     */
    public FieldValueBuilder withFieldValue(String fieldName, Object value) {
      checkState(values.isEmpty());
      return new FieldValueBuilder(schema, null).withFieldValue(fieldName, value);
    }

    /** Set a field value using the field id. */
    public FieldValueBuilder withFieldValue(Integer fieldId, Object value) {
      checkState(values.isEmpty());
      return new FieldValueBuilder(schema, null).withFieldValue(fieldId, value);
    }

    /** Set a field value using a FieldAccessDescriptor. */
    public FieldValueBuilder withFieldValue(
        FieldAccessDescriptor fieldAccessDescriptor, Object value) {
      checkState(values.isEmpty());
      return new FieldValueBuilder(schema, null).withFieldValue(fieldAccessDescriptor, value);
    }

    /**
     * Sets field values using the field names. Nested values can be set using the field selection
     * syntax.
     */
    public FieldValueBuilder withFieldValues(Map<String, Object> values) {
      checkState(this.values.isEmpty());
      return new FieldValueBuilder(schema, null).withFieldValues(values);
    }

    // The following methods allow appending a list of values to the Builder object. The values must
    // be in the same
    // order as the fields in the row. These methods cannot be used in conjunction with
    // withFieldValue or
    // withFieldValues.

    public Builder addValue(@Nullable Object value) {
      this.values.add(value);
      return this;
    }

    public Builder addValues(List<@Nullable Object> values) {
      this.values.addAll(values);
      return this;
    }

    public Builder addValues(Object... values) {
      return addValues(Arrays.asList(values));
    }

    public <T> Builder addArray(Collection<T> values) {
      this.values.add(values);
      return this;
    }

    public Builder addArray(Object... values) {
      addArray(Arrays.asList(values));
      return this;
    }

    public <T> Builder addIterable(Iterable<T> values) {
      this.values.add(values);
      return this;
    }

    // Values are attached. No verification is done, and no conversions are done. LogicalType values
    // must be specified as the base type. This method should be used with great care, as no
    // validation is done. If
    // incorrect values are passed in, it could result in strange errors later in the pipeline. This
    // method is largely
    // used internal to Beam.
    @Internal
    public Row attachValues(List<@Nullable Object> attachedValues) {
      checkState(this.values.isEmpty());
      return new RowWithStorage(schema, attachedValues);
    }

    public Row attachValues(@Nullable Object... values) {
      return attachValues(Arrays.asList(values));
    }

    public int nextFieldId() {
      return values.size();
    }

    @Internal
    public <T> Row withFieldValueGetters(
        Factory<List<FieldValueGetter<T, Object>>> fieldValueGetterFactory,
        T getterTarget,
        TypeDescriptor<?> getterTargetType) {
      checkState(getterTarget != null, "getters require withGetterTarget.");
      return new RowWithGetters<>(schema, fieldValueGetterFactory, getterTarget, getterTargetType);
    }

    public Row build() {
      checkNotNull(schema);

      if (!values.isEmpty() && values.size() != schema.getFieldCount()) {
        throw new IllegalArgumentException(
            "Row expected "
                + schema.getFieldCount()
                + String.format(
                    " fields (%s).",
                    schema.getFields().stream()
                        .map(Object::toString)
                        .collect(Collectors.joining(", ")))
                + " initialized with "
                + values.size()
                + " fields.");
      }

      if (!values.isEmpty()) {
        FieldOverrides fieldOverrides = new FieldOverrides(schema, this.values);
        if (!fieldOverrides.isEmpty()) {
          return (Row)
              new RowFieldMatcher()
                  .match(
                      new CapturingRowCases(schema, fieldOverrides),
                      FieldType.row(schema),
                      new RowPosition(FieldAccessDescriptor.create()),
                      null);
        }
      }
      return new RowWithStorage(schema, Collections.emptyList());
    }
  }

  /** Creates a {@link Row} from the list of values and {@link #getSchema()}. */
  public static <T> Collector<T, List<Object>, Row> toRow(Schema schema) {
    return Collector.of(
        () -> new ArrayList<>(schema.getFieldCount()),
        List::add,
        (left, right) -> {
          left.addAll(right);
          return left;
        },
        values -> Row.withSchema(schema).addValues(values).build());
  }

  /** Creates a new record filled with nulls. */
  public static Row nullRow(Schema schema) {
    return Row.withSchema(schema)
        .addValues(Collections.nCopies(schema.getFieldCount(), null))
        .build();
  }

  /** Returns an equivalent {@link Row} with fields lexicographically sorted by their name. */
  public Row sorted() {
    Schema sortedSchema = getSchema().sorted();
    return sortedSchema.getFields().stream()
        .map(
            field -> {
              if (field.getType().getRowSchema() != null) {
                Row innerRow = getValue(field.getName());
                if (innerRow != null) {
                  return innerRow.sorted();
                }
              }
              return (Object) getValue(field.getName());
            })
        .collect(Row.toRow(sortedSchema));
  }

  /** Returns an equivalent {@link Row} with `snake_case` field names. */
  public Row toSnakeCase() {
    return getSchema().getFields().stream()
        .map(
            field -> {
              if (field.getType().getRowSchema() != null) {
                Row innerRow = getValue(field.getName());
                if (innerRow != null) {
                  return innerRow.toSnakeCase();
                }
              }
              return (Object) getValue(field.getName());
            })
        .collect(toRow(getSchema().toSnakeCase()));
  }

  /** Returns an equivalent {@link Row} with `lowerCamelCase` field names. */
  public Row toCamelCase() {
    return getSchema().getFields().stream()
        .map(
            field -> {
              if (field.getType().getRowSchema() != null) {
                Row innerRow = getValue(field.getName());
                if (innerRow != null) {
                  return innerRow.toCamelCase();
                }
              }
              return (Object) getValue(field.getName());
            })
        .collect(toRow(getSchema().toCamelCase()));
  }
}
