<commit_before>from dtest import Tester, debug, DISABLE_VNODES
import unittest
from ccmlib.cluster import Cluster
from ccmlib.node import Node, NodeError, TimeoutError
import time, re
from assertions import assert_invalid, assert_all, assert_none, assert_one
import tempfile
import os

class TestCompaction(Tester):

    __test__= False

    def compaction_delete_test(self):
        """Test that executing a delete properly tombstones a row.
        Insert data, delete a partition of data and check that the requesite rows are tombstoned
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()

        cursor = self.patient_cql_connection(node1)
        self.create_ks(cursor, 'ks', 1)


        cursor.execute("create table ks.cf (key int PRIMARY KEY, val int) with compaction = {'class':'" + self.strategy + "'} and gc_grace_seconds = 30;")

        for x in range(0, 100):
            cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')

        node1.flush()
        for x in range(0, 10):
            cursor.execute('delete from cf where key = ' + str(x))

        node1.flush()
        for x in range(0, 10):
            assert_none(cursor, 'select * from cf where key = ' + str(x))

        json_path = tempfile.mkstemp(suffix='.json')
        jname = json_path[1]
        with open(jname, 'w') as f:
            node1.run_sstable2json(f)

        with open(jname, 'r') as g:
            jsoninfo = g.read()

        numfound = jsoninfo.count("markedForDeleteAt")

        self.assertEqual(numfound, 10)

    def data_size_test(self):
        """Ensure that data size does not have unwarranted increases after compaction.
        Insert data and check data size before and after a compaction.
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        node1.stress(['write', 'n=100000'])

        node1.flush()

        output = node1.nodetool('cfstats', True)[0]
        initialValue = 0
        if output.find("standard1") != -1:
            output = output[output.find("standard1"):]
            output = output[output.find("Space used (live):"):]
            initialValue = output[output.find(":")+1:output.find("\n")].strip()
        else:
            debug("datasize not found")
            debug(output)

        node1.nodetool('compact')

        output = node1.nodetool('cfstats', True)[0]
        finalValue = 0
        if output.find("standard1") != -1:
            output = output[output.find("standard1"):]
            output = output[output.find("Space used (live):"):]
            finalValue = output[output.find(":")+1:output.find("\n")].strip()
        else:
            debug("datasize not found")

        self.assertFalse(finalValue >= initialValue)

    def sstable_deletion_test(self):
        """Test that sstables are deleted properly when able to be.
        Insert data setting gc_grace_seconds to 0, and determine sstable
        is deleted upon data deletion.
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        self.create_ks(cursor, 'ks', 1)
        cursor.execute("create table cf (key int PRIMARY KEY, val int) with gc_grace_seconds = 0 and compaction= {'class':'" +self.strategy+"'}")

        for x in range(0, 100):
            cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')
        node1.flush()
        for x in range(0, 100):
            cursor.execute('delete from cf where key = ' + str(x))
        node1.flush()
        node1.nodetool('compact')

        try:
            cfs = os.listdir(node1.get_path() + "/data/ks")
            ssdir = os.listdir(node1.get_path() + "/data/ks/" + cfs[0])
            for afile in ssdir:
                self.assertFalse("Data" in afile)            

        except OSError:
            self.fail("Path to sstables not valid.")

    def dtcs_deletion_test(self):
        """Test that sstables are deleted properly when able to be.
        Insert data setting max_sstable_age_days low, and determine sstable
        is deleted upon data deletion past max_sstable_age_days.
        """
        if self.strategy == 'DateTieredCompactionStrategy':
            cluster = self.cluster
            cluster.populate(1).start()
            [node1] = cluster.nodelist()
            cursor = self.patient_cql_connection(node1)
            self.create_ks(cursor, 'ks', 1)
            # we set max_sstable_age_days to 10 seconds
            cursor.execute("create table cf (key int PRIMARY KEY, val int) with gc_grace_seconds = 0 and compaction= {'class':'DateTieredCompactionStrategy', 'max_sstable_age_days':0.000116, 'min_threshold':2}")

            #insert data
            for x in range(0, 300):
                cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')
            node1.flush()
            time.sleep(15)

            #write tombstones
            for x in range(0, 300):
                cursor.execute('delete from cf where key = ' + str(x))
            node1.flush()
            time.sleep(15)

            #get all sstables right now
            cfs = os.listdir(node1.get_path() + "/data/ks")
            ssdir = os.listdir(node1.get_path() + "/data/ks/" + cfs[0])
            initialsstables = []
            for afile in ssdir:
                if "Data" in afile:
                    initialsstables.append(afile)
            debug(initialsstables)

            #trigger minor compaction by writing more data
            for x in range(300, 400):
                cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')
            node1.flush()
            time.sleep(10)

            try:
                #should see no expired sstables from initial write and deletion
                finalsstables = []
                newssdir = os.listdir(node1.get_path() + "/data/ks/" + cfs[0])
                for anfile in newssdir:
                    if "Data" in anfile:
                        finalsstables.append(anfile)
                debug(finalsstables)

                #expired and tombstone sstables should be gone
                for expiredsstable in initialsstables:
                    self.assertFalse(expiredsstable in finalsstables, 'expired and tombstone sstables are not gone')
            except OSError:
                self.fail("Invalid path to sstables.")
        else:
            pass

    def compaction_throughput_test(self):
        """Test setting compaction throughput.
        Set throughput, insert data and ensure compaction performance corresponds.
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        node1.stress(['write', "n=100000"])
        node1.flush()

        threshold = "10"

        node1.nodetool('setcompactionthroughput -- ' + threshold)
        node1.nodetool('compact')

        matches = node1.watch_log_for("Compacted")

        stringline = matches[0]
        avgthroughput = stringline[stringline.find('=')+1:stringline.find("MB/s")]
        debug(avgthroughput)

        self.assertGreaterEqual(threshold, avgthroughput)

    def compaction_strategy_switching_test(self):
        """Ensure that switching strategies does not result in problems.
        Insert data, switch strategies, then check against data loss.
        """
        strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy']

        if self.strategy in strategies:
            strategies.remove(self.strategy)
            cluster = self.cluster
            cluster.populate(1).start()
            [node1] = cluster.nodelist()


            for strat in strategies:
                cursor = self.patient_cql_connection(node1)
                self.create_ks(cursor, 'ks', 1)

                cursor.execute("create table ks.cf (key int PRIMARY KEY, val int) with gc_grace_seconds = 0 and compaction= {'class':'" + self.strategy + "'};")

                for x in range(0, 100):
                    cursor.execute('insert into ks.cf (key, val) values (' + str(x) + ',1)')

                node1.flush()

                for x in range(0, 10):
                    cursor.execute('delete from cf where key = ' + str(x))

                cursor.execute("alter table ks.cf with compaction = {'class':'" + strat + "'};")

                for x in range(11,100):
                    assert_one(cursor, "select * from ks.cf where key =" + str(x),[x, 1])

                for x in range(0, 10):
                    assert_none(cursor, 'select * from cf where key = ' + str(x))

                node1.flush()
                cluster.clear()
                time.sleep(5)
                cluster.start()

strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy']
for strategy in strategies:
    cls_name = ('TestCompaction_with_' + strategy)
    vars()[cls_name] = type(cls_name, (TestCompaction,), {'strategy': strategy, '__test__':True})

<commit_msg>Fix up running dtcs deletion test by itself<commit_after>from dtest import Tester, debug, DISABLE_VNODES
import unittest
from ccmlib.cluster import Cluster
from ccmlib.node import Node, NodeError, TimeoutError
import time, re
from assertions import assert_invalid, assert_all, assert_none, assert_one
import tempfile
import os

class TestCompaction(Tester):

    __test__= False

    def compaction_delete_test(self):
        """Test that executing a delete properly tombstones a row.
        Insert data, delete a partition of data and check that the requesite rows are tombstoned
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()

        cursor = self.patient_cql_connection(node1)
        self.create_ks(cursor, 'ks', 1)


        cursor.execute("create table ks.cf (key int PRIMARY KEY, val int) with compaction = {'class':'" + self.strategy + "'} and gc_grace_seconds = 30;")

        for x in range(0, 100):
            cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')

        node1.flush()
        for x in range(0, 10):
            cursor.execute('delete from cf where key = ' + str(x))

        node1.flush()
        for x in range(0, 10):
            assert_none(cursor, 'select * from cf where key = ' + str(x))

        json_path = tempfile.mkstemp(suffix='.json')
        jname = json_path[1]
        with open(jname, 'w') as f:
            node1.run_sstable2json(f)

        with open(jname, 'r') as g:
            jsoninfo = g.read()

        numfound = jsoninfo.count("markedForDeleteAt")

        self.assertEqual(numfound, 10)

    def data_size_test(self):
        """Ensure that data size does not have unwarranted increases after compaction.
        Insert data and check data size before and after a compaction.
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        node1.stress(['write', 'n=100000'])

        node1.flush()

        output = node1.nodetool('cfstats', True)[0]
        initialValue = 0
        if output.find("standard1") != -1:
            output = output[output.find("standard1"):]
            output = output[output.find("Space used (live):"):]
            initialValue = output[output.find(":")+1:output.find("\n")].strip()
        else:
            debug("datasize not found")
            debug(output)

        node1.nodetool('compact')

        output = node1.nodetool('cfstats', True)[0]
        finalValue = 0
        if output.find("standard1") != -1:
            output = output[output.find("standard1"):]
            output = output[output.find("Space used (live):"):]
            finalValue = output[output.find(":")+1:output.find("\n")].strip()
        else:
            debug("datasize not found")

        self.assertFalse(finalValue >= initialValue)

    def sstable_deletion_test(self):
        """Test that sstables are deleted properly when able to be.
        Insert data setting gc_grace_seconds to 0, and determine sstable
        is deleted upon data deletion.
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        self.create_ks(cursor, 'ks', 1)
        cursor.execute("create table cf (key int PRIMARY KEY, val int) with gc_grace_seconds = 0 and compaction= {'class':'" +self.strategy+"'}")

        for x in range(0, 100):
            cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')
        node1.flush()
        for x in range(0, 100):
            cursor.execute('delete from cf where key = ' + str(x))
        node1.flush()
        node1.nodetool('compact')

        try:
            cfs = os.listdir(node1.get_path() + "/data/ks")
            ssdir = os.listdir(node1.get_path() + "/data/ks/" + cfs[0])
            for afile in ssdir:
                self.assertFalse("Data" in afile)            

        except OSError:
            self.fail("Path to sstables not valid.")

    def dtcs_deletion_test(self):
        """Test that sstables are deleted properly when able to be.
        Insert data setting max_sstable_age_days low, and determine sstable
        is deleted upon data deletion past max_sstable_age_days.
        """
        if not hasattr(self, 'strategy'):
            self.stategy = 'DateTieredCompactionStrategy'
        elif self.stategy != 'DateTieredCompactionStrategy':
            self.skip('Not implemented unless DateTieredCompactionStrategy is used')
            
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        self.create_ks(cursor, 'ks', 1)
        # we set max_sstable_age_days to 10 seconds
        cursor.execute("create table cf (key int PRIMARY KEY, val int) with gc_grace_seconds = 0 and compaction= {'class':'DateTieredCompactionStrategy', 'max_sstable_age_days':0.000116, 'min_threshold':2}")

        #insert data
        for x in range(0, 300):
            cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')
        node1.flush()
        time.sleep(15)

        #write tombstones
        for x in range(0, 300):
            cursor.execute('delete from cf where key = ' + str(x))
        node1.flush()
        time.sleep(15)

        #get all sstables right now
        cfs = os.listdir(node1.get_path() + "/data/ks")
        ssdir = os.listdir(node1.get_path() + "/data/ks/" + cfs[0])
        initialsstables = []
        for afile in ssdir:
            if "Data" in afile:
                initialsstables.append(afile)
        debug(initialsstables)

        #trigger minor compaction by writing more data
        for x in range(300, 400):
            cursor.execute('insert into cf (key, val) values (' + str(x) + ',1)')
        node1.flush()
        time.sleep(10)

        try:
            #should see no expired sstables from initial write and deletion
            finalsstables = []
            newssdir = os.listdir(node1.get_path() + "/data/ks/" + cfs[0])
            for anfile in newssdir:
                if "Data" in anfile:
                    finalsstables.append(anfile)
            debug(finalsstables)

            #expired and tombstone sstables should be gone
            for expiredsstable in initialsstables:
                self.assertFalse(expiredsstable in finalsstables, 'expired and tombstone sstables are not gone')
        except OSError:
            self.fail("Invalid path to sstables.")

    def compaction_throughput_test(self):
        """Test setting compaction throughput.
        Set throughput, insert data and ensure compaction performance corresponds.
        """
        cluster = self.cluster
        cluster.populate(1).start()
        [node1] = cluster.nodelist()
        cursor = self.patient_cql_connection(node1)
        node1.stress(['write', "n=100000"])
        node1.flush()

        threshold = "10"

        node1.nodetool('setcompactionthroughput -- ' + threshold)
        node1.nodetool('compact')

        matches = node1.watch_log_for("Compacted")

        stringline = matches[0]
        avgthroughput = stringline[stringline.find('=')+1:stringline.find("MB/s")]
        debug(avgthroughput)

        self.assertGreaterEqual(threshold, avgthroughput)

    def compaction_strategy_switching_test(self):
        """Ensure that switching strategies does not result in problems.
        Insert data, switch strategies, then check against data loss.
        """
        strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy']

        if self.strategy in strategies:
            strategies.remove(self.strategy)
            cluster = self.cluster
            cluster.populate(1).start()
            [node1] = cluster.nodelist()


            for strat in strategies:
                cursor = self.patient_cql_connection(node1)
                self.create_ks(cursor, 'ks', 1)

                cursor.execute("create table ks.cf (key int PRIMARY KEY, val int) with gc_grace_seconds = 0 and compaction= {'class':'" + self.strategy + "'};")

                for x in range(0, 100):
                    cursor.execute('insert into ks.cf (key, val) values (' + str(x) + ',1)')

                node1.flush()

                for x in range(0, 10):
                    cursor.execute('delete from cf where key = ' + str(x))

                cursor.execute("alter table ks.cf with compaction = {'class':'" + strat + "'};")

                for x in range(11,100):
                    assert_one(cursor, "select * from ks.cf where key =" + str(x),[x, 1])

                for x in range(0, 10):
                    assert_none(cursor, 'select * from cf where key = ' + str(x))

                node1.flush()
                cluster.clear()
                time.sleep(5)
                cluster.start()

strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy']
for strategy in strategies:
    cls_name = ('TestCompaction_with_' + strategy)
    vars()[cls_name] = type(cls_name, (TestCompaction,), {'strategy': strategy, '__test__':True})


{'fraction_non_alphanumeric': 0.11384482572569332, 'fraction_numerical': 0.01715765619941729, 'mean_word_length': 2.191287878787879, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '464162', 'n_tokens_mistral': 5641, 'n_tokens_neox': 5280, 'n_words': 1409}