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.db.rows;
  
  import java.util.*;
  
  import org.apache.cassandra.cache.IMeasurableMemory;
  import org.apache.cassandra.db.*;
  import org.apache.cassandra.utils.memory.ByteBufferCloner;
  
  /**
   * A marker for a range tombstone bound.
   * <p>
   * There is 2 types of markers: bounds (see {@link RangeTombstoneBoundMarker}) and boundaries (see {@link RangeTombstoneBoundaryMarker}).
   */
  public interface RangeTombstoneMarker extends Unfiltered, IMeasurableMemory
  {
      @Override
      public ClusteringBoundOrBoundary<?> clustering();
  
      public boolean isBoundary();
  
      public boolean isOpen(boolean reversed);
      public boolean isClose(boolean reversed);
  
      public DeletionTime openDeletionTime(boolean reversed);
      public DeletionTime closeDeletionTime(boolean reversed);
      public boolean openIsInclusive(boolean reversed);
      public boolean closeIsInclusive(boolean reversed);
  
      public ClusteringBound<?> openBound(boolean reversed);
      public ClusteringBound<?> closeBound(boolean reversed);
  
      public RangeTombstoneMarker clone(ByteBufferCloner cloner);
  
      default public boolean isEmpty()
      {
          // There is no such thing as an empty marker
          return false;
      }
  
      public RangeTombstoneMarker withNewOpeningDeletionTime(boolean reversed, DeletionTime newDeletionTime);
  
      /**
       * Utility class to help merging range tombstone markers coming from multiple inputs (UnfilteredRowIterators).
       * <p>
       * The assumption that each individual input must validate and that we must preserve in the output is that every
       * open marker has a corresponding close marker with the exact same deletion info, and that there is no other range
       * tombstone marker between those open and close marker (of course, they could be rows in between). In other word,
       * for any {@code UnfilteredRowIterator}, you only ever have to remenber the last open marker (if any) to have the
       * full picture of what is deleted by range tombstones at any given point of iterating that iterator.
       * <p>
       * Note that this class can merge both forward and reverse iterators. To deal with reverse, we just reverse how we
       * deal with open and close markers (in forward order, we'll get open-close, open-close, ..., while in reverse we'll
       * get close-open, close-open, ...).
       */
      public static class Merger
      {
          private final DeletionTime partitionDeletion;
          private final boolean reversed;
  
          private ClusteringBoundOrBoundary<?> bound;
          private final RangeTombstoneMarker[] markers;
  
          // For each iterator, what is the currently open marker deletion time (or null if there is no open marker on that iterator)
          private final DeletionTime[] openMarkers;
          // The index in openMarkers of the "biggest" marker, the one with the biggest deletion time. Is < 0 iff there is no open
          // marker on any iterator.
          private int biggestOpenMarker = -1;
  
          public Merger(int size, DeletionTime partitionDeletion, boolean reversed)
          {
              this.partitionDeletion = partitionDeletion;
              this.reversed = reversed;
  
              this.markers = new RangeTombstoneMarker[size];
              this.openMarkers = new DeletionTime[size];
          }
  
          public void clear()
          {
              Arrays.fill(markers, null);
          }
  
          public void add(int i, RangeTombstoneMarker marker)
          {
              bound = marker.clustering();
              markers[i] = marker;
          }
  
          public RangeTombstoneMarker merge()
          {
              /*
               * Merging of range tombstones works this way:
               *   1) We remember what is the currently open marker in the merged stream
               *   2) We update our internal states of what range is opened on the input streams based on the new markers to merge
               *   3) We compute what should be the state in the merge stream after 2)
               *   4) We return what marker should be issued on the merged stream based on the difference between the state from 1) and 3)
               */
  
              DeletionTime previousDeletionTimeInMerged = currentOpenDeletionTimeInMerged();
  
              updateOpenMarkers();
  
              DeletionTime newDeletionTimeInMerged = currentOpenDeletionTimeInMerged();
              if (previousDeletionTimeInMerged.equals(newDeletionTimeInMerged))
                  return null;
  
              boolean isBeforeClustering = bound.kind().comparedToClustering < 0;
              if (reversed)
                  isBeforeClustering = !isBeforeClustering;
  
              RangeTombstoneMarker merged;
              if (previousDeletionTimeInMerged.isLive())
              {
                  merged = isBeforeClustering
                         ? RangeTombstoneBoundMarker.inclusiveOpen(reversed, bound, newDeletionTimeInMerged)
                         : RangeTombstoneBoundMarker.exclusiveOpen(reversed, bound, newDeletionTimeInMerged);
              }
              else if (newDeletionTimeInMerged.isLive())
              {
                  merged = isBeforeClustering
                         ? RangeTombstoneBoundMarker.exclusiveClose(reversed, bound, previousDeletionTimeInMerged)
                         : RangeTombstoneBoundMarker.inclusiveClose(reversed, bound, previousDeletionTimeInMerged);
              }
              else
              {
                  merged = isBeforeClustering
                         ? RangeTombstoneBoundaryMarker.exclusiveCloseInclusiveOpen(reversed, bound, previousDeletionTimeInMerged, newDeletionTimeInMerged)
                         : RangeTombstoneBoundaryMarker.inclusiveCloseExclusiveOpen(reversed, bound, previousDeletionTimeInMerged, newDeletionTimeInMerged);
              }
  
              return merged;
          }
  
          public RangeTombstoneMarker[] mergedMarkers()
          {
              return markers;
          }
  
          private DeletionTime currentOpenDeletionTimeInMerged()
          {
              if (biggestOpenMarker < 0)
                  return DeletionTime.LIVE;
  
              DeletionTime biggestDeletionTime = openMarkers[biggestOpenMarker];
              // it's only open in the merged iterator if it doesn't supersedes the partition level deletion
              return !biggestDeletionTime.supersedes(partitionDeletion) ? DeletionTime.LIVE : biggestDeletionTime;
          }
  
          private void updateOpenMarkers()
          {
              for (int i = 0; i < markers.length; i++)
              {
                  RangeTombstoneMarker marker = markers[i];
                  if (marker == null)
                      continue;
  
                  // Note that we can have boundaries that are both open and close, but in that case all we care about
                  // is what it the open deletion after the marker, so we favor the opening part in this case.
                  if (marker.isOpen(reversed))
                      openMarkers[i] = marker.openDeletionTime(reversed);
                  else
                      openMarkers[i] = null;
              }
  
              // Recompute what is now the biggest open marker
              biggestOpenMarker = -1;
              for (int i = 0; i < openMarkers.length; i++)
              {
                  if (openMarkers[i] != null && (biggestOpenMarker < 0 || openMarkers[i].supersedes(openMarkers[biggestOpenMarker])))
                      biggestOpenMarker = i;
              }
          }
  
          public DeletionTime activeDeletion()
          {
              DeletionTime openMarker = currentOpenDeletionTimeInMerged();
              // We only have an open marker in the merged stream if it's not shadowed by the partition deletion (which can be LIVE itself), so
              // if have an open marker, we know it's the "active" deletion for the merged stream.
              return openMarker.isLive() ? partitionDeletion : openMarker;
          }
      }
  }
  