content: /*
   * 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.cassandra.service.paxos;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.*;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.collect.ImmutableList;
  
  import org.apache.cassandra.db.TypeSizes;
  import org.apache.cassandra.db.marshal.BytesType;
  import org.apache.cassandra.db.marshal.TupleType;
  import org.apache.cassandra.dht.IPartitioner;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.io.IVersionedSerializer;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.schema.Schema;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.utils.FBUtilities;
  
  import static java.lang.Math.min;
  import static org.apache.cassandra.service.paxos.Commit.isAfter;
  import static org.apache.cassandra.service.paxos.Commit.latest;
  
  public class PaxosRepairHistory
  {
      private static final TupleType TYPE = new TupleType(ImmutableList.of(BytesType.instance, BytesType.instance));
  
      /**
       * The following two fields represent the mapping of ranges to ballot lower bounds, for example:
       *
       *   ballotLowBound           = [ none(), b2, none(), b4, none() ]
       *   tokenInclusiveUpperBound = [ t1, t2, t3, t4 ]
       *
       * Correspond to the following token bounds:
       *
       *   (MIN_VALUE, t1] => none()
       *   (t1, t2]        => b2
       *   (t2, t3]        => none()
       *   (t3, t4]        => b4
       *   (t4, MAX_VALUE) => none()
       */
  
      public final IPartitioner partitioner;
      private final Token[] tokenInclusiveUpperBound;
      private final Ballot[] ballotLowBound; // always one longer to capture values up to "MAX_VALUE" (which in some cases doesn't exist, as is infinite)
  
      private static IPartitioner partitioner(String keyspace, String table)
      {
          TableMetadata tm = Schema.instance.getTableMetadata(keyspace, table);
          return tm != null ? tm.partitioner : IPartitioner.global();
      }
  
      @VisibleForTesting
      public static PaxosRepairHistory empty()
      {
          return empty(IPartitioner.global());
      }
  
      public static PaxosRepairHistory empty(String keyspace, String table)
      {
          return empty(partitioner(keyspace, table));
      }
  
      public static PaxosRepairHistory empty(IPartitioner partitioner)
      {
          return new PaxosRepairHistory(partitioner, new Token[0], new Ballot[] { Ballot.none() } );
      }
  
      PaxosRepairHistory(IPartitioner partitioner,
                         Token[] tokenInclusiveUpperBound, Ballot[] ballotLowBound)
      {
          assert ballotLowBound.length == tokenInclusiveUpperBound.length + 1;
          this.partitioner = partitioner;
          this.tokenInclusiveUpperBound = tokenInclusiveUpperBound;
          this.ballotLowBound = ballotLowBound;
      }
  
      public Ballot maxLowBound()
      {
          Ballot maxBallot = Ballot.none();
          for (Ballot lowBound : ballotLowBound)
          {
              maxBallot = Commit.latest(maxBallot, lowBound);
          }
          return maxBallot;
      }
  
      public String toString()
      {
          return "PaxosRepairHistory{" +
                  IntStream.range(0, ballotLowBound.length)
                          .filter(i -> !Ballot.none().equals(ballotLowBound[i]))
                          .mapToObj(i -> range(i) + "=" + ballotLowBound[i])
                          .collect(Collectors.joining(", ")) + '}';
      }
  
      public boolean equals(Object o)
      {
          if (this == o) return true;
          if (o == null || getClass() != o.getClass()) return false;
          PaxosRepairHistory that = (PaxosRepairHistory) o;
          return Arrays.equals(ballotLowBound, that.ballotLowBound) && Arrays.equals(tokenInclusiveUpperBound, that.tokenInclusiveUpperBound);
      }
  
      public int hashCode()
      {
          return Arrays.hashCode(ballotLowBound);
      }
  
      public Ballot ballotForToken(Token token)
      {
          int idx = Arrays.binarySearch(tokenInclusiveUpperBound, token);
          if (idx < 0) idx = -1 - idx;
          return ballotLowBound[idx];
      }
  
      private Ballot ballotForIndex(int idx)
      {
          if (idx < 0 || idx > size())
              throw new IndexOutOfBoundsException();
  
          return ballotLowBound[idx];
      }
  
      private int indexForToken(Token token)
      {
          int idx = Arrays.binarySearch(tokenInclusiveUpperBound, token);
          if (idx < 0) idx = -1 - idx;
          return idx;
      }
  
      private boolean contains(int idx, Token token)
      {
          if (idx < 0 || idx > size())
              throw new IndexOutOfBoundsException();
  
          return  (idx == 0      || tokenInclusiveUpperBound[idx - 1].compareTo(token) <  0)
               && (idx == size() || tokenInclusiveUpperBound[idx    ].compareTo(token) >= 0);
      }
  
      public int size()
      {
          return tokenInclusiveUpperBound.length;
      }
  
      private RangeIterator rangeIterator()
      {
          return new RangeIterator();
      }
  
      private Range<Token> range(int i)
      {
          return new Range<>(tokenExclusiveLowerBound(i), tokenInclusiveUpperBound(i));
      }
  
      public Searcher searcher()
      {
          return new Searcher();
      }
  
      private Token tokenExclusiveLowerBound(int i)
      {
          return i == 0 ? partitioner.getMinimumToken() : tokenInclusiveUpperBound[i - 1];
      }
  
      private Token tokenInclusiveUpperBound(int i)
      {
          return i == tokenInclusiveUpperBound.length ? partitioner.getMinimumToken() : tokenInclusiveUpperBound[i];
      }
  
      public List<ByteBuffer> toTupleBufferList()
      {
          List<ByteBuffer> tuples = new ArrayList<>(size() + 1);
          for (int i = 0 ; i < 1 + size() ; ++i)
              tuples.add(TYPE.pack(partitioner.getTokenFactory().toByteArray(tokenInclusiveUpperBound(i)), ballotLowBound[i].toBytes()));
          return tuples;
      }
  
      public static PaxosRepairHistory fromTupleBufferList(String keyspace, String table, List<ByteBuffer> tuples)
      {
          return fromTupleBufferList(partitioner(keyspace, table), tuples);
      }
  
      public static PaxosRepairHistory fromTupleBufferList(IPartitioner partitioner, List<ByteBuffer> tuples)
      {
          Token[] tokenInclusiveUpperBounds = new Token[tuples.size() - 1];
          Ballot[] ballotLowBounds = new Ballot[tuples.size()];
          for (int i = 0 ; i < tuples.size() ; ++i)
          {
              List<ByteBuffer> elements = TYPE.unpack(tuples.get(i));
              if (i < tokenInclusiveUpperBounds.length)
                  tokenInclusiveUpperBounds[i] = partitioner.getTokenFactory().fromByteArray(elements.get(0));
              ballotLowBounds[i] = Ballot.deserialize(elements.get(1));
          }
  
          return new PaxosRepairHistory(partitioner, tokenInclusiveUpperBounds, ballotLowBounds);
      }
  
      // append the item to the given list, modifying the underlying list
      // if the item makes previoud entries redundant
  
      public static PaxosRepairHistory merge(PaxosRepairHistory historyLeft, PaxosRepairHistory historyRight)
      {
          if (historyLeft == null)
              return historyRight;
  
          if (historyRight == null)
              return historyLeft;
  
          assert historyLeft.partitioner == historyRight.partitioner : String.format("Mismatching partitioners (%s != %s)",
                                                                                     historyLeft.partitioner,
                                                                                     historyRight.partitioner);
  
          Builder builder = new Builder(historyLeft.partitioner, historyLeft.size() + historyRight.size());
  
          RangeIterator left = historyLeft.rangeIterator();
          RangeIterator right = historyRight.rangeIterator();
          while (left.hasUpperBound() && right.hasUpperBound())
          {
              int cmp = left.tokenInclusiveUpperBound().compareTo(right.tokenInclusiveUpperBound());
  
              Ballot ballot = latest(left.ballotLowBound(), right.ballotLowBound());
              if (cmp == 0)
              {
                  builder.append(left.tokenInclusiveUpperBound(), ballot);
                  left.next();
                  right.next();
              }
              else
              {
                  RangeIterator firstIter = cmp < 0 ? left : right;
                  builder.append(firstIter.tokenInclusiveUpperBound(), ballot);
                  firstIter.next();
              }
          }
  
          while (left.hasUpperBound())
          {
              builder.append(left.tokenInclusiveUpperBound(), latest(left.ballotLowBound(), right.ballotLowBound()));
              left.next();
          }
  
          while (right.hasUpperBound())
          {
              builder.append(right.tokenInclusiveUpperBound(), latest(left.ballotLowBound(), right.ballotLowBound()));
              right.next();
          }
  
          builder.appendLast(latest(left.ballotLowBound(), right.ballotLowBound()));
          return builder.build();
      }
  
      @VisibleForTesting
      public static PaxosRepairHistory add(PaxosRepairHistory existing, Collection<Range<Token>> ranges, Ballot ballot)
      {
          ranges = Range.normalize(ranges);
          Builder builder = new Builder(existing.partitioner, ranges.size() * 2);
          for (Range<Token> range : ranges)
          {
              // don't add a point for an opening min token, since it
              // effectively leaves the bottom of the range unbounded
              builder.appendMaybeMin(range.left, Ballot.none());
              builder.appendMaybeMax(range.right, ballot);
          }
  
          return merge(existing, builder.build());
      }
  
      /**
       * returns a copy of this PaxosRepairHistory limited to the ranges supplied, with all other ranges reporting Ballot.none()
       */
      @VisibleForTesting
      static PaxosRepairHistory trim(PaxosRepairHistory existing, Collection<Range<Token>> ranges)
      {
          Builder builder = new Builder(existing.partitioner, existing.size());
  
          ranges = Range.normalize(ranges);
          for (Range<Token> select : ranges)
          {
              RangeIterator intersects = existing.intersects(select);
              while (intersects.hasNext())
              {
                  if (Ballot.none().equals(intersects.ballotLowBound()))
                  {
                      intersects.next();
                      continue;
                  }
  
                  Token exclusiveLowerBound = maxExclusiveLowerBound(select.left, intersects.tokenExclusiveLowerBound());
                  Token inclusiveUpperBound = minInclusiveUpperBound(select.right, intersects.tokenInclusiveUpperBound());
                  assert exclusiveLowerBound.compareTo(inclusiveUpperBound) < 0 || inclusiveUpperBound.isMinimum();
  
                  builder.appendMaybeMin(exclusiveLowerBound, Ballot.none());
                  builder.appendMaybeMax(inclusiveUpperBound, intersects.ballotLowBound());
                  intersects.next();
              }
          }
  
          return builder.build();
      }
  
      RangeIterator intersects(Range<Token> unwrapped)
      {
          int from = Arrays.binarySearch(tokenInclusiveUpperBound, unwrapped.left);
          if (from < 0) from = -1 - from; else ++from;
          int to = unwrapped.right.isMinimum() ? ballotLowBound.length - 1 : Arrays.binarySearch(tokenInclusiveUpperBound, unwrapped.right);
          if (to < 0) to = -1 - to;
          return new RangeIterator(from, min(1 + to, ballotLowBound.length));
      }
  
      private static Token maxExclusiveLowerBound(Token a, Token b)
      {
          return a.compareTo(b) < 0 ? b : a;
      }
  
      private static Token minInclusiveUpperBound(Token a, Token b)
      {
          if (!a.isMinimum() && !b.isMinimum()) return a.compareTo(b) <= 0 ? a : b;
          else if (!a.isMinimum()) return a;
          else if (!b.isMinimum()) return b;
          else return a;
      }
  
      public static final IVersionedSerializer<PaxosRepairHistory> serializer = new IVersionedSerializer<>()
      {
          public void serialize(PaxosRepairHistory history, DataOutputPlus out, int version) throws IOException
          {
              if (version >= MessagingService.VERSION_51)
                  out.writeUTF(history.partitioner.getClass().getCanonicalName());
              out.writeUnsignedVInt32(history.size());
              for (int i = 0; i < history.size() ; ++i)
              {
                  Token.serializer.serialize(history.tokenInclusiveUpperBound[i], out, version);
                  history.ballotLowBound[i].serialize(out);
              }
              history.ballotLowBound[history.size()].serialize(out);
          }
  
          public PaxosRepairHistory deserialize(DataInputPlus in, int version) throws IOException
          {
              IPartitioner partitioner = version >= MessagingService.VERSION_51
                                         ? FBUtilities.newPartitioner(in.readUTF())
                                         : IPartitioner.global();
              int size = in.readUnsignedVInt32();
              Token[] tokenInclusiveUpperBounds = new Token[size];
              Ballot[] ballotLowBounds = new Ballot[size + 1];
              for (int i = 0; i < size; i++)
              {
                  tokenInclusiveUpperBounds[i] = Token.serializer.deserialize(in, partitioner, version);
                  ballotLowBounds[i] = Ballot.deserialize(in);
              }
              ballotLowBounds[size] = Ballot.deserialize(in);
              return new PaxosRepairHistory(partitioner, tokenInclusiveUpperBounds, ballotLowBounds);
          }
  
          public long serializedSize(PaxosRepairHistory history, int version)
          {
              long size = 0;
              if (version >= MessagingService.VERSION_51)
                  size += TypeSizes.sizeof(history.partitioner.getClass().getCanonicalName());
              size += TypeSizes.sizeofUnsignedVInt(history.size());
              for (int i = 0; i < history.size() ; ++i)
              {
                  size += Token.serializer.serializedSize(history.tokenInclusiveUpperBound[i], version);
                  size += Ballot.sizeInBytes();
              }
              size += Ballot.sizeInBytes();
              return size;
          }
      };
  
      public class Searcher
      {
          int idx = -1;
  
          public Ballot ballotForToken(Token token)
          {
              if (idx < 0 || !contains(idx, token))
                  idx = indexForToken(token);
              return ballotForIndex(idx);
          }
      }
  
      class RangeIterator
      {
          final int end;
          int i;
  
          RangeIterator()
          {
              this.end = ballotLowBound.length;
          }
  
          RangeIterator(int from, int to)
          {
              this.i = from;
              this.end = to;
          }
  
          boolean hasNext()
          {
              return i < end;
          }
  
          boolean hasUpperBound()
          {
              return i < tokenInclusiveUpperBound.length;
          }
  
          void next()
          {
              ++i;
          }
  
          Token tokenExclusiveLowerBound()
          {
              return PaxosRepairHistory.this.tokenExclusiveLowerBound(i);
          }
  
          Token tokenInclusiveUpperBound()
          {
              return PaxosRepairHistory.this.tokenInclusiveUpperBound(i);
          }
  
          Ballot ballotLowBound()
          {
              return ballotLowBound[i];
          }
      }
  
      static class Builder
      {
          final IPartitioner partitioner;
          final List<Token> tokenInclusiveUpperBounds;
          final List<Ballot> ballotLowBounds;
  
          Builder(IPartitioner partitioner, int capacity)
          {
              this.partitioner = partitioner;
              this.tokenInclusiveUpperBounds = new ArrayList<>(capacity);
              this.ballotLowBounds = new ArrayList<>(capacity + 1);
          }
  
          void appendMaybeMin(Token inclusiveLowBound, Ballot ballotLowBound)
          {
              if (inclusiveLowBound.isMinimum())
                  assert ballotLowBound.equals(Ballot.none()) && ballotLowBounds.isEmpty();
              else
                  append(inclusiveLowBound, ballotLowBound);
          }
  
          void appendMaybeMax(Token inclusiveLowBound, Ballot ballotLowBound)
          {
              if (inclusiveLowBound.isMinimum())
                  appendLast(ballotLowBound);
              else
                  append(inclusiveLowBound, ballotLowBound);
          }
  
          void append(Token inclusiveLowBound, Ballot ballotLowBound)
          {
              int tailIdx = tokenInclusiveUpperBounds.size() - 1;
  
              assert tokenInclusiveUpperBounds.size() == ballotLowBounds.size();
              assert tailIdx < 0 || inclusiveLowBound.compareTo(tokenInclusiveUpperBounds.get(tailIdx)) >= 0;
  
              boolean sameAsTailToken = tailIdx >= 0 && inclusiveLowBound.equals(tokenInclusiveUpperBounds.get(tailIdx));
              boolean sameAsTailBallot = tailIdx >= 0 && ballotLowBound.equals(ballotLowBounds.get(tailIdx));
              if (sameAsTailToken || sameAsTailBallot)
              {
                  if (sameAsTailBallot)
                      tokenInclusiveUpperBounds.set(tailIdx, inclusiveLowBound);
                  else if (isAfter(ballotLowBound, ballotLowBounds.get(tailIdx)))
                      ballotLowBounds.set(tailIdx, ballotLowBound);
              }
              else
              {
                  tokenInclusiveUpperBounds.add(inclusiveLowBound);
                  ballotLowBounds.add(ballotLowBound);
              }
          }
  
          void appendLast(Ballot ballotLowBound)
          {
              assert ballotLowBounds.size() == tokenInclusiveUpperBounds.size();
              int tailIdx = tokenInclusiveUpperBounds.size() - 1;
              if (!ballotLowBounds.isEmpty() && ballotLowBound.equals(ballotLowBounds.get(tailIdx)))
                  tokenInclusiveUpperBounds.remove(tailIdx);
              else
                  ballotLowBounds.add(ballotLowBound);
          }
  
          PaxosRepairHistory build()
          {
              if (tokenInclusiveUpperBounds.size() == ballotLowBounds.size())
                  ballotLowBounds.add(Ballot.none());
              return new PaxosRepairHistory(partitioner,
                                            tokenInclusiveUpperBounds.toArray(new Token[0]),
                                            ballotLowBounds.toArray(new Ballot[0]));
          }
      }
  }
  