package com.jasongj.spark.reader;

import com.google.common.base.Preconditions;
import com.jasongj.spark.model.Bucket;
import com.jasongj.spark.model.TableMetaData;
import com.jasongj.spark.model.Tuple;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.apache.parquet.example.data.Group;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.validation.constraints.NotNull;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Date;
import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Jason Guo (jason.guo.vip@gmail.com)
 */
public abstract class BucketReaderIterator implements Iterator<Tuple> {
    protected RecordReader recordReader;
    protected Tuple header;
    protected List<String> fieldTypes;
    protected List<Integer> keyIndex;

    public BucketReaderIterator(Configuration hadoopConfiguration, TableMetaData tableMetaData, Integer bucketID){
        this.fieldTypes = tableMetaData.getFields().stream().map(FieldSchema::getType).collect(Collectors.toList());
        this.keyIndex = tableMetaData.getBucketColumns();
    }

    @Override
    public boolean hasNext() {
        if(this.recordReader == null) {
            return false;
        }
        return preFetch() != null;
    }

    public abstract Tuple preFetch();

    @Override
    public Tuple next() {
        if(this.recordReader == null) {
            return null;
        }
        if(this.header != null || preFetch() != null) {
            Tuple tuple = this.header;
            this.header = null;
            return tuple;
        } else {
            return null;
        }
    }

    @Override
    public void remove() {}

}

@@ -32,92 +32,8 @@
 r;%0A%0A
-import com.google.common.base.Preconditions;%0Aimport com.jasongj.spark.model.Bucket;%0A
 impo
@@ -116,52 +116,8 @@
 le;%0A
-import org.apache.commons.lang.StringUtils;%0A
 impo
@@ -161,42 +161,8 @@
 on;%0A
-import org.apache.hadoop.fs.Path;%0A
 impo
@@ -243,771 +243,32 @@
 oop.
-io.LongWritable;%0Aimport org.apache.hadoop.io.Text;%0Aimport org.apache.hadoop.mapreduce.RecordReader;%0Aimport org.apache.hadoop.mapreduce.TaskAttemptContext;%0Aimport org.apache.hadoop.mapreduce.TaskAttemptID;%0Aimport org.apache.hadoop.mapreduce.lib.input.FileSplit;%0Aimport org.apache.hadoop.mapreduce.lib.input.LineRecordReader;%0Aimport org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;%0Aimport org.apache.parquet.example.data.Group;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport javax.validation.constraints.NotNull;%0Aimport java.io.IOException;%0Aimport java.math.BigDecimal;%0Aimport java.sql.Date;%0Aimport java.sql.Timestamp;%0Aimport java.text.ParseException;%0Aimport java.text.SimpleDateFormat;%0Aimport java.util.ArrayList;%0Aimport java.util.Arrays;
+mapreduce.RecordReader;%0A
 %0Aimp
@@ -626,16 +626,59 @@
 eyIndex;
+%0A    protected TableMetaData tableMetaData;
 %0A%0A    pu
@@ -961,16 +961,60 @@
 umns();%0A
+        this.tableMetaData = tableMetaData;%0A
     %7D%0A%0A 
@@ -1057,32 +1057,34 @@
 ext() %7B%0A        
+/*
 if(this.recordRe
@@ -1126,32 +1126,34 @@
 false;%0A        %7D
+*/
 %0A        return 
@@ -1152,24 +1152,26 @@
  return 
-preFetch
+fetchTuple
 () != nu
@@ -1207,24 +1207,26 @@
 t Tuple 
-preFetch
+fetchTuple
 ();%0A%0A   
@@ -1262,32 +1262,34 @@
 ext() %7B%0A        
+/*
 if(this.recordRe
@@ -1334,24 +1334,26 @@
 l;%0A        %7D
+*/
 %0A        if(
@@ -1379,16 +1379,18 @@
  %7C%7C 
-preFetch
+fetchTuple
 () !

refactor BucketReaderIterator -- add TableMetaData as member variable so that each sub-class can use it directly