[HUDI-5047] Add partition value in HoodieLogRecordReader when hoodie.datasource.write.drop.partition.columns=true

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

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.model;
+
+public class HoodieLogFileWithPartition extends HoodieLogFile {
+
+  private  Object[] partitionFieldAndValues;
+
+  public HoodieLogFileWithPartition(HoodieLogFile logFile, Object[] partitionFieldAndValues) {
+    super(logFile);
+    this.partitionFieldAndValues = partitionFieldAndValues;
+  }
+
+  public Object[] getPartitionFieldAndValues() {
+    return this.partitionFieldAndValues;
+  }
+}

hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java

@@ -151,6 +151,12 @@ public abstract class AbstractHoodieLogRecordReader {
   // Use scanV2 method.
   private boolean useScanV2;
 
+  // Names of partition field
+  private Option<String[]> partitionFields;
+  // Names of partition field
+  private Option<Object[]> partitionValues;
+  private boolean dropPartitions = false;
+
   protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List<String> logFilePaths,
                                           Schema readerSchema,
                                           String latestInstantTime, boolean readBlocksLazily, boolean reverseReader,
@@ -164,14 +170,28 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List<Str
                                           Schema readerSchema, String latestInstantTime, boolean readBlocksLazily,
                                           boolean reverseReader, int bufferSize, Option<InstantRange> instantRange,
                                           boolean withOperationField, boolean forceFullScan,
-                                          Option<String> partitionName, InternalSchema internalSchema, boolean useScanV2) {
+                                          Option<String> partitionName, InternalSchema internalSchema,
+                                          boolean useScanV2, Option<Object[]> partitionValues) {
+    this(fs, basePath, logFilePaths, readerSchema, latestInstantTime, readBlocksLazily, reverseReader, bufferSize,
+        instantRange, withOperationField, forceFullScan, partitionName, internalSchema, useScanV2);
+    this.partitionValues = partitionValues;
+  }
+
+  protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List<String> logFilePaths,
+                                          Schema readerSchema, String latestInstantTime, boolean readBlocksLazily,
+                                          boolean reverseReader, int bufferSize, Option<InstantRange> instantRange,
+                                          boolean withOperationField, boolean forceFullScan,
+                                          Option<String> partitionName, InternalSchema internalSchema,
+                                          boolean useScanV2) {
     this.readerSchema = readerSchema;
     this.latestInstantTime = latestInstantTime;
     this.hoodieTableMetaClient = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(basePath).build();
     // load class from the payload fully qualified class name
     HoodieTableConfig tableConfig = this.hoodieTableMetaClient.getTableConfig();
     this.payloadClassFQN = tableConfig.getPayloadClass();
     this.preCombineField = tableConfig.getPreCombineField();
+    this.dropPartitions = tableConfig.getBooleanOrDefault(HoodieTableConfig.DROP_PARTITION_COLUMNS);
+    this.partitionFields = tableConfig.getPartitionFields();
     if (this.preCombineField != null) {
       this.payloadProps.setProperty(HoodiePayloadProps.PAYLOAD_ORDERING_FIELD_PROP_KEY, this.preCombineField);
     }
@@ -632,6 +652,16 @@ private void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpec
       while (recordIterator.hasNext()) {
         IndexedRecord currentRecord = recordIterator.next();
         IndexedRecord record = schemaOption.isPresent() ? HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get(), Collections.emptyMap()) : currentRecord;
+
+        if (this.dropPartitions && this.partitionFields.isPresent() && this.partitionValues.isPresent()) {
+          Schema schema = record.getSchema();
+          String[] partitionFieldArray = this.partitionFields.get();
+          Object[] partitionValueArray = this.partitionValues.get();
+          for (int i = 0; i < partitionFieldArray.length; i++) {
+            record.put(schema.getField(partitionFieldArray[i]).pos(), partitionValueArray[i]);
+          }
+        }
+
         processNextRecord(createHoodieRecord(record, this.hoodieTableMetaClient.getTableConfig(), this.payloadClassFQN,
             this.preCombineField, this.withOperationField, this.simpleKeyGenFields, this.partitionName));
         totalLogRecords.incrementAndGet();

hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java

@@ -107,6 +107,33 @@ protected HoodieMergedLogRecordScanner(FileSystem fs, String basePath, List<Stri
     }
   }
 
+  @SuppressWarnings("unchecked")
+  protected HoodieMergedLogRecordScanner(FileSystem fs, String basePath, List<String> logFilePaths, Schema readerSchema,
+                                         String latestInstantTime, Long maxMemorySizeInBytes, boolean readBlocksLazily,
+                                         boolean reverseReader, int bufferSize, String spillableMapBasePath,
+                                         Option<InstantRange> instantRange,
+                                         ExternalSpillableMap.DiskMapType diskMapType,
+                                         boolean isBitCaskDiskMapCompressionEnabled,
+                                         boolean withOperationField, boolean forceFullScan,
+                                         Option<String> partitionName, InternalSchema internalSchema,
+                                         boolean useScanV2, Option<Object[]> partitionValues) {
+    super(fs, basePath, logFilePaths, readerSchema, latestInstantTime, readBlocksLazily, reverseReader, bufferSize,
+        instantRange, withOperationField,
+        forceFullScan, partitionName, internalSchema, useScanV2, partitionValues);
+    try {
+      // Store merged records for all versions for this log file, set the in-memory footprint to maxInMemoryMapSize
+      this.records = new ExternalSpillableMap<>(maxMemorySizeInBytes, spillableMapBasePath, new DefaultSizeEstimator(),
+          new HoodieRecordSizeEstimator(readerSchema), diskMapType, isBitCaskDiskMapCompressionEnabled);
+      this.maxMemorySizeInBytes = maxMemorySizeInBytes;
+    } catch (IOException e) {
+      throw new HoodieIOException("IOException when creating ExternalSpillableMap at " + spillableMapBasePath, e);
+    }
+
+    if (forceFullScan) {
+      performScan();
+    }
+  }
+
   protected void performScan() {
     // Do the scan and merge
     timer.startTimer();
@@ -233,6 +260,8 @@ public static class Builder extends AbstractHoodieLogRecordReader.Builder {
     // Use scanV2 method.
     private boolean useScanV2 = false;
 
+    private Object[] partitionValues;
+
     @Override
     public Builder withFileSystem(FileSystem fs) {
       this.fs = fs;
@@ -331,6 +360,11 @@ public Builder withUseScanV2(boolean useScanV2) {
       return this;
     }
 
+    public Builder withPartitionValues(Object[] partitionValues) {
+      this.partitionValues = partitionValues;
+      return this;
+    }
+
     @Override
     public HoodieMergedLogRecordScanner build() {
       if (this.partitionName == null && CollectionUtils.nonEmpty(this.logFilePaths)) {
@@ -340,7 +374,7 @@ public HoodieMergedLogRecordScanner build() {
           latestInstantTime, maxMemorySizeInBytes, readBlocksLazily, reverseReader,
           bufferSize, spillableMapBasePath, instantRange,
           diskMapType, isBitCaskDiskMapCompressionEnabled, withOperationField, true,
-          Option.ofNullable(partitionName), internalSchema, useScanV2);
+          Option.ofNullable(partitionName), internalSchema, useScanV2, Option.ofNullable(partitionValues));
     }
   }
 }