Fixed MOR schema-pruning seq to prune only columns required for mergi…

…ng (and not ones required for incremental reads for ex)

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala

@@ -50,7 +50,7 @@ class MergeOnReadSnapshotRelation(sqlContext: SQLContext,
   override type FileSplit = HoodieMergeOnReadFileSplit
 
   /**
-   * NOTE: These are the fields that are required to properly fulfil Hudi's Merge-on-Read
+   * NOTE: These are the fields that are required to properly fulfil Merge-on-Read (MOR)
    *       semantic:
    *
    *       <ol>
@@ -65,9 +65,11 @@ class MergeOnReadSnapshotRelation(sqlContext: SQLContext,
    *       of primary-key or pre-combine-key are required to be fetched from storage (unless requested
    *       by the query), therefore saving on throughput
    */
-  override lazy val mandatoryFields: Seq[String] =
+  protected lazy val mandatoryFieldsForMerging: Seq[String] =
     Seq(recordKeyField) ++ preCombineFieldOpt.map(Seq(_)).getOrElse(Seq())
 
+  override lazy val mandatoryFields: Seq[String] = mandatoryFieldsForMerging
+
   protected val mergeType: String = optParams.getOrElse(DataSourceReadOptions.REALTIME_MERGE.key,
     DataSourceReadOptions.REALTIME_MERGE.defaultValue)
 
@@ -188,7 +190,7 @@ class MergeOnReadSnapshotRelation(sqlContext: SQLContext,
   }
 
   protected def pruneSchemaForMergeSkipping(requiredSchema: HoodieTableSchema): HoodieTableSchema = {
-    val mandatoryFieldNames = mandatoryFields.map(fieldName => HoodieAvroUtils.getRootLevelFieldName(fieldName))
+    val mandatoryFieldNames = mandatoryFieldsForMerging.map(fieldName => HoodieAvroUtils.getRootLevelFieldName(fieldName))
     val prunedStructSchema = StructType(
       requiredSchema.structTypeSchema.fields.filterNot(f => mandatoryFieldNames.contains(f.name))
     )