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.simulator.test;
  
  import java.util.List;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import accord.utils.Invariants;
  import org.apache.cassandra.distributed.Cluster;
  import org.apache.cassandra.distributed.api.ConsistencyLevel;
  import org.apache.cassandra.distributed.api.IInstance;
  import org.apache.cassandra.harry.gen.EntropySource;
  import org.apache.cassandra.harry.op.Visit;
  import org.apache.cassandra.harry.op.Operations;
  import org.apache.cassandra.harry.execution.CompiledStatement;
  import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
  import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
  import org.apache.cassandra.harry.execution.ResultSetRow;
  import org.apache.cassandra.harry.model.TokenPlacementModel;
  import org.apache.cassandra.harry.util.ByteUtils;
  import org.apache.cassandra.harry.util.TokenUtil;
  import org.apache.cassandra.simulator.systems.InterceptedExecution;
  import org.apache.cassandra.simulator.systems.InterceptingExecutor;
  import org.apache.cassandra.simulator.systems.SimulatedAction;
  
  import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
  
  public class HarryValidatingQuery extends SimulatedAction
  {
      private static final Logger logger = LoggerFactory.getLogger(HarryValidatingQuery.class);
  
      private final InterceptingExecutor on;
      private final Cluster cluster;
      private final List<TokenPlacementModel.Node> owernship;
      private final TokenPlacementModel.ReplicationFactor rf;
  
      private final HarrySimulatorTest.HarrySimulation simulation;
      private final Visit visit;
      private final QueryBuildingVisitExecutor queryBuilder;
      private final ConsistencyLevel consistencyLevel;
      private final EntropySource rng;
  
      public HarryValidatingQuery(HarrySimulatorTest.HarrySimulation simulation,
                                  Cluster cluster,
                                  TokenPlacementModel.ReplicationFactor rf,
                                  List<TokenPlacementModel.Node> owernship,
                                  Visit visit,
                                  QueryBuildingVisitExecutor queryBuilder,
                                  ConsistencyLevel consistencyLevel,
                                  EntropySource rng)
      {
          super(visit, Modifiers.RELIABLE_NO_TIMEOUTS, Modifiers.RELIABLE_NO_TIMEOUTS, null, simulation.simulated);
          this.rf = rf;
          this.cluster = cluster;
          this.on = (InterceptingExecutor) cluster.get(1).executor();
          this.owernship = owernship;
          this.visit = visit;
          this.queryBuilder = queryBuilder;
          this.simulation = simulation;
          this.consistencyLevel = consistencyLevel;
          this.rng = rng;
      }
  
      protected InterceptedExecution task()
      {
          return new InterceptedExecution.InterceptedTaskExecution(on)
          {
              public void run()
              {
                  try
                  {
                      if (consistencyLevel == ConsistencyLevel.NODE_LOCAL)
                      {
                          TokenPlacementModel.ReplicatedRanges ring = rf.replicate(owernship);
                          Invariants.require(visit.operations.length == 1);
                          Invariants.require(visit.operations[0] instanceof Operations.SelectStatement);
                          Operations.SelectStatement select = (Operations.SelectStatement) visit.operations[0];
                          for (TokenPlacementModel.Replica replica : ring.replicasFor(token(select.pd)))
                          {
                              CompiledStatement compiled = queryBuilder.compile(visit);
                              Object[][] objects = executeNodeLocal(compiled.cql(), replica.node(), compiled.bindings());
                              List<ResultSetRow> actualRows = InJvmDTestVisitExecutor.rowsToResultSet(simulation.schema, select, objects);
                              simulation.model.validate(select, actualRows);
                          }
                      }
                      else
                      {
                          Operations.SelectStatement select = (Operations.SelectStatement) visit.operations[0];
                          CompiledStatement compiled = queryBuilder.compile(visit);
                          Object[][] objects = execute(compiled.cql(),   rng.nextInt(cluster.size()) + 1, compiled.bindings());
                          List<ResultSetRow> actualRows = InJvmDTestVisitExecutor.rowsToResultSet(simulation.schema, select, objects);
                          simulation.model.validate(select, actualRows);
                      }
                  }
                  catch (Throwable t)
                  {
                      logger.error("Caught an exception while validating", t);
                      failWithOOM();
                      throw t;
                  }
              }
          };
      }
  
      protected long token(long pd)
      {
          return TokenUtil.token(ByteUtils.compose(ByteUtils.objectsToBytes(simulation.schema.valueGenerators.pkGen().inflate(pd))));
      }
  
      protected Object[][] executeNodeLocal(String statement, TokenPlacementModel.Node node, Object... bindings)
      {
          IInstance instance = cluster
                               .stream()
                               .filter((n) -> n.config().broadcastAddress().toString().equals(node.id()))
                               .findFirst()
                               .get();
          return instance.executeInternal(statement, bindings);
      }
  
      protected Object[][] execute(String statement, int id, Object... bindings)
      {
          IInstance instance = cluster.get(id);
          return instance.coordinator().execute(statement, consistencyLevel, bindings);
      }
  }
  