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.Comparator;
  import java.util.Iterator;
  
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.db.*;
  import org.apache.cassandra.db.filter.ColumnFilter;
  
  /**
   * An iterator that merges a source of rows with the range tombstone and partition level deletion of a give partition.
   * <p>
   * This is used by our {@code Partition} implementations to produce a {@code UnfilteredRowIterator} by merging the rows
   * and deletion infos that are kept separate. This has also 2 additional role:
   *   1) this make sure the row returned only includes the columns selected for the resulting iterator.
   *   2) this (optionally) remove any data that can be shadowed (see commet on 'removeShadowedData' below for more details)
   */
  public class RowAndDeletionMergeIterator extends AbstractUnfilteredRowIterator
  {
      // For some of our Partition implementation, we can't guarantee that the deletion information (partition level
      // deletion and range tombstones) don't shadow data in the rows. If that is the case, this class also take
      // cares of skipping such shadowed data (since it is the contract of an UnfilteredRowIterator that it doesn't
      // shadow its own data). Sometimes however, we know this can't happen, in which case we can skip that step.
      private final boolean removeShadowedData;
      private final Comparator<Clusterable> comparator;
      private final ColumnFilter selection;
  
      private final Iterator<Row> rows;
      private Row nextRow;
  
      private final Iterator<RangeTombstone> ranges;
      private RangeTombstone nextRange;
  
      // The currently open tombstone. Note that unless this is null, there is no point in checking nextRange.
      private RangeTombstone openRange;
  
      public RowAndDeletionMergeIterator(TableMetadata metadata,
                                         DecoratedKey partitionKey,
                                         DeletionTime partitionLevelDeletion,
                                         ColumnFilter selection,
                                         Row staticRow,
                                         boolean isReversed,
                                         EncodingStats stats,
                                         Iterator<Row> rows,
                                         Iterator<RangeTombstone> ranges,
                                         boolean removeShadowedData)
      {
          super(metadata, partitionKey, partitionLevelDeletion, selection.fetchedColumns(), staticRow, isReversed, stats);
          this.comparator = isReversed ? metadata.comparator.reversed() : metadata.comparator;
          this.selection = selection;
          this.removeShadowedData = removeShadowedData;
          this.rows = rows;
          this.ranges = ranges;
      }
  
      private Unfiltered computeNextInternal()
      {
          while (true)
          {
              updateNextRow();
              if (nextRow == null)
              {
                  if (openRange != null)
                      return closeOpenedRange();
  
                  updateNextRange();
                  return nextRange == null ? endOfData() : openRange();
              }
  
              // We have a next row
  
              if (openRange == null)
              {
                  // We have no currently open tombstone range. So check if we have a next range and if it sorts before this row.
                  // If it does, the opening of that range should go first. Otherwise, the row goes first.
                  updateNextRange();
                  if (nextRange != null && comparator.compare(openBound(nextRange), nextRow.clustering()) < 0)
                      return openRange();
  
                  Row row = consumeNextRow();
                  // it's possible for the row to be fully shadowed by the current range tombstone
                  if (row != null)
                      return row;
              }
              else
              {
                  // We have both a next row and a currently opened tombstone. Check which goes first between the range closing and the row.
                  if (comparator.compare(closeBound(openRange), nextRow.clustering()) < 0)
                      return closeOpenedRange();
  
                  Row row = consumeNextRow();
                  if (row != null)
                      return row;
              }
          }
      }
  
      /**
       * RangeTombstoneList doesn't correctly merge multiple superseded rts, or overlapping rts with the
       * same ts. This causes it to emit noop boundary markers which can cause unneeded read repairs and
       * repair over streaming. This should technically be fixed in RangeTombstoneList. However, fixing
       * it isn't trivial and that class is already so complicated that the fix would have a good chance
       * of adding a worse bug. So we just swallow the noop boundary markers here. See CASSANDRA-14894
       */
      private static boolean shouldSkip(Unfiltered unfiltered)
      {
          if (unfiltered == null || !unfiltered.isRangeTombstoneMarker())
              return false;
  
          RangeTombstoneMarker marker = (RangeTombstoneMarker) unfiltered;
  
          if (!marker.isBoundary())
              return false;
  
          DeletionTime open = marker.openDeletionTime(false);
          DeletionTime close = marker.closeDeletionTime(false);
  
          return open.equals(close);
  
      }
  
      @Override
      protected Unfiltered computeNext()
      {
          while (true)
          {
              Unfiltered next = computeNextInternal();
  
              if (shouldSkip(next))
                  continue;
  
              return next;
          }
      }
  
      private void updateNextRow()
      {
          if (nextRow == null && rows.hasNext())
              nextRow = rows.next();
      }
  
      private void updateNextRange()
      {
          while (nextRange == null && ranges.hasNext())
          {
              nextRange = ranges.next();
              // partition deletion will shadow range tombstone if partition deletion time is greater to equal to range
              // tombstone time.
              if ((removeShadowedData && !nextRange.deletionTime().supersedes(partitionLevelDeletion()))
                  || nextRange.deletedSlice().isEmpty(metadata.comparator))
                  nextRange = null;
          }
      }
  
      private Row consumeNextRow()
      {
          Row row = nextRow;
          nextRow = null;
          if (!removeShadowedData)
              return row.filter(selection, metadata());
  
          DeletionTime activeDeletion = openRange == null ? partitionLevelDeletion() : openRange.deletionTime();
          return row.filter(selection, activeDeletion, false, metadata());
      }
  
      private RangeTombstone consumeNextRange()
      {
          RangeTombstone range = nextRange;
          nextRange = null;
          return range;
      }
  
      private RangeTombstone consumeOpenRange()
      {
          RangeTombstone range = openRange;
          openRange = null;
          return range;
      }
  
      private ClusteringBound<?> openBound(RangeTombstone range)
      {
          return range.deletedSlice().open(isReverseOrder());
      }
  
      private ClusteringBound<?> closeBound(RangeTombstone range)
      {
          return range.deletedSlice().close(isReverseOrder());
      }
  
      private RangeTombstoneMarker closeOpenedRange()
      {
          // Check if that close if actually a boundary between markers
          updateNextRange();
          RangeTombstoneMarker marker;
          if (nextRange != null && comparator.compare(closeBound(openRange), openBound(nextRange)) == 0)
          {
              marker = RangeTombstoneBoundaryMarker.makeBoundary(isReverseOrder(), closeBound(openRange), openBound(nextRange), openRange.deletionTime(), nextRange.deletionTime());
              openRange = consumeNextRange();
          }
          else
          {
              RangeTombstone toClose = consumeOpenRange();
              marker = new RangeTombstoneBoundMarker(closeBound(toClose), toClose.deletionTime());
          }
          return marker;
      }
  
      private RangeTombstoneMarker openRange()
      {
          assert openRange == null && nextRange != null;
          openRange = consumeNextRange();
          return new RangeTombstoneBoundMarker(openBound(openRange), openRange.deletionTime());
      }
  
  }
  