content: /*
   * Copyright 2022 Alibaba Group Holding Limited.
   *
   * 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 com.alibaba.graphscope.format;
  
  import com.google.common.base.Charsets;
  
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.compress.CompressionCodec;
  import org.apache.hadoop.io.compress.CompressionCodecFactory;
  import org.apache.hadoop.io.compress.SplittableCompressionCodec;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.FileSplit;
  import org.apache.hadoop.mapred.InputSplit;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.JobConfigurable;
  import org.apache.hadoop.mapred.RecordReader;
  import org.apache.hadoop.mapred.Reporter;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  
  public class LongLongInputFormat extends FileInputFormat<LongWritable, LongLong>
          implements JobConfigurable {
  
      private Logger logger = LoggerFactory.getLogger(LongLongInputFormat.class.getName());
      private CompressionCodecFactory compressionCodecs = null;
  
      protected boolean isSplittable(FileSystem fs, Path file) {
          CompressionCodec codec = this.compressionCodecs.getCodec(file);
          return null == codec ? true : codec instanceof SplittableCompressionCodec;
      }
  
      public RecordReader<LongWritable, LongLong> getRecordReader(
              InputSplit genericSplit, JobConf job, Reporter reporter) throws IOException {
          reporter.setStatus(genericSplit.toString());
          String delimiter = job.get("textinputformat.record.delimiter");
          byte[] recordDelimiterBytes = null;
          if (null != delimiter) {
              recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
          }
          return new LongLongRecordReader(job, (FileSplit) genericSplit, recordDelimiterBytes);
      }
  
      @Override
      public void configure(JobConf jobConf) {
          this.compressionCodecs = new CompressionCodecFactory(jobConf);
      }
  }
  