View Javadoc

1   /*
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.io.hfile;
20  
21  import java.io.ByteArrayInputStream;
22  import java.io.ByteArrayOutputStream;
23  import java.io.DataInput;
24  import java.io.DataInputStream;
25  import java.io.DataOutputStream;
26  import java.io.IOException;
27  import java.nio.ByteBuffer;
28  
29  import org.apache.hadoop.hbase.util.ByteStringer;
30  import org.apache.hadoop.hbase.CellComparator.MetaCellComparator;
31  import org.apache.hadoop.hbase.classification.InterfaceAudience;
32  import org.apache.hadoop.fs.FSDataInputStream;
33  import org.apache.hadoop.hbase.CellComparator;
34  import org.apache.hadoop.hbase.KeyValue;
35  import org.apache.hadoop.hbase.io.compress.Compression;
36  import org.apache.hadoop.hbase.protobuf.generated.HFileProtos;
37  import org.apache.hadoop.hbase.util.Bytes;
38  
39  
40  /**
41   * The {@link HFile} has a fixed trailer which contains offsets to other
42   * variable parts of the file. Also includes basic metadata on this file. The
43   * trailer size is fixed within a given {@link HFile} format version only, but
44   * we always store the version number as the last four-byte integer of the file.
45   * The version number itself is split into two portions, a major 
46   * version and a minor version. 
47   * The last three bytes of a file is the major
48   * version and a single preceding byte is the minor number. The major version
49   * determines which readers/writers to use to read/write a hfile while a minor
50   * version determines smaller changes in hfile format that do not need a new
51   * reader/writer type.
52   */
53  @InterfaceAudience.Private
54  public class FixedFileTrailer {
55  
56    /**
57     * We store the comparator class name as a fixed-length field in the trailer.
58     */
59    private static final int MAX_COMPARATOR_NAME_LENGTH = 128;
60  
61    /**
62     * Offset to the fileinfo data, a small block of vitals. Necessary in v1 but
63     * only potentially useful for pretty-printing in v2.
64     */
65    private long fileInfoOffset;
66  
67    /**
68     * In version 1, the offset to the data block index. Starting from version 2,
69     * the meaning of this field is the offset to the section of the file that
70     * should be loaded at the time the file is being opened, and as of the time
71     * of writing, this happens to be the offset of the file info section.
72     */
73    private long loadOnOpenDataOffset;
74  
75    /** The number of entries in the root data index. */
76    private int dataIndexCount;
77  
78    /** Total uncompressed size of all blocks of the data index */
79    private long uncompressedDataIndexSize;
80  
81    /** The number of entries in the meta index */
82    private int metaIndexCount;
83  
84    /** The total uncompressed size of keys/values stored in the file. */
85    private long totalUncompressedBytes;
86  
87    /**
88     * The number of key/value pairs in the file. This field was int in version 1,
89     * but is now long.
90     */
91    private long entryCount;
92  
93    /** The compression codec used for all blocks. */
94    private Compression.Algorithm compressionCodec = Compression.Algorithm.NONE;
95  
96    /**
97     * The number of levels in the potentially multi-level data index. Used from
98     * version 2 onwards.
99     */
100   private int numDataIndexLevels;
101 
102   /** The offset of the first data block. */
103   private long firstDataBlockOffset;
104 
105   /**
106    * It is guaranteed that no key/value data blocks start after this offset in
107    * the file.
108    */
109   private long lastDataBlockOffset;
110 
111   /** Raw key comparator class name in version 3 */
112   // We could write the actual class name from 2.0 onwards and handle BC
113   private String comparatorClassName = CellComparator.COMPARATOR.getClass().getName();
114 
115   /** The encryption key */
116   private byte[] encryptionKey;
117 
118   /** The {@link HFile} format major version. */
119   private final int majorVersion;
120 
121   /** The {@link HFile} format minor version. */
122   private final int minorVersion;
123 
124   FixedFileTrailer(int majorVersion, int minorVersion) {
125     this.majorVersion = majorVersion;
126     this.minorVersion = minorVersion;
127     HFile.checkFormatVersion(majorVersion);
128   }
129 
130   private static int[] computeTrailerSizeByVersion() {
131     int versionToSize[] = new int[HFile.MAX_FORMAT_VERSION + 1];
132     // We support only 2 major versions now. ie. V2, V3
133     versionToSize[2] = 212;
134     for (int version = 3; version <= HFile.MAX_FORMAT_VERSION; version++) {
135       // Max FFT size for V3 and above is taken as 4KB for future enhancements
136       // if any.
137       // Unless the trailer size exceeds 4K this can continue
138       versionToSize[version] = 1024 * 4;
139     }
140     return versionToSize;
141   }
142 
143   private static int getMaxTrailerSize() {
144     int maxSize = 0;
145     for (int version = HFile.MIN_FORMAT_VERSION;
146          version <= HFile.MAX_FORMAT_VERSION;
147          ++version)
148       maxSize = Math.max(getTrailerSize(version), maxSize);
149     return maxSize;
150   }
151 
152   private static final int TRAILER_SIZE[] = computeTrailerSizeByVersion();
153   private static final int MAX_TRAILER_SIZE = getMaxTrailerSize();
154 
155   private static final int NOT_PB_SIZE = BlockType.MAGIC_LENGTH + Bytes.SIZEOF_INT;
156 
157   static int getTrailerSize(int version) {
158     return TRAILER_SIZE[version];
159   }
160 
161   public int getTrailerSize() {
162     return getTrailerSize(majorVersion);
163   }
164 
165   /**
166    * Write the trailer to a data stream. We support writing version 1 for
167    * testing and for determining version 1 trailer size. It is also easy to see
168    * what fields changed in version 2.
169    *
170    * @param outputStream
171    * @throws IOException
172    */
173   void serialize(DataOutputStream outputStream) throws IOException {
174     HFile.checkFormatVersion(majorVersion);
175 
176     ByteArrayOutputStream baos = new ByteArrayOutputStream();
177     DataOutputStream baosDos = new DataOutputStream(baos);
178 
179     BlockType.TRAILER.write(baosDos);
180     serializeAsPB(baosDos);
181 
182     // The last 4 bytes of the file encode the major and minor version universally
183     baosDos.writeInt(materializeVersion(majorVersion, minorVersion));
184 
185     baos.writeTo(outputStream);
186   }
187 
188   /**
189    * Write trailer data as protobuf
190    * @param outputStream
191    * @throws IOException
192    */
193   void serializeAsPB(DataOutputStream output) throws IOException {
194     ByteArrayOutputStream baos = new ByteArrayOutputStream();
195     HFileProtos.FileTrailerProto.Builder builder = HFileProtos.FileTrailerProto.newBuilder()
196       .setFileInfoOffset(fileInfoOffset)
197       .setLoadOnOpenDataOffset(loadOnOpenDataOffset)
198       .setUncompressedDataIndexSize(uncompressedDataIndexSize)
199       .setTotalUncompressedBytes(totalUncompressedBytes)
200       .setDataIndexCount(dataIndexCount)
201       .setMetaIndexCount(metaIndexCount)
202       .setEntryCount(entryCount)
203       .setNumDataIndexLevels(numDataIndexLevels)
204       .setFirstDataBlockOffset(firstDataBlockOffset)
205       .setLastDataBlockOffset(lastDataBlockOffset)
206       // TODO this is a classname encoded into an  HFile's trailer. We are going to need to have
207       // some compat code here.
208       .setComparatorClassName(comparatorClassName)
209       .setCompressionCodec(compressionCodec.ordinal());
210     if (encryptionKey != null) {
211       builder.setEncryptionKey(ByteStringer.wrap(encryptionKey));
212     }
213     // We need this extra copy unfortunately to determine the final size of the
214     // delimited output, see use of baos.size() below.
215     builder.build().writeDelimitedTo(baos);
216     baos.writeTo(output);
217     // Pad to make up the difference between variable PB encoding length and the
218     // length when encoded as writable under earlier V2 formats. Failure to pad
219     // properly or if the PB encoding is too big would mean the trailer wont be read
220     // in properly by HFile.
221     int padding = getTrailerSize() - NOT_PB_SIZE - baos.size();
222     if (padding < 0) {
223       throw new IOException("Pbuf encoding size exceeded fixed trailer size limit");
224     }
225     for (int i = 0; i < padding; i++) {
226       output.write(0);
227     }
228   }
229 
230   /**
231    * Deserialize the fixed file trailer from the given stream. The version needs
232    * to already be specified. Make sure this is consistent with
233    * {@link #serialize(DataOutputStream)}.
234    *
235    * @param inputStream
236    * @throws IOException
237    */
238   void deserialize(DataInputStream inputStream) throws IOException {
239     HFile.checkFormatVersion(majorVersion);
240 
241     BlockType.TRAILER.readAndCheck(inputStream);
242 
243     if (majorVersion > 2
244         || (majorVersion == 2 && minorVersion >= HFileReaderImpl.PBUF_TRAILER_MINOR_VERSION)) {
245       deserializeFromPB(inputStream);
246     } else {
247       deserializeFromWritable(inputStream);
248     }
249 
250     // The last 4 bytes of the file encode the major and minor version universally
251     int version = inputStream.readInt();
252     expectMajorVersion(extractMajorVersion(version));
253     expectMinorVersion(extractMinorVersion(version));
254   }
255 
256   /**
257    * Deserialize the file trailer as protobuf
258    * @param inputStream
259    * @throws IOException
260    */
261   void deserializeFromPB(DataInputStream inputStream) throws IOException {
262     // read PB and skip padding
263     int start = inputStream.available();
264     HFileProtos.FileTrailerProto trailerProto =
265         HFileProtos.FileTrailerProto.PARSER.parseDelimitedFrom(inputStream);
266     int size = start - inputStream.available();
267     inputStream.skip(getTrailerSize() - NOT_PB_SIZE - size);
268 
269     // process the PB
270     if (trailerProto.hasFileInfoOffset()) {
271       fileInfoOffset = trailerProto.getFileInfoOffset();
272     }
273     if (trailerProto.hasLoadOnOpenDataOffset()) {
274       loadOnOpenDataOffset = trailerProto.getLoadOnOpenDataOffset();
275     }
276     if (trailerProto.hasUncompressedDataIndexSize()) {
277       uncompressedDataIndexSize = trailerProto.getUncompressedDataIndexSize();
278     }
279     if (trailerProto.hasTotalUncompressedBytes()) {
280       totalUncompressedBytes = trailerProto.getTotalUncompressedBytes();
281     }
282     if (trailerProto.hasDataIndexCount()) {
283       dataIndexCount = trailerProto.getDataIndexCount();
284     }
285     if (trailerProto.hasMetaIndexCount()) {
286       metaIndexCount = trailerProto.getMetaIndexCount();
287     }
288     if (trailerProto.hasEntryCount()) {
289       entryCount = trailerProto.getEntryCount();
290     }
291     if (trailerProto.hasNumDataIndexLevels()) {
292       numDataIndexLevels = trailerProto.getNumDataIndexLevels();
293     }
294     if (trailerProto.hasFirstDataBlockOffset()) {
295       firstDataBlockOffset = trailerProto.getFirstDataBlockOffset();
296     }
297     if (trailerProto.hasLastDataBlockOffset()) {
298       lastDataBlockOffset = trailerProto.getLastDataBlockOffset();
299     }
300     if (trailerProto.hasComparatorClassName()) {
301       // TODO this is a classname encoded into an  HFile's trailer. We are going to need to have 
302       // some compat code here.
303       setComparatorClass(getComparatorClass(trailerProto.getComparatorClassName()));
304     }
305     if (trailerProto.hasCompressionCodec()) {
306       compressionCodec = Compression.Algorithm.values()[trailerProto.getCompressionCodec()];
307     } else {
308       compressionCodec = Compression.Algorithm.NONE;
309     }
310     if (trailerProto.hasEncryptionKey()) {
311       encryptionKey = trailerProto.getEncryptionKey().toByteArray();
312     }
313   }
314 
315   /**
316    * Deserialize the file trailer as writable data
317    * @param input
318    * @throws IOException
319    */
320   void deserializeFromWritable(DataInput input) throws IOException {
321     fileInfoOffset = input.readLong();
322     loadOnOpenDataOffset = input.readLong();
323     dataIndexCount = input.readInt();
324     uncompressedDataIndexSize = input.readLong();
325     metaIndexCount = input.readInt();
326 
327     totalUncompressedBytes = input.readLong();
328     entryCount = input.readLong();
329     compressionCodec = Compression.Algorithm.values()[input.readInt()];
330     numDataIndexLevels = input.readInt();
331     firstDataBlockOffset = input.readLong();
332     lastDataBlockOffset = input.readLong();
333     // TODO this is a classname encoded into an  HFile's trailer. We are going to need to have 
334     // some compat code here.
335     setComparatorClass(getComparatorClass(Bytes.readStringFixedSize(input,
336         MAX_COMPARATOR_NAME_LENGTH)));
337   }
338   
339   private void append(StringBuilder sb, String s) {
340     if (sb.length() > 0)
341       sb.append(", ");
342     sb.append(s);
343   }
344 
345   @Override
346   public String toString() {
347     StringBuilder sb = new StringBuilder();
348     append(sb, "fileinfoOffset=" + fileInfoOffset);
349     append(sb, "loadOnOpenDataOffset=" + loadOnOpenDataOffset);
350     append(sb, "dataIndexCount=" + dataIndexCount);
351     append(sb, "metaIndexCount=" + metaIndexCount);
352     append(sb, "totalUncomressedBytes=" + totalUncompressedBytes);
353     append(sb, "entryCount=" + entryCount);
354     append(sb, "compressionCodec=" + compressionCodec);
355     append(sb, "uncompressedDataIndexSize=" + uncompressedDataIndexSize);
356     append(sb, "numDataIndexLevels=" + numDataIndexLevels);
357     append(sb, "firstDataBlockOffset=" + firstDataBlockOffset);
358     append(sb, "lastDataBlockOffset=" + lastDataBlockOffset);
359     append(sb, "comparatorClassName=" + comparatorClassName);
360     if (majorVersion >= 3) {
361       append(sb, "encryptionKey=" + (encryptionKey != null ? "PRESENT" : "NONE"));
362     }
363     append(sb, "majorVersion=" + majorVersion);
364     append(sb, "minorVersion=" + minorVersion);
365 
366     return sb.toString();
367   }
368 
369   /**
370    * Reads a file trailer from the given file.
371    *
372    * @param istream the input stream with the ability to seek. Does not have to
373    *          be buffered, as only one read operation is made.
374    * @param fileSize the file size. Can be obtained using
375    *          {@link org.apache.hadoop.fs.FileSystem#getFileStatus(
376    *          org.apache.hadoop.fs.Path)}.
377    * @return the fixed file trailer read
378    * @throws IOException if failed to read from the underlying stream, or the
379    *           trailer is corrupted, or the version of the trailer is
380    *           unsupported
381    */
382   public static FixedFileTrailer readFromStream(FSDataInputStream istream,
383       long fileSize) throws IOException {
384     int bufferSize = MAX_TRAILER_SIZE;
385     long seekPoint = fileSize - bufferSize;
386     if (seekPoint < 0) {
387       // It is hard to imagine such a small HFile.
388       seekPoint = 0;
389       bufferSize = (int) fileSize;
390     }
391 
392     istream.seek(seekPoint);
393     ByteBuffer buf = ByteBuffer.allocate(bufferSize);
394     istream.readFully(buf.array(), buf.arrayOffset(),
395         buf.arrayOffset() + buf.limit());
396 
397     // Read the version from the last int of the file.
398     buf.position(buf.limit() - Bytes.SIZEOF_INT);
399     int version = buf.getInt();
400 
401     // Extract the major and minor versions.
402     int majorVersion = extractMajorVersion(version);
403     int minorVersion = extractMinorVersion(version);
404 
405     HFile.checkFormatVersion(majorVersion); // throws IAE if invalid
406 
407     int trailerSize = getTrailerSize(majorVersion);
408 
409     FixedFileTrailer fft = new FixedFileTrailer(majorVersion, minorVersion);
410     fft.deserialize(new DataInputStream(new ByteArrayInputStream(buf.array(),
411         buf.arrayOffset() + bufferSize - trailerSize, trailerSize)));
412     return fft;
413   }
414 
415   public void expectMajorVersion(int expected) {
416     if (majorVersion != expected) {
417       throw new IllegalArgumentException("Invalid HFile major version: "
418           + majorVersion 
419           + " (expected: " + expected + ")");
420     }
421   }
422 
423   public void expectMinorVersion(int expected) {
424     if (minorVersion != expected) {
425       throw new IllegalArgumentException("Invalid HFile minor version: "
426           + minorVersion + " (expected: " + expected + ")");
427     }
428   }
429 
430   public void expectAtLeastMajorVersion(int lowerBound) {
431     if (majorVersion < lowerBound) {
432       throw new IllegalArgumentException("Invalid HFile major version: "
433           + majorVersion
434           + " (expected: " + lowerBound + " or higher).");
435     }
436   }
437 
438   public long getFileInfoOffset() {
439     return fileInfoOffset;
440   }
441 
442   public void setFileInfoOffset(long fileInfoOffset) {
443     this.fileInfoOffset = fileInfoOffset;
444   }
445 
446   public long getLoadOnOpenDataOffset() {
447     return loadOnOpenDataOffset;
448   }
449 
450   public void setLoadOnOpenOffset(long loadOnOpenDataOffset) {
451     this.loadOnOpenDataOffset = loadOnOpenDataOffset;
452   }
453 
454   public int getDataIndexCount() {
455     return dataIndexCount;
456   }
457 
458   public void setDataIndexCount(int dataIndexCount) {
459     this.dataIndexCount = dataIndexCount;
460   }
461 
462   public int getMetaIndexCount() {
463     return metaIndexCount;
464   }
465 
466   public void setMetaIndexCount(int metaIndexCount) {
467     this.metaIndexCount = metaIndexCount;
468   }
469 
470   public long getTotalUncompressedBytes() {
471     return totalUncompressedBytes;
472   }
473 
474   public void setTotalUncompressedBytes(long totalUncompressedBytes) {
475     this.totalUncompressedBytes = totalUncompressedBytes;
476   }
477 
478   public long getEntryCount() {
479     return entryCount;
480   }
481 
482   public void setEntryCount(long newEntryCount) {
483     entryCount = newEntryCount;
484   }
485 
486   public Compression.Algorithm getCompressionCodec() {
487     return compressionCodec;
488   }
489 
490   public void setCompressionCodec(Compression.Algorithm compressionCodec) {
491     this.compressionCodec = compressionCodec;
492   }
493 
494   public int getNumDataIndexLevels() {
495     expectAtLeastMajorVersion(2);
496     return numDataIndexLevels;
497   }
498 
499   public void setNumDataIndexLevels(int numDataIndexLevels) {
500     expectAtLeastMajorVersion(2);
501     this.numDataIndexLevels = numDataIndexLevels;
502   }
503 
504   public long getLastDataBlockOffset() {
505     expectAtLeastMajorVersion(2);
506     return lastDataBlockOffset;
507   }
508 
509   public void setLastDataBlockOffset(long lastDataBlockOffset) {
510     expectAtLeastMajorVersion(2);
511     this.lastDataBlockOffset = lastDataBlockOffset;
512   }
513 
514   public long getFirstDataBlockOffset() {
515     expectAtLeastMajorVersion(2);
516     return firstDataBlockOffset;
517   }
518 
519   public void setFirstDataBlockOffset(long firstDataBlockOffset) {
520     expectAtLeastMajorVersion(2);
521     this.firstDataBlockOffset = firstDataBlockOffset;
522   }
523 
524   public String getComparatorClassName() {
525     return comparatorClassName;
526   }
527 
528   /**
529    * Returns the major version of this HFile format
530    */
531   public int getMajorVersion() {
532     return majorVersion;
533   }
534 
535   /**
536    * Returns the minor version of this HFile format
537    */
538   public int getMinorVersion() {
539     return minorVersion;
540   }
541 
542   public void setComparatorClass(Class<? extends CellComparator> klass) {
543     // Is the comparator instantiable?
544     try {
545       // If null, it should be the Bytes.BYTES_RAWCOMPARATOR
546       if (klass != null) {
547         CellComparator comp = klass.newInstance();
548         // if the name wasn't one of the legacy names, maybe its a legit new
549         // kind of comparator.
550         comparatorClassName = klass.getName();
551       }
552 
553     } catch (Exception e) {
554       throw new RuntimeException("Comparator class " + klass.getName() + " is not instantiable", e);
555     }
556   }
557 
558   @SuppressWarnings("unchecked")
559   private static Class<? extends CellComparator> getComparatorClass(String comparatorClassName)
560       throws IOException {
561     Class<? extends CellComparator> comparatorKlass;
562     if (comparatorClassName.equals(KeyValue.COMPARATOR.getLegacyKeyComparatorName())
563         || comparatorClassName.equals(KeyValue.COMPARATOR.getClass().getName())) {
564       comparatorKlass = CellComparator.class;
565     } else if (comparatorClassName.equals(KeyValue.META_COMPARATOR.getLegacyKeyComparatorName())
566         || comparatorClassName.equals(KeyValue.META_COMPARATOR.getClass().getName())) {
567       comparatorKlass = MetaCellComparator.class;
568     } else if (comparatorClassName.equals(KeyValue.RAW_COMPARATOR.getClass().getName())
569         || comparatorClassName.equals(KeyValue.RAW_COMPARATOR.getLegacyKeyComparatorName())) {
570       // When the comparator to be used is Bytes.BYTES_RAWCOMPARATOR, we just return null from here
571       // Bytes.BYTES_RAWCOMPARATOR is not a CellComparator
572       comparatorKlass = null;
573     } else {
574       // if the name wasn't one of the legacy names, maybe its a legit new kind of comparator.
575       try {
576         comparatorKlass = (Class<? extends CellComparator>) Class.forName(comparatorClassName);
577       } catch (ClassNotFoundException e) {
578         throw new IOException(e);
579       }
580     }
581     return comparatorKlass;
582   }
583 
584   public static CellComparator createComparator(
585       String comparatorClassName) throws IOException {
586     try {
587       Class<? extends CellComparator> comparatorClass = getComparatorClass(comparatorClassName);
588       return comparatorClass != null ? comparatorClass.newInstance() : null;
589     } catch (InstantiationException e) {
590       throw new IOException("Comparator class " + comparatorClassName +
591         " is not instantiable", e);
592     } catch (IllegalAccessException e) {
593       throw new IOException("Comparator class " + comparatorClassName +
594         " is not instantiable", e);
595     }
596   }
597 
598   CellComparator createComparator() throws IOException {
599     expectAtLeastMajorVersion(2);
600     return createComparator(comparatorClassName);
601   }
602 
603   public long getUncompressedDataIndexSize() {
604     return uncompressedDataIndexSize;
605   }
606 
607   public void setUncompressedDataIndexSize(
608       long uncompressedDataIndexSize) {
609     expectAtLeastMajorVersion(2);
610     this.uncompressedDataIndexSize = uncompressedDataIndexSize;
611   }
612 
613   public byte[] getEncryptionKey() {
614     // This is a v3 feature but if reading a v2 file the encryptionKey will just be null which
615     // if fine for this feature.
616     expectAtLeastMajorVersion(2);
617     return encryptionKey;
618   }
619 
620   public void setEncryptionKey(byte[] keyBytes) {
621     this.encryptionKey = keyBytes;
622   }
623 
624   /**
625    * Extracts the major version for a 4-byte serialized version data.
626    * The major version is the 3 least significant bytes
627    */
628   private static int extractMajorVersion(int serializedVersion) {
629     return (serializedVersion & 0x00ffffff);
630   }
631 
632   /**
633    * Extracts the minor version for a 4-byte serialized version data.
634    * The major version are the 3 the most significant bytes
635    */
636   private static int extractMinorVersion(int serializedVersion) {
637     return (serializedVersion >>> 24);
638   }
639 
640   /**
641    * Create a 4 byte serialized version number by combining the
642    * minor and major version numbers.
643    */
644   static int materializeVersion(int majorVersion, int minorVersion) {
645     return ((majorVersion & 0x00ffffff) | (minorVersion << 24));
646   }
647 }