/*
 * 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 java.io.File;
import java.io.IOException;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.List;
import java.util.Queue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.pig.PigConfiguration;
import org.apache.pig.classification.InterfaceAudience;
import org.apache.pig.classification.InterfaceStability;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.util.JavaCompilerHelper;
import org.apache.pig.impl.util.ObjectSerializer;

import com.google.common.collect.Lists;

/**
 * This class encapsulates the generation of SchemaTuples, as well as some logic
 * around shipping code to the distributed cache.
 */
@InterfaceAudience.Private
@InterfaceStability.Unstable
public class SchemaTupleClassGenerator {
    private static final Log LOG = LogFactory.getLog(SchemaTupleClassGenerator.class);

    private SchemaTupleClassGenerator() {}

    /**
     * The GenContext mechanism provides a level of control in where SchemaTupleFactories
     * are used. By attaching a GenContext enum type to the registration of a Schema,
     * the code can express the intent of where a SchemaTupleFactory is intended to be used.
     * In this way, if a load func and a join both involve Tuples of the same Schema, it's
     * possible to use SchemaTupleFactories in one but not in the other.
     */
    public static enum GenContext {
        /**
         * This context is used in UDF code. Currently, this is only used for
         * the inputs to UDF's.
         */
        UDF (PigConfiguration.PIG_SCHEMA_TUPLE_USE_IN_UDF, true, GenerateUdf.class),
        /**
         * This context is for POForEach. This will use the expected output of a ForEach
         * to return a typed Tuple.
         */
        FOREACH (PigConfiguration.PIG_SCHEMA_TUPLE_USE_IN_FOREACH, true, GenerateForeach.class),
        /**
         * This context controls whether or not SchemaTuples will be used in FR joins.
         * Currently, they will be used in the HashMap that FR Joins construct.
         */
        FR_JOIN (PigConfiguration.PIG_SCHEMA_TUPLE_USE_IN_FRJOIN, true, GenerateFrJoin.class),
        /**
         * This context controls whether or not SchemaTuples will be used in merge joins.
         */
        MERGE_JOIN (PigConfiguration.PIG_SCHEMA_TUPLE_USE_IN_MERGEJOIN, true, GenerateMergeJoin.class),
        /**
         * All registered Schemas will also be registered in one additional context.
         * This context will allow users to "force" the load of a SchemaTupleFactory
         * if one is present in any context.
         */
        FORCE_LOAD (PigConfiguration.PIG_SCHEMA_TUPLE_ALLOW_FORCE, true, GenerateForceLoad.class);

        /**
         * These annotations are used to mark a given SchemaTuple with
         * the context in which is was intended to be generated.
         */

        @Retention(RetentionPolicy.RUNTIME)
        @Target(ElementType.TYPE)
        public @interface GenerateUdf {}

        @Retention(RetentionPolicy.RUNTIME)
        @Target(ElementType.TYPE)
        public @interface GenerateForeach {}

        @Retention(RetentionPolicy.RUNTIME)
        @Target(ElementType.TYPE)
        public @interface GenerateFrJoin {}

        @Retention(RetentionPolicy.RUNTIME)
        @Target(ElementType.TYPE)
        public @interface GenerateMergeJoin {}

        @Retention(RetentionPolicy.RUNTIME)
        @Target(ElementType.TYPE)
        public @interface GenerateForceLoad {}

        private String key;
        private boolean defaultValue;
        private Class<?> annotation;

        GenContext(String key, boolean defaultValue, Class<?> annotation) {
            this.key = key;
            this.defaultValue = defaultValue;
            this.annotation = annotation;
        }

        public String key() {
            return key;
        }

        public String getAnnotationCanonicalName() {
            return annotation.getCanonicalName();
        }

        /**
         * Checks the generated class to see if the annotation
         * associated with this enum is present.
         * @param clazz
         * @return boolean type value
         */
        @SuppressWarnings({ "unchecked", "rawtypes" })
        public boolean shouldGenerate(Class clazz) {
            return clazz.getAnnotation(annotation) != null;
        }

        /**
         * Given a job configuration file, this checks to see
         * if the default value has been overriden.
         * @param conf
         * @return boolean type value
         */
        public boolean shouldGenerate(Configuration conf) {
            String shouldString = conf.get(key);
            if (shouldString == null) {
                return defaultValue;
            }
            return Boolean.parseBoolean(shouldString);
        }
    }

    /**
     * This value is used to distinguish all of the generated code.
     * The general naming scheme used is SchemaTupe_identifier. Note that
     * identifiers are incremented before code is actually generated.
     */
    private static int nextGlobalClassIdentifier = 0;

    protected static void resetGlobalClassIdentifier() {
        nextGlobalClassIdentifier = 0;
    }

    /**
     * This class actually generates the code for a given Schema.
     * @param   s as Schema
     * @param   appendable as boolean, true or false depending on whether it should be appendable
     * @param   id as int, id means identifier
     * @param   contexts which are a list of contexts in which the SchemaTuple is intended to be instantiated
     */
    protected static void generateSchemaTuple(Schema s, boolean appendable, int id, File codeDir, GenContext... contexts) {
        StringBuilder contextAnnotations = new StringBuilder();
        for (GenContext context : contexts) {
            LOG.info("Including context: " + context);
            contextAnnotations.append("@").append(context.getAnnotationCanonicalName()).append("\n");
        }

        String codeString = produceCodeString(s, appendable, id, contextAnnotations.toString(), codeDir);

        String name = "SchemaTuple_" + id;

        LOG.info("Compiling class " + name + " for Schema: " + s + ", and appendability: " + appendable);

        compileCodeString(name, codeString, codeDir);
    }

    private static int generateSchemaTuple(Schema s, boolean appendable, File codeDir, GenContext... contexts) {
        int id = SchemaTupleClassGenerator.getNextGlobalClassIdentifier();

        generateSchemaTuple(s, appendable, id, codeDir, contexts);

        return id;
    }

    /**
     * This method generates the actual SchemaTuple for the given Schema.
     * @param   schema
     * @param   whether the class should be appendable
     * @param   identifier
     * @return  the generated class's implementation
     */
    private static String produceCodeString(Schema s, boolean appendable, int id, String contextAnnotations, File codeDir) {
        TypeInFunctionStringOutFactory f = new TypeInFunctionStringOutFactory(s, id, appendable, contextAnnotations, codeDir);

        for (Schema.FieldSchema fs : s.getFields()) {
            f.process(fs);
        }

        return f.end();
    }

    protected static int getNextGlobalClassIdentifier() {
        return nextGlobalClassIdentifier++;
    }

    /**
     * This method takes generated code, and compiles it down to a class file. It will output
     * the generated class file to the static temporary directory for generated code. Note
     * that the compiler will use the classpath that Pig is instantiated with, as well as the
     * generated directory.
     *
     * @param String of generated code
     * @param name of class
     */
    //TODO in the future, we can use ASM to generate the bytecode directly.
    private static void compileCodeString(String className, String generatedCodeString, File codeDir) {
        JavaCompilerHelper compiler = new JavaCompilerHelper();
        String tempDir = codeDir.getAbsolutePath();
        compiler.addToClassPath(tempDir);
        LOG.debug("Compiling SchemaTuple code with classpath: " + compiler.getClassPath());
        compiler.compile(tempDir, new JavaCompilerHelper.JavaSourceFromString(className, generatedCodeString));
        LOG.info("Successfully compiled class: " + className);
    }

    static class CompareToSpecificString extends TypeInFunctionStringOut {
        private int id;

        public CompareToSpecificString(int id, boolean appendable) {
            super(appendable);
            this.id = id;
        }

        @Override
        public void prepare() {
            add("@Override");
            add("protected int generatedCodeCompareToSpecific(SchemaTuple_"+id+" t) {");
            add("    int i = 0;");
        }

        @Override
        public void process(int fieldNum, Schema.FieldSchema fs) {
            add("    i = compare(checkIfNull_" + fieldNum + "(), getPos_"
                    + fieldNum + "(), t.checkIfNull_" + fieldNum + "(), t.getPos_"
                    + fieldNum + "());");
            add("    if (i != 0) {");
            add("        return i;");
            add("    }");
        }

        @Override
        public void end() {
            add("    return i;");
            add("}");
        }
    }

    //TODO clear up how it deals with nulls etc. IE is the logic correct
    static class CompareToString extends TypeInFunctionStringOut {
        private int id;

        public CompareToString(int id) {
            this.id = id;
        }

        @Override
        public void prepare() {
            add("@Override");
            add("protected int generatedCodeCompareTo(SchemaTuple t, boolean checkType) {");
            add("    int i;");
        }

        boolean compTup = false;
        boolean compStr = false;
        boolean compIsNull = false;
        boolean compByte = false;

        @Override
        public void process(int fieldNum, Schema.FieldSchema fs) {
            add("        i = compareWithElementAtPos(checkIfNull_" + fieldNum + "(), getPos_" + fieldNum + "(), t, " + fieldNum + ");");
            add("        if (i != 0) {");
            add("            return i;");
            add("        }");
        }

        @Override
        public void end() {
            add("    return 0;");
            add("}");
        }
    }

    static class HashCode extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("public int generatedCodeHashCode() {");
            add("    int h = 17;");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("    h = hashCodePiece(h, getPos_" + fieldPos + "(), checkIfNull_" + fieldPos + "());");
        }

        @Override
        public void end() {
            add("    return h;");
            add("}");
        }
    }

    static class FieldString extends TypeInFunctionStringOut {
        private List<Queue<Integer>> listOfQueuesForIds;
        private Schema schema;

        private int primitives = 0;
        private int isNulls = 0;

        private int booleanBytes = 0;
        private int booleans = 0;
        private File codeDir;

        @Override
        public void prepare() {
            String s;
            try {
                s = ObjectSerializer.serialize(schema);
            } catch (IOException e) {
                throw new RuntimeException("Unable to serialize schema: " + schema, e);
            }
            add("private static Schema schema = staticSchemaGen(\"" + s + "\");");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            if (!isTuple()) {
                if (isPrimitive() && (primitives++ % 8 == 0)) {
                    add("private byte isNull_"+ isNulls++ +" = (byte)0xFF;");
                }

                if (isBoolean()) {
                    if (booleans++ % 8 == 0) {
                        add("private byte booleanByte_"+ booleanBytes++ +";");
                    }
                } else {
                    add("private "+typeName()+" pos_"+fieldPos+";");
                }
            } else {
                int id = SchemaTupleClassGenerator.generateSchemaTuple(fs.schema, isAppendable(), codeDir());

                for (Queue<Integer> q : listOfQueuesForIds) {
                    q.add(id);
                }

                add("private SchemaTuple_"+id+" pos_"+fieldPos+";");
            }
        }

        @Override
        public void end() {
            addBreak();
            add("@Override");
            add("public Schema getSchema() {");
            add("    return schema;");
            add("}");
            addBreak();
        }

        public FieldString(File codeDir, List<Queue<Integer>> listOfQueuesForIds, Schema schema, boolean appendable) {
            super(appendable);
            this.codeDir = codeDir;
            this.listOfQueuesForIds = listOfQueuesForIds;
            this.schema = schema;
        }

        public File codeDir() {
            return codeDir;
        }
    }

    static class SetPosString extends TypeInFunctionStringOut {
        private Queue<Integer> idQueue;

        private int byteField = 0; //this is for setting booleans
        private int byteIncr = 0; //this is for counting the booleans we've encountered

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            if (!isTuple()) {
                add("public void setPos_"+fieldPos+"("+typeName()+" v) {");
                if (isPrimitive()) {
                    add("    setNull_"+fieldPos+"(false);");
                }

                if (!isBoolean()) {
                    add("    pos_"+fieldPos+" = v;");
                } else {
                    add("    booleanByte_" + byteField + " = BytesHelper.setBitByPos(booleanByte_" + byteField + ", v, " + byteIncr++ + ");");

                    if (byteIncr % 8 == 0) {
                        byteIncr = 0;
                        byteField++;
                    }
                }

                add("}");
            } else {
                int nestedSchemaTupleId = idQueue.remove();
                add("public void setPos_"+fieldPos+"(SchemaTuple_"+nestedSchemaTupleId+" t) {");
                add("    pos_" + fieldPos + " = t;");
                add("}");
                addBreak();
                add("public void setPos_"+fieldPos+"(SchemaTuple t) {");
                add("    if (pos_"+fieldPos+" == null) {");
                add("        pos_"+fieldPos+" = new SchemaTuple_"+nestedSchemaTupleId+"();");
                add("    }");
                add("    pos_" + fieldPos + ".setAndCatch(t);");
                add("}");
                addBreak();
                add("public void setPos_"+fieldPos+"(Tuple t) {");
                add("    if (pos_"+fieldPos+" == null) {");
                add("        pos_"+fieldPos+" = new SchemaTuple_"+nestedSchemaTupleId+"();");
                add("    }");
                add("    pos_" + fieldPos + ".setAndCatch(t);");
                add("}");
            }
            addBreak();
        }

        public SetPosString(Queue<Integer> idQueue) {
            this.idQueue = idQueue;
        }
    }

    static class ListSetString extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("public void generatedCodeSetIterator(Iterator<Object> it) throws ExecException {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("    setPos_"+fieldPos+"(unbox(it.next(), getDummy_"+fieldPos+"()));");
        }

        @Override
        public void end() {
            add("}");
        }
    }

    static class GenericSetString extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("public void generatedCodeSetField(int fieldNum, Object val) throws ExecException {");
            add("    switch (fieldNum) {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("    case ("+fieldPos+"):");
            add("        if (val == null) {");
            add("            setNull_" + fieldPos + "(true);");
            add("            return;");
            add("        }");
            add("        setPos_"+fieldPos+"(unbox(val, getDummy_"+fieldPos+"()));");
            add("        break;");
        }

        @Override
        public void end() {
            add("    default:");
            add("        throw new ExecException(\"Invalid index given to set: \" + fieldNum);");
            add("    }");
            add("}");
        }
    }

    static class GenericGetString extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("public Object generatedCodeGetField(int fieldNum) throws ExecException {");
            add("    switch (fieldNum) {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("    case ("+fieldPos+"): return checkIfNull_"+fieldPos+"() ? null : box(getPos_"+fieldPos+"());");
        }

        @Override
        public void end() {
            add("    default: throw new ExecException(\"Invalid index given to get: \" + fieldNum);");
            add("    }");
            add("}");
        }
    }

    static class GeneralIsNullString extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("public boolean isGeneratedCodeFieldNull(int fieldNum) throws ExecException {");
            add("    switch (fieldNum) {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("    case ("+fieldPos+"): return checkIfNull_"+fieldPos+"();");
        }

        @Override
        public void end() {
            add("    default: throw new ExecException(\"Invalid index given: \" + fieldNum);");
            add("    }");
            add("}");
        }
    }

    static class CheckIfNullString extends TypeInFunctionStringOut {
        private int nullByte = 0; //the byte_ val
        private int byteIncr = 0; //the mask we're on

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("public boolean checkIfNull_" + fieldPos + "() {");
            if (isPrimitive()) {
                add("    return BytesHelper.getBitByPos(isNull_" + nullByte + ", " + byteIncr++ +");");
                if (byteIncr % 8 == 0) {
                    byteIncr = 0;
                    nullByte++;
                }
            } else {
               add("    return pos_" + fieldPos + " == null;");
            }
            add("}");
            addBreak();
        }
    }

   static class SetNullString extends TypeInFunctionStringOut {
        private int nullByte = 0; //the byte_ val
        private int byteIncr = 0; //the mask we're on

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("public void setNull_"+fieldPos+"(boolean b) {");
            if (isPrimitive()) {
                add("    isNull_" + nullByte + " = BytesHelper.setBitByPos(isNull_" + nullByte + ", b, " + byteIncr++ + ");");
                if (byteIncr % 8 == 0) {
                    byteIncr = 0;
                    nullByte++;
                }
            } else {
                add("    if (b) {");
                add("        pos_" + fieldPos + " = null;");
                add("    }");
            }
            add("}");
            addBreak();
        }
    }

    //TODO should this do something different if t is null?
    static class SetEqualToSchemaTupleSpecificString extends TypeInFunctionStringOut {
        private int id;

        @Override
        public void prepare() {
            add("@Override");
            add("protected SchemaTuple generatedCodeSetSpecific(SchemaTuple_"+id+" t) {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("    if (t.checkIfNull_" + fieldPos + "()) {");
            add("        setNull_" + fieldPos + "(true);");
            add("    } else {");
            add("        setPos_"+fieldPos+"(t.getPos_"+fieldPos+"());");
            add("    }");
            addBreak();
        }

        @Override
        public void end() {
            add("    return this;");
            add("}");
            addBreak();
        }

        public SetEqualToSchemaTupleSpecificString(int id) {
            this.id = id;
        }
    }

    static class IsSpecificSchemaTuple extends TypeInFunctionStringOut {
        private int id;

        public IsSpecificSchemaTuple(int id) {
            this.id = id;
        }

        @Override
        public void prepare() {
            add("@Override");
            add("public boolean isSpecificSchemaTuple(Object o) {");
            add("    return o instanceof SchemaTuple_" + id + ";");
            add("}");
        }
    }

    //this has to write the null state of all the fields, not just the null bytes, though those
    //will have to be reconstructed
    static class WriteNullsString extends TypeInFunctionStringOut {
        String s = "    boolean[] b = {\n";

        @Override
        public void prepare() {
            add("@Override");
            add("protected boolean[] generatedCodeNullsArray() throws IOException {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            s += "        checkIfNull_"+fieldPos+"(),\n";
        }

        @Override
        public void end() {
            s = s.substring(0, s.length() - 2) + "\n    };";
            add(s);
            add("    return b;");
            add("}");
            addBreak();
        }

        public WriteNullsString(boolean appendable) {
            super(appendable);
        }
    }

   static class ReadString extends TypeInFunctionStringOut {
        private Queue<Integer> idQueue;

        private int booleans = 0;

        @Override
        public void prepare() {
            add("@Override");
            add("protected void generatedCodeReadFields(DataInput in, boolean[] b) throws IOException {");
        }

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            if (isBoolean()) {
                booleans++;
                add("    if (b["+fieldPos+"]) {");
                add("        setNull_"+fieldPos+"(true);");
                add("    } else {");
                add("        setNull_"+fieldPos+"(false);");
                add("    }");
            } else if (!isTuple()) {
                add("    if (b["+fieldPos+"]) {");
                add("        setNull_"+fieldPos+"(true);");
                add("    } else {");
                add("        setPos_"+fieldPos+"(read(in, pos_"+fieldPos+"));");
                add("    }");
                addBreak();
            } else {
                int nestedSchemaTupleId = idQueue.remove();
                add("    if (b["+fieldPos+"]) {");
                add("        setNull_"+fieldPos+"(true);");
                add("    } else {");
                add("        SchemaTuple_"+nestedSchemaTupleId+" st = new SchemaTuple_"+nestedSchemaTupleId+"();");
                add("        st.readFields(in);");
                add("        setPos_"+fieldPos+"(st);");
                add("    }");
                addBreak();
            }
        }

        @Override
        public void end() {
            if (booleans > 0) {
                int i = 0;
                while (booleans > 0) {
                    add("    booleanByte_"+(i++)+" = in.readByte();");
                    booleans -= 8;
                }
            }
            add("}");
            addBreak();
        }

        public ReadString(Queue<Integer> idQueue, boolean appendable) {
            super(appendable);
            this.idQueue = idQueue;
        }
    }


    static class WriteString extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("protected void generatedCodeWriteElements(DataOutput out) throws IOException {");
        }

        private int booleans = 0;

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            if (isBoolean()) {
                booleans++;
            } else {
                add("    if (!checkIfNull_"+fieldPos+"()) {");
                add("        write(out, pos_"+fieldPos+");");
                add("    }");
                addBreak();
            }
        }

        @Override
        public void end() {
            if (booleans > 0) {
                int i = 0;
                while (booleans > 0) {
                    add("    out.writeByte(booleanByte_"+(i++)+");");
                    booleans -= 8;
                }
            }
            add("}");
            addBreak();
        }
    }

    //TODO need to include all of the objects from Schema (have it implement it's own getMemorySize()?
    static class MemorySizeString extends TypeInFunctionStringOut {
        private int size = 0;

        String s = "    return SizeUtil.roundToEight(";

        @Override
        public void prepare() {
            add("@Override");
            add("public long getGeneratedCodeMemorySize() {");
        }

        private int booleans = 0;
        private int primitives = 0;

        //TODO a null array or object variable still takes up space for the pointer, yes?
        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            if (isInt() || isFloat()) {
                size += 4;
            } else if (isLong() || isDouble()) {
                size += 8;
            } else if (isBytearray()) {
                size += 8; //the ptr
                s += "(pos_"+fieldPos+" == null ? 0 : SizeUtil.roundToEight(12 + pos_"+fieldPos+".length) * 8) + ";
            } else if (isBoolean()) {
                if (booleans++ % 8 == 0) {
                    size++; //accounts for the byte used to store boolean values
                }
            } else if (isDateTime()) {
                size += 10; // 8 for long and 2 for short
            } else if (isBag()) {
                size += 8; //the ptr
                s += "(pos_"+fieldPos+" == null ? 0 : pos_"+fieldPos+".getMemorySize()) + ";
            } else if (isMap() || isString() || isBigDecimal() || isBigInteger()) {
                size += 8; //the ptr
                s += "(pos_"+fieldPos+" == null ? 0 : SizeUtil.getPigObjMemSize(pos_"+fieldPos+")) + ";
            } else if (isTuple()) {
                size += 8; //the ptr
                s += "(pos_"+fieldPos+" == null ? 8 : pos_"+fieldPos+".getMemorySize()) + ";
            } else {
                throw new RuntimeException("Unsupported type found: " + fs);
            }

            if (isPrimitive() && primitives++ % 8 == 0) {
                size++; //accounts for the null byte
            }
        }

        @Override
        public void end() {
            s += size + ");";
            add(s);
            add("}");
            addBreak();
        }
    }

    static class GetDummyString extends TypeInFunctionStringOut {
        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            add("public "+typeName()+" getDummy_"+fieldPos+"() {");
            switch (fs.type) {
            case (DataType.INTEGER): add("    return 0;"); break;
            case (DataType.LONG): add("    return 0L;"); break;
            case (DataType.FLOAT): add("    return 0.0f;"); break;
            case (DataType.DOUBLE): add("    return 0.0;"); break;
            case (DataType.BOOLEAN): add("    return true;"); break;
            case (DataType.DATETIME): add("    return new DateTime();"); break;
            case (DataType.BIGDECIMAL): add("    return (BigDecimal)null;"); break;
            case (DataType.BIGINTEGER): add("    return (BigInteger)null;"); break;
            case (DataType.BYTEARRAY): add("    return (byte[])null;"); break;
            case (DataType.CHARARRAY): add("    return (String)null;"); break;
            case (DataType.TUPLE): add("    return (Tuple)null;"); break;
            case (DataType.BAG): add("    return (DataBag)null;"); break;
            case (DataType.MAP): add("    return (Map<String,Object>)null;"); break;
            default: throw new RuntimeException("Unsupported type");
            }
            add("}");
            addBreak();
        }
    }

    static class GetPosString extends TypeInFunctionStringOut {
        private Queue<Integer> idQueue;

        private int booleanByte = 0;
        private int booleans;

        @Override
        public void process(int fieldPos, Schema.FieldSchema fs) {
            if (!isTuple()) {
                add("public "+typeName()+" getPos_"+fieldPos+"() {");
            } else {
                int nestedSchemaTupleId = idQueue.remove();
                add("public SchemaTuple_" + nestedSchemaTupleId + " getPos_"+fieldPos+"() {");
            }
            if (isBoolean()) {
                add("    return BytesHelper.getBitByPos(booleanByte_" + booleanByte + ", " + booleans++ + ");");
                if (booleans % 8 == 0) {
                    booleanByte++;
                    booleans = 0;
                }
            } else {
                add("    return pos_"+fieldPos+";");
            }
            add("}");
            addBreak();
        }

        public GetPosString(Queue<Integer> idQueue) {
            this.idQueue = idQueue;
        }
    }

    static class GetSchemaTupleIdentifierString extends TypeInFunctionStringOut {
        private int id;

        @Override
        public void end() {
            add("@Override");
            add("public int getSchemaTupleIdentifier() {");
            add("    return "+id+";");
            add("}");
            addBreak();
        }

        public GetSchemaTupleIdentifierString(int id) {
            this.id = id;
        }
    }

    static class SchemaSizeString extends TypeInFunctionStringOut {
        int i = 0;

        @Override
        public void process(int fieldNum, Schema.FieldSchema fS) {
            i++;
        }

        @Override
        public void end() {
            add("@Override");
            add("protected int schemaSize() {");
            add("    return " + i + ";");
            add("}");
            addBreak();
        }
    }

    static class SizeString extends TypeInFunctionStringOut {
        int i = 0;

        @Override
        public void process(int fieldNum, Schema.FieldSchema fS) {
            i++;
        }

        @Override
        public void end() {
            add("@Override");
            add("protected int generatedCodeSize() {");
            add("    return " + i + ";");
            add("}");
            addBreak();
        }

        public SizeString(boolean appendable) {
            super(appendable);
        }
    }

    static class GetTypeString extends TypeInFunctionStringOut {
        @Override
        public void prepare() {
            add("@Override");
            add("public byte getGeneratedCodeFieldType(int fieldNum) throws ExecException {");
            add("    switch (fieldNum) {");
        }

        @Override
        public void process(int fieldNum, Schema.FieldSchema fs) {
            add("    case ("+fieldNum+"): return "+fs.type+";");
        }

        @Override
        public void end() {
            add("    default: throw new ExecException(\"Invalid index given: \" + fieldNum);");
            add("    }");
            add("}");
            addBreak();
        }
    }

    static class SetEqualToSchemaTupleString extends TypeInFunctionStringOut {
        int id;

        public SetEqualToSchemaTupleString(int id) {
            this.id = id;
        }

        @Override
        public void prepare() {
            add("@Override");
            add("protected SchemaTuple generatedCodeSet(SchemaTuple t, boolean checkClass) throws ExecException {");
            add("    if (checkClass && t instanceof SchemaTuple_"+id+") {");
            add("        return setSpecific((SchemaTuple_"+id+")t);");
            add("    }");
            addBreak();
            add("    if (t.size() < schemaSize()) {");
            add("        throw new ExecException(\"Given SchemaTuple does not have as many fields as \"+getClass()+\" (\"+t.size()+\" vs \"+schemaSize()+\")\");");
            add("    }");
            addBreak();
            add("    List<Schema.FieldSchema> theirFS = t.getSchema().getFields();");
            addBreak();
        }

        @Override
        public void process(int fieldNum, Schema.FieldSchema fs) {
            add("    if ("+fs.type+" != theirFS.get("+fieldNum+").type) {");
            add("        throw new ExecException(\"Given SchemaTuple does not match current in field " + fieldNum + ". Expected type: " + fs.type + ", found: \" + theirFS.get("+fieldNum+").type);");
            add("    }");
            add("    if (t.isNull("+fieldNum+")) {");
            add("        setNull_"+fieldNum+"(true);");
            add("    } else {");
            if (!isTuple()) {
                add("        setPos_"+fieldNum+"(t.get" + proper(fs.type) + "("+fieldNum+"));");
            } else {
                add("        setPos_"+fieldNum+"((Tuple)t.get("+fieldNum+"));");
            }
            add("    }");
            addBreak();
        }

        @Override
        public void end() {
            add("    return this;");
            add("}");
        }
    }

   static class TypeAwareGetString extends TypeAwareSetString {
        public TypeAwareGetString(byte type) {
            super(type);
        }

        @Override
        public void prepare() {
            add("@Override");
            add("protected "+name()+" generatedCodeGet"+properName()+"(int fieldNum) throws ExecException {");
            add("    switch(fieldNum) {");
        }

        @Override
        public void process(int fieldNum, Schema.FieldSchema fs) {
            if (fs.type==thisType()) {
                add("    case ("+fieldNum+"): return returnUnlessNull(checkIfNull_"+fieldNum+"(), getPos_"+fieldNum+"());");
            }
        }

        @Override
        public void end() {
            add("    default:");
            add("        return unbox"+properName()+"(getTypeAwareBase(fieldNum, \""+name()+"\"));");
            add("    }");
            add("}");
        }
    }

    static class TypeAwareSetString extends TypeInFunctionStringOut {
        private byte type;

        public TypeAwareSetString(byte type) {
            this.type = type;
        }

        public byte thisType() {
            return type;
        }

        public String name() {
            return typeName(type);
        }

        public String properName() {
            return proper(thisType());
        }

        @Override
        public void prepare() {
            add("@Override");
            add("protected void generatedCodeSet"+properName()+"(int fieldNum, "+name()+" val) throws ExecException {");
            add("    switch(fieldNum) {");
        }

        @Override
        public void process(int fieldNum, Schema.FieldSchema fs) {
            if (fs.type==thisType())
                add("    case ("+fieldNum+"): setPos_"+fieldNum+"(val); break;");
        }

        @Override
        public void end() {
            add("    default: setTypeAwareBase(fieldNum, val, \""+name()+"\");");
            add("    }");
            add("}");
        }
    }

    //TODO need to use StringBuilder for all concatenation, not +
    static class TypeInFunctionStringOutFactory {
        private List<TypeInFunctionStringOut> listOfFutureMethods = Lists.newArrayList();
        private int id;
        private boolean appendable;
        private String contextAnnotations;

        public TypeInFunctionStringOutFactory(Schema s, int id, boolean appendable, String contextAnnotations, File codeDir) {
            this.id = id;
            this.appendable = appendable;
            this.contextAnnotations = contextAnnotations;

            Queue<Integer> nextNestedSchemaIdForSetPos = Lists.newLinkedList();
            Queue<Integer> nextNestedSchemaIdForGetPos = Lists.newLinkedList();
            Queue<Integer> nextNestedSchemaIdForReadField = Lists.newLinkedList();

            List<Queue<Integer>> listOfQueuesForIds = Lists.newArrayList(nextNestedSchemaIdForSetPos, nextNestedSchemaIdForGetPos, nextNestedSchemaIdForReadField);

            listOfFutureMethods.add(new FieldString(codeDir, listOfQueuesForIds, s, appendable)); //has to be run first
            listOfFutureMethods.add(new SetPosString(nextNestedSchemaIdForSetPos));
            listOfFutureMethods.add(new GetPosString(nextNestedSchemaIdForGetPos));
            listOfFutureMethods.add(new GetDummyString());
            listOfFutureMethods.add(new GenericSetString());
            listOfFutureMethods.add(new GenericGetString());
            listOfFutureMethods.add(new GeneralIsNullString());
            listOfFutureMethods.add(new CheckIfNullString());
            listOfFutureMethods.add(new SetNullString());
            listOfFutureMethods.add(new SetEqualToSchemaTupleSpecificString(id));
            listOfFutureMethods.add(new WriteNullsString(appendable));
            listOfFutureMethods.add(new ReadString(nextNestedSchemaIdForReadField, appendable));
            listOfFutureMethods.add(new WriteString());
            listOfFutureMethods.add(new SizeString(appendable));
            listOfFutureMethods.add(new MemorySizeString());
            listOfFutureMethods.add(new GetSchemaTupleIdentifierString(id));
            listOfFutureMethods.add(new HashCode());
            listOfFutureMethods.add(new SchemaSizeString());
            listOfFutureMethods.add(new GetTypeString());
            listOfFutureMethods.add(new CompareToString(id));
            listOfFutureMethods.add(new CompareToSpecificString(id, appendable));
            listOfFutureMethods.add(new SetEqualToSchemaTupleString(id));
            listOfFutureMethods.add(new IsSpecificSchemaTuple(id));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.INTEGER));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.LONG));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.FLOAT));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.DOUBLE));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.BYTEARRAY));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.CHARARRAY));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.BOOLEAN));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.DATETIME));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.BIGDECIMAL));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.BIGINTEGER));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.TUPLE));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.BAG));
            listOfFutureMethods.add(new TypeAwareSetString(DataType.MAP));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.INTEGER));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.LONG));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.FLOAT));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.DOUBLE));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.BYTEARRAY));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.CHARARRAY));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.BOOLEAN));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.DATETIME));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.BIGDECIMAL));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.BIGINTEGER));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.TUPLE));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.BAG));
            listOfFutureMethods.add(new TypeAwareGetString(DataType.MAP));
            listOfFutureMethods.add(new ListSetString());

            for (TypeInFunctionStringOut t : listOfFutureMethods) {
                t.prepare();
            }
        }

        public void process(Schema.FieldSchema fs) {
            for (TypeInFunctionStringOut t : listOfFutureMethods)
                t.prepareProcess(fs);
        }

        public String end() {
            StringBuilder head =
                new StringBuilder()
                    .append("import java.util.List;\n")
                    .append("import java.util.Map;\n")
                    .append("import java.util.Iterator;\n")
                    .append("import java.io.DataOutput;\n")
                    .append("import java.io.DataInput;\n")
                    .append("import java.io.IOException;\n")
                    .append("import java.math.BigDecimal;\n")
                    .append("import java.math.BigInteger;\n")
                    .append("\n")
                    .append("import com.google.common.collect.Lists;\n")
                    .append("\n")
                    .append("import org.joda.time.DateTime;")
                    .append("\n")
                    .append("import org.apache.pig.data.DataType;\n")
                    .append("import org.apache.pig.data.DataBag;\n")
                    .append("import org.apache.pig.data.Tuple;\n")
                    .append("import org.apache.pig.data.SchemaTuple;\n")
                    .append("import org.apache.pig.data.AppendableSchemaTuple;\n")
                    .append("import org.apache.pig.data.utils.SedesHelper;\n")
                    .append("import org.apache.pig.data.utils.BytesHelper;\n")
                    .append("import org.apache.pig.data.DataByteArray;\n")
                    .append("import org.apache.pig.data.BinInterSedes;\n")
                    .append("import org.apache.pig.impl.util.Utils;\n")
                    .append("import org.apache.pig.impl.logicalLayer.schema.Schema;\n")
                    .append("import org.apache.pig.impl.logicalLayer.FrontendException;\n")
                    .append("import org.apache.pig.backend.executionengine.ExecException;\n")
                    .append("import org.apache.pig.data.SizeUtil;\n")
                    .append("import org.apache.pig.data.SchemaTuple.SchemaTupleQuickGenerator;\n")
                    .append("\n")
                    .append(contextAnnotations);

            if (appendable) {
                head.append("public class SchemaTuple_"+id+" extends AppendableSchemaTuple<SchemaTuple_"+id+"> {\n");
            } else {
                head.append("public class SchemaTuple_"+id+" extends SchemaTuple<SchemaTuple_"+id+"> {\n");
            }

            for (TypeInFunctionStringOut t : listOfFutureMethods) {
                t.end();
                head.append(t.getContent());
            }

            head.append("\n")
                .append("    @Override\n")
                .append("    public SchemaTupleQuickGenerator<SchemaTuple_" + id + "> getQuickGenerator() {\n")
                .append("        return new SchemaTupleQuickGenerator<SchemaTuple_" + id + ">() {\n")
                .append("            @Override\n")
                .append("            public SchemaTuple_" + id + " make() {\n")
                .append("                return new SchemaTuple_" + id + "();\n")
                .append("            }\n")
                .append("        };\n")
                .append("    }\n");

            return head.append("}").toString();
        }
    }

    static class TypeInFunctionStringOut {
        private int fieldPos = 0;
        private StringBuilder content = new StringBuilder();
        private byte type;

        public void prepare() {}
        public void process(int fieldPos, Schema.FieldSchema fs) {}
        public void end() {}

        public int appendable = -1;

        public StringBuilder getContent() {
            return content;
        }

        public TypeInFunctionStringOut() {
            add("// this code generated by " + getClass());
            addBreak();
        }

        public boolean isAppendable() {
            if (appendable == -1) {
                throw new RuntimeException("Need to be given appendable status in " + getClass());
            }
            return appendable == 1;
        }

        public TypeInFunctionStringOut(boolean appendable) {
            this();
            this.appendable = appendable ? 1 : 0;
        }

        public StringBuilder spaces(int indent) {
            StringBuilder out = new StringBuilder();
            String space = "    ";
            for (int i = 0; i < indent; i++) {
                out.append(space);
            }
            return out;
        }

        public void add(String s) {
            for (String str : s.split("\\n")) {
                content.append(spaces(1).append(str).append("\n"));
            }
        }

        public void addBreak() {
            content.append("\n");
        }

        public void prepareProcess(Schema.FieldSchema fs) {
            type = fs.type;

            process(fieldPos, fs);
            fieldPos++;
        }

        public boolean isInt() {
            return type == DataType.INTEGER;
        }

        public boolean isLong() {
            return type == DataType.LONG;
        }

        public boolean isFloat() {
            return type == DataType.FLOAT;
        }

        public boolean isDouble() {
            return type == DataType.DOUBLE;
        }

        public boolean isDateTime() {
            return type == DataType.DATETIME;
        }

        public boolean isBigDecimal() {
            return type == DataType.BIGDECIMAL;
        }

        public boolean isBigInteger() {
            return type == DataType.BIGINTEGER;
        }

        public boolean isPrimitive() {
            return isInt() || isLong() || isFloat() || isDouble() || isBoolean();
        }

        public boolean isBoolean() {
            return type == DataType.BOOLEAN;
        }

        public boolean isString() {
            return type == DataType.CHARARRAY;
        }

        public boolean isBytearray() {
            return type == DataType.BYTEARRAY;
        }

        public boolean isTuple() {
            return type == DataType.TUPLE;
        }

        public boolean isBag() {
            return type == DataType.BAG;
        }

        public boolean isMap() {
            return type == DataType.MAP;
        }

        public boolean isObject() {
            return !isPrimitive();
        }

        public String typeName() {
            return typeName(type);
        }

        public String typeName(byte type) {
            switch(type) {
                case (DataType.INTEGER): return "int";
                case (DataType.LONG): return "long";
                case (DataType.FLOAT): return "float";
                case (DataType.DOUBLE): return "double";
                case (DataType.BYTEARRAY): return "byte[]";
                case (DataType.CHARARRAY): return "String";
                case (DataType.BOOLEAN): return "boolean";
                case (DataType.DATETIME): return "DateTime";
                case (DataType.BIGDECIMAL): return "BigDecimal";
                case (DataType.BIGINTEGER): return "BigInteger";
                case (DataType.TUPLE): return "Tuple";
                case (DataType.BAG): return "DataBag";
                case (DataType.MAP): return "Map";
                default: throw new RuntimeException("Can't return String for given type " + DataType.findTypeName(type));
            }
        }

        public String proper(byte type) {
            String s = typeName(type);
            switch (type) {
            case DataType.BYTEARRAY: return "Bytes";
            default: return s.substring(0,1).toUpperCase() + s.substring(1);
            }
        }
    }
}