[HUDI-8892] Introduce projection push down for payload mode

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecordMerger.java

@@ -20,13 +20,19 @@
 
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.HoodieRecordUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.common.util.collection.Pair;
 
 import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.IndexedRecord;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Properties;
 
 /**
@@ -66,4 +72,31 @@ private Option<IndexedRecord> combineAndGetUpdateValue(HoodieRecord older, Hoodi
   public HoodieRecordMerger asPreCombiningMode() {
     return HoodiePreCombineAvroRecordMerger.INSTANCE;
   }
+
+  /**
+   * Check if the projection is compatible with the record merger according to the payload class.
+   */
+  @Override
+  public boolean isProjectionCompatible(HoodieTableConfig cfg, TypedProperties properties) {
+    String payloadClass = cfg.getPayloadClass();
+    ValidationUtils.checkArgument(payloadClass != null, "Payload class must be set in HoodieTableConfig for avro record merger");
+    HoodieRecordPayload dummyInstance = HoodieRecordUtils.loadPayload(payloadClass, new Object[] {null, 0}, GenericRecord.class, Comparable.class);
+    return dummyInstance.enableProjectionPushDown();
+  }
+
+  /**
+   * Get the mandatory fields for merging according to the payload class.
+   */
+  @Override
+  public String[] getMandatoryFieldsForMerging(Schema dataSchema, HoodieTableConfig cfg, TypedProperties properties) {
+    String payloadClass = cfg.getPayloadClass();
+    ValidationUtils.checkArgument(payloadClass != null, "Payload class must be set in HoodieTableConfig for avro record merger");
+    HoodieRecordPayload dummyInstance = HoodieRecordUtils.loadPayload(payloadClass, new Object[] {null, 0}, GenericRecord.class, Comparable.class);
+    String[] specifiedMandatoryFields = dummyInstance.mandatoryFields();
+    String[] commonMandatoryFields = HoodieRecordMerger.super.getMandatoryFieldsForMerging(dataSchema, cfg, properties);
+    List<String> allNeedMandatoryFields = Arrays.asList(commonMandatoryFields);
+    Arrays.stream(specifiedMandatoryFields).filter(f -> !allNeedMandatoryFields.contains(f)).forEach(allNeedMandatoryFields::add);
+    return allNeedMandatoryFields.toArray(new String[0]);
+  }
+
 }

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordMerger.java

@@ -146,7 +146,7 @@ default List<Pair<HoodieRecord, Schema>> fullOuterMerge(HoodieRecord older, Sche
    * If true, mor merging can be done without all columns. The columns required can be configured
    * by overriding getMandatoryFieldsForMerging
    */
-  default boolean isProjectionCompatible() {
+  default boolean isProjectionCompatible(HoodieTableConfig cfg, TypedProperties properties) {
     return false;
   }
 

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordPayload.java

@@ -146,6 +146,24 @@ default Comparable<?> getOrderingValue() {
     return 0;
   }
 
+  /**
+   * This method can be used to enable projection push down for the payload to trim columns.
+   * @return true if projection push down is enabled, false otherwise.
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  default boolean enableProjectionPushDown() {
+    return false;
+  }
+
+  /**
+   * This method can be used to specify the mandatory fields required for merging.
+   * @return the mandatory fields required for merging.
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  default String[] mandatoryFields() {
+    return new String[]{};
+  }
+
   static String getAvroPayloadForMergeMode(RecordMergeMode mergeMode) {
     switch (mergeMode) {
       //TODO: After we have merge mode working for writing, we should have a dummy payload that will throw exception when used

hudi-common/src/main/java/org/apache/hudi/common/model/PartialUpdateAvroPayload.java

@@ -282,4 +282,10 @@ private static boolean isRecordNewer(Comparable orderingVal, IndexedRecord recor
     }
     return false;
   }
+
+  @Override
+  public boolean enableProjectionPushDown() {
+    return true;
+  }
+
 }

hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReaderSchemaHandler.java

@@ -156,7 +156,7 @@ private Schema generateRequiredSchema() {
     }
 
     if (hoodieTableConfig.getRecordMergeMode() == RecordMergeMode.CUSTOM) {
-      if (!recordMerger.get().isProjectionCompatible()) {
+      if (!recordMerger.get().isProjectionCompatible(hoodieTableConfig, properties)) {
         return dataSchema;
       }
     }