1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mob;
20
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.Date;
24 import java.util.List;
25
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.classification.InterfaceAudience;
29 import org.apache.hadoop.conf.Configuration;
30 import org.apache.hadoop.fs.Path;
31 import org.apache.hadoop.hbase.Cell;
32 import org.apache.hadoop.hbase.HConstants;
33 import org.apache.hadoop.hbase.KeyValue;
34 import org.apache.hadoop.hbase.KeyValueUtil;
35 import org.apache.hadoop.hbase.Tag;
36 import org.apache.hadoop.hbase.TagType;
37 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
38 import org.apache.hadoop.hbase.regionserver.DefaultStoreFlusher;
39 import org.apache.hadoop.hbase.regionserver.HMobStore;
40 import org.apache.hadoop.hbase.regionserver.InternalScanner;
41 import org.apache.hadoop.hbase.regionserver.MemStoreSnapshot;
42 import org.apache.hadoop.hbase.regionserver.ScannerContext;
43 import org.apache.hadoop.hbase.regionserver.Store;
44 import org.apache.hadoop.hbase.regionserver.StoreFile;
45 import org.apache.hadoop.hbase.util.Bytes;
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62 @InterfaceAudience.Private
63 public class DefaultMobStoreFlusher extends DefaultStoreFlusher {
64
65 private static final Log LOG = LogFactory.getLog(DefaultMobStoreFlusher.class);
66 private final Object flushLock = new Object();
67 private long mobCellValueSizeThreshold = 0;
68 private Path targetPath;
69 private HMobStore mobStore;
70
71 public DefaultMobStoreFlusher(Configuration conf, Store store) throws IOException {
72 super(conf, store);
73 mobCellValueSizeThreshold = store.getFamily().getMobThreshold();
74 this.targetPath = MobUtils.getMobFamilyPath(conf, store.getTableName(),
75 store.getColumnFamilyName());
76 if (!this.store.getFileSystem().exists(targetPath)) {
77 this.store.getFileSystem().mkdirs(targetPath);
78 }
79 this.mobStore = (HMobStore) store;
80 }
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96 @Override
97 public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushId,
98 MonitoredTask status) throws IOException {
99 ArrayList<Path> result = new ArrayList<Path>();
100 int cellsCount = snapshot.getCellsCount();
101 if (cellsCount == 0) return result;
102
103
104 long smallestReadPoint = store.getSmallestReadPoint();
105 InternalScanner scanner = createScanner(snapshot.getScanner(), smallestReadPoint);
106 if (scanner == null) {
107 return result;
108 }
109 StoreFile.Writer writer;
110 try {
111
112
113 synchronized (flushLock) {
114 status.setStatus("Flushing " + store + ": creating writer");
115
116 writer = store.createWriterInTmp(cellsCount, store.getFamily().getCompression(),
117 false, true, true);
118 writer.setTimeRangeTracker(snapshot.getTimeRangeTracker());
119 try {
120
121
122 performMobFlush(snapshot, cacheFlushId, scanner, writer, status);
123 } finally {
124 finalizeWriter(writer, cacheFlushId, status);
125 }
126 }
127 } finally {
128 scanner.close();
129 }
130 LOG.info("Flushed, sequenceid=" + cacheFlushId + ", memsize="
131 + snapshot.getSize() + ", hasBloomFilter=" + writer.hasGeneralBloom()
132 + ", into tmp file " + writer.getPath());
133 result.add(writer.getPath());
134 return result;
135 }
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154 protected void performMobFlush(MemStoreSnapshot snapshot, long cacheFlushId,
155 InternalScanner scanner, StoreFile.Writer writer, MonitoredTask status) throws IOException {
156 StoreFile.Writer mobFileWriter = null;
157 int compactionKVMax = conf.getInt(HConstants.COMPACTION_KV_MAX,
158 HConstants.COMPACTION_KV_MAX_DEFAULT);
159 long mobCount = 0;
160 long mobSize = 0;
161 long time = snapshot.getTimeRangeTracker().getMaximumTimestamp();
162 mobFileWriter = mobStore.createWriterInTmp(new Date(time), snapshot.getCellsCount(),
163 store.getFamily().getCompression(), store.getRegionInfo().getStartKey());
164
165
166 byte[] fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
167 try {
168 Tag tableNameTag = new Tag(TagType.MOB_TABLE_NAME_TAG_TYPE, store.getTableName()
169 .getName());
170 List<Cell> cells = new ArrayList<Cell>();
171 boolean hasMore;
172 ScannerContext scannerContext =
173 ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
174
175 do {
176 hasMore = scanner.next(cells, scannerContext);
177 if (!cells.isEmpty()) {
178 for (Cell c : cells) {
179
180
181
182 KeyValue kv = KeyValueUtil.ensureKeyValue(c);
183 if (kv.getValueLength() <= mobCellValueSizeThreshold || MobUtils.isMobReferenceCell(kv)
184 || kv.getTypeByte() != KeyValue.Type.Put.getCode()) {
185 writer.append(kv);
186 } else {
187
188 mobFileWriter.append(kv);
189 mobSize += kv.getValueLength();
190 mobCount++;
191
192
193
194 KeyValue reference = MobUtils.createMobRefKeyValue(kv, fileName, tableNameTag);
195 writer.append(reference);
196 }
197 }
198 cells.clear();
199 }
200 } while (hasMore);
201 } finally {
202 status.setStatus("Flushing mob file " + store + ": appending metadata");
203 mobFileWriter.appendMetadata(cacheFlushId, false, mobCount);
204 status.setStatus("Flushing mob file " + store + ": closing flushed file");
205 mobFileWriter.close();
206 }
207
208 if (mobCount > 0) {
209
210
211
212
213 mobStore.commitFile(mobFileWriter.getPath(), targetPath);
214 mobStore.updateMobFlushCount();
215 mobStore.updateMobFlushedCellsCount(mobCount);
216 mobStore.updateMobFlushedCellsSize(mobSize);
217 } else {
218 try {
219
220 store.getFileSystem().delete(mobFileWriter.getPath(), true);
221 } catch (IOException e) {
222 LOG.error("Fail to delete the temp mob file", e);
223 }
224 }
225 }
226 }