CassandraPagingDemo.java

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.util.NoSuchElementException;
import java.util.Scanner;
import com.cleartrail.keyspacemigrator.management.CassandraDal;
import com.datastax.driver.core.PagingState;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.NoHostAvailableException;


public class CassandraPagingDemo
{
	public void readPaginatedData() throws Exception
	{
		boolean readdone = false;
		int totalrows = 0;
		int fetchsize = 1000;
		String keyspace = "ks";
		String table = "tab";
		
		String selectquery = "SELECT * FROM "+keyspace+"."+table+";";
		
		Statement selectstatement = new SimpleStatement(selectquery);
		selectstatement.setFetchSize(fetchsize);
		String initialpagingstate = readPagingState();
		
		// This will be absent for the first page
		if (initialpagingstate != null)
		{
			if(initialpagingstate.equals("Done"))
			{
				readdone = true;
			}
			else
			{
				selectstatement.setPagingState(PagingState.fromString(initialpagingstate));
			}
		}
		ResultSet results = null;
		boolean moredata = true;
		while(true)
		{
			if(moredata)
			{
				try
				{
					results = CassandraDal.getInstance().getSession().execute(selectstatement);
				}
				catch(NoHostAvailableException nhae)
				{
					throw nhae;
				}
				catch(Exception e)
				{
					throw e;
				}
				PagingState pagingState = results.getExecutionInfo().getPagingState();
				int remaining = results.getAvailableWithoutFetching();
				for(Row row : results)
				{
					totalrows++;
					//do something
					if(--remaining == 0)
					{
						break;
					}
				}
				System.out.println("Read count = " + totalrows);
				if(pagingState == null)
				{
					writePagingState("Done");
					moredata = false;
					System.out.println("Done!");
					readdone = true;
				}
				else
				{
					String pagingstatestring = pagingState.toString();
					writePagingState(pagingstatestring);
					selectstatement.setPagingState(pagingState);
				}
			}
			else
			{
				System.out.println("All fetched");
				break;
			}
		}
	}
	
	public String readPagingState() throws Exception
	{
		String pagingstatestring = null;
		String filename = "pagingstatedata.txt";
		try
		{
			pagingstatestring = new Scanner(new File(filename)).next();
		} 
		catch (FileNotFoundException e)
		{
			throw e;
		}
		catch(NoSuchElementException nsee)
		{
			throw nsee;
		}
		catch (Exception e)
		{
			throw e;
		}
		return pagingstatestring;
	}
	
	public void writePagingState(String pagingstate) throws IOException
	{
		String filename = "pagingstatedata.txt";
		try
		{
			File file = new File(filename);

			// if file doesnt exists, then create it
			if (!file.exists())
			{
				file.createNewFile();
			}
			else
			{
				FileWriter fw = new FileWriter(file.getAbsoluteFile());
				fw.write("");
				fw.close();
			}
			FileWriter fw = new FileWriter(file.getAbsoluteFile());
			BufferedWriter bw = new BufferedWriter(fw);
			bw.write(pagingstate);
			bw.close();
		}
		catch (IOException e)
		{
			throw e;
		}
	}
}