Added extensions to Read and Write mode.

Added support for key prefixes

3 files changed, 771 insertions, 0 deletions

diff --git a/src/test/java/parallelai/spyglass/hbase/AllTests.java b/src/test/java/parallelai/spyglass/hbase/AllTests.java
new file mode 100644
index 0000000..e1b875f
--- /dev/null
+++ b/src/test/java/parallelai/spyglass/hbase/AllTests.java
@@ -0,0 +1,11 @@
+package parallelai.spyglass.hbase;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+import org.junit.runners.Suite.SuiteClasses;
+
+@RunWith(Suite.class)
+@SuiteClasses({ HBaseSalterTester.class })
+public class AllTests {
+
+}
diff --git a/src/test/java/parallelai/spyglass/hbase/GenerateTestingHTables.java b/src/test/java/parallelai/spyglass/hbase/GenerateTestingHTables.java
new file mode 100644
index 0000000..471e1fe
--- /dev/null
+++ b/src/test/java/parallelai/spyglass/hbase/GenerateTestingHTables.java
@@ -0,0 +1,261 @@
+package parallelai.spyglass.hbase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableExistsException;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.hfile.Compression;
+import org.apache.hadoop.hbase.regionserver.StoreFile;
+import org.apache.hadoop.hbase.util.Bytes;
+import junit.framework.Assert;
+import org.junit.Test;
+
+/**
+ * Class generates TWO tables in HBase 'TABLE_01' and 'TABLE_02'
+ * 
+ * Those tables are used by the 'integration-testing' of HBaseSource
+ * in file HBaseSourceShouldRead.scala
+ * 
+ * Run with:
+ * mvn -Dtest=bskyb.commons.hbase.skybase.GenerateTestingHTables test
+ * 
+ * @author Antwnis@gmail.com
+ */
+public class GenerateTestingHTables {
+
+	private static Configuration config = HBaseConfiguration.create();
+	private static final String QUORUM = "cldmgr.prod.bigdata.bskyb.com";
+	private static final String QUORUM_PORT = "2181";
+	private static final Long STARTING_TIMESTAMP = 1260000000000L;
+
+	public static enum TestingTable {
+		TABLE_01, TABLE_02
+	}
+
+	private static final Log LOG = LogFactory.getLog(GenerateTestingHTables.class);
+
+	@Test
+	public void fakeTest() {
+
+		// Connect to Quorum
+		LOG.info("Connecting to " + QUORUM + ":" + QUORUM_PORT);
+		config.clear();
+		config.set("hbase.zookeeper.quorum", QUORUM);
+		config.set("hbase.zookeeper.property.clientPort", QUORUM_PORT);
+
+		// Delete test tables
+		try {
+			deleteTestTable(TestingTable.TABLE_01.name());
+			deleteTestTable(TestingTable.TABLE_02.name());
+	
+			// Generate test tables
+			createTestTable(TestingTable.TABLE_01);
+			createTestTable(TestingTable.TABLE_02);
+	
+			// Populate test tables
+			populateTestTable(TestingTable.TABLE_01);
+			populateTestTable(TestingTable.TABLE_02);
+	
+			// Print content of test table
+			printHTable(TestingTable.TABLE_01);
+			
+			// If we've reached here - the testing data are in
+			Assert.assertEquals("true", "true");
+		} catch (IOException e) {
+			LOG.error(e.toString());
+		}
+
+	}
+
+	private static void populateTestTable(TestingTable testingTable)
+			throws IOException {
+		// Load up HBase table
+		HTable table = new HTable(config, testingTable.name());
+		
+		LOG.info("Populating table: " + testingTable.name());
+
+		// Table_01
+		if (testingTable == TestingTable.TABLE_01) {
+			Put put1 = new Put("2000-01-01 10:00:10".getBytes());
+			put1.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP, "1".getBytes());
+			Put put2 = new Put("2000-01-01 10:05:00".getBytes());
+			put2.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP, "2".getBytes());
+			Put put3 = new Put("2000-01-01 10:10:00".getBytes());
+			put3.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP, "3".getBytes());
+			table.put(put1);
+			table.put(put2);
+			table.put(put3);
+		} else
+		// Table_02
+		if (testingTable == TestingTable.TABLE_02) {
+
+			// 3 versions at 10 o'clock
+			byte[] k1 = "2000-01-01 10:00:00".getBytes();
+			Put put1 = new Put(k1);
+			put1.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP        , "1".getBytes());
+			Put put2 = new Put(k1);
+			put2.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP + 1000L, "2".getBytes());
+			Put put3 = new Put(k1);
+			put3.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP + 2000L, "3".getBytes());
+
+			// 3 versions at 11 o'clock
+			byte[] k2 = "2000-01-01 11:00:00".getBytes();
+			Put put4 = new Put(k2);
+			put4.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP        , "4".getBytes());
+			Put put5 = new Put(k2);
+			put5.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP + 1000L, "5".getBytes());
+			Put put6 = new Put(k2);
+			put6.add("data".getBytes(), "column1".getBytes(), STARTING_TIMESTAMP + 2000L, "6".getBytes());
+
+			// Generate list of puts
+			List<Put> puts = new ArrayList<Put>();
+			
+			puts.add(put1);
+			puts.add(put2);
+			puts.add(put3);
+			puts.add(put4);
+			puts.add(put5);
+			puts.add(put6);
+			
+			table.put(puts);
+		}
+
+		table.close();
+	}
+
+	private static void createTestTable(TestingTable testingTable)
+			throws IOException {
+
+		// Reset configuration
+		config.clear();
+		config.set("hbase.zookeeper.quorum", QUORUM);
+		config.set("hbase.zookeeper.property.clientPort", QUORUM_PORT);
+
+		HBaseAdmin hbase = new HBaseAdmin(config);
+
+		// Get and set the name of the new table
+		String tableName = testingTable.name();
+		HTableDescriptor newTable = new HTableDescriptor(tableName);
+
+		// Table1
+		if (testingTable == TestingTable.TABLE_01) {
+			HColumnDescriptor meta = new HColumnDescriptor("data");
+			meta.setMaxVersions(3)
+ 		        .setCompressionType(Compression.Algorithm.NONE)
+			    .setInMemory(HColumnDescriptor.DEFAULT_IN_MEMORY)
+				.setBlockCacheEnabled(HColumnDescriptor.DEFAULT_BLOCKCACHE)
+				.setTimeToLive(HColumnDescriptor.DEFAULT_TTL)
+				.setBloomFilterType(StoreFile.BloomType.NONE);
+
+			newTable.addFamily(meta);
+			// Table2
+		} else if (testingTable == TestingTable.TABLE_02) {
+			HColumnDescriptor meta = new HColumnDescriptor("data".getBytes());
+			meta.setMaxVersions(3)
+		        .setCompressionType(Compression.Algorithm.NONE)
+		        .setInMemory(HColumnDescriptor.DEFAULT_IN_MEMORY)
+			    .setBlockCacheEnabled(HColumnDescriptor.DEFAULT_BLOCKCACHE)
+			    .setTimeToLive(HColumnDescriptor.DEFAULT_TTL)
+			    .setBloomFilterType(StoreFile.BloomType.NONE);
+
+//			HColumnDescriptor prefix = new HColumnDescriptor("account".getBytes());
+//			newTable.addFamily(prefix);
+			newTable.addFamily(meta);
+			LOG.info("scan 'TABLE_02' , { VERSIONS => 3 }");
+		}
+
+		try {
+			LOG.info("Creating table " + tableName);
+			hbase.createTable(newTable);
+		} catch (TableExistsException et) {
+			LOG.error("TableExistsException for table: " + tableName);
+			LOG.debug(et.toString());
+		} catch (IOException e) {
+			LOG.error("IOException: " + e.toString());
+		}
+
+		hbase.close();
+	}
+
+	/**
+	 * Method to disable and delete HBase Tables i.e. "int-test-01"
+	 */
+	private static void deleteTestTable(String tableName) throws IOException {
+
+		// Reset configuration
+		config.clear();
+		config.set("hbase.zookeeper.quorum", QUORUM);
+		config.set("hbase.zookeeper.property.clientPort", QUORUM_PORT);
+
+		HBaseAdmin hbase = new HBaseAdmin(config);
+
+		if (hbase.tableExists(tableName)) {
+			LOG.info("Table: " + tableName + " exists.");
+			hbase.disableTable(tableName);
+			hbase.deleteTable(tableName);
+			LOG.info("Table: " + tableName + " disabled and deleted.");
+		} else {
+			LOG.info("Table: " + tableName + " does not exist.");
+		}
+
+		hbase.close();
+	}
+
+	/**
+	 * Method to print-out an HTable
+	 */
+	private static void printHTable(TestingTable testingTable)
+			throws IOException {
+
+		HTable table = new HTable(config, testingTable.name());
+
+		Scan s = new Scan();
+		// Let scanner know which columns we are interested in
+		ResultScanner scanner = table.getScanner(s);
+
+		LOG.info("Printing HTable: " + Bytes.toString(table.getTableName()));
+
+		try {
+			// Iterate results
+			for (Result rr = scanner.next(); rr != null; rr = scanner.next()) {
+				String key = Bytes.toString(rr.getRow());
+				Iterator<KeyValue> iter = rr.list().iterator();
+
+				String header = "Key:\t";
+				String data = key + "\t";
+
+				while (iter.hasNext()) {
+					KeyValue kv = iter.next();
+					header += Bytes.toString(kv.getFamily()) + ":"
+							+ Bytes.toString(kv.getQualifier()) + "\t";
+					data += Bytes.toString(kv.getValue()) + "\t";
+				}
+
+				LOG.info(header);
+				LOG.info(data);
+			}
+			System.out.println();
+		} finally {
+			// Make sure you close your scanners when you are done!
+			// Thats why we have it inside a try/finally clause
+			scanner.close();
+			table.close();
+		}
+	}
+
+}
\ No newline at end of file
diff --git a/src/test/java/parallelai/spyglass/hbase/HBaseSalterTester.java b/src/test/java/parallelai/spyglass/hbase/HBaseSalterTester.java
new file mode 100644
index 0000000..34d7ab4
--- /dev/null
+++ b/src/test/java/parallelai/spyglass/hbase/HBaseSalterTester.java
@@ -0,0 +1,499 @@
+package parallelai.spyglass.hbase;
+
+import static org.junit.Assert.*;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.junit.Test;
+
+import parallelai.spyglass.hbase.HBaseSalter;
+
+
+public class HBaseSalterTester {
+
+	@Test
+	public void addSaltPrefix() throws IOException {
+		String keyStr = "1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		byte [] expected = Bytes.toBytes("1_1021");
+		byte [] actual = HBaseSalter.addSaltPrefix(keyBytes);
+		
+		assertArrayEquals(actual, expected);
+		
+		String actualStr = HBaseSalter.addSaltPrefix(keyStr);
+		
+		System.out.println(Bytes.toString(expected) + " -> " + actualStr );
+		
+		assertEquals(Bytes.toString(expected), actualStr);
+
+	}
+
+	@Test
+	public void delSaltPrefix() throws IOException {
+		String keyStr = "1_1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		byte [] expected = Bytes.toBytes("1021");
+		byte [] actual = HBaseSalter.delSaltPrefix(keyBytes);
+		
+		assertArrayEquals(actual, expected);
+		
+		String actualStr = HBaseSalter.delSaltPrefix(keyStr);
+		
+		assertEquals(Bytes.toString(expected), actualStr);
+
+	}
+
+	@Test
+	public void getAllKeys() throws IOException {
+		String keyStr = "1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		
+		char [] prefixArr = HBaseSalter.DEFAULT_PREFIX_LIST.toCharArray();
+		
+		byte [][] expected = new byte[prefixArr.length][];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expected[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStr);
+		}
+		
+		byte [][] actual = HBaseSalter.getAllKeys(keyBytes);
+		
+		assertEquals(expected.length, actual.length);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+	}
+
+	@Test
+	public void getAllKeysWithPrefix() throws IOException {
+		String keyStr = "1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		String prefix = "0123456789";
+		
+		char [] prefixArr = prefix.toCharArray();
+		
+		byte [][] expected = new byte[prefixArr.length][];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expected[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStr);
+		}
+		
+		byte [][] actual = HBaseSalter.getAllKeys(keyBytes, prefix);
+		
+		assertEquals(expected.length, actual.length);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+	}
+
+	@Test
+	public void getAllKeysWithPrefixAndRange() throws IOException {
+		String keyStr = "1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		String prefix = "1234";
+		String fullPrefix = "0123456789";
+		
+		char [] prefixArr = prefix.toCharArray();
+		Byte [] prefixBytes = new Byte[prefixArr.length]; 
+		
+		byte [][] expected = new byte[prefixArr.length][];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expected[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStr);
+			prefixBytes[i] = (byte)prefixArr[i];
+		}
+		
+		byte [][] actual = HBaseSalter.getAllKeysInRange(keyBytes, fullPrefix, (byte)'1', (byte)'5');
+		
+		assertEquals(expected.length, actual.length);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+		
+		actual = HBaseSalter.getAllKeys(keyBytes, prefixBytes);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+		
+	}
+
+
+	@Test
+	public void getAllKeysWithPrefixWithStart() throws IOException {
+		String keyStr = "1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		String prefix = "3456789";
+		String fullPrefix = "0123456789";
+		
+		char [] prefixArr = prefix.toCharArray();
+		Byte [] prefixBytes = new Byte[prefixArr.length]; 
+		
+		byte [][] expected = new byte[prefixArr.length][];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expected[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStr);
+			prefixBytes[i] = (byte)prefixArr[i];
+		}
+		
+		byte [][] actual = HBaseSalter.getAllKeysWithStart(keyBytes, fullPrefix, (byte)'3');
+		
+		assertEquals(expected.length , actual.length);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+		
+		actual = HBaseSalter.getAllKeys(keyBytes, prefixBytes);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+		
+	}
+
+	@Test
+	public void getAllKeysWithPrefixWithStop() throws IOException {
+		String keyStr = "1021";
+		byte [] keyBytes = Bytes.toBytes(keyStr);
+		String prefix = "01234";
+		String fullPrefix = "0123456789";
+		
+		char [] prefixArr = prefix.toCharArray();
+		Byte [] prefixBytes = new Byte[prefixArr.length]; 
+		
+		byte [][] expected = new byte[prefixArr.length][];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expected[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStr);
+			prefixBytes[i] = (byte)prefixArr[i];
+		}
+		
+		byte [][] actual = HBaseSalter.getAllKeysWithStop(keyBytes, fullPrefix, (byte)'5');
+		
+		assertEquals(expected.length , actual.length);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+		
+		actual = HBaseSalter.getAllKeys(keyBytes, prefixBytes);
+		
+		for( int i = 0; i < expected.length; i++) {
+			assertArrayEquals(expected[i], actual[i]);
+		}
+	}
+	
+	@Test
+	public void getDistributedIntervals() throws IOException {
+		String keyStrStart = "1021";
+		byte [] keyBytesStart = Bytes.toBytes(keyStrStart);
+
+		String keyStrStop = "1022";
+		byte [] keyBytesStop = Bytes.toBytes(keyStrStop);
+
+		char [] prefixArr = HBaseSalter.DEFAULT_PREFIX_LIST.toCharArray();
+		
+		byte [][] expectedStart = new byte[prefixArr.length][];
+		byte [][] expectedStop = new byte[prefixArr.length][];
+		Pair<byte[], byte[]> expectedPairs [] = new Pair[prefixArr.length];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expectedStart[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStart);
+			expectedStop[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStop);
+			expectedPairs[i] = new Pair<byte[], byte[]>(expectedStart[i], expectedStop[i]);
+		}
+		
+		Pair<byte[], byte[]> actualPairs [] = HBaseSalter.getDistributedIntervals(keyBytesStart, keyBytesStop);
+		
+		assertEquals(expectedPairs.length, actualPairs.length);
+		
+		for( int i = 0; i < expectedPairs.length; i++ ) {
+//			System.out.println("".format("FIRST: EXPECTED: (%s) ACTUAL: (%s)", 
+//					Bytes.toString(expectedPairs[i].getFirst()), Bytes.toString(actualPairs[i].getFirst()) ));
+//
+//			System.out.println("".format("SECOND: EXPECTED: (%s) ACTUAL: (%s)", 
+//					Bytes.toString(expectedPairs[i].getSecond()), Bytes.toString(actualPairs[i].getSecond()) ));
+
+			assertArrayEquals(expectedPairs[i].getFirst(), actualPairs[i].getFirst());
+			assertArrayEquals(expectedPairs[i].getSecond(), actualPairs[i].getSecond());
+		}
+	}
+	
+	
+	@Test
+	public void getDistributedIntervalsWithPrefix() throws IOException {
+		String keyStrStart = "1021";
+		byte [] keyBytesStart = Bytes.toBytes(keyStrStart);
+
+		String keyStrStop = "1022";
+		byte [] keyBytesStop = Bytes.toBytes(keyStrStop);
+
+		String prefix = "0123";
+		char [] prefixArr = prefix.toCharArray();
+		
+		byte [][] expectedStart = new byte[prefixArr.length][];
+		byte [][] expectedStop = new byte[prefixArr.length][];
+		Pair<byte[], byte[]> expectedPairs [] = new Pair[prefixArr.length];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expectedStart[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStart);
+			expectedStop[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStop);
+			expectedPairs[i] = new Pair<byte[], byte[]>(expectedStart[i], expectedStop[i]);
+		}
+		
+		Pair<byte[], byte[]> actualPairs [] = HBaseSalter.getDistributedIntervals(keyBytesStart, keyBytesStop, prefix);
+		
+		assertEquals(expectedPairs.length, actualPairs.length);
+		
+		for( int i = 0; i < expectedPairs.length; i++ ) {
+			System.out.println("".format("FIRST: EXPECTED: (%s) ACTUAL: (%s)", 
+					Bytes.toString(expectedPairs[i].getFirst()), Bytes.toString(actualPairs[i].getFirst()) ));
+
+			System.out.println("".format("SECOND: EXPECTED: (%s) ACTUAL: (%s)", 
+					Bytes.toString(expectedPairs[i].getSecond()), Bytes.toString(actualPairs[i].getSecond()) ));
+
+			assertArrayEquals(expectedPairs[i].getFirst(), actualPairs[i].getFirst());
+			assertArrayEquals(expectedPairs[i].getSecond(), actualPairs[i].getSecond());
+		}
+	}
+	
+	@Test
+	public void getDistributedIntervalsWithRegionsStartStop() throws IOException {
+		String keyStrStart = "1021";
+		byte [] keyBytesStart = Bytes.toBytes(keyStrStart);
+
+		String keyStrStop = "1022";
+		byte [] keyBytesStop = Bytes.toBytes(keyStrStop);
+		
+		byte [] regionStart = Bytes.toBytes("1");
+		byte [] regionsStop = Bytes.toBytes("4");
+
+		String expectedPrefix = "123";
+		char [] prefixArr = expectedPrefix.toCharArray();
+		
+		byte [][] expectedStart = new byte[prefixArr.length][];
+		byte [][] expectedStop = new byte[prefixArr.length][];
+		Pair<byte[], byte[]> expectedPairs [] = new Pair[prefixArr.length];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expectedStart[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStart);
+			expectedStop[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStop);
+			expectedPairs[i] = new Pair<byte[], byte[]>(expectedStart[i], expectedStop[i]);
+		}
+		
+		Pair<byte[], byte[]> actualPairs [] = HBaseSalter.getDistributedIntervals(keyBytesStart, keyBytesStop, regionStart, regionsStop, HBaseSalter.DEFAULT_PREFIX_LIST);
+		
+		assertEquals(expectedPairs.length, actualPairs.length);
+		
+		for( int i = 0; i < expectedPairs.length; i++ ) {
+			System.out.println("".format("FIRST: EXPECTED: (%s) ACTUAL: (%s)", 
+					Bytes.toString(expectedPairs[i].getFirst()), Bytes.toString(actualPairs[i].getFirst()) ));
+
+			System.out.println("".format("SECOND: EXPECTED: (%s) ACTUAL: (%s)", 
+					Bytes.toString(expectedPairs[i].getSecond()), Bytes.toString(actualPairs[i].getSecond()) ));
+
+			assertArrayEquals(expectedPairs[i].getFirst(), actualPairs[i].getFirst());
+			assertArrayEquals(expectedPairs[i].getSecond(), actualPairs[i].getSecond());
+		}
+	}
+	
+	
+	@Test
+	public void getDistributedIntervalsWithRegionsStartStopWithPrefixAll() throws IOException {
+		System.out.println("------------ TEST 20 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+		    "1020", "1021",
+		    "1_1021", "3_1023",
+		    "12", "012345"
+				);
+
+		System.out.println("------------ TEST 21 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			    "1020", "1021",
+			    "2_1021", Bytes.toString(HConstants.EMPTY_END_ROW),
+			    "2345", "012345"
+					);
+
+		System.out.println("------------ TEST 22 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			    "1020", "1021",
+			    Bytes.toString(HConstants.EMPTY_START_ROW), "3_1023",
+			    "012", "012345"
+					);
+
+		System.out.println("------------ TEST 23 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			    "1020", "1021",
+			    Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+			    "012345", "012345"
+					);
+
+		System.out.println("------------ TEST 24 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			Bytes.toString(HConstants.EMPTY_START_ROW), "1021",
+		    "1_1021", "3_1023",
+		    "12", "012345"
+				);
+
+		System.out.println("------------ TEST 25 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			"1020", Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "1_1021", "3_1023",
+		    "12", "012345"
+				);
+
+		System.out.println("------------ TEST 26 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+				Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "1_1021", "3_1023",
+		    "12", "012345"
+				);
+
+		System.out.println("------------ TEST 27 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			Bytes.toString(HConstants.EMPTY_START_ROW), "1021",
+			Bytes.toString(HConstants.EMPTY_START_ROW), "3_1023",
+		    "012", "012345"
+				);
+
+		System.out.println("------------ TEST 28 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			"1020", Bytes.toString(HConstants.EMPTY_END_ROW),
+			Bytes.toString(HConstants.EMPTY_START_ROW), "3_1023",
+		    "012", "012345"
+				);
+
+		System.out.println("------------ TEST 29 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+			Bytes.toString(HConstants.EMPTY_START_ROW), "3_1023",
+		    "012", "012345"
+				);
+
+		System.out.println("------------ TEST 30 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			Bytes.toString(HConstants.EMPTY_START_ROW), "1021",
+		    "1_1021", Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "12345", "012345"
+				);
+
+		System.out.println("------------ TEST 31 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			"1020", Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "1_1021", Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "12345", "012345"
+				);
+
+		System.out.println("------------ TEST 32 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+				Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "1_1021", Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "12345", "012345"
+				);
+
+		System.out.println("------------ TEST 33 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			Bytes.toString(HConstants.EMPTY_START_ROW), "1021",
+			Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "012345", "012345"
+				);
+
+		System.out.println("------------ TEST 34 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			"1020", Bytes.toString(HConstants.EMPTY_END_ROW),
+			Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "012345", "012345"
+				);
+
+		System.out.println("------------ TEST 35 --------------");
+		getDistributedIntervalsWithRegionsStartStopWithPrefix(
+				Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+				Bytes.toString(HConstants.EMPTY_START_ROW), Bytes.toString(HConstants.EMPTY_END_ROW),
+		    "012345", "012345"
+				);
+
+	}
+	
+	private void getDistributedIntervalsWithRegionsStartStopWithPrefix(
+			String keyStrStart, String keyStrStop,
+			String regionStrStart, String regionStrStop,
+			String expectedPrefix, String sendPrefix) throws IOException {
+
+		byte [] keyBytesStart = Bytes.toBytes(keyStrStart);
+		byte [] keyBytesStop = Bytes.toBytes(keyStrStop);
+		
+		byte [] regionStart = Bytes.toBytes(regionStrStart);
+		byte [] regionsStop = Bytes.toBytes(regionStrStop);
+
+		char [] prefixArr = expectedPrefix.toCharArray();
+		
+		byte [][] expectedStart = new byte[prefixArr.length][];
+		byte [][] expectedStop = new byte[prefixArr.length][];
+		Pair<byte[], byte[]> expectedPairs [] = new Pair[prefixArr.length];
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expectedStart[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStart);
+			expectedStop[i] = Bytes.toBytes(prefixArr[i] + "_" + keyStrStop);
+		}
+		
+		if( Arrays.equals(keyBytesStart, HConstants.EMPTY_START_ROW)
+				&& Arrays.equals(keyBytesStop, HConstants.EMPTY_END_ROW) ) {
+			for( int i = expectedStart.length - 1; i >=1; i--) {
+				expectedStart[i] = expectedStart[i - 1];
+			}
+			
+			expectedStart[0] = HConstants.EMPTY_START_ROW;
+			expectedStop[expectedStop.length - 1] = HConstants.EMPTY_END_ROW;
+		} else if(Arrays.equals(keyBytesStart, HConstants.EMPTY_START_ROW)) {
+			for( int i = expectedStart.length - 1; i >=1; i--) {
+				expectedStart[i] = expectedStart[i - 1];
+			}
+			
+			expectedStart[0] = HConstants.EMPTY_START_ROW;
+		} else if (Arrays.equals(keyBytesStop, HConstants.EMPTY_END_ROW)) {
+			for(int i = 0; i < expectedStop.length - 1; i++ ) {
+				expectedStop[i] = expectedStop[i + 1];
+			}
+			expectedStop[expectedStop.length - 1] = HConstants.EMPTY_END_ROW;
+		}
+		
+		for(int i = 0; i < prefixArr.length; i++ ) {
+			expectedPairs[i] = new Pair<byte[], byte[]>(expectedStart[i], expectedStop[i]);
+		}
+		
+		Pair<byte[], byte[]> actualPairs [] = HBaseSalter.getDistributedIntervals(keyBytesStart, keyBytesStop, regionStart, regionsStop, sendPrefix);
+		
+		for(Pair<byte[], byte[]> p : expectedPairs ) {
+			System.out.println("- EXPECTED " + Bytes.toString(p.getFirst())
+					+ " -> " + Bytes.toString(p.getSecond()));
+		}
+		
+		for(Pair<byte[], byte[]> p : actualPairs ) {
+			System.out.println("- ACTUAL " + Bytes.toString(p.getFirst())
+					+ " -> " + Bytes.toString(p.getSecond()));
+		}
+		
+		assertEquals(expectedPairs.length, actualPairs.length);
+		
+		for( int i = 0; i < expectedPairs.length; i++ ) {
+//			System.out.println("".format("FIRST: EXPECTED: (%s) ACTUAL: (%s)", 
+//					Bytes.toString(expectedPairs[i].getFirst()), Bytes.toString(actualPairs[i].getFirst()) ));
+//
+//			System.out.println("".format("SECOND: EXPECTED: (%s) ACTUAL: (%s)", 
+//					Bytes.toString(expectedPairs[i].getSecond()), Bytes.toString(actualPairs[i].getSecond()) ));
+
+			assertArrayEquals(expectedPairs[i].getFirst(), actualPairs[i].getFirst());
+			assertArrayEquals(expectedPairs[i].getSecond(), actualPairs[i].getSecond());
+		}
+	}
+	
+	
+}