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.index.sasi.conf;
  
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.Set;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.stream.Collectors;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.db.marshal.AbstractType;
  import org.apache.cassandra.index.sasi.SSTableIndex;
  import org.apache.cassandra.index.sasi.conf.view.View;
  import org.apache.cassandra.io.sstable.format.SSTableReader;
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.utils.Pair;
  
  /** a pared-down version of DataTracker and DT.View. need one for each index of each column family */
  public class DataTracker
  {
      private static final Logger logger = LoggerFactory.getLogger(DataTracker.class);
  
      private final AbstractType<?> keyValidator;
      private final ColumnIndex columnIndex;
      private final AtomicReference<View> view = new AtomicReference<>();
  
      public DataTracker(AbstractType<?> keyValidator, ColumnIndex index)
      {
          this.keyValidator = keyValidator;
          this.columnIndex = index;
          this.view.set(new View(index, Collections.<SSTableIndex>emptySet()));
      }
  
      public View getView()
      {
          return view.get();
      }
  
      /**
       * Replaces old SSTables with new by creating new immutable tracker.
       *
       * @param oldSSTables A set of SSTables to remove.
       * @param newSSTables A set of SSTables to add to tracker.
       *
       * @return A collection of SSTables which don't have component attached for current index.
       */
      public Iterable<SSTableReader> update(Collection<SSTableReader> oldSSTables, Collection<SSTableReader> newSSTables)
      {
          final Pair<Set<SSTableIndex>, Set<SSTableReader>> built = getBuiltIndexes(newSSTables);
          final Set<SSTableIndex> newIndexes = built.left;
          final Set<SSTableReader> indexedSSTables = built.right;
  
          View currentView, newView;
          do
          {
              currentView = view.get();
              newView = new View(columnIndex, currentView.getIndexes(), oldSSTables, newIndexes);
          }
          while (!view.compareAndSet(currentView, newView));
  
          for (SSTableReader sstable : indexedSSTables)
          {
                  sstable.addComponents(Collections.singleton(columnIndex.getComponent()));
          }
  
          return newSSTables.stream().filter(sstable -> !indexedSSTables.contains(sstable)).collect(Collectors.toList());
      }
  
      public boolean hasSSTable(SSTableReader sstable)
      {
          View currentView = view.get();
          for (SSTableIndex index : currentView)
          {
              if (index.getSSTable().equals(sstable))
                  return true;
          }
  
          return false;
      }
  
      public void dropData(Collection<SSTableReader> sstablesToRebuild)
      {
          View currentView = view.get();
          if (currentView == null)
              return;
  
          Set<SSTableReader> toRemove = new HashSet<>(sstablesToRebuild);
          for (SSTableIndex index : currentView)
          {
              SSTableReader sstable = index.getSSTable();
              if (!sstablesToRebuild.contains(sstable))
                  continue;
  
              index.markObsolete();
          }
  
          update(toRemove, Collections.<SSTableReader>emptyList());
      }
  
      public void dropData(long truncateUntil)
      {
          View currentView = view.get();
          if (currentView == null)
              return;
  
          Set<SSTableReader> toRemove = new HashSet<>();
          for (SSTableIndex index : currentView)
          {
              SSTableReader sstable = index.getSSTable();
              if (sstable.getMaxTimestamp() > truncateUntil)
                  continue;
  
              index.markObsolete();
              toRemove.add(sstable);
          }
  
          update(toRemove, Collections.<SSTableReader>emptyList());
      }
  
      private Pair<Set<SSTableIndex>, Set<SSTableReader>> getBuiltIndexes(Collection<SSTableReader> sstables)
      {
          Set<SSTableIndex> indexes = new HashSet<>(sstables.size());
          Set<SSTableReader> builtSSTables = new HashSet<>(sstables.size());
          for (SSTableReader sstable : sstables)
          {
              if (sstable.isMarkedCompacted())
                  continue;
  
              File indexFile = sstable.descriptor.fileFor(columnIndex.getComponent());
              if (!indexFile.exists())
                  continue;
  
              // if the index file is empty, we have to ignore it to avoid re-building, but it doesn't take
              // a part in query process
              if (indexFile.length() == 0)
              {
                  builtSSTables.add(sstable);
                  continue;
              }
  
              SSTableIndex index = null;
  
              try
              {
                  index = new SSTableIndex(columnIndex, indexFile, sstable);
  
                  logger.info("SSTableIndex.open(column: {}, minTerm: {}, maxTerm: {}, minKey: {}, maxKey: {}, sstable: {})",
                              columnIndex.getColumnName(),
                              columnIndex.getValidator().getString(index.minTerm()),
                              columnIndex.getValidator().getString(index.maxTerm()),
                              keyValidator.getString(index.minKey()),
                              keyValidator.getString(index.maxKey()),
                              index.getSSTable());
  
                  // Try to add new index to the set, if set already has such index, we'll simply release and move on.
                  // This covers situation when sstable collection has the same sstable multiple
                  // times because we don't know what kind of collection it actually is.
                  if (indexes.add(index))
                      builtSSTables.add(sstable);
                  else
                      index.release();
              }
              catch (Throwable t)
              {
                  logger.error("Can't open index file at " + indexFile.absolutePath() + ", skipping.", t);
                  if (index != null)
                      index.release();
              }
          }
  
          return Pair.create(indexes, builtSSTables);
      }
  }
  