cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From pa...@apache.org
Subject [25/50] cassandra git commit: CASSANDRA-10130 (#1486)
Date Tue, 05 Sep 2017 06:05:39 GMT
CASSANDRA-10130 (#1486)

* Add test case for CASSANDRA-10130

* Address comments by @sbtourist

* Add more tests for index status management

* Ad missed `@staticmethod` annotation

* Add @since annotations for 4.0

* Update failing index build failures

* Fix code style removing trailing whitespaces and blank lines with whitespaces


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/50e1e7b1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/50e1e7b1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/50e1e7b1

Branch: refs/heads/master
Commit: 50e1e7b13a1eef3e9347aee7806dc40569ab17ad
Parents: 6847bc1
Author: Andrés de la Peña <adelapena@users.noreply.github.com>
Authored: Mon Jun 26 13:18:55 2017 +0100
Committer: Philip Thompson <ptnapoleon@gmail.com>
Committed: Mon Jun 26 14:18:55 2017 +0200

----------------------------------------------------------------------
 byteman/index_build_failure.btm    |  13 +++
 secondary_indexes_test.py          | 174 +++++++++++++++++++++++++++++---
 sstable_generation_loading_test.py | 122 +++++++++++++++++-----
 3 files changed, 271 insertions(+), 38 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/50e1e7b1/byteman/index_build_failure.btm
----------------------------------------------------------------------
diff --git a/byteman/index_build_failure.btm b/byteman/index_build_failure.btm
new file mode 100644
index 0000000..8f5183d
--- /dev/null
+++ b/byteman/index_build_failure.btm
@@ -0,0 +1,13 @@
+#
+# Sleep 5s during index update
+#
+RULE fail during index building
+CLASS org.apache.cassandra.db.compaction.CompactionManager
+METHOD submitIndexBuild
+AT ENTRY
+# set flag to only run this rule once.
+IF NOT flagged("done")
+DO
+   flag("done");
+   throw new java.lang.RuntimeException("Index building failure")
+ENDRULE

http://git-wip-us.apache.org/repos/asf/cassandra/blob/50e1e7b1/secondary_indexes_test.py
----------------------------------------------------------------------
diff --git a/secondary_indexes_test.py b/secondary_indexes_test.py
index b73e94d..1edd30e 100644
--- a/secondary_indexes_test.py
+++ b/secondary_indexes_test.py
@@ -13,7 +13,7 @@ from cassandra.query import BatchStatement, SimpleStatement
 
 from dtest import (DISABLE_VNODES, OFFHEAP_MEMTABLES, DtestTimeoutError,
                    Tester, debug, CASSANDRA_VERSION_FROM_BUILD, create_ks, create_cf)
-from tools.assertions import assert_bootstrap_state, assert_invalid, assert_one, assert_row_count
+from tools.assertions import assert_bootstrap_state, assert_invalid, assert_none, assert_one,
assert_row_count
 from tools.data import index_is_built, rows_to_list
 from tools.decorators import since
 from tools.misc import new_node
@@ -21,6 +21,16 @@ from tools.misc import new_node
 
 class TestSecondaryIndexes(Tester):
 
+    @staticmethod
+    def _index_sstables_files(node, keyspace, table, index):
+        files = []
+        for data_dir in node.data_directories():
+            data_dir = os.path.join(data_dir, keyspace)
+            base_tbl_dir = os.path.join(data_dir, [s for s in os.listdir(data_dir) if s.startswith(table)][0])
+            index_sstables_dir = os.path.join(base_tbl_dir, '.' + index)
+            files.extend(os.listdir(index_sstables_dir))
+        return set(files)
+
     def data_created_before_index_not_returned_in_where_query_test(self):
         """
         @jira_ticket CASSANDRA-3367
@@ -307,14 +317,7 @@ class TestSecondaryIndexes(Tester):
 
         stmt = session.prepare('select * from standard1 where "C0" = ?')
         self.assertEqual(1, len(list(session.execute(stmt, [lookup_value]))))
-        before_files = []
-        index_sstables_dirs = []
-        for data_dir in node1.data_directories():
-            data_dir = os.path.join(data_dir, 'keyspace1')
-            base_tbl_dir = os.path.join(data_dir, [s for s in os.listdir(data_dir) if s.startswith("standard1")][0])
-            index_sstables_dir = os.path.join(base_tbl_dir, '.ix_c0')
-            before_files.extend(os.listdir(index_sstables_dir))
-            index_sstables_dirs.append(index_sstables_dir)
+        before_files = self._index_sstables_files(node1, 'keyspace1', 'standard1', 'ix_c0')
 
         node1.nodetool("rebuild_index keyspace1 standard1 ix_c0")
         start = time.time()
@@ -326,15 +329,160 @@ class TestSecondaryIndexes(Tester):
         else:
             raise DtestTimeoutError()
 
-        after_files = []
-        for index_sstables_dir in index_sstables_dirs:
-            after_files.extend(os.listdir(index_sstables_dir))
-        self.assertNotEqual(set(before_files), set(after_files))
+        after_files = self._index_sstables_files(node1, 'keyspace1', 'standard1', 'ix_c0')
+        self.assertNotEqual(before_files, after_files)
         self.assertEqual(1, len(list(session.execute(stmt, [lookup_value]))))
 
         # verify that only the expected row is present in the build indexes table
         self.assertEqual(1, len(list(session.execute("""SELECT * FROM system."IndexInfo";"""))))
 
+    @since('4.0')
+    def test_failing_manual_rebuild_index(self):
+        """
+        @jira_ticket CASSANDRA-10130
+
+        Tests the management of index status during manual index rebuilding failures.
+        """
+
+        cluster = self.cluster
+        cluster.populate(1, install_byteman=True).start(wait_for_binary_proto=True)
+        node = cluster.nodelist()[0]
+
+        session = self.patient_cql_connection(node)
+        create_ks(session, 'k', 1)
+        session.execute("CREATE TABLE k.t (k int PRIMARY KEY, v int)")
+        session.execute("CREATE INDEX idx ON k.t(v)")
+        session.execute("INSERT INTO k.t(k, v) VALUES (0, 1)")
+        session.execute("INSERT INTO k.t(k, v) VALUES (2, 3)")
+
+        # Verify that the index is marked as built and it can answer queries
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
+        # Simulate a failing index rebuild
+        before_files = self._index_sstables_files(node, 'k', 't', 'idx')
+        node.byteman_submit(['./byteman/index_build_failure.btm'])
+        with self.assertRaises(Exception):
+            node.nodetool("rebuild_index k t idx")
+        after_files = self._index_sstables_files(node, 'k', 't', 'idx')
+
+        # Verify that the index is not rebuilt, not marked as built, and it still can answer
queries
+        self.assertEqual(before_files, after_files)
+        assert_none(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""")
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
+        # Restart the node to trigger the scheduled index rebuild
+        before_files = after_files
+        node.nodetool('drain')
+        node.stop()
+        cluster.start()
+        session = self.patient_cql_connection(node)
+        session.execute("USE k")
+        after_files = self._index_sstables_files(node, 'k', 't', 'idx')
+
+        # Verify that, the index is rebuilt, marked as built, and it can answer queries
+        self.assertNotEqual(before_files, after_files)
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
+        # Simulate another failing index rebuild
+        before_files = self._index_sstables_files(node, 'k', 't', 'idx')
+        node.byteman_submit(['./byteman/index_build_failure.btm'])
+        with self.assertRaises(Exception):
+            node.nodetool("rebuild_index k t idx")
+        after_files = self._index_sstables_files(node, 'k', 't', 'idx')
+
+        # Verify that the index is not rebuilt, not marked as built, and it still can answer
queries
+        self.assertEqual(before_files, after_files)
+        assert_none(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""")
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
+        # Successfully rebuild the index
+        before_files = after_files
+        node.nodetool("rebuild_index k t idx")
+        cluster.wait_for_compactions()
+        after_files = self._index_sstables_files(node, 'k', 't', 'idx')
+
+        # Verify that the index is rebuilt, marked as built, and it can answer queries
+        self.assertNotEqual(before_files, after_files)
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
+    @since('4.0')
+    def test_drop_index_while_building(self):
+        """
+        asserts that indexes deleted before they have been completely build are invalidated
and not built after restart
+        """
+        cluster = self.cluster
+        cluster.populate(1).start()
+        node = cluster.nodelist()[0]
+        session = self.patient_cql_connection(node)
+
+        # Create some thousands of rows to guarantee a long index building
+        node.stress(['write', 'n=50K', 'no-warmup'])
+        session.execute("USE keyspace1")
+
+        # Create an index and immediately drop it, without waiting for index building
+        session.execute('CREATE INDEX idx ON standard1("C0")')
+        session.execute('DROP INDEX idx')
+        cluster.wait_for_compactions()
+
+        # Check that the index is not marked as built nor queryable
+        assert_none(session, """SELECT * FROM system."IndexInfo" WHERE table_name='keyspace1'""")
+        assert_invalid(session,
+                       'SELECT * FROM standard1 WHERE "C0" = 0x00',
+                       'Cannot execute this query as it might involve data filtering')
+
+        # Restart the node to trigger any eventual unexpected index rebuild
+        node.nodetool('drain')
+        node.stop()
+        cluster.start()
+        session = self.patient_cql_connection(node)
+        session.execute("USE keyspace1")
+
+        # The index should remain not built nor queryable after restart
+        assert_none(session, """SELECT * FROM system."IndexInfo" WHERE table_name='keyspace1'""")
+        assert_invalid(session,
+                       'SELECT * FROM standard1 WHERE "C0" = 0x00',
+                       'Cannot execute this query as it might involve data filtering')
+
+    @since('4.0')
+    def test_index_is_not_always_rebuilt_at_start(self):
+        """
+        @jira_ticket CASSANDRA-10130
+
+        Tests the management of index status during manual index rebuilding failures.
+        """
+
+        cluster = self.cluster
+        cluster.populate(1, install_byteman=True).start(wait_for_binary_proto=True)
+        node = cluster.nodelist()[0]
+
+        session = self.patient_cql_connection(node)
+        create_ks(session, 'k', 1)
+        session.execute("CREATE TABLE k.t (k int PRIMARY KEY, v int)")
+        session.execute("CREATE INDEX idx ON k.t(v)")
+        session.execute("INSERT INTO k.t(k, v) VALUES (0, 1)")
+        session.execute("INSERT INTO k.t(k, v) VALUES (2, 3)")
+
+        # Verify that the index is marked as built and it can answer queries
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
+        # Restart the node to trigger any eventual undesired index rebuild
+        before_files = self._index_sstables_files(node, 'k', 't', 'idx')
+        node.nodetool('drain')
+        node.stop()
+        cluster.start()
+        session = self.patient_cql_connection(node)
+        session.execute("USE k")
+        after_files = self._index_sstables_files(node, 'k', 't', 'idx')
+
+        # Verify that, the index is not rebuilt, marked as built, and it can answer queries
+        self.assertNotEqual(before_files, after_files)
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 1", [0, 1])
+
     def test_multi_index_filtering_query(self):
         """
         asserts that having multiple indexes that cover all predicates still requires ALLOW
FILTERING to also be present

http://git-wip-us.apache.org/repos/asf/cassandra/blob/50e1e7b1/sstable_generation_loading_test.py
----------------------------------------------------------------------
diff --git a/sstable_generation_loading_test.py b/sstable_generation_loading_test.py
index e04d714..37ea8e3 100644
--- a/sstable_generation_loading_test.py
+++ b/sstable_generation_loading_test.py
@@ -6,7 +6,8 @@ from distutils import dir_util
 from ccmlib import common as ccmcommon
 
 from dtest import Tester, debug, create_ks, create_cf
-from tools.assertions import assert_one
+from tools.assertions import assert_all, assert_none, assert_one
+from tools.decorators import since
 
 
 # WARNING: sstableloader tests should be added to TestSSTableGenerationAndLoading (below),
@@ -68,6 +69,34 @@ class BaseSStableLoaderTest(Tester):
 
         self.load_sstable_with_configuration(ks='"Keyspace1"', create_schema=create_schema_with_mv)
 
+    def copy_sstables(self, cluster, node):
+        for x in xrange(0, cluster.data_dir_count):
+            data_dir = os.path.join(node.get_path(), 'data{0}'.format(x))
+            copy_root = os.path.join(node.get_path(), 'data{0}_copy'.format(x))
+            for ddir in os.listdir(data_dir):
+                keyspace_dir = os.path.join(data_dir, ddir)
+                if os.path.isdir(keyspace_dir) and ddir != 'system':
+                    copy_dir = os.path.join(copy_root, ddir)
+                    dir_util.copy_tree(keyspace_dir, copy_dir)
+
+    def load_sstables(self, cluster, node, ks):
+        cdir = node.get_install_dir()
+        sstableloader = os.path.join(cdir, 'bin', ccmcommon.platform_binary('sstableloader'))
+        env = ccmcommon.make_cassandra_env(cdir, node.get_path())
+        host = node.address()
+        for x in xrange(0, cluster.data_dir_count):
+            sstablecopy_dir = os.path.join(node.get_path(), 'data{0}_copy'.format(x), ks.strip('"'))
+            for cf_dir in os.listdir(sstablecopy_dir):
+                full_cf_dir = os.path.join(sstablecopy_dir, cf_dir)
+                if os.path.isdir(full_cf_dir):
+                    cmd_args = [sstableloader, '--nodes', host, full_cf_dir]
+                    p = subprocess.Popen(cmd_args, stderr=subprocess.PIPE, stdout=subprocess.PIPE,
env=env)
+                    exit_status = p.wait()
+                    debug('stdout: {out}'.format(out=p.stdout))
+                    debug('stderr: {err}'.format(err=p.stderr))
+                    self.assertEqual(0, exit_status,
+                                     "sstableloader exited with a non-zero status: {}".format(exit_status))
+
     def load_sstable_with_configuration(self, pre_compression=None, post_compression=None,
ks="ks", create_schema=create_schema):
         """
         tests that the sstableloader works by using it to load data.
@@ -112,14 +141,7 @@ class BaseSStableLoaderTest(Tester):
 
         debug("Making a copy of the sstables")
         # make a copy of the sstables
-        for x in xrange(0, cluster.data_dir_count):
-            data_dir = os.path.join(node1.get_path(), 'data{0}'.format(x))
-            copy_root = os.path.join(node1.get_path(), 'data{0}_copy'.format(x))
-            for ddir in os.listdir(data_dir):
-                keyspace_dir = os.path.join(data_dir, ddir)
-                if os.path.isdir(keyspace_dir) and ddir != 'system':
-                    copy_dir = os.path.join(copy_root, ddir)
-                    dir_util.copy_tree(keyspace_dir, copy_dir)
+        self.copy_sstables(cluster, node1)
 
         debug("Wiping out the data and restarting cluster")
         # wipe out the node data.
@@ -140,22 +162,7 @@ class BaseSStableLoaderTest(Tester):
 
         debug("Calling sstableloader")
         # call sstableloader to re-load each cf.
-        cdir = node1.get_install_dir()
-        sstableloader = os.path.join(cdir, 'bin', ccmcommon.platform_binary('sstableloader'))
-        env = ccmcommon.make_cassandra_env(cdir, node1.get_path())
-        host = node1.address()
-        for x in xrange(0, cluster.data_dir_count):
-            sstablecopy_dir = os.path.join(node1.get_path(), 'data{0}_copy'.format(x), ks.strip('"'))
-            for cf_dir in os.listdir(sstablecopy_dir):
-                full_cf_dir = os.path.join(sstablecopy_dir, cf_dir)
-                if os.path.isdir(full_cf_dir):
-                    cmd_args = [sstableloader, '--nodes', host, full_cf_dir]
-                    p = subprocess.Popen(cmd_args, stderr=subprocess.PIPE, stdout=subprocess.PIPE,
env=env)
-                    exit_status = p.wait()
-                    debug('stdout: {out}'.format(out=p.stdout))
-                    debug('stderr: {err}'.format(err=p.stderr))
-                    self.assertEqual(0, exit_status,
-                                     "sstableloader exited with a non-zero status: {}".format(exit_status))
+        self.load_sstables(cluster, node1, ks)
 
         def read_and_validate_data(session):
             for i in range(NUM_KEYS):
@@ -287,3 +294,68 @@ class TestSSTableGenerationAndLoading(BaseSStableLoaderTest):
                             "PRIMARY KEY (v)")
 
         self.load_sstable_with_configuration(ks='"Keyspace1"', create_schema=create_schema_with_mv)
+
+    @since('4.0')
+    def sstableloader_with_failing_2i_test(self):
+        """
+        @jira_ticket CASSANDRA-10130
+
+        Simulates an index building failure during SSTables load.
+        The table data should be loaded and the index should be marked for rebuilding during
the next node start.
+        """
+        def create_schema_with_2i(session):
+            create_ks(session, 'k', 1)
+            session.execute("CREATE TABLE k.t (p int, c int, v int, PRIMARY KEY(p, c))")
+            session.execute("CREATE INDEX idx ON k.t(v)")
+
+        cluster = self.cluster
+        cluster.populate(1, install_byteman=True).start(wait_for_binary_proto=True)
+        node = cluster.nodelist()[0]
+
+        session = self.patient_cql_connection(node)
+        create_schema_with_2i(session)
+        session.execute("INSERT INTO k.t(p, c, v) VALUES (0, 1, 8)")
+
+        # Stop node and copy SSTables
+        node.nodetool('drain')
+        node.stop()
+        self.copy_sstables(cluster, node)
+
+        # Wipe out data and restart
+        cluster.clear()
+        cluster.start()
+
+        # Restore the schema
+        session = self.patient_cql_connection(node)
+        create_schema_with_2i(session)
+
+        # The table should exist and be empty, and the index should be empty and marked as
built
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_none(session, "SELECT * FROM k.t")
+        assert_none(session, "SELECT * FROM k.t WHERE v = 8")
+
+        # Add some additional data before loading the SSTable, to check that it will be still
accessible
+        session.execute("INSERT INTO k.t(p, c, v) VALUES (0, 2, 8)")
+        assert_one(session, "SELECT * FROM k.t", [0, 2, 8])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 8", [0, 2, 8])
+
+        # Load SSTables with a failure during index creation
+        node.byteman_submit(['./byteman/index_build_failure.btm'])
+        with self.assertRaises(Exception):
+            self.load_sstables(cluster, node, 'k')
+
+        # Check that the index isn't marked as built and the old SSTable data has been loaded
but not indexed
+        assert_none(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""")
+        assert_all(session, "SELECT * FROM k.t", [[0, 1, 8], [0, 2, 8]])
+        assert_one(session, "SELECT * FROM k.t WHERE v = 8", [0, 2, 8])
+
+        # Restart the node to trigger index rebuild
+        node.nodetool('drain')
+        node.stop()
+        cluster.start()
+        session = self.patient_cql_connection(node)
+
+        # Check that the index is marked as built and the index has been rebuilt
+        assert_one(session, """SELECT * FROM system."IndexInfo" WHERE table_name='k'""",
['k', 'idx'])
+        assert_all(session, "SELECT * FROM k.t", [[0, 1, 8], [0, 2, 8]])
+        assert_all(session, "SELECT * FROM k.t WHERE v = 8", [[0, 1, 8], [0, 2, 8]])


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message