/*
 * 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 java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Arrays;
import org.apache.iceberg.avro.SupportsIndexProjection;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ByteBuffers;

/** Mutable {@link StructLike} implementation of {@link DeletionVector}. */
class DeletionVectorStruct extends SupportsIndexProjection implements DeletionVector, Serializable {
  private static final Types.StructType BASE_TYPE =
      Types.StructType.of(
          DeletionVector.LOCATION,
          DeletionVector.OFFSET,
          DeletionVector.SIZE_IN_BYTES,
          DeletionVector.CARDINALITY,
          DeletionVector.KEY_METADATA);

  private String location = null;
  private long offset = -1L;
  private long sizeInBytes = -1L;
  private long cardinality = -1L;
  private byte[] keyMetadata = null;

  DeletionVectorStruct(Types.StructType type) {
    super(BASE_TYPE, type);
  }

  private DeletionVectorStruct(DeletionVectorStruct toCopy) {
    super(toCopy);
    this.location = toCopy.location;
    this.offset = toCopy.offset;
    this.sizeInBytes = toCopy.sizeInBytes;
    this.cardinality = toCopy.cardinality;
    this.keyMetadata =
        toCopy.keyMetadata != null
            ? Arrays.copyOf(toCopy.keyMetadata, toCopy.keyMetadata.length)
            : null;
  }

  private DeletionVectorStruct(
      String location, long offset, long sizeInBytes, long cardinality, ByteBuffer keyMetadata) {
    super(BASE_TYPE.fields().size());
    this.location = location;
    this.offset = offset;
    this.sizeInBytes = sizeInBytes;
    this.cardinality = cardinality;
    this.keyMetadata = ByteBuffers.toByteArray(keyMetadata);
  }

  @Override
  public String location() {
    return location;
  }

  // Package-private only so the manifest reader can store the location resolved against the
  // table location; other callers must go through construction.
  void setLocation(String newLocation) {
    this.location = newLocation;
  }

  @Override
  public long offset() {
    return offset;
  }

  @Override
  public long sizeInBytes() {
    return sizeInBytes;
  }

  @Override
  public long cardinality() {
    return cardinality;
  }

  @Override
  public ByteBuffer keyMetadata() {
    return keyMetadata != null ? ByteBuffer.wrap(keyMetadata) : null;
  }

  @Override
  public DeletionVectorStruct copy() {
    return new DeletionVectorStruct(this);
  }

  @Override
  protected <T> T internalGet(int pos, Class<T> javaClass) {
    return javaClass.cast(getByPos(pos));
  }

  private Object getByPos(int pos) {
    return switch (pos) {
      case 0 -> location;
      case 1 -> offset;
      case 2 -> sizeInBytes;
      case 3 -> cardinality;
      case 4 -> keyMetadata();
      default -> throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
    };
  }

  @Override
  protected <T> void internalSet(int pos, T value) {
    switch (pos) {
        // always coerce to String for Serializable
      case 0 -> this.location = value.toString();
      case 1 -> this.offset = (Long) value;
      case 2 -> this.sizeInBytes = (Long) value;
      case 3 -> this.cardinality = (Long) value;
      case 4 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer) value);
      default -> {
        // ignore the object, it must be from a newer version of the format
      }
    }
  }

  static Builder builder() {
    return new Builder();
  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
        .add("location", location)
        .add("offset", offset)
        .add("size_in_bytes", sizeInBytes)
        .add("cardinality", cardinality)
        .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
        .toString();
  }

  static class Builder {
    private String location = null;
    private Long offset = null;
    private Long sizeInBytes = null;
    private Long cardinality = null;
    private ByteBuffer keyMetadata = null;

    Builder location(String dvLocation) {
      Preconditions.checkArgument(dvLocation != null, "Invalid location: null");
      this.location = dvLocation;
      return this;
    }

    Builder offset(long dvOffset) {
      Preconditions.checkArgument(dvOffset >= 0, "Invalid offset: %s (must be >= 0)", dvOffset);
      this.offset = dvOffset;
      return this;
    }

    Builder sizeInBytes(long dvSizeInBytes) {
      Preconditions.checkArgument(
          dvSizeInBytes >= 0, "Invalid size in bytes: %s (must be >= 0)", dvSizeInBytes);
      this.sizeInBytes = dvSizeInBytes;
      return this;
    }

    Builder cardinality(long dvCardinality) {
      Preconditions.checkArgument(
          dvCardinality >= 0, "Invalid cardinality: %s (must be >= 0)", dvCardinality);
      this.cardinality = dvCardinality;
      return this;
    }

    Builder keyMetadata(ByteBuffer dvKeyMetadata) {
      this.keyMetadata = dvKeyMetadata;
      return this;
    }

    DeletionVectorStruct build() {
      Preconditions.checkArgument(location != null, "Missing required value: location");
      Preconditions.checkArgument(offset != null, "Missing required value: offset");
      Preconditions.checkArgument(sizeInBytes != null, "Missing required value: size in bytes");
      Preconditions.checkArgument(cardinality != null, "Missing required value: cardinality");
      return new DeletionVectorStruct(location, offset, sizeInBytes, cardinality, keyMetadata);
    }
  }
}
