001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one
003     * or more contributor license agreements.  See the NOTICE file
004     * distributed with this work for additional information
005     * regarding copyright ownership.  The ASF licenses this file
006     * to you under the Apache License, Version 2.0 (the
007     * "License"); you may not use this file except in compliance
008     * with the License.  You may obtain a copy of the License at
009     *
010     *     http://www.apache.org/licenses/LICENSE-2.0
011     *
012     * Unless required by applicable law or agreed to in writing, software
013     * distributed under the License is distributed on an "AS IS" BASIS,
014     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015     * See the License for the specific language governing permissions and
016     * limitations under the License.
017     */
018    
019    package org.apache.hadoop.io;
020    
021    import java.io.*;
022    import java.util.*;
023    import java.rmi.server.UID;
024    import java.security.MessageDigest;
025    import org.apache.commons.logging.*;
026    import org.apache.hadoop.util.Options;
027    import org.apache.hadoop.fs.*;
028    import org.apache.hadoop.fs.Options.CreateOpts;
029    import org.apache.hadoop.io.compress.CodecPool;
030    import org.apache.hadoop.io.compress.CompressionCodec;
031    import org.apache.hadoop.io.compress.CompressionInputStream;
032    import org.apache.hadoop.io.compress.CompressionOutputStream;
033    import org.apache.hadoop.io.compress.Compressor;
034    import org.apache.hadoop.io.compress.Decompressor;
035    import org.apache.hadoop.io.compress.DefaultCodec;
036    import org.apache.hadoop.io.compress.GzipCodec;
037    import org.apache.hadoop.io.compress.zlib.ZlibFactory;
038    import org.apache.hadoop.io.serializer.Deserializer;
039    import org.apache.hadoop.io.serializer.Serializer;
040    import org.apache.hadoop.io.serializer.SerializationFactory;
041    import org.apache.hadoop.classification.InterfaceAudience;
042    import org.apache.hadoop.classification.InterfaceStability;
043    import org.apache.hadoop.conf.*;
044    import org.apache.hadoop.util.Progressable;
045    import org.apache.hadoop.util.Progress;
046    import org.apache.hadoop.util.ReflectionUtils;
047    import org.apache.hadoop.util.NativeCodeLoader;
048    import org.apache.hadoop.util.MergeSort;
049    import org.apache.hadoop.util.PriorityQueue;
050    import org.apache.hadoop.util.Time;
051    
052    /** 
053     * <code>SequenceFile</code>s are flat files consisting of binary key/value 
054     * pairs.
055     * 
056     * <p><code>SequenceFile</code> provides {@link Writer}, {@link Reader} and
057     * {@link Sorter} classes for writing, reading and sorting respectively.</p>
058     * 
059     * There are three <code>SequenceFile</code> <code>Writer</code>s based on the 
060     * {@link CompressionType} used to compress key/value pairs:
061     * <ol>
062     *   <li>
063     *   <code>Writer</code> : Uncompressed records.
064     *   </li>
065     *   <li>
066     *   <code>RecordCompressWriter</code> : Record-compressed files, only compress 
067     *                                       values.
068     *   </li>
069     *   <li>
070     *   <code>BlockCompressWriter</code> : Block-compressed files, both keys & 
071     *                                      values are collected in 'blocks' 
072     *                                      separately and compressed. The size of 
073     *                                      the 'block' is configurable.
074     * </ol>
075     * 
076     * <p>The actual compression algorithm used to compress key and/or values can be
077     * specified by using the appropriate {@link CompressionCodec}.</p>
078     * 
079     * <p>The recommended way is to use the static <tt>createWriter</tt> methods
080     * provided by the <code>SequenceFile</code> to chose the preferred format.</p>
081     *
082     * <p>The {@link Reader} acts as the bridge and can read any of the above 
083     * <code>SequenceFile</code> formats.</p>
084     *
085     * <h4 id="Formats">SequenceFile Formats</h4>
086     * 
087     * <p>Essentially there are 3 different formats for <code>SequenceFile</code>s
088     * depending on the <code>CompressionType</code> specified. All of them share a
089     * <a href="#Header">common header</a> described below.
090     * 
091     * <h5 id="Header">SequenceFile Header</h5>
092     * <ul>
093     *   <li>
094     *   version - 3 bytes of magic header <b>SEQ</b>, followed by 1 byte of actual 
095     *             version number (e.g. SEQ4 or SEQ6)
096     *   </li>
097     *   <li>
098     *   keyClassName -key class
099     *   </li>
100     *   <li>
101     *   valueClassName - value class
102     *   </li>
103     *   <li>
104     *   compression - A boolean which specifies if compression is turned on for 
105     *                 keys/values in this file.
106     *   </li>
107     *   <li>
108     *   blockCompression - A boolean which specifies if block-compression is 
109     *                      turned on for keys/values in this file.
110     *   </li>
111     *   <li>
112     *   compression codec - <code>CompressionCodec</code> class which is used for  
113     *                       compression of keys and/or values (if compression is 
114     *                       enabled).
115     *   </li>
116     *   <li>
117     *   metadata - {@link Metadata} for this file.
118     *   </li>
119     *   <li>
120     *   sync - A sync marker to denote end of the header.
121     *   </li>
122     * </ul>
123     * 
124     * <h5 id="#UncompressedFormat">Uncompressed SequenceFile Format</h5>
125     * <ul>
126     * <li>
127     * <a href="#Header">Header</a>
128     * </li>
129     * <li>
130     * Record
131     *   <ul>
132     *     <li>Record length</li>
133     *     <li>Key length</li>
134     *     <li>Key</li>
135     *     <li>Value</li>
136     *   </ul>
137     * </li>
138     * <li>
139     * A sync-marker every few <code>100</code> bytes or so.
140     * </li>
141     * </ul>
142     *
143     * <h5 id="#RecordCompressedFormat">Record-Compressed SequenceFile Format</h5>
144     * <ul>
145     * <li>
146     * <a href="#Header">Header</a>
147     * </li>
148     * <li>
149     * Record
150     *   <ul>
151     *     <li>Record length</li>
152     *     <li>Key length</li>
153     *     <li>Key</li>
154     *     <li><i>Compressed</i> Value</li>
155     *   </ul>
156     * </li>
157     * <li>
158     * A sync-marker every few <code>100</code> bytes or so.
159     * </li>
160     * </ul>
161     * 
162     * <h5 id="#BlockCompressedFormat">Block-Compressed SequenceFile Format</h5>
163     * <ul>
164     * <li>
165     * <a href="#Header">Header</a>
166     * </li>
167     * <li>
168     * Record <i>Block</i>
169     *   <ul>
170     *     <li>Uncompressed number of records in the block</li>
171     *     <li>Compressed key-lengths block-size</li>
172     *     <li>Compressed key-lengths block</li>
173     *     <li>Compressed keys block-size</li>
174     *     <li>Compressed keys block</li>
175     *     <li>Compressed value-lengths block-size</li>
176     *     <li>Compressed value-lengths block</li>
177     *     <li>Compressed values block-size</li>
178     *     <li>Compressed values block</li>
179     *   </ul>
180     * </li>
181     * <li>
182     * A sync-marker every block.
183     * </li>
184     * </ul>
185     * 
186     * <p>The compressed blocks of key lengths and value lengths consist of the 
187     * actual lengths of individual keys/values encoded in ZeroCompressedInteger 
188     * format.</p>
189     * 
190     * @see CompressionCodec
191     */
192    @InterfaceAudience.Public
193    @InterfaceStability.Stable
194    public class SequenceFile {
195      private static final Log LOG = LogFactory.getLog(SequenceFile.class);
196    
197      private SequenceFile() {}                         // no public ctor
198    
199      private static final byte BLOCK_COMPRESS_VERSION = (byte)4;
200      private static final byte CUSTOM_COMPRESS_VERSION = (byte)5;
201      private static final byte VERSION_WITH_METADATA = (byte)6;
202      private static byte[] VERSION = new byte[] {
203        (byte)'S', (byte)'E', (byte)'Q', VERSION_WITH_METADATA
204      };
205    
206      private static final int SYNC_ESCAPE = -1;      // "length" of sync entries
207      private static final int SYNC_HASH_SIZE = 16;   // number of bytes in hash 
208      private static final int SYNC_SIZE = 4+SYNC_HASH_SIZE; // escape + hash
209    
210      /** The number of bytes between sync points.*/
211      public static final int SYNC_INTERVAL = 100*SYNC_SIZE; 
212    
213      /** 
214       * The compression type used to compress key/value pairs in the 
215       * {@link SequenceFile}.
216       * 
217       * @see SequenceFile.Writer
218       */
219      public static enum CompressionType {
220        /** Do not compress records. */
221        NONE, 
222        /** Compress values only, each separately. */
223        RECORD,
224        /** Compress sequences of records together in blocks. */
225        BLOCK
226      }
227    
228      /**
229       * Get the compression type for the reduce outputs
230       * @param job the job config to look in
231       * @return the kind of compression to use
232       */
233      static public CompressionType getDefaultCompressionType(Configuration job) {
234        String name = job.get("io.seqfile.compression.type");
235        return name == null ? CompressionType.RECORD : 
236          CompressionType.valueOf(name);
237      }
238      
239      /**
240       * Set the default compression type for sequence files.
241       * @param job the configuration to modify
242       * @param val the new compression type (none, block, record)
243       */
244      static public void setDefaultCompressionType(Configuration job, 
245                                                   CompressionType val) {
246        job.set("io.seqfile.compression.type", val.toString());
247      }
248    
249      /**
250       * Create a new Writer with the given options.
251       * @param conf the configuration to use
252       * @param opts the options to create the file with
253       * @return a new Writer
254       * @throws IOException
255       */
256      public static Writer createWriter(Configuration conf, Writer.Option... opts
257                                        ) throws IOException {
258        Writer.CompressionOption compressionOption = 
259          Options.getOption(Writer.CompressionOption.class, opts);
260        CompressionType kind;
261        if (compressionOption != null) {
262          kind = compressionOption.getValue();
263        } else {
264          kind = getDefaultCompressionType(conf);
265          opts = Options.prependOptions(opts, Writer.compression(kind));
266        }
267        switch (kind) {
268          default:
269          case NONE:
270            return new Writer(conf, opts);
271          case RECORD:
272            return new RecordCompressWriter(conf, opts);
273          case BLOCK:
274            return new BlockCompressWriter(conf, opts);
275        }
276      }
277    
278      /**
279       * Construct the preferred type of SequenceFile Writer.
280       * @param fs The configured filesystem. 
281       * @param conf The configuration.
282       * @param name The name of the file. 
283       * @param keyClass The 'key' type.
284       * @param valClass The 'value' type.
285       * @return Returns the handle to the constructed SequenceFile Writer.
286       * @throws IOException
287       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
288       *     instead.
289       */
290      @Deprecated
291      public static Writer 
292        createWriter(FileSystem fs, Configuration conf, Path name, 
293                     Class keyClass, Class valClass) throws IOException {
294        return createWriter(conf, Writer.filesystem(fs),
295                            Writer.file(name), Writer.keyClass(keyClass),
296                            Writer.valueClass(valClass));
297      }
298      
299      /**
300       * Construct the preferred type of SequenceFile Writer.
301       * @param fs The configured filesystem. 
302       * @param conf The configuration.
303       * @param name The name of the file. 
304       * @param keyClass The 'key' type.
305       * @param valClass The 'value' type.
306       * @param compressionType The compression type.
307       * @return Returns the handle to the constructed SequenceFile Writer.
308       * @throws IOException
309       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
310       *     instead.
311       */
312      @Deprecated
313      public static Writer 
314        createWriter(FileSystem fs, Configuration conf, Path name, 
315                     Class keyClass, Class valClass, 
316                     CompressionType compressionType) throws IOException {
317        return createWriter(conf, Writer.filesystem(fs),
318                            Writer.file(name), Writer.keyClass(keyClass),
319                            Writer.valueClass(valClass), 
320                            Writer.compression(compressionType));
321      }
322      
323      /**
324       * Construct the preferred type of SequenceFile Writer.
325       * @param fs The configured filesystem. 
326       * @param conf The configuration.
327       * @param name The name of the file. 
328       * @param keyClass The 'key' type.
329       * @param valClass The 'value' type.
330       * @param compressionType The compression type.
331       * @param progress The Progressable object to track progress.
332       * @return Returns the handle to the constructed SequenceFile Writer.
333       * @throws IOException
334       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
335       *     instead.
336       */
337      @Deprecated
338      public static Writer
339        createWriter(FileSystem fs, Configuration conf, Path name, 
340                     Class keyClass, Class valClass, CompressionType compressionType,
341                     Progressable progress) throws IOException {
342        return createWriter(conf, Writer.file(name),
343                            Writer.filesystem(fs),
344                            Writer.keyClass(keyClass),
345                            Writer.valueClass(valClass), 
346                            Writer.compression(compressionType),
347                            Writer.progressable(progress));
348      }
349    
350      /**
351       * Construct the preferred type of SequenceFile Writer.
352       * @param fs The configured filesystem. 
353       * @param conf The configuration.
354       * @param name The name of the file. 
355       * @param keyClass The 'key' type.
356       * @param valClass The 'value' type.
357       * @param compressionType The compression type.
358       * @param codec The compression codec.
359       * @return Returns the handle to the constructed SequenceFile Writer.
360       * @throws IOException
361       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
362       *     instead.
363       */
364      @Deprecated
365      public static Writer 
366        createWriter(FileSystem fs, Configuration conf, Path name, 
367                     Class keyClass, Class valClass, CompressionType compressionType, 
368                     CompressionCodec codec) throws IOException {
369        return createWriter(conf, Writer.file(name),
370                            Writer.filesystem(fs),
371                            Writer.keyClass(keyClass),
372                            Writer.valueClass(valClass), 
373                            Writer.compression(compressionType, codec));
374      }
375      
376      /**
377       * Construct the preferred type of SequenceFile Writer.
378       * @param fs The configured filesystem. 
379       * @param conf The configuration.
380       * @param name The name of the file. 
381       * @param keyClass The 'key' type.
382       * @param valClass The 'value' type.
383       * @param compressionType The compression type.
384       * @param codec The compression codec.
385       * @param progress The Progressable object to track progress.
386       * @param metadata The metadata of the file.
387       * @return Returns the handle to the constructed SequenceFile Writer.
388       * @throws IOException
389       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
390       *     instead.
391       */
392      @Deprecated
393      public static Writer
394        createWriter(FileSystem fs, Configuration conf, Path name, 
395                     Class keyClass, Class valClass, 
396                     CompressionType compressionType, CompressionCodec codec,
397                     Progressable progress, Metadata metadata) throws IOException {
398        return createWriter(conf, Writer.file(name),
399                            Writer.filesystem(fs),
400                            Writer.keyClass(keyClass),
401                            Writer.valueClass(valClass),
402                            Writer.compression(compressionType, codec),
403                            Writer.progressable(progress),
404                            Writer.metadata(metadata));
405      }
406    
407      /**
408       * Construct the preferred type of SequenceFile Writer.
409       * @param fs The configured filesystem.
410       * @param conf The configuration.
411       * @param name The name of the file.
412       * @param keyClass The 'key' type.
413       * @param valClass The 'value' type.
414       * @param bufferSize buffer size for the underlaying outputstream.
415       * @param replication replication factor for the file.
416       * @param blockSize block size for the file.
417       * @param compressionType The compression type.
418       * @param codec The compression codec.
419       * @param progress The Progressable object to track progress.
420       * @param metadata The metadata of the file.
421       * @return Returns the handle to the constructed SequenceFile Writer.
422       * @throws IOException
423       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
424       *     instead.
425       */
426      @Deprecated
427      public static Writer
428        createWriter(FileSystem fs, Configuration conf, Path name,
429                     Class keyClass, Class valClass, int bufferSize,
430                     short replication, long blockSize,
431                     CompressionType compressionType, CompressionCodec codec,
432                     Progressable progress, Metadata metadata) throws IOException {
433        return createWriter(conf, Writer.file(name),
434                            Writer.filesystem(fs),
435                            Writer.keyClass(keyClass),
436                            Writer.valueClass(valClass), 
437                            Writer.bufferSize(bufferSize), 
438                            Writer.replication(replication),
439                            Writer.blockSize(blockSize),
440                            Writer.compression(compressionType, codec),
441                            Writer.progressable(progress),
442                            Writer.metadata(metadata));
443      }
444    
445      /**
446       * Construct the preferred type of SequenceFile Writer.
447       * @param fs The configured filesystem.
448       * @param conf The configuration.
449       * @param name The name of the file.
450       * @param keyClass The 'key' type.
451       * @param valClass The 'value' type.
452       * @param bufferSize buffer size for the underlaying outputstream.
453       * @param replication replication factor for the file.
454       * @param blockSize block size for the file.
455       * @param createParent create parent directory if non-existent
456       * @param compressionType The compression type.
457       * @param codec The compression codec.
458       * @param metadata The metadata of the file.
459       * @return Returns the handle to the constructed SequenceFile Writer.
460       * @throws IOException
461       */
462      @Deprecated
463      public static Writer
464      createWriter(FileSystem fs, Configuration conf, Path name,
465                   Class keyClass, Class valClass, int bufferSize,
466                   short replication, long blockSize, boolean createParent,
467                   CompressionType compressionType, CompressionCodec codec,
468                   Metadata metadata) throws IOException {
469        return createWriter(FileContext.getFileContext(fs.getUri(), conf),
470            conf, name, keyClass, valClass, compressionType, codec,
471            metadata, EnumSet.of(CreateFlag.CREATE,CreateFlag.OVERWRITE),
472            CreateOpts.bufferSize(bufferSize),
473            createParent ? CreateOpts.createParent()
474                         : CreateOpts.donotCreateParent(),
475            CreateOpts.repFac(replication),
476            CreateOpts.blockSize(blockSize)
477          );
478      }
479    
480      /**
481       * Construct the preferred type of SequenceFile Writer.
482       * @param fc The context for the specified file.
483       * @param conf The configuration.
484       * @param name The name of the file.
485       * @param keyClass The 'key' type.
486       * @param valClass The 'value' type.
487       * @param compressionType The compression type.
488       * @param codec The compression codec.
489       * @param metadata The metadata of the file.
490       * @param createFlag gives the semantics of create: overwrite, append etc.
491       * @param opts file creation options; see {@link CreateOpts}.
492       * @return Returns the handle to the constructed SequenceFile Writer.
493       * @throws IOException
494       */
495      public static Writer
496      createWriter(FileContext fc, Configuration conf, Path name,
497                   Class keyClass, Class valClass,
498                   CompressionType compressionType, CompressionCodec codec,
499                   Metadata metadata,
500                   final EnumSet<CreateFlag> createFlag, CreateOpts... opts)
501                   throws IOException {
502        return createWriter(conf, fc.create(name, createFlag, opts),
503              keyClass, valClass, compressionType, codec, metadata).ownStream();
504      }
505    
506      /**
507       * Construct the preferred type of SequenceFile Writer.
508       * @param fs The configured filesystem. 
509       * @param conf The configuration.
510       * @param name The name of the file. 
511       * @param keyClass The 'key' type.
512       * @param valClass The 'value' type.
513       * @param compressionType The compression type.
514       * @param codec The compression codec.
515       * @param progress The Progressable object to track progress.
516       * @return Returns the handle to the constructed SequenceFile Writer.
517       * @throws IOException
518       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
519       *     instead.
520       */
521      @Deprecated
522      public static Writer
523        createWriter(FileSystem fs, Configuration conf, Path name, 
524                     Class keyClass, Class valClass, 
525                     CompressionType compressionType, CompressionCodec codec,
526                     Progressable progress) throws IOException {
527        return createWriter(conf, Writer.file(name),
528                            Writer.filesystem(fs),
529                            Writer.keyClass(keyClass),
530                            Writer.valueClass(valClass),
531                            Writer.compression(compressionType, codec),
532                            Writer.progressable(progress));
533      }
534    
535      /**
536       * Construct the preferred type of 'raw' SequenceFile Writer.
537       * @param conf The configuration.
538       * @param out The stream on top which the writer is to be constructed.
539       * @param keyClass The 'key' type.
540       * @param valClass The 'value' type.
541       * @param compressionType The compression type.
542       * @param codec The compression codec.
543       * @param metadata The metadata of the file.
544       * @return Returns the handle to the constructed SequenceFile Writer.
545       * @throws IOException
546       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
547       *     instead.
548       */
549      @Deprecated
550      public static Writer
551        createWriter(Configuration conf, FSDataOutputStream out, 
552                     Class keyClass, Class valClass,
553                     CompressionType compressionType,
554                     CompressionCodec codec, Metadata metadata) throws IOException {
555        return createWriter(conf, Writer.stream(out), Writer.keyClass(keyClass),
556                            Writer.valueClass(valClass), 
557                            Writer.compression(compressionType, codec),
558                            Writer.metadata(metadata));
559      }
560      
561      /**
562       * Construct the preferred type of 'raw' SequenceFile Writer.
563       * @param conf The configuration.
564       * @param out The stream on top which the writer is to be constructed.
565       * @param keyClass The 'key' type.
566       * @param valClass The 'value' type.
567       * @param compressionType The compression type.
568       * @param codec The compression codec.
569       * @return Returns the handle to the constructed SequenceFile Writer.
570       * @throws IOException
571       * @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
572       *     instead.
573       */
574      @Deprecated
575      public static Writer
576        createWriter(Configuration conf, FSDataOutputStream out, 
577                     Class keyClass, Class valClass, CompressionType compressionType,
578                     CompressionCodec codec) throws IOException {
579        return createWriter(conf, Writer.stream(out), Writer.keyClass(keyClass),
580                            Writer.valueClass(valClass),
581                            Writer.compression(compressionType, codec));
582      }
583      
584    
585      /** The interface to 'raw' values of SequenceFiles. */
586      public static interface ValueBytes {
587    
588        /** Writes the uncompressed bytes to the outStream.
589         * @param outStream : Stream to write uncompressed bytes into.
590         * @throws IOException
591         */
592        public void writeUncompressedBytes(DataOutputStream outStream)
593          throws IOException;
594    
595        /** Write compressed bytes to outStream. 
596         * Note: that it will NOT compress the bytes if they are not compressed.
597         * @param outStream : Stream to write compressed bytes into.
598         */
599        public void writeCompressedBytes(DataOutputStream outStream) 
600          throws IllegalArgumentException, IOException;
601    
602        /**
603         * Size of stored data.
604         */
605        public int getSize();
606      }
607      
608      private static class UncompressedBytes implements ValueBytes {
609        private int dataSize;
610        private byte[] data;
611        
612        private UncompressedBytes() {
613          data = null;
614          dataSize = 0;
615        }
616        
617        private void reset(DataInputStream in, int length) throws IOException {
618          if (data == null) {
619            data = new byte[length];
620          } else if (length > data.length) {
621            data = new byte[Math.max(length, data.length * 2)];
622          }
623          dataSize = -1;
624          in.readFully(data, 0, length);
625          dataSize = length;
626        }
627        
628        @Override
629        public int getSize() {
630          return dataSize;
631        }
632        
633        @Override
634        public void writeUncompressedBytes(DataOutputStream outStream)
635          throws IOException {
636          outStream.write(data, 0, dataSize);
637        }
638    
639        @Override
640        public void writeCompressedBytes(DataOutputStream outStream) 
641          throws IllegalArgumentException, IOException {
642          throw 
643            new IllegalArgumentException("UncompressedBytes cannot be compressed!");
644        }
645    
646      } // UncompressedBytes
647      
648      private static class CompressedBytes implements ValueBytes {
649        private int dataSize;
650        private byte[] data;
651        DataInputBuffer rawData = null;
652        CompressionCodec codec = null;
653        CompressionInputStream decompressedStream = null;
654    
655        private CompressedBytes(CompressionCodec codec) {
656          data = null;
657          dataSize = 0;
658          this.codec = codec;
659        }
660    
661        private void reset(DataInputStream in, int length) throws IOException {
662          if (data == null) {
663            data = new byte[length];
664          } else if (length > data.length) {
665            data = new byte[Math.max(length, data.length * 2)];
666          } 
667          dataSize = -1;
668          in.readFully(data, 0, length);
669          dataSize = length;
670        }
671        
672        @Override
673        public int getSize() {
674          return dataSize;
675        }
676        
677        @Override
678        public void writeUncompressedBytes(DataOutputStream outStream)
679          throws IOException {
680          if (decompressedStream == null) {
681            rawData = new DataInputBuffer();
682            decompressedStream = codec.createInputStream(rawData);
683          } else {
684            decompressedStream.resetState();
685          }
686          rawData.reset(data, 0, dataSize);
687    
688          byte[] buffer = new byte[8192];
689          int bytesRead = 0;
690          while ((bytesRead = decompressedStream.read(buffer, 0, 8192)) != -1) {
691            outStream.write(buffer, 0, bytesRead);
692          }
693        }
694    
695        @Override
696        public void writeCompressedBytes(DataOutputStream outStream) 
697          throws IllegalArgumentException, IOException {
698          outStream.write(data, 0, dataSize);
699        }
700    
701      } // CompressedBytes
702      
703      /**
704       * The class encapsulating with the metadata of a file.
705       * The metadata of a file is a list of attribute name/value
706       * pairs of Text type.
707       *
708       */
709      public static class Metadata implements Writable {
710    
711        private TreeMap<Text, Text> theMetadata;
712        
713        public Metadata() {
714          this(new TreeMap<Text, Text>());
715        }
716        
717        public Metadata(TreeMap<Text, Text> arg) {
718          if (arg == null) {
719            this.theMetadata = new TreeMap<Text, Text>();
720          } else {
721            this.theMetadata = arg;
722          }
723        }
724        
725        public Text get(Text name) {
726          return this.theMetadata.get(name);
727        }
728        
729        public void set(Text name, Text value) {
730          this.theMetadata.put(name, value);
731        }
732        
733        public TreeMap<Text, Text> getMetadata() {
734          return new TreeMap<Text, Text>(this.theMetadata);
735        }
736        
737        @Override
738        public void write(DataOutput out) throws IOException {
739          out.writeInt(this.theMetadata.size());
740          Iterator<Map.Entry<Text, Text>> iter =
741            this.theMetadata.entrySet().iterator();
742          while (iter.hasNext()) {
743            Map.Entry<Text, Text> en = iter.next();
744            en.getKey().write(out);
745            en.getValue().write(out);
746          }
747        }
748    
749        @Override
750        public void readFields(DataInput in) throws IOException {
751          int sz = in.readInt();
752          if (sz < 0) throw new IOException("Invalid size: " + sz + " for file metadata object");
753          this.theMetadata = new TreeMap<Text, Text>();
754          for (int i = 0; i < sz; i++) {
755            Text key = new Text();
756            Text val = new Text();
757            key.readFields(in);
758            val.readFields(in);
759            this.theMetadata.put(key, val);
760          }    
761        }
762    
763        @Override
764        public boolean equals(Object other) {
765          if (other == null) {
766            return false;
767          }
768          if (other.getClass() != this.getClass()) {
769            return false;
770          } else {
771            return equals((Metadata)other);
772          }
773        }
774        
775        public boolean equals(Metadata other) {
776          if (other == null) return false;
777          if (this.theMetadata.size() != other.theMetadata.size()) {
778            return false;
779          }
780          Iterator<Map.Entry<Text, Text>> iter1 =
781            this.theMetadata.entrySet().iterator();
782          Iterator<Map.Entry<Text, Text>> iter2 =
783            other.theMetadata.entrySet().iterator();
784          while (iter1.hasNext() && iter2.hasNext()) {
785            Map.Entry<Text, Text> en1 = iter1.next();
786            Map.Entry<Text, Text> en2 = iter2.next();
787            if (!en1.getKey().equals(en2.getKey())) {
788              return false;
789            }
790            if (!en1.getValue().equals(en2.getValue())) {
791              return false;
792            }
793          }
794          if (iter1.hasNext() || iter2.hasNext()) {
795            return false;
796          }
797          return true;
798        }
799    
800        @Override
801        public int hashCode() {
802          assert false : "hashCode not designed";
803          return 42; // any arbitrary constant will do 
804        }
805        
806        @Override
807        public String toString() {
808          StringBuilder sb = new StringBuilder();
809          sb.append("size: ").append(this.theMetadata.size()).append("\n");
810          Iterator<Map.Entry<Text, Text>> iter =
811            this.theMetadata.entrySet().iterator();
812          while (iter.hasNext()) {
813            Map.Entry<Text, Text> en = iter.next();
814            sb.append("\t").append(en.getKey().toString()).append("\t").append(en.getValue().toString());
815            sb.append("\n");
816          }
817          return sb.toString();
818        }
819      }
820      
821      /** Write key/value pairs to a sequence-format file. */
822      public static class Writer implements java.io.Closeable, Syncable {
823        private Configuration conf;
824        FSDataOutputStream out;
825        boolean ownOutputStream = true;
826        DataOutputBuffer buffer = new DataOutputBuffer();
827    
828        Class keyClass;
829        Class valClass;
830    
831        private final CompressionType compress;
832        CompressionCodec codec = null;
833        CompressionOutputStream deflateFilter = null;
834        DataOutputStream deflateOut = null;
835        Metadata metadata = null;
836        Compressor compressor = null;
837        
838        protected Serializer keySerializer;
839        protected Serializer uncompressedValSerializer;
840        protected Serializer compressedValSerializer;
841        
842        // Insert a globally unique 16-byte value every few entries, so that one
843        // can seek into the middle of a file and then synchronize with record
844        // starts and ends by scanning for this value.
845        long lastSyncPos;                     // position of last sync
846        byte[] sync;                          // 16 random bytes
847        {
848          try {                                       
849            MessageDigest digester = MessageDigest.getInstance("MD5");
850            long time = Time.now();
851            digester.update((new UID()+"@"+time).getBytes());
852            sync = digester.digest();
853          } catch (Exception e) {
854            throw new RuntimeException(e);
855          }
856        }
857    
858        public static interface Option {}
859        
860        static class FileOption extends Options.PathOption 
861                                        implements Option {
862          FileOption(Path path) {
863            super(path);
864          }
865        }
866    
867        /**
868         * @deprecated only used for backwards-compatibility in the createWriter methods
869         * that take FileSystem.
870         */
871        @Deprecated
872        private static class FileSystemOption implements Option {
873          private final FileSystem value;
874          protected FileSystemOption(FileSystem value) {
875            this.value = value;
876          }
877          public FileSystem getValue() {
878            return value;
879          }
880        }
881    
882        static class StreamOption extends Options.FSDataOutputStreamOption 
883                                  implements Option {
884          StreamOption(FSDataOutputStream stream) {
885            super(stream);
886          }
887        }
888    
889        static class BufferSizeOption extends Options.IntegerOption
890                                      implements Option {
891          BufferSizeOption(int value) {
892            super(value);
893          }
894        }
895        
896        static class BlockSizeOption extends Options.LongOption implements Option {
897          BlockSizeOption(long value) {
898            super(value);
899          }
900        }
901    
902        static class ReplicationOption extends Options.IntegerOption
903                                       implements Option {
904          ReplicationOption(int value) {
905            super(value);
906          }
907        }
908    
909        static class KeyClassOption extends Options.ClassOption implements Option {
910          KeyClassOption(Class<?> value) {
911            super(value);
912          }
913        }
914    
915        static class ValueClassOption extends Options.ClassOption
916                                              implements Option {
917          ValueClassOption(Class<?> value) {
918            super(value);
919          }
920        }
921    
922        static class MetadataOption implements Option {
923          private final Metadata value;
924          MetadataOption(Metadata value) {
925            this.value = value;
926          }
927          Metadata getValue() {
928            return value;
929          }
930        }
931    
932        static class ProgressableOption extends Options.ProgressableOption
933                                        implements Option {
934          ProgressableOption(Progressable value) {
935            super(value);
936          }
937        }
938    
939        private static class CompressionOption implements Option {
940          private final CompressionType value;
941          private final CompressionCodec codec;
942          CompressionOption(CompressionType value) {
943            this(value, null);
944          }
945          CompressionOption(CompressionType value, CompressionCodec codec) {
946            this.value = value;
947            this.codec = (CompressionType.NONE != value && null == codec)
948              ? new DefaultCodec()
949              : codec;
950          }
951          CompressionType getValue() {
952            return value;
953          }
954          CompressionCodec getCodec() {
955            return codec;
956          }
957        }
958        
959        public static Option file(Path value) {
960          return new FileOption(value);
961        }
962    
963        /**
964         * @deprecated only used for backwards-compatibility in the createWriter methods
965         * that take FileSystem.
966         */
967        @Deprecated
968        private static Option filesystem(FileSystem fs) {
969          return new SequenceFile.Writer.FileSystemOption(fs);
970        }
971        
972        public static Option bufferSize(int value) {
973          return new BufferSizeOption(value);
974        }
975        
976        public static Option stream(FSDataOutputStream value) {
977          return new StreamOption(value);
978        }
979        
980        public static Option replication(short value) {
981          return new ReplicationOption(value);
982        }
983        
984        public static Option blockSize(long value) {
985          return new BlockSizeOption(value);
986        }
987        
988        public static Option progressable(Progressable value) {
989          return new ProgressableOption(value);
990        }
991    
992        public static Option keyClass(Class<?> value) {
993          return new KeyClassOption(value);
994        }
995        
996        public static Option valueClass(Class<?> value) {
997          return new ValueClassOption(value);
998        }
999        
1000        public static Option metadata(Metadata value) {
1001          return new MetadataOption(value);
1002        }
1003    
1004        public static Option compression(CompressionType value) {
1005          return new CompressionOption(value);
1006        }
1007    
1008        public static Option compression(CompressionType value,
1009            CompressionCodec codec) {
1010          return new CompressionOption(value, codec);
1011        }
1012        
1013        /**
1014         * Construct a uncompressed writer from a set of options.
1015         * @param conf the configuration to use
1016         * @param options the options used when creating the writer
1017         * @throws IOException if it fails
1018         */
1019        Writer(Configuration conf, 
1020               Option... opts) throws IOException {
1021          BlockSizeOption blockSizeOption = 
1022            Options.getOption(BlockSizeOption.class, opts);
1023          BufferSizeOption bufferSizeOption = 
1024            Options.getOption(BufferSizeOption.class, opts);
1025          ReplicationOption replicationOption = 
1026            Options.getOption(ReplicationOption.class, opts);
1027          ProgressableOption progressOption = 
1028            Options.getOption(ProgressableOption.class, opts);
1029          FileOption fileOption = Options.getOption(FileOption.class, opts);
1030          FileSystemOption fsOption = Options.getOption(FileSystemOption.class, opts);
1031          StreamOption streamOption = Options.getOption(StreamOption.class, opts);
1032          KeyClassOption keyClassOption = 
1033            Options.getOption(KeyClassOption.class, opts);
1034          ValueClassOption valueClassOption = 
1035            Options.getOption(ValueClassOption.class, opts);
1036          MetadataOption metadataOption = 
1037            Options.getOption(MetadataOption.class, opts);
1038          CompressionOption compressionTypeOption =
1039            Options.getOption(CompressionOption.class, opts);
1040          // check consistency of options
1041          if ((fileOption == null) == (streamOption == null)) {
1042            throw new IllegalArgumentException("file or stream must be specified");
1043          }
1044          if (fileOption == null && (blockSizeOption != null ||
1045                                     bufferSizeOption != null ||
1046                                     replicationOption != null ||
1047                                     progressOption != null)) {
1048            throw new IllegalArgumentException("file modifier options not " +
1049                                               "compatible with stream");
1050          }
1051    
1052          FSDataOutputStream out;
1053          boolean ownStream = fileOption != null;
1054          if (ownStream) {
1055            Path p = fileOption.getValue();
1056            FileSystem fs;
1057            if (fsOption != null) {
1058              fs = fsOption.getValue();
1059            } else {
1060              fs = p.getFileSystem(conf);
1061            }
1062            int bufferSize = bufferSizeOption == null ? getBufferSize(conf) :
1063              bufferSizeOption.getValue();
1064            short replication = replicationOption == null ? 
1065              fs.getDefaultReplication(p) :
1066              (short) replicationOption.getValue();
1067            long blockSize = blockSizeOption == null ? fs.getDefaultBlockSize(p) :
1068              blockSizeOption.getValue();
1069            Progressable progress = progressOption == null ? null :
1070              progressOption.getValue();
1071            out = fs.create(p, true, bufferSize, replication, blockSize, progress);
1072          } else {
1073            out = streamOption.getValue();
1074          }
1075          Class<?> keyClass = keyClassOption == null ?
1076              Object.class : keyClassOption.getValue();
1077          Class<?> valueClass = valueClassOption == null ?
1078              Object.class : valueClassOption.getValue();
1079          Metadata metadata = metadataOption == null ?
1080              new Metadata() : metadataOption.getValue();
1081          this.compress = compressionTypeOption.getValue();
1082          final CompressionCodec codec = compressionTypeOption.getCodec();
1083          if (codec != null &&
1084              (codec instanceof GzipCodec) &&
1085              !NativeCodeLoader.isNativeCodeLoaded() &&
1086              !ZlibFactory.isNativeZlibLoaded(conf)) {
1087            throw new IllegalArgumentException("SequenceFile doesn't work with " +
1088                                               "GzipCodec without native-hadoop " +
1089                                               "code!");
1090          }
1091          init(conf, out, ownStream, keyClass, valueClass, codec, metadata);
1092        }
1093    
1094        /** Create the named file.
1095         * @deprecated Use 
1096         *   {@link SequenceFile#createWriter(Configuration, Writer.Option...)} 
1097         *   instead.
1098         */
1099        @Deprecated
1100        public Writer(FileSystem fs, Configuration conf, Path name, 
1101                      Class keyClass, Class valClass) throws IOException {
1102          this.compress = CompressionType.NONE;
1103          init(conf, fs.create(name), true, keyClass, valClass, null, 
1104               new Metadata());
1105        }
1106        
1107        /** Create the named file with write-progress reporter.
1108         * @deprecated Use 
1109         *   {@link SequenceFile#createWriter(Configuration, Writer.Option...)} 
1110         *   instead.
1111         */
1112        @Deprecated
1113        public Writer(FileSystem fs, Configuration conf, Path name, 
1114                      Class keyClass, Class valClass,
1115                      Progressable progress, Metadata metadata) throws IOException {
1116          this.compress = CompressionType.NONE;
1117          init(conf, fs.create(name, progress), true, keyClass, valClass,
1118               null, metadata);
1119        }
1120        
1121        /** Create the named file with write-progress reporter. 
1122         * @deprecated Use 
1123         *   {@link SequenceFile#createWriter(Configuration, Writer.Option...)} 
1124         *   instead.
1125         */
1126        @Deprecated
1127        public Writer(FileSystem fs, Configuration conf, Path name,
1128                      Class keyClass, Class valClass,
1129                      int bufferSize, short replication, long blockSize,
1130                      Progressable progress, Metadata metadata) throws IOException {
1131          this.compress = CompressionType.NONE;
1132          init(conf,
1133               fs.create(name, true, bufferSize, replication, blockSize, progress),
1134               true, keyClass, valClass, null, metadata);
1135        }
1136    
1137        boolean isCompressed() { return compress != CompressionType.NONE; }
1138        boolean isBlockCompressed() { return compress == CompressionType.BLOCK; }
1139        
1140        Writer ownStream() { this.ownOutputStream = true; return this;  }
1141    
1142        /** Write and flush the file header. */
1143        private void writeFileHeader() 
1144          throws IOException {
1145          out.write(VERSION);
1146          Text.writeString(out, keyClass.getName());
1147          Text.writeString(out, valClass.getName());
1148          
1149          out.writeBoolean(this.isCompressed());
1150          out.writeBoolean(this.isBlockCompressed());
1151          
1152          if (this.isCompressed()) {
1153            Text.writeString(out, (codec.getClass()).getName());
1154          }
1155          this.metadata.write(out);
1156          out.write(sync);                       // write the sync bytes
1157          out.flush();                           // flush header
1158        }
1159        
1160        /** Initialize. */
1161        @SuppressWarnings("unchecked")
1162        void init(Configuration conf, FSDataOutputStream out, boolean ownStream,
1163                  Class keyClass, Class valClass,
1164                  CompressionCodec codec, Metadata metadata) 
1165          throws IOException {
1166          this.conf = conf;
1167          this.out = out;
1168          this.ownOutputStream = ownStream;
1169          this.keyClass = keyClass;
1170          this.valClass = valClass;
1171          this.codec = codec;
1172          this.metadata = metadata;
1173          SerializationFactory serializationFactory = new SerializationFactory(conf);
1174          this.keySerializer = serializationFactory.getSerializer(keyClass);
1175          if (this.keySerializer == null) {
1176            throw new IOException(
1177                "Could not find a serializer for the Key class: '"
1178                    + keyClass.getCanonicalName() + "'. "
1179                    + "Please ensure that the configuration '" +
1180                    CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
1181                    + "properly configured, if you're using"
1182                    + "custom serialization.");
1183          }
1184          this.keySerializer.open(buffer);
1185          this.uncompressedValSerializer = serializationFactory.getSerializer(valClass);
1186          if (this.uncompressedValSerializer == null) {
1187            throw new IOException(
1188                "Could not find a serializer for the Value class: '"
1189                    + valClass.getCanonicalName() + "'. "
1190                    + "Please ensure that the configuration '" +
1191                    CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
1192                    + "properly configured, if you're using"
1193                    + "custom serialization.");
1194          }
1195          this.uncompressedValSerializer.open(buffer);
1196          if (this.codec != null) {
1197            ReflectionUtils.setConf(this.codec, this.conf);
1198            this.compressor = CodecPool.getCompressor(this.codec);
1199            this.deflateFilter = this.codec.createOutputStream(buffer, compressor);
1200            this.deflateOut = 
1201              new DataOutputStream(new BufferedOutputStream(deflateFilter));
1202            this.compressedValSerializer = serializationFactory.getSerializer(valClass);
1203            if (this.compressedValSerializer == null) {
1204              throw new IOException(
1205                  "Could not find a serializer for the Value class: '"
1206                      + valClass.getCanonicalName() + "'. "
1207                      + "Please ensure that the configuration '" +
1208                      CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
1209                      + "properly configured, if you're using"
1210                      + "custom serialization.");
1211            }
1212            this.compressedValSerializer.open(deflateOut);
1213          }
1214          writeFileHeader();
1215        }
1216        
1217        /** Returns the class of keys in this file. */
1218        public Class getKeyClass() { return keyClass; }
1219    
1220        /** Returns the class of values in this file. */
1221        public Class getValueClass() { return valClass; }
1222    
1223        /** Returns the compression codec of data in this file. */
1224        public CompressionCodec getCompressionCodec() { return codec; }
1225        
1226        /** create a sync point */
1227        public void sync() throws IOException {
1228          if (sync != null && lastSyncPos != out.getPos()) {
1229            out.writeInt(SYNC_ESCAPE);                // mark the start of the sync
1230            out.write(sync);                          // write sync
1231            lastSyncPos = out.getPos();               // update lastSyncPos
1232          }
1233        }
1234    
1235        /**
1236         * flush all currently written data to the file system
1237         * @deprecated Use {@link #hsync()} or {@link #hflush()} instead
1238         */
1239        @Deprecated
1240        public void syncFs() throws IOException {
1241          if (out != null) {
1242            out.sync();                               // flush contents to file system
1243          }
1244        }
1245    
1246        @Override
1247        public void hsync() throws IOException {
1248          if (out != null) {
1249            out.hsync();
1250          }
1251        }
1252    
1253        @Override
1254        public void hflush() throws IOException {
1255          if (out != null) {
1256            out.hflush();
1257          }
1258        }
1259        
1260        /** Returns the configuration of this file. */
1261        Configuration getConf() { return conf; }
1262        
1263        /** Close the file. */
1264        @Override
1265        public synchronized void close() throws IOException {
1266          keySerializer.close();
1267          uncompressedValSerializer.close();
1268          if (compressedValSerializer != null) {
1269            compressedValSerializer.close();
1270          }
1271    
1272          CodecPool.returnCompressor(compressor);
1273          compressor = null;
1274          
1275          if (out != null) {
1276            
1277            // Close the underlying stream iff we own it...
1278            if (ownOutputStream) {
1279              out.close();
1280            } else {
1281              out.flush();
1282            }
1283            out = null;
1284          }
1285        }
1286    
1287        synchronized void checkAndWriteSync() throws IOException {
1288          if (sync != null &&
1289              out.getPos() >= lastSyncPos+SYNC_INTERVAL) { // time to emit sync
1290            sync();
1291          }
1292        }
1293    
1294        /** Append a key/value pair. */
1295        public void append(Writable key, Writable val)
1296          throws IOException {
1297          append((Object) key, (Object) val);
1298        }
1299    
1300        /** Append a key/value pair. */
1301        @SuppressWarnings("unchecked")
1302        public synchronized void append(Object key, Object val)
1303          throws IOException {
1304          if (key.getClass() != keyClass)
1305            throw new IOException("wrong key class: "+key.getClass().getName()
1306                                  +" is not "+keyClass);
1307          if (val.getClass() != valClass)
1308            throw new IOException("wrong value class: "+val.getClass().getName()
1309                                  +" is not "+valClass);
1310    
1311          buffer.reset();
1312    
1313          // Append the 'key'
1314          keySerializer.serialize(key);
1315          int keyLength = buffer.getLength();
1316          if (keyLength < 0)
1317            throw new IOException("negative length keys not allowed: " + key);
1318    
1319          // Append the 'value'
1320          if (compress == CompressionType.RECORD) {
1321            deflateFilter.resetState();
1322            compressedValSerializer.serialize(val);
1323            deflateOut.flush();
1324            deflateFilter.finish();
1325          } else {
1326            uncompressedValSerializer.serialize(val);
1327          }
1328    
1329          // Write the record out
1330          checkAndWriteSync();                                // sync
1331          out.writeInt(buffer.getLength());                   // total record length
1332          out.writeInt(keyLength);                            // key portion length
1333          out.write(buffer.getData(), 0, buffer.getLength()); // data
1334        }
1335    
1336        public synchronized void appendRaw(byte[] keyData, int keyOffset,
1337            int keyLength, ValueBytes val) throws IOException {
1338          if (keyLength < 0)
1339            throw new IOException("negative length keys not allowed: " + keyLength);
1340    
1341          int valLength = val.getSize();
1342    
1343          checkAndWriteSync();
1344          
1345          out.writeInt(keyLength+valLength);          // total record length
1346          out.writeInt(keyLength);                    // key portion length
1347          out.write(keyData, keyOffset, keyLength);   // key
1348          val.writeUncompressedBytes(out);            // value
1349        }
1350    
1351        /** Returns the current length of the output file.
1352         *
1353         * <p>This always returns a synchronized position.  In other words,
1354         * immediately after calling {@link SequenceFile.Reader#seek(long)} with a position
1355         * returned by this method, {@link SequenceFile.Reader#next(Writable)} may be called.  However
1356         * the key may be earlier in the file than key last written when this
1357         * method was called (e.g., with block-compression, it may be the first key
1358         * in the block that was being written when this method was called).
1359         */
1360        public synchronized long getLength() throws IOException {
1361          return out.getPos();
1362        }
1363    
1364      } // class Writer
1365    
1366      /** Write key/compressed-value pairs to a sequence-format file. */
1367      static class RecordCompressWriter extends Writer {
1368        
1369        RecordCompressWriter(Configuration conf, 
1370                             Option... options) throws IOException {
1371          super(conf, options);
1372        }
1373    
1374        /** Append a key/value pair. */
1375        @Override
1376        @SuppressWarnings("unchecked")
1377        public synchronized void append(Object key, Object val)
1378          throws IOException {
1379          if (key.getClass() != keyClass)
1380            throw new IOException("wrong key class: "+key.getClass().getName()
1381                                  +" is not "+keyClass);
1382          if (val.getClass() != valClass)
1383            throw new IOException("wrong value class: "+val.getClass().getName()
1384                                  +" is not "+valClass);
1385    
1386          buffer.reset();
1387    
1388          // Append the 'key'
1389          keySerializer.serialize(key);
1390          int keyLength = buffer.getLength();
1391          if (keyLength < 0)
1392            throw new IOException("negative length keys not allowed: " + key);
1393    
1394          // Compress 'value' and append it
1395          deflateFilter.resetState();
1396          compressedValSerializer.serialize(val);
1397          deflateOut.flush();
1398          deflateFilter.finish();
1399    
1400          // Write the record out
1401          checkAndWriteSync();                                // sync
1402          out.writeInt(buffer.getLength());                   // total record length
1403          out.writeInt(keyLength);                            // key portion length
1404          out.write(buffer.getData(), 0, buffer.getLength()); // data
1405        }
1406    
1407        /** Append a key/value pair. */
1408        @Override
1409        public synchronized void appendRaw(byte[] keyData, int keyOffset,
1410            int keyLength, ValueBytes val) throws IOException {
1411    
1412          if (keyLength < 0)
1413            throw new IOException("negative length keys not allowed: " + keyLength);
1414    
1415          int valLength = val.getSize();
1416          
1417          checkAndWriteSync();                        // sync
1418          out.writeInt(keyLength+valLength);          // total record length
1419          out.writeInt(keyLength);                    // key portion length
1420          out.write(keyData, keyOffset, keyLength);   // 'key' data
1421          val.writeCompressedBytes(out);              // 'value' data
1422        }
1423        
1424      } // RecordCompressionWriter
1425    
1426      /** Write compressed key/value blocks to a sequence-format file. */
1427      static class BlockCompressWriter extends Writer {
1428        
1429        private int noBufferedRecords = 0;
1430        
1431        private DataOutputBuffer keyLenBuffer = new DataOutputBuffer();
1432        private DataOutputBuffer keyBuffer = new DataOutputBuffer();
1433    
1434        private DataOutputBuffer valLenBuffer = new DataOutputBuffer();
1435        private DataOutputBuffer valBuffer = new DataOutputBuffer();
1436    
1437        private final int compressionBlockSize;
1438        
1439        BlockCompressWriter(Configuration conf,
1440                            Option... options) throws IOException {
1441          super(conf, options);
1442          compressionBlockSize = 
1443            conf.getInt("io.seqfile.compress.blocksize", 1000000);
1444          keySerializer.close();
1445          keySerializer.open(keyBuffer);
1446          uncompressedValSerializer.close();
1447          uncompressedValSerializer.open(valBuffer);
1448        }
1449    
1450        /** Workhorse to check and write out compressed data/lengths */
1451        private synchronized 
1452          void writeBuffer(DataOutputBuffer uncompressedDataBuffer) 
1453          throws IOException {
1454          deflateFilter.resetState();
1455          buffer.reset();
1456          deflateOut.write(uncompressedDataBuffer.getData(), 0, 
1457                           uncompressedDataBuffer.getLength());
1458          deflateOut.flush();
1459          deflateFilter.finish();
1460          
1461          WritableUtils.writeVInt(out, buffer.getLength());
1462          out.write(buffer.getData(), 0, buffer.getLength());
1463        }
1464        
1465        /** Compress and flush contents to dfs */
1466        @Override
1467        public synchronized void sync() throws IOException {
1468          if (noBufferedRecords > 0) {
1469            super.sync();
1470            
1471            // No. of records
1472            WritableUtils.writeVInt(out, noBufferedRecords);
1473            
1474            // Write 'keys' and lengths
1475            writeBuffer(keyLenBuffer);
1476            writeBuffer(keyBuffer);
1477            
1478            // Write 'values' and lengths
1479            writeBuffer(valLenBuffer);
1480            writeBuffer(valBuffer);
1481            
1482            // Flush the file-stream
1483            out.flush();
1484            
1485            // Reset internal states
1486            keyLenBuffer.reset();
1487            keyBuffer.reset();
1488            valLenBuffer.reset();
1489            valBuffer.reset();
1490            noBufferedRecords = 0;
1491          }
1492          
1493        }
1494        
1495        /** Close the file. */
1496        @Override
1497        public synchronized void close() throws IOException {
1498          if (out != null) {
1499            sync();
1500          }
1501          super.close();
1502        }
1503    
1504        /** Append a key/value pair. */
1505        @Override
1506        @SuppressWarnings("unchecked")
1507        public synchronized void append(Object key, Object val)
1508          throws IOException {
1509          if (key.getClass() != keyClass)
1510            throw new IOException("wrong key class: "+key+" is not "+keyClass);
1511          if (val.getClass() != valClass)
1512            throw new IOException("wrong value class: "+val+" is not "+valClass);
1513    
1514          // Save key/value into respective buffers 
1515          int oldKeyLength = keyBuffer.getLength();
1516          keySerializer.serialize(key);
1517          int keyLength = keyBuffer.getLength() - oldKeyLength;
1518          if (keyLength < 0)
1519            throw new IOException("negative length keys not allowed: " + key);
1520          WritableUtils.writeVInt(keyLenBuffer, keyLength);
1521    
1522          int oldValLength = valBuffer.getLength();
1523          uncompressedValSerializer.serialize(val);
1524          int valLength = valBuffer.getLength() - oldValLength;
1525          WritableUtils.writeVInt(valLenBuffer, valLength);
1526          
1527          // Added another key/value pair
1528          ++noBufferedRecords;
1529          
1530          // Compress and flush?
1531          int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();
1532          if (currentBlockSize >= compressionBlockSize) {
1533            sync();
1534          }
1535        }
1536        
1537        /** Append a key/value pair. */
1538        @Override
1539        public synchronized void appendRaw(byte[] keyData, int keyOffset,
1540            int keyLength, ValueBytes val) throws IOException {
1541          
1542          if (keyLength < 0)
1543            throw new IOException("negative length keys not allowed");
1544    
1545          int valLength = val.getSize();
1546          
1547          // Save key/value data in relevant buffers
1548          WritableUtils.writeVInt(keyLenBuffer, keyLength);
1549          keyBuffer.write(keyData, keyOffset, keyLength);
1550          WritableUtils.writeVInt(valLenBuffer, valLength);
1551          val.writeUncompressedBytes(valBuffer);
1552    
1553          // Added another key/value pair
1554          ++noBufferedRecords;
1555    
1556          // Compress and flush?
1557          int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength(); 
1558          if (currentBlockSize >= compressionBlockSize) {
1559            sync();
1560          }
1561        }
1562      
1563      } // BlockCompressionWriter
1564    
1565      /** Get the configured buffer size */
1566      private static int getBufferSize(Configuration conf) {
1567        return conf.getInt("io.file.buffer.size", 4096);
1568      }
1569    
1570      /** Reads key/value pairs from a sequence-format file. */
1571      public static class Reader implements java.io.Closeable {
1572        private String filename;
1573        private FSDataInputStream in;
1574        private DataOutputBuffer outBuf = new DataOutputBuffer();
1575    
1576        private byte version;
1577    
1578        private String keyClassName;
1579        private String valClassName;
1580        private Class keyClass;
1581        private Class valClass;
1582    
1583        private CompressionCodec codec = null;
1584        private Metadata metadata = null;
1585        
1586        private byte[] sync = new byte[SYNC_HASH_SIZE];
1587        private byte[] syncCheck = new byte[SYNC_HASH_SIZE];
1588        private boolean syncSeen;
1589    
1590        private long headerEnd;
1591        private long end;
1592        private int keyLength;
1593        private int recordLength;
1594    
1595        private boolean decompress;
1596        private boolean blockCompressed;
1597        
1598        private Configuration conf;
1599    
1600        private int noBufferedRecords = 0;
1601        private boolean lazyDecompress = true;
1602        private boolean valuesDecompressed = true;
1603        
1604        private int noBufferedKeys = 0;
1605        private int noBufferedValues = 0;
1606        
1607        private DataInputBuffer keyLenBuffer = null;
1608        private CompressionInputStream keyLenInFilter = null;
1609        private DataInputStream keyLenIn = null;
1610        private Decompressor keyLenDecompressor = null;
1611        private DataInputBuffer keyBuffer = null;
1612        private CompressionInputStream keyInFilter = null;
1613        private DataInputStream keyIn = null;
1614        private Decompressor keyDecompressor = null;
1615    
1616        private DataInputBuffer valLenBuffer = null;
1617        private CompressionInputStream valLenInFilter = null;
1618        private DataInputStream valLenIn = null;
1619        private Decompressor valLenDecompressor = null;
1620        private DataInputBuffer valBuffer = null;
1621        private CompressionInputStream valInFilter = null;
1622        private DataInputStream valIn = null;
1623        private Decompressor valDecompressor = null;
1624        
1625        private Deserializer keyDeserializer;
1626        private Deserializer valDeserializer;
1627    
1628        /**
1629         * A tag interface for all of the Reader options
1630         */
1631        public static interface Option {}
1632        
1633        /**
1634         * Create an option to specify the path name of the sequence file.
1635         * @param value the path to read
1636         * @return a new option
1637         */
1638        public static Option file(Path value) {
1639          return new FileOption(value);
1640        }
1641        
1642        /**
1643         * Create an option to specify the stream with the sequence file.
1644         * @param value the stream to read.
1645         * @return a new option
1646         */
1647        public static Option stream(FSDataInputStream value) {
1648          return new InputStreamOption(value);
1649        }
1650        
1651        /**
1652         * Create an option to specify the starting byte to read.
1653         * @param value the number of bytes to skip over
1654         * @return a new option
1655         */
1656        public static Option start(long value) {
1657          return new StartOption(value);
1658        }
1659        
1660        /**
1661         * Create an option to specify the number of bytes to read.
1662         * @param value the number of bytes to read
1663         * @return a new option
1664         */
1665        public static Option length(long value) {
1666          return new LengthOption(value);
1667        }
1668        
1669        /**
1670         * Create an option with the buffer size for reading the given pathname.
1671         * @param value the number of bytes to buffer
1672         * @return a new option
1673         */
1674        public static Option bufferSize(int value) {
1675          return new BufferSizeOption(value);
1676        }
1677    
1678        private static class FileOption extends Options.PathOption 
1679                                        implements Option {
1680          private FileOption(Path value) {
1681            super(value);
1682          }
1683        }
1684        
1685        private static class InputStreamOption
1686            extends Options.FSDataInputStreamOption 
1687            implements Option {
1688          private InputStreamOption(FSDataInputStream value) {
1689            super(value);
1690          }
1691        }
1692    
1693        private static class StartOption extends Options.LongOption
1694                                         implements Option {
1695          private StartOption(long value) {
1696            super(value);
1697          }
1698        }
1699    
1700        private static class LengthOption extends Options.LongOption
1701                                          implements Option {
1702          private LengthOption(long value) {
1703            super(value);
1704          }
1705        }
1706    
1707        private static class BufferSizeOption extends Options.IntegerOption
1708                                          implements Option {
1709          private BufferSizeOption(int value) {
1710            super(value);
1711          }
1712        }
1713    
1714        // only used directly
1715        private static class OnlyHeaderOption extends Options.BooleanOption 
1716                                              implements Option {
1717          private OnlyHeaderOption() {
1718            super(true);
1719          }
1720        }
1721    
1722        public Reader(Configuration conf, Option... opts) throws IOException {
1723          // Look up the options, these are null if not set
1724          FileOption fileOpt = Options.getOption(FileOption.class, opts);
1725          InputStreamOption streamOpt = 
1726            Options.getOption(InputStreamOption.class, opts);
1727          StartOption startOpt = Options.getOption(StartOption.class, opts);
1728          LengthOption lenOpt = Options.getOption(LengthOption.class, opts);
1729          BufferSizeOption bufOpt = Options.getOption(BufferSizeOption.class,opts);
1730          OnlyHeaderOption headerOnly = 
1731            Options.getOption(OnlyHeaderOption.class, opts);
1732          // check for consistency
1733          if ((fileOpt == null) == (streamOpt == null)) {
1734            throw new 
1735              IllegalArgumentException("File or stream option must be specified");
1736          }
1737          if (fileOpt == null && bufOpt != null) {
1738            throw new IllegalArgumentException("buffer size can only be set when" +
1739                                               " a file is specified.");
1740          }
1741          // figure out the real values
1742          Path filename = null;
1743          FSDataInputStream file;
1744          final long len;
1745          if (fileOpt != null) {
1746            filename = fileOpt.getValue();
1747            FileSystem fs = filename.getFileSystem(conf);
1748            int bufSize = bufOpt == null ? getBufferSize(conf): bufOpt.getValue();
1749            len = null == lenOpt
1750              ? fs.getFileStatus(filename).getLen()
1751              : lenOpt.getValue();
1752            file = openFile(fs, filename, bufSize, len);
1753          } else {
1754            len = null == lenOpt ? Long.MAX_VALUE : lenOpt.getValue();
1755            file = streamOpt.getValue();
1756          }
1757          long start = startOpt == null ? 0 : startOpt.getValue();
1758          // really set up
1759          initialize(filename, file, start, len, conf, headerOnly != null);
1760        }
1761    
1762        /**
1763         * Construct a reader by opening a file from the given file system.
1764         * @param fs The file system used to open the file.
1765         * @param file The file being read.
1766         * @param conf Configuration
1767         * @throws IOException
1768         * @deprecated Use Reader(Configuration, Option...) instead.
1769         */
1770        @Deprecated
1771        public Reader(FileSystem fs, Path file, 
1772                      Configuration conf) throws IOException {
1773          this(conf, file(file.makeQualified(fs)));
1774        }
1775    
1776        /**
1777         * Construct a reader by the given input stream.
1778         * @param in An input stream.
1779         * @param buffersize unused
1780         * @param start The starting position.
1781         * @param length The length being read.
1782         * @param conf Configuration
1783         * @throws IOException
1784         * @deprecated Use Reader(Configuration, Reader.Option...) instead.
1785         */
1786        @Deprecated
1787        public Reader(FSDataInputStream in, int buffersize,
1788            long start, long length, Configuration conf) throws IOException {
1789          this(conf, stream(in), start(start), length(length));
1790        }
1791    
1792        /** Common work of the constructors. */
1793        private void initialize(Path filename, FSDataInputStream in,
1794                                long start, long length, Configuration conf,
1795                                boolean tempReader) throws IOException {
1796          if (in == null) {
1797            throw new IllegalArgumentException("in == null");
1798          }
1799          this.filename = filename == null ? "<unknown>" : filename.toString();
1800          this.in = in;
1801          this.conf = conf;
1802          boolean succeeded = false;
1803          try {
1804            seek(start);
1805            this.end = this.in.getPos() + length;
1806            // if it wrapped around, use the max
1807            if (end < length) {
1808              end = Long.MAX_VALUE;
1809            }
1810            init(tempReader);
1811            succeeded = true;
1812          } finally {
1813            if (!succeeded) {
1814              IOUtils.cleanup(LOG, this.in);
1815            }
1816          }
1817        }
1818    
1819        /**
1820         * Override this method to specialize the type of
1821         * {@link FSDataInputStream} returned.
1822         * @param fs The file system used to open the file.
1823         * @param file The file being read.
1824         * @param bufferSize The buffer size used to read the file.
1825         * @param length The length being read if it is >= 0.  Otherwise,
1826         *               the length is not available.
1827         * @return The opened stream.
1828         * @throws IOException
1829         */
1830        protected FSDataInputStream openFile(FileSystem fs, Path file,
1831            int bufferSize, long length) throws IOException {
1832          return fs.open(file, bufferSize);
1833        }
1834        
1835        /**
1836         * Initialize the {@link Reader}
1837         * @param tmpReader <code>true</code> if we are constructing a temporary
1838         *                  reader {@link SequenceFile.Sorter.cloneFileAttributes}, 
1839         *                  and hence do not initialize every component; 
1840         *                  <code>false</code> otherwise.
1841         * @throws IOException
1842         */
1843        private void init(boolean tempReader) throws IOException {
1844          byte[] versionBlock = new byte[VERSION.length];
1845          in.readFully(versionBlock);
1846    
1847          if ((versionBlock[0] != VERSION[0]) ||
1848              (versionBlock[1] != VERSION[1]) ||
1849              (versionBlock[2] != VERSION[2]))
1850            throw new IOException(this + " not a SequenceFile");
1851    
1852          // Set 'version'
1853          version = versionBlock[3];
1854          if (version > VERSION[3])
1855            throw new VersionMismatchException(VERSION[3], version);
1856    
1857          if (version < BLOCK_COMPRESS_VERSION) {
1858            UTF8 className = new UTF8();
1859    
1860            className.readFields(in);
1861            keyClassName = className.toStringChecked(); // key class name
1862    
1863            className.readFields(in);
1864            valClassName = className.toStringChecked(); // val class name
1865          } else {
1866            keyClassName = Text.readString(in);
1867            valClassName = Text.readString(in);
1868          }
1869    
1870          if (version > 2) {                          // if version > 2
1871            this.decompress = in.readBoolean();       // is compressed?
1872          } else {
1873            decompress = false;
1874          }
1875    
1876          if (version >= BLOCK_COMPRESS_VERSION) {    // if version >= 4
1877            this.blockCompressed = in.readBoolean();  // is block-compressed?
1878          } else {
1879            blockCompressed = false;
1880          }
1881          
1882          // if version >= 5
1883          // setup the compression codec
1884          if (decompress) {
1885            if (version >= CUSTOM_COMPRESS_VERSION) {
1886              String codecClassname = Text.readString(in);
1887              try {
1888                Class<? extends CompressionCodec> codecClass
1889                  = conf.getClassByName(codecClassname).asSubclass(CompressionCodec.class);
1890                this.codec = ReflectionUtils.newInstance(codecClass, conf);
1891              } catch (ClassNotFoundException cnfe) {
1892                throw new IllegalArgumentException("Unknown codec: " + 
1893                                                   codecClassname, cnfe);
1894              }
1895            } else {
1896              codec = new DefaultCodec();
1897              ((Configurable)codec).setConf(conf);
1898            }
1899          }
1900          
1901          this.metadata = new Metadata();
1902          if (version >= VERSION_WITH_METADATA) {    // if version >= 6
1903            this.metadata.readFields(in);
1904          }
1905          
1906          if (version > 1) {                          // if version > 1
1907            in.readFully(sync);                       // read sync bytes
1908            headerEnd = in.getPos();                  // record end of header
1909          }
1910          
1911          // Initialize... *not* if this we are constructing a temporary Reader
1912          if (!tempReader) {
1913            valBuffer = new DataInputBuffer();
1914            if (decompress) {
1915              valDecompressor = CodecPool.getDecompressor(codec);
1916              valInFilter = codec.createInputStream(valBuffer, valDecompressor);
1917              valIn = new DataInputStream(valInFilter);
1918            } else {
1919              valIn = valBuffer;
1920            }
1921    
1922            if (blockCompressed) {
1923              keyLenBuffer = new DataInputBuffer();
1924              keyBuffer = new DataInputBuffer();
1925              valLenBuffer = new DataInputBuffer();
1926    
1927              keyLenDecompressor = CodecPool.getDecompressor(codec);
1928              keyLenInFilter = codec.createInputStream(keyLenBuffer, 
1929                                                       keyLenDecompressor);
1930              keyLenIn = new DataInputStream(keyLenInFilter);
1931    
1932              keyDecompressor = CodecPool.getDecompressor(codec);
1933              keyInFilter = codec.createInputStream(keyBuffer, keyDecompressor);
1934              keyIn = new DataInputStream(keyInFilter);
1935    
1936              valLenDecompressor = CodecPool.getDecompressor(codec);
1937              valLenInFilter = codec.createInputStream(valLenBuffer, 
1938                                                       valLenDecompressor);
1939              valLenIn = new DataInputStream(valLenInFilter);
1940            }
1941            
1942            SerializationFactory serializationFactory =
1943              new SerializationFactory(conf);
1944            this.keyDeserializer =
1945              getDeserializer(serializationFactory, getKeyClass());
1946            if (this.keyDeserializer == null) {
1947              throw new IOException(
1948                  "Could not find a deserializer for the Key class: '"
1949                      + getKeyClass().getCanonicalName() + "'. "
1950                      + "Please ensure that the configuration '" +
1951                      CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
1952                      + "properly configured, if you're using "
1953                      + "custom serialization.");
1954            }
1955            if (!blockCompressed) {
1956              this.keyDeserializer.open(valBuffer);
1957            } else {
1958              this.keyDeserializer.open(keyIn);
1959            }
1960            this.valDeserializer =
1961              getDeserializer(serializationFactory, getValueClass());
1962            if (this.valDeserializer == null) {
1963              throw new IOException(
1964                  "Could not find a deserializer for the Value class: '"
1965                      + getValueClass().getCanonicalName() + "'. "
1966                      + "Please ensure that the configuration '" +
1967                      CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
1968                      + "properly configured, if you're using "
1969                      + "custom serialization.");
1970            }
1971            this.valDeserializer.open(valIn);
1972          }
1973        }
1974        
1975        @SuppressWarnings("unchecked")
1976        private Deserializer getDeserializer(SerializationFactory sf, Class c) {
1977          return sf.getDeserializer(c);
1978        }
1979        
1980        /** Close the file. */
1981        @Override
1982        public synchronized void close() throws IOException {
1983          // Return the decompressors to the pool
1984          CodecPool.returnDecompressor(keyLenDecompressor);
1985          CodecPool.returnDecompressor(keyDecompressor);
1986          CodecPool.returnDecompressor(valLenDecompressor);
1987          CodecPool.returnDecompressor(valDecompressor);
1988          keyLenDecompressor = keyDecompressor = null;
1989          valLenDecompressor = valDecompressor = null;
1990          
1991          if (keyDeserializer != null) {
1992            keyDeserializer.close();
1993          }
1994          if (valDeserializer != null) {
1995            valDeserializer.close();
1996          }
1997          
1998          // Close the input-stream
1999          in.close();
2000        }
2001    
2002        /** Returns the name of the key class. */
2003        public String getKeyClassName() {
2004          return keyClassName;
2005        }
2006    
2007        /** Returns the class of keys in this file. */
2008        public synchronized Class<?> getKeyClass() {
2009          if (null == keyClass) {
2010            try {
2011              keyClass = WritableName.getClass(getKeyClassName(), conf);
2012            } catch (IOException e) {
2013              throw new RuntimeException(e);
2014            }
2015          }
2016          return keyClass;
2017        }
2018    
2019        /** Returns the name of the value class. */
2020        public String getValueClassName() {
2021          return valClassName;
2022        }
2023    
2024        /** Returns the class of values in this file. */
2025        public synchronized Class<?> getValueClass() {
2026          if (null == valClass) {
2027            try {
2028              valClass = WritableName.getClass(getValueClassName(), conf);
2029            } catch (IOException e) {
2030              throw new RuntimeException(e);
2031            }
2032          }
2033          return valClass;
2034        }
2035    
2036        /** Returns true if values are compressed. */
2037        public boolean isCompressed() { return decompress; }
2038        
2039        /** Returns true if records are block-compressed. */
2040        public boolean isBlockCompressed() { return blockCompressed; }
2041        
2042        /** Returns the compression codec of data in this file. */
2043        public CompressionCodec getCompressionCodec() { return codec; }
2044        
2045        /**
2046         * Get the compression type for this file.
2047         * @return the compression type
2048         */
2049        public CompressionType getCompressionType() {
2050          if (decompress) {
2051            return blockCompressed ? CompressionType.BLOCK : CompressionType.RECORD;
2052          } else {
2053            return CompressionType.NONE;
2054          }
2055        }
2056    
2057        /** Returns the metadata object of the file */
2058        public Metadata getMetadata() {
2059          return this.metadata;
2060        }
2061        
2062        /** Returns the configuration used for this file. */
2063        Configuration getConf() { return conf; }
2064        
2065        /** Read a compressed buffer */
2066        private synchronized void readBuffer(DataInputBuffer buffer, 
2067                                             CompressionInputStream filter) throws IOException {
2068          // Read data into a temporary buffer
2069          DataOutputBuffer dataBuffer = new DataOutputBuffer();
2070    
2071          try {
2072            int dataBufferLength = WritableUtils.readVInt(in);
2073            dataBuffer.write(in, dataBufferLength);
2074          
2075            // Set up 'buffer' connected to the input-stream
2076            buffer.reset(dataBuffer.getData(), 0, dataBuffer.getLength());
2077          } finally {
2078            dataBuffer.close();
2079          }
2080    
2081          // Reset the codec
2082          filter.resetState();
2083        }
2084        
2085        /** Read the next 'compressed' block */
2086        private synchronized void readBlock() throws IOException {
2087          // Check if we need to throw away a whole block of 
2088          // 'values' due to 'lazy decompression' 
2089          if (lazyDecompress && !valuesDecompressed) {
2090            in.seek(WritableUtils.readVInt(in)+in.getPos());
2091            in.seek(WritableUtils.readVInt(in)+in.getPos());
2092          }
2093          
2094          // Reset internal states
2095          noBufferedKeys = 0; noBufferedValues = 0; noBufferedRecords = 0;
2096          valuesDecompressed = false;
2097    
2098          //Process sync
2099          if (sync != null) {
2100            in.readInt();
2101            in.readFully(syncCheck);                // read syncCheck
2102            if (!Arrays.equals(sync, syncCheck))    // check it
2103              throw new IOException("File is corrupt!");
2104          }
2105          syncSeen = true;
2106    
2107          // Read number of records in this block
2108          noBufferedRecords = WritableUtils.readVInt(in);
2109          
2110          // Read key lengths and keys
2111          readBuffer(keyLenBuffer, keyLenInFilter);
2112          readBuffer(keyBuffer, keyInFilter);
2113          noBufferedKeys = noBufferedRecords;
2114          
2115          // Read value lengths and values
2116          if (!lazyDecompress) {
2117            readBuffer(valLenBuffer, valLenInFilter);
2118            readBuffer(valBuffer, valInFilter);
2119            noBufferedValues = noBufferedRecords;
2120            valuesDecompressed = true;
2121          }
2122        }
2123    
2124        /** 
2125         * Position valLenIn/valIn to the 'value' 
2126         * corresponding to the 'current' key 
2127         */
2128        private synchronized void seekToCurrentValue() throws IOException {
2129          if (!blockCompressed) {
2130            if (decompress) {
2131              valInFilter.resetState();
2132            }
2133            valBuffer.reset();
2134          } else {
2135            // Check if this is the first value in the 'block' to be read
2136            if (lazyDecompress && !valuesDecompressed) {
2137              // Read the value lengths and values
2138              readBuffer(valLenBuffer, valLenInFilter);
2139              readBuffer(valBuffer, valInFilter);
2140              noBufferedValues = noBufferedRecords;
2141              valuesDecompressed = true;
2142            }
2143            
2144            // Calculate the no. of bytes to skip
2145            // Note: 'current' key has already been read!
2146            int skipValBytes = 0;
2147            int currentKey = noBufferedKeys + 1;          
2148            for (int i=noBufferedValues; i > currentKey; --i) {
2149              skipValBytes += WritableUtils.readVInt(valLenIn);
2150              --noBufferedValues;
2151            }
2152            
2153            // Skip to the 'val' corresponding to 'current' key
2154            if (skipValBytes > 0) {
2155              if (valIn.skipBytes(skipValBytes) != skipValBytes) {
2156                throw new IOException("Failed to seek to " + currentKey + 
2157                                      "(th) value!");
2158              }
2159            }
2160          }
2161        }
2162    
2163        /**
2164         * Get the 'value' corresponding to the last read 'key'.
2165         * @param val : The 'value' to be read.
2166         * @throws IOException
2167         */
2168        public synchronized void getCurrentValue(Writable val) 
2169          throws IOException {
2170          if (val instanceof Configurable) {
2171            ((Configurable) val).setConf(this.conf);
2172          }
2173    
2174          // Position stream to 'current' value
2175          seekToCurrentValue();
2176    
2177          if (!blockCompressed) {
2178            val.readFields(valIn);
2179            
2180            if (valIn.read() > 0) {
2181              LOG.info("available bytes: " + valIn.available());
2182              throw new IOException(val+" read "+(valBuffer.getPosition()-keyLength)
2183                                    + " bytes, should read " +
2184                                    (valBuffer.getLength()-keyLength));
2185            }
2186          } else {
2187            // Get the value
2188            int valLength = WritableUtils.readVInt(valLenIn);
2189            val.readFields(valIn);
2190            
2191            // Read another compressed 'value'
2192            --noBufferedValues;
2193            
2194            // Sanity check
2195            if ((valLength < 0) && LOG.isDebugEnabled()) {
2196              LOG.debug(val + " is a zero-length value");
2197            }
2198          }
2199    
2200        }
2201        
2202        /**
2203         * Get the 'value' corresponding to the last read 'key'.
2204         * @param val : The 'value' to be read.
2205         * @throws IOException
2206         */
2207        public synchronized Object getCurrentValue(Object val) 
2208          throws IOException {
2209          if (val instanceof Configurable) {
2210            ((Configurable) val).setConf(this.conf);
2211          }
2212    
2213          // Position stream to 'current' value
2214          seekToCurrentValue();
2215    
2216          if (!blockCompressed) {
2217            val = deserializeValue(val);
2218            
2219            if (valIn.read() > 0) {
2220              LOG.info("available bytes: " + valIn.available());
2221              throw new IOException(val+" read "+(valBuffer.getPosition()-keyLength)
2222                                    + " bytes, should read " +
2223                                    (valBuffer.getLength()-keyLength));
2224            }
2225          } else {
2226            // Get the value
2227            int valLength = WritableUtils.readVInt(valLenIn);
2228            val = deserializeValue(val);
2229            
2230            // Read another compressed 'value'
2231            --noBufferedValues;
2232            
2233            // Sanity check
2234            if ((valLength < 0) && LOG.isDebugEnabled()) {
2235              LOG.debug(val + " is a zero-length value");
2236            }
2237          }
2238          return val;
2239    
2240        }
2241    
2242        @SuppressWarnings("unchecked")
2243        private Object deserializeValue(Object val) throws IOException {
2244          return valDeserializer.deserialize(val);
2245        }
2246        
2247        /** Read the next key in the file into <code>key</code>, skipping its
2248         * value.  True if another entry exists, and false at end of file. */
2249        public synchronized boolean next(Writable key) throws IOException {
2250          if (key.getClass() != getKeyClass())
2251            throw new IOException("wrong key class: "+key.getClass().getName()
2252                                  +" is not "+keyClass);
2253    
2254          if (!blockCompressed) {
2255            outBuf.reset();
2256            
2257            keyLength = next(outBuf);
2258            if (keyLength < 0)
2259              return false;
2260            
2261            valBuffer.reset(outBuf.getData(), outBuf.getLength());
2262            
2263            key.readFields(valBuffer);
2264            valBuffer.mark(0);
2265            if (valBuffer.getPosition() != keyLength)
2266              throw new IOException(key + " read " + valBuffer.getPosition()
2267                                    + " bytes, should read " + keyLength);
2268          } else {
2269            //Reset syncSeen
2270            syncSeen = false;
2271            
2272            if (noBufferedKeys == 0) {
2273              try {
2274                readBlock();
2275              } catch (EOFException eof) {
2276                return false;
2277              }
2278            }
2279            
2280            int keyLength = WritableUtils.readVInt(keyLenIn);
2281            
2282            // Sanity check
2283            if (keyLength < 0) {
2284              return false;
2285            }
2286            
2287            //Read another compressed 'key'
2288            key.readFields(keyIn);
2289            --noBufferedKeys;
2290          }
2291    
2292          return true;
2293        }
2294    
2295        /** Read the next key/value pair in the file into <code>key</code> and
2296         * <code>val</code>.  Returns true if such a pair exists and false when at
2297         * end of file */
2298        public synchronized boolean next(Writable key, Writable val)
2299          throws IOException {
2300          if (val.getClass() != getValueClass())
2301            throw new IOException("wrong value class: "+val+" is not "+valClass);
2302    
2303          boolean more = next(key);
2304          
2305          if (more) {
2306            getCurrentValue(val);
2307          }
2308    
2309          return more;
2310        }
2311        
2312        /**
2313         * Read and return the next record length, potentially skipping over 
2314         * a sync block.
2315         * @return the length of the next record or -1 if there is no next record
2316         * @throws IOException
2317         */
2318        private synchronized int readRecordLength() throws IOException {
2319          if (in.getPos() >= end) {
2320            return -1;
2321          }      
2322          int length = in.readInt();
2323          if (version > 1 && sync != null &&
2324              length == SYNC_ESCAPE) {              // process a sync entry
2325            in.readFully(syncCheck);                // read syncCheck
2326            if (!Arrays.equals(sync, syncCheck))    // check it
2327              throw new IOException("File is corrupt!");
2328            syncSeen = true;
2329            if (in.getPos() >= end) {
2330              return -1;
2331            }
2332            length = in.readInt();                  // re-read length
2333          } else {
2334            syncSeen = false;
2335          }
2336          
2337          return length;
2338        }
2339        
2340        /** Read the next key/value pair in the file into <code>buffer</code>.
2341         * Returns the length of the key read, or -1 if at end of file.  The length
2342         * of the value may be computed by calling buffer.getLength() before and
2343         * after calls to this method. */
2344        /** @deprecated Call {@link #nextRaw(DataOutputBuffer,SequenceFile.ValueBytes)}. */
2345        @Deprecated
2346        synchronized int next(DataOutputBuffer buffer) throws IOException {
2347          // Unsupported for block-compressed sequence files
2348          if (blockCompressed) {
2349            throw new IOException("Unsupported call for block-compressed" +
2350                                  " SequenceFiles - use SequenceFile.Reader.next(DataOutputStream, ValueBytes)");
2351          }
2352          try {
2353            int length = readRecordLength();
2354            if (length == -1) {
2355              return -1;
2356            }
2357            int keyLength = in.readInt();
2358            buffer.write(in, length);
2359            return keyLength;
2360          } catch (ChecksumException e) {             // checksum failure
2361            handleChecksumException(e);
2362            return next(buffer);
2363          }
2364        }
2365    
2366        public ValueBytes createValueBytes() {
2367          ValueBytes val = null;
2368          if (!decompress || blockCompressed) {
2369            val = new UncompressedBytes();
2370          } else {
2371            val = new CompressedBytes(codec);
2372          }
2373          return val;
2374        }
2375    
2376        /**
2377         * Read 'raw' records.
2378         * @param key - The buffer into which the key is read
2379         * @param val - The 'raw' value
2380         * @return Returns the total record length or -1 for end of file
2381         * @throws IOException
2382         */
2383        public synchronized int nextRaw(DataOutputBuffer key, ValueBytes val) 
2384          throws IOException {
2385          if (!blockCompressed) {
2386            int length = readRecordLength();
2387            if (length == -1) {
2388              return -1;
2389            }
2390            int keyLength = in.readInt();
2391            int valLength = length - keyLength;
2392            key.write(in, keyLength);
2393            if (decompress) {
2394              CompressedBytes value = (CompressedBytes)val;
2395              value.reset(in, valLength);
2396            } else {
2397              UncompressedBytes value = (UncompressedBytes)val;
2398              value.reset(in, valLength);
2399            }
2400            
2401            return length;
2402          } else {
2403            //Reset syncSeen
2404            syncSeen = false;
2405            
2406            // Read 'key'
2407            if (noBufferedKeys == 0) {
2408              if (in.getPos() >= end) 
2409                return -1;
2410    
2411              try { 
2412                readBlock();
2413              } catch (EOFException eof) {
2414                return -1;
2415              }
2416            }
2417            int keyLength = WritableUtils.readVInt(keyLenIn);
2418            if (keyLength < 0) {
2419              throw new IOException("zero length key found!");
2420            }
2421            key.write(keyIn, keyLength);
2422            --noBufferedKeys;
2423            
2424            // Read raw 'value'
2425            seekToCurrentValue();
2426            int valLength = WritableUtils.readVInt(valLenIn);
2427            UncompressedBytes rawValue = (UncompressedBytes)val;
2428            rawValue.reset(valIn, valLength);
2429            --noBufferedValues;
2430            
2431            return (keyLength+valLength);
2432          }
2433          
2434        }
2435    
2436        /**
2437         * Read 'raw' keys.
2438         * @param key - The buffer into which the key is read
2439         * @return Returns the key length or -1 for end of file
2440         * @throws IOException
2441         */
2442        public synchronized int nextRawKey(DataOutputBuffer key) 
2443          throws IOException {
2444          if (!blockCompressed) {
2445            recordLength = readRecordLength();
2446            if (recordLength == -1) {
2447              return -1;
2448            }
2449            keyLength = in.readInt();
2450            key.write(in, keyLength);
2451            return keyLength;
2452          } else {
2453            //Reset syncSeen
2454            syncSeen = false;
2455            
2456            // Read 'key'
2457            if (noBufferedKeys == 0) {
2458              if (in.getPos() >= end) 
2459                return -1;
2460    
2461              try { 
2462                readBlock();
2463              } catch (EOFException eof) {
2464                return -1;
2465              }
2466            }
2467            int keyLength = WritableUtils.readVInt(keyLenIn);
2468            if (keyLength < 0) {
2469              throw new IOException("zero length key found!");
2470            }
2471            key.write(keyIn, keyLength);
2472            --noBufferedKeys;
2473            
2474            return keyLength;
2475          }
2476          
2477        }
2478    
2479        /** Read the next key in the file, skipping its
2480         * value.  Return null at end of file. */
2481        public synchronized Object next(Object key) throws IOException {
2482          if (key != null && key.getClass() != getKeyClass()) {
2483            throw new IOException("wrong key class: "+key.getClass().getName()
2484                                  +" is not "+keyClass);
2485          }
2486    
2487          if (!blockCompressed) {
2488            outBuf.reset();
2489            
2490            keyLength = next(outBuf);
2491            if (keyLength < 0)
2492              return null;
2493            
2494            valBuffer.reset(outBuf.getData(), outBuf.getLength());
2495            
2496            key = deserializeKey(key);
2497            valBuffer.mark(0);
2498            if (valBuffer.getPosition() != keyLength)
2499              throw new IOException(key + " read " + valBuffer.getPosition()
2500                                    + " bytes, should read " + keyLength);
2501          } else {
2502            //Reset syncSeen
2503            syncSeen = false;
2504            
2505            if (noBufferedKeys == 0) {
2506              try {
2507                readBlock();
2508              } catch (EOFException eof) {
2509                return null;
2510              }
2511            }
2512            
2513            int keyLength = WritableUtils.readVInt(keyLenIn);
2514            
2515            // Sanity check
2516            if (keyLength < 0) {
2517              return null;
2518            }
2519            
2520            //Read another compressed 'key'
2521            key = deserializeKey(key);
2522            --noBufferedKeys;
2523          }
2524    
2525          return key;
2526        }
2527    
2528        @SuppressWarnings("unchecked")
2529        private Object deserializeKey(Object key) throws IOException {
2530          return keyDeserializer.deserialize(key);
2531        }
2532    
2533        /**
2534         * Read 'raw' values.
2535         * @param val - The 'raw' value
2536         * @return Returns the value length
2537         * @throws IOException
2538         */
2539        public synchronized int nextRawValue(ValueBytes val) 
2540          throws IOException {
2541          
2542          // Position stream to current value
2543          seekToCurrentValue();
2544     
2545          if (!blockCompressed) {
2546            int valLength = recordLength - keyLength;
2547            if (decompress) {
2548              CompressedBytes value = (CompressedBytes)val;
2549              value.reset(in, valLength);
2550            } else {
2551              UncompressedBytes value = (UncompressedBytes)val;
2552              value.reset(in, valLength);
2553            }
2554             
2555            return valLength;
2556          } else {
2557            int valLength = WritableUtils.readVInt(valLenIn);
2558            UncompressedBytes rawValue = (UncompressedBytes)val;
2559            rawValue.reset(valIn, valLength);
2560            --noBufferedValues;
2561            return valLength;
2562          }
2563          
2564        }
2565    
2566        private void handleChecksumException(ChecksumException e)
2567          throws IOException {
2568          if (this.conf.getBoolean("io.skip.checksum.errors", false)) {
2569            LOG.warn("Bad checksum at "+getPosition()+". Skipping entries.");
2570            sync(getPosition()+this.conf.getInt("io.bytes.per.checksum", 512));
2571          } else {
2572            throw e;
2573          }
2574        }
2575    
2576        /** disables sync. often invoked for tmp files */
2577        synchronized void ignoreSync() {
2578          sync = null;
2579        }
2580        
2581        /** Set the current byte position in the input file.
2582         *
2583         * <p>The position passed must be a position returned by {@link
2584         * SequenceFile.Writer#getLength()} when writing this file.  To seek to an arbitrary
2585         * position, use {@link SequenceFile.Reader#sync(long)}.
2586         */
2587        public synchronized void seek(long position) throws IOException {
2588          in.seek(position);
2589          if (blockCompressed) {                      // trigger block read
2590            noBufferedKeys = 0;
2591            valuesDecompressed = true;
2592          }
2593        }
2594    
2595        /** Seek to the next sync mark past a given position.*/
2596        public synchronized void sync(long position) throws IOException {
2597          if (position+SYNC_SIZE >= end) {
2598            seek(end);
2599            return;
2600          }
2601    
2602          if (position < headerEnd) {
2603            // seek directly to first record
2604            in.seek(headerEnd);
2605            // note the sync marker "seen" in the header
2606            syncSeen = true;
2607            return;
2608          }
2609    
2610          try {
2611            seek(position+4);                         // skip escape
2612            in.readFully(syncCheck);
2613            int syncLen = sync.length;
2614            for (int i = 0; in.getPos() < end; i++) {
2615              int j = 0;
2616              for (; j < syncLen; j++) {
2617                if (sync[j] != syncCheck[(i+j)%syncLen])
2618                  break;
2619              }
2620              if (j == syncLen) {
2621                in.seek(in.getPos() - SYNC_SIZE);     // position before sync
2622                return;
2623              }
2624              syncCheck[i%syncLen] = in.readByte();
2625            }
2626          } catch (ChecksumException e) {             // checksum failure
2627            handleChecksumException(e);
2628          }
2629        }
2630    
2631        /** Returns true iff the previous call to next passed a sync mark.*/
2632        public synchronized boolean syncSeen() { return syncSeen; }
2633    
2634        /** Return the current byte position in the input file. */
2635        public synchronized long getPosition() throws IOException {
2636          return in.getPos();
2637        }
2638    
2639        /** Returns the name of the file. */
2640        @Override
2641        public String toString() {
2642          return filename;
2643        }
2644    
2645      }
2646    
2647      /** Sorts key/value pairs in a sequence-format file.
2648       *
2649       * <p>For best performance, applications should make sure that the {@link
2650       * Writable#readFields(DataInput)} implementation of their keys is
2651       * very efficient.  In particular, it should avoid allocating memory.
2652       */
2653      public static class Sorter {
2654    
2655        private RawComparator comparator;
2656    
2657        private MergeSort mergeSort; //the implementation of merge sort
2658        
2659        private Path[] inFiles;                     // when merging or sorting
2660    
2661        private Path outFile;
2662    
2663        private int memory; // bytes
2664        private int factor; // merged per pass
2665    
2666        private FileSystem fs = null;
2667    
2668        private Class keyClass;
2669        private Class valClass;
2670    
2671        private Configuration conf;
2672        private Metadata metadata;
2673        
2674        private Progressable progressable = null;
2675    
2676        /** Sort and merge files containing the named classes. */
2677        public Sorter(FileSystem fs, Class<? extends WritableComparable> keyClass,
2678                      Class valClass, Configuration conf)  {
2679          this(fs, WritableComparator.get(keyClass, conf), keyClass, valClass, conf);
2680        }
2681    
2682        /** Sort and merge using an arbitrary {@link RawComparator}. */
2683        public Sorter(FileSystem fs, RawComparator comparator, Class keyClass, 
2684                      Class valClass, Configuration conf) {
2685          this(fs, comparator, keyClass, valClass, conf, new Metadata());
2686        }
2687    
2688        /** Sort and merge using an arbitrary {@link RawComparator}. */
2689        public Sorter(FileSystem fs, RawComparator comparator, Class keyClass,
2690                      Class valClass, Configuration conf, Metadata metadata) {
2691          this.fs = fs;
2692          this.comparator = comparator;
2693          this.keyClass = keyClass;
2694          this.valClass = valClass;
2695          this.memory = conf.getInt("io.sort.mb", 100) * 1024 * 1024;
2696          this.factor = conf.getInt("io.sort.factor", 100);
2697          this.conf = conf;
2698          this.metadata = metadata;
2699        }
2700    
2701        /** Set the number of streams to merge at once.*/
2702        public void setFactor(int factor) { this.factor = factor; }
2703    
2704        /** Get the number of streams to merge at once.*/
2705        public int getFactor() { return factor; }
2706    
2707        /** Set the total amount of buffer memory, in bytes.*/
2708        public void setMemory(int memory) { this.memory = memory; }
2709    
2710        /** Get the total amount of buffer memory, in bytes.*/
2711        public int getMemory() { return memory; }
2712    
2713        /** Set the progressable object in order to report progress. */
2714        public void setProgressable(Progressable progressable) {
2715          this.progressable = progressable;
2716        }
2717        
2718        /** 
2719         * Perform a file sort from a set of input files into an output file.
2720         * @param inFiles the files to be sorted
2721         * @param outFile the sorted output file
2722         * @param deleteInput should the input files be deleted as they are read?
2723         */
2724        public void sort(Path[] inFiles, Path outFile,
2725                         boolean deleteInput) throws IOException {
2726          if (fs.exists(outFile)) {
2727            throw new IOException("already exists: " + outFile);
2728          }
2729    
2730          this.inFiles = inFiles;
2731          this.outFile = outFile;
2732    
2733          int segments = sortPass(deleteInput);
2734          if (segments > 1) {
2735            mergePass(outFile.getParent());
2736          }
2737        }
2738    
2739        /** 
2740         * Perform a file sort from a set of input files and return an iterator.
2741         * @param inFiles the files to be sorted
2742         * @param tempDir the directory where temp files are created during sort
2743         * @param deleteInput should the input files be deleted as they are read?
2744         * @return iterator the RawKeyValueIterator
2745         */
2746        public RawKeyValueIterator sortAndIterate(Path[] inFiles, Path tempDir, 
2747                                                  boolean deleteInput) throws IOException {
2748          Path outFile = new Path(tempDir + Path.SEPARATOR + "all.2");
2749          if (fs.exists(outFile)) {
2750            throw new IOException("already exists: " + outFile);
2751          }
2752          this.inFiles = inFiles;
2753          //outFile will basically be used as prefix for temp files in the cases
2754          //where sort outputs multiple sorted segments. For the single segment
2755          //case, the outputFile itself will contain the sorted data for that
2756          //segment
2757          this.outFile = outFile;
2758    
2759          int segments = sortPass(deleteInput);
2760          if (segments > 1)
2761            return merge(outFile.suffix(".0"), outFile.suffix(".0.index"), 
2762                         tempDir);
2763          else if (segments == 1)
2764            return merge(new Path[]{outFile}, true, tempDir);
2765          else return null;
2766        }
2767    
2768        /**
2769         * The backwards compatible interface to sort.
2770         * @param inFile the input file to sort
2771         * @param outFile the sorted output file
2772         */
2773        public void sort(Path inFile, Path outFile) throws IOException {
2774          sort(new Path[]{inFile}, outFile, false);
2775        }
2776        
2777        private int sortPass(boolean deleteInput) throws IOException {
2778          if(LOG.isDebugEnabled()) {
2779            LOG.debug("running sort pass");
2780          }
2781          SortPass sortPass = new SortPass();         // make the SortPass
2782          sortPass.setProgressable(progressable);
2783          mergeSort = new MergeSort(sortPass.new SeqFileComparator());
2784          try {
2785            return sortPass.run(deleteInput);         // run it
2786          } finally {
2787            sortPass.close();                         // close it
2788          }
2789        }
2790    
2791        private class SortPass {
2792          private int memoryLimit = memory/4;
2793          private int recordLimit = 1000000;
2794          
2795          private DataOutputBuffer rawKeys = new DataOutputBuffer();
2796          private byte[] rawBuffer;
2797    
2798          private int[] keyOffsets = new int[1024];
2799          private int[] pointers = new int[keyOffsets.length];
2800          private int[] pointersCopy = new int[keyOffsets.length];
2801          private int[] keyLengths = new int[keyOffsets.length];
2802          private ValueBytes[] rawValues = new ValueBytes[keyOffsets.length];
2803          
2804          private ArrayList segmentLengths = new ArrayList();
2805          
2806          private Reader in = null;
2807          private FSDataOutputStream out = null;
2808          private FSDataOutputStream indexOut = null;
2809          private Path outName;
2810    
2811          private Progressable progressable = null;
2812    
2813          public int run(boolean deleteInput) throws IOException {
2814            int segments = 0;
2815            int currentFile = 0;
2816            boolean atEof = (currentFile >= inFiles.length);
2817            CompressionType compressionType;
2818            CompressionCodec codec = null;
2819            segmentLengths.clear();
2820            if (atEof) {
2821              return 0;
2822            }
2823            
2824            // Initialize
2825            in = new Reader(fs, inFiles[currentFile], conf);
2826            compressionType = in.getCompressionType();
2827            codec = in.getCompressionCodec();
2828            
2829            for (int i=0; i < rawValues.length; ++i) {
2830              rawValues[i] = null;
2831            }
2832            
2833            while (!atEof) {
2834              int count = 0;
2835              int bytesProcessed = 0;
2836              rawKeys.reset();
2837              while (!atEof && 
2838                     bytesProcessed < memoryLimit && count < recordLimit) {
2839    
2840                // Read a record into buffer
2841                // Note: Attempt to re-use 'rawValue' as far as possible
2842                int keyOffset = rawKeys.getLength();       
2843                ValueBytes rawValue = 
2844                  (count == keyOffsets.length || rawValues[count] == null) ? 
2845                  in.createValueBytes() : 
2846                  rawValues[count];
2847                int recordLength = in.nextRaw(rawKeys, rawValue);
2848                if (recordLength == -1) {
2849                  in.close();
2850                  if (deleteInput) {
2851                    fs.delete(inFiles[currentFile], true);
2852                  }
2853                  currentFile += 1;
2854                  atEof = currentFile >= inFiles.length;
2855                  if (!atEof) {
2856                    in = new Reader(fs, inFiles[currentFile], conf);
2857                  } else {
2858                    in = null;
2859                  }
2860                  continue;
2861                }
2862    
2863                int keyLength = rawKeys.getLength() - keyOffset;
2864    
2865                if (count == keyOffsets.length)
2866                  grow();
2867    
2868                keyOffsets[count] = keyOffset;                // update pointers
2869                pointers[count] = count;
2870                keyLengths[count] = keyLength;
2871                rawValues[count] = rawValue;
2872    
2873                bytesProcessed += recordLength; 
2874                count++;
2875              }
2876    
2877              // buffer is full -- sort & flush it
2878              if(LOG.isDebugEnabled()) {
2879                LOG.debug("flushing segment " + segments);
2880              }
2881              rawBuffer = rawKeys.getData();
2882              sort(count);
2883              // indicate we're making progress
2884              if (progressable != null) {
2885                progressable.progress();
2886              }
2887              flush(count, bytesProcessed, compressionType, codec, 
2888                    segments==0 && atEof);
2889              segments++;
2890            }
2891            return segments;
2892          }
2893    
2894          public void close() throws IOException {
2895            if (in != null) {
2896              in.close();
2897            }
2898            if (out != null) {
2899              out.close();
2900            }
2901            if (indexOut != null) {
2902              indexOut.close();
2903            }
2904          }
2905    
2906          private void grow() {
2907            int newLength = keyOffsets.length * 3 / 2;
2908            keyOffsets = grow(keyOffsets, newLength);
2909            pointers = grow(pointers, newLength);
2910            pointersCopy = new int[newLength];
2911            keyLengths = grow(keyLengths, newLength);
2912            rawValues = grow(rawValues, newLength);
2913          }
2914    
2915          private int[] grow(int[] old, int newLength) {
2916            int[] result = new int[newLength];
2917            System.arraycopy(old, 0, result, 0, old.length);
2918            return result;
2919          }
2920          
2921          private ValueBytes[] grow(ValueBytes[] old, int newLength) {
2922            ValueBytes[] result = new ValueBytes[newLength];
2923            System.arraycopy(old, 0, result, 0, old.length);
2924            for (int i=old.length; i < newLength; ++i) {
2925              result[i] = null;
2926            }
2927            return result;
2928          }
2929    
2930          private void flush(int count, int bytesProcessed, 
2931                             CompressionType compressionType, 
2932                             CompressionCodec codec, 
2933                             boolean done) throws IOException {
2934            if (out == null) {
2935              outName = done ? outFile : outFile.suffix(".0");
2936              out = fs.create(outName);
2937              if (!done) {
2938                indexOut = fs.create(outName.suffix(".index"));
2939              }
2940            }
2941    
2942            long segmentStart = out.getPos();
2943            Writer writer = createWriter(conf, Writer.stream(out), 
2944                Writer.keyClass(keyClass), Writer.valueClass(valClass),
2945                Writer.compression(compressionType, codec),
2946                Writer.metadata(done ? metadata : new Metadata()));
2947            
2948            if (!done) {
2949              writer.sync = null;                     // disable sync on temp files
2950            }
2951    
2952            for (int i = 0; i < count; i++) {         // write in sorted order
2953              int p = pointers[i];
2954              writer.appendRaw(rawBuffer, keyOffsets[p], keyLengths[p], rawValues[p]);
2955            }
2956            writer.close();
2957            
2958            if (!done) {
2959              // Save the segment length
2960              WritableUtils.writeVLong(indexOut, segmentStart);
2961              WritableUtils.writeVLong(indexOut, (out.getPos()-segmentStart));
2962              indexOut.flush();
2963            }
2964          }
2965    
2966          private void sort(int count) {
2967            System.arraycopy(pointers, 0, pointersCopy, 0, count);
2968            mergeSort.mergeSort(pointersCopy, pointers, 0, count);
2969          }
2970          class SeqFileComparator implements Comparator<IntWritable> {
2971            @Override
2972            public int compare(IntWritable I, IntWritable J) {
2973              return comparator.compare(rawBuffer, keyOffsets[I.get()], 
2974                                        keyLengths[I.get()], rawBuffer, 
2975                                        keyOffsets[J.get()], keyLengths[J.get()]);
2976            }
2977          }
2978          
2979          /** set the progressable object in order to report progress */
2980          public void setProgressable(Progressable progressable)
2981          {
2982            this.progressable = progressable;
2983          }
2984          
2985        } // SequenceFile.Sorter.SortPass
2986    
2987        /** The interface to iterate over raw keys/values of SequenceFiles. */
2988        public static interface RawKeyValueIterator {
2989          /** Gets the current raw key
2990           * @return DataOutputBuffer
2991           * @throws IOException
2992           */
2993          DataOutputBuffer getKey() throws IOException; 
2994          /** Gets the current raw value
2995           * @return ValueBytes 
2996           * @throws IOException
2997           */
2998          ValueBytes getValue() throws IOException; 
2999          /** Sets up the current key and value (for getKey and getValue)
3000           * @return true if there exists a key/value, false otherwise 
3001           * @throws IOException
3002           */
3003          boolean next() throws IOException;
3004          /** closes the iterator so that the underlying streams can be closed
3005           * @throws IOException
3006           */
3007          void close() throws IOException;
3008          /** Gets the Progress object; this has a float (0.0 - 1.0) 
3009           * indicating the bytes processed by the iterator so far
3010           */
3011          Progress getProgress();
3012        }    
3013        
3014        /**
3015         * Merges the list of segments of type <code>SegmentDescriptor</code>
3016         * @param segments the list of SegmentDescriptors
3017         * @param tmpDir the directory to write temporary files into
3018         * @return RawKeyValueIterator
3019         * @throws IOException
3020         */
3021        public RawKeyValueIterator merge(List <SegmentDescriptor> segments, 
3022                                         Path tmpDir) 
3023          throws IOException {
3024          // pass in object to report progress, if present
3025          MergeQueue mQueue = new MergeQueue(segments, tmpDir, progressable);
3026          return mQueue.merge();
3027        }
3028    
3029        /**
3030         * Merges the contents of files passed in Path[] using a max factor value
3031         * that is already set
3032         * @param inNames the array of path names
3033         * @param deleteInputs true if the input files should be deleted when 
3034         * unnecessary
3035         * @param tmpDir the directory to write temporary files into
3036         * @return RawKeyValueIteratorMergeQueue
3037         * @throws IOException
3038         */
3039        public RawKeyValueIterator merge(Path [] inNames, boolean deleteInputs,
3040                                         Path tmpDir) 
3041          throws IOException {
3042          return merge(inNames, deleteInputs, 
3043                       (inNames.length < factor) ? inNames.length : factor,
3044                       tmpDir);
3045        }
3046    
3047        /**
3048         * Merges the contents of files passed in Path[]
3049         * @param inNames the array of path names
3050         * @param deleteInputs true if the input files should be deleted when 
3051         * unnecessary
3052         * @param factor the factor that will be used as the maximum merge fan-in
3053         * @param tmpDir the directory to write temporary files into
3054         * @return RawKeyValueIteratorMergeQueue
3055         * @throws IOException
3056         */
3057        public RawKeyValueIterator merge(Path [] inNames, boolean deleteInputs,
3058                                         int factor, Path tmpDir) 
3059          throws IOException {
3060          //get the segments from inNames
3061          ArrayList <SegmentDescriptor> a = new ArrayList <SegmentDescriptor>();
3062          for (int i = 0; i < inNames.length; i++) {
3063            SegmentDescriptor s = new SegmentDescriptor(0,
3064                fs.getFileStatus(inNames[i]).getLen(), inNames[i]);
3065            s.preserveInput(!deleteInputs);
3066            s.doSync();
3067            a.add(s);
3068          }
3069          this.factor = factor;
3070          MergeQueue mQueue = new MergeQueue(a, tmpDir, progressable);
3071          return mQueue.merge();
3072        }
3073    
3074        /**
3075         * Merges the contents of files passed in Path[]
3076         * @param inNames the array of path names
3077         * @param tempDir the directory for creating temp files during merge
3078         * @param deleteInputs true if the input files should be deleted when 
3079         * unnecessary
3080         * @return RawKeyValueIteratorMergeQueue
3081         * @throws IOException
3082         */
3083        public RawKeyValueIterator merge(Path [] inNames, Path tempDir, 
3084                                         boolean deleteInputs) 
3085          throws IOException {
3086          //outFile will basically be used as prefix for temp files for the
3087          //intermediate merge outputs           
3088          this.outFile = new Path(tempDir + Path.SEPARATOR + "merged");
3089          //get the segments from inNames
3090          ArrayList <SegmentDescriptor> a = new ArrayList <SegmentDescriptor>();
3091          for (int i = 0; i < inNames.length; i++) {
3092            SegmentDescriptor s = new SegmentDescriptor(0,
3093                fs.getFileStatus(inNames[i]).getLen(), inNames[i]);
3094            s.preserveInput(!deleteInputs);
3095            s.doSync();
3096            a.add(s);
3097          }
3098          factor = (inNames.length < factor) ? inNames.length : factor;
3099          // pass in object to report progress, if present
3100          MergeQueue mQueue = new MergeQueue(a, tempDir, progressable);
3101          return mQueue.merge();
3102        }
3103    
3104        /**
3105         * Clones the attributes (like compression of the input file and creates a 
3106         * corresponding Writer
3107         * @param inputFile the path of the input file whose attributes should be 
3108         * cloned
3109         * @param outputFile the path of the output file 
3110         * @param prog the Progressable to report status during the file write
3111         * @return Writer
3112         * @throws IOException
3113         */
3114        public Writer cloneFileAttributes(Path inputFile, Path outputFile, 
3115                                          Progressable prog) throws IOException {
3116          Reader reader = new Reader(conf,
3117                                     Reader.file(inputFile),
3118                                     new Reader.OnlyHeaderOption());
3119          CompressionType compress = reader.getCompressionType();
3120          CompressionCodec codec = reader.getCompressionCodec();
3121          reader.close();
3122    
3123          Writer writer = createWriter(conf, 
3124                                       Writer.file(outputFile), 
3125                                       Writer.keyClass(keyClass), 
3126                                       Writer.valueClass(valClass), 
3127                                       Writer.compression(compress, codec), 
3128                                       Writer.progressable(prog));
3129          return writer;
3130        }
3131    
3132        /**
3133         * Writes records from RawKeyValueIterator into a file represented by the 
3134         * passed writer
3135         * @param records the RawKeyValueIterator
3136         * @param writer the Writer created earlier 
3137         * @throws IOException
3138         */
3139        public void writeFile(RawKeyValueIterator records, Writer writer) 
3140          throws IOException {
3141          while(records.next()) {
3142            writer.appendRaw(records.getKey().getData(), 0, 
3143                             records.getKey().getLength(), records.getValue());
3144          }
3145          writer.sync();
3146        }
3147            
3148        /** Merge the provided files.
3149         * @param inFiles the array of input path names
3150         * @param outFile the final output file
3151         * @throws IOException
3152         */
3153        public void merge(Path[] inFiles, Path outFile) throws IOException {
3154          if (fs.exists(outFile)) {
3155            throw new IOException("already exists: " + outFile);
3156          }
3157          RawKeyValueIterator r = merge(inFiles, false, outFile.getParent());
3158          Writer writer = cloneFileAttributes(inFiles[0], outFile, null);
3159          
3160          writeFile(r, writer);
3161    
3162          writer.close();
3163        }
3164    
3165        /** sort calls this to generate the final merged output */
3166        private int mergePass(Path tmpDir) throws IOException {
3167          if(LOG.isDebugEnabled()) {
3168            LOG.debug("running merge pass");
3169          }
3170          Writer writer = cloneFileAttributes(
3171                                              outFile.suffix(".0"), outFile, null);
3172          RawKeyValueIterator r = merge(outFile.suffix(".0"), 
3173                                        outFile.suffix(".0.index"), tmpDir);
3174          writeFile(r, writer);
3175    
3176          writer.close();
3177          return 0;
3178        }
3179    
3180        /** Used by mergePass to merge the output of the sort
3181         * @param inName the name of the input file containing sorted segments
3182         * @param indexIn the offsets of the sorted segments
3183         * @param tmpDir the relative directory to store intermediate results in
3184         * @return RawKeyValueIterator
3185         * @throws IOException
3186         */
3187        private RawKeyValueIterator merge(Path inName, Path indexIn, Path tmpDir) 
3188          throws IOException {
3189          //get the segments from indexIn
3190          //we create a SegmentContainer so that we can track segments belonging to
3191          //inName and delete inName as soon as we see that we have looked at all
3192          //the contained segments during the merge process & hence don't need 
3193          //them anymore
3194          SegmentContainer container = new SegmentContainer(inName, indexIn);
3195          MergeQueue mQueue = new MergeQueue(container.getSegmentList(), tmpDir, progressable);
3196          return mQueue.merge();
3197        }
3198        
3199        /** This class implements the core of the merge logic */
3200        private class MergeQueue extends PriorityQueue 
3201          implements RawKeyValueIterator {
3202          private boolean compress;
3203          private boolean blockCompress;
3204          private DataOutputBuffer rawKey = new DataOutputBuffer();
3205          private ValueBytes rawValue;
3206          private long totalBytesProcessed;
3207          private float progPerByte;
3208          private Progress mergeProgress = new Progress();
3209          private Path tmpDir;
3210          private Progressable progress = null; //handle to the progress reporting object
3211          private SegmentDescriptor minSegment;
3212          
3213          //a TreeMap used to store the segments sorted by size (segment offset and
3214          //segment path name is used to break ties between segments of same sizes)
3215          private Map<SegmentDescriptor, Void> sortedSegmentSizes =
3216            new TreeMap<SegmentDescriptor, Void>();
3217                
3218          @SuppressWarnings("unchecked")
3219          public void put(SegmentDescriptor stream) throws IOException {
3220            if (size() == 0) {
3221              compress = stream.in.isCompressed();
3222              blockCompress = stream.in.isBlockCompressed();
3223            } else if (compress != stream.in.isCompressed() || 
3224                       blockCompress != stream.in.isBlockCompressed()) {
3225              throw new IOException("All merged files must be compressed or not.");
3226            } 
3227            super.put(stream);
3228          }
3229          
3230          /**
3231           * A queue of file segments to merge
3232           * @param segments the file segments to merge
3233           * @param tmpDir a relative local directory to save intermediate files in
3234           * @param progress the reference to the Progressable object
3235           */
3236          public MergeQueue(List <SegmentDescriptor> segments,
3237              Path tmpDir, Progressable progress) {
3238            int size = segments.size();
3239            for (int i = 0; i < size; i++) {
3240              sortedSegmentSizes.put(segments.get(i), null);
3241            }
3242            this.tmpDir = tmpDir;
3243            this.progress = progress;
3244          }
3245          @Override
3246          protected boolean lessThan(Object a, Object b) {
3247            // indicate we're making progress
3248            if (progress != null) {
3249              progress.progress();
3250            }
3251            SegmentDescriptor msa = (SegmentDescriptor)a;
3252            SegmentDescriptor msb = (SegmentDescriptor)b;
3253            return comparator.compare(msa.getKey().getData(), 0, 
3254                                      msa.getKey().getLength(), msb.getKey().getData(), 0, 
3255                                      msb.getKey().getLength()) < 0;
3256          }
3257          @Override
3258          public void close() throws IOException {
3259            SegmentDescriptor ms;                           // close inputs
3260            while ((ms = (SegmentDescriptor)pop()) != null) {
3261              ms.cleanup();
3262            }
3263            minSegment = null;
3264          }
3265          @Override
3266          public DataOutputBuffer getKey() throws IOException {
3267            return rawKey;
3268          }
3269          @Override
3270          public ValueBytes getValue() throws IOException {
3271            return rawValue;
3272          }
3273          @Override
3274          public boolean next() throws IOException {
3275            if (size() == 0)
3276              return false;
3277            if (minSegment != null) {
3278              //minSegment is non-null for all invocations of next except the first
3279              //one. For the first invocation, the priority queue is ready for use
3280              //but for the subsequent invocations, first adjust the queue 
3281              adjustPriorityQueue(minSegment);
3282              if (size() == 0) {
3283                minSegment = null;
3284                return false;
3285              }
3286            }
3287            minSegment = (SegmentDescriptor)top();
3288            long startPos = minSegment.in.getPosition(); // Current position in stream
3289            //save the raw key reference
3290            rawKey = minSegment.getKey();
3291            //load the raw value. Re-use the existing rawValue buffer
3292            if (rawValue == null) {
3293              rawValue = minSegment.in.createValueBytes();
3294            }
3295            minSegment.nextRawValue(rawValue);
3296            long endPos = minSegment.in.getPosition(); // End position after reading value
3297            updateProgress(endPos - startPos);
3298            return true;
3299          }
3300          
3301          @Override
3302          public Progress getProgress() {
3303            return mergeProgress; 
3304          }
3305    
3306          private void adjustPriorityQueue(SegmentDescriptor ms) throws IOException{
3307            long startPos = ms.in.getPosition(); // Current position in stream
3308            boolean hasNext = ms.nextRawKey();
3309            long endPos = ms.in.getPosition(); // End position after reading key
3310            updateProgress(endPos - startPos);
3311            if (hasNext) {
3312              adjustTop();
3313            } else {
3314              pop();
3315              ms.cleanup();
3316            }
3317          }
3318    
3319          private void updateProgress(long bytesProcessed) {
3320            totalBytesProcessed += bytesProcessed;
3321            if (progPerByte > 0) {
3322              mergeProgress.set(totalBytesProcessed * progPerByte);
3323            }
3324          }
3325          
3326          /** This is the single level merge that is called multiple times 
3327           * depending on the factor size and the number of segments
3328           * @return RawKeyValueIterator
3329           * @throws IOException
3330           */
3331          public RawKeyValueIterator merge() throws IOException {
3332            //create the MergeStreams from the sorted map created in the constructor
3333            //and dump the final output to a file
3334            int numSegments = sortedSegmentSizes.size();
3335            int origFactor = factor;
3336            int passNo = 1;
3337            LocalDirAllocator lDirAlloc = new LocalDirAllocator("io.seqfile.local.dir");
3338            do {
3339              //get the factor for this pass of merge
3340              factor = getPassFactor(passNo, numSegments);
3341              List<SegmentDescriptor> segmentsToMerge =
3342                new ArrayList<SegmentDescriptor>();
3343              int segmentsConsidered = 0;
3344              int numSegmentsToConsider = factor;
3345              while (true) {
3346                //extract the smallest 'factor' number of segment pointers from the 
3347                //TreeMap. Call cleanup on the empty segments (no key/value data)
3348                SegmentDescriptor[] mStream = 
3349                  getSegmentDescriptors(numSegmentsToConsider);
3350                for (int i = 0; i < mStream.length; i++) {
3351                  if (mStream[i].nextRawKey()) {
3352                    segmentsToMerge.add(mStream[i]);
3353                    segmentsConsidered++;
3354                    // Count the fact that we read some bytes in calling nextRawKey()
3355                    updateProgress(mStream[i].in.getPosition());
3356                  }
3357                  else {
3358                    mStream[i].cleanup();
3359                    numSegments--; //we ignore this segment for the merge
3360                  }
3361                }
3362                //if we have the desired number of segments
3363                //or looked at all available segments, we break
3364                if (segmentsConsidered == factor || 
3365                    sortedSegmentSizes.size() == 0) {
3366                  break;
3367                }
3368                  
3369                numSegmentsToConsider = factor - segmentsConsidered;
3370              }
3371              //feed the streams to the priority queue
3372              initialize(segmentsToMerge.size()); clear();
3373              for (int i = 0; i < segmentsToMerge.size(); i++) {
3374                put(segmentsToMerge.get(i));
3375              }
3376              //if we have lesser number of segments remaining, then just return the
3377              //iterator, else do another single level merge
3378              if (numSegments <= factor) {
3379                //calculate the length of the remaining segments. Required for 
3380                //calculating the merge progress
3381                long totalBytes = 0;
3382                for (int i = 0; i < segmentsToMerge.size(); i++) {
3383                  totalBytes += segmentsToMerge.get(i).segmentLength;
3384                }
3385                if (totalBytes != 0) //being paranoid
3386                  progPerByte = 1.0f / (float)totalBytes;
3387                //reset factor to what it originally was
3388                factor = origFactor;
3389                return this;
3390              } else {
3391                //we want to spread the creation of temp files on multiple disks if 
3392                //available under the space constraints
3393                long approxOutputSize = 0; 
3394                for (SegmentDescriptor s : segmentsToMerge) {
3395                  approxOutputSize += s.segmentLength + 
3396                                      ChecksumFileSystem.getApproxChkSumLength(
3397                                      s.segmentLength);
3398                }
3399                Path tmpFilename = 
3400                  new Path(tmpDir, "intermediate").suffix("." + passNo);
3401    
3402                Path outputFile =  lDirAlloc.getLocalPathForWrite(
3403                                                    tmpFilename.toString(),
3404                                                    approxOutputSize, conf);
3405                if(LOG.isDebugEnabled()) { 
3406                  LOG.debug("writing intermediate results to " + outputFile);
3407                }
3408                Writer writer = cloneFileAttributes(
3409                                                    fs.makeQualified(segmentsToMerge.get(0).segmentPathName), 
3410                                                    fs.makeQualified(outputFile), null);
3411                writer.sync = null; //disable sync for temp files
3412                writeFile(this, writer);
3413                writer.close();
3414                
3415                //we finished one single level merge; now clean up the priority 
3416                //queue
3417                this.close();
3418                
3419                SegmentDescriptor tempSegment = 
3420                  new SegmentDescriptor(0,
3421                      fs.getFileStatus(outputFile).getLen(), outputFile);
3422                //put the segment back in the TreeMap
3423                sortedSegmentSizes.put(tempSegment, null);
3424                numSegments = sortedSegmentSizes.size();
3425                passNo++;
3426              }
3427              //we are worried about only the first pass merge factor. So reset the 
3428              //factor to what it originally was
3429              factor = origFactor;
3430            } while(true);
3431          }
3432      
3433          //Hadoop-591
3434          public int getPassFactor(int passNo, int numSegments) {
3435            if (passNo > 1 || numSegments <= factor || factor == 1) 
3436              return factor;
3437            int mod = (numSegments - 1) % (factor - 1);
3438            if (mod == 0)
3439              return factor;
3440            return mod + 1;
3441          }
3442          
3443          /** Return (& remove) the requested number of segment descriptors from the
3444           * sorted map.
3445           */
3446          public SegmentDescriptor[] getSegmentDescriptors(int numDescriptors) {
3447            if (numDescriptors > sortedSegmentSizes.size())
3448              numDescriptors = sortedSegmentSizes.size();
3449            SegmentDescriptor[] SegmentDescriptors = 
3450              new SegmentDescriptor[numDescriptors];
3451            Iterator iter = sortedSegmentSizes.keySet().iterator();
3452            int i = 0;
3453            while (i < numDescriptors) {
3454              SegmentDescriptors[i++] = (SegmentDescriptor)iter.next();
3455              iter.remove();
3456            }
3457            return SegmentDescriptors;
3458          }
3459        } // SequenceFile.Sorter.MergeQueue
3460    
3461        /** This class defines a merge segment. This class can be subclassed to 
3462         * provide a customized cleanup method implementation. In this 
3463         * implementation, cleanup closes the file handle and deletes the file 
3464         */
3465        public class SegmentDescriptor implements Comparable {
3466          
3467          long segmentOffset; //the start of the segment in the file
3468          long segmentLength; //the length of the segment
3469          Path segmentPathName; //the path name of the file containing the segment
3470          boolean ignoreSync = true; //set to true for temp files
3471          private Reader in = null; 
3472          private DataOutputBuffer rawKey = null; //this will hold the current key
3473          private boolean preserveInput = false; //delete input segment files?
3474          
3475          /** Constructs a segment
3476           * @param segmentOffset the offset of the segment in the file
3477           * @param segmentLength the length of the segment
3478           * @param segmentPathName the path name of the file containing the segment
3479           */
3480          public SegmentDescriptor (long segmentOffset, long segmentLength, 
3481                                    Path segmentPathName) {
3482            this.segmentOffset = segmentOffset;
3483            this.segmentLength = segmentLength;
3484            this.segmentPathName = segmentPathName;
3485          }
3486          
3487          /** Do the sync checks */
3488          public void doSync() {ignoreSync = false;}
3489          
3490          /** Whether to delete the files when no longer needed */
3491          public void preserveInput(boolean preserve) {
3492            preserveInput = preserve;
3493          }
3494    
3495          public boolean shouldPreserveInput() {
3496            return preserveInput;
3497          }
3498          
3499          @Override
3500          public int compareTo(Object o) {
3501            SegmentDescriptor that = (SegmentDescriptor)o;
3502            if (this.segmentLength != that.segmentLength) {
3503              return (this.segmentLength < that.segmentLength ? -1 : 1);
3504            }
3505            if (this.segmentOffset != that.segmentOffset) {
3506              return (this.segmentOffset < that.segmentOffset ? -1 : 1);
3507            }
3508            return (this.segmentPathName.toString()).
3509              compareTo(that.segmentPathName.toString());
3510          }
3511    
3512          @Override
3513          public boolean equals(Object o) {
3514            if (!(o instanceof SegmentDescriptor)) {
3515              return false;
3516            }
3517            SegmentDescriptor that = (SegmentDescriptor)o;
3518            if (this.segmentLength == that.segmentLength &&
3519                this.segmentOffset == that.segmentOffset &&
3520                this.segmentPathName.toString().equals(
3521                  that.segmentPathName.toString())) {
3522              return true;
3523            }
3524            return false;
3525          }
3526    
3527          @Override
3528          public int hashCode() {
3529            return 37 * 17 + (int) (segmentOffset^(segmentOffset>>>32));
3530          }
3531    
3532          /** Fills up the rawKey object with the key returned by the Reader
3533           * @return true if there is a key returned; false, otherwise
3534           * @throws IOException
3535           */
3536          public boolean nextRawKey() throws IOException {
3537            if (in == null) {
3538              int bufferSize = getBufferSize(conf); 
3539              Reader reader = new Reader(conf,
3540                                         Reader.file(segmentPathName), 
3541                                         Reader.bufferSize(bufferSize),
3542                                         Reader.start(segmentOffset), 
3543                                         Reader.length(segmentLength));
3544            
3545              //sometimes we ignore syncs especially for temp merge files
3546              if (ignoreSync) reader.ignoreSync();
3547    
3548              if (reader.getKeyClass() != keyClass)
3549                throw new IOException("wrong key class: " + reader.getKeyClass() +
3550                                      " is not " + keyClass);
3551              if (reader.getValueClass() != valClass)
3552                throw new IOException("wrong value class: "+reader.getValueClass()+
3553                                      " is not " + valClass);
3554              this.in = reader;
3555              rawKey = new DataOutputBuffer();
3556            }
3557            rawKey.reset();
3558            int keyLength = 
3559              in.nextRawKey(rawKey);
3560            return (keyLength >= 0);
3561          }
3562    
3563          /** Fills up the passed rawValue with the value corresponding to the key
3564           * read earlier
3565           * @param rawValue
3566           * @return the length of the value
3567           * @throws IOException
3568           */
3569          public int nextRawValue(ValueBytes rawValue) throws IOException {
3570            int valLength = in.nextRawValue(rawValue);
3571            return valLength;
3572          }
3573          
3574          /** Returns the stored rawKey */
3575          public DataOutputBuffer getKey() {
3576            return rawKey;
3577          }
3578          
3579          /** closes the underlying reader */
3580          private void close() throws IOException {
3581            this.in.close();
3582            this.in = null;
3583          }
3584    
3585          /** The default cleanup. Subclasses can override this with a custom 
3586           * cleanup 
3587           */
3588          public void cleanup() throws IOException {
3589            close();
3590            if (!preserveInput) {
3591              fs.delete(segmentPathName, true);
3592            }
3593          }
3594        } // SequenceFile.Sorter.SegmentDescriptor
3595        
3596        /** This class provisions multiple segments contained within a single
3597         *  file
3598         */
3599        private class LinkedSegmentsDescriptor extends SegmentDescriptor {
3600    
3601          SegmentContainer parentContainer = null;
3602    
3603          /** Constructs a segment
3604           * @param segmentOffset the offset of the segment in the file
3605           * @param segmentLength the length of the segment
3606           * @param segmentPathName the path name of the file containing the segment
3607           * @param parent the parent SegmentContainer that holds the segment
3608           */
3609          public LinkedSegmentsDescriptor (long segmentOffset, long segmentLength, 
3610                                           Path segmentPathName, SegmentContainer parent) {
3611            super(segmentOffset, segmentLength, segmentPathName);
3612            this.parentContainer = parent;
3613          }
3614          /** The default cleanup. Subclasses can override this with a custom 
3615           * cleanup 
3616           */
3617          @Override
3618          public void cleanup() throws IOException {
3619            super.close();
3620            if (super.shouldPreserveInput()) return;
3621            parentContainer.cleanup();
3622          }
3623          
3624          @Override
3625          public boolean equals(Object o) {
3626            if (!(o instanceof LinkedSegmentsDescriptor)) {
3627              return false;
3628            }
3629            return super.equals(o);
3630          }
3631        } //SequenceFile.Sorter.LinkedSegmentsDescriptor
3632    
3633        /** The class that defines a container for segments to be merged. Primarily
3634         * required to delete temp files as soon as all the contained segments
3635         * have been looked at */
3636        private class SegmentContainer {
3637          private int numSegmentsCleanedUp = 0; //track the no. of segment cleanups
3638          private int numSegmentsContained; //# of segments contained
3639          private Path inName; //input file from where segments are created
3640          
3641          //the list of segments read from the file
3642          private ArrayList <SegmentDescriptor> segments = 
3643            new ArrayList <SegmentDescriptor>();
3644          /** This constructor is there primarily to serve the sort routine that 
3645           * generates a single output file with an associated index file */
3646          public SegmentContainer(Path inName, Path indexIn) throws IOException {
3647            //get the segments from indexIn
3648            FSDataInputStream fsIndexIn = fs.open(indexIn);
3649            long end = fs.getFileStatus(indexIn).getLen();
3650            while (fsIndexIn.getPos() < end) {
3651              long segmentOffset = WritableUtils.readVLong(fsIndexIn);
3652              long segmentLength = WritableUtils.readVLong(fsIndexIn);
3653              Path segmentName = inName;
3654              segments.add(new LinkedSegmentsDescriptor(segmentOffset, 
3655                                                        segmentLength, segmentName, this));
3656            }
3657            fsIndexIn.close();
3658            fs.delete(indexIn, true);
3659            numSegmentsContained = segments.size();
3660            this.inName = inName;
3661          }
3662    
3663          public List <SegmentDescriptor> getSegmentList() {
3664            return segments;
3665          }
3666          public void cleanup() throws IOException {
3667            numSegmentsCleanedUp++;
3668            if (numSegmentsCleanedUp == numSegmentsContained) {
3669              fs.delete(inName, true);
3670            }
3671          }
3672        } //SequenceFile.Sorter.SegmentContainer
3673    
3674      } // SequenceFile.Sorter
3675    
3676    } // SequenceFile