Skip to content

Commit 2173a42

Browse files
author
Mufti, Saad
committed
Merge pull request #1 in DMP/haeinsa from multiget to hbase_1_2_0
* commit '9f64cf7aac92dfcede0611bbf73ee7e42560c1dc': add support for multiget and remove some usage of deprecated API's
2 parents 31b8b76 + 9f64cf7 commit 2173a42

11 files changed

Lines changed: 454 additions & 49 deletions

‎pom.xml‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525

2626
<groupId>kr.co.vcnc.haeinsa</groupId>
2727
<artifactId>haeinsa</artifactId>
28-
<version>1.0.6-hbase12-SNAPSHOT</version>
28+
<version>1.0.6-hbase12</version>
2929
<packaging>jar</packaging>
3030

3131
<name>haeinsa</name>
@@ -144,10 +144,10 @@
144144
<plugin>
145145
<groupId>org.apache.maven.plugins</groupId>
146146
<artifactId>maven-compiler-plugin</artifactId>
147-
<version>3.1</version>
147+
<version>3.5.1</version>
148148
<configuration>
149-
<source>1.7</source>
150-
<target>1.7</target>
149+
<source>1.8</source>
150+
<target>1.8</target>
151151
<encoding>UTF-8</encoding>
152152
</configuration>
153153
</plugin>
@@ -173,7 +173,7 @@
173173
<plugin>
174174
<groupId>org.apache.maven.plugins</groupId>
175175
<artifactId>maven-checkstyle-plugin</artifactId>
176-
<version>2.10</version>
176+
<version>2.16</version>
177177
<executions>
178178
<execution>
179179
<id>checkstyle</id>
@@ -304,7 +304,7 @@
304304
<pluginExecutionFilter>
305305
<groupId>org.apache.thrift.tools</groupId>
306306
<artifactId>maven-thrift-plugin</artifactId>
307-
<versionRange>[0.1.10,)</versionRange>
307+
<versionRange>[0.1.11,)</versionRange>
308308
<goals>
309309
<goal>compile</goal>
310310
<goal>testCompile</goal>

‎src/main/java/kr/co/vcnc/haeinsa/HaeinsaKeyValue.java‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,12 @@
1515
*/
1616
package kr.co.vcnc.haeinsa;
1717

18+
import java.util.Arrays;
1819
import java.util.Comparator;
1920

2021
import kr.co.vcnc.haeinsa.utils.NullableComparator;
2122

23+
import org.apache.hadoop.hbase.Cell;
2224
import org.apache.hadoop.hbase.KeyValue;
2325
import org.apache.hadoop.hbase.KeyValue.Type;
2426
import org.apache.hadoop.hbase.util.Bytes;
@@ -62,9 +64,12 @@ public int compare(HaeinsaKeyValue o1, HaeinsaKeyValue o2) {
6264
public HaeinsaKeyValue() {
6365
}
6466

65-
public HaeinsaKeyValue(KeyValue keyValue) {
66-
this(keyValue.getRow(), keyValue.getFamily(), keyValue.getQualifier(), keyValue.getValue(),
67-
KeyValue.Type.codeToType(keyValue.getType()));
67+
public HaeinsaKeyValue(Cell keyValue) {
68+
this(Arrays.copyOfRange(keyValue.getRowArray(), keyValue.getRowOffset(), keyValue.getRowOffset() + keyValue.getRowLength()),
69+
Arrays.copyOfRange(keyValue.getFamilyArray(), keyValue.getFamilyOffset(), keyValue.getFamilyOffset() + keyValue.getFamilyLength()),
70+
Arrays.copyOfRange(keyValue.getQualifierArray(), keyValue.getQualifierOffset(), keyValue.getQualifierOffset() + keyValue.getQualifierLength()),
71+
Arrays.copyOfRange(keyValue.getValueArray(), keyValue.getValueOffset(), keyValue.getValueOffset() + keyValue.getValueLength()),
72+
KeyValue.Type.codeToType(keyValue.getTypeByte()));
6873
}
6974

7075
public HaeinsaKeyValue(byte[] row, byte[] family, byte[] qualifier, byte[] value, Type type) {

‎src/main/java/kr/co/vcnc/haeinsa/HaeinsaResult.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import java.util.Collections;
1919
import java.util.List;
2020

21+
import org.apache.hadoop.hbase.Cell;
2122
import org.apache.hadoop.hbase.KeyValue;
2223
import org.apache.hadoop.hbase.client.Result;
2324

@@ -97,10 +98,10 @@ private static List<HaeinsaKeyValue> toHaeinsaKVs(Result result) {
9798
List<HaeinsaKeyValue> sorted = Collections.emptyList();
9899
if (!result.isEmpty()) {
99100
sorted = Lists.transform(
100-
result.list(),
101-
new Function<KeyValue, HaeinsaKeyValue>() {
101+
result.listCells(),
102+
new Function<Cell, HaeinsaKeyValue>() {
102103
@Override
103-
public HaeinsaKeyValue apply(KeyValue kv) {
104+
public HaeinsaKeyValue apply(Cell kv) {
104105
return new HaeinsaKeyValue(kv);
105106
}
106107
});

‎src/main/java/kr/co/vcnc/haeinsa/HaeinsaTable.java‎

Lines changed: 122 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,10 @@
2222

2323
import java.io.IOException;
2424
import java.nio.ByteBuffer;
25+
import java.util.ArrayList;
2526
import java.util.Collections;
2627
import java.util.Iterator;
28+
import java.util.LinkedList;
2729
import java.util.List;
2830
import java.util.Map;
2931
import java.util.Map.Entry;
@@ -115,6 +117,35 @@ private HaeinsaResult getWithoutTx(HaeinsaGet get) throws IOException {
115117
return new HaeinsaResult(result);
116118
}
117119

120+
/**
121+
* Get data from HBase without transaction.
122+
* {@link HaeinsaTransaction#commit()} to check or mutate lock column of the row scanned by this method.
123+
* This method can be used when read performance is important or strict consistency of the result is not matter.
124+
*/
125+
private HaeinsaResult[] getWithoutTx(List<HaeinsaGet> gets) throws IOException {
126+
final List<Get> hGets = new ArrayList<>(gets.size());
127+
for (final HaeinsaGet get : gets) {
128+
final Get hGet = new Get(get.getRow());
129+
for (Entry<byte[], NavigableSet<byte[]>> entry : get.getFamilyMap().entrySet()) {
130+
if (entry.getValue() == null) {
131+
hGet.addFamily(entry.getKey());
132+
} else {
133+
for (byte[] qualifier : entry.getValue()) {
134+
hGet.addColumn(entry.getKey(), qualifier);
135+
}
136+
}
137+
}
138+
hGets.add(hGet);
139+
}
140+
141+
final Result[] results = table.get(hGets);
142+
final HaeinsaResult[] hResults = new HaeinsaResult[results.length];
143+
for (int ix = 0; ix < results.length; ix++) {
144+
hResults[ix] = results[ix] == null ? null : new HaeinsaResult(results[ix]);
145+
}
146+
return hResults;
147+
}
148+
118149
@Override
119150
public HaeinsaResult get(@Nullable HaeinsaTransaction tx, HaeinsaGet get) throws IOException {
120151
Preconditions.checkNotNull(get);
@@ -177,6 +208,83 @@ public HaeinsaResult get(@Nullable HaeinsaTransaction tx, HaeinsaGet get) throws
177208
return hResult;
178209
}
179210

211+
@Override
212+
public HaeinsaResult[] get(@Nullable HaeinsaTransaction tx, List<HaeinsaGet> gets) throws IOException {
213+
Preconditions.checkNotNull(gets);
214+
if (tx == null) {
215+
return getWithoutTx(gets);
216+
}
217+
218+
final HaeinsaTableTransaction tableState = tx.createOrGetTableState(this.table.getName().getName());
219+
final List<Get> hGets = new LinkedList<>();
220+
221+
for (final HaeinsaGet get : gets) {
222+
final byte[] row = get.getRow();
223+
final HaeinsaRowTransaction rowState = tableState.getRowStates().get(row);
224+
Get hGet = new Get(get.getRow());
225+
hGet.setCacheBlocks(get.getCacheBlocks());
226+
hGets.add(hGet);
227+
228+
for (Entry<byte[], NavigableSet<byte[]>> entry : get.getFamilyMap().entrySet()) {
229+
if (entry.getValue() == null) {
230+
hGet.addFamily(entry.getKey());
231+
} else {
232+
for (byte[] qualifier : entry.getValue()) {
233+
hGet.addColumn(entry.getKey(), qualifier);
234+
}
235+
}
236+
}
237+
238+
if (rowState == null) {
239+
if (hGet.hasFamilies()) {
240+
hGet.addColumn(LOCK_FAMILY, LOCK_QUALIFIER);
241+
}
242+
}
243+
}
244+
245+
Result[] results = table.get(hGets);
246+
final HaeinsaResult[] hResults = new HaeinsaResult[results.length];
247+
248+
for (int ix = 0; ix < results.length; ix++) {
249+
if (results[ix] == null) {
250+
hResults[ix] = null;
251+
continue;
252+
}
253+
254+
HaeinsaRowTransaction rowState = tableState.getRowStates().get(gets.get(ix).getRow());
255+
List<HaeinsaKeyValueScanner> scanners = Lists.newArrayList();
256+
if (rowState != null) {
257+
scanners.addAll(rowState.getScanners());
258+
}
259+
scanners.add(new HBaseGetScanner(results[ix], Long.MAX_VALUE));
260+
261+
HaeinsaResult hResult = null;
262+
// Scanners at this moment is:
263+
// union( muationScanners from RowTransaction, Scanner of get)
264+
try (ClientScanner scanner = new ClientScanner(tx, scanners, gets.get(ix).getFamilyMap(), rowState == null)) {
265+
hResult = scanner.next();
266+
}
267+
if (hResult == null) {
268+
/*
269+
* if specific row is empty and there was no puts at all, initialize ClientScanner make empty scanners
270+
* variable.
271+
* There will be no rowState associated to the row, and transaction will not operate normally.
272+
* Therefore, create rowState if there was no HBase operation accessed to the row before.
273+
*/
274+
rowState = tableState.createOrGetRowState(gets.get(ix).getRow());
275+
if (rowState.getCurrent() == null) {
276+
rowState.setCurrent(TRowLocks.deserialize(null));
277+
}
278+
279+
List<HaeinsaKeyValue> emptyList = Collections.emptyList();
280+
hResult = new HaeinsaResult(emptyList);
281+
}
282+
283+
hResults[ix] = hResult;
284+
}
285+
return hResults;
286+
}
287+
180288
@Override
181289
public HaeinsaResultScanner getScanner(@Nullable HaeinsaTransaction tx, byte[] family) throws IOException {
182290
Preconditions.checkNotNull(family);
@@ -507,10 +615,10 @@ public void commitSingleRowPutOnly(HaeinsaRowTransaction rowState, byte[] row) t
507615
Put put = new Put(row);
508616
HaeinsaPut haeinsaPut = (HaeinsaPut) rowState.getMutations().remove(0);
509617
for (HaeinsaKeyValue kv : Iterables.concat(haeinsaPut.getFamilyMap().values())) {
510-
put.add(kv.getFamily(), kv.getQualifier(), tx.getCommitTimestamp(), kv.getValue());
618+
put.addColumn(kv.getFamily(), kv.getQualifier(), tx.getCommitTimestamp(), kv.getValue());
511619
}
512620
TRowLock newRowLock = new TRowLock(ROW_LOCK_VERSION, TRowLockState.STABLE, tx.getCommitTimestamp());
513-
put.add(LOCK_FAMILY, LOCK_QUALIFIER, tx.getCommitTimestamp(), TRowLocks.serialize(newRowLock));
621+
put.addColumn(LOCK_FAMILY, LOCK_QUALIFIER, tx.getCommitTimestamp(), TRowLocks.serialize(newRowLock));
514622

515623
byte[] currentRowLockBytes = TRowLocks.serialize(rowState.getCurrent());
516624
if (!table.checkAndPut(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, put)) {
@@ -554,7 +662,7 @@ public void prewrite(HaeinsaRowTransaction rowState, byte[] row, boolean isPrima
554662
if (rowState.getMutations().get(0) instanceof HaeinsaPut) {
555663
HaeinsaPut haeinsaPut = (HaeinsaPut) rowState.getMutations().remove(0);
556664
for (HaeinsaKeyValue kv : Iterables.concat(haeinsaPut.getFamilyMap().values())) {
557-
put.add(kv.getFamily(), kv.getQualifier(), tx.getPrewriteTimestamp(), kv.getValue());
665+
put.addColumn(kv.getFamily(), kv.getQualifier(), tx.getPrewriteTimestamp(), kv.getValue());
558666
TCellKey cellKey = new TCellKey();
559667
cellKey.setFamily(kv.getFamily());
560668
cellKey.setQualifier(kv.getQualifier());
@@ -587,7 +695,7 @@ public void prewrite(HaeinsaRowTransaction rowState, byte[] row, boolean isPrima
587695
newRowLock.setPrewritten(Lists.newArrayList(prewritten));
588696
newRowLock.setMutations(remaining);
589697
newRowLock.setExpiry(tx.getExpiry());
590-
put.add(LOCK_FAMILY, LOCK_QUALIFIER, tx.getPrewriteTimestamp(), TRowLocks.serialize(newRowLock));
698+
put.addColumn(LOCK_FAMILY, LOCK_QUALIFIER, tx.getPrewriteTimestamp(), TRowLocks.serialize(newRowLock));
591699

592700
byte[] currentRowLockBytes = TRowLocks.serialize(rowState.getCurrent());
593701

@@ -631,9 +739,9 @@ public void applyMutations(HaeinsaRowTransaction rowTxState, byte[] row) throws
631739
// Maintain prewritten state and extend lock by ROW_LOCK_TIMEOUT
632740
newRowLock.setExpiry(tx.getExpiry());
633741
Put put = new Put(row);
634-
put.add(LOCK_FAMILY, LOCK_QUALIFIER, newRowLock.getCurrentTimestamp(), TRowLocks.serialize(newRowLock));
742+
put.addColumn(LOCK_FAMILY, LOCK_QUALIFIER, newRowLock.getCurrentTimestamp(), TRowLocks.serialize(newRowLock));
635743
for (TKeyValue kv : mutation.getPut().getValues()) {
636-
put.add(kv.getKey().getFamily(), kv.getKey().getQualifier(), newRowLock.getCurrentTimestamp(), kv.getValue());
744+
put.addColumn(kv.getKey().getFamily(), kv.getKey().getQualifier(), newRowLock.getCurrentTimestamp(), kv.getValue());
637745
}
638746
if (!table.checkAndPut(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, put)) {
639747
// Consider as conflict because another transaction might acquire lock of this row.
@@ -647,12 +755,12 @@ public void applyMutations(HaeinsaRowTransaction rowTxState, byte[] row) throws
647755
Delete delete = new Delete(row);
648756
if (mutation.getRemove().getRemoveFamiliesSize() > 0) {
649757
for (ByteBuffer removeFamily : mutation.getRemove().getRemoveFamilies()) {
650-
delete.deleteFamily(removeFamily.array(), mutationTimestamp);
758+
delete.addFamily(removeFamily.array(), mutationTimestamp);
651759
}
652760
}
653761
if (mutation.getRemove().getRemoveCellsSize() > 0) {
654762
for (TCellKey removeCell : mutation.getRemove().getRemoveCells()) {
655-
delete.deleteColumns(removeCell.getFamily(), removeCell.getQualifier(), mutationTimestamp);
763+
delete.addColumns(removeCell.getFamily(), removeCell.getQualifier(), mutationTimestamp);
656764
}
657765
}
658766
if (!table.checkAndDelete(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, delete)) {
@@ -676,7 +784,7 @@ public void makeStable(HaeinsaRowTransaction rowTxState, byte[] row) throws IOEx
676784
TRowLock newRowLock = new TRowLock(ROW_LOCK_VERSION, TRowLockState.STABLE, commitTimestamp);
677785
byte[] newRowLockBytes = TRowLocks.serialize(newRowLock);
678786
Put put = new Put(row);
679-
put.add(LOCK_FAMILY, LOCK_QUALIFIER, commitTimestamp, newRowLockBytes);
787+
put.addColumn(LOCK_FAMILY, LOCK_QUALIFIER, commitTimestamp, newRowLockBytes);
680788

681789
if (!table.checkAndPut(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, put)) {
682790
// Consider as success because another transaction might already stabilize this row.
@@ -705,7 +813,7 @@ public void commitPrimary(HaeinsaRowTransaction rowTxState, byte[] row) throws I
705813

706814
byte[] newRowLockBytes = TRowLocks.serialize(newRowLock);
707815
Put put = new Put(row);
708-
put.add(LOCK_FAMILY, LOCK_QUALIFIER, newRowLock.getCurrentTimestamp(), newRowLockBytes);
816+
put.addColumn(LOCK_FAMILY, LOCK_QUALIFIER, newRowLock.getCurrentTimestamp(), newRowLockBytes);
709817

710818
if (!table.checkAndPut(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, put)) {
711819
// We don't need abort current transaction. Because the transaction is already aborted.
@@ -746,7 +854,7 @@ public void abortPrimary(HaeinsaRowTransaction rowTxState, byte[] row) throws IO
746854

747855
byte[] newRowLockBytes = TRowLocks.serialize(newRowLock);
748856
Put put = new Put(row);
749-
put.add(LOCK_FAMILY, LOCK_QUALIFIER, newRowLock.getCurrentTimestamp(), newRowLockBytes);
857+
put.addColumn(LOCK_FAMILY, LOCK_QUALIFIER, newRowLock.getCurrentTimestamp(), newRowLockBytes);
750858

751859
if (!table.checkAndPut(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, put)) {
752860
// Consider as conflict because another transaction might acquire lock of primary row.
@@ -767,7 +875,7 @@ public void deletePrewritten(HaeinsaRowTransaction rowTxState, byte[] row) throw
767875
rowTxState.getCurrent().getPrewriteTimestamp() : rowTxState.getCurrent().getCurrentTimestamp();
768876
Delete delete = new Delete(row);
769877
for (TCellKey cellKey : rowTxState.getCurrent().getPrewritten()) {
770-
delete.deleteColumn(cellKey.getFamily(), cellKey.getQualifier(), prewriteTimestamp);
878+
delete.addColumn(cellKey.getFamily(), cellKey.getQualifier(), prewriteTimestamp);
771879
}
772880
if (!table.checkAndDelete(row, LOCK_FAMILY, LOCK_QUALIFIER, currentRowLockBytes, delete)) {
773881
// Consider as conflict because another transaction might acquire lock of this row.
@@ -1204,7 +1312,7 @@ public HaeinsaKeyValue peek() {
12041312
}
12051313
// First scan or next() was called last time so move
12061314
// resultIndex.
1207-
current = new HaeinsaKeyValue(currentResult.raw()[resultIndex]);
1315+
current = new HaeinsaKeyValue(currentResult.rawCells()[resultIndex]);
12081316
resultIndex++;
12091317

12101318
return current;
@@ -1304,7 +1412,7 @@ public HaeinsaKeyValue peek() {
13041412
if (result == null) {
13051413
return null;
13061414
}
1307-
current = new HaeinsaKeyValue(result.list().get(resultIndex));
1415+
current = new HaeinsaKeyValue(result.listCells().get(resultIndex));
13081416
resultIndex++;
13091417
return current;
13101418
}

‎src/main/java/kr/co/vcnc/haeinsa/HaeinsaTableIface.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,30 @@ public interface HaeinsaTableIface extends Closeable {
5757
*/
5858
HTableDescriptor getTableDescriptor() throws IOException;
5959

60+
/**
61+
* Extracts certain cells from a given row.
62+
*
63+
* @param tx HaeinsaTransaction which this operation is participated in.
64+
* It can be null if user don't want to execute get inside transaction.
65+
* @param gets The list of objects that specify what data to fetch and from which row.
66+
* @return The list of results with data coming from the specified row, if it exists.
67+
* If the row specified doesn't exist, the {@link HaeinsaResult} instance returned
68+
* won't contain any {@link HaeinsaKeyValue}, as indicated by {@link HaeinsaResult#isEmpty()}.
69+
* @throws IOException if a remote or network exception occurs.
70+
*/
71+
default HaeinsaResult[] get(@Nullable HaeinsaTransaction tx, List<HaeinsaGet> gets) throws IOException {
72+
if (gets.isEmpty()) {
73+
return null;
74+
}
75+
76+
final HaeinsaResult[] results = new HaeinsaResult[gets.size()];
77+
for (int ix = 0; ix < gets.size(); ix++) {
78+
results[ix] = get(tx, gets.get(ix));
79+
}
80+
81+
return results;
82+
}
83+
6084
/**
6185
* Extracts certain cells from a given row.
6286
*

0 commit comments

Comments
 (0)