codope commented on code in PR #11415: URL: https://github.com/apache/hudi/pull/11415#discussion_r1631913348
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPositionBasedMergingFallback.scala: ########## @@ -0,0 +1,192 @@ +/* + * 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.functional + +import org.apache.hadoop.fs.FileSystem +import org.apache.hudi.DataSourceWriteOptions +import org.apache.hudi.DataSourceWriteOptions.{OPERATION, PRECOMBINE_FIELD, RECORDKEY_FIELD, TABLE_TYPE} +import org.apache.hudi.HoodieConversionUtils.toJavaOption +import org.apache.hudi.common.config.{HoodieReaderConfig, HoodieStorageConfig} +import org.apache.hudi.common.model.HoodieRecordMerger +import org.apache.hudi.common.util +import org.apache.hudi.config.HoodieWriteConfig +import org.apache.hudi.testutils.HoodieSparkClientTestBase +import org.apache.hudi.util.JFunction +import org.apache.spark.sql.SaveMode.{Append, Overwrite} +import org.apache.spark.sql.SparkSessionExtensions +import org.apache.spark.sql.hudi.HoodieSparkSessionExtension +import org.apache.spark.sql.internal.SQLConf +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.{AfterEach, BeforeEach} +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.{Arguments, MethodSource} + +import java.util.function.Consumer + +class TestPositionBasedMergingFallback extends HoodieSparkClientTestBase { + override def getSparkSessionExtensionsInjector: util.Option[Consumer[SparkSessionExtensions]] = + toJavaOption( + Some( + JFunction.toJavaConsumer((receiver: SparkSessionExtensions) => new HoodieSparkSessionExtension().apply(receiver))) + ) + + @BeforeEach override def setUp(): Unit = { + initPath() + initSparkContexts() + sparkSession.conf.set(SQLConf.PARQUET_RECORD_FILTER_ENABLED.key, "true") + initTestDataGenerator() + initHoodieStorage() + } + + @AfterEach override def tearDown(): Unit = { + cleanupSparkContexts() + cleanupTestDataGenerator() + cleanupFileSystem() + FileSystem.closeAll() + System.gc() Review Comment: let's avoid System.gc() ########## hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedFileGroupRecordBuffer.java: ########## @@ -174,20 +189,97 @@ public boolean containsLogRecord(String recordKey) { } @Override - protected boolean doHasNext() throws IOException { - ValidationUtils.checkState(baseFileIterator != null, "Base file iterator has not been set yet"); - - // Handle merging. - while (baseFileIterator.hasNext()) { - T baseRecord = baseFileIterator.next(); - nextRecordPosition = readerContext.extractRecordPosition(baseRecord, readerSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME, nextRecordPosition); - Pair<Option<T>, Map<String, Object>> logRecordInfo = records.remove(nextRecordPosition++); - if (hasNextBaseRecord(baseRecord, logRecordInfo)) { - return true; + protected boolean hasNextBaseRecord(T baseRecord) throws IOException { + if (!readerContext.getShouldMergeUseRecordPosition()) { + return doHasNextFallbackBaseRecord(baseRecord); + } + + nextRecordPosition = readerContext.extractRecordPosition(baseRecord, readerSchema, + ROW_INDEX_COLUMN_NAME, nextRecordPosition); + Pair<Option<T>, Map<String, Object>> logRecordInfo = records.remove(nextRecordPosition++); + + Map<String, Object> metadata = readerContext.generateMetadataForRecord( + baseRecord, readerSchema); + + Option<T> resultRecord = logRecordInfo != null + ? merge(Option.of(baseRecord), metadata, logRecordInfo.getLeft(), logRecordInfo.getRight()) + : merge(Option.empty(), Collections.emptyMap(), Option.of(baseRecord), metadata); + if (resultRecord.isPresent()) { + nextRecord = readerContext.seal(resultRecord.get()); + return true; + } + return false; + } + + private boolean doHasNextFallbackBaseRecord(T baseRecord) throws IOException { + if (needToDoHybridStrategy) { Review Comment: let's test this logic as well. ########## hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedFileGroupRecordBuffer.java: ########## @@ -123,46 +142,42 @@ public void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpecO } } - @Override - public void processNextDataRecord(T record, Map<String, Object> metadata, Serializable recordPosition) throws IOException { - Pair<Option<T>, Map<String, Object>> existingRecordMetadataPair = records.get(recordPosition); - Option<Pair<T, Map<String, Object>>> mergedRecordAndMetadata = - doProcessNextDataRecord(record, metadata, existingRecordMetadataPair); - if (mergedRecordAndMetadata.isPresent()) { - records.put(recordPosition, Pair.of( - Option.ofNullable(readerContext.seal(mergedRecordAndMetadata.get().getLeft())), - mergedRecordAndMetadata.get().getRight())); + private void fallbackToKeyBasedBuffer() { + readerContext.setShouldMergeUseRecordPosition(false); + //need to make a copy of the keys to avoid concurrent modification exception + ArrayList<Serializable> positions = new ArrayList<>(records.keySet()); Review Comment: rename `positions` to `recordKeys` and `position` to `recordKey`, or something like that.. because this is not position right. ########## hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedFileGroupRecordBuffer.java: ########## @@ -174,20 +189,97 @@ public boolean containsLogRecord(String recordKey) { } @Override - protected boolean doHasNext() throws IOException { - ValidationUtils.checkState(baseFileIterator != null, "Base file iterator has not been set yet"); - - // Handle merging. - while (baseFileIterator.hasNext()) { - T baseRecord = baseFileIterator.next(); - nextRecordPosition = readerContext.extractRecordPosition(baseRecord, readerSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME, nextRecordPosition); - Pair<Option<T>, Map<String, Object>> logRecordInfo = records.remove(nextRecordPosition++); - if (hasNextBaseRecord(baseRecord, logRecordInfo)) { - return true; + protected boolean hasNextBaseRecord(T baseRecord) throws IOException { + if (!readerContext.getShouldMergeUseRecordPosition()) { + return doHasNextFallbackBaseRecord(baseRecord); + } + + nextRecordPosition = readerContext.extractRecordPosition(baseRecord, readerSchema, + ROW_INDEX_COLUMN_NAME, nextRecordPosition); + Pair<Option<T>, Map<String, Object>> logRecordInfo = records.remove(nextRecordPosition++); + + Map<String, Object> metadata = readerContext.generateMetadataForRecord( + baseRecord, readerSchema); + + Option<T> resultRecord = logRecordInfo != null + ? merge(Option.of(baseRecord), metadata, logRecordInfo.getLeft(), logRecordInfo.getRight()) + : merge(Option.empty(), Collections.emptyMap(), Option.of(baseRecord), metadata); + if (resultRecord.isPresent()) { + nextRecord = readerContext.seal(resultRecord.get()); + return true; + } + return false; + } + + private boolean doHasNextFallbackBaseRecord(T baseRecord) throws IOException { + if (needToDoHybridStrategy) { + //see if there is a delete block with record positions + nextRecordPosition = readerContext.extractRecordPosition(baseRecord, readerSchema, + ROW_INDEX_TEMPORARY_COLUMN_NAME, nextRecordPosition); + Pair<Option<T>, Map<String, Object>> logRecordInfo = records.remove(nextRecordPosition++); Review Comment: Does it need to be thread-safe? Or can multiple threads update the nextRecordPosition? Maybe make it AtomicLong? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPositionBasedMergingFallback.scala: ########## @@ -0,0 +1,192 @@ +/* + * 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.functional + +import org.apache.hadoop.fs.FileSystem +import org.apache.hudi.DataSourceWriteOptions +import org.apache.hudi.DataSourceWriteOptions.{OPERATION, PRECOMBINE_FIELD, RECORDKEY_FIELD, TABLE_TYPE} +import org.apache.hudi.HoodieConversionUtils.toJavaOption +import org.apache.hudi.common.config.{HoodieReaderConfig, HoodieStorageConfig} +import org.apache.hudi.common.model.HoodieRecordMerger +import org.apache.hudi.common.util +import org.apache.hudi.config.HoodieWriteConfig +import org.apache.hudi.testutils.HoodieSparkClientTestBase +import org.apache.hudi.util.JFunction +import org.apache.spark.sql.SaveMode.{Append, Overwrite} +import org.apache.spark.sql.SparkSessionExtensions +import org.apache.spark.sql.hudi.HoodieSparkSessionExtension +import org.apache.spark.sql.internal.SQLConf +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.{AfterEach, BeforeEach} +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.{Arguments, MethodSource} + +import java.util.function.Consumer + +class TestPositionBasedMergingFallback extends HoodieSparkClientTestBase { + override def getSparkSessionExtensionsInjector: util.Option[Consumer[SparkSessionExtensions]] = + toJavaOption( + Some( + JFunction.toJavaConsumer((receiver: SparkSessionExtensions) => new HoodieSparkSessionExtension().apply(receiver))) + ) + + @BeforeEach override def setUp(): Unit = { + initPath() + initSparkContexts() + sparkSession.conf.set(SQLConf.PARQUET_RECORD_FILTER_ENABLED.key, "true") + initTestDataGenerator() + initHoodieStorage() + } + + @AfterEach override def tearDown(): Unit = { + cleanupSparkContexts() + cleanupTestDataGenerator() + cleanupFileSystem() + FileSystem.closeAll() + System.gc() + } + + @ParameterizedTest + @MethodSource(Array("testArgs")) + def testPositionFallback(updateWithRecordPositions: String, deleteWithRecordPositions: String, secondUpdateWithPositions: String): Unit = { + val columns = Seq("ts", "key", "name", "_hoodie_is_deleted") + val data = Seq( + (10, "1", "A", false), + (10, "2", "B", false), + (10, "3", "C", false), + (10, "4", "D", false), + (10, "5", "E", false)) + + val inserts = sparkSession.createDataFrame(data).toDF(columns: _*) + inserts.write.format("hudi"). + option(RECORDKEY_FIELD.key(), "key"). + option(PRECOMBINE_FIELD.key(), "ts"). + option("hoodie.table.name", "test_table"). + option(TABLE_TYPE.key(), "MERGE_ON_READ"). + option(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet"). + option(DataSourceWriteOptions.RECORD_MERGER_STRATEGY.key(), HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID). + option(DataSourceWriteOptions.RECORD_MERGER_IMPLS.key(), "org.apache.hudi.HoodieSparkRecordMerger"). + option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true"). + option(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), "true"). + mode(Overwrite). + save(basePath) + + val updateData = Seq((11, "1", "A_1", false), (9, "2", "B_1", false)) + + val updates = sparkSession.createDataFrame(updateData).toDF(columns: _*) + + updates.write.format("hudi"). + option(RECORDKEY_FIELD.key(), "key"). Review Comment: let's extract the options to separate methods such as `getWriterOptions` and `getReaderOptions` which can take the same arguments as passed to the test. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPositionBasedMergingFallback.scala: ########## @@ -0,0 +1,192 @@ +/* + * 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.functional + +import org.apache.hadoop.fs.FileSystem +import org.apache.hudi.DataSourceWriteOptions +import org.apache.hudi.DataSourceWriteOptions.{OPERATION, PRECOMBINE_FIELD, RECORDKEY_FIELD, TABLE_TYPE} +import org.apache.hudi.HoodieConversionUtils.toJavaOption +import org.apache.hudi.common.config.{HoodieReaderConfig, HoodieStorageConfig} +import org.apache.hudi.common.model.HoodieRecordMerger +import org.apache.hudi.common.util +import org.apache.hudi.config.HoodieWriteConfig +import org.apache.hudi.testutils.HoodieSparkClientTestBase +import org.apache.hudi.util.JFunction +import org.apache.spark.sql.SaveMode.{Append, Overwrite} +import org.apache.spark.sql.SparkSessionExtensions +import org.apache.spark.sql.hudi.HoodieSparkSessionExtension +import org.apache.spark.sql.internal.SQLConf +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.{AfterEach, BeforeEach} +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.{Arguments, MethodSource} + +import java.util.function.Consumer + +class TestPositionBasedMergingFallback extends HoodieSparkClientTestBase { + override def getSparkSessionExtensionsInjector: util.Option[Consumer[SparkSessionExtensions]] = + toJavaOption( + Some( + JFunction.toJavaConsumer((receiver: SparkSessionExtensions) => new HoodieSparkSessionExtension().apply(receiver))) + ) + + @BeforeEach override def setUp(): Unit = { + initPath() + initSparkContexts() + sparkSession.conf.set(SQLConf.PARQUET_RECORD_FILTER_ENABLED.key, "true") + initTestDataGenerator() + initHoodieStorage() + } + + @AfterEach override def tearDown(): Unit = { + cleanupSparkContexts() + cleanupTestDataGenerator() + cleanupFileSystem() + FileSystem.closeAll() + System.gc() + } + + @ParameterizedTest + @MethodSource(Array("testArgs")) + def testPositionFallback(updateWithRecordPositions: String, deleteWithRecordPositions: String, secondUpdateWithPositions: String): Unit = { + val columns = Seq("ts", "key", "name", "_hoodie_is_deleted") + val data = Seq( + (10, "1", "A", false), + (10, "2", "B", false), + (10, "3", "C", false), + (10, "4", "D", false), + (10, "5", "E", false)) + + val inserts = sparkSession.createDataFrame(data).toDF(columns: _*) + inserts.write.format("hudi"). + option(RECORDKEY_FIELD.key(), "key"). + option(PRECOMBINE_FIELD.key(), "ts"). + option("hoodie.table.name", "test_table"). + option(TABLE_TYPE.key(), "MERGE_ON_READ"). + option(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet"). + option(DataSourceWriteOptions.RECORD_MERGER_STRATEGY.key(), HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID). + option(DataSourceWriteOptions.RECORD_MERGER_IMPLS.key(), "org.apache.hudi.HoodieSparkRecordMerger"). + option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true"). + option(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), "true"). + mode(Overwrite). + save(basePath) + + val updateData = Seq((11, "1", "A_1", false), (9, "2", "B_1", false)) + + val updates = sparkSession.createDataFrame(updateData).toDF(columns: _*) + + updates.write.format("hudi"). + option(RECORDKEY_FIELD.key(), "key"). + option(PRECOMBINE_FIELD.key(), "ts"). + option("hoodie.table.name", "test_table"). + option(TABLE_TYPE.key(), "MERGE_ON_READ"). + option(OPERATION.key(), "upsert"). + option(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet"). + option(DataSourceWriteOptions.RECORD_MERGER_STRATEGY.key(), HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID). + option(DataSourceWriteOptions.RECORD_MERGER_IMPLS.key(), "org.apache.hudi.HoodieSparkRecordMerger"). + option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true"). + option(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), updateWithRecordPositions). + mode(Append). + save(basePath) + + val deletesData = Seq((10, "4", "D", true), (10, "3", "C", true)) + + val deletes = sparkSession.createDataFrame(deletesData).toDF(columns: _*) + deletes.write.format("hudi"). + option(RECORDKEY_FIELD.key(), "key"). + option(PRECOMBINE_FIELD.key(), "ts"). + option("hoodie.table.name", "test_table"). + option(TABLE_TYPE.key(), "MERGE_ON_READ"). + option(OPERATION.key(), "upsert"). + option(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet"). + option(DataSourceWriteOptions.RECORD_MERGER_STRATEGY.key(), HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID). + option(DataSourceWriteOptions.RECORD_MERGER_IMPLS.key(), "org.apache.hudi.HoodieSparkRecordMerger"). + option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true"). + option(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), deleteWithRecordPositions). + mode(Append). + save(basePath) + + + val secondUpdateData = Seq((14, "5", "E_3", false), (3, "3", "C_3", false)) + val secondUpdates = sparkSession.createDataFrame(secondUpdateData).toDF(columns: _*) + secondUpdates.write.format("hudi"). + option(RECORDKEY_FIELD.key(), "key"). + option(PRECOMBINE_FIELD.key(), "ts"). + option("hoodie.table.name", "test_table"). + option(TABLE_TYPE.key(), "MERGE_ON_READ"). + option(OPERATION.key(), "upsert"). + option(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet"). + option(DataSourceWriteOptions.RECORD_MERGER_STRATEGY.key(), HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID). + option(DataSourceWriteOptions.RECORD_MERGER_IMPLS.key(), "org.apache.hudi.HoodieSparkRecordMerger"). + option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true"). + option(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), secondUpdateWithPositions). + mode(Append). + save(basePath) + + val df = sparkSession.read.format("hudi"). + option(DataSourceWriteOptions.RECORD_MERGER_STRATEGY.key(), HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID). + option(DataSourceWriteOptions.RECORD_MERGER_IMPLS.key(), "org.apache.hudi.HoodieSparkRecordMerger"). + option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key(), "true"). + option(HoodieReaderConfig.MERGE_USE_RECORD_POSITIONS.key(), "true").load(basePath) + val finalDf = df.select("ts", "key", "name") + val finalColumns = Seq("ts", "key", "name") + + val finalExpectedData = Seq( + (11, "1", "A_1"), + (10, "2", "B"), + (14, "5", "E_3")) + + val expectedDf = sparkSession.createDataFrame(finalExpectedData).toDF(finalColumns: _*) + + assertEquals(0, finalDf.except(expectedDf).count()) + assertEquals(0, expectedDf.except(finalDf).count()) Review Comment: this is cool but do we also test logic under `needToDoHybridStrategy` anywhere? Basically, the scenario "if it's a delete record and the key is null, then we need to still use positions". -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@hudi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
