<commit_before>/**
 * (c) Copyright 2012 WibiData, Inc.
 *
 * See the NOTICE file distributed with this work for additional
 * information regarding copyright ownership.
 *
 * Licensed 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.kiji.mapreduce.output;

import static org.apache.hadoop.hbase.util.Bytes.toBytes;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.kiji.mapreduce.HFileKeyValue;
import org.kiji.mapreduce.KijiConfKeys;
import org.kiji.schema.Kiji;
import org.kiji.schema.KijiAdmin;
import org.kiji.schema.KijiInstaller;
import org.kiji.schema.KijiURI;
import org.kiji.schema.layout.KijiTableLayout;
import org.kiji.schema.layout.KijiTableLayouts;
import org.kiji.schema.layout.impl.ColumnId;

/** Tests for KijiHFileOutputFormat. */
public class TestKijiHFileOutputFormat {
  private static final Logger LOG = LoggerFactory.getLogger(TestKijiHFileOutputFormat.class);

  /** Counter for fake instance IDs. */
  private static final AtomicLong FAKE_INSTANCE_COUNTER = new AtomicLong(0);

  /** NullWritable shortcut. */
  private static final NullWritable NW = NullWritable.get();

  /**
   * Makes a dummy byte array.
   *
   * @param value Byte value to repeat.
   * @param nbytes Number of bytes.
   * @return a byte array with the specified number of bytes and the specified byte value.
   */
  private static byte[] makeBytes(int value, int nbytes) {
    final byte[] bytes = new byte[nbytes];
    for (int i = 0; i < nbytes; ++i) {
      bytes[i] = (byte) value;
    }
    return bytes;
  }

  /**
   * Makes an HFile entry (KeyValue writable-comparable).
   *
   * @param row Row key.
   * @param family HBase family (as a Kiji locality group column ID).
   * @param qualifier HBase qualifier.
   * @param timestamp Cell timestamp.
   * @param value Cell content bytes.
   * @return a new HFileKeyValue with the specified parameters.
   */
  private static HFileKeyValue entry(
      String row, ColumnId family, String qualifier, long timestamp, byte[] value) {
    return new HFileKeyValue(
        toBytes(row), family.toByteArray(), toBytes(qualifier), timestamp, value);
  }

  /**
   * Loads an HFile content into a list of KeyValue entries.
   *
   * @param path Path of the HFile to load.
   * @param conf Configuration.
   * @return the content of the specified HFile, as an ordered list of KeyValue entries.
   * @throws IOException on I/O error.
   */
  private static List<KeyValue> loadHFile(Path path, Configuration conf) throws IOException {
    final FileSystem fs = path.getFileSystem(conf);
    final CacheConfig cacheConf = new CacheConfig(conf);
    final HFile.Reader reader = HFile.createReader(fs, path, cacheConf);
    final HFileScanner scanner = reader.getScanner(false, false);
    final List<KeyValue> kvs = Lists.newArrayListWithCapacity((int) reader.getEntries());
    boolean hasNext = scanner.seekTo();
    while (hasNext) {
      kvs.add(scanner.getKeyValue());
      hasNext = scanner.next();
    }
    reader.close();
    return kvs;
  }

  /**
   * Asserts the content of an HFile.
   *
   * @param path Path of the HFile to validate the content of.
   * @param values Expected KeyValue entries, in order.
   * @throws IOException on I/O error.
   */
  private void assertHFileContent(Path path, KeyValue... values) throws IOException {
    final FileSystem fs = path.getFileSystem(mConf);
    assertTrue(String.format("HFile '%s' does not exist.", path), fs.exists(path));
    final List<KeyValue> kvs = loadHFile(path, mConf);
    assertEquals(kvs.size(), values.length);
    for (int i = 0; i < values.length; ++i) {
      assertEquals(kvs.get(i), values[i]);
    }
  }

  private Configuration mConf;
  private KijiURI mTableURI;
  private Kiji mKiji;
  private File mTempDir;
  private KijiTableLayout mLayout;
  private KijiHFileOutputFormat mFormat;

  private ColumnId mDefaultLGId;
  private ColumnId mInMemoryLGId;

  @Before
  public void setUp() throws Exception {
    mConf = HBaseConfiguration.create();
    mTableURI = KijiURI.newBuilder(String.format(
        "kiji://.fake.%s/default/user", FAKE_INSTANCE_COUNTER.getAndIncrement())).build();

    mTempDir = File.createTempFile("test-" + System.currentTimeMillis() + "-", "");
    Preconditions.checkState(mTempDir.delete());
    Preconditions.checkState(mTempDir.mkdir());

    mConf.set("fs.defaultFS", "file://" + mTempDir.toString());
    mConf.set("mapred.output.dir", "file://" + mTempDir.toString());

    KijiInstaller.install(mTableURI, mConf);
    mKiji = Kiji.Factory.open(mTableURI);

    final KijiAdmin admin = mKiji.getAdmin();
    mLayout = new KijiTableLayout(KijiTableLayouts.getLayout(KijiTableLayouts.FULL_FEATURED), null);
    admin.createTable("user", mLayout, false);

    mDefaultLGId = mLayout.getLocalityGroupMap().get("default").getId();
    mInMemoryLGId = mLayout.getLocalityGroupMap().get("inMemory").getId();

    mConf.set(KijiConfKeys.KIJI_OUTPUT_TABLE_URI, mTableURI.toString());

    mFormat = new KijiHFileOutputFormat();
  }

  @After
  public void tearDown() throws Exception {
    FileUtils.deleteDirectory(mTempDir);
  }

  @Test
  public void testMaxHFileSizeSameRow() throws Exception {
    final HFileKeyValue entry1 = entry("row-key", mDefaultLGId, "a", 1L, makeBytes(0, 1024));<commit_msg>SCHEMA-109. Update to reflect API changes to KijiInstaller.<commit_after>/**
 * (c) Copyright 2012 WibiData, Inc.
 *
 * See the NOTICE file distributed with this work for additional
 * information regarding copyright ownership.
 *
 * Licensed 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.kiji.mapreduce.output;

import static org.apache.hadoop.hbase.util.Bytes.toBytes;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.kiji.mapreduce.HFileKeyValue;
import org.kiji.mapreduce.KijiConfKeys;
import org.kiji.schema.Kiji;
import org.kiji.schema.KijiAdmin;
import org.kiji.schema.KijiInstaller;
import org.kiji.schema.KijiURI;
import org.kiji.schema.layout.KijiTableLayout;
import org.kiji.schema.layout.KijiTableLayouts;
import org.kiji.schema.layout.impl.ColumnId;

/** Tests for KijiHFileOutputFormat. */
public class TestKijiHFileOutputFormat {
  private static final Logger LOG = LoggerFactory.getLogger(TestKijiHFileOutputFormat.class);

  /** Counter for fake instance IDs. */
  private static final AtomicLong FAKE_INSTANCE_COUNTER = new AtomicLong(0);

  /** NullWritable shortcut. */
  private static final NullWritable NW = NullWritable.get();

  /**
   * Makes a dummy byte array.
   *
   * @param value Byte value to repeat.
   * @param nbytes Number of bytes.
   * @return a byte array with the specified number of bytes and the specified byte value.
   */
  private static byte[] makeBytes(int value, int nbytes) {
    final byte[] bytes = new byte[nbytes];
    for (int i = 0; i < nbytes; ++i) {
      bytes[i] = (byte) value;
    }
    return bytes;
  }

  /**
   * Makes an HFile entry (KeyValue writable-comparable).
   *
   * @param row Row key.
   * @param family HBase family (as a Kiji locality group column ID).
   * @param qualifier HBase qualifier.
   * @param timestamp Cell timestamp.
   * @param value Cell content bytes.
   * @return a new HFileKeyValue with the specified parameters.
   */
  private static HFileKeyValue entry(
      String row, ColumnId family, String qualifier, long timestamp, byte[] value) {
    return new HFileKeyValue(
        toBytes(row), family.toByteArray(), toBytes(qualifier), timestamp, value);
  }

  /**
   * Loads an HFile content into a list of KeyValue entries.
   *
   * @param path Path of the HFile to load.
   * @param conf Configuration.
   * @return the content of the specified HFile, as an ordered list of KeyValue entries.
   * @throws IOException on I/O error.
   */
  private static List<KeyValue> loadHFile(Path path, Configuration conf) throws IOException {
    final FileSystem fs = path.getFileSystem(conf);
    final CacheConfig cacheConf = new CacheConfig(conf);
    final HFile.Reader reader = HFile.createReader(fs, path, cacheConf);
    final HFileScanner scanner = reader.getScanner(false, false);
    final List<KeyValue> kvs = Lists.newArrayListWithCapacity((int) reader.getEntries());
    boolean hasNext = scanner.seekTo();
    while (hasNext) {
      kvs.add(scanner.getKeyValue());
      hasNext = scanner.next();
    }
    reader.close();
    return kvs;
  }

  /**
   * Asserts the content of an HFile.
   *
   * @param path Path of the HFile to validate the content of.
   * @param values Expected KeyValue entries, in order.
   * @throws IOException on I/O error.
   */
  private void assertHFileContent(Path path, KeyValue... values) throws IOException {
    final FileSystem fs = path.getFileSystem(mConf);
    assertTrue(String.format("HFile '%s' does not exist.", path), fs.exists(path));
    final List<KeyValue> kvs = loadHFile(path, mConf);
    assertEquals(kvs.size(), values.length);
    for (int i = 0; i < values.length; ++i) {
      assertEquals(kvs.get(i), values[i]);
    }
  }

  private Configuration mConf;
  private KijiURI mTableURI;
  private Kiji mKiji;
  private File mTempDir;
  private KijiTableLayout mLayout;
  private KijiHFileOutputFormat mFormat;

  private ColumnId mDefaultLGId;
  private ColumnId mInMemoryLGId;

  @Before
  public void setUp() throws Exception {
    mConf = HBaseConfiguration.create();
    mTableURI = KijiURI.newBuilder(String.format(
        "kiji://.fake.%s/default/user", FAKE_INSTANCE_COUNTER.getAndIncrement())).build();

    mTempDir = File.createTempFile("test-" + System.currentTimeMillis() + "-", "");
    Preconditions.checkState(mTempDir.delete());
    Preconditions.checkState(mTempDir.mkdir());

    mConf.set("fs.defaultFS", "file://" + mTempDir.toString());
    mConf.set("mapred.output.dir", "file://" + mTempDir.toString());

    KijiInstaller.get().install(mTableURI, mConf);
    mKiji = Kiji.Factory.open(mTableURI);

    final KijiAdmin admin = mKiji.getAdmin();
    mLayout = new KijiTableLayout(KijiTableLayouts.getLayout(KijiTableLayouts.FULL_FEATURED), null);
    admin.createTable("user", mLayout, false);

    mDefaultLGId = mLayout.getLocalityGroupMap().get("default").getId();
    mInMemoryLGId = mLayout.getLocalityGroupMap().get("inMemory").getId();

    mConf.set(KijiConfKeys.KIJI_OUTPUT_TABLE_URI, mTableURI.toString());

    mFormat = new KijiHFileOutputFormat();
  }

  @After
  public void tearDown() throws Exception {
    FileUtils.deleteDirectory(mTempDir);
  }

  @Test
  public void testMaxHFileSizeSameRow() throws Exception {
    final HFileKeyValue entry1 = entry("row-key", mDefaultLGId, "a", 1L, makeBytes(0, 1024));
{'fraction_non_alphanumeric': 0.10989167616875713, 'fraction_numerical': 0.00306442417331813, 'mean_word_length': 4.114067055393586, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4048816', 'n_tokens_mistral': 4372, 'n_tokens_neox': 4279, 'n_words': 1166}