/**
 * 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.hcatalog.data;

import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hcatalog.common.HCatException;
import org.apache.hcatalog.data.schema.HCatSchema;
import org.apache.hcatalog.data.schema.HCatSchemaUtils;

import junit.framework.Assert;
import junit.framework.TestCase;

public class TestDefaultHCatRecord extends TestCase{

  public void testRYW() throws IOException{

    File f = new File("binary.dat");
    f.delete();
    f.createNewFile();
    f.deleteOnExit();

    OutputStream fileOutStream = new FileOutputStream(f);
    DataOutput outStream = new DataOutputStream(fileOutStream);

    HCatRecord[]  recs = getHCatRecords();
    for(int i =0; i < recs.length; i++){
      recs[i].write(outStream);
    }
    fileOutStream.flush();
    fileOutStream.close();

    InputStream fInStream = new FileInputStream(f);
    DataInput inpStream = new DataInputStream(fInStream);

    for(int i =0; i < recs.length; i++){
      HCatRecord rec = new DefaultHCatRecord();
      rec.readFields(inpStream);
      Assert.assertTrue(HCatDataCheckUtil.recordsEqual(recs[i],rec));
    }

    Assert.assertEquals(fInStream.available(), 0);
    fInStream.close();

  }

  public void testCompareTo() {
    HCatRecord[] recs = getHCatRecords();
    Assert.assertTrue(HCatDataCheckUtil.compareRecords(recs[0],recs[1]) == 0);
    Assert.assertTrue(HCatDataCheckUtil.compareRecords(recs[4],recs[5]) == 0);
  }

  public void testEqualsObject() {

    HCatRecord[] recs = getHCatRecords();
    Assert.assertTrue(HCatDataCheckUtil.recordsEqual(recs[0],recs[1]));
    Assert.assertTrue(HCatDataCheckUtil.recordsEqual(recs[4],recs[5]));
  }

  /**
   * Test get and set calls with type
   * @throws HCatException
   */
  public void testGetSetByType1() throws HCatException{
    HCatRecord inpRec = getHCatRecords()[0];
    HCatRecord newRec = new DefaultHCatRecord(inpRec.size());
    HCatSchema hsch = 
        HCatSchemaUtils.getHCatSchema(
            "a:tinyint,b:smallint,c:int,d:bigint,e:float,f:double,g:boolean,h:string,i:binary,j:string");
    

    newRec.setByte("a", hsch, inpRec.getByte("a", hsch) );
    newRec.setShort("b", hsch, inpRec.getShort("b", hsch) );
    newRec.setInteger("c", hsch, inpRec.getInteger("c", hsch) );
    newRec.setLong("d", hsch, inpRec.getLong("d", hsch) );
    newRec.setFloat("e", hsch, inpRec.getFloat("e", hsch) );
    newRec.setDouble("f", hsch, inpRec.getDouble("f", hsch) );
    newRec.setBoolean("g", hsch, inpRec.getBoolean("g", hsch) );
    newRec.setString("h", hsch, inpRec.getString("h", hsch) );
    newRec.setByteArray("i", hsch, inpRec.getByteArray("i", hsch) );
    newRec.setString("j", hsch, inpRec.getString("j", hsch) );
    
    Assert.assertTrue(HCatDataCheckUtil.recordsEqual(newRec,inpRec));
    
    
  }
  
  /**
   * Test get and set calls with type
   * @throws HCatException
   */
  public void testGetSetByType2() throws HCatException{
    HCatRecord inpRec = getGetSet2InpRec();
    
    HCatRecord newRec = new DefaultHCatRecord(inpRec.size());
    HCatSchema hsch = 
        HCatSchemaUtils.getHCatSchema("a:binary,b:map<string,string>,c:array<int>,d:struct<i:int>");
    

    newRec.setByteArray("a", hsch, inpRec.getByteArray("a", hsch) );
    newRec.setMap("b", hsch, inpRec.getMap("b", hsch) );
    newRec.setList("c", hsch, inpRec.getList("c", hsch) );
    newRec.setStruct("d", hsch, inpRec.getStruct("d", hsch) );

    Assert.assertTrue(HCatDataCheckUtil.recordsEqual(newRec,inpRec));
  }
  
  
  private HCatRecord getGetSet2InpRec() {
    List<Object> rlist = new ArrayList<Object>();
    
    rlist.add(new byte[]{1,2,3});
    
    Map<Short, String> mapcol = new HashMap<Short, String>(3);
    mapcol.put(new Short("2"), "hcat is cool");
    mapcol.put(new Short("3"), "is it?");
    mapcol.put(new Short("4"), "or is it not?");
    rlist.add(mapcol);

    List<Integer> listcol = new ArrayList<Integer>();
    listcol.add(314);
    listcol.add(007);
    rlist.add( listcol);//list
    rlist.add( listcol);//struct
    return new DefaultHCatRecord(rlist);
  }

  private HCatRecord[] getHCatRecords(){

    List<Object> rec_1 = new ArrayList<Object>(8);
    rec_1.add(new Byte("123"));
    rec_1.add(new Short("456"));
    rec_1.add( new Integer(789));
    rec_1.add( new Long(1000L));
    rec_1.add( new Float(5.3F));
    rec_1.add( new Double(5.3D));
    rec_1.add( new Boolean(true));
    rec_1.add( new String("hcat and hadoop"));
    rec_1.add( null);
    rec_1.add( "null");

    HCatRecord tup_1 = new DefaultHCatRecord(rec_1);

    List<Object> rec_2 = new ArrayList<Object>(8);
    rec_2.add( new Byte("123"));
    rec_2.add( new Short("456"));
    rec_2.add( new Integer(789));
    rec_2.add( new Long(1000L));
    rec_2.add( new Float(5.3F));
    rec_2.add( new Double(5.3D));
    rec_2.add( new Boolean(true));
    rec_2.add( new String("hcat and hadoop"));
    rec_2.add( null);
    rec_2.add( "null");
    HCatRecord tup_2 = new DefaultHCatRecord(rec_2);

    List<Object> rec_3 = new ArrayList<Object>(10);
    rec_3.add(new Byte("123"));
    rec_3.add(new Short("456"));
    rec_3.add( new Integer(789));
    rec_3.add( new Long(1000L));
    rec_3.add( new Double(5.3D));
    rec_3.add( new String("hcat and hadoop"));
    rec_3.add( null);
    List<Integer> innerList = new ArrayList<Integer>();
    innerList.add(314);
    innerList.add(007);
    rec_3.add( innerList);
    Map<Short, String> map = new HashMap<Short, String>(3);
    map.put(new Short("2"), "hcat is cool");
    map.put(new Short("3"), "is it?");
    map.put(new Short("4"), "or is it not?");
    rec_3.add(map);

    HCatRecord tup_3 = new DefaultHCatRecord(rec_3);

    List<Object> rec_4 = new ArrayList<Object>(8);
    rec_4.add( new Byte("123"));
    rec_4.add( new Short("456"));
    rec_4.add( new Integer(789));
    rec_4.add( new Long(1000L));
    rec_4.add( new Double(5.3D));
    rec_4.add( new String("hcat and hadoop"));
    rec_4.add( null);
    rec_4.add( "null");

    Map<Short, String> map2 = new HashMap<Short, String>(3);
    map2.put(new Short("2"), "hcat is cool");
    map2.put(new Short("3"), "is it?");
    map2.put(new Short("4"), "or is it not?");
    rec_4.add(map2);
    List<Integer> innerList2 = new ArrayList<Integer>();
    innerList2.add(314);
    innerList2.add(007);
    rec_4.add( innerList2);
    HCatRecord tup_4 = new DefaultHCatRecord(rec_4);

    
    List<Object> rec_5 = new ArrayList<Object>(3);
    rec_5.add( getByteArray());
    rec_5.add( getStruct());
    rec_5.add( getList());
    HCatRecord tup_5 = new DefaultHCatRecord(rec_5);
    

    List<Object> rec_6 = new ArrayList<Object>(3);
    rec_6.add( getByteArray());
    rec_6.add( getStruct());
    rec_6.add( getList());
    HCatRecord tup_6 = new DefaultHCatRecord(rec_6);

    
    return  new HCatRecord[]{tup_1,tup_2,tup_3,tup_4,tup_5,tup_6};

  }

  private Object getList() {
    return getStruct();
  }

  private Object getByteArray() {
    return new byte[]{1,2,3,4};
  }

  private List<?> getStruct() {
    List<Object> struct = new ArrayList<Object>();
    struct.add(new Integer(1));
    struct.add(new String("x"));
    return struct;
  }
}