Apache/hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-15287 mapreduce.RowCounter returns incorrect result with binary row key inputs (Matt Warhaftig)

Browse Source
This commit is contained in:
tedyu 2016-04-16 18:11:27 -07:00
parent fd49923350
commit 7dacf9f10c
11 changed files with 77 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hbase-server/src/main/java/org/apache/hadoop/hbase/mapred/GroupingTableMap.java
View File

@ -154,6 +154,6 @@ implements TableMap<ImmutableBytesWritable,Result> {
}
sb.append(Bytes.toString(vals[i]));
}
return new ImmutableBytesWritable(Bytes.toBytes(sb.toString()));
return new ImmutableBytesWritable(Bytes.toBytesBinary(sb.toString()));
}
}

2
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/CellCounter.java
View File

@ -236,7 +236,7 @@ public class CellCounter {
String regexPattern = filterCriteria.substring(1, filterCriteria.length());
rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regexPattern));
} else {
rowFilter = new PrefixFilter(Bytes.toBytes(filterCriteria));
rowFilter = new PrefixFilter(Bytes.toBytesBinary(filterCriteria));
}
return rowFilter;
}

4
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/CopyTable.java
View File

@ -116,11 +116,11 @@ public class CopyTable extends Configured implements Tool {
}
if (startRow != null) {
scan.setStartRow(Bytes.toBytes(startRow));
scan.setStartRow(Bytes.toBytesBinary(startRow));
}
if (stopRow != null) {
scan.setStopRow(Bytes.toBytes(stopRow));
scan.setStopRow(Bytes.toBytesBinary(stopRow));
}
if(families != null) {

6
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/Export.java
View File

@ -96,10 +96,10 @@ public class Export {
s.setCacheBlocks(false);
// set Start and Stop row
if (conf.get(TableInputFormat.SCAN_ROW_START) != null) {
s.setStartRow(Bytes.toBytes(conf.get(TableInputFormat.SCAN_ROW_START)));
s.setStartRow(Bytes.toBytesBinary(conf.get(TableInputFormat.SCAN_ROW_START)));
}
if (conf.get(TableInputFormat.SCAN_ROW_STOP) != null) {
s.setStopRow(Bytes.toBytes(conf.get(TableInputFormat.SCAN_ROW_STOP)));
s.setStopRow(Bytes.toBytesBinary(conf.get(TableInputFormat.SCAN_ROW_STOP)));
}
// Set Scan Column Family
boolean raw = Boolean.parseBoolean(conf.get(RAW_SCAN));
@ -138,7 +138,7 @@ public class Export {
String regexPattern = filterCriteria.substring(1, filterCriteria.length());
exportFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(regexPattern));
} else {
exportFilter = new PrefixFilter(Bytes.toBytes(filterCriteria));
exportFilter = new PrefixFilter(Bytes.toBytesBinary(filterCriteria));
}
return exportFilter;
}

2
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/GroupingTableMapper.java
View File

@ -145,7 +145,7 @@ extends TableMapper<ImmutableBytesWritable,Result> implements Configurable {
}
sb.append(Bytes.toString(vals[i]));
}
return new ImmutableBytesWritable(Bytes.toBytes(sb.toString()));
return new ImmutableBytesWritable(Bytes.toBytesBinary(sb.toString()));
}
/**

4
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/RowCounter.java
View File

@ -132,10 +132,10 @@ public class RowCounter {
Scan scan = new Scan();
scan.setCacheBlocks(false);
if (startKey != null && !startKey.equals("")) {
scan.setStartRow(Bytes.toBytes(startKey));
scan.setStartRow(Bytes.toBytesBinary(startKey));
}
if (endKey != null && !endKey.equals("")) {
scan.setStopRow(Bytes.toBytes(endKey));
scan.setStopRow(Bytes.toBytesBinary(endKey));
}
if (sb.length() > 0) {
for (String columnName : sb.toString().trim().split(" ")) {

2
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/SimpleTotalOrderPartitioner.java
View File

@ -94,7 +94,7 @@ implements Configurable {
}
LOG.warn("Using deprecated configuration " + deprecatedKey +
" - please use static accessor methods instead.");
return Bytes.toBytes(oldStyleVal);
return Bytes.toBytesBinary(oldStyleVal);
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/TableInputFormat.java
View File

@ -129,11 +129,11 @@ implements Configurable {
scan = new Scan();
if (conf.get(SCAN_ROW_START) != null) {
scan.setStartRow(Bytes.toBytes(conf.get(SCAN_ROW_START)));
scan.setStartRow(Bytes.toBytesBinary(conf.get(SCAN_ROW_START)));
}
if (conf.get(SCAN_ROW_STOP) != null) {
scan.setStopRow(Bytes.toBytes(conf.get(SCAN_ROW_STOP)));
scan.setStopRow(Bytes.toBytesBinary(conf.get(SCAN_ROW_STOP)));
}
if (conf.get(SCAN_COLUMNS) != null) {

42
hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestCellCounter.java
View File

@ -45,8 +45,8 @@ import static org.junit.Assert.fail;
@Category(LargeTests.class)
public class TestCellCounter {
private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
private static final byte[] ROW1 = Bytes.toBytes("row1");
private static final byte[] ROW2 = Bytes.toBytes("row2");
private static final byte[] ROW1 = Bytes.toBytesBinary("\\x01row1");
private static final byte[] ROW2 = Bytes.toBytesBinary("\\x01row2");
private static final String FAMILY_A_STRING = "a";
private static final String FAMILY_B_STRING = "b";
private static final byte[] FAMILY_A = Bytes.toBytes(FAMILY_A_STRING);
@ -111,6 +111,44 @@ public class TestCellCounter {
}
/**
* Test CellCounter all data should print to output
*/
@Test(timeout = 300000)
public void testCellCounterPrefix() throws Exception {
String sourceTable = "testCellCounterPrefix";
byte[][] families = { FAMILY_A, FAMILY_B };
Table t = UTIL.createTable(Bytes.toBytes(sourceTable), families);
try {
Put p = new Put(ROW1);
p.addColumn(FAMILY_A, QUALIFIER, now, Bytes.toBytes("Data11"));
p.addColumn(FAMILY_B, QUALIFIER, now + 1, Bytes.toBytes("Data12"));
p.addColumn(FAMILY_A, QUALIFIER, now + 2, Bytes.toBytes("Data13"));
t.put(p);
p = new Put(ROW2);
p.addColumn(FAMILY_B, QUALIFIER, now, Bytes.toBytes("Dat21"));
p.addColumn(FAMILY_A, QUALIFIER, now + 1, Bytes.toBytes("Data22"));
p.addColumn(FAMILY_B, QUALIFIER, now + 2, Bytes.toBytes("Data23"));
t.put(p);
String[] args = { sourceTable, FQ_OUTPUT_DIR.toString(), ";", "\\x01row1" };
runCount(args);
FileInputStream inputStream =
new FileInputStream(OUTPUT_DIR + File.separator + "part-r-00000");
String data = IOUtils.toString(inputStream);
inputStream.close();
assertTrue(data.contains("Total Families Across all Rows" + "\t" + "2"));
assertTrue(data.contains("Total Qualifiers across all Rows" + "\t" + "2"));
assertTrue(data.contains("Total ROWS" + "\t" + "1"));
assertTrue(data.contains("b;q" + "\t" + "1"));
assertTrue(data.contains("a;q" + "\t" + "1"));
assertTrue(data.contains("row1;a;q_Versions" + "\t" + "1"));
assertTrue(data.contains("row1;b;q_Versions" + "\t" + "1"));
} finally {
t.close();
FileUtil.fullyDelete(new File(OUTPUT_DIR));
}
}
/**
* Test CellCounter with time range all data should print to output
*/

20
hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestCopyTable.java
View File

@ -106,7 +106,7 @@ public class TestCopyTable {
assertEquals(1, r.size());
assertTrue(CellUtil.matchingQualifier(r.rawCells()[0], COLUMN1));
}
t1.close();
t2.close();
TEST_UTIL.deleteTable(TABLENAME1);
@ -121,7 +121,7 @@ public class TestCopyTable {
public void testCopyTable() throws Exception {
doCopyTableTest(false);
}
/**
* Simple end-to-end test with bulkload.
*/
@ -129,16 +129,16 @@ public class TestCopyTable {
public void testCopyTableWithBulkload() throws Exception {
doCopyTableTest(true);
}
@Test
public void testStartStopRow() throws Exception {
final TableName TABLENAME1 = TableName.valueOf("testStartStopRow1");
final TableName TABLENAME2 = TableName.valueOf("testStartStopRow2");
final byte[] FAMILY = Bytes.toBytes("family");
final byte[] COLUMN1 = Bytes.toBytes("c1");
final byte[] ROW0 = Bytes.toBytes("row0");
final byte[] ROW1 = Bytes.toBytes("row1");
final byte[] ROW2 = Bytes.toBytes("row2");
final byte[] ROW0 = Bytes.toBytesBinary("\\x01row0");
final byte[] ROW1 = Bytes.toBytesBinary("\\x01row1");
final byte[] ROW2 = Bytes.toBytesBinary("\\x01row2");
Table t1 = TEST_UTIL.createTable(TABLENAME1, FAMILY);
Table t2 = TEST_UTIL.createTable(TABLENAME2, FAMILY);
@ -157,8 +157,8 @@ public class TestCopyTable {
CopyTable copy = new CopyTable(TEST_UTIL.getConfiguration());
assertEquals(
0,
copy.run(new String[] { "--new.name=" + TABLENAME2, "--startrow=row1",
"--stoprow=row2", TABLENAME1.getNameAsString() }));
copy.run(new String[] { "--new.name=" + TABLENAME2, "--startrow=\\x01row1",
"--stoprow=\\x01row2", TABLENAME1.getNameAsString() }));
// verify the data was copied into table 2
// row1 exist, row0, row2 do not exist
@ -170,11 +170,11 @@ public class TestCopyTable {
g = new Get(ROW0);
r = t2.get(g);
assertEquals(0, r.size());
g = new Get(ROW2);
r = t2.get(g);
assertEquals(0, r.size());
t1.close();
t2.close();
TEST_UTIL.deleteTable(TABLENAME1);

16
hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestImportExport.java
View File

@ -90,8 +90,9 @@ import org.mockito.stubbing.Answer;
public class TestImportExport {
private static final Log LOG = LogFactory.getLog(TestImportExport.class);
private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
private static final byte[] ROW1 = Bytes.toBytes("row1");
private static final byte[] ROW2 = Bytes.toBytes("row2");
private static final byte[] ROW1 = Bytes.toBytesBinary("\\x32row1");
private static final byte[] ROW2 = Bytes.toBytesBinary("\\x32row2");
private static final byte[] ROW3 = Bytes.toBytesBinary("\\x32row3");
private static final String FAMILYA_STRING = "a";
private static final String FAMILYB_STRING = "b";
private static final byte[] FAMILYA = Bytes.toBytes(FAMILYA_STRING);
@ -179,9 +180,17 @@ public class TestImportExport {
p.add(FAMILYA, QUAL, now+1, QUAL);
p.add(FAMILYA, QUAL, now+2, QUAL);
t.put(p);
p = new Put(ROW3);
p.add(FAMILYA, QUAL, now, QUAL);
p.add(FAMILYA, QUAL, now + 1, QUAL);
p.add(FAMILYA, QUAL, now + 2, QUAL);
t.put(p);
}
String[] args = new String[] {
// Only export row1 & row2.
"-D" + TableInputFormat.SCAN_ROW_START + "=\\x32row1",
"-D" + TableInputFormat.SCAN_ROW_STOP + "=\\x32row3",
EXPORT_TABLE,
FQ_OUTPUT_DIR,
"1000", // max number of key versions per key to export
@ -205,6 +214,9 @@ public class TestImportExport {
g.setMaxVersions();
r = t.get(g);
assertEquals(3, r.size());
g = new Get(ROW3);
r = t.get(g);
assertEquals(0, r.size());
}
}