LoadCSVConfiguration.java

package edu.hawaii.ics.yucheng;

import java.util.ArrayList;
import java.util.Properties;

/**
 * An immutable class that contains the configuration information for a cluster
 * of nodes and a catalog. They are parsed from the 'clustercfg' file, and works
 * specifically as the configuration for uploading CSVs to a table on this
 * distributed system. 
 * 
 * @author     Cheng Jade
 * @assignment ICS 421 Assignment 4
 * @date       Mar 22, 2010
 * @bugs       None
 */
public class LoadCSVConfiguration {
    public final ConfigurationNode catalog;
    public final String tableName;
    public final String partitionMethod;
    public final String partitionColumn;
    public final int nodeNumber;
    public final ArrayList<NodeParamPair> nodes = new ArrayList<NodeParamPair>();
    
    /**
     * a inner class that groups the partition parameters together with its node.
     */
    public class NodeParamPair{
        public final ConfigurationNode node;
        public final String param1;
        public final String param2;
        
        /**
         * initialize an instance of this object.
         */
        public NodeParamPair(
                final ConfigurationNode node,
                final String param1,
                final String param2) throws ProgramException{
            if (null == node)
                throw new ProgramException("Invalide Configuration, mismatch node id");
            if (null == param1)
                throw new ProgramException("Invalide Configuration, param1 not found on " + node.name);
            
            this.node = node;
            this.param1 = param1;
            this.param2 = param2;
        }

        /**
         * return a string with class data.
         */
        public String toString() {
            StringBuilder builder = new StringBuilder();
            builder.append(node);
            builder.append("\n    param1: " + param1);
            builder.append("\n    param2: " + param2);
            return builder.toString();
        }
    }
    
    /**
     * return a string with class data.
     */
    public String toString() {
        StringBuilder builder = new StringBuilder();
        builder.append("- catalog:\n" + this.catalog);
        builder.append("\n- table name: " + this.tableName);
        builder.append("\n- partition method: " + this.partitionMethod);
        builder.append("\n- partition column: " + this.partitionColumn);
        builder.append("\n- relevant nodes with partition params:");
        for (NodeParamPair pair : nodes)
            builder.append("\n  node: \n" + pair);
        return builder.toString();
    }
    
    /**
     * initialize an instance of this object
     */
    public LoadCSVConfiguration(final Properties properties,  ConfigurationNode catalog)
            throws ProgramException {
        this.catalog = catalog;
        this.tableName = properties.getProperty("tablename");
        this.partitionMethod = properties.getProperty("partition.method");
        this.partitionColumn = properties.getProperty("partition.column");
        final ArrayList<ConfigurationNode> plainNodes = this.catalog.getNodesFromCatalog(this.tableName);
        
        String nodeNumber = null;
        if (this.partitionMethod.equalsIgnoreCase("range")) {
            nodeNumber = properties.getProperty("numnodes");
            for (int i = 1; i <= plainNodes.size(); i++) {
                final String param1 = properties.getProperty("partition.node" + i + ".param1");
                final String param2 = properties.getProperty("partition.node" + i + ".param2");
                for (int j = 0; j < plainNodes.size(); j++) {
                    if (plainNodes.get(j).name.substring(4).equalsIgnoreCase(Integer.toString(i))) {
                        final ConfigurationNode node = plainNodes.get(j);
                        nodes.add(new NodeParamPair(node, param1, param2));
                        break;
                    }             
                }
            }
        }
        
        else if (this.partitionMethod.equalsIgnoreCase("hash")) {
            nodeNumber = properties.getProperty("partition.param1");
            for (int i = 0; i < plainNodes.size(); i++)
                nodes.add(new NodeParamPair(plainNodes.get(i), nodeNumber, null));
        }
        
        else
            throw new ProgramException("Invalid Cofiguration, only range and hash partition methods are supported");
        
        try {
            this.nodeNumber = Integer.parseInt(nodeNumber);
            if (this.nodeNumber != nodes.size())
                throw new ProgramException("Invalid Configuration, mismatch node number");
        } catch (final NumberFormatException e) {
            throw new ProgramException("Invaild Configuration, invalid node number format");
        }

        if (this.catalog == null)
            throw new ProgramException("Invalid Configuratin, missing catalog");
        if (this.tableName == null)
            throw new ProgramException("Invalid Configuratin, missing table name");
        if (this.partitionMethod == null)
            throw new ProgramException("Invalid Configuratin, missing partition method");
        if (this.partitionColumn == null)
            throw new ProgramException("Invalid Configuratin, missing partition column");
    }

}