def test_composite_pk(self):
        table = 'join_rdd_test_composite_pk'

        self.session.execute('''
            CREATE TABLE IF NOT EXISTS ''' + table + ''' (
                pk text, cc text, value text,
                primary key (pk, cc)
            )
        ''')
        self.session.execute(table % 'TRUNCATE %s')

        rows = [
           # (pk, cc, pk + '-' + cc)
           (unicode(pk), unicode(cc), unicode(pk + '-' + cc))
           for pk in string.ascii_lowercase[:3]
           for cc in (str(i) for i in range(3))
        ]

        for row in rows:
            self.session.execute(
                'INSERT INTO ' + table + ' (pk, cc, value) values (%s, %s, %s)', row
            )

        rdd = self.sc.parallelize(rows)

        joined = rdd.joinWithCassandraTable(self.keyspace, table).on('pk', 'cc')
        self.assertEqual(sorted(zip(rows, rows)), sorted(joined.map(tuple).collect()))

        joined = rdd.joinWithCassandraTable(self.keyspace, table).on('pk')
        self.assertEqual(len(rows) * sqrt(len(rows)), joined.count())


        # TODO test
        # .where()
        # .limit()
dataset/ETHPy150Open TargetHolding/pyspark-cassandra/python/pyspark_cassandra/tests.py/JoinRDDTest.test_composite_pk/swapped operands of `%`