1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.io.hfile;
21
22 import java.io.DataOutput;
23 import java.io.DataOutputStream;
24 import java.io.IOException;
25 import java.util.ArrayList;
26 import java.util.List;
27
28 import org.apache.commons.logging.Log;
29 import org.apache.commons.logging.LogFactory;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.fs.FSDataOutputStream;
32 import org.apache.hadoop.fs.FileSystem;
33 import org.apache.hadoop.fs.Path;
34 import org.apache.hadoop.hbase.Cell;
35 import org.apache.hadoop.hbase.CellComparator;
36 import org.apache.hadoop.hbase.CellUtil;
37 import org.apache.hadoop.hbase.KeyValue.KVComparator;
38 import org.apache.hadoop.hbase.classification.InterfaceAudience;
39 import org.apache.hadoop.hbase.io.hfile.HFile.Writer;
40 import org.apache.hadoop.hbase.io.hfile.HFileBlock.BlockWritable;
41 import org.apache.hadoop.hbase.util.BloomFilterWriter;
42 import org.apache.hadoop.hbase.util.Bytes;
43 import org.apache.hadoop.io.Writable;
44
45
46
47
48 @InterfaceAudience.Private
49 public class HFileWriterV2 extends AbstractHFileWriter {
50 static final Log LOG = LogFactory.getLog(HFileWriterV2.class);
51
52
53 public static final byte [] MAX_MEMSTORE_TS_KEY =
54 Bytes.toBytes("MAX_MEMSTORE_TS_KEY");
55
56
57 public static final byte [] KEY_VALUE_VERSION =
58 Bytes.toBytes("KEY_VALUE_VERSION");
59
60
61 public static final int KEY_VALUE_VER_WITH_MEMSTORE = 1;
62
63
64 private List<InlineBlockWriter> inlineBlockWriters =
65 new ArrayList<InlineBlockWriter>();
66
67
68 protected HFileBlock.Writer fsBlockWriter;
69
70 private HFileBlockIndex.BlockIndexWriter dataBlockIndexWriter;
71 private HFileBlockIndex.BlockIndexWriter metaBlockIndexWriter;
72
73
74 private long firstDataBlockOffset = -1;
75
76
77 protected long lastDataBlockOffset;
78
79
80
81
82
83 private Cell lastCellOfPreviousBlock = null;
84
85
86 private List<BlockWritable> additionalLoadOnOpenData =
87 new ArrayList<BlockWritable>();
88
89 protected long maxMemstoreTS = 0;
90
91
92 private static boolean warnCellWithTags = true;
93
94 static class WriterFactoryV2 extends HFile.WriterFactory {
95 WriterFactoryV2(Configuration conf, CacheConfig cacheConf) {
96 super(conf, cacheConf);
97 }
98
99 @Override
100 public Writer createWriter(FileSystem fs, Path path,
101 FSDataOutputStream ostream,
102 KVComparator comparator, HFileContext context) throws IOException {
103 context.setIncludesTags(false);
104 return new HFileWriterV2(conf, cacheConf, fs, path, ostream,
105 comparator, context);
106 }
107 }
108
109
110 public HFileWriterV2(Configuration conf, CacheConfig cacheConf,
111 FileSystem fs, Path path, FSDataOutputStream ostream,
112 final KVComparator comparator, final HFileContext context) throws IOException {
113 super(cacheConf,
114 ostream == null ? createOutputStream(conf, fs, path, null) : ostream,
115 path, comparator, context);
116 finishInit(conf);
117 }
118
119
120 protected void finishInit(final Configuration conf) {
121 if (fsBlockWriter != null)
122 throw new IllegalStateException("finishInit called twice");
123
124 fsBlockWriter = new HFileBlock.Writer(blockEncoder, hFileContext);
125
126
127 boolean cacheIndexesOnWrite = cacheConf.shouldCacheIndexesOnWrite();
128 dataBlockIndexWriter = new HFileBlockIndex.BlockIndexWriter(fsBlockWriter,
129 cacheIndexesOnWrite ? cacheConf : null,
130 cacheIndexesOnWrite ? name : null);
131 dataBlockIndexWriter.setMaxChunkSize(
132 HFileBlockIndex.getMaxChunkSize(conf));
133 inlineBlockWriters.add(dataBlockIndexWriter);
134
135
136 metaBlockIndexWriter = new HFileBlockIndex.BlockIndexWriter();
137 if (LOG.isTraceEnabled()) LOG.trace("Initialized with " + cacheConf);
138 }
139
140
141
142
143
144
145 protected void checkBlockBoundary() throws IOException {
146 if (fsBlockWriter.blockSizeWritten() < hFileContext.getBlocksize())
147 return;
148
149 finishBlock();
150 writeInlineBlocks(false);
151 newBlock();
152 }
153
154
155 private void finishBlock() throws IOException {
156 if (!fsBlockWriter.isWriting() || fsBlockWriter.blockSizeWritten() == 0)
157 return;
158
159
160 if (firstDataBlockOffset == -1) {
161 firstDataBlockOffset = outputStream.getPos();
162 }
163
164 lastDataBlockOffset = outputStream.getPos();
165 fsBlockWriter.writeHeaderAndData(outputStream);
166 int onDiskSize = fsBlockWriter.getOnDiskSizeWithHeader();
167
168 Cell indexEntry =
169 CellComparator.getMidpoint(this.comparator, lastCellOfPreviousBlock, firstCellInBlock);
170 dataBlockIndexWriter.addEntry(CellUtil.getCellKeySerializedAsKeyValueKey(indexEntry),
171 lastDataBlockOffset, onDiskSize);
172 totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
173 if (cacheConf.shouldCacheDataOnWrite()) {
174 doCacheOnWrite(lastDataBlockOffset);
175 }
176 }
177
178
179 private void writeInlineBlocks(boolean closing) throws IOException {
180 for (InlineBlockWriter ibw : inlineBlockWriters) {
181 while (ibw.shouldWriteBlock(closing)) {
182 long offset = outputStream.getPos();
183 boolean cacheThisBlock = ibw.getCacheOnWrite();
184 ibw.writeInlineBlock(fsBlockWriter.startWriting(
185 ibw.getInlineBlockType()));
186 fsBlockWriter.writeHeaderAndData(outputStream);
187 ibw.blockWritten(offset, fsBlockWriter.getOnDiskSizeWithHeader(),
188 fsBlockWriter.getUncompressedSizeWithoutHeader());
189 totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
190
191 if (cacheThisBlock) {
192 doCacheOnWrite(offset);
193 }
194 }
195 }
196 }
197
198
199
200
201
202
203 private void doCacheOnWrite(long offset) {
204 HFileBlock cacheFormatBlock = fsBlockWriter.getBlockForCaching(cacheConf);
205 cacheConf.getBlockCache().cacheBlock(
206 new BlockCacheKey(name, offset), cacheFormatBlock);
207 }
208
209
210
211
212
213
214 protected void newBlock() throws IOException {
215
216 fsBlockWriter.startWriting(BlockType.DATA);
217 firstCellInBlock = null;
218 if (lastCell != null) {
219 lastCellOfPreviousBlock = lastCell;
220 }
221 }
222
223
224
225
226
227
228
229
230
231
232
233
234 @Override
235 public void appendMetaBlock(String metaBlockName, Writable content) {
236 byte[] key = Bytes.toBytes(metaBlockName);
237 int i;
238 for (i = 0; i < metaNames.size(); ++i) {
239
240 byte[] cur = metaNames.get(i);
241 if (Bytes.BYTES_RAWCOMPARATOR.compare(cur, 0, cur.length, key, 0,
242 key.length) > 0) {
243 break;
244 }
245 }
246 metaNames.add(i, key);
247 metaData.add(i, content);
248 }
249
250
251
252
253
254
255
256
257 @Override
258 public void append(final Cell cell) throws IOException {
259 byte[] value = cell.getValueArray();
260 int voffset = cell.getValueOffset();
261 int vlength = cell.getValueLength();
262
263 boolean dupKey = checkKey(cell);
264 checkValue(value, voffset, vlength);
265 if (!dupKey) {
266 checkBlockBoundary();
267 }
268
269 if (!fsBlockWriter.isWriting()) {
270 newBlock();
271 }
272
273 if (warnCellWithTags && getFileContext().isIncludesTags()) {
274 LOG.warn("A minimum HFile version of " + HFile.MIN_FORMAT_VERSION_WITH_TAGS
275 + " is required to support cell attributes/tags. Consider setting "
276 + HFile.FORMAT_VERSION_KEY + " accordingly.");
277 warnCellWithTags = false;
278 }
279
280 fsBlockWriter.write(cell);
281
282 totalKeyLength += CellUtil.estimatedSerializedSizeOfKey(cell);
283 totalValueLength += vlength;
284
285
286 if (firstCellInBlock == null) {
287
288
289 firstCellInBlock = cell;
290 }
291
292
293
294 lastCell = cell;
295 entryCount++;
296 this.maxMemstoreTS = Math.max(this.maxMemstoreTS, cell.getSequenceId());
297 }
298
299 @Override
300 public void close() throws IOException {
301 if (outputStream == null) {
302 return;
303 }
304
305 blockEncoder.saveMetadata(this);
306
307
308
309 finishBlock();
310 writeInlineBlocks(true);
311
312 FixedFileTrailer trailer = new FixedFileTrailer(getMajorVersion(), getMinorVersion());
313
314
315 if (!metaNames.isEmpty()) {
316 for (int i = 0; i < metaNames.size(); ++i) {
317
318 long offset = outputStream.getPos();
319
320 DataOutputStream dos = fsBlockWriter.startWriting(BlockType.META);
321 metaData.get(i).write(dos);
322
323 fsBlockWriter.writeHeaderAndData(outputStream);
324 totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
325
326
327 metaBlockIndexWriter.addEntry(metaNames.get(i), offset,
328 fsBlockWriter.getOnDiskSizeWithHeader());
329 }
330 }
331
332
333
334
335
336
337
338
339
340
341 long rootIndexOffset = dataBlockIndexWriter.writeIndexBlocks(outputStream);
342 trailer.setLoadOnOpenOffset(rootIndexOffset);
343
344
345 metaBlockIndexWriter.writeSingleLevelIndex(fsBlockWriter.startWriting(
346 BlockType.ROOT_INDEX), "meta");
347 fsBlockWriter.writeHeaderAndData(outputStream);
348 totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
349
350 if (this.hFileContext.isIncludesMvcc()) {
351 appendFileInfo(MAX_MEMSTORE_TS_KEY, Bytes.toBytes(maxMemstoreTS));
352 appendFileInfo(KEY_VALUE_VERSION, Bytes.toBytes(KEY_VALUE_VER_WITH_MEMSTORE));
353 }
354
355
356 writeFileInfo(trailer, fsBlockWriter.startWriting(BlockType.FILE_INFO));
357 fsBlockWriter.writeHeaderAndData(outputStream);
358 totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
359
360
361 for (BlockWritable w : additionalLoadOnOpenData){
362 fsBlockWriter.writeBlock(w, outputStream);
363 totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
364 }
365
366
367 trailer.setNumDataIndexLevels(dataBlockIndexWriter.getNumLevels());
368 trailer.setUncompressedDataIndexSize(
369 dataBlockIndexWriter.getTotalUncompressedSize());
370 trailer.setFirstDataBlockOffset(firstDataBlockOffset);
371 trailer.setLastDataBlockOffset(lastDataBlockOffset);
372 trailer.setComparatorClass(comparator.getClass());
373 trailer.setDataIndexCount(dataBlockIndexWriter.getNumRootEntries());
374
375
376 finishClose(trailer);
377
378 fsBlockWriter.release();
379 }
380
381 @Override
382 public void addInlineBlockWriter(InlineBlockWriter ibw) {
383 inlineBlockWriters.add(ibw);
384 }
385
386 @Override
387 public void addGeneralBloomFilter(final BloomFilterWriter bfw) {
388 this.addBloomFilter(bfw, BlockType.GENERAL_BLOOM_META);
389 }
390
391 @Override
392 public void addDeleteFamilyBloomFilter(final BloomFilterWriter bfw) {
393 this.addBloomFilter(bfw, BlockType.DELETE_FAMILY_BLOOM_META);
394 }
395
396 private void addBloomFilter(final BloomFilterWriter bfw,
397 final BlockType blockType) {
398 if (bfw.getKeyCount() <= 0)
399 return;
400
401 if (blockType != BlockType.GENERAL_BLOOM_META &&
402 blockType != BlockType.DELETE_FAMILY_BLOOM_META) {
403 throw new RuntimeException("Block Type: " + blockType.toString() +
404 "is not supported");
405 }
406 additionalLoadOnOpenData.add(new BlockWritable() {
407 @Override
408 public BlockType getBlockType() {
409 return blockType;
410 }
411
412 @Override
413 public void writeToBlock(DataOutput out) throws IOException {
414 bfw.getMetaWriter().write(out);
415 Writable dataWriter = bfw.getDataWriter();
416 if (dataWriter != null)
417 dataWriter.write(out);
418 }
419 });
420 }
421
422 protected int getMajorVersion() {
423 return 2;
424 }
425
426 protected int getMinorVersion() {
427 return HFileReaderV2.MAX_MINOR_VERSION;
428 }
429
430 @Override
431 public HFileContext getFileContext() {
432 return hFileContext;
433 }
434 }