<?php
class Iterator_Cassandra
{

    protected $initialRows;

    public function __construct(Cassandra\Rows $rows) {
        $this->initialRows = $rows;
    }

    public function count() {
        // Paginate and reset
        $startMem = memory_get_usage();
        $count = $this->initialRows->count();
        while ($this->initialRows = $this->initialRows->nextPage()) {
            if ($count % 1000 == 0) {
                $endMem = memory_get_usage();
                $memUsage = $endMem - $startMem;
                print "Memory usage = Delta: $memUsage, Start: $startMem, End: $endMem\n";
            }
            $count += $this->initialRows->count();
        }

        return (int)$count;
    }
}


$cluster = Cassandra::cluster()->build();
$session = $cluster->connect();

$session->execute(new Cassandra\SimpleStatement(
    "CREATE KEYSPACE IF NOT EXISTS totalsocial
     WITH replication = { 'class': 'SimpleStrategy', 'replication_factor': '3' }"));

$session->execute(new Cassandra\SimpleStatement(
    "CREATE TYPE IF NOT EXISTS totalsocial.field_totalsocial_score (
        \"v\" float,
        \"m\" float,
        \"b\" float
    )"
));

$session->execute(new Cassandra\SimpleStatement(
    "CREATE TYPE IF NOT EXISTS totalsocial.field_totalsocial_sentiment (
        \"n\" float,
        \"p\" float,
        \"ne\" float,
        \"m\" float
    );"
));

$session->execute(new Cassandra\SimpleStatement(
    "CREATE TABLE IF NOT EXISTS totalsocial.\"totalsocial_report_talktrack_week\" (
        \"i\" text,
        \"iv\" text,
        \"sd\" timestamp,
        \"ed\" timestamp,
        \"ms\" double,
        \"is\" double,
        \"tm\" double,
        \"ti\" double,
        \"ts\" double,
        \"sv\" frozen<field_totalsocial_score>,
        \"st\" frozen<field_totalsocial_sentiment>,
        \"rs\" double,
        \"ss\" frozen<field_totalsocial_score>,
        \"mmr\" double,
        \"mmb\" double,
        \"rbs\" double,
        \"sbs\" frozen<field_totalsocial_score>,
        \"cs\" double,
        \"tc\" double,
        \"pc\" double,
        \"us\" double,
        \"tu\" double,
        \"pu\" double,
        \"ri\" double,
        \"si\" frozen<field_totalsocial_score>,
        PRIMARY KEY (\"i\", \"iv\", \"sd\")
    );"
));

$tableName = "totalsocial.totalsocial_report_talktrack_week";

$scoreType = Cassandra\Type::userType(
    "v", Cassandra\Type::float(),
    "m", Cassandra\Type::float(),
    "b", Cassandra\Type::float()
);

$sentimentType = Cassandra\Type::userType(
    "n", Cassandra\Type::float(),
    "p", Cassandra\Type::float(),
    "ne", Cassandra\Type::float(),
    "m", Cassandra\Type::float()
);

$query = "SELECT * FROM $tableName LIMIT 1";
$statement = new Cassandra\SimpleStatement($query);
$rows = $session->execute($statement);

if ($rows->count() === 0) {
    print "Populating table...";
    $query = "INSERT INTO $tableName
        (i, iv, sd, ed, ms,
         \"is\", tm, ti, ts, sv,
         st, rs, ss, mmr, mmb,
         rbs, sbs, cs, tc, pc,
         us, tu, pu, ri, si) VALUES
        (?, ?, ?, ?, ?,
         ?, ?, ?, ?, ?,
         ?, ?, ?, ?, ?,
         ?, ?, ?, ?, ?,
         ?, ?, ?, ?, ?)";
    for ($i = 0; $i < 100000; $i++) {
        $statement = new Cassandra\SimpleStatement($query);

        $i = $iv = "$i";
        $sd = $ed = new Cassandra\Timestamp();
        $ms = $is = $tm = $ti = $ts = 1.0;
        $sv = $scoreType->create(
            "v", new Cassandra\Float(2.0),
            "m", new Cassandra\Float(3.0),
            "b", new Cassandra\Float(4.0)
        );
        $st = $sentimentType->create(
            "n", new Cassandra\Float(5.0),
            "p", new Cassandra\Float(6.0),
            "ne", new Cassandra\Float(7.0),
            "m", new Cassandra\Float(8.0)
        );
        $rs = 9.0;
        $ss = $scoreType->create(
            "v", new Cassandra\Float(10.0),
            "m", new Cassandra\Float(11.0),
            "b", new Cassandra\Float(12.0)
        );
        $mmr = $mmb = $rbs = 13.0;
        $sbs = $scoreType->create(
            "v", new Cassandra\Float(14.0),
            "m", new Cassandra\Float(15.0),
            "b", new Cassandra\Float(16.0)
        );
        $cs = $tc = $pc = $us = $tu = $pu = $ri = 17.0;
        $si = $scoreType->create(
            "v", new Cassandra\Float(18.0),
            "m", new Cassandra\Float(19.0),
            "b", new Cassandra\Float(20.0)
        );

        $options = array('arguments' =>
            array(
                $i, $iv, $sd, $ed, $ms,
                $is, $tm, $ti, $ts, $sv,
                $st, $rs, $ss, $mmr, $mmb,
                $rbs, $sbs, $cs, $tc, $pc,
                $us, $tu, $pu, $ri, $si
        ));

        $options = new Cassandra\ExecutionOptions($options);
        $session->execute($statement, $options);
    }
    print "Done.\n";
}

print "Paging table...\n";
$query = "SELECT * FROM $tableName";
$statement = new Cassandra\SimpleStatement($query);
$options = array('page_size' => 10);
$options = new Cassandra\ExecutionOptions($options);

$startMem = memory_get_usage();
$rows    = $session->execute($statement, $options);
$iterator = new Iterator_Cassandra($rows);
$count = $iterator->count();
print "Done.\n";