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.view;
  
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import com.google.common.collect.Maps;
  import com.google.common.collect.Sets;
  import com.google.common.util.concurrent.FutureCallback;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.concurrent.ScheduledExecutors;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.SystemKeyspace;
  import org.apache.cassandra.db.compaction.CompactionInterruptedException;
  import org.apache.cassandra.db.compaction.CompactionManager;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.locator.RangesAtEndpoint;
  import org.apache.cassandra.locator.Replicas;
  import org.apache.cassandra.schema.SystemDistributedKeyspace;
  import org.apache.cassandra.service.StorageService;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.Pair;
  import org.apache.cassandra.utils.concurrent.Future;
  import org.apache.cassandra.utils.concurrent.FutureCombiner;
  import org.apache.cassandra.utils.concurrent.ImmediateFuture;
  
  import static java.util.stream.Collectors.toList;
  
  /**
   * Builds a materialized view for the local token ranges.
   * <p>
   * The build is split in at least {@link #NUM_TASKS} {@link ViewBuilderTask tasks}, suitable of being parallelized by
   * the {@link CompactionManager} which will execute them.
   */
  class ViewBuilder
  {
      private static final Logger logger = LoggerFactory.getLogger(ViewBuilder.class);
  
      private static final int NUM_TASKS = Runtime.getRuntime().availableProcessors() * 4;
  
      private final ColumnFamilyStore baseCfs;
      private final View view;
      private final String ksName;
      private final UUID localHostId;
      private final Set<Range<Token>> builtRanges = Sets.newConcurrentHashSet();
      private final Map<Range<Token>, Pair<Token, Long>> pendingRanges = Maps.newConcurrentMap();
      private final Set<ViewBuilderTask> tasks = Sets.newConcurrentHashSet();
      private volatile long keysBuilt = 0;
      private volatile boolean isStopped = false;
      private volatile Future<?> future = ImmediateFuture.success(null);
  
      ViewBuilder(ColumnFamilyStore baseCfs, View view)
      {
          this.baseCfs = baseCfs;
          this.view = view;
          ksName = baseCfs.metadata.keyspace;
          this.localHostId = ClusterMetadata.current().myNodeId().toUUID();
      }
  
      public void start()
      {
          if (SystemKeyspace.isViewBuilt(ksName, view.name))
          {
              logger.debug("View already marked built for {}.{}", ksName, view.name);
              if (!SystemKeyspace.isViewStatusReplicated(ksName, view.name))
                  updateDistributed();
          }
          else
          {
              SystemDistributedKeyspace.startViewBuild(ksName, view.name, localHostId);
  
              logger.debug("Starting build of view({}.{}). Flushing base table {}.{}",
                           ksName, view.name, ksName, baseCfs.name);
              baseCfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.VIEW_BUILD_STARTED);
  
              loadStatusAndBuild();
          }
      }
  
      private void loadStatusAndBuild()
      {
          loadStatus();
          build();
      }
  
      private void loadStatus()
      {
          builtRanges.clear();
          pendingRanges.clear();
          SystemKeyspace.getViewBuildStatus(ksName, view.name)
                        .forEach((range, pair) ->
                                 {
                                     Token lastToken = pair.left;
                                     if (lastToken != null && lastToken.equals(range.right))
                                     {
                                         builtRanges.add(range);
                                         keysBuilt += pair.right;
                                     }
                                     else
                                     {
                                         pendingRanges.put(range, pair);
                                     }
                                 });
      }
  
      private synchronized void build()
      {
          if (isStopped)
          {
              logger.debug("Stopped build for view({}.{}) after covering {} keys", ksName, view.name, keysBuilt);
              return;
          }
  
          // Get the local ranges for which the view hasn't already been built nor it's building
          RangesAtEndpoint replicatedRanges = StorageService.instance.getLocalReplicas(ksName);
          Replicas.temporaryAssertFull(replicatedRanges);
          Set<Range<Token>> newRanges = replicatedRanges.ranges()
                                                        .stream()
                                                        .map(r -> r.subtractAll(builtRanges))
                                                        .flatMap(Set::stream)
                                                        .map(r -> r.subtractAll(pendingRanges.keySet()))
                                                        .flatMap(Set::stream)
                                                        .collect(Collectors.toSet());
          // If there are no new nor pending ranges we should finish the build
          if (newRanges.isEmpty() && pendingRanges.isEmpty())
          {
              finish();
              return;
          }
  
          // Split the new local ranges and add them to the pending set
          DatabaseDescriptor.getPartitioner()
                            .splitter()
                            .map(s -> s.split(newRanges, NUM_TASKS))
                            .orElse(newRanges)
                            .forEach(r -> pendingRanges.put(r, Pair.<Token, Long>create(null, 0L)));
  
          // Submit a new view build task for each building range.
          // We keep record of all the submitted tasks to be able of stopping them.
          List<Future<Long>> futures = pendingRanges.entrySet()
                                                    .stream()
                                                    .map(e -> new ViewBuilderTask(baseCfs,
                                                                                  view,
                                                                                  e.getKey(),
                                                                                  e.getValue().left,
                                                                                  e.getValue().right))
                                                    .peek(tasks::add)
                                                    .map(CompactionManager.instance::submitViewBuilder)
                                                    .collect(toList());
  
          // Add a callback to process any eventual new local range and mark the view as built, doing a delayed retry if
          // the tasks don't succeed
          Future<List<Long>> future = FutureCombiner.allOf(futures);
          future.addCallback(new FutureCallback<List<Long>>()
          {
              public void onSuccess(List<Long> result)
              {
                  keysBuilt += result.stream().mapToLong(x -> x).sum();
                  builtRanges.addAll(pendingRanges.keySet());
                  pendingRanges.clear();
                  build();
              }
  
              public void onFailure(Throwable t)
              {
                  if (t instanceof CompactionInterruptedException)
                  {
                      internalStop(true);
                      keysBuilt = tasks.stream().mapToLong(ViewBuilderTask::keysBuilt).sum();
                      logger.info("Interrupted build for view({}.{}) after covering {} keys", ksName, view.name, keysBuilt);
                  }
                  else
                  {
                      ScheduledExecutors.nonPeriodicTasks.schedule(() -> loadStatusAndBuild(), 5, TimeUnit.MINUTES);
                      logger.warn("Materialized View failed to complete, sleeping 5 minutes before restarting", t);
                  }
              }
          });
          this.future = future;
      }
  
      private void finish()
      {
          logger.debug("Marking view({}.{}) as built after covering {} keys ", ksName, view.name, keysBuilt);
          SystemKeyspace.finishViewBuildStatus(ksName, view.name);
          updateDistributed();
      }
  
      private void updateDistributed()
      {
          try
          {
              SystemDistributedKeyspace.successfulViewBuild(ksName, view.name, localHostId);
              SystemKeyspace.setViewBuiltReplicated(ksName, view.name);
          }
          catch (Exception e)
          {
              ScheduledExecutors.nonPeriodicTasks.schedule(this::updateDistributed, 5, TimeUnit.MINUTES);
              logger.warn("Failed to update the distributed status of view, sleeping 5 minutes before retrying", e);
          }
      }
  
      /**
       * Stops the view building.
       */
      void stop()
      {
          boolean wasStopped;
          synchronized (this)
          {
              wasStopped = isStopped;
              internalStop(false);
          }
          // TODO: very unclear what the goal is here. why do we wait only if we were the first to invoke stop?
          // but we wait outside the synchronized block to avoid a deadlock with `build` in the future callback
          if (!wasStopped)
              FBUtilities.waitOnFuture(future);
      }
  
      private void internalStop(boolean isCompactionInterrupted)
      {
          isStopped = true;
          tasks.forEach(task -> task.stop(isCompactionInterrupted));
      }
  }
  