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

import static org.apache.iceberg.TestHelpers.ALL_VERSIONS;
import static org.apache.iceberg.types.Conversions.fromByteBuffer;
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assumptions.assumeThat;

import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.BinaryType;
import org.apache.iceberg.types.Types.BooleanType;
import org.apache.iceberg.types.Types.DateType;
import org.apache.iceberg.types.Types.DecimalType;
import org.apache.iceberg.types.Types.DoubleType;
import org.apache.iceberg.types.Types.FixedType;
import org.apache.iceberg.types.Types.FloatType;
import org.apache.iceberg.types.Types.IntegerType;
import org.apache.iceberg.types.Types.ListType;
import org.apache.iceberg.types.Types.LongType;
import org.apache.iceberg.types.Types.MapType;
import org.apache.iceberg.types.Types.StringType;
import org.apache.iceberg.types.Types.StructType;
import org.apache.iceberg.types.Types.TimeType;
import org.apache.iceberg.types.Types.TimestampType;
import org.apache.iceberg.util.DateTimeUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.io.TempDir;

/** Tests for Metrics. */
public abstract class TestMetrics {

  @Parameters(name = "formatVersion = {0}")
  protected static List<Integer> formatVersions() {
    return ALL_VERSIONS;
  }

  @TempDir protected Path temp;
  @TempDir private File tableDir;

  private static final StructType LEAF_STRUCT_TYPE =
      StructType.of(
          optional(5, "leafLongCol", LongType.get()),
          optional(6, "leafBinaryCol", BinaryType.get()));

  private static final StructType NESTED_STRUCT_TYPE =
      StructType.of(
          required(3, "longCol", LongType.get()),
          required(4, "leafStructCol", LEAF_STRUCT_TYPE),
          required(7, "doubleCol", DoubleType.get()));

  private static final Schema NESTED_SCHEMA =
      new Schema(
          required(1, "intCol", IntegerType.get()),
          required(2, "nestedStructCol", NESTED_STRUCT_TYPE));

  private static final Schema SIMPLE_SCHEMA =
      new Schema(
          optional(1, "booleanCol", BooleanType.get()),
          required(2, "intCol", IntegerType.get()),
          optional(3, "longCol", LongType.get()),
          required(4, "floatCol", FloatType.get()),
          optional(5, "doubleCol", DoubleType.get()),
          optional(6, "decimalCol", DecimalType.of(10, 2)),
          required(7, "stringCol", StringType.get()),
          optional(8, "dateCol", DateType.get()),
          required(9, "timeCol", TimeType.get()),
          required(10, "timestampColAboveEpoch", TimestampType.withoutZone()),
          required(11, "fixedCol", FixedType.ofLength(4)),
          required(12, "binaryCol", BinaryType.get()),
          required(13, "timestampColBelowEpoch", TimestampType.withoutZone()));

  private static final Schema FLOAT_DOUBLE_ONLY_SCHEMA =
      new Schema(
          optional(1, "floatCol", FloatType.get()), optional(2, "doubleCol", DoubleType.get()));

  private static final Record FLOAT_DOUBLE_RECORD_1 = createRecordWithFloatAndDouble(1.2F, 3.4D);
  private static final Record FLOAT_DOUBLE_RECORD_2 = createRecordWithFloatAndDouble(5.6F, 7.8D);
  private static final Record NAN_ONLY_RECORD =
      createRecordWithFloatAndDouble(Float.NaN, Double.NaN);

  @Parameter private int formatVersion;
  private final byte[] fixed = "abcd".getBytes(StandardCharsets.UTF_8);

  @AfterEach
  public void after() {
    TestTables.clearTables();
  }

  private static Record createRecordWithFloatAndDouble(float floatValue, double doubleValue) {
    Record record = GenericRecord.create(FLOAT_DOUBLE_ONLY_SCHEMA);
    record.setField("floatCol", floatValue);
    record.setField("doubleCol", doubleValue);
    return record;
  }

  public abstract FileFormat fileFormat();

  public abstract Metrics getMetrics(Schema schema, MetricsConfig metricsConfig, Record... records)
      throws IOException;

  public abstract Metrics getMetrics(Schema schema, Record... records) throws IOException;

  protected abstract Metrics getMetricsForRecordsWithSmallRowGroups(
      Schema schema, OutputFile outputFile, Record... records) throws IOException;

  public abstract int splitCount(InputFile inputFile) throws IOException;

  public boolean supportsSmallRowGroups() {
    return false;
  }

  protected abstract OutputFile createOutputFile() throws IOException;

  @TestTemplate
  public void testMetricsForRepeatedValues() throws IOException {
    Record record = GenericRecord.create(SIMPLE_SCHEMA);
    record.setField("booleanCol", true);
    record.setField("intCol", 3);
    record.setField("longCol", null);
    record.setField("floatCol", Float.NaN);
    record.setField("doubleCol", 2.0D);
    record.setField("decimalCol", new BigDecimal("3.50"));
    record.setField("stringCol", "AAA");
    record.setField("dateCol", DateTimeUtil.dateFromDays(1500));
    record.setField("timeCol", DateTimeUtil.timeFromMicros(2000L));
    record.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(0L));
    record.setField("fixedCol", fixed);
    record.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
    record.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(0L));

    Metrics metrics = getMetrics(SIMPLE_SCHEMA, record, record);
    assertThat(metrics.recordCount()).isEqualTo(2L);
    assertCounts(1, 2L, 0L, metrics);
    assertCounts(2, 2L, 0L, metrics);
    assertCounts(3, 2L, 2L, metrics);
    assertCounts(4, 2L, 0L, 2L, metrics);
    assertCounts(5, 2L, 0L, 0L, metrics);
    assertCounts(6, 2L, 0L, metrics);
    assertCounts(7, 2L, 0L, metrics);
    assertCounts(8, 2L, 0L, metrics);
    assertCounts(9, 2L, 0L, metrics);
    assertCounts(10, 2L, 0L, metrics);
    assertCounts(11, 2L, 0L, metrics);
    assertCounts(12, 2L, 0L, metrics);
    assertCounts(13, 2L, 0L, metrics);
  }

  @TestTemplate
  public void testMetricsForTopLevelFields() throws IOException {
    Record firstRecord = GenericRecord.create(SIMPLE_SCHEMA);
    firstRecord.setField("booleanCol", true);
    firstRecord.setField("intCol", 3);
    firstRecord.setField("longCol", 5L);
    firstRecord.setField("floatCol", 2.0F);
    firstRecord.setField("doubleCol", 2.0D);
    firstRecord.setField("decimalCol", new BigDecimal("3.50"));
    firstRecord.setField("stringCol", "AAA");
    firstRecord.setField("dateCol", DateTimeUtil.dateFromDays(1500));
    firstRecord.setField("timeCol", DateTimeUtil.timeFromMicros(2000L));
    firstRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(0L));
    firstRecord.setField("fixedCol", fixed);
    firstRecord.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
    firstRecord.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(-1_900_300L));
    Record secondRecord = GenericRecord.create(SIMPLE_SCHEMA);
    secondRecord.setField("booleanCol", false);
    secondRecord.setField("intCol", Integer.MIN_VALUE);
    secondRecord.setField("longCol", null);
    secondRecord.setField("floatCol", 1.0F);
    secondRecord.setField("doubleCol", null);
    secondRecord.setField("decimalCol", null);
    secondRecord.setField("stringCol", "ZZZ");
    secondRecord.setField("dateCol", null);
    secondRecord.setField("timeCol", DateTimeUtil.timeFromMicros(3000L));
    secondRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(900L));
    secondRecord.setField("fixedCol", fixed);
    secondRecord.setField("binaryCol", ByteBuffer.wrap("W".getBytes()));
    secondRecord.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(-7_000L));

    Metrics metrics = getMetrics(SIMPLE_SCHEMA, firstRecord, secondRecord);
    assertThat(metrics.recordCount()).isEqualTo(2L);
    assertCounts(1, 2L, 0L, metrics);
    assertBounds(1, BooleanType.get(), false, true, metrics);
    assertCounts(2, 2L, 0L, metrics);
    assertBounds(2, IntegerType.get(), Integer.MIN_VALUE, 3, metrics);
    assertCounts(3, 2L, 1L, metrics);
    assertBounds(3, LongType.get(), 5L, 5L, metrics);
    assertCounts(4, 2L, 0L, 0L, metrics);
    assertBounds(4, FloatType.get(), 1.0F, 2.0F, metrics);
    assertCounts(5, 2L, 1L, 0L, metrics);
    assertBounds(5, DoubleType.get(), 2.0D, 2.0D, metrics);
    assertCounts(6, 2L, 1L, metrics);
    assertBounds(6, DecimalType.of(10, 2), new BigDecimal("3.50"), new BigDecimal("3.50"), metrics);
    assertCounts(7, 2L, 0L, metrics);
    assertBounds(7, StringType.get(), CharBuffer.wrap("AAA"), CharBuffer.wrap("ZZZ"), metrics);
    assertCounts(8, 2L, 1L, metrics);
    assertBounds(8, DateType.get(), 1500, 1500, metrics);
    assertCounts(9, 2L, 0L, metrics);
    assertBounds(9, TimeType.get(), 2000L, 3000L, metrics);
    assertCounts(10, 2L, 0L, metrics);
    assertBounds(10, TimestampType.withoutZone(), 0L, 900L, metrics);
    assertCounts(11, 2L, 0L, metrics);
    assertBounds(
        11, FixedType.ofLength(4), ByteBuffer.wrap(fixed), ByteBuffer.wrap(fixed), metrics);
    assertCounts(12, 2L, 0L, metrics);
    assertBounds(
        12,
        BinaryType.get(),
        ByteBuffer.wrap("S".getBytes()),
        ByteBuffer.wrap("W".getBytes()),
        metrics);
    if (fileFormat() == FileFormat.ORC) {
      // TODO: The special condition for ORC can be removed when ORC-342 is fixed
      // ORC-342: ORC writer creates inaccurate timestamp data and stats 1 sec below epoch
      // Values in the range `[1969-12-31 23:59:59.000,1969-12-31 23:59:59.999]` will have 1 sec
      // added to them
      // So the upper bound value of -7_000 micros becomes 993_000 micros
      assertBounds(13, TimestampType.withoutZone(), -1_900_300L, 993_000L, metrics);
    } else {
      assertBounds(13, TimestampType.withoutZone(), -1_900_300L, -7_000L, metrics);
    }
  }

  @TestTemplate
  public void testMetricsForGeospatialTypes() throws IOException {
    // Geometry and geography are only written by the Parquet path; other formats do not support
    // the geo types yet.
    assumeThat(fileFormat()).isEqualTo(FileFormat.PARQUET);

    Schema schema =
        new Schema(
            required(1, "id", LongType.get()),
            optional(2, "geom", Types.GeometryType.crs84()),
            optional(3, "geog", Types.GeographyType.crs84()));

    Record first = GenericRecord.create(schema);
    first.setField("id", 1L);
    first.setField("geom", wkbPoint(30, 10));
    first.setField("geog", wkbPoint(-5, 40));
    Record second = GenericRecord.create(schema);
    second.setField("id", 2L);
    // both geo columns are left null

    Metrics metrics = getMetrics(schema, first, second);
    assertThat(metrics.recordCount()).isEqualTo(2L);

    // geometry and geography keep value/null counts but no bounds: lexicographic WKB min/max is not
    // meaningful, so bounds are intentionally skipped (spatial bounds are a separate follow-up).
    assertCounts(2, 2L, 1L, metrics);
    assertBounds(2, Types.GeometryType.crs84(), null, null, metrics);
    assertCounts(3, 2L, 1L, metrics);
    assertBounds(3, Types.GeographyType.crs84(), null, null, metrics);
  }

  @TestTemplate
  public void testMetricsForDecimals() throws IOException {
    Schema schema =
        new Schema(
            required(1, "decimalAsInt32", DecimalType.of(4, 2)),
            required(2, "decimalAsInt64", DecimalType.of(14, 2)),
            required(3, "decimalAsFixed", DecimalType.of(22, 2)));

    Record record = GenericRecord.create(schema);
    record.setField("decimalAsInt32", new BigDecimal("2.55"));
    record.setField("decimalAsInt64", new BigDecimal("4.75"));
    record.setField("decimalAsFixed", new BigDecimal("5.80"));

    Metrics metrics = getMetrics(schema, record);
    assertThat(metrics.recordCount()).isEqualTo(1);
    assertCounts(1, 1L, 0L, metrics);
    assertBounds(1, DecimalType.of(4, 2), new BigDecimal("2.55"), new BigDecimal("2.55"), metrics);
    assertCounts(2, 1L, 0L, metrics);
    assertBounds(2, DecimalType.of(14, 2), new BigDecimal("4.75"), new BigDecimal("4.75"), metrics);
    assertCounts(3, 1L, 0L, metrics);
    assertBounds(3, DecimalType.of(22, 2), new BigDecimal("5.80"), new BigDecimal("5.80"), metrics);
  }

  @TestTemplate
  public void testMetricsForNestedStructFields() throws IOException {
    Metrics metrics = getMetrics(NESTED_SCHEMA, buildNestedTestRecord());
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertCounts(1, 1L, 0L, metrics);
    assertBounds(1, IntegerType.get(), Integer.MAX_VALUE, Integer.MAX_VALUE, metrics);
    assertCounts(3, 1L, 0L, metrics);
    assertBounds(3, LongType.get(), 100L, 100L, metrics);
    assertCounts(5, 1L, 0L, metrics);
    assertBounds(5, LongType.get(), 20L, 20L, metrics);
    assertCounts(6, 1L, 0L, metrics);
    assertBounds(
        6,
        BinaryType.get(),
        ByteBuffer.wrap("A".getBytes()),
        ByteBuffer.wrap("A".getBytes()),
        metrics);
    assertCounts(7, 1L, 0L, 1L, metrics);
    assertBounds(7, DoubleType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testMetricsModeForNestedStructFields() throws IOException {
    Map<String, String> properties =
        ImmutableMap.of(
            TableProperties.DEFAULT_WRITE_METRICS_MODE,
            MetricsModes.None.get().toString(),
            TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "nestedStructCol.longCol",
            MetricsModes.Full.get().toString());
    MetricsConfig config = MetricsConfig.from(properties, NESTED_SCHEMA, null);

    Metrics metrics = getMetrics(NESTED_SCHEMA, config, buildNestedTestRecord());
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertThat(metrics.lowerBounds()).hasSize(1);
    assertThat(metrics.upperBounds()).hasSize(1);
    assertBounds(3, LongType.get(), 100L, 100L, metrics);
  }

  private Record buildNestedTestRecord() {
    Record leafStruct = GenericRecord.create(LEAF_STRUCT_TYPE);
    leafStruct.setField("leafLongCol", 20L);
    leafStruct.setField("leafBinaryCol", ByteBuffer.wrap("A".getBytes()));
    Record nestedStruct = GenericRecord.create(NESTED_STRUCT_TYPE);
    nestedStruct.setField("longCol", 100L);
    nestedStruct.setField("leafStructCol", leafStruct);
    nestedStruct.setField("doubleCol", Double.NaN);
    Record record = GenericRecord.create(NESTED_SCHEMA);
    record.setField("intCol", Integer.MAX_VALUE);
    record.setField("nestedStructCol", nestedStruct);

    return record;
  }

  @TestTemplate
  public void testMetricsForListAndMapElements() throws IOException {
    StructType structType =
        StructType.of(
            required(1, "leafIntCol", IntegerType.get()),
            optional(2, "leafStringCol", StringType.get()));
    Schema schema =
        new Schema(
            optional(3, "intListCol", ListType.ofRequired(4, IntegerType.get())),
            optional(5, "mapCol", MapType.ofRequired(6, 7, StringType.get(), structType)));

    Record record = GenericRecord.create(schema);
    record.setField("intListCol", Lists.newArrayList(10, 11, 12));
    Record struct = GenericRecord.create(structType);
    struct.setField("leafIntCol", 1);
    struct.setField("leafStringCol", "BBB");
    Map<String, Record> map = Maps.newHashMap();
    map.put("4", struct);
    record.set(1, map);

    Metrics metrics = getMetrics(schema, record);
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertCounts(1, null, null, metrics);
    assertCounts(2, null, null, metrics);
    assertCounts(4, null, null, metrics);
    assertCounts(6, null, null, metrics);
    assertBounds(1, IntegerType.get(), null, null, metrics);
    assertBounds(2, StringType.get(), null, null, metrics);
    assertBounds(4, IntegerType.get(), null, null, metrics);
    assertBounds(6, StringType.get(), null, null, metrics);
    assertBounds(7, structType, null, null, metrics);
  }

  @TestTemplate
  public void testMetricsForNullColumns() throws IOException {
    Schema schema = new Schema(optional(1, "intCol", IntegerType.get()));
    Record firstRecord = GenericRecord.create(schema);
    firstRecord.setField("intCol", null);
    Record secondRecord = GenericRecord.create(schema);
    secondRecord.setField("intCol", null);

    Metrics metrics = getMetrics(schema, firstRecord, secondRecord);
    assertThat(metrics.recordCount()).isEqualTo(2L);
    assertCounts(1, 2L, 2L, metrics);
    assertBounds(1, IntegerType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testMetricsForNaNColumns() throws IOException {
    Metrics metrics = getMetrics(FLOAT_DOUBLE_ONLY_SCHEMA, NAN_ONLY_RECORD, NAN_ONLY_RECORD);
    assertThat(metrics.recordCount()).isEqualTo(2L);
    assertCounts(1, 2L, 0L, 2L, metrics);
    assertCounts(2, 2L, 0L, 2L, metrics);

    assertBounds(1, FloatType.get(), null, null, metrics);
    assertBounds(2, DoubleType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testColumnBoundsWithNaNValueAtFront() throws IOException {
    Metrics metrics =
        getMetrics(
            FLOAT_DOUBLE_ONLY_SCHEMA,
            NAN_ONLY_RECORD,
            FLOAT_DOUBLE_RECORD_1,
            FLOAT_DOUBLE_RECORD_2);
    assertThat(metrics.recordCount()).isEqualTo(3L);
    assertCounts(1, 3L, 0L, 1L, metrics);
    assertCounts(2, 3L, 0L, 1L, metrics);

    assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
    assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
  }

  @TestTemplate
  public void testColumnBoundsWithNaNValueInMiddle() throws IOException {
    Metrics metrics =
        getMetrics(
            FLOAT_DOUBLE_ONLY_SCHEMA,
            FLOAT_DOUBLE_RECORD_1,
            NAN_ONLY_RECORD,
            FLOAT_DOUBLE_RECORD_2);
    assertThat(metrics.recordCount()).isEqualTo(3L);
    assertCounts(1, 3L, 0L, 1L, metrics);
    assertCounts(2, 3L, 0L, 1L, metrics);

    assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
    assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
  }

  @TestTemplate
  public void testColumnBoundsWithNaNValueAtEnd() throws IOException {
    Metrics metrics =
        getMetrics(
            FLOAT_DOUBLE_ONLY_SCHEMA,
            FLOAT_DOUBLE_RECORD_1,
            FLOAT_DOUBLE_RECORD_2,
            NAN_ONLY_RECORD);
    assertThat(metrics.recordCount()).isEqualTo(3L);
    assertCounts(1, 3L, 0L, 1L, metrics);
    assertCounts(2, 3L, 0L, 1L, metrics);

    assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
    assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
  }

  @TestTemplate
  public void testMetricsForTopLevelWithMultipleRowGroup() throws Exception {
    assumeThat(supportsSmallRowGroups())
        .as("Skip test for formats that do not support small row groups")
        .isTrue();

    int recordCount = 201;
    List<Record> records = Lists.newArrayListWithExpectedSize(recordCount);

    for (int i = 0; i < recordCount; i++) {
      Record newRecord = GenericRecord.create(SIMPLE_SCHEMA);
      newRecord.setField("booleanCol", i != 0);
      newRecord.setField("intCol", i + 1);
      newRecord.setField("longCol", i == 0 ? null : i + 1L);
      newRecord.setField("floatCol", i + 1.0F);
      newRecord.setField("doubleCol", i == 0 ? null : i + 1.0D);
      newRecord.setField(
          "decimalCol", i == 0 ? null : new BigDecimal(i + "").add(new BigDecimal("1.00")));
      newRecord.setField("stringCol", "AAA");
      newRecord.setField("dateCol", DateTimeUtil.dateFromDays(i + 1));
      newRecord.setField("timeCol", DateTimeUtil.timeFromMicros(i + 1L));
      newRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(i + 1L));
      newRecord.setField("fixedCol", fixed);
      newRecord.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
      newRecord.setField(
          "timestampColBelowEpoch", DateTimeUtil.timestampFromMicros((i + 1L) * -1L));
      records.add(newRecord);
    }

    // create file with multiple row groups. by using smaller number of bytes
    OutputFile outputFile = createOutputFile();
    Metrics metrics =
        getMetricsForRecordsWithSmallRowGroups(
            SIMPLE_SCHEMA, outputFile, records.toArray(new Record[0]));
    InputFile recordsFile = outputFile.toInputFile();

    assertThat(recordsFile).isNotNull();
    // rowgroup size should be > 1
    assertThat(splitCount(recordsFile)).isEqualTo(3);

    assertThat(metrics.recordCount()).isEqualTo(201L);
    assertCounts(1, 201L, 0L, metrics);
    assertBounds(1, Types.BooleanType.get(), false, true, metrics);
    assertBounds(2, Types.IntegerType.get(), 1, 201, metrics);
    assertCounts(3, 201L, 1L, metrics);
    assertBounds(3, Types.LongType.get(), 2L, 201L, metrics);
    assertCounts(4, 201L, 0L, 0L, metrics);
    assertBounds(4, Types.FloatType.get(), 1.0F, 201.0F, metrics);
    assertCounts(5, 201L, 1L, 0L, metrics);
    assertBounds(5, Types.DoubleType.get(), 2.0D, 201.0D, metrics);
    assertCounts(6, 201L, 1L, metrics);
    assertBounds(
        6, Types.DecimalType.of(10, 2), new BigDecimal("2.00"), new BigDecimal("201.00"), metrics);
  }

  @TestTemplate
  public void testMetricsForNestedStructFieldsWithMultipleRowGroup() throws IOException {
    assumeThat(supportsSmallRowGroups())
        .as("Skip test for formats that do not support small row groups")
        .isTrue();

    int recordCount = 201;
    List<Record> records = Lists.newArrayListWithExpectedSize(recordCount);

    for (int i = 0; i < recordCount; i++) {
      Record newLeafStruct = GenericRecord.create(LEAF_STRUCT_TYPE);
      newLeafStruct.setField("leafLongCol", i + 1L);
      newLeafStruct.setField("leafBinaryCol", ByteBuffer.wrap("A".getBytes()));
      Record newNestedStruct = GenericRecord.create(NESTED_STRUCT_TYPE);
      newNestedStruct.setField("longCol", i + 1L);
      newNestedStruct.setField("leafStructCol", newLeafStruct);
      newNestedStruct.setField("doubleCol", Double.NaN);
      Record newRecord = GenericRecord.create(NESTED_SCHEMA);
      newRecord.setField("intCol", i + 1);
      newRecord.setField("nestedStructCol", newNestedStruct);
      records.add(newRecord);
    }

    // create file with multiple row groups. by using smaller number of bytes
    OutputFile outputFile = createOutputFile();
    Metrics metrics =
        getMetricsForRecordsWithSmallRowGroups(
            NESTED_SCHEMA, outputFile, records.toArray(new Record[0]));
    InputFile recordsFile = outputFile.toInputFile();

    assertThat(recordsFile).isNotNull();
    // rowgroup size should be > 1
    assertThat(splitCount(recordsFile)).isEqualTo(3);

    assertThat(metrics.recordCount()).isEqualTo(201L);
    assertCounts(1, 201L, 0L, metrics);
    assertBounds(1, IntegerType.get(), 1, 201, metrics);
    assertCounts(3, 201L, 0L, metrics);
    assertBounds(3, LongType.get(), 1L, 201L, metrics);
    assertCounts(5, 201L, 0L, metrics);
    assertBounds(5, LongType.get(), 1L, 201L, metrics);
    assertCounts(6, 201L, 0L, metrics);
    assertBounds(
        6,
        BinaryType.get(),
        ByteBuffer.wrap("A".getBytes()),
        ByteBuffer.wrap("A".getBytes()),
        metrics);
    assertCounts(7, 201L, 0L, 201L, metrics);
    assertBounds(7, DoubleType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testNoneMetricsMode() throws IOException {
    Metrics metrics =
        getMetrics(
            NESTED_SCHEMA,
            MetricsConfig.from(
                ImmutableMap.of("write.metadata.metrics.default", "none"), NESTED_SCHEMA, null),
            buildNestedTestRecord());
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertThat(metrics.columnSizes()).isEmpty();
    assertCounts(1, null, null, metrics);
    assertBounds(1, Types.IntegerType.get(), null, null, metrics);
    assertCounts(3, null, null, metrics);
    assertBounds(3, Types.LongType.get(), null, null, metrics);
    assertCounts(5, null, null, metrics);
    assertBounds(5, Types.LongType.get(), null, null, metrics);
    assertCounts(6, null, null, metrics);
    assertBounds(6, Types.BinaryType.get(), null, null, metrics);
    assertCounts(7, null, null, metrics);
    assertBounds(7, Types.DoubleType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testCountsMetricsMode() throws IOException {
    Metrics metrics =
        getMetrics(
            NESTED_SCHEMA,
            MetricsConfig.from(
                ImmutableMap.of("write.metadata.metrics.default", "counts"), NESTED_SCHEMA, null),
            buildNestedTestRecord());
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertThat(metrics.columnSizes()).doesNotContainValue(null);
    assertThat(metrics.columnSizes()).isNotEmpty();
    assertCounts(1, 1L, 0L, metrics);
    assertBounds(1, Types.IntegerType.get(), null, null, metrics);
    assertCounts(3, 1L, 0L, metrics);
    assertBounds(3, Types.LongType.get(), null, null, metrics);
    assertCounts(5, 1L, 0L, metrics);
    assertBounds(5, Types.LongType.get(), null, null, metrics);
    assertCounts(6, 1L, 0L, metrics);
    assertBounds(6, Types.BinaryType.get(), null, null, metrics);
    assertCounts(7, 1L, 0L, 1L, metrics);
    assertBounds(7, Types.DoubleType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testFullMetricsMode() throws IOException {
    Metrics metrics =
        getMetrics(
            NESTED_SCHEMA,
            MetricsConfig.from(
                ImmutableMap.of("write.metadata.metrics.default", "full"), NESTED_SCHEMA, null),
            buildNestedTestRecord());
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertThat(metrics.columnSizes()).doesNotContainValue(null);
    assertThat(metrics.columnSizes()).isNotEmpty();
    assertCounts(1, 1L, 0L, metrics);
    assertBounds(1, Types.IntegerType.get(), Integer.MAX_VALUE, Integer.MAX_VALUE, metrics);
    assertCounts(3, 1L, 0L, metrics);
    assertBounds(3, Types.LongType.get(), 100L, 100L, metrics);
    assertCounts(5, 1L, 0L, metrics);
    assertBounds(5, Types.LongType.get(), 20L, 20L, metrics);
    assertCounts(6, 1L, 0L, metrics);
    assertBounds(
        6,
        Types.BinaryType.get(),
        ByteBuffer.wrap("A".getBytes()),
        ByteBuffer.wrap("A".getBytes()),
        metrics);
    assertCounts(7, 1L, 0L, 1L, metrics);
    assertBounds(7, Types.DoubleType.get(), null, null, metrics);
  }

  @TestTemplate
  public void testTruncateStringMetricsMode() throws IOException {
    String colName = "str_to_truncate";
    Schema singleStringColSchema = new Schema(required(1, colName, Types.StringType.get()));

    String value = "Lorem ipsum dolor sit amet";
    Record record = GenericRecord.create(singleStringColSchema);
    record.setField(colName, value);

    Metrics metrics =
        getMetrics(
            singleStringColSchema,
            MetricsConfig.from(
                ImmutableMap.of("write.metadata.metrics.default", "truncate(10)"),
                singleStringColSchema,
                null),
            record);

    CharBuffer expectedMinBound = CharBuffer.wrap("Lorem ipsu");
    CharBuffer expectedMaxBound = CharBuffer.wrap("Lorem ipsv");
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertThat(metrics.columnSizes()).doesNotContainValue(null);
    assertThat(metrics.columnSizes()).isNotEmpty();
    assertCounts(1, 1L, 0L, metrics);
    assertBounds(1, Types.StringType.get(), expectedMinBound, expectedMaxBound, metrics);
  }

  @TestTemplate
  public void testTruncateBinaryMetricsMode() throws IOException {
    String colName = "bin_to_truncate";
    Schema singleBinaryColSchema = new Schema(required(1, colName, Types.BinaryType.get()));

    byte[] value = new byte[] {0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9, 0x10, 0xA, 0xB};
    Record record = GenericRecord.create(singleBinaryColSchema);
    record.setField(colName, ByteBuffer.wrap(value));

    Metrics metrics =
        getMetrics(
            singleBinaryColSchema,
            MetricsConfig.from(
                ImmutableMap.of("write.metadata.metrics.default", "truncate(5)"),
                singleBinaryColSchema,
                null),
            record);

    ByteBuffer expectedMinBounds = ByteBuffer.wrap(new byte[] {0x1, 0x2, 0x3, 0x4, 0x5});
    ByteBuffer expectedMaxBounds = ByteBuffer.wrap(new byte[] {0x1, 0x2, 0x3, 0x4, 0x6});
    assertThat(metrics.recordCount()).isEqualTo(1L);
    assertThat(metrics.columnSizes()).doesNotContainValue(null);
    assertThat(metrics.columnSizes()).isNotEmpty();
    assertCounts(1, 1L, 0L, metrics);
    assertBounds(1, Types.BinaryType.get(), expectedMinBounds, expectedMaxBounds, metrics);
  }

  @TestTemplate
  public void testSortedColumnMetrics() throws IOException {
    SortOrder sortOrder =
        SortOrder.builderFor(SIMPLE_SCHEMA)
            .asc("booleanCol")
            .asc("intCol")
            .asc("longCol")
            .asc("decimalCol")
            .asc("stringCol")
            .asc("dateCol")
            .build();
    PartitionSpec spec = PartitionSpec.unpartitioned();
    Table table =
        TestTables.create(tableDir, "test", SIMPLE_SCHEMA, spec, sortOrder, formatVersion);
    table.updateProperties().set(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none").commit();

    Record firstRecord = GenericRecord.create(SIMPLE_SCHEMA);
    firstRecord.setField("booleanCol", true);
    firstRecord.setField("intCol", Integer.MIN_VALUE);
    firstRecord.setField("longCol", Long.MIN_VALUE);
    firstRecord.setField("floatCol", Float.NaN);
    firstRecord.setField("doubleCol", 2.0D);
    firstRecord.setField("decimalCol", new BigDecimal("0.00"));
    firstRecord.setField("stringCol", "AAA");
    firstRecord.setField("dateCol", DateTimeUtil.dateFromDays(1500));
    firstRecord.setField("timeCol", DateTimeUtil.timeFromMicros(2000L));
    firstRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(0L));
    firstRecord.setField("fixedCol", fixed);
    firstRecord.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
    firstRecord.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(0L));

    Record secondRecord = GenericRecord.create(SIMPLE_SCHEMA);

    secondRecord.setField("booleanCol", false);
    secondRecord.setField("intCol", Integer.MAX_VALUE);
    secondRecord.setField("longCol", Long.MAX_VALUE);
    secondRecord.setField("floatCol", Float.NaN);
    secondRecord.setField("doubleCol", 2.0D);
    secondRecord.setField("decimalCol", new BigDecimal("10.00"));
    secondRecord.setField("stringCol", "ZZZ");
    secondRecord.setField("dateCol", DateTimeUtil.dateFromDays(3000));
    secondRecord.setField("timeCol", DateTimeUtil.timeFromMicros(2000L));
    secondRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(0L));
    secondRecord.setField("fixedCol", fixed);
    secondRecord.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
    secondRecord.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(0L));

    Metrics metrics =
        getMetrics(SIMPLE_SCHEMA, MetricsConfig.forTable(table), firstRecord, secondRecord);

    assertThat(metrics.recordCount()).isEqualTo(2L);
    assertBounds(1, BooleanType.get(), false, true, metrics);
    assertBounds(2, IntegerType.get(), Integer.MIN_VALUE, Integer.MAX_VALUE, metrics);
    assertBounds(3, LongType.get(), Long.MIN_VALUE, Long.MAX_VALUE, metrics);
    assertBounds(
        6, DecimalType.of(10, 2), new BigDecimal("0.00"), new BigDecimal("10.00"), metrics);
    assertBounds(7, StringType.get(), CharBuffer.wrap("AAA"), CharBuffer.wrap("ZZZ"), metrics);
    assertBounds(8, DateType.get(), 1500, 3000, metrics);
  }

  @TestTemplate
  public void testMetricsForSortedNestedStructFields() throws IOException {
    SortOrder sortOrder =
        SortOrder.builderFor(NESTED_SCHEMA)
            .asc("nestedStructCol.longCol")
            .asc("nestedStructCol.leafStructCol.leafLongCol")
            .build();
    PartitionSpec spec = PartitionSpec.unpartitioned();
    Table table =
        TestTables.create(tableDir, "nested", NESTED_SCHEMA, spec, sortOrder, formatVersion);

    Record leafStruct = GenericRecord.create(LEAF_STRUCT_TYPE);
    leafStruct.setField("leafLongCol", Long.MAX_VALUE);
    leafStruct.setField("leafBinaryCol", ByteBuffer.wrap("A".getBytes()));
    Record nestedStruct = GenericRecord.create(NESTED_STRUCT_TYPE);
    nestedStruct.setField("longCol", Long.MAX_VALUE);
    nestedStruct.setField("leafStructCol", leafStruct);
    nestedStruct.setField("doubleCol", Double.NaN);
    Record record = GenericRecord.create(NESTED_SCHEMA);
    record.setField("intCol", Integer.MAX_VALUE);
    record.setField("nestedStructCol", nestedStruct);

    Metrics metrics = getMetrics(NESTED_SCHEMA, MetricsConfig.forTable(table), record);

    assertBounds(3, LongType.get(), Long.MAX_VALUE, Long.MAX_VALUE, metrics);
    assertBounds(5, LongType.get(), Long.MAX_VALUE, Long.MAX_VALUE, metrics);
  }

  private static ByteBuffer wkbPoint(double xCoord, double yCoord) {
    // little-endian WKB encoding of a point
    return ByteBuffer.wrap(
        ByteBuffer.allocate(21)
            .order(ByteOrder.LITTLE_ENDIAN)
            .put((byte) 1) // byte order: little endian
            .putInt(1) // WKB geometry type: Point
            .putDouble(xCoord)
            .putDouble(yCoord)
            .array());
  }

  protected void assertCounts(int fieldId, Long valueCount, Long nullValueCount, Metrics metrics) {
    assertCounts(fieldId, valueCount, nullValueCount, null, metrics);
  }

  protected void assertCounts(
      int fieldId, Long valueCount, Long nullValueCount, Long nanValueCount, Metrics metrics) {
    assertThat(metrics.valueCounts().get(fieldId)).isEqualTo(valueCount);
    assertThat(metrics.nullValueCounts().get(fieldId)).isEqualTo(nullValueCount);
    assertThat(metrics.nanValueCounts().get(fieldId)).isEqualTo(nanValueCount);
  }

  protected <T> void assertBounds(
      int fieldId, Type type, T lowerBound, T upperBound, Metrics metrics) {
    Map<Integer, ByteBuffer> lowerBounds = metrics.lowerBounds();
    Map<Integer, ByteBuffer> upperBounds = metrics.upperBounds();
    if (null != lowerBound || null != upperBound) {
      // if there's an expected lower/upper bound, then the original type should be available
      assertThat(metrics.originalTypes().get(fieldId)).isEqualTo(type);
    }

    if (lowerBounds.containsKey(fieldId)) {
      assertThat((Object) fromByteBuffer(type, lowerBounds.get(fieldId))).isEqualTo(lowerBound);
    } else {
      assertThat(lowerBound).isNull();
    }

    if (upperBounds.containsKey(fieldId)) {
      assertThat((Object) fromByteBuffer(type, upperBounds.get(fieldId))).isEqualTo(upperBound);
    } else {
      assertThat(upperBound).isNull();
    }
  }
}
