Skip to content

Commit 997e4da

Browse files
apurtellbusbey
authored and
Huaxiang Sun
committed
HBASE-23723 Ensure MOB compaction works in optimized mode after snapshot clone (apache#4617)
* HBASE-23723 Ensure MOB compaction works in optimized mode after snapshot clone (apache#1446) * Reorganize MOB compaction tests for more reuse. * Add tests for mob compaction after snapshot clone operations * note the original table used to write a given mob hfile and use that to find it later. Signed-off-by: Esteban Gutierrez <[email protected]> * spotless:apply to fix HBaseTestingUtility * Fix error-prone errors Signed-off-by: Esteban Gutierrez <[email protected]> Co-authored-by: Sean Busbey <[email protected]>
1 parent da472aa commit 997e4da

File tree

16 files changed

+739
-220
lines changed

16 files changed

+739
-220
lines changed

‎hbase-common/src/main/java/org/apache/hadoop/hbase/PrivateCellUtil.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -916,7 +916,7 @@ public static List<Tag> getTags(Cell cell) {
916916
* Retrieve Cell's first tag, matching the passed in type
917917
* @param cell The Cell
918918
* @param type Type of the Tag to retrieve
919-
* @return null if there is no tag of the passed in tag type
919+
* @return Optional, empty if there is no tag of the passed in tag type
920920
*/
921921
public static Optional<Tag> getTag(Cell cell, byte type) {
922922
boolean bufferBacked = cell instanceof ByteBufferExtendedCell;

‎hbase-common/src/main/java/org/apache/hadoop/hbase/TableName.java‎

Lines changed: 52 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,12 @@
2222
import java.util.Arrays;
2323
import java.util.Set;
2424
import java.util.concurrent.CopyOnWriteArraySet;
25+
import org.apache.commons.lang3.ArrayUtils;
2526
import org.apache.hadoop.hbase.util.Bytes;
2627
import org.apache.yetus.audience.InterfaceAudience;
2728

29+
import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
30+
2831
/**
2932
* Immutable POJO class for representing a table name. Which is of the form: &lt;table
3033
* namespace&gt;:&lt;table qualifier&gt; Two special namespaces: 1. hbase - system namespace, used
@@ -386,28 +389,69 @@ public static TableName valueOf(String namespaceAsString, String qualifierAsStri
386389
}
387390

388391
/**
392+
* @param fullName will use the entire byte array
389393
* @throws IllegalArgumentException if fullName equals old root or old meta. Some code depends on
390394
* this. The test is buried in the table creation to save on
391395
* array comparison when we're creating a standard table object
392396
* that will be in the cache.
393397
*/
394398
public static TableName valueOf(byte[] fullName) throws IllegalArgumentException {
399+
return valueOf(fullName, 0, fullName.length);
400+
}
401+
402+
/**
403+
* @param fullName byte array to look into
404+
* @param offset within said array
405+
* @param length within said array
406+
* @throws IllegalArgumentException if fullName equals old root or old meta.
407+
*/
408+
public static TableName valueOf(byte[] fullName, int offset, int length)
409+
throws IllegalArgumentException {
410+
Preconditions.checkArgument(offset >= 0, "offset must be non-negative but was %s", offset);
411+
Preconditions.checkArgument(offset < fullName.length, "offset (%s) must be < array length (%s)",
412+
offset, fullName.length);
413+
Preconditions.checkArgument(length <= fullName.length,
414+
"length (%s) must be <= array length (%s)", length, fullName.length);
395415
for (TableName tn : tableCache) {
396-
if (Arrays.equals(tn.getName(), fullName)) {
416+
final byte[] tnName = tn.getName();
417+
if (Bytes.equals(tnName, 0, tnName.length, fullName, offset, length)) {
397418
return tn;
398419
}
399420
}
400421

401-
int namespaceDelimIndex = org.apache.hbase.thirdparty.com.google.common.primitives.Bytes
402-
.lastIndexOf(fullName, (byte) NAMESPACE_DELIM);
422+
int namespaceDelimIndex = ArrayUtils.lastIndexOf(fullName, (byte) NAMESPACE_DELIM);
403423

404-
if (namespaceDelimIndex < 0) {
424+
if (namespaceDelimIndex < offset) {
405425
return createTableNameIfNecessary(ByteBuffer.wrap(NamespaceDescriptor.DEFAULT_NAMESPACE_NAME),
406-
ByteBuffer.wrap(fullName));
426+
ByteBuffer.wrap(fullName, offset, length));
427+
} else {
428+
return createTableNameIfNecessary(ByteBuffer.wrap(fullName, offset, namespaceDelimIndex),
429+
ByteBuffer.wrap(fullName, namespaceDelimIndex + 1, length - (namespaceDelimIndex + 1)));
430+
}
431+
}
432+
433+
/**
434+
* @param fullname of a table, possibly with a leading namespace and ':' as delimiter.
435+
* @throws IllegalArgumentException if fullName equals old root or old meta.
436+
*/
437+
public static TableName valueOf(ByteBuffer fullname) {
438+
fullname = fullname.duplicate();
439+
fullname.mark();
440+
boolean miss = true;
441+
while (fullname.hasRemaining() && miss) {
442+
miss = ((byte) NAMESPACE_DELIM) != fullname.get();
443+
}
444+
if (miss) {
445+
fullname.reset();
446+
return valueOf(null, fullname);
407447
} else {
408-
return createTableNameIfNecessary(ByteBuffer.wrap(fullName, 0, namespaceDelimIndex),
409-
ByteBuffer.wrap(fullName, namespaceDelimIndex + 1,
410-
fullName.length - (namespaceDelimIndex + 1)));
448+
ByteBuffer qualifier = fullname.slice();
449+
int delimiterIndex = fullname.position() - 1;
450+
fullname.reset();
451+
// changing variable name for clarity
452+
ByteBuffer namespace = fullname.duplicate();
453+
namespace.limit(delimiterIndex);
454+
return valueOf(namespace, qualifier);
411455
}
412456
}
413457

‎hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFilePrettyPrinter.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import java.util.List;
4242
import java.util.Locale;
4343
import java.util.Map;
44+
import java.util.Optional;
4445
import java.util.Set;
4546
import java.util.SortedMap;
4647
import java.util.TimeZone;
@@ -422,17 +423,16 @@ private void scanKeysValues(Path file, KeyValueStatsCollector fileStats, HFileSc
422423
}
423424
// check if mob files are missing.
424425
if (checkMobIntegrity && MobUtils.isMobReferenceCell(cell)) {
425-
Tag tnTag = MobUtils.getTableNameTag(cell);
426-
if (tnTag == null) {
426+
Optional<TableName> tn = MobUtils.getTableName(cell);
427+
if (!tn.isPresent()) {
427428
System.err.println(
428429
"ERROR, wrong tag format in mob reference cell " + CellUtil.getCellKeyAsString(cell));
429430
} else if (!MobUtils.hasValidMobRefCellValue(cell)) {
430431
System.err.println(
431432
"ERROR, wrong value format in mob reference cell " + CellUtil.getCellKeyAsString(cell));
432433
} else {
433-
TableName tn = TableName.valueOf(Tag.cloneValue(tnTag));
434434
String mobFileName = MobUtils.getMobFileName(cell);
435-
boolean exist = mobFileExists(fs, tn, mobFileName,
435+
boolean exist = mobFileExists(fs, tn.get(), mobFileName,
436436
Bytes.toString(CellUtil.cloneFamily(cell)), foundMobFiles, missingMobFiles);
437437
if (!exist) {
438438
// report error

‎hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java‎

Lines changed: 87 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,9 @@
2626
import java.util.ArrayList;
2727
import java.util.Date;
2828
import java.util.HashMap;
29-
import java.util.HashSet;
3029
import java.util.List;
31-
import java.util.Set;
30+
import java.util.Map;
31+
import java.util.Optional;
3232
import java.util.function.Consumer;
3333
import org.apache.hadoop.conf.Configuration;
3434
import org.apache.hadoop.fs.FileStatus;
@@ -37,9 +37,11 @@
3737
import org.apache.hadoop.hbase.Cell;
3838
import org.apache.hadoop.hbase.KeyValue;
3939
import org.apache.hadoop.hbase.PrivateCellUtil;
40+
import org.apache.hadoop.hbase.TableName;
4041
import org.apache.hadoop.hbase.regionserver.CellSink;
4142
import org.apache.hadoop.hbase.regionserver.HMobStore;
4243
import org.apache.hadoop.hbase.regionserver.HStore;
44+
import org.apache.hadoop.hbase.regionserver.HStoreFile;
4345
import org.apache.hadoop.hbase.regionserver.InternalScanner;
4446
import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
4547
import org.apache.hadoop.hbase.regionserver.ScanInfo;
@@ -62,7 +64,10 @@
6264
import org.slf4j.Logger;
6365
import org.slf4j.LoggerFactory;
6466

67+
import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap;
68+
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSetMultimap;
6569
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
70+
import org.apache.hbase.thirdparty.com.google.common.collect.SetMultimap;
6671

6772
/**
6873
* Compact passed set of files in the mob-enabled column family.
@@ -82,12 +87,8 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
8287
* compaction process.
8388
*/
8489

85-
static ThreadLocal<Set<String>> mobRefSet = new ThreadLocal<Set<String>>() {
86-
@Override
87-
protected Set<String> initialValue() {
88-
return new HashSet<String>();
89-
}
90-
};
90+
static ThreadLocal<SetMultimap<TableName, String>> mobRefSet =
91+
ThreadLocal.withInitial(HashMultimap::create);
9192

9293
/*
9394
* Is it user or system-originated request.
@@ -192,34 +193,72 @@ public List<Path> compact(CompactionRequestImpl request,
192193
// Check if I/O optimized MOB compaction
193194
if (ioOptimizedMode) {
194195
if (request.isMajor() && request.getPriority() == HStore.PRIORITY_USER) {
195-
Path mobDir =
196-
MobUtils.getMobFamilyPath(conf, store.getTableName(), store.getColumnFamilyName());
197-
List<Path> mobFiles = MobUtils.getReferencedMobFiles(request.getFiles(), mobDir);
198-
// reset disableIO
199-
disableIO.set(Boolean.FALSE);
200-
if (mobFiles.size() > 0) {
201-
calculateMobLengthMap(mobFiles);
196+
try {
197+
final SetMultimap<TableName, String> mobRefs = request.getFiles().stream().map(file -> {
198+
byte[] value = file.getMetadataValue(HStoreFile.MOB_FILE_REFS);
199+
ImmutableSetMultimap.Builder<TableName, String> builder;
200+
if (value == null) {
201+
builder = ImmutableSetMultimap.builder();
202+
} else {
203+
try {
204+
builder = MobUtils.deserializeMobFileRefs(value);
205+
} catch (RuntimeException exception) {
206+
throw new RuntimeException("failure getting mob references for hfile " + file,
207+
exception);
208+
}
209+
}
210+
return builder;
211+
}).reduce((a, b) -> a.putAll(b.build())).orElseGet(ImmutableSetMultimap::builder).build();
212+
// reset disableIO
213+
disableIO.set(Boolean.FALSE);
214+
if (!mobRefs.isEmpty()) {
215+
calculateMobLengthMap(mobRefs);
216+
}
217+
LOG.info(
218+
"Table={} cf={} region={}. I/O optimized MOB compaction. "
219+
+ "Total referenced MOB files: {}",
220+
tableName, familyName, regionName, mobRefs.size());
221+
} catch (RuntimeException exception) {
222+
throw new IOException("Failed to get list of referenced hfiles for request " + request,
223+
exception);
202224
}
203-
LOG.info("Table={} cf={} region={}. I/O optimized MOB compaction. "
204-
+ "Total referenced MOB files: {}", tableName, familyName, regionName, mobFiles.size());
205225
}
206226
}
207227

208228
return compact(request, scannerFactory, writerFactory, throughputController, user);
209229
}
210230

211-
private void calculateMobLengthMap(List<Path> mobFiles) throws IOException {
231+
/**
232+
* @param mobRefs multimap of original table name -> mob hfile
233+
*/
234+
private void calculateMobLengthMap(SetMultimap<TableName, String> mobRefs) throws IOException {
212235
FileSystem fs = store.getFileSystem();
213236
HashMap<String, Long> map = mobLengthMap.get();
214237
map.clear();
215-
for (Path p : mobFiles) {
216-
if (MobFileName.isOldMobFileName(p.getName())) {
238+
for (Map.Entry<TableName, String> reference : mobRefs.entries()) {
239+
final TableName table = reference.getKey();
240+
final String mobfile = reference.getValue();
241+
if (MobFileName.isOldMobFileName(mobfile)) {
217242
disableIO.set(Boolean.TRUE);
218243
}
219-
FileStatus st = fs.getFileStatus(p);
220-
long size = st.getLen();
221-
LOG.debug("Referenced MOB file={} size={}", p, size);
222-
map.put(p.getName(), fs.getFileStatus(p).getLen());
244+
List<Path> locations = mobStore.getLocations(table);
245+
for (Path p : locations) {
246+
try {
247+
FileStatus st = fs.getFileStatus(new Path(p, mobfile));
248+
long size = st.getLen();
249+
LOG.debug("Referenced MOB file={} size={}", mobfile, size);
250+
map.put(mobfile, size);
251+
break;
252+
} catch (FileNotFoundException exception) {
253+
LOG.debug("Mob file {} was not in location {}. May have other locations to try.", mobfile,
254+
p);
255+
}
256+
}
257+
if (!map.containsKey(mobfile)) {
258+
throw new FileNotFoundException("Could not find mob file " + mobfile + " in the list of "
259+
+ "expected locations: " + locations);
260+
}
261+
223262
}
224263
}
225264

@@ -395,8 +434,15 @@ protected boolean performCompaction(FileDetails fd, InternalScanner scanner, Cel
395434
// We leave large MOB file as is (is not compacted),
396435
// then we update set of MOB file references
397436
// and append mob cell directly to the store's writer
398-
mobRefSet.get().add(fName);
399-
writer.append(mobCell);
437+
Optional<TableName> refTable = MobUtils.getTableName(c);
438+
if (refTable.isPresent()) {
439+
mobRefSet.get().put(refTable.get(), fName);
440+
writer.append(c);
441+
} else {
442+
throw new IOException("MOB cell did not contain a tablename "
443+
+ "tag. should not be possible. see ref guide on mob troubleshooting. "
444+
+ "store=" + getStoreInfo() + " cell=" + c);
445+
}
400446
}
401447
}
402448
} else {
@@ -444,9 +490,15 @@ protected boolean performCompaction(FileDetails fd, InternalScanner scanner, Cel
444490
if (MobUtils.hasValidMobRefCellValue(c)) {
445491
// We do not check mobSizeThreshold during normal compaction,
446492
// leaving it to a MOB compaction run
447-
writer.append(c);
448-
// Add MOB reference to a MOB reference set
449-
mobRefSet.get().add(MobUtils.getMobFileName(c));
493+
Optional<TableName> refTable = MobUtils.getTableName(c);
494+
if (refTable.isPresent()) {
495+
mobRefSet.get().put(refTable.get(), MobUtils.getMobFileName(c));
496+
writer.append(c);
497+
} else {
498+
throw new IOException("MOB cell did not contain a tablename "
499+
+ "tag. should not be possible. see ref guide on mob troubleshooting. " + "store="
500+
+ getStoreInfo() + " cell=" + c);
501+
}
450502
} else {
451503
String errMsg = String.format("Corrupted MOB reference: %s", c.toString());
452504
throw new IOException(errMsg);
@@ -525,7 +577,7 @@ protected boolean performCompaction(FileDetails fd, InternalScanner scanner, Cel
525577
throughputController.finish(compactionName);
526578
if (!finished && mobFileWriter != null) {
527579
// Remove all MOB references because compaction failed
528-
mobRefSet.get().clear();
580+
clearThreadLocals();
529581
// Abort writer
530582
LOG.debug("Aborting writer for {} because of a compaction failure, Store {}",
531583
mobFileWriter.getPath(), getStoreInfo());
@@ -543,16 +595,13 @@ protected boolean performCompaction(FileDetails fd, InternalScanner scanner, Cel
543595
return true;
544596
}
545597

546-
private String getStoreInfo() {
598+
protected String getStoreInfo() {
547599
return String.format("[table=%s family=%s region=%s]", store.getTableName().getNameAsString(),
548600
store.getColumnFamilyName(), store.getRegionInfo().getEncodedName());
549601
}
550602

551603
private void clearThreadLocals() {
552-
Set<String> set = mobRefSet.get();
553-
if (set != null) {
554-
set.clear();
555-
}
604+
mobRefSet.get().clear();
556605
HashMap<String, Long> map = mobLengthMap.get();
557606
if (map != null) {
558607
map.clear();
@@ -567,7 +616,7 @@ private StoreFileWriter newMobWriter(FileDetails fd, boolean major) throws IOExc
567616
LOG.debug("New MOB writer created={} store={}", mobFileWriter.getPath().getName(),
568617
getStoreInfo());
569618
// Add reference we get for compact MOB
570-
mobRefSet.get().add(mobFileWriter.getPath().getName());
619+
mobRefSet.get().put(store.getTableName(), mobFileWriter.getPath().getName());
571620
return mobFileWriter;
572621
} catch (IOException e) {
573622
// Bailing out
@@ -599,7 +648,7 @@ private void commitOrAbortMobWriter(StoreFileWriter mobFileWriter, long maxSeqId
599648
LOG.debug("Aborting writer for {} because there are no MOB cells, store={}",
600649
mobFileWriter.getPath(), getStoreInfo());
601650
// Remove MOB file from reference set
602-
mobRefSet.get().remove(mobFileWriter.getPath().getName());
651+
mobRefSet.get().remove(store.getTableName(), mobFileWriter.getPath().getName());
603652
abortWriter(mobFileWriter);
604653
}
605654
} else {
@@ -612,9 +661,7 @@ protected List<Path> commitWriter(StoreFileWriter writer, FileDetails fd,
612661
CompactionRequestImpl request) throws IOException {
613662
List<Path> newFiles = Lists.newArrayList(writer.getPath());
614663
writer.appendMetadata(fd.maxSeqId, request.isAllFiles(), request.getFiles());
615-
// Append MOB references
616-
Set<String> refSet = mobRefSet.get();
617-
writer.appendMobMetadata(refSet);
664+
writer.appendMobMetadata(mobRefSet.get());
618665
writer.close();
619666
clearThreadLocals();
620667
return newFiles;

‎hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.hadoop.hbase.Cell;
3131
import org.apache.hadoop.hbase.HConstants;
3232
import org.apache.hadoop.hbase.KeyValue;
33+
import org.apache.hadoop.hbase.TableName;
3334
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
3435
import org.apache.hadoop.hbase.regionserver.DefaultStoreFlusher;
3536
import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker;
@@ -47,6 +48,8 @@
4748
import org.slf4j.Logger;
4849
import org.slf4j.LoggerFactory;
4950

51+
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSetMultimap;
52+
5053
/**
5154
* An implementation of the StoreFlusher. It extends the DefaultStoreFlusher. If the store is not a
5255
* mob store, the flusher flushes the MemStore the same with DefaultStoreFlusher, If the store is a
@@ -277,7 +280,8 @@ protected void finalizeWriter(StoreFileWriter writer, long cacheFlushSeqNum, Mon
277280
// The hfile is current up to and including cacheFlushSeqNum.
278281
status.setStatus("Flushing " + store + ": appending metadata");
279282
writer.appendMetadata(cacheFlushSeqNum, false);
280-
writer.appendMobMetadata(mobRefSet.get());
283+
writer.appendMobMetadata(ImmutableSetMultimap.<TableName, String> builder()
284+
.putAll(store.getTableName(), mobRefSet.get()).build());
281285
status.setStatus("Flushing " + store + ": closing flushed file");
282286
writer.close();
283287
}

0 commit comments

Comments
 (0)