/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with this
 * work for additional information regarding copyright ownership. The ASF
 * licenses this file to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.hadoop.io.file.tfile;

import java.io.IOException;
import java.util.Random;

import org.junit.Test;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.file.tfile.TFile.Reader;
import org.apache.hadoop.io.file.tfile.TFile.Writer;
import org.apache.hadoop.io.file.tfile.TFile.Reader.Scanner;
import org.apache.hadoop.test.GenericTestUtils;

public class TestTFileSplit {
  private static String ROOT = GenericTestUtils.getTestDir().getAbsolutePath();

  private final static int BLOCK_SIZE = 64 * 1024;

  private static final String KEY = "key";
  private static final String VALUE = "value";

  private FileSystem fs;
  private Configuration conf;
  private Path path;
  private Random random = new Random();

  private String comparator = "memcmp";
  private String outputFile = "TestTFileSplit";

  void createFile(int count, String compress) throws IOException {
    conf = new Configuration();
    path = new Path(ROOT, outputFile + "." + compress);
    fs = path.getFileSystem(conf);
    FSDataOutputStream out = fs.create(path);
    Writer writer = new Writer(out, BLOCK_SIZE, compress, comparator, conf);

    int nx;
    for (nx = 0; nx < count; nx++) {
      byte[] key = composeSortedKey(KEY, count, nx).getBytes();
      byte[] value = (VALUE + nx).getBytes();
      writer.append(key, value);
    }
    writer.close();
    out.close();
  }

  void readFile() throws IOException {
    long fileLength = fs.getFileStatus(path).getLen();
    int numSplit = 10;
    long splitSize = fileLength / numSplit + 1;

    Reader reader =
        new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
    long offset = 0;
    long rowCount = 0;
    BytesWritable key, value;
    for (int i = 0; i < numSplit; ++i, offset += splitSize) {
      Scanner scanner = reader.createScannerByByteRange(offset, splitSize);
      int count = 0;
      key = new BytesWritable();
      value = new BytesWritable();
      while (!scanner.atEnd()) {
        scanner.entry().get(key, value);
        ++count;
        scanner.advance();
      }
      scanner.close();
      assertTrue(count > 0);
      rowCount += count;
    }
    assertEquals(rowCount, reader.getEntryCount());
    reader.close();
  }

  /* Similar to readFile(), tests the scanner created 
   * by record numbers rather than the offsets.
   */
  void readRowSplits(int numSplits) throws IOException {

    Reader reader =
      new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf);
    
    long totalRecords = reader.getEntryCount();
    for (int i=0; i<numSplits; i++) {
      long startRec = i*totalRecords/numSplits;
      long endRec = (i+1)*totalRecords/numSplits;
      if (i == numSplits-1) {
        endRec = totalRecords;
      }
      Scanner scanner = reader.createScannerByRecordNum(startRec, endRec);
      int count = 0;
      BytesWritable key = new BytesWritable();
      BytesWritable value = new BytesWritable();
      long x=startRec;
      while (!scanner.atEnd()) {
        assertEquals("Incorrect RecNum returned by scanner", scanner.getRecordNum(), x);
        scanner.entry().get(key, value);
        ++count;
        assertEquals("Incorrect RecNum returned by scanner", scanner.getRecordNum(), x);
        scanner.advance();
        ++x;
      }
      scanner.close();
      assertTrue(count == (endRec - startRec));
    }
    // make sure specifying range at the end gives zero records.
    Scanner scanner = reader.createScannerByRecordNum(totalRecords, -1);
    assertTrue(scanner.atEnd());
  }
  
  static String composeSortedKey(String prefix, int total, int value) {
    return String.format("%s%010d", prefix, value);
  }
  
  void checkRecNums() throws IOException {
    long fileLen = fs.getFileStatus(path).getLen();
    Reader reader = new Reader(fs.open(path), fileLen, conf);
    long totalRecs = reader.getEntryCount();
    long begin = random.nextLong() % (totalRecs / 2);
    if (begin < 0)
      begin += (totalRecs / 2);
    long end = random.nextLong() % (totalRecs / 2);
    if (end < 0)
      end += (totalRecs / 2);
    end += (totalRecs / 2) + 1;

    assertEquals("RecNum for offset=0 should be 0", 0, reader
        .getRecordNumNear(0));
    for (long x : new long[] { fileLen, fileLen + 1, 2 * fileLen }) {
      assertEquals("RecNum for offset>=fileLen should be total entries",
          totalRecs, reader.getRecordNumNear(x));
    }

    for (long i = 0; i < 100; ++i) {
      assertEquals("Locaton to RecNum conversion not symmetric", i, reader
          .getRecordNumByLocation(reader.getLocationByRecordNum(i)));
    }

    for (long i = 1; i < 100; ++i) {
      long x = totalRecs - i;
      assertEquals("Locaton to RecNum conversion not symmetric", x, reader
          .getRecordNumByLocation(reader.getLocationByRecordNum(x)));
    }

    for (long i = begin; i < end; ++i) {
      assertEquals("Locaton to RecNum conversion not symmetric", i, reader
          .getRecordNumByLocation(reader.getLocationByRecordNum(i)));
    }

    for (int i = 0; i < 1000; ++i) {
      long x = random.nextLong() % totalRecs;
      if (x < 0) x += totalRecs;
      assertEquals("Locaton to RecNum conversion not symmetric", x, reader
          .getRecordNumByLocation(reader.getLocationByRecordNum(x)));
    }
  }

  @Test
  public void testSplit() throws IOException {
    System.out.println("testSplit");
    createFile(100000, Compression.Algorithm.NONE.getName());
    checkRecNums();   
    readFile();
    readRowSplits(10);
    fs.delete(path, true);
    createFile(500000, Compression.Algorithm.GZ.getName());
    checkRecNums();
    readFile();
    readRowSplits(83);
    fs.delete(path, true);
  }
}