<commit_before>/**
 * 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 com.pinterest.secor.io.impl;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.CompressionCodec;

import com.pinterest.secor.common.LogFilePath;
import com.pinterest.secor.io.FileReaderWriter;
import com.pinterest.secor.io.KeyValue;

/**
 * 
 * Sequence file writer implementation
 * 
 * @author <NAME> (<EMAIL>)
 *
 */
public class SequenceFileReaderWriter implements FileReaderWriter {

    private final SequenceFile.Writer writer;
    private final SequenceFile.Reader reader;
    private final LongWritable key;
    private final BytesWritable value;

    // constructor
    public SequenceFileReaderWriter(LogFilePath path, CompressionCodec codec,
            FileReaderWriter.Type type) throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        Path fsPath = new Path(path.getLogFilePath());

        if (type == FileReaderWriter.Type.Reader) {
            this.reader = new SequenceFile.Reader(fs, fsPath, config);
            this.key = (LongWritable) reader.getKeyClass().newInstance();
            this.value = (BytesWritable) reader.getValueClass().newInstance();
            this.writer = null;
        } else if (type == FileReaderWriter.Type.Writer) {
            if (codec != null) {
                this.writer = SequenceFile.createWriter(fs, config, fsPath,
                        LongWritable.class, BytesWritable.class,
                        SequenceFile.CompressionType.BLOCK, codec);
            } else {
                this.writer = SequenceFile.createWriter(fs, config, fsPath,
                        LongWritable.class, BytesWritable.class);
            }
            this.reader = null;
            this.key = null;
            this.value = null;
        } else {
            throw new IllegalArgumentException("Undefined File Type: " + type);
        }

    }

    @Override
    public void close() throws IOException {
        if (this.writer == null) {
            this.writer.close();
        }
        if (this.reader == null) {
            this.reader.close();
        }
    }

    @Override
    public long getLength() throws IOException {
        return this.writer.getLength();
    }

    @Override
    public void write(long key, byte[] value) throws IOException {
        LongWritable writeableKey = new LongWritable(key);
        BytesWritable writeableValue = new BytesWritable(value);
        this.writer.append(writeableKey, writeableValue);
    }

    @Override
    public KeyValue next() throws IOException {
        if (reader.next(key, value)) {
            return new KeyValue(key.get(), value.getBytes());
        } else {
            return null;
        }
    }

}<commit_msg>NPE fix<commit_after>/**
 * 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 com.pinterest.secor.io.impl;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.CompressionCodec;

import com.pinterest.secor.common.LogFilePath;
import com.pinterest.secor.io.FileReaderWriter;
import com.pinterest.secor.io.KeyValue;

/**
 * 
 * Sequence file writer implementation
 * 
 * @author <NAME> (<EMAIL>)
 *
 */
public class SequenceFileReaderWriter implements FileReaderWriter {

    private final SequenceFile.Writer writer;
    private final SequenceFile.Reader reader;
    private final LongWritable key;
    private final BytesWritable value;

    // constructor
    public SequenceFileReaderWriter(LogFilePath path, CompressionCodec codec,
            FileReaderWriter.Type type) throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        Path fsPath = new Path(path.getLogFilePath());

        if (type == FileReaderWriter.Type.Reader) {
            this.reader = new SequenceFile.Reader(fs, fsPath, config);
            this.key = (LongWritable) reader.getKeyClass().newInstance();
            this.value = (BytesWritable) reader.getValueClass().newInstance();
            this.writer = null;
        } else if (type == FileReaderWriter.Type.Writer) {
            if (codec != null) {
                this.writer = SequenceFile.createWriter(fs, config, fsPath,
                        LongWritable.class, BytesWritable.class,
                        SequenceFile.CompressionType.BLOCK, codec);
            } else {
                this.writer = SequenceFile.createWriter(fs, config, fsPath,
                        LongWritable.class, BytesWritable.class);
            }
            this.reader = null;
            this.key = null;
            this.value = null;
        } else {
            throw new IllegalArgumentException("Undefined File Type: " + type);
        }

    }

    @Override
    public void close() throws IOException {
        if (this.writer != null) {
            this.writer.close();
        }
        if (this.reader != null) {
            this.reader.close();
        }
    }

    @Override
    public long getLength() throws IOException {
        return this.writer.getLength();
    }

    @Override
    public void write(long key, byte[] value) throws IOException {
        LongWritable writeableKey = new LongWritable(key);
        BytesWritable writeableValue = new BytesWritable(value);
        this.writer.append(writeableKey, writeableValue);
    }

    @Override
    public KeyValue next() throws IOException {
        if (reader.next(key, value)) {
            return new KeyValue(key.get(), value.getBytes());
        } else {
            return null;
        }
    }

}
{'fraction_non_alphanumeric': 0.09223044397463002, 'fraction_numerical': 0.0010570824524312897, 'mean_word_length': 2.652992277992278, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '4825348', 'n_tokens_mistral': 1997, 'n_tokens_neox': 1943, 'n_words': 589}