diff --git a/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java b/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
index c07f08b..5471420 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
@@ -75,7 +75,7 @@
   /**
    * DataType converter for different computing engines
    */
-  private static final ThreadLocal<DataTypeConverter> converter = new ThreadLocal<>();
+  private static DataTypeConverter converter;
 
   /**
    * This method will convert a given value to its specific type
@@ -84,8 +84,8 @@
    * @param dataType
    * @return
    */
-  public static Object getMeasureValueBasedOnDataType(String msrValue, DataType dataType,
-      int scale, int precision) {
+  public static Object getMeasureValueBasedOnDataType(String msrValue, DataType dataType, int scale,
+      int precision) {
     return getMeasureValueBasedOnDataType(msrValue, dataType, scale, precision, false);
   }
 
@@ -105,7 +105,7 @@
           new BigDecimal(msrValue).setScale(scale, RoundingMode.HALF_UP);
       BigDecimal decimal = normalizeDecimalValue(bigDecimal, precision);
       if (useConverter) {
-        return getDataTypeConverter().convertFromBigDecimalToDecimal(decimal);
+        return converter.convertFromBigDecimalToDecimal(decimal);
       } else {
         return decimal;
       }
@@ -140,11 +140,10 @@
     if (dataType == DataTypes.BOOLEAN) {
       return BooleanConvert.parseBoolean(dimValue);
     } else if (DataTypes.isDecimal(dataType)) {
-      BigDecimal bigDecimal =
-          new BigDecimal(dimValue).setScale(scale, RoundingMode.HALF_UP);
+      BigDecimal bigDecimal = new BigDecimal(dimValue).setScale(scale, RoundingMode.HALF_UP);
       BigDecimal decimal = normalizeDecimalValue(bigDecimal, precision);
       if (useConverter) {
-        return getDataTypeConverter().convertFromBigDecimalToDecimal(decimal);
+        return converter.convertFromBigDecimalToDecimal(decimal);
       } else {
         return decimal;
       }
@@ -457,7 +456,7 @@
       }
     } else {
       // Default action for String/Varchar
-      return getDataTypeConverter().convertFromStringToUTF8String(dimensionValue);
+      return converter.convertFromStringToUTF8String(dimensionValue);
     }
   }
 
@@ -518,7 +517,7 @@
     } else if (actualDataType == DataTypes.LONG) {
       return ByteUtil.toXorBytes((Long) dimensionValue);
     } else if (actualDataType == DataTypes.TIMESTAMP) {
-      return ByteUtil.toXorBytes((Long)dimensionValue);
+      return ByteUtil.toXorBytes((Long) dimensionValue);
     } else {
       // Default action for String/Varchar
       return ByteUtil.toBytes(dimensionValue.toString());
@@ -970,11 +969,12 @@
 
   /**
    * set the data type converter as per computing engine
+   *
    * @param converterLocal
    */
   public static void setDataTypeConverter(DataTypeConverter converterLocal) {
     if (converterLocal != null) {
-      converter.set(converterLocal);
+      converter = converterLocal;
       timeStampformatter.remove();
       dateformatter.remove();
     }
@@ -989,17 +989,10 @@
   }
 
   public static DataTypeConverter getDataTypeConverter() {
-    DataTypeConverter dataTypeConverter = converter.get();
-    if (dataTypeConverter == null) {
-      synchronized (converter) {
-        dataTypeConverter = converter.get();
-        if (dataTypeConverter == null) {
-          dataTypeConverter = new DataTypeConverterImpl();
-          converter.set(dataTypeConverter);
-        }
-      }
+    if (converter == null) {
+      converter = new DataTypeConverterImpl();
     }
-    return dataTypeConverter;
+    return converter;
   }
 
   public static DataType valueOf(String name) {
diff --git a/integration/flink/src/main/java/org/apache/carbon/flink/CarbonLocalWriter.java b/integration/flink/src/main/java/org/apache/carbon/flink/CarbonLocalWriter.java
index c24c3bf..ac39bd0 100644
--- a/integration/flink/src/main/java/org/apache/carbon/flink/CarbonLocalWriter.java
+++ b/integration/flink/src/main/java/org/apache/carbon/flink/CarbonLocalWriter.java
@@ -113,6 +113,9 @@
     synchronized (this) {
       if (!this.flushed) {
         this.closeWriters();
+        this.commit();
+        this.writerFactory.reset();
+        this.writeCount.set(0);
         this.flushed = true;
       }
     }
diff --git a/integration/flink/src/main/java/org/apache/carbon/flink/CarbonS3Writer.java b/integration/flink/src/main/java/org/apache/carbon/flink/CarbonS3Writer.java
index 0c8ccbd..1d3ec6b 100644
--- a/integration/flink/src/main/java/org/apache/carbon/flink/CarbonS3Writer.java
+++ b/integration/flink/src/main/java/org/apache/carbon/flink/CarbonS3Writer.java
@@ -120,6 +120,9 @@
     synchronized (this) {
       if (!this.flushed) {
         this.closeWriters();
+        this.commit();
+        this.writerFactory.reset();
+        this.writeCount.set(0);
         this.flushed = true;
       }
     }
diff --git a/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonPartitionWriter.scala b/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonPartitionWriter.scala
index cc3c4b4..447e83e 100644
--- a/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonPartitionWriter.scala
+++ b/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonPartitionWriter.scala
@@ -71,11 +71,11 @@
       environment.enableCheckpointing(2000L)
       environment.setRestartStrategy(RestartStrategies.noRestart)
 
-      val dataCount = 10000
+      val dataCount = 1000
       val source = new TestSource(dataCount) {
         @throws[InterruptedException]
         override def get(index: Int): Array[AnyRef] = {
-          val data = new Array[AnyRef](5)
+          val data = new Array[AnyRef](7)
           data(0) = "test" + index
           data(1) = index.asInstanceOf[AnyRef]
           data(2) = 12345.asInstanceOf[AnyRef]
@@ -86,7 +86,7 @@
 
         @throws[InterruptedException]
         override def onFinish(): Unit = {
-          Thread.sleep(30000L)
+          Thread.sleep(5000L)
         }
       }
       val stream = environment.addSource(source)
@@ -118,18 +118,99 @@
       assertResult(false)(FileFactory
         .getCarbonFile(CarbonTablePath.getStageDir(tablePath)).listFiles().isEmpty)
 
-      // ensure the carbon data file count in data directory
-      // is same of the data file count which stage files recorded.
-      assertResult(true)(FileFactory.getCarbonFile(dataLocation).listFiles().length ==
-        collectStageInputs(CarbonTablePath.getStageDir(tablePath)).map(
-          stageInput =>
-            stageInput.getLocations.asScala.map(location => location.getFiles.size()).sum
-        ).sum
+      sql(s"INSERT INTO $tableName STAGE")
+
+      checkAnswer(sql(s"select count(1) from $tableName"), Seq(Row(1000)))
+
+    } finally {
+      sql(s"drop table if exists $tableName").collect()
+      delDir(new File(dataPath))
+    }
+  }
+
+  @Test
+  def testComplexType(): Unit = {
+    sql(s"drop table if exists $tableName").collect()
+    sql(
+      s"""
+         | CREATE TABLE $tableName (stringField string, intField int, shortField short,
+         | structField struct<value1:string,value2:int,value3:int>, binaryField struct<value1:binary>)
+         | STORED AS carbondata
+         | PARTITIONED BY (hour_ string, date_ string)
+         | TBLPROPERTIES ('SORT_COLUMNS'='hour_,date_,stringField', 'SORT_SCOPE'='GLOBAL_SORT')
+      """.stripMargin
+    ).collect()
+
+    val rootPath = System.getProperty("user.dir") + "/target/test-classes"
+
+    val dataTempPath = rootPath + "/data/temp/"
+    val dataPath = rootPath + "/data/"
+    delDir(new File(dataPath))
+    new File(dataPath).mkdir()
+
+    try {
+      val tablePath = storeLocation + "/" + tableName + "/"
+
+      val writerProperties = newWriterProperties(dataTempPath, dataPath, storeLocation)
+      val carbonProperties = newCarbonProperties(storeLocation)
+
+      val environment = StreamExecutionEnvironment.getExecutionEnvironment
+      environment.setParallelism(6)
+      environment.enableCheckpointing(2000L)
+      environment.setRestartStrategy(RestartStrategies.noRestart)
+
+      val dataCount = 1000
+      val source = new TestSource(dataCount) {
+        @throws[InterruptedException]
+        override def get(index: Int): Array[AnyRef] = {
+          val data = new Array[AnyRef](7)
+          data(0) = "test" + index
+          data(1) = index.asInstanceOf[AnyRef]
+          data(2) = 12345.asInstanceOf[AnyRef]
+          data(3) = "test\0011\0012"
+          data(4) = "test"
+          data(5) = Integer.toString(TestSource.randomCache.get().nextInt(24))
+          data(6) = "20191218"
+          data
+        }
+
+        @throws[InterruptedException]
+        override def onFinish(): Unit = {
+          Thread.sleep(5000L)
+        }
+      }
+      val stream = environment.addSource(source)
+      val factory = CarbonWriterFactory.builder("Local").build(
+        "default",
+        tableName,
+        tablePath,
+        new Properties,
+        writerProperties,
+        carbonProperties
       )
+      val streamSink = StreamingFileSink.forBulkFormat(new Path(ProxyFileSystem.DEFAULT_URI), factory).build
+
+      stream.keyBy(new KeySelector[Array[AnyRef], AnyRef] {
+        override def getKey(value: Array[AnyRef]): AnyRef = value(3) // return hour_
+      }).addSink(streamSink)
+
+      try environment.execute
+      catch {
+        case exception: Exception =>
+          // TODO
+          throw new UnsupportedOperationException(exception)
+      }
+
+      val dataLocation = dataPath + "default" + CarbonCommonConstants.FILE_SEPARATOR +
+                         tableName + CarbonCommonConstants.FILE_SEPARATOR
+
+      assertResult(true)(FileFactory.isFileExist(dataLocation))
+      assertResult(false)(FileFactory
+        .getCarbonFile(CarbonTablePath.getStageDir(tablePath)).listFiles().isEmpty)
 
       sql(s"INSERT INTO $tableName STAGE")
 
-      checkAnswer(sql(s"select count(1) from $tableName"), Seq(Row(10000)))
+      checkAnswer(sql(s"select count(1) from $tableName"), Seq(Row(1000)))
 
     } finally {
       sql(s"drop table if exists $tableName").collect()
diff --git a/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonWriter.scala b/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonWriter.scala
index 67c7bab..9195863 100644
--- a/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonWriter.scala
+++ b/integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonWriter.scala
@@ -64,7 +64,7 @@
       environment.enableCheckpointing(2000L)
       environment.setRestartStrategy(RestartStrategies.noRestart)
 
-      val dataCount = 10000
+      val dataCount = 1000
       val source = new TestSource(dataCount) {
         @throws[InterruptedException]
         override def get(index: Int): Array[AnyRef] = {
@@ -103,7 +103,7 @@
 
       sql(s"INSERT INTO $tableName STAGE")
 
-      checkAnswer(sql(s"select count(1) from $tableName"), Seq(Row(10000)))
+      checkAnswer(sql(s"select count(1) from $tableName"), Seq(Row(1000)))
 
       // ensure the stage snapshot file and all stage files are deleted
       assertResult(false)(FileFactory.isFileExist(CarbonTablePath.getStageSnapshotFile(tablePath)))
@@ -116,9 +116,9 @@
   }
 
   private def newWriterProperties(
-                                   dataTempPath: String,
-                                   dataPath: String,
-                                   storeLocation: String) = {
+    dataTempPath: String,
+    dataPath: String,
+    storeLocation: String) = {
     val properties = new Properties
     properties.setProperty(CarbonLocalProperty.DATA_TEMP_PATH, dataTempPath)
     properties.setProperty(CarbonLocalProperty.DATA_PATH, dataPath)
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
index f7b8668..bb5e946 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
@@ -26,15 +26,17 @@
 import org.apache.spark.{Accumulator, CarbonInputMetrics, DataSkewRangePartitioner, TaskContext}
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, Row, SparkSession}
-import org.apache.spark.sql.catalyst.expressions.GenericRow
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
 import org.apache.spark.sql.execution.command.ExecutionErrors
 import org.apache.spark.sql.util.{SparkSQLUtil, SparkTypeConverter}
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.unsafe.types.UTF8String
 
 import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants}
+import org.apache.carbondata.converter.SparkDataTypeConverterImpl
+import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.datastore.row.CarbonRow
 import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes, StructField, StructType}
 import org.apache.carbondata.core.metadata.schema.table.CarbonTable
@@ -99,7 +101,7 @@
       ThreadLocalSessionInfo.setConfigurationToCurrentThread(conf.value.value)
       DataLoadProcessorStepOnSpark.convertFunc(rows, index, modelBroadcast, partialSuccessAccum,
         convertStepRowCounter)
-    }.filter(_ != null)// Filter the bad record
+    }.filter(_ != null) // Filter the bad record
 
     // 3. Sort
     val configuration = DataLoadProcessBuilder.createConfiguration(model)
@@ -269,7 +271,7 @@
     val configuration = DataLoadProcessBuilder.createConfiguration(model)
     val header = configuration.getHeader
     val rangeColumn = model.getRangePartitionColumn
-    val rangeColumnIndex = (0 until header.length).find{
+    val rangeColumnIndex = (0 until header.length).find {
       index =>
         header(index).equalsIgnoreCase(rangeColumn.getColName)
     }.get
@@ -427,7 +429,7 @@
       .map(_.getColName)
       .toArray
     val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
-    val rdd: RDD[Row] = new CarbonScanRDD[CarbonRow](
+    val rdd: RDD[InternalRow] = new CarbonScanRDD[CarbonRow](
       sparkSession,
       columnProjection = new CarbonProjection(columns),
       null,
@@ -436,13 +438,13 @@
       carbonTable.getTableInfo,
       new CarbonInputMetrics,
       null,
-      null,
+      classOf[SparkDataTypeConverterImpl],
       classOf[CarbonRowReadSupport],
       splits.asJava)
       .map { row =>
-        new GenericRow(row.getData.asInstanceOf[Array[Any]])
+        new GenericInternalRow(row.getData.asInstanceOf[Array[Any]])
       }
-    sparkSession.createDataFrame(rdd, schema)
+    SparkSQLUtil.execute(rdd, schema, sparkSession)
   }
 }
 
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
index 8f39f9b..13e7c45 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
@@ -22,13 +22,17 @@
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.SparkContext
 import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.EmptyRule
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeSeq, Cast, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeSeq, NamedExpression}
 import org.apache.spark.sql.catalyst.plans.logical
 import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LogicalPlan, Statistics}
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.LogicalRDD
 import org.apache.spark.sql.internal.{SessionState, SQLConf}
+import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.{CarbonReflectionUtils, SerializableConfiguration, SparkUtil, Utils}
 
 object SparkSQLUtil {
@@ -38,6 +42,10 @@
     Dataset.ofRows(sparkSession, logicalPlan)
   }
 
+  def execute(rdd: RDD[InternalRow], schema: StructType, sparkSession: SparkSession): DataFrame = {
+    execute(LogicalRDD(schema.toAttributes, rdd)(sparkSession), sparkSession)
+  }
+
   def getSparkSession: SparkSession = {
     SparkSession.getDefaultSession.get
   }
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
index eb63d03..24e7765 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
@@ -32,12 +32,12 @@
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{CarbonEnv, DataFrame, Row, SparkSession}
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.GenericRow
 import org.apache.spark.sql.execution.command.{Checker, DataCommand}
 import org.apache.spark.sql.util.{SparkSQLUtil, SparkTypeConverter}
 
 import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
 import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.converter.SparkDataTypeConverterImpl
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.datastore.filesystem.CarbonFile
 import org.apache.carbondata.core.datastore.impl.FileFactory
@@ -85,7 +85,6 @@
     val tablePath = table.getTablePath
     val stagePath = CarbonTablePath.getStageDir(tablePath)
     val snapshotFilePath = CarbonTablePath.getStageSnapshotFile(tablePath)
-    var loadModel: CarbonLoadModel = null
     val lock = acquireIngestLock(table)
 
     try {
@@ -133,44 +132,21 @@
       val executorService = Executors.newFixedThreadPool(numThreads)
       val stageInputs = collectStageInputs(executorService, stageFiles)
 
-      // 3) add new segment with INSERT_IN_PROGRESS into table status
-      loadModel = DataLoadProcessBuilderOnSpark.createLoadModelForGlobalSort(spark, table)
-      CarbonLoaderUtil.recordNewLoadMetadata(loadModel)
-
-      // 4) write all existing stage file names and segmentId into a new snapshot file
-      // The content of snapshot file is: first line is segmentId, followed by each line is
-      // one stage file name
-      val content =
-        (Seq(loadModel.getSegmentId) ++ stageFiles.map(_._1.getAbsolutePath)).mkString("\n")
-      FileFactory.writeFile(content, snapshotFilePath)
-
-      // 5) perform data loading
+      // 3) perform data loading
       if (table.isHivePartitionTable) {
-        startLoadingWithPartition(spark, table, loadModel, stageInputs)
+        startLoadingWithPartition(spark, table, stageInputs, stageFiles, snapshotFilePath)
       } else {
-        startLoading(spark, table, loadModel, stageInputs)
+        startLoading(spark, table, stageInputs, stageFiles, snapshotFilePath)
       }
 
-      // 6) write segment file and update the segment entry to SUCCESS
-      val segmentFileName = SegmentFileStore.writeSegmentFile(
-        table, loadModel.getSegmentId, loadModel.getFactTimeStamp.toString)
-      SegmentFileStore.updateTableStatusFile(
-        table, loadModel.getSegmentId, segmentFileName,
-        table.getCarbonTableIdentifier.getTableId,
-        new SegmentFileStore(table.getTablePath, segmentFileName),
-        SegmentStatus.SUCCESS)
-
-      // 7) delete stage files
+      // 4) delete stage files
       deleteStageFiles(executorService, stageFiles)
 
-      // 8) delete the snapshot file
+      // 5) delete the snapshot file
       FileFactory.getCarbonFile(snapshotFilePath).delete()
     } catch {
       case ex: Throwable =>
         LOGGER.error(s"failed to insert ${table.getDatabaseName}.${table.getTableName}", ex)
-        if (loadModel != null) {
-          CarbonLoaderUtil.updateTableStatusForFailure(loadModel)
-        }
         throw ex
     } finally {
       lock.unlock()
@@ -266,24 +242,55 @@
   private def startLoading(
       spark: SparkSession,
       table: CarbonTable,
-      loadModel: CarbonLoadModel,
-      stageInput: Seq[StageInput]
+      stageInput: Seq[StageInput],
+      stageFiles: Array[(CarbonFile, CarbonFile)],
+      snapshotFilePath: String
   ): Unit = {
-    val splits = stageInput.flatMap(_.createSplits().asScala)
-    LOGGER.info(s"start to load ${splits.size} files into " +
-                s"${table.getDatabaseName}.${table.getTableName}")
-    val start = System.currentTimeMillis()
-    val dataFrame = DataLoadProcessBuilderOnSpark.createInputDataFrame(spark, table, splits)
-    DataLoadProcessBuilderOnSpark.loadDataUsingGlobalSort(
-      spark,
-      Option(dataFrame),
-      loadModel,
-      SparkSQLUtil.sessionState(spark).newHadoopConf()
-    ).map { row =>
-        (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-    }
+    var loadModel: CarbonLoadModel = null
+    try {
+      // 1) add new segment with INSERT_IN_PROGRESS into table status
+      loadModel = DataLoadProcessBuilderOnSpark.createLoadModelForGlobalSort(spark, table)
+      CarbonLoaderUtil.recordNewLoadMetadata(loadModel)
 
-    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+      // 2) write all existing stage file names and segmentId into a new snapshot file
+      // The content of snapshot file is: first line is segmentId, followed by each line is
+      // one stage file name
+      val content =
+      (Seq(loadModel.getSegmentId) ++ stageFiles.map(_._1.getAbsolutePath)).mkString("\n")
+      FileFactory.writeFile(content, snapshotFilePath)
+
+      // 3) do loading.
+      val splits = stageInput.flatMap(_.createSplits().asScala)
+      LOGGER.info(s"start to load ${splits.size} files into " +
+                  s"${table.getDatabaseName}.${table.getTableName}")
+      val start = System.currentTimeMillis()
+      val dataFrame = DataLoadProcessBuilderOnSpark.createInputDataFrame(spark, table, splits)
+      DataLoadProcessBuilderOnSpark.loadDataUsingGlobalSort(
+        spark,
+        Option(dataFrame),
+        loadModel,
+        SparkSQLUtil.sessionState(spark).newHadoopConf()
+      ).map { row =>
+          (row._1, FailureCauses.NONE == row._2._2.failureCauses)
+      }
+      LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+
+      // 4) write segment file and update the segment entry to SUCCESS
+      val segmentFileName = SegmentFileStore.writeSegmentFile(
+        table, loadModel.getSegmentId, loadModel.getFactTimeStamp.toString)
+      SegmentFileStore.updateTableStatusFile(
+        table, loadModel.getSegmentId, segmentFileName,
+        table.getCarbonTableIdentifier.getTableId,
+        new SegmentFileStore(table.getTablePath, segmentFileName),
+        SegmentStatus.SUCCESS)
+    } catch {
+      case ex: Throwable =>
+        LOGGER.error(s"failed to insert ${table.getDatabaseName}.${table.getTableName}", ex)
+        if (loadModel != null) {
+          CarbonLoaderUtil.updateTableStatusForFailure(loadModel)
+        }
+        throw ex
+    }
   }
 
   /**
@@ -292,15 +299,18 @@
   private def startLoadingWithPartition(
       spark: SparkSession,
       table: CarbonTable,
-      loadModel: CarbonLoadModel,
-      stageInput: Seq[StageInput]
+      stageInput: Seq[StageInput],
+      stageFiles: Array[(CarbonFile, CarbonFile)],
+      snapshotFilePath: String
     ): Unit = {
     val partitionDataList = listPartitionFiles(stageInput)
+
+    val content = stageFiles.map(_._1.getAbsolutePath).mkString("\n")
+    FileFactory.writeFile(content, snapshotFilePath)
+
     val start = System.currentTimeMillis()
-    var index = 0
     partitionDataList.map {
       case (partition, splits) =>
-        index = index + 1
         LOGGER.info(s"start to load ${splits.size} files into " +
           s"${table.getDatabaseName}.${table.getTableName}. " +
           s"Partition information: ${partition.mkString(",")}")
@@ -484,22 +494,20 @@
       .map(_.getColName)
       .toArray
     val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
-    val rdd: RDD[Row] = new CarbonScanRDD[InternalRow](
-        sparkSession,
-        columnProjection = new CarbonProjection(columns),
-        null,
-        carbonTable.getAbsoluteTableIdentifier,
-        carbonTable.getTableInfo.serialize,
-        carbonTable.getTableInfo,
-        new CarbonInputMetrics,
-        null,
-        null,
-        classOf[SparkRowReadSupportImpl],
-        splits.asJava
-      ).map { row =>
-        new GenericRow(row.toSeq(schema).toArray)
-      }
-    sparkSession.createDataFrame(rdd, schema)
+    val rdd: RDD[InternalRow] = new CarbonScanRDD[InternalRow](
+      sparkSession,
+      columnProjection = new CarbonProjection(columns),
+      null,
+      carbonTable.getAbsoluteTableIdentifier,
+      carbonTable.getTableInfo.serialize,
+      carbonTable.getTableInfo,
+      new CarbonInputMetrics,
+      null,
+      classOf[SparkDataTypeConverterImpl],
+      classOf[SparkRowReadSupportImpl],
+      splits.asJava
+    )
+    SparkSQLUtil.execute(rdd, schema, sparkSession)
   }
 
   override protected def opName: String = "INSERT STAGE"
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index 0309e91..1334178 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -52,7 +52,6 @@
 import org.apache.carbondata.core.datamap.DataMapStoreManager
 import org.apache.carbondata.core.datastore.compression.CompressorFactory
 import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.datastore.row.CarbonRow
 import org.apache.carbondata.core.indexstore.PartitionSpec
 import org.apache.carbondata.core.metadata.SegmentFileStore
 import org.apache.carbondata.core.metadata.encoder.Encoding
@@ -662,34 +661,14 @@
       curAttributes: Seq[AttributeReference],
       sortScope: SortScopeOptions.SortScope,
       isDataFrame: Boolean): (LogicalPlan, Int, Option[RDD[InternalRow]]) = {
+    val catalogAttributes = catalogTable.schema.toAttributes
     // Converts the data as per the loading steps before give it to writer or sorter
-    val convertedRdd = convertData(
+    val updatedRdd = convertData(
       rdd,
       sparkSession,
       loadModel,
       isDataFrame,
       partitionValues)
-    val updatedRdd = if (isDataFrame) {
-      val columnCount = loadModel.getCsvHeaderColumns.length
-      convertedRdd.map { row =>
-        val array = new Array[AnyRef](columnCount)
-        val data = row.getData
-        var i = 0
-        while (i < columnCount) {
-          data(i) match {
-            case string: String =>
-              array(i) = UTF8String.fromString(string)
-            case _ =>
-              array(i) = data(i)
-          }
-          i = i + 1
-        }
-        array
-      }.map(row => InternalRow.fromSeq(row))
-    } else {
-      convertedRdd.map(row => InternalRow.fromSeq(row.getData))
-    }
-    val catalogAttributes = catalogTable.schema.toAttributes
     var attributes = curAttributes.map(a => {
       catalogAttributes.find(_.name.equalsIgnoreCase(a.name)).get
     })
@@ -783,7 +762,7 @@
       sparkSession: SparkSession,
       model: CarbonLoadModel,
       isDataFrame: Boolean,
-      partitionValues: Array[String]): RDD[CarbonRow] = {
+      partitionValues: Array[String]): RDD[InternalRow] = {
     val sc = sparkSession.sparkContext
     val info =
       model.getCarbonDataLoadSchema.getCarbonTable.getTableInfo.getFactTable.getPartitionInfo
@@ -827,7 +806,7 @@
           partialSuccessAccum,
           inputStepRowCounter,
           keepActualData = true)
-      }.filter(_ != null)
+      }.filter(_ != null).map(row => InternalRow.fromSeq(row.getData))
 
     finalRDD
   }