RangeTombstoneMarker and PartitionDeletion is not properly included in MV

    @Test
    public void testExistingRangeTombstoneWithFlush() throws Throwable
    {
        testExistingRangeTombstone(true);
    }

    @Test
    public void testExistingRangeTombstoneWithoutFlush() throws Throwable
    {
        testExistingRangeTombstone(false);
    }

    public void testExistingRangeTombstone(boolean flush) throws Throwable
    {
        createTable("CREATE TABLE %s (k1 int, c1 int, c2 int, v1 int, v2 int, PRIMARY KEY (k1, c1, c2))");

        execute("USE " + keyspace());
        executeNet(protocolVersion, "USE " + keyspace());

        createView("view1",
                   "CREATE MATERIALIZED VIEW view1 AS SELECT * FROM %%s WHERE k1 IS NOT NULL AND c1 IS NOT NULL AND c2 IS NOT NULL PRIMARY KEY (k1, c2, c1)");

        updateView("DELETE FROM %s USING TIMESTAMP 10 WHERE k1 = 1 and c1=1");


        if (flush)
            Keyspace.open(keyspace()).getColumnFamilyStore(currentTable()).forceBlockingFlush();

        String table = KEYSPACE + "." + currentTable();
        updateView("BEGIN BATCH " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 0, 0, 0, 0) USING TIMESTAMP 5; " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 0, 1, 0, 1) USING TIMESTAMP 5; " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 1, 0, 1, 0) USING TIMESTAMP 5; " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 1, 1, 1, 1) USING TIMESTAMP 5; " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 1, 2, 1, 2) USING TIMESTAMP 5; " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 1, 3, 1, 3) USING TIMESTAMP 5; " +
                "INSERT INTO " + table + " (k1, c1, c2, v1, v2) VALUES (1, 2, 0, 2, 0) USING TIMESTAMP 5; " +
                "APPLY BATCH");

        assertRowsIgnoringOrder(execute("select * from %s"),
                                row(1, 0, 0, 0, 0),
                                row(1, 0, 1, 0, 1),
                                row(1, 2, 0, 2, 0));
        assertRowsIgnoringOrder(execute("select k1,c1,c2,v1,v2 from view1"),
                                row(1, 0, 0, 0, 0),
                                row(1, 0, 1, 0, 1),
                                row(1, 2, 0, 2, 0));
    }

    @Test
    public void testExistingParitionDeletionWithFlush() throws Throwable
    {
        testExistingParitionDeletion(true);
    }

    @Test
    public void testExistingParitionDeletionWithoutFlush() throws Throwable
    {
        testExistingParitionDeletion(false);
    }

    public void testExistingParitionDeletion(boolean flush) throws Throwable
    {
        // for partition range deletion, need to know that existing row is shadowed instead of not existed.
        createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a))");

        execute("USE " + keyspace());
        executeNet(protocolVersion, "USE " + keyspace());

        createView("mv_test1",
                   "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (a, b)");

        Keyspace ks = Keyspace.open(keyspace());
        ks.getColumnFamilyStore("mv_test1").disableAutoCompaction();

        execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?) using timestamp 0", 1, 1, 1, 1);
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());

        assertRowsIgnoringOrder(execute("SELECT * FROM mv_test1"), row(1, 1, 1, 1));

        // remove view row
        updateView("UPDATE %s using timestamp 1 set b = null WHERE a=1");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());

        assertRowsIgnoringOrder(execute("SELECT * FROM mv_test1"));
        // remove base row, no view updated generated.
        updateView("DELETE FROM %s using timestamp 2 where a=1");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());

        assertRowsIgnoringOrder(execute("SELECT * FROM mv_test1"));

        // restor view row with b,c column. d is still tombstone
        updateView("UPDATE %s using timestamp 3 set b = 1,c = 1 where a=1"); // upsert
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());

        assertRowsIgnoringOrder(execute("SELECT * FROM mv_test1"), row(1, 1, 1, null));
    }