<commit_before>/***********************************************************************************************************************
 *
 * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu)
 *
 * Licensed 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 eu.stratosphere.pact.common.io;


import java.io.IOException;
import java.io.UnsupportedEncodingException;

import eu.stratosphere.nephele.configuration.Configuration;
import eu.stratosphere.pact.common.contract.FileDataSink;
import eu.stratosphere.pact.common.type.PactRecord;


/**
 * The base class for output formats that serialize their records into a delimited sequence.
 */
public abstract class DelimitedOutputFormat extends FileOutputFormat
{
	/**
	 * The configuration key for the entry that defines the record delimiter.
	 */
	public static final String RECORD_DELIMITER = "pact.output.delimited.delimiter";

	/**
	 * The configuration key to set the record delimiter encoding.
	 */
	private static final String RECORD_DELIMITER_ENCODING = "pact.output.delimited.delimiter-encoding";
	
	/**
	 * The configuration key for the entry that defines the write-buffer size.
	 */
	public static final String WRITE_BUFFER_SIZE = "pact.output.delimited.buffersize";
	
	/**
	 * The default write-buffer size. 64 KiByte. 
	 */
	private static final int DEFAULT_WRITE_BUFFER_SIZE = 64 * 1024;
	
	/**
	 * The minimal write-buffer size, 1 KiByte.
	 */
	private static final int MIN_WRITE_BUFFER_SIZE = 1024;
	
	// --------------------------------------------------------------------------------------------
	
	private byte[] delimiter;
	
	private byte[] buffer;
	
	private byte[] targetArray = new byte[64];
	
	private int pos;
	
	private int bufferSize;
	
	// --------------------------------------------------------------------------------------------
	
	
	/**
	 * Calls the super classes to configure themselves and reads the config parameters for the delimiter and
	 * the write buffer size.
	 * 
	 *  @param config The configuration to read the parameters from.
	 *  
	 * @see eu.stratosphere.pact.common.io.FileOutputFormat#configure(eu.stratosphere.nephele.configuration.Configuration)
	 */
	public void configure(Configuration config)
	{
		super.configure(config);
		
		final String delim = config.getString(RECORD_DELIMITER, "\n");
		final String charsetName = config.getString(RECORD_DELIMITER_ENCODING, null);		
		if (delim == null) {
			throw new IllegalArgumentException("The delimiter in the DelimitedOutputFormat must not be null.");
		}
		try {
			this.delimiter = charsetName == null ? delim.getBytes() : delim.getBytes(charsetName);
		} catch (UnsupportedEncodingException useex) {
			throw new IllegalArgumentException("The charset with the name '" + charsetName + 
				"' is not supported on this TaskManager instance.", useex);
		}
		
		this.bufferSize = config.getInteger(WRITE_BUFFER_SIZE, DEFAULT_WRITE_BUFFER_SIZE);
		if (this.bufferSize < MIN_WRITE_BUFFER_SIZE) {
			throw new IllegalArgumentException("The write buffer size must not be less than " + MIN_WRITE_BUFFER_SIZE
				+ " bytes.");
		}
	}
	
	/* (non-Javadoc)
	 * @see eu.stratosphere.pact.common.io.FileOutputFormat#open(int)
	 */
	public void open(int taskNumber) throws IOException
	{
		super.open(taskNumber);
		
		if (this.buffer == null) {
			this.buffer = new byte[this.bufferSize];
		}
		if (this.targetArray == null) {
			this.targetArray = new byte[64];
		}
		this.pos = 0;
	}
	
	// --------------------------------------------------------------------------------------------
	
	/* (non-Javadoc)
	 * @see eu.stratosphere.pact.common.io.FileOutputFormat#close()
	 */
	@Override
	public void close() throws IOException
	{
		this.stream.write(this.buffer, 0, this.pos);
		
		// close file stream
		super.close();
	}

	/**
	 * This method is called for every record so serialize itself into the given target array. The method should
	 * return the number of bytes occupied in the target array. If the target array is not large enough, a negative 
	 * value should be returned.
	 * <p>
	 * The absolute value of the returned integer can be given as a hint how large an array is required. The array is
	 * resized to the return value's absolute value, if that is larger than the current array size. Otherwise, the
	 * array size is simply doubled.
	 * 
	 * @param rec The record to be serialized.
	 * @param target The array to serialize the record into.
	 * @return The length of the serialized contents, or a negative value, indicating that the array is too small.
	 * 
	 * @throws Exception If the user code produces an exception that prevents processing the record, it should
	 *                   throw it such that the engine recognizes the situation as a fault.<commit_msg>Fixed error during closing of delimited input format.<commit_after>/***********************************************************************************************************************
 *
 * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu)
 *
 * Licensed 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 eu.stratosphere.pact.common.io;


import java.io.IOException;
import java.io.UnsupportedEncodingException;

import eu.stratosphere.nephele.configuration.Configuration;
import eu.stratosphere.pact.common.contract.FileDataSink;
import eu.stratosphere.pact.common.type.PactRecord;


/**
 * The base class for output formats that serialize their records into a delimited sequence.
 */
public abstract class DelimitedOutputFormat extends FileOutputFormat
{
	/**
	 * The configuration key for the entry that defines the record delimiter.
	 */
	public static final String RECORD_DELIMITER = "pact.output.delimited.delimiter";

	/**
	 * The configuration key to set the record delimiter encoding.
	 */
	private static final String RECORD_DELIMITER_ENCODING = "pact.output.delimited.delimiter-encoding";
	
	/**
	 * The configuration key for the entry that defines the write-buffer size.
	 */
	public static final String WRITE_BUFFER_SIZE = "pact.output.delimited.buffersize";
	
	/**
	 * The default write-buffer size. 64 KiByte. 
	 */
	private static final int DEFAULT_WRITE_BUFFER_SIZE = 64 * 1024;
	
	/**
	 * The minimal write-buffer size, 1 KiByte.
	 */
	private static final int MIN_WRITE_BUFFER_SIZE = 1024;
	
	// --------------------------------------------------------------------------------------------
	
	private byte[] delimiter;
	
	private byte[] buffer;
	
	private byte[] targetArray = new byte[64];
	
	private int pos;
	
	private int bufferSize;
	
	// --------------------------------------------------------------------------------------------
	
	
	/**
	 * Calls the super classes to configure themselves and reads the config parameters for the delimiter and
	 * the write buffer size.
	 * 
	 *  @param config The configuration to read the parameters from.
	 *  
	 * @see eu.stratosphere.pact.common.io.FileOutputFormat#configure(eu.stratosphere.nephele.configuration.Configuration)
	 */
	public void configure(Configuration config)
	{
		super.configure(config);
		
		final String delim = config.getString(RECORD_DELIMITER, "\n");
		final String charsetName = config.getString(RECORD_DELIMITER_ENCODING, null);		
		if (delim == null) {
			throw new IllegalArgumentException("The delimiter in the DelimitedOutputFormat must not be null.");
		}
		try {
			this.delimiter = charsetName == null ? delim.getBytes() : delim.getBytes(charsetName);
		} catch (UnsupportedEncodingException useex) {
			throw new IllegalArgumentException("The charset with the name '" + charsetName + 
				"' is not supported on this TaskManager instance.", useex);
		}
		
		this.bufferSize = config.getInteger(WRITE_BUFFER_SIZE, DEFAULT_WRITE_BUFFER_SIZE);
		if (this.bufferSize < MIN_WRITE_BUFFER_SIZE) {
			throw new IllegalArgumentException("The write buffer size must not be less than " + MIN_WRITE_BUFFER_SIZE
				+ " bytes.");
		}
	}
	
	/* (non-Javadoc)
	 * @see eu.stratosphere.pact.common.io.FileOutputFormat#open(int)
	 */
	public void open(int taskNumber) throws IOException
	{
		super.open(taskNumber);
		
		if (this.buffer == null) {
			this.buffer = new byte[this.bufferSize];
		}
		if (this.targetArray == null) {
			this.targetArray = new byte[64];
		}
		this.pos = 0;
	}
	
	// --------------------------------------------------------------------------------------------
	
	/* (non-Javadoc)
	 * @see eu.stratosphere.pact.common.io.FileOutputFormat#close()
	 */
	@Override
	public void close() throws IOException {
		if (this.stream != null) {
			this.stream.write(this.buffer, 0, this.pos);
		}
		
		// close file stream
		super.close();
	}

	/**
	 * This method is called for every record so serialize itself into the given target array. The method should
	 * return the number of bytes occupied in the target array. If the target array is not large enough, a negative 
	 * value should be returned.
	 * <p>
	 * The absolute value of the returned integer can be given as a hint how large an array is required. The array is
	 * resized to the return value's absolute value, if that is larger than the current array size. Otherwise, the
	 * array size is simply doubled.
	 * 
	 * @param rec The record to be serialized.
	 * @param target The array to serialize the record into.
	 * @return The length of the serialized contents, or a negative value, indicating that the array is too small.
	 * 
	 * @throws Exception If the user code produces an exception that prevents processing the record, it should
	 *                   throw it such that the engine recognizes the situation as a fault.
{'fraction_non_alphanumeric': 0.17344575187621605, 'fraction_numerical': 0.005003242842583156, 'mean_word_length': 3.1073059360730593, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2774391', 'n_tokens_mistral': 3044, 'n_tokens_neox': 2681, 'n_words': 1017}