public void testFailureNegativeOffset() throws IOException {
  if (skip)
    return;
  writeRecords(2, true, true);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  byte[] buf = new byte[K];
  try {
    scanner.entry().getKey(buf, -1);
    Assert.fail("Failed to handle key negative offset.");
  }
  catch (Exception e) {
    // noop, expecting exceptions
  }
  finally {
  }
  scanner.close();
  reader.close();
}
 

private int readAndCheckbytes(Scanner scanner, int start, int n)
    throws IOException {
  String value = "value";
  for (int i = start; i < (start + n); i++) {
    byte[] key = readKey(scanner);
    byte[] val = readValue(scanner);
    String keyStr = String.format(localFormatter, i);
    String valStr = value + keyStr;
    assertTrue("btyes for keys do not match " + keyStr + " "
        + new String(key), Arrays.equals(keyStr.getBytes(), key));
    assertTrue("bytes for vals do not match " + valStr + " "
        + new String(val), Arrays.equals(
        valStr.getBytes(), val));
    assertTrue(scanner.advance());
    key = readKey(scanner);
    val = readValue(scanner);
    assertTrue("btyes for keys do not match", Arrays.equals(
        keyStr.getBytes(), key));
    assertTrue("bytes for vals do not match", Arrays.equals(
        valStr.getBytes(), val));
    assertTrue(scanner.advance());
  }
  return (start + n);
}
 

private int readPrepWithUnknownLength(Scanner scanner, int start, int n)
    throws IOException {
  for (int i = start; i < start; i++) {
    String key = String.format(localFormatter, i);
    byte[] read = readKey(scanner);
    assertTrue("keys not equal", Arrays.equals(key.getBytes(), read));
    try {
      read = readValue(scanner);
      assertTrue(false);
    }
    catch (IOException ie) {
      // should have thrown exception
    }
    String value = "value" + key;
    read = readLongValue(scanner, value.getBytes().length);
    assertTrue("values nto equal", Arrays.equals(read, value.getBytes()));
    scanner.advance();
  }
  return (start + n);
}
 

void unsortedWithSomeCodec(String codec) throws IOException {
  Path uTfile = new Path(ROOT, "unsorted.tfile");
  FSDataOutputStream fout = createFSOutput(uTfile);
  Writer writer = new Writer(fout, minBlockSize, codec, null, conf);
  writeRecords(writer);
  writer.close();
  fout.close();
  FSDataInputStream fin = fs.open(uTfile);
  Reader reader =
      new Reader(fs.open(uTfile), fs.getFileStatus(uTfile).getLen(), conf);

  Scanner scanner = reader.createScanner();
  readAllRecords(scanner);
  scanner.close();
  reader.close();
  fin.close();
  fs.delete(uTfile, true);
}
 

public void testFailureNegativeOffset() throws IOException {
  if (skip)
    return;
  writeRecords(2, true, true);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  byte[] buf = new byte[K];
  try {
    scanner.entry().getKey(buf, -1);
    Assert.fail("Failed to handle key negative offset.");
  }
  catch (Exception e) {
    // noop, expecting exceptions
  }
  finally {
  }
  scanner.close();
  reader.close();
}
 

@Test
public void testFailureReadValueManyTimes() throws IOException {
  if (skip)
    return;
  writeRecords(5);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();

  byte[] vbuf = new byte[BUF_SIZE];
  int vlen = scanner.entry().getValueLength();
  scanner.entry().getValue(vbuf);
  Assert.assertEquals(new String(vbuf, 0, vlen), VALUE + 0);
  try {
    scanner.entry().getValue(vbuf);
    Assert.fail("Cannot get the value mlutiple times.");
  } catch (Exception e) {
    // noop, expecting exceptions
  }

  scanner.close();
  reader.close();
}
 

@Test
public void testFailureNegativeOffset_2() throws IOException {
  if (skip)
    return;
  closeOutput();

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  try {
    scanner.lowerBound("keyX".getBytes(), -1, 4);
    Assert.fail("Error on handling negative offset.");
  } catch (Exception e) {
    // noop, expecting exceptions
  } finally {
    reader.close();
    scanner.close();
  }
  closeOutput();
}
 

@Test
public void testFailureNegativeLength_2() throws IOException {
  if (skip)
    return;
  closeOutput();

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  try {
    scanner.lowerBound("keyX".getBytes(), 0, -1);
    Assert.fail("Error on handling negative length.");
  } catch (Exception e) {
    // noop, expecting exceptions
  } finally {
    scanner.close();
    reader.close();
  }
  closeOutput();
}
 

private void readValueBeforeKey(int recordIndex)
    throws IOException {
  Reader reader =
      new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner =
      reader.createScannerByKey(composeSortedKey(KEY, recordIndex)
          .getBytes(), null);

  try {
    byte[] vbuf = new byte[BUF_SIZE];
    int vlen = scanner.entry().getValueLength();
    scanner.entry().getValue(vbuf);
    Assert.assertEquals(new String(vbuf, 0, vlen), VALUE + recordIndex);

    byte[] kbuf = new byte[BUF_SIZE];
    int klen = scanner.entry().getKeyLength();
    scanner.entry().getKey(kbuf);
    Assert.assertEquals(new String(kbuf, 0, klen), composeSortedKey(KEY,
        recordIndex));
  } finally {
    scanner.close();
    reader.close();
  }
}
 

private void readValueWithoutKey(int recordIndex)
    throws IOException {
  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);

  Scanner scanner =
      reader.createScannerByKey(composeSortedKey(KEY, recordIndex)
          .getBytes(), null);

  byte[] vbuf1 = new byte[BUF_SIZE];
  int vlen1 = scanner.entry().getValueLength();
  scanner.entry().getValue(vbuf1);
  Assert.assertEquals(new String(vbuf1, 0, vlen1), VALUE + recordIndex);

  if (scanner.advance() && !scanner.atEnd()) {
    byte[] vbuf2 = new byte[BUF_SIZE];
    int vlen2 = scanner.entry().getValueLength();
    scanner.entry().getValue(vbuf2);
    Assert.assertEquals(new String(vbuf2, 0, vlen2), VALUE
        + (recordIndex + 1));
  }

  scanner.close();
  reader.close();
}
 

public void testFailureScannerWithKeys() throws IOException {
  Reader reader =
      new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Assert.assertFalse(reader.isSorted());
  Assert.assertEquals((int) reader.getEntryCount(), 4);

  try {
    Scanner scanner =
        reader.createScannerByKey("aaa".getBytes(), "zzz".getBytes());
    Assert
        .fail("Failed to catch creating scanner with keys on unsorted file.");
  }
  catch (RuntimeException e) {
  }
  finally {
    reader.close();
  }
}
 

private int readAndCheckbytes(Scanner scanner, int start, int n)
    throws IOException {
  String value = "value";
  for (int i = start; i < (start + n); i++) {
    byte[] key = readKey(scanner);
    byte[] val = readValue(scanner);
    String keyStr = String.format(localFormatter, i);
    String valStr = value + keyStr;
    assertTrue("btyes for keys do not match " + keyStr + " "
        + new String(key), Arrays.equals(keyStr.getBytes(), key));
    assertTrue("bytes for vals do not match " + valStr + " "
        + new String(val), Arrays.equals(
        valStr.getBytes(), val));
    assertTrue(scanner.advance());
    key = readKey(scanner);
    val = readValue(scanner);
    assertTrue("btyes for keys do not match", Arrays.equals(
        keyStr.getBytes(), key));
    assertTrue("bytes for vals do not match", Arrays.equals(
        valStr.getBytes(), val));
    assertTrue(scanner.advance());
  }
  return (start + n);
}
 

private int readPrepWithUnknownLength(Scanner scanner, int start, int n)
    throws IOException {
  for (int i = start; i < start; i++) {
    String key = String.format(localFormatter, i);
    byte[] read = readKey(scanner);
    assertTrue("keys not equal", Arrays.equals(key.getBytes(), read));
    try {
      read = readValue(scanner);
      assertTrue(false);
    }
    catch (IOException ie) {
      // should have thrown exception
    }
    String value = "value" + key;
    read = readLongValue(scanner, value.getBytes().length);
    assertTrue("values nto equal", Arrays.equals(read, value.getBytes()));
    scanner.advance();
  }
  return (start + n);
}
 

void unsortedWithSomeCodec(String codec) throws IOException {
  Path uTfile = new Path(ROOT, "unsorted.tfile");
  FSDataOutputStream fout = createFSOutput(uTfile);
  Writer writer = new Writer(fout, minBlockSize, codec, null, conf);
  writeRecords(writer);
  writer.close();
  fout.close();
  FSDataInputStream fin = fs.open(uTfile);
  Reader reader =
      new Reader(fs.open(uTfile), fs.getFileStatus(uTfile).getLen(), conf);

  Scanner scanner = reader.createScanner();
  readAllRecords(scanner);
  scanner.close();
  reader.close();
  fin.close();
  fs.delete(uTfile, true);
}
 

public void testFailureNegativeOffset() throws IOException {
  if (skip)
    return;
  writeRecords(2, true, true);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  byte[] buf = new byte[K];
  try {
    scanner.entry().getKey(buf, -1);
    Assert.fail("Failed to handle key negative offset.");
  }
  catch (Exception e) {
    // noop, expecting exceptions
  }
  finally {
  }
  scanner.close();
  reader.close();
}
 

void unsortedWithSomeCodec(String codec) throws IOException {
  Path uTfile = new Path(ROOT, "unsorted.tfile");
  FSDataOutputStream fout = createFSOutput(uTfile);
  Writer writer = new Writer(fout, minBlockSize, codec, null, conf);
  writeRecords(writer);
  writer.close();
  fout.close();
  FSDataInputStream fin = fs.open(uTfile);
  Reader reader =
      new Reader(fs.open(uTfile), fs.getFileStatus(uTfile).getLen(), conf);

  Scanner scanner = reader.createScanner();
  readAllRecords(scanner);
  scanner.close();
  reader.close();
  fin.close();
  fs.delete(uTfile, true);
}
 

@Test
public void testFailureReadValueManyTimes() throws IOException {
  if (skip)
    return;
  writeRecords(5);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();

  byte[] vbuf = new byte[BUF_SIZE];
  int vlen = scanner.entry().getValueLength();
  scanner.entry().getValue(vbuf);
  Assert.assertEquals(new String(vbuf, 0, vlen), VALUE + 0);
  try {
    scanner.entry().getValue(vbuf);
    Assert.fail("Cannot get the value mlutiple times.");
  } catch (Exception e) {
    // noop, expecting exceptions
  }

  scanner.close();
  reader.close();
}
 

@Test
public void testFailureNegativeOffset_2() throws IOException {
  if (skip)
    return;
  closeOutput();

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  try {
    scanner.lowerBound("keyX".getBytes(), -1, 4);
    Assert.fail("Error on handling negative offset.");
  } catch (Exception e) {
    // noop, expecting exceptions
  } finally {
    reader.close();
    scanner.close();
  }
  closeOutput();
}
 

@Test
public void testFailureNegativeLength_2() throws IOException {
  if (skip)
    return;
  closeOutput();

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  try {
    scanner.lowerBound("keyX".getBytes(), 0, -1);
    Assert.fail("Error on handling negative length.");
  } catch (Exception e) {
    // noop, expecting exceptions
  } finally {
    scanner.close();
    reader.close();
  }
  closeOutput();
}
 

private void readValueBeforeKey(int recordIndex)
    throws IOException {
  Reader reader =
      new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner =
      reader.createScannerByKey(composeSortedKey(KEY, recordIndex)
          .getBytes(), null);

  try {
    byte[] vbuf = new byte[BUF_SIZE];
    int vlen = scanner.entry().getValueLength();
    scanner.entry().getValue(vbuf);
    Assert.assertEquals(new String(vbuf, 0, vlen), VALUE + recordIndex);

    byte[] kbuf = new byte[BUF_SIZE];
    int klen = scanner.entry().getKeyLength();
    scanner.entry().getKey(kbuf);
    Assert.assertEquals(new String(kbuf, 0, klen), composeSortedKey(KEY,
        recordIndex));
  } finally {
    scanner.close();
    reader.close();
  }
}
 

private void readValueBeforeKey(int count, int recordIndex)
    throws IOException {
  Reader reader =
      new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner =
      reader.createScanner(composeSortedKey(KEY, count, recordIndex)
          .getBytes(), null);

  try {
    byte[] vbuf = new byte[BUF_SIZE];
    int vlen = scanner.entry().getValueLength();
    scanner.entry().getValue(vbuf);
    Assert.assertEquals(new String(vbuf, 0, vlen), VALUE + recordIndex);

    byte[] kbuf = new byte[BUF_SIZE];
    int klen = scanner.entry().getKeyLength();
    scanner.entry().getKey(kbuf);
    Assert.assertEquals(new String(kbuf, 0, klen), composeSortedKey(KEY,
        count, recordIndex));
  }
  finally {
    scanner.close();
    reader.close();
  }
}
 

private int readAndCheckbytes(Scanner scanner, int start, int n)
    throws IOException {
  String value = "value";
  for (int i = start; i < (start + n); i++) {
    byte[] key = readKey(scanner);
    byte[] val = readValue(scanner);
    String keyStr = String.format(localFormatter, i);
    String valStr = value + keyStr;
    assertTrue("btyes for keys do not match " + keyStr + " "
        + new String(key), Arrays.equals(keyStr.getBytes(), key));
    assertTrue("bytes for vals do not match " + valStr + " "
        + new String(val), Arrays.equals(
        valStr.getBytes(), val));
    assertTrue(scanner.advance());
    key = readKey(scanner);
    val = readValue(scanner);
    assertTrue("btyes for keys do not match", Arrays.equals(
        keyStr.getBytes(), key));
    assertTrue("bytes for vals do not match", Arrays.equals(
        valStr.getBytes(), val));
    assertTrue(scanner.advance());
  }
  return (start + n);
}
 

public void testFailureNegativeLength_2() throws IOException {
  if (skip)
    return;
  closeOutput();

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  try {
    scanner.lowerBound("keyX".getBytes(), 0, -1);
    Assert.fail("Error on handling negative length.");
  }
  catch (Exception e) {
    // noop, expecting exceptions
  }
  finally {
    scanner.close();
    reader.close();
  }
  closeOutput();
}
 

void unsortedWithSomeCodec(String codec) throws IOException {
  Path uTfile = new Path(ROOT, "unsorted.tfile");
  FSDataOutputStream fout = createFSOutput(uTfile);
  Writer writer = new Writer(fout, minBlockSize, codec, null, conf);
  writeRecords(writer);
  writer.close();
  fout.close();
  FSDataInputStream fin = fs.open(uTfile);
  Reader reader =
      new Reader(fs.open(uTfile), fs.getFileStatus(uTfile).getLen(), conf);

  Scanner scanner = reader.createScanner();
  readAllRecords(scanner);
  scanner.close();
  reader.close();
  fin.close();
  fs.delete(uTfile, true);
}
 

public void testFailureNegativeOffset() throws IOException {
  if (skip)
    return;
  writeRecords(2, true, true);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  byte[] buf = new byte[K];
  try {
    scanner.entry().getKey(buf, -1);
    Assert.fail("Failed to handle key negative offset.");
  }
  catch (Exception e) {
    // noop, expecting exceptions
  }
  finally {
  }
  scanner.close();
  reader.close();
}
 

public void testLocate() throws IOException {
  if (skip)
    return;
  writeRecords(3 * records1stBlock);
  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  Location loc2 =
      locate(scanner, composeSortedKey(KEY, 3 * records1stBlock, 2)
          .getBytes());
  Location locLastIn1stBlock =
      locate(scanner, composeSortedKey(KEY, 3 * records1stBlock,
          records1stBlock - 1).getBytes());
  Location locFirstIn2ndBlock =
      locate(scanner, composeSortedKey(KEY, 3 * records1stBlock,
          records1stBlock).getBytes());
  Location locX = locate(scanner, "keyX".getBytes());
  Assert.assertEquals(scanner.endLocation, locX);
  scanner.close();
  reader.close();
}
 

@Test
public void testFailureNegativeOffset_2() throws IOException {
  if (skip)
    return;
  closeOutput();

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();
  try {
    scanner.lowerBound("keyX".getBytes(), -1, 4);
    Assert.fail("Error on handling negative offset.");
  } catch (Exception e) {
    // noop, expecting exceptions
  } finally {
    reader.close();
    scanner.close();
  }
  closeOutput();
}
 

public void testFailureReadValueManyTimes() throws IOException {
  if (skip)
    return;
  writeRecords(5);

  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
  Scanner scanner = reader.createScanner();

  byte[] vbuf = new byte[BUF_SIZE];
  int vlen = scanner.entry().getValueLength();
  scanner.entry().getValue(vbuf);
  Assert.assertEquals(new String(vbuf, 0, vlen), VALUE + 0);
  try {
    scanner.entry().getValue(vbuf);
    Assert.fail("Cannot get the value mlutiple times.");
  }
  catch (Exception e) {
    // noop, expecting exceptions
  }

  scanner.close();
  reader.close();
}
 

private void readValueWithoutKey(int recordIndex)
    throws IOException {
  Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);

  Scanner scanner =
      reader.createScannerByKey(composeSortedKey(KEY, recordIndex)
          .getBytes(), null);

  byte[] vbuf1 = new byte[BUF_SIZE];
  int vlen1 = scanner.entry().getValueLength();
  scanner.entry().getValue(vbuf1);
  Assert.assertEquals(new String(vbuf1, 0, vlen1), VALUE + recordIndex);

  if (scanner.advance() && !scanner.atEnd()) {
    byte[] vbuf2 = new byte[BUF_SIZE];
    int vlen2 = scanner.entry().getValueLength();
    scanner.entry().getValue(vbuf2);
    Assert.assertEquals(new String(vbuf2, 0, vlen2), VALUE
        + (recordIndex + 1));
  }

  scanner.close();
  reader.close();
}
 

public byte[] readLongValue(Scanner scanner, int len) throws IOException {
  DataInputStream din = scanner.entry().getValueStream();
  byte[] b = new byte[len];
  din.readFully(b);
  din.close();
  return b;
}