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

import static org.apache.pig.builtin.mock.Storage.resetData;
import static org.apache.pig.builtin.mock.Storage.tuple;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

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.math.BigDecimal;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.pig.ExecType;
import org.apache.pig.PigConfiguration;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
import org.apache.pig.builtin.mock.Storage.Data;
import org.apache.pig.data.SchemaTupleClassGenerator.GenContext;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.InterRecordReader;
import org.apache.pig.impl.io.InterRecordWriter;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
import org.apache.pig.impl.util.PropertiesUtil;
import org.apache.pig.impl.util.Utils;
import org.joda.time.DateTime;
import org.junit.Before;
import org.junit.Test;

import com.google.common.collect.Maps;

public class TestSchemaTuple {
    private Properties props;
    private Configuration conf;
    private PigContext pigContext;
    private static final BinInterSedes bis = new BinInterSedes();

    @Before
    public void perTestInitialize() {
        props = new Properties();
        props.setProperty(PigConfiguration.PIG_SCHEMA_TUPLE_ENABLED, "true");

        conf = ConfigurationUtil.toConfiguration(props);

        pigContext = new PigContext(ExecType.LOCAL, props);
    }

    @Test
    public void testCompileAndResolve() throws Exception {
        //frontend
        Schema udfSchema = Utils.getSchemaFromString("a:int");
        boolean isAppendable = false;
        GenContext context = GenContext.UDF;
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString("a:long");
        isAppendable = true;
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString("a:chararray,(a:chararray)");
        isAppendable = false;
        context = GenContext.FOREACH;
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString("a:int,(a:int,(a:int,(a:int,(a:int,(a:int,(a:int))))))");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString("((a:int,b:int),(a:int,b:int),(a:int,b:int)),((a:int,b:int),(a:int,b:int),(a:int,b:int))");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString(
                "a:int, b:long, c:chararray, d:boolean, e:bytearray, f:float, g:double, dt:datetime, bd: bigdecimal, bi: biginteger,"
                +"(a:int, b:long, c:chararray, d:boolean, e:bytearray, f:float, g:double, dt:datetime, bd: bigdecimal, bi: biginteger,"
                +"(a:int, b:long, c:chararray, d:boolean, e:bytearray, f:float, g:double, dt:datetime, bd: bigdecimal, bi: biginteger))");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString("boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        udfSchema = Utils.getSchemaFromString(
                "int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger))");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        isAppendable = true;
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        isAppendable = false;
        udfSchema = Utils.getSchemaFromString("int, b:bag{(int,int,int)}");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        isAppendable = false;
        udfSchema = Utils.getSchemaFromString("int, m:map[(int,int,int)]");
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        isAppendable = false;
        udfSchema = new Schema();
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        isAppendable = false;
        udfSchema = new Schema(new FieldSchema(null, DataType.BAG));
        SchemaTupleFrontend.registerToGenerateIfPossible(udfSchema, isAppendable, context);

        // this compiles and "ships"
        SchemaTupleFrontend.copyAllGeneratedToDistributedCache(pigContext, conf);

        //backend
        SchemaTupleBackend.initialize(conf, pigContext);

        udfSchema = Utils.getSchemaFromString("a:int");
        isAppendable = false;
        context = GenContext.UDF;
        SchemaTupleFactory tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        context = GenContext.MERGE_JOIN;
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        assertNull(tf);

        udfSchema = Utils.getSchemaFromString("a:long");
        context = GenContext.UDF;
        isAppendable = true;

        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        isAppendable = false;
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        assertNull(tf);

        udfSchema = Utils.getSchemaFromString("a:chararray,(a:chararray)");
        isAppendable = false;
        context = GenContext.FOREACH;
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        udfSchema = Utils.getSchemaFromString("(a:chararray)");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        assertNull(tf);

        udfSchema = Utils.getSchemaFromString("a:int,(a:int,(a:int,(a:int,(a:int,(a:int,(a:int))))))");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        udfSchema = Utils.getSchemaFromString("((a:int,b:int),(a:int,b:int),(a:int,b:int)),((a:int,b:int),(a:int,b:int),(a:int,b:int))");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        udfSchema = Utils.getSchemaFromString(
                "a:int, b:long, c:chararray, d:boolean, e:bytearray, f:float, g:double, dt: datetime, bd: bigdecimal, bi: biginteger,"
                +"(a:int, b:long, c:chararray, d:boolean, e:bytearray, f:float, g:double, dt: datetime, bd: bigdecimal, bi: biginteger,"
                +"(a:int, b:long, c:chararray, d:boolean, e:bytearray, f:float, g:double, dt: datetime, bd: bigdecimal, bi: biginteger))");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        udfSchema = Utils.getSchemaFromString("boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean,"
                + "boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean, boolean");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        udfSchema = Utils.getSchemaFromString(
                "int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger)),"
                +"int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger,"
                +"(int, long, chararray, boolean, bytearray, float, double, datetime, bigdecimal, biginteger))");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        isAppendable = true;
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        isAppendable = false;
        udfSchema = Utils.getSchemaFromString("int, b:bag{(int,int,int)}");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        isAppendable = false;
        udfSchema = Utils.getSchemaFromString("int, m:map[(int,int,int)]");
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);

        isAppendable = false;
        udfSchema = new Schema();
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        assertNull(tf);

        isAppendable = false;
        udfSchema = new Schema(new FieldSchema(null, DataType.BAG));
        tf = SchemaTupleFactory.getInstance(udfSchema, isAppendable, context);
        putThroughPaces(tf, udfSchema, isAppendable);
    }

    private void putThroughPaces(SchemaTupleFactory tf, Schema udfSchema, boolean isAppendable) throws Exception {
        assertNotNull(tf);
        if (isAppendable) {
            assertTrue(tf.newTuple() instanceof AppendableSchemaTuple);
        }

        testNotAppendable(tf, udfSchema);
        if (isAppendable) {
            testAppendable(tf, udfSchema);
        }
    }

    private void testAppendable(SchemaTupleFactory tf, Schema udfSchema) {
        SchemaTuple<?> st = tf.newTuple();

        st.append("woah");
        assertEquals(udfSchema.size() + 1, st.size());

    }

    private void testNotAppendable(SchemaTupleFactory tf, Schema udfSchema) throws Exception {
        SchemaTuple<?> st = tf.newTuple();
        Schema.equals(udfSchema, st.getSchema(), false, true);

        assertEquals(udfSchema.size(), st.size());

        testHashCodeEqualAsNormalTuple(tf);

        shouldAllBeNull(tf);

        copyThenCompare(tf);

        testSerDe(tf);
        testInterStorageSerDe(tf);
    }

    private void copyThenCompare(SchemaTupleFactory tf) throws ExecException {
        SchemaTuple<?> st = tf.newTuple();
        SchemaTuple<?> st2 = tf.newTuple();
        fillWithData(st);
        st2.set(st);
        assertTrue(st.equals(st2));
        assertEquals(st.compareTo(st2), 0);
        st.set(0, null);
        assertFalse(st.equals(st2));
        assertEquals(st.compareTo(st2), -1);
        assertTrue(st.isNull(0));
        st2.set(0, null);
        assertTrue(st.equals(st2));
        assertEquals(st.compareTo(st2), 0);
    }

    /**
     * This ensures that a fresh Tuple out of a TupleFactory
     * will be full of null fields.
     * @param   tf a TupleFactory
     * @throws  ExecException
     */
    private void shouldAllBeNull(SchemaTupleFactory tf) throws ExecException {
        Tuple t = tf.newTuple();
        for (Object o : t) {
            assertNull(o);
        }
        for (int i = 0; i < t.size(); i++) {
            assertNull(t.get(i));
            assertTrue(t.isNull(i));
        }
    }

    private void fillWithData(SchemaTuple<?> st) throws ExecException {
        Schema udfSchema = st.getSchema();
        int pos = 0;
        for (FieldSchema fs : udfSchema.getFields()) {
            Object val;
            if (fs.type == DataType.TUPLE) {
                val = SchemaTupleFactory
                            .getInstance(fs.schema, false, GenContext.FORCE_LOAD)
                            .newTuple();
                fillWithData((SchemaTuple<?>)val);
            } else {
                val = randData(fs);
            }
            st.set(pos++, val);
        }
    }

    private Random r = new Random(100L);

    private Object randData(FieldSchema fs) throws ExecException {
        switch (fs.type) {
        case DataType.BIGDECIMAL: return new BigDecimal(r.nextDouble());
        case DataType.BIGINTEGER: return new BigInteger(130, r);
        case DataType.BOOLEAN: return r.nextBoolean();
        case DataType.BYTEARRAY: return new DataByteArray(new BigInteger(130, r).toByteArray());
        case DataType.CHARARRAY: return new BigInteger(130, r).toString(32);
        case DataType.INTEGER: return r.nextInt();
        case DataType.LONG: return r.nextLong();
        case DataType.FLOAT: return r.nextFloat();
        case DataType.DOUBLE: return r.nextDouble();
        case DataType.DATETIME: return new DateTime(r.nextLong());
        case DataType.BAG:
            DataBag db = BagFactory.getInstance().newDefaultBag();
            int sz = r.nextInt(100);
            for (int i = 0; i < sz; i++) {
                int tSz = r.nextInt(10);
                Tuple t = TupleFactory.getInstance().newTuple(tSz);
                for (int j = 0; j < tSz; j++) {
                    t.set(j, r.nextInt());
                }
                db.add(t);
            }
            return db;
        case DataType.MAP:
            Map<String, Object> map = Maps.newHashMap();
            int keys = r.nextInt(10);
            for (int i = 0; i < keys; i++) {
                String key = (String) randData(new FieldSchema(null, DataType.CHARARRAY));
                int values = r.nextInt(100);
                for (int j = 0; j < values; j++) {
                    int tSz = r.nextInt(10);
                    Tuple t = TupleFactory.getInstance().newTuple(tSz);
                    for (int k = 0; k < tSz; k++) {
                        t.set(k, r.nextInt());
                    }
                    map.put(key,  t);
                }
            }
            return map;
        default: throw new RuntimeException("Cannot generate data for given FieldSchema: " + fs);
        }
    }

    public void testTypeAwareGetSetting(TupleFactory tf) throws ExecException {
        SchemaTuple<?> st = (SchemaTuple<?>)tf.newTuple();
        checkNullGetThrowsError(st);
    }

    private void checkNullGetThrowsError(SchemaTuple<?> st) throws ExecException {
        Schema schema = st.getSchema();
        int i = 0;
        for (Schema.FieldSchema fs : schema.getFields()) {
            boolean fieldIsNull = false;
            try {
                switch (fs.type) {
                case DataType.BIGDECIMAL: st.getBigDecimal(i); break;
                case DataType.BIGINTEGER: st.getBigInteger(i); break;
                case DataType.BOOLEAN: st.getBoolean(i); break;
                case DataType.BYTEARRAY: st.getBytes(i); break;
                case DataType.CHARARRAY: st.getString(i); break;
                case DataType.INTEGER: st.getInt(i); break;
                case DataType.LONG: st.getLong(i); break;
                case DataType.FLOAT: st.getFloat(i); break;
                case DataType.DOUBLE: st.getDouble(i); break;
                case DataType.DATETIME: st.getDateTime(i); break;
                case DataType.TUPLE: st.getTuple(i); break;
                case DataType.BAG: st.getDataBag(i); break;
                case DataType.MAP: st.getMap(i); break;
                default: throw new RuntimeException("Unsupported FieldSchema in SchemaTuple: " + fs);
                }
            } catch (FieldIsNullException e) {
                fieldIsNull = true;
            }
            assertTrue(fieldIsNull);
            i++;
        }
    }

    /**
     * This tests that InterStorage will work as expected with the Tuples generated by
     * the given TupleFactory.
     * @param tf
     * @throws Exception
     */
    public void testInterStorageSerDe(SchemaTupleFactory tf) throws Exception {
        int sz = 4096;
        List<Tuple> written = new ArrayList<Tuple>(sz);

        File temp = File.createTempFile("tmp", "tmp");
        temp.deleteOnExit();
        FileOutputStream fos = new FileOutputStream(temp);
        FSDataOutputStream dos = new FSDataOutputStream(fos, null);

        InterRecordWriter writer = new InterRecordWriter(dos,
                PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT,
                PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL_DEFAULT);

        // This test does not cover the case of overlapping record bytes that may be present at the
        // beginning of a split, for that see org.apache.pig.test.TestBinInterSedes#testInterStorageSyncMarker()

        for (int i = 0; i < sz; i++) {
            SchemaTuple<?> st = (SchemaTuple<?>)tf.newTuple();
            fillWithData(st);
            writer.write(null, st);
            written.add(st);
        }
        writer.close(null);

        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "file:///");

        TaskAttemptID taskId = HadoopShims.createTaskAttemptID("jt", 1, true, 1, 1);
        conf.set(MRConfiguration.TASK_ID, taskId.toString());

        InputSplit is = new FileSplit(new Path(temp.getAbsolutePath()), 0, temp.length(), null);

        InterRecordReader reader = new InterRecordReader(PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT,
                                                         PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL_DEFAULT);
        reader.initialize(is, HadoopShims.createTaskAttemptContext(conf, taskId));

        for (int i = 0; i < sz; i++) {
            assertTrue(reader.nextKeyValue());
            SchemaTuple<?> st = (SchemaTuple<?>)reader.getCurrentValue();
            assertEquals(written.get(i), st);
        }
        reader.close();

    }

    public void testSerDe(SchemaTupleFactory tf) throws IOException {
        int sz = 4096;
        List<Tuple> written = new ArrayList<Tuple>(sz);

        File temp = File.createTempFile("tmp", "tmp");
        temp.deleteOnExit();
        FileOutputStream fos = new FileOutputStream(temp);
        DataOutput dos = new DataOutputStream(fos);

        for (int i = 0; i < sz; i++) {
            SchemaTuple<?> st = (SchemaTuple<?>)tf.newTuple();
            fillWithData(st);
            bis.writeDatum(dos, st);
            written.add(st);
        }
        fos.close();

        assertEquals(sz, written.size());

        FileInputStream fis = new FileInputStream(temp);
        DataInput din = new DataInputStream(fis);
        for (int i = 0; i < sz; i++) {
            SchemaTuple<?> st = (SchemaTuple<?>)bis.readDatum(din);
            assertEquals(written.get(i), st);
        }
        fis.close();
    }

    @Test
    public void testFRJoinWithSchemaTuple() throws Exception {
        testJoinType("replicated", false);
    }

    @Test
    public void testMergeJoinWithSchemaTuple() throws Exception {
        testJoinType("merge", true);
    }

    public void testJoinType(String joinType, boolean preSort) throws Exception {
        Properties props = PropertiesUtil.loadDefaultProperties();
        props.setProperty("pig.schematuple", "true");
        PigServer pigServer = new PigServer(ExecType.LOCAL, props);

        Data data = resetData(pigServer);

        data.set("foo1",
            tuple(0, 0),
            tuple(1, 1),
            tuple(2, 2),
            tuple(3, 3),
            tuple(4, 4),
            tuple(5, 5),
            tuple(6, 6),
            tuple(7, 7),
            tuple(8, 8),
            tuple(9, 9)
            );

        data.set("foo2",
            tuple(0, 0),
            tuple(1, 1),
            tuple(2, 2),
            tuple(3, 3),
            tuple(4, 4),
            tuple(5, 5),
            tuple(6, 6),
            tuple(7, 7),
            tuple(8, 8),
            tuple(9, 9)
            );

        pigServer.registerQuery("A = LOAD 'foo1' USING mock.Storage() as (x:int, y:int);");
        pigServer.registerQuery("B = LOAD 'foo2' USING mock.Storage() as (x:int, y:int);");
        if (preSort) {
            pigServer.registerQuery("A = ORDER A BY x ASC;");
            pigServer.registerQuery("B = ORDER B BY x ASC;");
        }
        pigServer.registerQuery("C = JOIN A by x, B by x using '"+joinType+"';");
        pigServer.registerQuery("D = ORDER C BY $0 ASC;");

        Iterator<Tuple> out = pigServer.openIterator("D");
        for (int i = 0; i < 10; i++) {
            if (!out.hasNext()) {
                throw new Exception("Output should have had more elements! Failed on element: " + i);
            }
            assertEquals(tuple(i, i, i, i), out.next());
        }
        assertFalse(out.hasNext());

        pigServer.registerQuery("STORE D INTO 'bar1' USING mock.Storage();");
        pigServer.registerQuery("E = JOIN A by (x, y),  B by (x, y) using '"+joinType+"';");
        pigServer.registerQuery("F = ORDER E BY $0 ASC;");
        pigServer.registerQuery("STORE F INTO 'bar2' USING mock.Storage();");

        List<Tuple> bar1 = data.get("bar1");
        List<Tuple> bar2 = data.get("bar2");

        assertEquals("Output does not have enough elements! List: " + bar1, 10, bar1.size());
        assertEquals("Output does not have enough elements! List: " + bar2, 10, bar2.size());

        for (int i = 0; i < 10; i++) {
            assertEquals(tuple(i, i, i, i), bar1.get(i));
            assertEquals(tuple(i, i, i, i), bar2.get(i));
        }

    }

    public void testHashCodeEqualAsNormalTuple(SchemaTupleFactory tf) {
        SchemaTuple<?> st = tf.newTuple();
        Tuple t = TupleFactory.getInstance().newTuple();
        for (Object o : st) {
            t.append(o);
        }
        assertEquals(t.hashCode(), st.hashCode());
        assertEquals(st, t);
        assertEquals(t, st);
    }

}