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.transform;
  
  import java.util.Collections;
  import java.util.List;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.*;
  import org.apache.cassandra.db.compaction.OperationType;
  import org.apache.cassandra.db.partitions.PartitionIterator;
  import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
  import org.apache.cassandra.db.rows.*;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.utils.DiagnosticSnapshotService;
  import org.apache.cassandra.utils.FBUtilities;
  
  public class DuplicateRowChecker extends Transformation<BaseRowIterator<?>>
  {
      private static final Logger logger = LoggerFactory.getLogger(DuplicateRowChecker.class);
  
      Clustering<?> previous = null;
      int duplicatesDetected = 0;
      boolean hadNonEqualDuplicates = false;
  
      final String stage;
      final List<InetAddressAndPort> replicas;
      final TableMetadata metadata;
      final DecoratedKey key;
      final boolean snapshotOnDuplicate;
  
      DuplicateRowChecker(final DecoratedKey key,
                          final TableMetadata metadata,
                          final String stage,
                          final boolean snapshotOnDuplicate,
                          final List<InetAddressAndPort> replicas)
      {
          this.key = key;
          this.metadata = metadata;
          this.stage = stage;
          this.snapshotOnDuplicate = snapshotOnDuplicate;
          this.replicas = replicas;
      }
  
      protected DeletionTime applyToDeletion(DeletionTime deletionTime)
      {
          return deletionTime;
      }
  
      protected RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
      {
          return marker;
      }
  
      protected Row applyToStatic(Row row)
      {
          return row;
      }
  
      protected Row applyToRow(Row row)
      {
          if (null != previous && metadata.comparator.compare(row.clustering(), previous) == 0)
          {
              duplicatesDetected++;
              hadNonEqualDuplicates |= !row.clustering().equals(previous);
          }
  
          previous = row.clustering();
          return row;
      }
  
      protected void onPartitionClose()
      {
          if (duplicatesDetected > 0)
          {
              logger.warn("Detected {} duplicate rows for {} during {}.{}",
                          duplicatesDetected,
                          metadata.partitionKeyType.getString(key.getKey()),
                          stage,
                          hadNonEqualDuplicates ? " Some duplicates had different byte representation." : "");
              if (snapshotOnDuplicate)
                  DiagnosticSnapshotService.duplicateRows(metadata, replicas);
          }
          duplicatesDetected = 0;
          previous = null;
          super.onPartitionClose();
      }
  
      public static UnfilteredPartitionIterator duringCompaction(final UnfilteredPartitionIterator iterator, OperationType type)
      {
          if (!DatabaseDescriptor.checkForDuplicateRowsDuringCompaction())
              return iterator;
          final List<InetAddressAndPort> address = Collections.singletonList(FBUtilities.getBroadcastAddressAndPort());
          final boolean snapshot = DatabaseDescriptor.snapshotOnDuplicateRowDetection();
          return Transformation.apply(iterator, new Transformation<UnfilteredRowIterator>()
          {
              protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
              {
                  return Transformation.apply(partition, new DuplicateRowChecker(partition.partitionKey(),
                                                                                 partition.metadata(),
                                                                                 type.toString(),
                                                                                 snapshot,
                                                                                 address));
              }
          });
      }
  
      public static PartitionIterator duringRead(final PartitionIterator iterator, final List<InetAddressAndPort> replicas)
      {
          if (!DatabaseDescriptor.checkForDuplicateRowsDuringReads())
              return iterator;
          final boolean snapshot = DatabaseDescriptor.snapshotOnDuplicateRowDetection();
          return Transformation.apply(iterator, new Transformation<RowIterator>()
          {
              protected RowIterator applyToPartition(RowIterator partition)
              {
                  return Transformation.apply(partition, new DuplicateRowChecker(partition.partitionKey(),
                                                                                 partition.metadata(),
                                                                                 "Read",
                                                                                 snapshot,
                                                                                 replicas));
              }
          });
      }
  }
  