public void readFields(DataInput in) throws IOException {
  byte version = in.readByte();
  switch(version) {
  case 1:
    code = in.readByte();
    lastModified = in.readLong();
    args = WritableUtils.readCompressedStringArray(in);
    break;
  case VERSION:
    code = in.readByte();
    lastModified = in.readLong();
    args = WritableUtils.readStringArray(in);
    break;
  default:
    throw new VersionMismatchException(VERSION, version);
  }
}
 

public void readFields(DataInput in) throws IOException {
   byte version = in.readByte();
   switch(version) {
   case 1:
     majorCode = in.readByte();
     minorCode = in.readShort();
     args = WritableUtils.readCompressedStringArray(in);
     break;
   case 2:
     majorCode = in.readByte();
     minorCode = in.readShort();
     args = WritableUtils.readStringArray(in);
     break;
   default:
     throw new VersionMismatchException(VERSION, version);
   }
}
 

public void readFields(DataInput in) throws IOException {
  fields.clear();
  byte version = in.readByte();
  if (version != VERSION) {
    throw new VersionMismatchException(VERSION, version);
  }
  int size = WritableUtils.readVInt(in);
  for (int i = 0; i < size; i++) {
    String name = Text.readString(in);
    NutchField field = new NutchField();
    field.readFields(in);
    fields.put(name, field);
  }
  weight = in.readFloat();
  documentMeta.readFields(in);
}
 

public void readFields(DataInput in) throws IOException {
  byte version = in.readByte();
  switch(version) {
  case 1:
    code = in.readByte();
    lastModified = in.readLong();
    args = WritableUtils.readCompressedStringArray(in);
    break;
  case VERSION:
    code = in.readByte();
    lastModified = in.readLong();
    args = WritableUtils.readStringArray(in);
    break;
  default:
    throw new VersionMismatchException(VERSION, version);
  }
}
 

public void readFields(DataInput in) throws IOException {
   byte version = in.readByte();
   switch(version) {
   case 1:
     majorCode = in.readByte();
     minorCode = in.readShort();
     args = WritableUtils.readCompressedStringArray(in);
     break;
   case 2:
     majorCode = in.readByte();
     minorCode = in.readShort();
     args = WritableUtils.readStringArray(in);
     break;
   default:
     throw new VersionMismatchException(VERSION, version);
   }
}
 

public void readFields(DataInput in) throws IOException {
  fields.clear();
  byte version = in.readByte();
  if (version != VERSION) {
    throw new VersionMismatchException(VERSION, version);
  }
  int size = WritableUtils.readVInt(in);
  for (int i = 0; i < size; i++) {
    String name = Text.readString(in);
    NutchField field = new NutchField();
    field.readFields(in);
    fields.put(name, field);
  }
  weight = in.readFloat();
  documentMeta.readFields(in);
}
 

private final void readFieldsCompressed(DataInput in) throws IOException {
  byte oldVersion = in.readByte();
  switch (oldVersion) {
  case 0:
  case 1:
    url = Text.readString(in); // read url
    base = Text.readString(in); // read base

    content = new byte[in.readInt()]; // read content
    in.readFully(content);

    contentType = Text.readString(in); // read contentType
    // reconstruct metadata
    int keySize = in.readInt();
    String key;
    for (int i = 0; i < keySize; i++) {
      key = Text.readString(in);
      int valueSize = in.readInt();
      for (int j = 0; j < valueSize; j++) {
        metadata.add(key, Text.readString(in));
      }
    }
    break;
  case 2:
    url = Text.readString(in); // read url
    base = Text.readString(in); // read base

    content = new byte[in.readInt()]; // read content
    in.readFully(content);

    contentType = Text.readString(in); // read contentType
    metadata.readFields(in); // read meta data
    break;
  default:
    throw new VersionMismatchException((byte)2, oldVersion);
  }

}
 

public final void readFields(DataInput in) throws IOException {
  metadata.clear();
  int sizeOrVersion = in.readInt();
  if (sizeOrVersion < 0) { // version
    version = sizeOrVersion;
    switch (version) {
    case VERSION:
      url = Text.readString(in);
      base = Text.readString(in);

      content = new byte[in.readInt()];
      in.readFully(content);

      contentType = Text.readString(in);
      metadata.readFields(in);
      break;
    default:
      throw new VersionMismatchException((byte)VERSION, (byte)version);
    }
  } else { // size
    byte[] compressed = new byte[sizeOrVersion];
    in.readFully(compressed, 0, compressed.length);
    ByteArrayInputStream deflated = new ByteArrayInputStream(compressed);
    DataInput inflater =
      new DataInputStream(new InflaterInputStream(deflated));
    readFieldsCompressed(inflater);
  }
}
 

private final void readFieldsCompressed(DataInput in) throws IOException {
  byte oldVersion = in.readByte();
  switch (oldVersion) {
  case 0:
  case 1:
    url = Text.readString(in); // read url
    base = Text.readString(in); // read base

    content = new byte[in.readInt()]; // read content
    in.readFully(content);

    contentType = Text.readString(in); // read contentType
    // reconstruct metadata
    int keySize = in.readInt();
    String key;
    for (int i = 0; i < keySize; i++) {
      key = Text.readString(in);
      int valueSize = in.readInt();
      for (int j = 0; j < valueSize; j++) {
        metadata.add(key, Text.readString(in));
      }
    }
    break;
  case 2:
    url = Text.readString(in); // read url
    base = Text.readString(in); // read base

    content = new byte[in.readInt()]; // read content
    in.readFully(content);

    contentType = Text.readString(in); // read contentType
    metadata.readFields(in); // read meta data
    break;
  default:
    throw new VersionMismatchException((byte)2, oldVersion);
  }

}
 

public final void readFields(DataInput in) throws IOException {
  metadata.clear();
  int sizeOrVersion = in.readInt();
  if (sizeOrVersion < 0) { // version
    version = sizeOrVersion;
    switch (version) {
    case VERSION:
      url = Text.readString(in);
      base = Text.readString(in);

      content = new byte[in.readInt()];
      in.readFully(content);

      contentType = Text.readString(in);
      metadata.readFields(in);
      break;
    default:
      throw new VersionMismatchException((byte)VERSION, (byte)version);
    }
  } else { // size
    byte[] compressed = new byte[sizeOrVersion];
    in.readFully(compressed, 0, compressed.length);
    ByteArrayInputStream deflated = new ByteArrayInputStream(compressed);
    DataInput inflater =
      new DataInputStream(new InflaterInputStream(deflated));
    readFieldsCompressed(inflater);
  }
}
 

Writer(Configuration conf, Option... opts) throws IOException {
  BlockSizeOption blockSizeOption =
      Options.getOption(BlockSizeOption.class, opts);
  BufferSizeOption bufferSizeOption =
      Options.getOption(BufferSizeOption.class, opts);
  ReplicationOption replicationOption =
      Options.getOption(ReplicationOption.class, opts);

  FileOption fileOption = Options.getOption(FileOption.class, opts);
  AppendIfExistsOption appendIfExistsOption = Options.getOption(
      AppendIfExistsOption.class, opts);
  StreamOption streamOption = Options.getOption(StreamOption.class, opts);

  // check consistency of options
  if ((fileOption == null) == (streamOption == null)) {
    throw new IllegalArgumentException("file or stream must be specified");
  }
  if (fileOption == null && (blockSizeOption != null ||
                             bufferSizeOption != null ||
                             replicationOption != null)) {
    throw new IllegalArgumentException("file modifier options not " +
                                       "compatible with stream");
  }

  FSDataOutputStream out;
  boolean ownStream = fileOption != null;
  if (ownStream) {
    Path p = fileOption.getValue();
    FileSystem fs;
    fs = p.getFileSystem(conf);
    int bufferSize = bufferSizeOption == null ? getBufferSize(conf) :
                     bufferSizeOption.getValue();
    short replication = replicationOption == null ?
                        fs.getDefaultReplication(p) :
                        (short) replicationOption.getValue();
    long blockSize = blockSizeOption == null ? fs.getDefaultBlockSize(p) :
                     blockSizeOption.getValue();

    if (appendIfExistsOption != null && appendIfExistsOption.getValue()
        && fs.exists(p)) {
      // Read the file and verify header details
      try (WALFile.Reader reader =
               new WALFile.Reader(conf, WALFile.Reader.file(p), new Reader.OnlyHeaderOption())){
        if (reader.getVersion() != VERSION[3]) {
          throw new VersionMismatchException(VERSION[3], reader.getVersion());
        }
        sync = reader.getSync();
      }
      out = fs.append(p, bufferSize);
      this.appendMode = true;
    } else {
      out = fs.create(p, true, bufferSize, replication, blockSize);
    }
  } else {
    out = streamOption.getValue();
  }

  init(conf, out, ownStream);
}
 

/**
 * Initialize the {@link Reader}
 *
 * @param tempReader <code>true</code> if we are constructing a temporary and hence do not
 *                   initialize every component; <code>false</code> otherwise.
 */
private void init(boolean tempReader) throws IOException {
  byte[] versionBlock = new byte[VERSION.length];
  in.readFully(versionBlock);

  if ((versionBlock[0] != VERSION[0]) ||
      (versionBlock[1] != VERSION[1]) ||
      (versionBlock[2] != VERSION[2])) {
    throw new IOException(this + " not a WALFile");
  }

  // Set 'version'
  version = versionBlock[3];
  if (version > VERSION[3]) {
    throw new VersionMismatchException(VERSION[3], version);
  }

  in.readFully(sync);                       // read sync bytes
  headerEnd = in.getPos();                  // record end of header

  // Initialize... *not* if this we are constructing a temporary Reader
  if (!tempReader) {
    valBuffer = new DataInputBuffer();
    valIn = valBuffer;

    SerializationFactory serializationFactory =
        new SerializationFactory(conf);
    this.keyDeserializer =
        getDeserializer(serializationFactory, WALEntry.class);
    if (this.keyDeserializer == null) {
      throw new IOException(
          "Could not find a deserializer for the Key class: '"
          + WALFile.class.getCanonicalName() + "'. "
          + "Please ensure that the configuration '" +
          CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
          + "properly configured, if you're using "
          + "custom serialization.");
    }

    this.keyDeserializer.open(valBuffer);

    this.valDeserializer =
        getDeserializer(serializationFactory, WALEntry.class);
    if (this.valDeserializer == null) {
      throw new IOException(
          "Could not find a deserializer for the Value class: '"
          + WALEntry.class.getCanonicalName() + "'. "
          + "Please ensure that the configuration '" +
          CommonConfigurationKeys.IO_SERIALIZATIONS_KEY + "' is "
          + "properly configured, if you're using "
          + "custom serialization.");
    }
    this.valDeserializer.open(valIn);
  }
}