diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java
index 2d7bc84..90354f7 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java
@@ -131,29 +131,33 @@
     }
 
     public void populateCache(JobConf jobConf) {
-        String cache = jobConf.get(MAPRED_CACHE_FILES);
-        System.out.println("cache:" + cache);
-        if (cache == null) {
-            return;
-        }
-        String localCache = jobConf.get(MAPRED_CACHE_LOCALFILES);
-        System.out.println("localCache:" + localCache);
-        if (localCache != null) {
-            return;
-        }
-        localCache = "";
-        StringTokenizer cacheTokenizer = new StringTokenizer(cache, ",");
-        while (cacheTokenizer.hasMoreTokens()) {
-            if (!"".equals(localCache)) {
-                localCache += ",";
+        try {
+            String cache = jobConf.get(MAPRED_CACHE_FILES);
+            System.out.println("cache:" + cache);
+            if (cache == null) {
+                return;
             }
-            try {
-                localCache += DCacheClient.get().get(cacheTokenizer.nextToken());
-            } catch (IOException e) {
-                throw new RuntimeException(e);
+            String localCache = jobConf.get(MAPRED_CACHE_LOCALFILES);
+            System.out.println("localCache:" + localCache);
+            if (localCache != null) {
+                return;
             }
+            localCache = "";
+            StringTokenizer cacheTokenizer = new StringTokenizer(cache, ",");
+            while (cacheTokenizer.hasMoreTokens()) {
+                if (!"".equals(localCache)) {
+                    localCache += ",";
+                }
+                try {
+                    localCache += DCacheClient.get().get(cacheTokenizer.nextToken());
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+            jobConf.set(MAPRED_CACHE_LOCALFILES, localCache);
+            System.out.println("localCache:" + localCache);
+        } catch (Exception e) {
+
         }
-        jobConf.set(MAPRED_CACHE_LOCALFILES, localCache);
-        System.out.println("localCache:" + localCache);
     }
 }
\ No newline at end of file
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java
index 6ebe0b9..e169a9f 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java
@@ -23,6 +23,7 @@
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.Mapper;
 import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
 
 import edu.uci.ics.hyracks.api.context.IHyracksContext;
@@ -65,6 +66,7 @@
         public void open() throws HyracksDataException {
             jobConf = getJobConf();
             populateCache(jobConf);
+            Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
             try {
                 mapper = createMapper();
             } catch (Exception e) {
@@ -117,6 +119,7 @@
     private static final long serialVersionUID = 1L;
     private Class<? extends Mapper> mapperClass;
     private InputSplitsProxy inputSplitsProxy;
+    private transient InputSplit[] inputSplits;
 
     private void initializeSplitInfo(InputSplit[] splits) throws IOException {
         jobConf = super.getJobConf();
@@ -165,9 +168,23 @@
 
     @Override
     public IOperatorNodePushable createPushRuntime(IHyracksContext ctx, IOperatorEnvironment env,
-            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
-        return new DeserializedOperatorNodePushable(ctx, new MapperOperator(partition), recordDescProvider
-                .getInputRecordDescriptor(getOperatorId(), 0));
+            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
+        RecordDescriptor recordDescriptor = null;
+        JobConf conf = getJobConf();
+        Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
+        try {
+            if (inputSplits == null) {
+                inputSplits = inputSplitsProxy.toInputSplits(conf);
+            }
+            RecordReader reader = conf.getInputFormat().getRecordReader(inputSplits[partition], conf,
+                    super.createReporter());
+            recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor((Class<? extends Writable>) reader
+                    .createKey().getClass(), (Class<? extends Writable>) reader.createValue().getClass());
+        } catch (Exception e) {
+            throw new HyracksDataException(e);
+        }
+
+        return new DeserializedOperatorNodePushable(ctx, new MapperOperator(partition), recordDescriptor);
     }
 
     public Class<? extends Mapper> getMapperClass() {
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java
index 871ea25..a75e6a9 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java
@@ -51,6 +51,14 @@
     private String inputFormatClassName;
     private Map<String, String> jobConfMap;
     private InputSplitsProxy inputSplitsProxy;
+    private transient JobConf jobConf;
+
+    public JobConf getJobConf() {
+        if (jobConf == null) {
+            jobConf = DatatypeHelper.map2JobConf(jobConfMap);
+        }
+        return jobConf;
+    }
 
     public HadoopReadOperatorDescriptor(JobConf jobConf, JobSpecification spec, InputSplit[] splits) throws IOException {
         super(spec, 0, 1);
@@ -59,9 +67,35 @@
         RecordReader recordReader = inputFormat.getRecordReader(splits[0], jobConf, createReporter());
         recordDescriptors[0] = DatatypeHelper.createKeyValueRecordDescriptor((Class<? extends Writable>) recordReader
                 .createKey().getClass(), (Class<? extends Writable>) recordReader.createValue().getClass());
+        System.out.println(" READER : KEY : " + recordReader.createKey().getClass().getName());
+        System.out.println(" READER : VALUE : " + recordReader.createValue().getClass().getName());
+        System.out.println(" Record descriptor : " + recordDescriptors[0].getFields()[0].getClass().getName());
+        System.out.println(" Record descriptor : " + recordDescriptors[0].getFields()[1].getClass().getName());
         this.setPartitionConstraint(new PartitionCountConstraint(splits.length));
         inputSplitsProxy = new InputSplitsProxy(splits);
         this.inputFormatClassName = inputFormat.getClass().getName();
+        try {
+            checkSplits(jobConf);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public InputSplit[] getInputSplits() throws InstantiationException, IllegalAccessException, IOException {
+        return inputSplitsProxy.toInputSplits(getJobConf());
+    }
+
+    private void checkSplits(JobConf conf) throws Exception {
+        InputSplit[] splits = inputSplitsProxy.toInputSplits(conf);
+        for (InputSplit inputSplit : splits) {
+            Class inputFormatClass = Class.forName(inputFormatClassName);
+            InputFormat inputFormat = (InputFormat) ReflectionUtils.newInstance(inputFormatClass, conf);
+            RecordReader hadoopRecordReader = (RecordReader) inputFormat.getRecordReader(inputSplit, conf,
+                    createReporter());
+            System.out.println("split :" + inputSplit);
+            System.out.println(" READER (proxy): KEY : " + hadoopRecordReader.createKey().getClass().getName());
+            System.out.println(" READER (proxy): VALUE : " + hadoopRecordReader.createValue().getClass().getName());
+        }
     }
 
     protected Reporter createReporter() {
@@ -113,10 +147,11 @@
             public void initialize() throws HyracksDataException {
                 try {
                     JobConf conf = DatatypeHelper.map2JobConf((HashMap) jobConfMap);
+                    Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
                     conf.setClassLoader(this.getClass().getClassLoader());
                     RecordReader hadoopRecordReader;
-                    Writable key;
-                    Writable value;
+                    Object key;
+                    Object value;
                     InputSplit[] splits = inputSplitsProxy.toInputSplits(conf);
                     InputSplit inputSplit = splits[partition];
                     Class inputFormatClass = Class.forName(inputFormatClassName);
@@ -131,13 +166,19 @@
                         inputKeyClass = hadoopRecordReader.createKey().getClass();
                         inputValueClass = hadoopRecordReader.createValue().getClass();
                     }
-                    key = (Writable) inputKeyClass.newInstance();
-                    value = (Writable) inputValueClass.newInstance();
+
+                    key = hadoopRecordReader.createKey();
+                    value = hadoopRecordReader.createValue();
                     ByteBuffer outBuffer = ctx.getResourceManager().allocateFrame();
                     FrameTupleAppender appender = new FrameTupleAppender(ctx);
                     appender.reset(outBuffer, true);
+                    RecordDescriptor outputRecordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor(
+                            (Class<? extends Writable>) hadoopRecordReader.createKey().getClass(),
+                            (Class<? extends Writable>) hadoopRecordReader.createValue().getClass());
+                    /*
                     RecordDescriptor outputRecordDescriptor = recordDescProvider.getOutputRecordDescriptor(
                             getOperatorId(), 0);
+                    */
                     int nFields = outputRecordDescriptor.getFields().length;
                     ArrayTupleBuilder tb = new ArrayTupleBuilder(nFields);
                     while (hadoopRecordReader.next(key, value)) {
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java
index 8a22641..7dbed84 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java
@@ -56,6 +56,7 @@
 
         public ReducerAggregator(Reducer<K2, V2, K3, V3> reducer) {
             this.reducer = reducer;
+            Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
             reducer.configure(getJobConf());
             output = new DataWritingOutputCollector<K3, V3>();
             reporter = new Reporter() {
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java
index 090bca1..a2dac50 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java
@@ -31,6 +31,7 @@
 import org.apache.hadoop.mapred.FileOutputFormat;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.hadoop.mapred.lib.NullOutputFormat;
 
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
@@ -150,16 +151,27 @@
 
     private static FileSplit[] getOutputSplits(JobConf conf, int noOfMappers) {
         int numOutputters = conf.getNumReduceTasks() != 0 ? conf.getNumReduceTasks() : noOfMappers;
-        FileSplit[] outputFileSplits = new FileSplit[numOutputters];
-        String absolutePath = FileOutputFormat.getOutputPath(conf).toString();
-        for (int index = 0; index < numOutputters; index++) {
-            String suffix = new String("part-00000");
-            suffix = new String(suffix.substring(0, suffix.length() - ("" + index).length()));
-            suffix = suffix + index;
-            String outputPath = absolutePath + "/" + suffix;
-            outputFileSplits[index] = new FileSplit("localhost", new File(outputPath));
+        if (conf.getOutputFormat() instanceof NullOutputFormat) {
+            FileSplit[] outputFileSplits = new FileSplit[numOutputters];
+            for (int i = 0; i < numOutputters; i++) {
+                String outputPath = "/tmp/" + System.currentTimeMillis() + i;
+                outputFileSplits[i] = new FileSplit("localhost", new File(outputPath));
+            }
+            return outputFileSplits;
+        } else {
+
+            FileSplit[] outputFileSplits = new FileSplit[numOutputters];
+            String absolutePath = FileOutputFormat.getOutputPath(conf).toString();
+            for (int index = 0; index < numOutputters; index++) {
+                String suffix = new String("part-00000");
+                suffix = new String(suffix.substring(0, suffix.length() - ("" + index).length()));
+                suffix = suffix + index;
+                String outputPath = absolutePath + "/" + suffix;
+                outputFileSplits[index] = new FileSplit("localhost", new File(outputPath));
+            }
+            return outputFileSplits;
         }
-        return outputFileSplits;
+
     }
 
     public HadoopWriteOperatorDescriptor(JobSpecification jobSpec, JobConf jobConf, int numMapTasks) throws Exception {
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/ClasspathBasedHadoopClassFactory.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/ClasspathBasedHadoopClassFactory.java
index 56ddbe5..8137c62 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/ClasspathBasedHadoopClassFactory.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/ClasspathBasedHadoopClassFactory.java
@@ -19,23 +19,22 @@
 
 public class ClasspathBasedHadoopClassFactory implements IHadoopClassFactory {
 
-	@Override
-	public Mapper createMapper(String mapClassName) throws Exception {
-		Class clazz = loadClass(mapClassName);
-		return (Mapper)clazz.newInstance();
-	}
+    @Override
+    public Mapper createMapper(String mapClassName) throws Exception {
+        Class clazz = loadClass(mapClassName);
+        return (Mapper) clazz.newInstance();
+    }
 
-	@Override
-	public Reducer createReducer(String reduceClassName) throws Exception {
-		Class clazz = loadClass(reduceClassName);
-		return (Reducer)clazz.newInstance();
-	}
+    @Override
+    public Reducer createReducer(String reduceClassName) throws Exception {
+        Class clazz = loadClass(reduceClassName);
+        return (Reducer) clazz.newInstance();
+    }
 
-	@Override
-	public Class loadClass(String className) throws Exception {
-		Class clazz = Class.forName(className);
-		return clazz;
-	}
+    @Override
+    public Class loadClass(String className) throws Exception {
+        Class clazz = Class.forName(className);
+        return clazz;
+    }
 
-	
 }
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/DatatypeHelper.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/DatatypeHelper.java
index 23f7f6e..10f29c8 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/DatatypeHelper.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/DatatypeHelper.java
@@ -63,7 +63,7 @@
                 o.readFields(in);
             } catch (IOException e) {
                 e.printStackTrace();
-                // throw new HyracksDataException(e);
+                throw new HyracksDataException(e);
             }
             return o;
         }
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/IHadoopClassFactory.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/IHadoopClassFactory.java
index d465471..1ce78ef 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/IHadoopClassFactory.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/util/IHadoopClassFactory.java
@@ -19,11 +19,11 @@
 import org.apache.hadoop.mapred.Mapper;
 import org.apache.hadoop.mapred.Reducer;
 
-public interface IHadoopClassFactory extends Serializable{
+public interface IHadoopClassFactory extends Serializable {
 
-	public Mapper createMapper(String mapClassName) throws Exception;
-	
-	public Reducer createReducer(String reduceClassName) throws Exception;
-	
-	public Class loadClass(String className) throws Exception;
+    public Mapper createMapper(String mapClassName) throws Exception;
+
+    public Reducer createReducer(String reduceClassName) throws Exception;
+
+    public Class loadClass(String className) throws Exception;
 }