fix execution error in non-native parquet sink tasks

spark-extension-shims-spark3/src/main/scala/org/apache/spark/sql/execution/blaze/plan/NativeParquetInsertIntoHiveTableExec.scala

@@ -113,9 +113,17 @@ case class NativeParquetInsertIntoHiveTableExec(
       new BasicWriteJobStatsTracker(serializableHadoopConf, metrics) {
         override def newTaskInstance(): WriteTaskStatsTracker = {
           new BasicWriteTaskStatsTracker(serializableHadoopConf.value) {
-            override def newRow(_filePath: String, _row: InternalRow): Unit = {}
+            override def newRow(filePath: String, row: InternalRow): Unit = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.newRow(filePath, row)
+              }
+            }
 
             override def closeFile(filePath: String): Unit = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.closeFile(filePath)
+              }
+
               val outputFileStat = ParquetSinkTaskContext.get.processedOutputFiles.remove()
               for (_ <- 0L until outputFileStat.numRows) {
                 super.newRow(filePath, null)
@@ -147,12 +155,23 @@ case class NativeParquetInsertIntoHiveTableExec(
               mutable.ArrayBuffer.empty
 
             override def newPartition(partitionValues: InternalRow): Unit = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.newPartition(partitionValues)
+              }
               partitions.append(partitionValues)
             }
 
-            override def newRow(_row: InternalRow): Unit = {}
+            override def newRow(row: InternalRow): Unit = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.newRow(row)
+              }
+            }
 
             override def getFinalStats(): WriteTaskStats = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.getFinalStats()
+              }
+
               val outputFileStat = ParquetSinkTaskContext.get.processedOutputFiles.remove()
               BasicWriteTaskStats(
                 partitions = partitions,
@@ -179,9 +198,17 @@ case class NativeParquetInsertIntoHiveTableExec(
       new BasicWriteJobStatsTracker(serializableHadoopConf, metrics) {
         override def newTaskInstance(): WriteTaskStatsTracker = {
           new BasicWriteTaskStatsTracker(serializableHadoopConf.value) {
-            override def newRow(_row: InternalRow): Unit = {}
+            override def newRow(row: InternalRow): Unit = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.newRow(row)
+              }
+            }
 
             override def getFinalStats(): WriteTaskStats = {
+              if (!ParquetSinkTaskContext.get.isNative) {
+                return super.getFinalStats()
+              }
+
               val outputFileStat = ParquetSinkTaskContext.get.processedOutputFiles.remove()
               BasicWriteTaskStats(
                 numPartitions = 1,

spark-extension/src/main/scala/org/apache/spark/sql/execution/blaze/plan/NativeParquetInsertIntoHiveTableBase.scala

@@ -165,6 +165,7 @@ class BlazeMapredParquetOutputFormat
 case class OutputFileStat(path: String, numRows: Long, numBytes: Long)
 
 class ParquetSinkTaskContext {
+  var isNative: Boolean = false
   val processingOutputFiles = new LinkedBlockingDeque[String]()
   val processedOutputFiles = new util.ArrayDeque[OutputFileStat]()
 }

spark-extension/src/main/scala/org/apache/spark/sql/execution/blaze/plan/NativeParquetSinkBase.scala

@@ -97,6 +97,9 @@ abstract class NativeParquetSinkBase(
       inputRDD.isShuffleReadFull,
       (partition, context) => {
 
+        // mark for native parquet sink
+        ParquetSinkTaskContext.get.isNative = true
+
         // init hadoop fs
         val resourceId = s"NativeParquetSinkExec:${UUID.randomUUID().toString}"
         JniBridge.resourcesMap.put(