diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieLogFileWithPartition.java b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieLogFileWithPartition.java new file mode 100644 index 0000000000000..60ec8b051ab50 --- /dev/null +++ b/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; + } +} diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java index 88da6aa1f0669..ffe16896d95f4 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java +++ b/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 partitionFields; + // Names of partition field + private Option partitionValues; + private boolean dropPartitions = false; + protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List logFilePaths, Schema readerSchema, String latestInstantTime, boolean readBlocksLazily, boolean reverseReader, @@ -164,7 +170,19 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List instantRange, boolean withOperationField, boolean forceFullScan, - Option partitionName, InternalSchema internalSchema, boolean useScanV2) { + Option partitionName, InternalSchema internalSchema, + boolean useScanV2, Option 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 logFilePaths, + Schema readerSchema, String latestInstantTime, boolean readBlocksLazily, + boolean reverseReader, int bufferSize, Option instantRange, + boolean withOperationField, boolean forceFullScan, + Option partitionName, InternalSchema internalSchema, + boolean useScanV2) { this.readerSchema = readerSchema; this.latestInstantTime = latestInstantTime; this.hoodieTableMetaClient = HoodieTableMetaClient.builder().setConf(fs.getConf()).setBasePath(basePath).build(); @@ -172,6 +190,8 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List 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(); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java index 756e031cea6c7..5261493dff2ad 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java @@ -107,6 +107,33 @@ protected HoodieMergedLogRecordScanner(FileSystem fs, String basePath, List logFilePaths, Schema readerSchema, + String latestInstantTime, Long maxMemorySizeInBytes, boolean readBlocksLazily, + boolean reverseReader, int bufferSize, String spillableMapBasePath, + Option instantRange, + ExternalSpillableMap.DiskMapType diskMapType, + boolean isBitCaskDiskMapCompressionEnabled, + boolean withOperationField, boolean forceFullScan, + Option partitionName, InternalSchema internalSchema, + boolean useScanV2, Option 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)); } } } diff --git a/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogRecordReader.java b/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogRecordReader.java new file mode 100644 index 0000000000000..44d38505b6eb1 --- /dev/null +++ b/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogRecordReader.java @@ -0,0 +1,260 @@ +/* + * 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.functional; + +import org.apache.hudi.avro.AvroSchemaUtils; +import org.apache.hudi.avro.HoodieAvroUtils; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieAvroPayload; +import org.apache.hudi.common.model.HoodieLogFile; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.log.HoodieLogFormat; +import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner; +import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock; +import org.apache.hudi.common.table.log.block.HoodieDataBlock; +import org.apache.hudi.common.table.log.block.HoodieLogBlock; +import org.apache.hudi.common.testutils.FileCreateUtils; +import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.common.testutils.SchemaTestUtil; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.generic.IndexedRecord; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.apache.hudi.common.testutils.SchemaTestUtil.getSimpleSchema; +import static org.apache.hudi.common.util.collection.ExternalSpillableMap.DiskMapType.BITCASK; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TestHoodieLogRecordReader { + + private FileSystem fs; + private Path tmpDir; + private String basePath; + + @BeforeEach + public void setUp() throws Exception { + this.tmpDir = new Path("/tmp/" + UUID.randomUUID()); + this.basePath = tmpDir.toString(); + + fs = new Path(basePath).getFileSystem(new Configuration()); + } + + @AfterEach + public void tearDown() throws IOException { + fs.delete(this.tmpDir, true); + } + + @Test + public void testLogRecordReaderExtractPartitionFromPath() throws IOException, URISyntaxException, InterruptedException { + String partitionValue1 = "str_partition"; + int partitionValue2 = 100; + String partitionField1 = "partition1"; + String partitionField2 = "partition2"; + Path partitionPath = new Path(tmpDir.toString(), partitionValue1 + "/" + partitionValue2); + Object[] partitionValues = {"str_partition", 100}; + + Properties properties = new Properties(); + properties.setProperty(HoodieTableConfig.DROP_PARTITION_COLUMNS.key(), "true"); + properties.setProperty(HoodieTableConfig.PARTITION_FIELDS.key(), partitionField1 + "," + partitionField2); + HoodieTestUtils.init(fs.getConf(), basePath, HoodieTableType.MERGE_ON_READ, properties); + + Schema schema = HoodieAvroUtils.addMetadataFields(getSimpleSchema()); + HoodieLogFormat.Writer writer = + HoodieLogFormat.newWriterBuilder().onParentPath(partitionPath).withFileExtension(HoodieLogFile.DELTA_EXTENSION) + .withFileId("test-fileid1").overBaseCommit("100").withFs(fs).withSizeThreshold(500).build(); + // Write 1 + List records1 = SchemaTestUtil.generateHoodieTestRecords(0, 100); + List copyOfRecords1 = records1.stream() + .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record, schema)).collect(Collectors.toList()); + + Map header = new HashMap<>(); + header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100"); + header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString()); + HoodieDataBlock dataBlock = new HoodieAvroDataBlock(records1, header, HoodieRecord.RECORD_KEY_METADATA_FIELD); + writer.appendBlock(dataBlock); + + // Write 2 + List records2 = SchemaTestUtil.generateHoodieTestRecords(0, 100); + List copyOfRecords2 = records2.stream() + .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record, schema)).collect(Collectors.toList()); + header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString()); + dataBlock = new HoodieAvroDataBlock(records2, header, HoodieRecord.RECORD_KEY_METADATA_FIELD); + writer.appendBlock(dataBlock); + writer.close(); + + List allLogFiles = + FSUtils.getAllLogFiles(fs, partitionPath, "test-fileid1", HoodieLogFile.DELTA_EXTENSION, "100") + .map(s -> s.getPath().toString()).collect(Collectors.toList()); + + FileCreateUtils.createDeltaCommit(basePath, "100", fs); + + // generate reader schema + + List newFields = new ArrayList<>(); + newFields.add(new Schema.Field(partitionField1, Schema.create(Schema.Type.STRING), null, "")); + newFields.add(new Schema.Field(partitionField2, Schema.create(Schema.Type.INT), null, 0)); + Schema readerSchema = AvroSchemaUtils.appendFieldsToSchema(schema, newFields); + + HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder() + .withFileSystem(fs) + .withBasePath(basePath) + .withLogFilePaths(allLogFiles) + .withReaderSchema(readerSchema) + .withLatestInstantTime("102") + .withMaxMemorySizeInBytes(10240L) + .withReverseReader(false) + .withBufferSize(4096) + .withDiskMapType(BITCASK) + .withPartition(partitionValue1 + "/" + partitionValue2) + .withPartitionValues(partitionValues) + .build(); + assertEquals(200, scanner.getTotalLogRecords()); + Set readKeys = new HashSet<>(200); + scanner.forEach(s -> { + readKeys.add(s.getKey().getRecordKey()); + try { + IndexedRecord record = ((HoodieAvroPayload)s.getData()).getInsertValue(readerSchema, new Properties()).get(); + String partition1 = record.get(record.getSchema().getField(partitionField1).pos()).toString(); + assertEquals(partitionValue1, partition1); + + int partition2 = Integer.parseInt(record.get(record.getSchema().getField(partitionField2).pos()).toString()); + assertEquals(partitionValue2, partition2); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + + assertEquals(200, readKeys.size(), "Stream collect should return all 200 records"); + copyOfRecords1.addAll(copyOfRecords2); + Set originalKeys = + copyOfRecords1.stream().map(s -> ((GenericRecord) s).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString()) + .collect(Collectors.toSet()); + assertEquals(originalKeys, readKeys, "CompositeAvroLogReader should return 200 records from 2 versions"); + } + + @Test + public void testLogRecordReaderExtractHiveStylePartitionFromPath() throws IOException, URISyntaxException, InterruptedException { + String partitionValue1 = "str_partition"; + int partitionValue2 = 100; + String partitionField1 = "partition1"; + String partitionField2 = "partition2"; + Object[] partitionValues = {"str_partition", 100}; + + String hiveStylePartitionValue = partitionField1 + "=" + partitionValue1 + "/" + partitionField2 + "=" + partitionValue2; + Path partitionPath = new Path(tmpDir.toString(), hiveStylePartitionValue); + + Properties properties = new Properties(); + properties.setProperty(HoodieTableConfig.DROP_PARTITION_COLUMNS.key(), "true"); + properties.setProperty(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE.key(), "true"); + properties.setProperty(HoodieTableConfig.PARTITION_FIELDS.key(), partitionField1 + "," + partitionField2); + HoodieTestUtils.init(fs.getConf(), basePath, HoodieTableType.MERGE_ON_READ, properties); + + Schema schema = HoodieAvroUtils.addMetadataFields(getSimpleSchema()); + HoodieLogFormat.Writer writer = + HoodieLogFormat.newWriterBuilder().onParentPath(partitionPath).withFileExtension(HoodieLogFile.DELTA_EXTENSION) + .withFileId("test-fileid1").overBaseCommit("100").withFs(fs).withSizeThreshold(500).build(); + // Write 1 + List records1 = SchemaTestUtil.generateHoodieTestRecords(0, 100); + List copyOfRecords1 = records1.stream() + .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record, schema)).collect(Collectors.toList()); + + Map header = new HashMap<>(); + header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100"); + header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString()); + HoodieDataBlock dataBlock = new HoodieAvroDataBlock(records1, header, HoodieRecord.RECORD_KEY_METADATA_FIELD); + writer.appendBlock(dataBlock); + + // Write 2 + List records2 = SchemaTestUtil.generateHoodieTestRecords(0, 100); + List copyOfRecords2 = records2.stream() + .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record, schema)).collect(Collectors.toList()); + header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString()); + dataBlock = new HoodieAvroDataBlock(records2, header, HoodieRecord.RECORD_KEY_METADATA_FIELD); + writer.appendBlock(dataBlock); + writer.close(); + + List allLogFiles = + FSUtils.getAllLogFiles(fs, partitionPath, "test-fileid1", HoodieLogFile.DELTA_EXTENSION, "100") + .map(s -> s.getPath().toString()).collect(Collectors.toList()); + + FileCreateUtils.createDeltaCommit(basePath, "100", fs); + + // generate reader schema + List newFields = new ArrayList<>(); + newFields.add(new Schema.Field(partitionField1, Schema.create(Schema.Type.STRING), null, "")); + newFields.add(new Schema.Field(partitionField2, Schema.create(Schema.Type.INT), null, 0)); + Schema readerSchema = AvroSchemaUtils.appendFieldsToSchema(schema, newFields); + + HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder() + .withFileSystem(fs) + .withBasePath(basePath) + .withLogFilePaths(allLogFiles) + .withReaderSchema(readerSchema) + .withLatestInstantTime("102") + .withMaxMemorySizeInBytes(10240L) + .withReverseReader(false) + .withBufferSize(4096) + .withDiskMapType(BITCASK) + .withPartition(hiveStylePartitionValue) + .withPartitionValues(partitionValues) + .build(); + assertEquals(200, scanner.getTotalLogRecords()); + Set readKeys = new HashSet<>(200); + scanner.forEach(s -> { + readKeys.add(s.getKey().getRecordKey()); + try { + IndexedRecord record = ((HoodieAvroPayload)s.getData()).getInsertValue(readerSchema, new Properties()).get(); + String partition1 = record.get(record.getSchema().getField(partitionField1).pos()).toString(); + assertEquals(partitionValue1, partition1); + + int partition2 = Integer.parseInt(record.get(record.getSchema().getField(partitionField2).pos()).toString()); + assertEquals(partitionValue2, partition2); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + + assertEquals(200, readKeys.size(), "Stream collect should return all 200 records"); + copyOfRecords1.addAll(copyOfRecords2); + Set originalKeys = + copyOfRecords1.stream().map(s -> ((GenericRecord) s).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString()) + .collect(Collectors.toSet()); + assertEquals(originalKeys, readKeys, "CompositeAvroLogReader should return 200 records from 2 versions"); + } +} diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/LogFileIterator.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/LogFileIterator.scala index 07a0ce7f239be..dcde00d6f41b1 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/LogFileIterator.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/LogFileIterator.scala @@ -21,7 +21,7 @@ package org.apache.hudi import org.apache.hudi.HoodieBaseRelation.{BaseFileReader, generateUnsafeProjection} import org.apache.hudi.HoodieConversionUtils.{toJavaOption, toScalaOption} import org.apache.hudi.HoodieDataSourceHelper.AvroDeserializerSupport -import org.apache.hudi.common.model.{HoodieLogFile, HoodieRecord, HoodieRecordPayload} +import org.apache.hudi.common.model.{HoodieLogFile, HoodieLogFileWithPartition, HoodieRecord, HoodieRecordPayload} import org.apache.hudi.config.HoodiePayloadConfig import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes import org.apache.hudi.LogFileIterator._ @@ -34,20 +34,16 @@ import org.apache.hudi.hadoop.config.HoodieRealtimeConfig import org.apache.hudi.internal.schema.InternalSchema import org.apache.hudi.metadata.{HoodieBackedTableMetadata, HoodieTableMetadata} import org.apache.hudi.metadata.HoodieTableMetadata.getDataTableBasePathFromMetadataTable - import org.apache.avro.Schema import org.apache.avro.generic.{GenericRecord, GenericRecordBuilder, IndexedRecord} - import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path import org.apache.hadoop.mapred.JobConf - import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.types.StructType import java.io.Closeable import java.util.Properties - import scala.annotation.tailrec import scala.collection.JavaConverters._ import scala.util.Try @@ -300,6 +296,10 @@ object LogFileIterator { if (logFiles.nonEmpty) { logRecordScannerBuilder.withPartition( getRelativePartitionPath(new Path(tableState.tablePath), logFiles.head.getPath.getParent)) + + if (logFiles.head.isInstanceOf[HoodieLogFileWithPartition]) { + logRecordScannerBuilder.withPartitionValues(logFiles.head.asInstanceOf[HoodieLogFileWithPartition].getPartitionFieldAndValues) + } } logRecordScannerBuilder.build() diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala index 1d0d533e5bb81..7d218ec7a6472 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala @@ -25,7 +25,7 @@ import org.apache.hudi.HoodieConversionUtils.toScalaOption import org.apache.hudi.MergeOnReadSnapshotRelation.getFilePath import org.apache.hudi.avro.HoodieAvroUtils import org.apache.hudi.common.fs.FSUtils.getRelativePartitionPath -import org.apache.hudi.common.model.{FileSlice, HoodieLogFile} +import org.apache.hudi.common.model.{FileSlice, HoodieLogFile, HoodieLogFileWithPartition} import org.apache.hudi.common.table.HoodieTableMetaClient import org.apache.hudi.common.table.view.HoodieTableFileSystemView import org.apache.spark.execution.datasources.HoodieInMemoryFileIndex @@ -36,6 +36,7 @@ import org.apache.spark.sql.catalyst.expressions.Expression import org.apache.spark.sql.execution.datasources.PartitionedFile import org.apache.spark.sql.sources.Filter import org.apache.spark.sql.types.StructType +import org.apache.spark.unsafe.types.UTF8String import scala.collection.JavaConverters._ @@ -213,7 +214,24 @@ class MergeOnReadSnapshotRelation(sqlContext: SQLContext, PartitionedFile(getPartitionColumnsAsInternalRow(file.getFileStatus), filePath, 0, file.getFileLen) } - HoodieMergeOnReadFileSplit(partitionedBaseFile, logFiles) + if (shouldExtractPartitionValuesFromPartitionPath && logFiles.nonEmpty) { + val partitionColumnValues = fileIndex.parsePartitionColumnValues(partitionColumns, + getRelativePartitionPath(new Path(basePath), logFiles.head.getPath.getParent)).map( + // "String" type in schema is String + par => if (par.isInstanceOf[UTF8String]) { + String.valueOf(par) + } else { + par + } + ) + + val logFileWithPartition = logFiles.map(log => { + new HoodieLogFileWithPartition(log, partitionColumnValues) + }) + HoodieMergeOnReadFileSplit(partitionedBaseFile, logFileWithPartition) + } else { + HoodieMergeOnReadFileSplit(partitionedBaseFile, logFiles) + } }.toList } } diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala index a9a38f5f82bdd..a329d007a4b42 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala @@ -213,7 +213,7 @@ class SparkHoodieTableFileIndex(spark: SparkSession, } } - protected def parsePartitionColumnValues(partitionColumns: Array[String], partitionPath: String): Array[Object] = { + def parsePartitionColumnValues(partitionColumns: Array[String], partitionPath: String): Array[Object] = { if (partitionColumns.length == 0) { // This is a non-partitioned table Array.empty