From 8b09437530fce0720b9fc35a7f27a8f45cbb2eab Mon Sep 17 00:00:00 2001 From: xicm Date: Tue, 18 Oct 2022 16:28:47 +0800 Subject: [PATCH 01/10] [HUDI-5047] Add partition value in HoodieLogRecordReader when hoodie.datasource.write.drop.partition.columns=true --- .../log/AbstractHoodieLogRecordReader.java | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) 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 5bfb395dbc54c..60356243c9d9b 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 @@ -141,6 +141,11 @@ public abstract class AbstractHoodieLogRecordReader { // Populate meta fields for the records private boolean populateMetaFields = true; + private Option partitionFields; + private Option partitionValues; + private boolean dropPartitions = false; + private boolean hiveStylePartition = false; + protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List logFilePaths, Schema readerSchema, String latestInstantTime, boolean readBlocksLazily, boolean reverseReader, @@ -162,6 +167,9 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List getPartitionValues() { + if (!this.partitionName.isPresent()) { + return Option.empty(); + } + + String[] partitionValues = this.partitionName.get().split("/"); + if (this.hiveStylePartition) { + return Option.of(Arrays.stream(partitionValues) + .map(partition -> partition.split("=")) + .filter(partition -> partition.length ==2) + .map(partition -> partition[1]) + .toArray(String[]::new)); + } else { + return Option.of(partitionValues); + } } protected String getKeyField() { @@ -387,6 +413,19 @@ private void processDataBlock(HoodieDataBlock dataBlock, Option keySpec while (recordIterator.hasNext()) { IndexedRecord currentRecord = recordIterator.next(); IndexedRecord record = schemaOption.isPresent() ? HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get(), Collections.emptyMap()) : currentRecord; + + Schema schema = record.getSchema(); + if (this.dropPartitions && this.partitionFields.isPresent() && this.partitionValues.isPresent()) { + String[] partitionFields = this.partitionFields.get(); + String[] partitionValues = this.partitionValues.get(); + + if (partitionFields.length == partitionValues.length) { + for (int i = 0; i < partitionValues.length; i++) { + record.put(schema.getField(partitionFields[i]).pos(), partitionValues[i]); + } + } + } + processNextRecord(createHoodieRecord(record, this.hoodieTableMetaClient.getTableConfig(), this.payloadClassFQN, this.preCombineField, this.withOperationField, this.simpleKeyGenFields, this.partitionName)); totalLogRecords.incrementAndGet(); From 6c3f427967e17da7ed960a8dc7944f2f3d39fe76 Mon Sep 17 00:00:00 2001 From: xicm Date: Wed, 19 Oct 2022 15:26:53 +0800 Subject: [PATCH 02/10] add ut --- .../log/AbstractHoodieLogRecordReader.java | 2 +- .../functional/TestHoodieLogRecordReader.java | 180 ++++++++++++++++++ 2 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogRecordReader.java 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 60356243c9d9b..3d6cd8218b8c4 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 @@ -204,7 +204,7 @@ private Option getPartitionValues() { if (this.hiveStylePartition) { return Option.of(Arrays.stream(partitionValues) .map(partition -> partition.split("=")) - .filter(partition -> partition.length ==2) + .filter(partition -> partition.length == 2) .map(partition -> partition[1]) .toArray(String[]::new)); } else { 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..65801163b88bf --- /dev/null +++ b/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogRecordReader.java @@ -0,0 +1,180 @@ +/* + * 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.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.HoodieCommonTestHarness; +import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.common.testutils.SchemaTestUtil; +import org.apache.hudi.common.testutils.minicluster.MiniClusterUtil; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.generic.IndexedRecord; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +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.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; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestHoodieLogRecordReader extends HoodieCommonTestHarness { + + private FileSystem fs; + private Path partitionPath; + private String partitionValue = "my_partition"; + private String partitionName = "partition"; + + @BeforeAll + public static void setUpClass() throws IOException, InterruptedException { + MiniClusterUtil.setUp(); + } + + @AfterAll + public static void tearDownClass() { + MiniClusterUtil.shutdown(); + } + + @BeforeEach + public void setUp() throws IOException, InterruptedException { + this.fs = MiniClusterUtil.fileSystem; + + assertTrue(fs.mkdirs(new Path(tempDir.toAbsolutePath().toString()))); + this.partitionPath = new Path(tempDir.toAbsolutePath().toString(), partitionValue); + this.basePath = tempDir.getParent().toString(); + } + + @AfterEach + public void tearDown() throws IOException { + fs.delete(partitionPath.getParent(), true); + } + + @Test + public void testLogRecordReaderExtractPartitionFromPath() throws IOException, URISyntaxException, InterruptedException { + + Properties properties = new Properties(); + properties.setProperty(HoodieTableConfig.DROP_PARTITION_COLUMNS.key(), "true"); + properties.setProperty(HoodieTableConfig.PARTITION_FIELDS.key(), partitionName); + HoodieTestUtils.init(MiniClusterUtil.configuration, 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 fields = new ArrayList<>(); + fields.add(new Schema.Field(partitionName, Schema.create(Schema.Type.STRING), null, "")); + for (Schema.Field field : schema.getFields()) { + Schema.Field newField = new Schema.Field(field.name(), field.schema(), field.doc(), field.defaultVal()); + for (Map.Entry prop : field.getObjectProps().entrySet()) { + newField.addProp(prop.getKey(), prop.getValue()); + } + fields.add(newField); + } + Schema readerSchema = Schema.createRecord(schema.getName(), schema.getDoc(), schema.getNamespace(), false); + readerSchema.setFields(fields); + + HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder() + .withFileSystem(fs) + .withBasePath(basePath) + .withLogFilePaths(allLogFiles) + .withReaderSchema(readerSchema) + .withLatestInstantTime("102") + .withMaxMemorySizeInBytes(10240L) + .withReverseReader(false) + .withBufferSize(4096) + .withDiskMapType(BITCASK) + .withPartition(partitionValue) + .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 partition = record.get(record.getSchema().getField(partitionName).pos()).toString(); + assertEquals(partitionValue, partition); + } 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"); + } +} From 970ad798c5690ef0ddcd8d32e480f49959e16de9 Mon Sep 17 00:00:00 2001 From: xicm Date: Wed, 19 Oct 2022 22:15:30 +0800 Subject: [PATCH 03/10] fix ut --- .../functional/TestHoodieLogRecordReader.java | 41 +++++++++++++------ 1 file changed, 29 insertions(+), 12 deletions(-) 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 index 65801163b88bf..44df54806359b 100644 --- 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 @@ -31,22 +31,25 @@ 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.HoodieCommonTestHarness; import org.apache.hudi.common.testutils.HoodieTestUtils; import org.apache.hudi.common.testutils.SchemaTestUtil; -import org.apache.hudi.common.testutils.minicluster.MiniClusterUtil; 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.FileUtil; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.MiniDFSCluster; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.File; import java.io.IOException; import java.net.URISyntaxException; import java.util.ArrayList; @@ -56,42 +59,55 @@ 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; -import static org.junit.jupiter.api.Assertions.assertTrue; -public class TestHoodieLogRecordReader extends HoodieCommonTestHarness { +public class TestHoodieLogRecordReader { + + private static File dfsBaseDir; + private static MiniDFSCluster cluster; private FileSystem fs; + private String basePath; private Path partitionPath; private String partitionValue = "my_partition"; private String partitionName = "partition"; @BeforeAll public static void setUpClass() throws IOException, InterruptedException { - MiniClusterUtil.setUp(); + dfsBaseDir = new File("/tmp/" + UUID.randomUUID().toString()); + FileUtil.fullyDelete(dfsBaseDir); + // Append is not supported in LocalFileSystem. HDFS needs to be setup. + Configuration conf = new Configuration(); + // lower heartbeat interval for fast recognition of DN + conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, dfsBaseDir.getAbsolutePath()); + conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, 1000); + conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1); + conf.setInt(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, 3000); + cluster = new MiniDFSCluster.Builder(conf).checkExitOnShutdown(true).numDataNodes(4).build(); } @AfterAll public static void tearDownClass() { - MiniClusterUtil.shutdown(); + cluster.shutdown(true); + FileUtil.fullyDelete(dfsBaseDir); } @BeforeEach public void setUp() throws IOException, InterruptedException { - this.fs = MiniClusterUtil.fileSystem; + this.fs = cluster.getFileSystem(); - assertTrue(fs.mkdirs(new Path(tempDir.toAbsolutePath().toString()))); - this.partitionPath = new Path(tempDir.toAbsolutePath().toString(), partitionValue); - this.basePath = tempDir.getParent().toString(); + this.partitionPath = new Path(dfsBaseDir.toString(), partitionValue); + this.basePath = dfsBaseDir.toString(); } @AfterEach public void tearDown() throws IOException { - fs.delete(partitionPath.getParent(), true); + fs.delete(partitionPath, true); } @Test @@ -100,7 +116,7 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR Properties properties = new Properties(); properties.setProperty(HoodieTableConfig.DROP_PARTITION_COLUMNS.key(), "true"); properties.setProperty(HoodieTableConfig.PARTITION_FIELDS.key(), partitionName); - HoodieTestUtils.init(MiniClusterUtil.configuration, basePath, HoodieTableType.MERGE_ON_READ, properties); + HoodieTestUtils.init(fs.getConf(), basePath, HoodieTableType.MERGE_ON_READ, properties); Schema schema = HoodieAvroUtils.addMetadataFields(getSimpleSchema()); HoodieLogFormat.Writer writer = @@ -164,6 +180,7 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR try { IndexedRecord record = ((HoodieAvroPayload)s.getData()).getInsertValue(readerSchema, new Properties()).get(); String partition = record.get(record.getSchema().getField(partitionName).pos()).toString(); + // The value of field "partition" should be "my_partition" passed to scanner assertEquals(partitionValue, partition); } catch (IOException e) { throw new RuntimeException(e); From 088cfee9a091af9f0327d4b432fcb3aa9a6e22ca Mon Sep 17 00:00:00 2001 From: xicm Date: Thu, 20 Oct 2022 16:34:18 +0800 Subject: [PATCH 04/10] fix flaky test --- .../apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala index f5435e09adfc2..681a2ce094bbf 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala +++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala @@ -593,8 +593,8 @@ class TestCDCDataFrameSuite extends HoodieCDCTestBase { // it will write out >1 cdc log files due to rollover. assert(cdcLogFiles2.size > 1) // with a small value for 'hoodie.logfile.data.block.max.size', - // it will write out >1 cdc data blocks in one single cdc log file. - assert(getCDCBlocks(cdcLogFiles2.head, cdcSchema).size > 1) + // it will write out >= 1 cdc data blocks in one single cdc log file. + assert(getCDCBlocks(cdcLogFiles2.head, cdcSchema).size >= 1) // check cdc data val cdcDataFromCDCLogFile2 = cdcLogFiles2.flatMap(readCDCLogFile(_, cdcSchema)) From 74787e65d82a88ece5a1a9ece86a207d9ca2125e Mon Sep 17 00:00:00 2001 From: xicm Date: Tue, 25 Oct 2022 16:53:21 +0800 Subject: [PATCH 05/10] convert partition value to schema specified type --- .../org/apache/hudi/avro/HoodieAvroUtils.java | 23 +++ .../log/AbstractHoodieLogRecordReader.java | 58 ++++--- .../functional/TestHoodieLogRecordReader.java | 163 ++++++++++++------ 3 files changed, 168 insertions(+), 76 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java index a352e86b96c73..a841d0e4e6d69 100644 --- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java @@ -427,6 +427,7 @@ private static void copyOldValueOrSetDefault(GenericRecord oldRecord, GenericRec Schema oldSchema = oldRecord.getSchema(); Object fieldValue = oldSchema.getField(field.name()) == null ? null : oldRecord.get(field.name()); + if (fieldValue != null) { // In case field's value is a nested record, we have to rewrite it as well Object newFieldValue; @@ -718,6 +719,17 @@ public static Object getRecordColumnValues(HoodieRecord partitionFields; - private Option partitionValues; + // Partition field name and value + private Map partitionNameAndValues = new HashMap<>(); private boolean dropPartitions = false; private boolean hiveStylePartition = false; @@ -203,24 +205,36 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List getPartitionValues() { - if (!this.partitionName.isPresent()) { - return Option.empty(); + private static Map getPartitionValues( + Option partitionFields, + Option partitionName, + boolean hiveStylePartition + ) { + Map result = new HashMap<>(); + if (!partitionName.isPresent() || !partitionFields.isPresent()) { + return result; } - String[] partitionValues = this.partitionName.get().split("/"); - if (this.hiveStylePartition) { - return Option.of(Arrays.stream(partitionValues) - .map(partition -> partition.split("=")) - .filter(partition -> partition.length == 2) - .map(partition -> partition[1]) - .toArray(String[]::new)); - } else { - return Option.of(partitionValues); + String[] partitionField = partitionFields.get(); + String[] partitionValue = partitionName.get().split("/"); + if (partitionField.length != partitionValue.length) { + throw new HoodieException("The length of partitionField and partitionValue must be the same. " + + "partitionField = " + String.join(",", partitionField) + + " ,partitionValue = " + String.join(",", partitionValue)); + } + for (int i = 0; i < partitionValue.length; i++) { + if (hiveStylePartition) { + result.put(partitionField[i], partitionValue[i].substring(partitionValue[i].indexOf('=') + 1)); + } else { + result.put(partitionField[i], partitionValue[i]); + } } + + return result; } protected String getKeyField() { @@ -659,16 +673,12 @@ private void processDataBlock(HoodieDataBlock dataBlock, Option keySpec IndexedRecord currentRecord = recordIterator.next(); IndexedRecord record = schemaOption.isPresent() ? HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get(), Collections.emptyMap()) : currentRecord; - Schema schema = record.getSchema(); - if (this.dropPartitions && this.partitionFields.isPresent() && this.partitionValues.isPresent()) { - String[] partitionFields = this.partitionFields.get(); - String[] partitionValues = this.partitionValues.get(); - - if (partitionFields.length == partitionValues.length) { - for (int i = 0; i < partitionValues.length; i++) { - record.put(schema.getField(partitionFields[i]).pos(), partitionValues[i]); - } - } + if (this.dropPartitions) { + Schema schema = record.getSchema(); + this.partitionNameAndValues.forEach((k, v) -> { + Object partitionValue = HoodieAvroUtils.convertStringToSchemaSpecifiedType(v, schema.getField(k).schema()); + record.put(schema.getField(k).pos(), partitionValue); + }); } processNextRecord(createHoodieRecord(record, this.hoodieTableMetaClient.getTableConfig(), this.payloadClassFQN, 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 index 44df54806359b..cf3e107115d64 100644 --- 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 @@ -18,6 +18,7 @@ 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; @@ -39,17 +40,11 @@ import org.apache.avro.generic.IndexedRecord; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.FileUtil; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hdfs.DFSConfigKeys; -import org.apache.hadoop.hdfs.MiniDFSCluster; -import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import java.io.File; import java.io.IOException; import java.net.URISyntaxException; import java.util.ArrayList; @@ -68,54 +63,34 @@ public class TestHoodieLogRecordReader { - private static File dfsBaseDir; - private static MiniDFSCluster cluster; - private FileSystem fs; + private Path tmpDir; private String basePath; - private Path partitionPath; - private String partitionValue = "my_partition"; - private String partitionName = "partition"; - - @BeforeAll - public static void setUpClass() throws IOException, InterruptedException { - dfsBaseDir = new File("/tmp/" + UUID.randomUUID().toString()); - FileUtil.fullyDelete(dfsBaseDir); - // Append is not supported in LocalFileSystem. HDFS needs to be setup. - Configuration conf = new Configuration(); - // lower heartbeat interval for fast recognition of DN - conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, dfsBaseDir.getAbsolutePath()); - conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, 1000); - conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1); - conf.setInt(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, 3000); - cluster = new MiniDFSCluster.Builder(conf).checkExitOnShutdown(true).numDataNodes(4).build(); - } - - @AfterAll - public static void tearDownClass() { - cluster.shutdown(true); - FileUtil.fullyDelete(dfsBaseDir); - } @BeforeEach - public void setUp() throws IOException, InterruptedException { - this.fs = cluster.getFileSystem(); + public void setUp() throws Exception { + this.tmpDir = new Path("/tmp/" + UUID.randomUUID()); + this.basePath = tmpDir.toString(); - this.partitionPath = new Path(dfsBaseDir.toString(), partitionValue); - this.basePath = dfsBaseDir.toString(); + fs = new Path(basePath).getFileSystem(new Configuration()); } @AfterEach public void tearDown() throws IOException { - fs.delete(partitionPath, true); + 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); Properties properties = new Properties(); properties.setProperty(HoodieTableConfig.DROP_PARTITION_COLUMNS.key(), "true"); - properties.setProperty(HoodieTableConfig.PARTITION_FIELDS.key(), partitionName); + properties.setProperty(HoodieTableConfig.PARTITION_FIELDS.key(), partitionField1 + "," + partitionField2); HoodieTestUtils.init(fs.getConf(), basePath, HoodieTableType.MERGE_ON_READ, properties); Schema schema = HoodieAvroUtils.addMetadataFields(getSimpleSchema()); @@ -149,17 +124,99 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR FileCreateUtils.createDeltaCommit(basePath, "100", fs); // generate reader schema - List fields = new ArrayList<>(); - fields.add(new Schema.Field(partitionName, Schema.create(Schema.Type.STRING), null, "")); - for (Schema.Field field : schema.getFields()) { - Schema.Field newField = new Schema.Field(field.name(), field.schema(), field.doc(), field.defaultVal()); - for (Map.Entry prop : field.getObjectProps().entrySet()) { - newField.addProp(prop.getKey(), prop.getValue()); + + 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) + .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); } - fields.add(newField); - } - Schema readerSchema = Schema.createRecord(schema.getName(), schema.getDoc(), schema.getNamespace(), false); - readerSchema.setFields(fields); + }); + + 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"; + + 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) @@ -171,7 +228,7 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR .withReverseReader(false) .withBufferSize(4096) .withDiskMapType(BITCASK) - .withPartition(partitionValue) + .withPartition(hiveStylePartitionValue) .build(); assertEquals(200, scanner.getTotalLogRecords()); Set readKeys = new HashSet<>(200); @@ -179,9 +236,11 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR readKeys.add(s.getKey().getRecordKey()); try { IndexedRecord record = ((HoodieAvroPayload)s.getData()).getInsertValue(readerSchema, new Properties()).get(); - String partition = record.get(record.getSchema().getField(partitionName).pos()).toString(); - // The value of field "partition" should be "my_partition" passed to scanner - assertEquals(partitionValue, partition); + 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); } From 240b1a43e85fb85791e5c9bcb87aa9e95201c1cb Mon Sep 17 00:00:00 2001 From: xicm Date: Wed, 26 Oct 2022 10:17:29 +0800 Subject: [PATCH 06/10] remove blank line --- .../src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java | 1 - 1 file changed, 1 deletion(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java index a841d0e4e6d69..595cb6a111365 100644 --- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java @@ -427,7 +427,6 @@ private static void copyOldValueOrSetDefault(GenericRecord oldRecord, GenericRec Schema oldSchema = oldRecord.getSchema(); Object fieldValue = oldSchema.getField(field.name()) == null ? null : oldRecord.get(field.name()); - if (fieldValue != null) { // In case field's value is a nested record, we have to rewrite it as well Object newFieldValue; From 1d34a26c34ced9fead6af9725e68d8bb86cc934e Mon Sep 17 00:00:00 2001 From: xicm Date: Wed, 26 Oct 2022 14:06:49 +0800 Subject: [PATCH 07/10] convert UNION to its real type --- .../main/java/org/apache/hudi/avro/HoodieAvroUtils.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java index 595cb6a111365..a8ac87d24b0d2 100644 --- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java @@ -719,14 +719,16 @@ public static Object getRecordColumnValues(HoodieRecord Date: Mon, 31 Oct 2022 18:04:55 +0800 Subject: [PATCH 08/10] get partition values with SparkParsePartitionUtil --- .../org/apache/hudi/avro/HoodieAvroUtils.java | 24 -------- .../model/HoodieLogFileWithPartition.java | 33 ++++++++++ .../log/AbstractHoodieLogRecordReader.java | 60 +++++++------------ .../log/HoodieMergedLogRecordScanner.java | 36 ++++++++++- .../functional/TestHoodieLogRecordReader.java | 4 ++ .../org/apache/hudi/LogFileIterator.scala | 10 ++-- .../hudi/MergeOnReadSnapshotRelation.scala | 22 ++++++- .../hudi/SparkHoodieTableFileIndex.scala | 2 +- 8 files changed, 118 insertions(+), 73 deletions(-) create mode 100644 hudi-common/src/main/java/org/apache/hudi/common/model/HoodieLogFileWithPartition.java diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java index a8ac87d24b0d2..a352e86b96c73 100644 --- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java @@ -718,19 +718,6 @@ public static Object getRecordColumnValues(HoodieRecord partitionFields; - // Partition field name and value - private Map partitionNameAndValues = new HashMap<>(); + // Names of partition field + private Object[] partitionValues; private boolean dropPartitions = false; - private boolean hiveStylePartition = false; protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List logFilePaths, Schema readerSchema, @@ -171,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, Object[] partitionValues) { + this(fs, basePath, logFilePaths, readerSchema, latestInstantTime, readBlocksLazily, reverseReader, bufferSize, + instantRange, withOperationField, true, Option.empty(), InternalSchema.getEmptyInternalSchema(), false); + 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(); @@ -181,7 +192,6 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List getPartitionValues( - Option partitionFields, - Option partitionName, - boolean hiveStylePartition - ) { - Map result = new HashMap<>(); - if (!partitionName.isPresent() || !partitionFields.isPresent()) { - return result; - } - - String[] partitionField = partitionFields.get(); - String[] partitionValue = partitionName.get().split("/"); - if (partitionField.length != partitionValue.length) { - throw new HoodieException("The length of partitionField and partitionValue must be the same. " - + "partitionField = " + String.join(",", partitionField) - + " ,partitionValue = " + String.join(",", partitionValue)); - } - for (int i = 0; i < partitionValue.length; i++) { - if (hiveStylePartition) { - result.put(partitionField[i], partitionValue[i].substring(partitionValue[i].indexOf('=') + 1)); - } else { - result.put(partitionField[i], partitionValue[i]); - } - } - - return result; } protected String getKeyField() { @@ -673,12 +653,12 @@ private void processDataBlock(HoodieDataBlock dataBlock, Option keySpec IndexedRecord currentRecord = recordIterator.next(); IndexedRecord record = schemaOption.isPresent() ? HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get(), Collections.emptyMap()) : currentRecord; - if (this.dropPartitions) { + if (this.dropPartitions && this.partitionFields.isPresent()) { Schema schema = record.getSchema(); - this.partitionNameAndValues.forEach((k, v) -> { - Object partitionValue = HoodieAvroUtils.convertStringToSchemaSpecifiedType(v, schema.getField(k).schema()); - record.put(schema.getField(k).pos(), partitionValue); - }); + String[] partitionFieldArray = this.partitionFields.get(); + for (int i = 0; i < partitionFieldArray.length; i++) { + record.put(schema.getField(partitionFieldArray[i]).pos(), this.partitionValues[i]); + } } processNextRecord(createHoodieRecord(record, this.hoodieTableMetaClient.getTableConfig(), this.payloadClassFQN, 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..6b0b7175166ed 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, 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, 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 index cf3e107115d64..44d38505b6eb1 100644 --- 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 @@ -87,6 +87,7 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR 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"); @@ -141,6 +142,7 @@ public void testLogRecordReaderExtractPartitionFromPath() throws IOException, UR .withBufferSize(4096) .withDiskMapType(BITCASK) .withPartition(partitionValue1 + "/" + partitionValue2) + .withPartitionValues(partitionValues) .build(); assertEquals(200, scanner.getTotalLogRecords()); Set readKeys = new HashSet<>(200); @@ -172,6 +174,7 @@ public void testLogRecordReaderExtractHiveStylePartitionFromPath() throws IOExce 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); @@ -229,6 +232,7 @@ public void testLogRecordReaderExtractHiveStylePartitionFromPath() throws IOExce .withBufferSize(4096) .withDiskMapType(BITCASK) .withPartition(hiveStylePartitionValue) + .withPartitionValues(partitionValues) .build(); assertEquals(200, scanner.getTotalLogRecords()); Set readKeys = new HashSet<>(200); 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..efa2f9e043621 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) { + 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 From 599f8a287f7780cb543e491dbd963bf5dea50db6 Mon Sep 17 00:00:00 2001 From: xicm Date: Tue, 1 Nov 2022 10:44:34 +0800 Subject: [PATCH 09/10] fix params --- .../hudi/common/table/log/AbstractHoodieLogRecordReader.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 f2a22e2126685..5510addc86122 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 @@ -173,7 +173,7 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List partitionName, InternalSchema internalSchema, boolean useScanV2, Object[] partitionValues) { this(fs, basePath, logFilePaths, readerSchema, latestInstantTime, readBlocksLazily, reverseReader, bufferSize, - instantRange, withOperationField, true, Option.empty(), InternalSchema.getEmptyInternalSchema(), false); + instantRange, withOperationField, forceFullScan, partitionName, internalSchema, useScanV2); this.partitionValues = partitionValues; } From cd21d9804de19bc035e6bd8f1665bcd88ffa342f Mon Sep 17 00:00:00 2001 From: xicm Date: Tue, 1 Nov 2022 18:05:50 +0800 Subject: [PATCH 10/10] fix param type --- .../common/table/log/AbstractHoodieLogRecordReader.java | 9 +++++---- .../common/table/log/HoodieMergedLogRecordScanner.java | 4 ++-- .../org/apache/hudi/MergeOnReadSnapshotRelation.scala | 4 ++-- 3 files changed, 9 insertions(+), 8 deletions(-) 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 5510addc86122..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 @@ -154,7 +154,7 @@ public abstract class AbstractHoodieLogRecordReader { // Names of partition field private Option partitionFields; // Names of partition field - private Object[] partitionValues; + private Option partitionValues; private boolean dropPartitions = false; protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List logFilePaths, @@ -171,7 +171,7 @@ protected AbstractHoodieLogRecordReader(FileSystem fs, String basePath, List instantRange, boolean withOperationField, boolean forceFullScan, Option partitionName, InternalSchema internalSchema, - boolean useScanV2, Object[] partitionValues) { + boolean useScanV2, Option partitionValues) { this(fs, basePath, logFilePaths, readerSchema, latestInstantTime, readBlocksLazily, reverseReader, bufferSize, instantRange, withOperationField, forceFullScan, partitionName, internalSchema, useScanV2); this.partitionValues = partitionValues; @@ -653,11 +653,12 @@ private void processDataBlock(HoodieDataBlock dataBlock, Option keySpec IndexedRecord currentRecord = recordIterator.next(); IndexedRecord record = schemaOption.isPresent() ? HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get(), Collections.emptyMap()) : currentRecord; - if (this.dropPartitions && this.partitionFields.isPresent()) { + 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(), this.partitionValues[i]); + record.put(schema.getField(partitionFieldArray[i]).pos(), partitionValueArray[i]); } } 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 6b0b7175166ed..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 @@ -116,7 +116,7 @@ protected HoodieMergedLogRecordScanner(FileSystem fs, String basePath, List partitionName, InternalSchema internalSchema, - boolean useScanV2, Object[] partitionValues) { + boolean useScanV2, Option partitionValues) { super(fs, basePath, logFilePaths, readerSchema, latestInstantTime, readBlocksLazily, reverseReader, bufferSize, instantRange, withOperationField, forceFullScan, partitionName, internalSchema, useScanV2, partitionValues); @@ -374,7 +374,7 @@ public HoodieMergedLogRecordScanner build() { latestInstantTime, maxMemorySizeInBytes, readBlocksLazily, reverseReader, bufferSize, spillableMapBasePath, instantRange, diskMapType, isBitCaskDiskMapCompressionEnabled, withOperationField, true, - Option.ofNullable(partitionName), internalSchema, useScanV2, partitionValues); + Option.ofNullable(partitionName), internalSchema, useScanV2, Option.ofNullable(partitionValues)); } } } 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 efa2f9e043621..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 @@ -214,8 +214,8 @@ class MergeOnReadSnapshotRelation(sqlContext: SQLContext, PartitionedFile(getPartitionColumnsAsInternalRow(file.getFileStatus), filePath, 0, file.getFileLen) } - if (shouldExtractPartitionValuesFromPartitionPath) { - val partitionColumnValues =fileIndex.parsePartitionColumnValues(partitionColumns, + 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]) {