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.cql3;
  
  import java.util.Arrays;
  
  import org.junit.Test;
  
  import org.apache.cassandra.Util;
  import org.apache.cassandra.db.Keyspace;
  
  import static org.junit.Assert.assertEquals;
  
  /* ViewComplexTest class has been split into multiple ones because of timeout issues (CASSANDRA-16670, CASSANDRA-17167)
   * Any changes here check if they apply to the other classes:
   * - ViewComplexUpdatesTest
   * - ViewComplexDeletionsTest
   * - ViewComplexTTLTest
   * - ViewComplexTest
   * - ViewComplexLivenessTest
   * - ...
   * - ViewComplex*Test
   */
  public class ViewComplexLivenessLimitTest extends ViewAbstractParameterizedTest
  {
      @Test
      public void testExpiredLivenessLimitWithFlush() throws Throwable
      {
          // CASSANDRA-13883
          testExpiredLivenessLimit(true);
      }
  
      @Test
      public void testExpiredLivenessLimitWithoutFlush() throws Throwable
      {
          // CASSANDRA-13883
          testExpiredLivenessLimit(false);
      }
  
      private void testExpiredLivenessLimit(boolean flush) throws Throwable
      {
          createTable("CREATE TABLE %s (k int PRIMARY KEY, a int, b int);");
  
          Keyspace ks = Keyspace.open(keyspace());
  
          String mv1 = createView("CREATE MATERIALIZED VIEW %s AS SELECT * FROM %s " +
                                  "WHERE k IS NOT NULL AND a IS NOT NULL PRIMARY KEY (k, a)");
          String mv2 = createView("CREATE MATERIALIZED VIEW %s AS SELECT * FROM %s " +
                                  "WHERE k IS NOT NULL AND a IS NOT NULL PRIMARY KEY (a, k)");
          ks.getColumnFamilyStore(mv1).disableAutoCompaction();
          ks.getColumnFamilyStore(mv2).disableAutoCompaction();
  
          for (int i = 1; i <= 100; i++)
              updateView("INSERT INTO %s(k, a, b) VALUES (?, ?, ?);", i, i, i);
          for (int i = 1; i <= 100; i++)
          {
              if (i % 50 == 0)
                  continue;
              // create expired liveness
              updateView("DELETE a FROM %s WHERE k = ?;", i);
          }
  
          if (flush)
          {
              Util.flushTable(ks, mv1);
              Util.flushTable(ks, mv2);
          }
  
          for (String view : Arrays.asList(mv1, mv2))
          {
              // paging
              assertEquals(1, executeNetWithPaging(String.format("SELECT k,a,b FROM %s limit 1", view), 1).all().size());
              assertEquals(2, executeNetWithPaging(String.format("SELECT k,a,b FROM %s limit 2", view), 1).all().size());
              assertEquals(2, executeNetWithPaging(String.format("SELECT k,a,b FROM %s", view), 1).all().size());
              assertRowsNet(executeNetWithPaging(String.format("SELECT k,a,b FROM %s ", view), 1),
                            row(50, 50, 50),
                            row(100, 100, 100));
              // limit
              assertEquals(1, execute(String.format("SELECT k,a,b FROM %s limit 1", view)).size());
              assertRowsIgnoringOrder(execute(String.format("SELECT k,a,b FROM %s limit 2", view)),
                                      row(50, 50, 50),
                                      row(100, 100, 100));
          }
      }
  }
  