/*
 * 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.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Test;

class TestDeletionVectorStruct {

  private static final ByteBuffer KEY_METADATA = ByteBuffer.wrap(new byte[] {1, 2, 3});

  @Test
  void fieldAccess() {
    DeletionVectorStruct dv =
        DeletionVectorStruct.builder()
            .location("s3://bucket/data/dv.puffin")
            .offset(256L)
            .sizeInBytes(128L)
            .cardinality(42L)
            .keyMetadata(KEY_METADATA)
            .build();

    assertThat(dv.location()).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(dv.offset()).isEqualTo(256L);
    assertThat(dv.sizeInBytes()).isEqualTo(128L);
    assertThat(dv.cardinality()).isEqualTo(42L);
    assertThat(dv.keyMetadata()).isEqualTo(KEY_METADATA);
  }

  @Test
  void copy() {
    DeletionVectorStruct dv =
        DeletionVectorStruct.builder()
            .location("s3://bucket/data/dv.puffin")
            .offset(256L)
            .sizeInBytes(128L)
            .cardinality(42L)
            .keyMetadata(KEY_METADATA)
            .build();

    DeletionVectorStruct copy = dv.copy();

    assertThat(copy.location()).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(copy.offset()).isEqualTo(256L);
    assertThat(copy.sizeInBytes()).isEqualTo(128L);
    assertThat(copy.cardinality()).isEqualTo(42L);
    assertThat(copy.keyMetadata()).isEqualTo(KEY_METADATA);
  }

  @Test
  void size() {
    DeletionVectorStruct dv = new DeletionVectorStruct(DeletionVector.schema());
    assertThat(dv.size()).isEqualTo(5);
  }

  @Test
  void projectedStructLike() {
    // project only location (field ID 155) and cardinality (field ID 156)
    Types.StructType projection =
        Types.StructType.of(DeletionVector.LOCATION, DeletionVector.CARDINALITY);

    DeletionVectorStruct dv = new DeletionVectorStruct(projection);
    assertThat(dv.size()).isEqualTo(2);

    // projected position 0 maps to internal position 0 (location)
    // projected position 1 maps to internal position 3 (cardinality)
    dv.set(0, "s3://bucket/data/dv.puffin");
    dv.set(1, 42L);

    assertThat(dv.location()).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(dv.cardinality()).isEqualTo(42L);
    assertThat(dv.get(0, String.class)).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(dv.get(1, Long.class)).isEqualTo(42L);
  }

  @Test
  void internalSetIgnoresUnknownOrdinal() {
    DeletionVectorStruct dv =
        DeletionVectorStruct.builder()
            .location("s3://bucket/data/dv.puffin")
            .offset(100L)
            .sizeInBytes(512L)
            .cardinality(42L)
            .keyMetadata(KEY_METADATA)
            .build();

    // unknown ordinals from a newer format version are silently ignored
    dv.internalSet(99, "value from a newer format");

    // every field is unchanged
    assertThat(dv.location()).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(dv.offset()).isEqualTo(100L);
    assertThat(dv.sizeInBytes()).isEqualTo(512L);
    assertThat(dv.cardinality()).isEqualTo(42L);
    assertThat(dv.keyMetadata()).isEqualTo(KEY_METADATA);
  }

  @Test
  void javaSerializationRoundTrip() throws IOException, ClassNotFoundException {
    DeletionVectorStruct dv =
        DeletionVectorStruct.builder()
            .location("s3://bucket/data/dv.puffin")
            .offset(256L)
            .sizeInBytes(128L)
            .cardinality(42L)
            .keyMetadata(KEY_METADATA)
            .build();

    DeletionVectorStruct deserialized = TestHelpers.roundTripSerialize(dv);

    assertThat(deserialized.location()).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(deserialized.offset()).isEqualTo(256L);
    assertThat(deserialized.sizeInBytes()).isEqualTo(128L);
    assertThat(deserialized.cardinality()).isEqualTo(42L);
    assertThat(deserialized.keyMetadata()).isEqualTo(KEY_METADATA);
  }

  @Test
  void builderMissingRequiredFields() {
    assertThatThrownBy(
            () -> DeletionVectorStruct.builder().offset(0).sizeInBytes(1).cardinality(1).build())
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Missing required value: location");

    assertThatThrownBy(
            () ->
                DeletionVectorStruct.builder()
                    .location("s3://bucket/dv.puffin")
                    .sizeInBytes(1)
                    .cardinality(1)
                    .build())
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Missing required value: offset");

    assertThatThrownBy(
            () ->
                DeletionVectorStruct.builder()
                    .location("s3://bucket/dv.puffin")
                    .offset(0)
                    .cardinality(1)
                    .build())
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Missing required value: size in bytes");

    assertThatThrownBy(
            () ->
                DeletionVectorStruct.builder()
                    .location("s3://bucket/dv.puffin")
                    .offset(0)
                    .sizeInBytes(1)
                    .build())
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Missing required value: cardinality");
  }

  @Test
  void builderRejectsInvalidValuesAtSetter() {
    assertThatThrownBy(() -> DeletionVectorStruct.builder().location(null))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Invalid location: null");

    assertThatThrownBy(() -> DeletionVectorStruct.builder().offset(-1))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Invalid offset: -1 (must be >= 0)");

    assertThatThrownBy(() -> DeletionVectorStruct.builder().sizeInBytes(-1))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Invalid size in bytes: -1 (must be >= 0)");

    assertThatThrownBy(() -> DeletionVectorStruct.builder().cardinality(-1))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessage("Invalid cardinality: -1 (must be >= 0)");
  }

  @Test
  void kryoSerializationRoundTrip() throws IOException {
    DeletionVectorStruct dv =
        DeletionVectorStruct.builder()
            .location("s3://bucket/data/dv.puffin")
            .offset(256L)
            .sizeInBytes(128L)
            .cardinality(42L)
            .keyMetadata(KEY_METADATA)
            .build();

    DeletionVectorStruct deserialized = TestHelpers.KryoHelpers.roundTripSerialize(dv);

    assertThat(deserialized.location()).isEqualTo("s3://bucket/data/dv.puffin");
    assertThat(deserialized.offset()).isEqualTo(256L);
    assertThat(deserialized.sizeInBytes()).isEqualTo(128L);
    assertThat(deserialized.cardinality()).isEqualTo(42L);
    assertThat(deserialized.keyMetadata()).isEqualTo(KEY_METADATA);
  }
}
