import com.mysql.cluster.ndbj.*;
import com.mysql.cluster.ndbj.NdbOperation.LockMode;

/*
 * Setup:
 * create table email_example (
 * 		id int not null, emailaddr varchar(100),
 * 		primary key (id), key (emailaddr)) engine=ndbcluster;
 * 
 * insert into email_example values (1,'a@a.com'),(2,'a@a.com'),(3,'b@b.com'),
 *            (4,'c@c.com'),(5,'c@c.com'),(6,'c@com'),(7,'d@d.com'),(8,'d@d.com'),
 *            (9,'e@e.com'),(10,'f@f.com'),(11,'f@f.com'),(12,'f@f.com');
 */
public class Example
{
	/*
	 * Simulate the SQL query:
	 * select count(*),emailaddr from email_example
	 * group by emailaddr having count(emailaddr) > 1;
	 */
	static void getEmails(Ndb ndb, int x) throws Exception
	{
		NdbTransaction trans = ndb.startTransaction();
		/* perform a scan on the ordered index on the email address */
		NdbIndexScanOperation op = trans.getSelectIndexScanOperation("emailaddr",
				"email_example", LockMode.LM_Read);
		op.getValue("emailaddr"); /* read only the emailaddr field */
		NdbResultSet rs = op.resultData();
		trans.executeCommit();

		int emailCount = 0;
		String email = "";
		while(rs.next())
		{
			String curEmail = rs.getString("emailaddr");
			/* if we have a new email, finish the previous one */
			if(!curEmail.equals(email) && !email.equals(""))
			{
				/* print if it matches */
				if(emailCount > x)
					System.out.println("Email: " + email + ", count = " + emailCount);
				/* reset the count for the next address */
				emailCount = 0;
			}

			/* set the previous to the current value */
			email = curEmail;
			/* increment this email count */
			emailCount++;
		}

		/* after the loop we still have to check if the last address matched */
		if(emailCount > x)
			System.out.println("Email: " + email + ", count = " + emailCount);
	}

	public static void main(String args[]) throws Exception
	{
		NdbClusterConnection conn = NdbClusterConnection.create("localhost");
		conn.connect(0, 0, true);
		conn.waitUntilReady(5, 0);
		Ndb ndb = conn.createNdb("test", 1);

		System.out.println("Connected. Running query.");
		getEmails(ndb, 1);

		ndb.close();
		conn.close();
	}
}