<commit_before>package rapcap.hadoop.mr1.lzo;

import java.io.DataInputStream;
import java.io.IOException;

import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

import com.hadoop.compression.lzo.LzopCodec;
import com.hadoop.compression.lzo.LzopDecompressor;

public class LzopRecordReader implements RecordReader<LongWritable, BytesWritable> {
	
	Seekable baseStream;
	DataInputStream stream;
	Reporter reporter;

	long start, next_start;
	public PublicLzopInputStream decompressor_stream;
	
	public LzopRecordReader(long start, long next_start, Seekable baseStream, DataInputStream stream, Reporter reporter) throws IOException {
		this.baseStream = baseStream;
		this.stream = stream;
		this.start = start;
		this.next_start = next_start;
		this.reporter = reporter;
		
		int buffer_size = LzopCodec.DEFAULT_LZO_BUFFER_SIZE;
		LzopDecompressor decompressor = new LzopDecompressor(buffer_size);
		
		baseStream.seek(0);
		decompressor_stream = new PublicLzopInputStream(stream, decompressor, buffer_size);
		baseStream.seek(start);
	}

	public void close() throws IOException {
		stream.close();
		decompressor_stream.close();
	}

	public LongWritable createKey() {
		return new LongWritable();
	}

	public BytesWritable createValue() {
		return new BytesWritable();
	}

	public long getPos() throws IOException {
		return baseStream.getPos();
	}

	public float getProgress() throws IOException {
		if (getPos() >= next_start)
			return 1;
		
		return Math.min(1.0f, (getPos() - start) / (float)(next_start - start));
	}

	public boolean next(LongWritable key, BytesWritable value) throws IOException {
		if (getPos() >= next_start)
			return false;

		long start_pos = getPos();
		int decompressed_size = stream.readInt();
		baseStream.seek(start_pos);

/*
		if (decompressor_buffer == null
		 || decompressor_buffer.length < decompressed_size) {
			decompressor_buffer = new byte[decompressed_size];
		}
*/		
		value.setCapacity(decompressed_size);
		
		try {
		decompressor_stream.decompress(value.getBytes(), 0, decompressed_size);
		}
		catch (IOException e) {
			throw new IOException("start=" + start + ", pos=" + start_pos + ", size=" + decompressed_size + ", " + e.getMessage());
		}

		key.set(getPos());

		reporter.setStatus("Read " + getPos() + " of " + next_start + " bytes");
		reporter.progress();

		return true;
	}
}

<commit_msg>debugging messages for Lzop RecordReader<commit_after>package rapcap.hadoop.mr1.lzo;

import java.io.DataInputStream;
import java.io.IOException;

import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

import com.hadoop.compression.lzo.LzopCodec;
import com.hadoop.compression.lzo.LzopDecompressor;

public class LzopRecordReader implements RecordReader<LongWritable, BytesWritable> {
	
	Seekable baseStream;
	DataInputStream stream;
	Reporter reporter;

	long start, next_start;
	public PublicLzopInputStream decompressor_stream;
	
	public LzopRecordReader(long start, long next_start, Seekable baseStream, DataInputStream stream, Reporter reporter) throws IOException {
		this.baseStream = baseStream;
		this.stream = stream;
		this.start = start;
		this.next_start = next_start;
		this.reporter = reporter;
		
		int buffer_size = LzopCodec.DEFAULT_LZO_BUFFER_SIZE;
		LzopDecompressor decompressor = new LzopDecompressor(buffer_size);
		
		baseStream.seek(0);
		decompressor_stream = new PublicLzopInputStream(stream, decompressor, buffer_size);
		baseStream.seek(start);
	}

	public void close() throws IOException {
		stream.close();
		decompressor_stream.close();
	}

	public LongWritable createKey() {
		return new LongWritable();
	}

	public BytesWritable createValue() {
		return new BytesWritable();
	}

	public long getPos() throws IOException {
		return baseStream.getPos();
	}

	public float getProgress() throws IOException {
		if (getPos() >= next_start)
			return 1;
		
		return Math.min(1.0f, (getPos() - start) / (float)(next_start - start));
	}

	public boolean next(LongWritable key, BytesWritable value) throws IOException {
		if (getPos() >= next_start)
			return false;

		long start_pos = getPos();
		int decompressed_size = stream.readInt();
		baseStream.seek(start_pos);

/*
		if (decompressor_buffer == null
		 || decompressor_buffer.length < decompressed_size) {
			decompressor_buffer = new byte[decompressed_size];
		}
*/		
		value.setCapacity(decompressed_size);
		
		try {
		decompressor_stream.decompress(value.getBytes(), 0, decompressed_size);
		}
		catch (Exception e) {
			throw new IOException("start=" + start + ", pos=" + start_pos + ", size=" + decompressed_size + ", " + e.getMessage());
		}

		key.set(getPos());

		reporter.setStatus("Read " + getPos() + " of " + next_start + " bytes");
		reporter.progress();

		return true;
	}
}


{'fraction_non_alphanumeric': 0.11148581630628844, 'fraction_numerical': 0.0023804800634794686, 'mean_word_length': 4.815455594002307, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6276019', 'n_tokens_mistral': 1765, 'n_tokens_neox': 1611, 'n_words': 346}