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 harry.model;
  
  import java.util.*;
  import java.util.function.Supplier;
  
  import harry.core.Run;
  import harry.data.ResultSetRow;
  import harry.ddl.ColumnSpec;
  import harry.ddl.SchemaSpec;
  import harry.model.sut.SystemUnderTest;
  import harry.reconciler.PartitionState;
  import harry.operations.Query;
  import harry.reconciler.Reconciler;
  import harry.runner.DataTracker;
  
  import static harry.generators.DataGenerators.NIL_DESCR;
  import static harry.generators.DataGenerators.UNSET_DESCR;
  
  public class QuiescentChecker implements Model
  {
      protected final OpSelectors.Clock clock;
  
      protected final DataTracker tracker;
      protected final SystemUnderTest sut;
      protected final Reconciler reconciler;
      protected final SchemaSpec schema;
  
      public QuiescentChecker(Run run)
      {
          this(run, new Reconciler(run));
      }
  
      public QuiescentChecker(Run run, Reconciler reconciler)
      {
          this(run.clock, run.sut, run.tracker, run.schemaSpec, reconciler);
      }
  
      public QuiescentChecker(OpSelectors.Clock clock,
                              SystemUnderTest sut,
                              DataTracker tracker,
                              SchemaSpec schema,
                              Reconciler reconciler)
      {
          this.clock = clock;
          this.sut = sut;
          this.reconciler = reconciler;
          this.tracker = tracker;
          this.schema = schema;
      }
  
      public void validate(Query query)
      {
          tracker.beginValidation(query.pd);
          validate(() -> SelectHelper.execute(sut, clock, query), query);
          tracker.endValidation(query.pd);
      }
  
      protected void validate(Supplier<List<ResultSetRow>> rowsSupplier, Query query)
      {
          List<ResultSetRow> actualRows = rowsSupplier.get();
          PartitionState partitionState = reconciler.inflatePartitionState(query.pd, tracker, query);
          validate(schema, tracker, partitionState, actualRows, query);
      }
  
      public static void validate(SchemaSpec schema, DataTracker tracker, PartitionState partitionState, List<ResultSetRow> actualRows, Query query)
      {
          Set<ColumnSpec<?>> columns = new HashSet<>();
          columns.addAll(schema.allColumns);
          validate(schema, tracker, columns, partitionState, actualRows, query);
      }
  
      public static Reconciler.RowState adjustForSelection(Reconciler.RowState row, SchemaSpec schema, Set<ColumnSpec<?>> selection, boolean isStatic)
      {
          if (selection.size() == schema.allColumns.size())
              return row;
  
          List<ColumnSpec<?>> columns = isStatic ? schema.staticColumns : schema.regularColumns;
          Reconciler.RowState newRowState = row.clone();
          assert newRowState.vds.length == columns.size();
          for (int i = 0; i < columns.size(); i++)
          {
              if (!selection.contains(columns.get(i)))
              {
                  newRowState.vds[i] = UNSET_DESCR;
                  newRowState.lts[i] = NO_TIMESTAMP;
              }
          }
          return newRowState;
      }
  
      public static void validate(SchemaSpec schema, DataTracker tracker, Set<ColumnSpec<?>> selection, PartitionState partitionState, List<ResultSetRow> actualRows, Query query)
      {
          boolean isWildcardQuery = selection == null;
          String trackerBefore = tracker.toString();
          if (isWildcardQuery)
              selection = new HashSet<>(schema.allColumns);
  
          Iterator<ResultSetRow> actual = actualRows.iterator();
          Collection<Reconciler.RowState> expectedRows = partitionState.rows(query.reverse);
  
          Iterator<Reconciler.RowState> expected = expectedRows.iterator();
  
          String trackerState = String.format("Tracker before: %s, Tracker after: %s", trackerBefore, tracker);
  
          // It is possible that we only get a single row in response, and it is equal to static row
          if (partitionState.isEmpty() && partitionState.staticRow() != null && actual.hasNext())
          {
              ResultSetRow actualRowState = actual.next();
              if (actualRowState.cd != UNSET_DESCR && actualRowState.cd != partitionState.staticRow().cd)
              {
                  throw new ValidationException(trackerState,
                                                partitionState.toString(schema),
                                                toString(actualRows),
                                                "Found a row while model predicts statics only:" +
                                                "\nExpected: %s" +
                                                "\nActual: %s" +
                                                "\nQuery: %s",
                                                partitionState.staticRow(),
                                                actualRowState,
                                                query.toSelectStatement());
              }
  
              for (int i = 0; i < actualRowState.vds.length; i++)
              {
                  if (actualRowState.vds[i] != NIL_DESCR || actualRowState.lts[i] != NO_TIMESTAMP)
                      throw new ValidationException(trackerState,
                                                    partitionState.toString(schema),
                                                    toString(actualRows),
                                                    "Found a row while model predicts statics only:" +
                                                    "\nActual: %s" +
                                                    "\nQuery: %s",
                                                    actualRowState, query.toSelectStatement());
              }
  
              assertStaticRow(partitionState, actualRows,
                              adjustForSelection(partitionState.staticRow(), schema, selection, true),
                              actualRowState, query, trackerState, schema, isWildcardQuery);
          }
  
          while (actual.hasNext() && expected.hasNext())
          {
              ResultSetRow actualRowState = actual.next();
              Reconciler.RowState originalExpectedRowState = expected.next();
              Reconciler.RowState expectedRowState = adjustForSelection(originalExpectedRowState, schema, selection, false);
  
              if (schema.trackLts)
                  partitionState.compareVisitedLts(actualRowState.visited_lts);
  
              // TODO: this is not necessarily true. It can also be that ordering is incorrect.
              if (actualRowState.cd != UNSET_DESCR && actualRowState.cd != expectedRowState.cd)
              {
                  throw new ValidationException(trackerState,
                                                partitionState.toString(schema),
                                                toString(actualRows),
                                                "Found a row in the model that is not present in the resultset:" +
                                                "\nExpected: %s" +
                                                "\nActual: %s" +
                                                "\nQuery: %s",
                                                expectedRowState.toString(schema),
                                                actualRowState, query.toSelectStatement());
              }
  
              if (!Arrays.equals(actualRowState.vds, expectedRowState.vds))
                  throw new ValidationException(trackerState,
                                                partitionState.toString(schema),
                                                toString(actualRows),
                                                "Returned row state doesn't match the one predicted by the model:" +
                                                "\nExpected: %s (%s)" +
                                                "\nActual:   %s (%s)." +
                                                "\nQuery: %s",
                                                descriptorsToString(expectedRowState.vds), expectedRowState.toString(schema),
                                                descriptorsToString(actualRowState.vds), actualRowState,
                                                query.toSelectStatement());
  
              // Wildcard queries do not include timestamps
              if (!isWildcardQuery && !Arrays.equals(actualRowState.lts, expectedRowState.lts))
                  throw new ValidationException(trackerState,
                                                partitionState.toString(schema),
                                                toString(actualRows),
                                                "Timestamps in the row state don't match ones predicted by the model:" +
                                                "\nExpected: %s (%s)" +
                                                "\nActual:   %s (%s)." +
                                                "\nQuery: %s",
                                                Arrays.toString(expectedRowState.lts), expectedRowState.toString(schema),
                                                Arrays.toString(actualRowState.lts), actualRowState,
                                                query.toSelectStatement());
  
              if (partitionState.staticRow() != null || actualRowState.hasStaticColumns())
              {
                  Reconciler.RowState expectedStaticRowState = adjustForSelection(partitionState.staticRow(), schema, selection, true);
                  assertStaticRow(partitionState, actualRows, expectedStaticRowState, actualRowState, query, trackerState, schema, isWildcardQuery);
              }
          }
  
          if (actual.hasNext() || expected.hasNext())
          {
              throw new ValidationException(trackerState,
                                            partitionState.toString(schema),
                                            toString(actualRows),
                                            "Expected results to have the same number of results, but %s result iterator has more results." +
                                            "\nExpected: %s" +
                                            "\nActual:   %s" +
                                            "\nQuery: %s",
                                            actual.hasNext() ? "actual" : "expected",
                                            expectedRows,
                                            actualRows,
                                            query.toSelectStatement());
          }
      }
  
      public static void assertStaticRow(PartitionState partitionState,
                                         List<ResultSetRow> actualRows,
                                         Reconciler.RowState staticRow,
                                         ResultSetRow actualRowState,
                                         Query query,
                                         String trackerState,
                                         SchemaSpec schemaSpec,
                                         boolean isWildcardQuery)
      {
          if (!Arrays.equals(staticRow.vds, actualRowState.sds))
              throw new ValidationException(trackerState,
                                            partitionState.toString(schemaSpec),
                                            toString(actualRows),
                                            "Returned static row state doesn't match the one predicted by the model:" +
                                            "\nExpected: %s (%s)" +
                                            "\nActual:   %s (%s)." +
                                            "\nQuery: %s",
                                            descriptorsToString(staticRow.vds), staticRow.toString(schemaSpec),
                                            descriptorsToString(actualRowState.sds), actualRowState,
                                            query.toSelectStatement());
  
          if (!isWildcardQuery && !Arrays.equals(staticRow.lts, actualRowState.slts))
              throw new ValidationException(trackerState,
                                            partitionState.toString(schemaSpec),
                                            toString(actualRows),
                                            "Timestamps in the static row state don't match ones predicted by the model:" +
                                            "\nExpected: %s (%s)" +
                                            "\nActual:   %s (%s)." +
                                            "\nQuery: %s",
                                            Arrays.toString(staticRow.lts), staticRow.toString(schemaSpec),
                                            Arrays.toString(actualRowState.slts), actualRowState,
                                            query.toSelectStatement());
      }
  
      public static String descriptorsToString(long[] descriptors)
      {
          StringBuilder sb = new StringBuilder();
          for (int i = 0; i < descriptors.length; i++)
          {
              if (descriptors[i] == NIL_DESCR)
                  sb.append("NIL");
              if (descriptors[i] == UNSET_DESCR)
                  sb.append("UNSET");
              else
                  sb.append(descriptors[i]);
              if (i > 0)
                  sb.append(", ");
          }
          return sb.toString();
      }
      public static String toString(Collection<Reconciler.RowState> collection, SchemaSpec schema)
      {
          StringBuilder builder = new StringBuilder();
  
          for (Reconciler.RowState rowState : collection)
              builder.append(rowState.toString(schema)).append("\n");
          return builder.toString();
      }
  
      public static String toString(List<ResultSetRow> collection)
      {
          StringBuilder builder = new StringBuilder();
  
          for (ResultSetRow rowState : collection)
              builder.append(rowState.toString()).append("\n");
          return builder.toString();
      }
  }
  