<commit_before>/*
*    jETeL/Clover - Java based ETL application framework.
*    Copyright (C) 2002-04  <NAME> <<EMAIL>>
*    
*    This library is free software; you can redistribute it and/or
*    modify it under the terms of the GNU Lesser General Public
*    License as published by the Free Software Foundation; either
*    version 2.1 of the License, or (at your option) any later version.
*    
*    This library is distributed in the hope that it will be useful,
*    but WITHOUT ANY WARRANTY; without even the implied warranty of
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU    
*    Lesser General Public License for more details.
*    
*    You should have received a copy of the GNU Lesser General Public
*    License along with this library; if not, write to the Free Software
*    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*
*/
package org.jetel.component;

import java.io.*;
import java.util.Arrays;
import org.jetel.graph.*;
import org.jetel.data.DataField;
import org.jetel.data.DataRecord;
import org.jetel.data.HashKey;
import org.jetel.data.RecordKey;
import org.jetel.data.Defaults;
import org.jetel.util.ComponentXMLAttributes;
import org.jetel.exception.ComponentNotReadyException;
import org.w3c.dom.Element;

/**
 *  <h3>Partition Component</h3> <!-- Partitions input data into
 * set of partitions (each connected output port becomes one partition.
 * Data is partitioned using different algorithms. Three (RoundRobin,Hash,Rage) are
 * implemented -->
 *
 * <table border="1">
 * <th>Component:</th>
 * <tr><td><h4><i>Name:</i></h4></td>
 * <td>Partition</td></tr>
 * <tr><td><h4><i>Category:</i></h4></td>
 * <td></td></tr>
 * <tr><td><h4><i>Description:</i></h4></td>
 * <td>Partitions data into distinct flows. Each connected port becomes
 * one flow. For each record read, the partition function chosen denotes which output
 * flow (port) will be the record sent to.<br>
 * Which partition algorithm becomes active depends on following:
 * <ul>
 * <li>If <code>partitionKey</code> is <b>NOT</b> specified/defined, RoundRobin algorithm is used.
 * <li>If <code>partitionKey</code> <b>IS</b> specified and <b>NO</b> <code>ranges</code> is specified, then
 * partition by calculated hash value is used. The formula used is: <code>hashValue / MAX_HASH_VALUE * #connected_output_ports) MOD #connected_output_ports</code>
 * <li>If <b>BOTH</b> <code>partitionKey</code> and <code>ranges</code> are specified, then partition by
 * range is used - <i>partitionKey's</i> value is sequentially compared with defined range boundaris. If
 * the value is less or equal to specified boundary, then the record is sent out through the port corresponding
 * to that boundary.<br><i>Note: when boundaries are used, only 1 field can be specified as partitionKey.</i>
 * </ul>
 * </td></tr>
 * <tr><td><h4><i>Inputs:</i></h4></td>
 * <td>[0]- input records</td></tr>
 * <tr><td><h4><i>Outputs:</i></h4></td>
 * <td>[0..n] - one or more output ports connected</td></tr>
 * <tr><td><h4><i>Comment:</i></h4></td>
 * <td></td></tr>
 * </table>
 *  <br>
 *  <table border="1">
 *  <th>XML attributes:</th>
 *  <tr><td><b>type</b></td><td>"PARTITION"</td></tr>
 *  <tr><td><b>id</b></td><td>component identification</td>
 *  <tr><td><b>partitionKey</b><br><i>optional</i></td><td>key which specifies which fields (one or more) will be used for calculating hash valu
 * used for determinig output port. If ranges attribute is not defined, then partition method partition by hash will be used.</td>
 * <tr><td><b>ranges</b><br><i>optional</i></td><td>definition of intervals against which
 * partitionKey will be checked. If key's value belongs to interval then interval's order number is converted
 * to output port number. <br>If this option is used, the partitioning is range partitioning and partition key can
 * be composed of ONE field only (actually 2nd and additional fields will be ignored).<br>
 * <i>Note:Boundaries are first sorted in ascending order, then checked.When checking interval boundaries, the
 * &lt;= operator is used.</i></td>
 *  </tr>
 *  </table>
 *
 *  <h4>Example:</h4>
 *  <pre>&lt;Node id="PARTITION_BY_KEY" type="PARTITION" partitionKey="age"/&gt;</pre>
 *  <pre>&lt;Node id="PARTITION_BY_RANGE" type="PARTITION" partitionKey="age" ranges="18;30;60;80"/&gt;</pre>
 *
 * @author      dpavlis
 * @since       February 28, 2005
 * @revision    $Revision$
 */
public class Partition extends Node {

	/**  Description of the Field */
	public final static String COMPONENT_TYPE = "PARTITION";

	private final static int READ_FROM_PORT=0;
	
	private String[] partitionKeyNames;
	private String[] partitionRanges;

	private RecordKey partitionKey;
	private HashKey hashKey;

	//	 instantiate proper partitioning function
	private PartitionFunction partitionFce;

	private static final String XML_PARTITIONKEY_ATTRIBUTE = "partitionKey";

	private static final String XML_RANGES_ATTRIBUTE = "ranges";

	/**
	 *  Constructor for the Partition object
	 *
	 * @param  id         Description of the Parameter
	 */
	public Partition(String id) {
		super(id);
		partitionFce=null;
	}


    /**
     * @param partitionKeyNames The partitionKeyNames to set.
     */
    public void setPartitionKeyNames(String[] partitionKeyNames) {
        this.partitionKeyNames = partitionKeyNames;
    }
	
	public void setPartitionRanges(String[] partitionRanges){
	    this.partitionRanges=partitionRanges;
	}
    
	/**
	 * Method which can be used to set custom partitioning function
	 * 
	 * @param fce class implementing PartitionFunction interface
	 */
	public void setPartitionFunction(PartitionFunction fce){
	    this.partitionFce=fce;
	}

	/**
	 *  Main processing method for the SimpleCopy object
	 *
	 * @since    April 4, 2002
	 */
	public void run() {
		InputPort inPort;
		OutputPort[] outPorts;

		inPort=getInputPort(READ_FROM_PORT);
		//get array of all output ports defined/connected - use collection Collection - getOutPorts();
		outPorts = (OutputPort[]) getOutPorts().toArray(new OutputPort[0]);
		//create array holding incoming records
		DataRecord inRecord = new DataRecord(inPort.getMetadata());
		inRecord.init();

		while (inRecord!=null && runIt ) {
		    try{
		        inRecord=inPort.readRecord(inRecord);
		        if (inRecord!=null){
		            outPorts[partitionFce.getOutputPort(inRecord)].writeRecord(inRecord);
		        }
		    } catch (IOException ex) {
		        resultMsg = ex.getMessage();
		        resultCode = Node.RESULT_ERROR;
		        closeAllOutputPorts();
		        return;
		    } catch (Exception ex) {
		        resultMsg = ex.getMessage();
		        resultCode = Node.RESULT_FATAL_ERROR;
		        return;
		    }
		    yield();
		}
		broadcastEOF();
		if (runIt) {
			resultMsg = "OK";
		} else {
			resultMsg = "STOPPED";
		}
		resultCode = Node.RESULT_OK;
	}


	/**
	 *  Description of the Method
	 *
	 * @exception  ComponentNotReadyException  Description of the Exception
	 * @since                                  April 4, 2002
	 */
	public void init() throws ComponentNotReadyException {
		// test that we have at least one input port and one output
		if (inPorts.size() < 1) {
			throw new ComponentNotReadyException("At least one input port has to be defined!");
		} else if (outPorts.size() < 1) {
			throw new ComponentNotReadyException("At least one output port has to be defined!");
		}
		// initialize partition key - if defined
		if (partitionKeyNames!=null){<commit_msg>improved reporting of error in case of Exception in run() method<commit_after>/*
*    jETeL/Clover - Java based ETL application framework.
*    Copyright (C) 2002-04  <NAME> <<EMAIL>>
*    
*    This library is free software; you can redistribute it and/or
*    modify it under the terms of the GNU Lesser General Public
*    License as published by the Free Software Foundation; either
*    version 2.1 of the License, or (at your option) any later version.
*    
*    This library is distributed in the hope that it will be useful,
*    but WITHOUT ANY WARRANTY; without even the implied warranty of
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU    
*    Lesser General Public License for more details.
*    
*    You should have received a copy of the GNU Lesser General Public
*    License along with this library; if not, write to the Free Software
*    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*
*/
package org.jetel.component;

import java.io.*;
import java.util.Arrays;
import org.jetel.graph.*;
import org.jetel.data.DataField;
import org.jetel.data.DataRecord;
import org.jetel.data.HashKey;
import org.jetel.data.RecordKey;
import org.jetel.data.Defaults;
import org.jetel.util.ComponentXMLAttributes;
import org.jetel.exception.ComponentNotReadyException;
import org.w3c.dom.Element;

/**
 *  <h3>Partition Component</h3> <!-- Partitions input data into
 * set of partitions (each connected output port becomes one partition.
 * Data is partitioned using different algorithms. Three (RoundRobin,Hash,Rage) are
 * implemented -->
 *
 * <table border="1">
 * <th>Component:</th>
 * <tr><td><h4><i>Name:</i></h4></td>
 * <td>Partition</td></tr>
 * <tr><td><h4><i>Category:</i></h4></td>
 * <td></td></tr>
 * <tr><td><h4><i>Description:</i></h4></td>
 * <td>Partitions data into distinct flows. Each connected port becomes
 * one flow. For each record read, the partition function chosen denotes which output
 * flow (port) will be the record sent to.<br>
 * Which partition algorithm becomes active depends on following:
 * <ul>
 * <li>If <code>partitionKey</code> is <b>NOT</b> specified/defined, RoundRobin algorithm is used.
 * <li>If <code>partitionKey</code> <b>IS</b> specified and <b>NO</b> <code>ranges</code> is specified, then
 * partition by calculated hash value is used. The formula used is: <code>hashValue / MAX_HASH_VALUE * #connected_output_ports) MOD #connected_output_ports</code>
 * <li>If <b>BOTH</b> <code>partitionKey</code> and <code>ranges</code> are specified, then partition by
 * range is used - <i>partitionKey's</i> value is sequentially compared with defined range boundaris. If
 * the value is less or equal to specified boundary, then the record is sent out through the port corresponding
 * to that boundary.<br><i>Note: when boundaries are used, only 1 field can be specified as partitionKey.</i>
 * </ul>
 * </td></tr>
 * <tr><td><h4><i>Inputs:</i></h4></td>
 * <td>[0]- input records</td></tr>
 * <tr><td><h4><i>Outputs:</i></h4></td>
 * <td>[0..n] - one or more output ports connected</td></tr>
 * <tr><td><h4><i>Comment:</i></h4></td>
 * <td></td></tr>
 * </table>
 *  <br>
 *  <table border="1">
 *  <th>XML attributes:</th>
 *  <tr><td><b>type</b></td><td>"PARTITION"</td></tr>
 *  <tr><td><b>id</b></td><td>component identification</td>
 *  <tr><td><b>partitionKey</b><br><i>optional</i></td><td>key which specifies which fields (one or more) will be used for calculating hash valu
 * used for determinig output port. If ranges attribute is not defined, then partition method partition by hash will be used.</td>
 * <tr><td><b>ranges</b><br><i>optional</i></td><td>definition of intervals against which
 * partitionKey will be checked. If key's value belongs to interval then interval's order number is converted
 * to output port number. <br>If this option is used, the partitioning is range partitioning and partition key can
 * be composed of ONE field only (actually 2nd and additional fields will be ignored).<br>
 * <i>Note:Boundaries are first sorted in ascending order, then checked.When checking interval boundaries, the
 * &lt;= operator is used.</i></td>
 *  </tr>
 *  </table>
 *
 *  <h4>Example:</h4>
 *  <pre>&lt;Node id="PARTITION_BY_KEY" type="PARTITION" partitionKey="age"/&gt;</pre>
 *  <pre>&lt;Node id="PARTITION_BY_RANGE" type="PARTITION" partitionKey="age" ranges="18;30;60;80"/&gt;</pre>
 *
 * @author      dpavlis
 * @since       February 28, 2005
 * @revision    $Revision$
 */
public class Partition extends Node {

	/**  Description of the Field */
	public final static String COMPONENT_TYPE = "PARTITION";

	private final static int READ_FROM_PORT=0;
	
	private String[] partitionKeyNames;
	private String[] partitionRanges;

	private RecordKey partitionKey;
	private HashKey hashKey;

	//	 instantiate proper partitioning function
	private PartitionFunction partitionFce;

	private static final String XML_PARTITIONKEY_ATTRIBUTE = "partitionKey";

	private static final String XML_RANGES_ATTRIBUTE = "ranges";

	/**
	 *  Constructor for the Partition object
	 *
	 * @param  id         Description of the Parameter
	 */
	public Partition(String id) {
		super(id);
		partitionFce=null;
	}


    /**
     * @param partitionKeyNames The partitionKeyNames to set.
     */
    public void setPartitionKeyNames(String[] partitionKeyNames) {
        this.partitionKeyNames = partitionKeyNames;
    }
	
	public void setPartitionRanges(String[] partitionRanges){
	    this.partitionRanges=partitionRanges;
	}
    
	/**
	 * Method which can be used to set custom partitioning function
	 * 
	 * @param fce class implementing PartitionFunction interface
	 */
	public void setPartitionFunction(PartitionFunction fce){
	    this.partitionFce=fce;
	}

	/**
	 *  Main processing method for the SimpleCopy object
	 *
	 * @since    April 4, 2002
	 */
	public void run() {
		InputPort inPort;
		OutputPort[] outPorts;

		inPort=getInputPort(READ_FROM_PORT);
		//get array of all output ports defined/connected - use collection Collection - getOutPorts();
		outPorts = (OutputPort[]) getOutPorts().toArray(new OutputPort[0]);
		//create array holding incoming records
		DataRecord inRecord = new DataRecord(inPort.getMetadata());
		inRecord.init();

		while (inRecord!=null && runIt ) {
		    try{
		        inRecord=inPort.readRecord(inRecord);
		        if (inRecord!=null){
		            outPorts[partitionFce.getOutputPort(inRecord)].writeRecord(inRecord);
		        }
		    } catch (IOException ex) {
		        resultMsg = ex.getMessage();
		        resultCode = Node.RESULT_ERROR;
		        closeAllOutputPorts();
		        return;
		    } catch (Exception ex) {
		        resultMsg = ex.getClass().getName()+" : "+ex.getMessage();
		        resultCode = Node.RESULT_FATAL_ERROR;
		        return;
		    }
		    yield();
		}
		broadcastEOF();
		if (runIt) {
			resultMsg = "OK";
		} else {
			resultMsg = "STOPPED";
		}
		resultCode = Node.RESULT_OK;
	}


	/**
	 *  Description of the Method
	 *
	 * @exception  ComponentNotReadyException  Description of the Exception
	 * @since                                  April 4, 2002
	 */
	public void init() throws ComponentNotReadyException {
		// test that we have at least one input port and one output
		if (inPorts.size() < 1) {
			throw new ComponentNotReadyException("At least one input port has to be defined!");
		} else if (outPorts.size() < 1) {
			throw new ComponentNotReadyException("At least one output port has to be defined!");
		}
		// initialize partition key - if defined
		if (partitionKeyNames!=null){
{'fraction_non_alphanumeric': 0.12510762302139214, 'fraction_numerical': 0.009669514537386582, 'mean_word_length': 3.740973312401884, 'pattern_counts': {'":': 0, '<': 311, '<?xml version=': 0, '>': 307, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2365140', 'n_tokens_mistral': 4785, 'n_tokens_neox': 4464, 'n_words': 1448}