danny0405 commented on code in PR #19353: URL: https://github.com/apache/hudi/pull/19353#discussion_r3642687180
########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieNativeLogAppendHandle.java: ########## @@ -0,0 +1,286 @@ +/* + * 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.io; + +import org.apache.hudi.common.config.RecordMergeMode; +import org.apache.hudi.common.engine.LocalTaskContextSupplier; +import org.apache.hudi.common.engine.RecordContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieDeltaWriteStat; +import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieLogFile; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.common.table.log.AppendResult; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieAppendException; +import org.apache.hudi.io.cdc.HoodieNativeLogFormatWriter; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.table.HoodieTable; + +import org.junit.jupiter.api.Test; +import org.mockito.MockedConstruction; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class TestHoodieNativeLogAppendHandle { + + private static final String SCHEMA = "{\"type\":\"record\",\"name\":\"trip\",\"fields\":[" + + "{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"ts\",\"type\":\"long\"}]}"; + + @Test + public void testCreatesWriterAndRoutesDataAndDeleteRecords() throws Exception { + HoodieWriteConfig config = config(); + HoodieTable table = table(config); + HoodieRecord inputRecord = mock(HoodieRecord.class); + HoodieRecord populatedRecord = mock(HoodieRecord.class); + when(inputRecord.prependMetaFields(any(HoodieSchema.class), any(HoodieSchema.class), any(), any())) + .thenReturn(populatedRecord); + + try (MockedConstruction<HoodieNativeLogFormatWriter> writers = mockConstruction( + HoodieNativeLogFormatWriter.class, (writer, context) -> { + when(writer.canWriteDataFile()).thenReturn(true); + when(writer.canWriteDeleteFile()).thenReturn(true); + when(writer.hasPendingWrites()).thenReturn(true); + when(writer.getLastAppendResults()).thenReturn(Collections.emptyList()); + when(writer.getLogFile()).thenReturn(new HoodieLogFile(new StoragePath("/tmp/native.log.parquet"))); + })) { + TestableNativeLogAppendHandle handle = new TestableNativeLogAppendHandle(config, table); + handle.createWriter(); + HoodieNativeLogFormatWriter writer = writers.constructed().get(0); + + handle.writeData(inputRecord, true); + verify(writer).appendRecord(eq(populatedRecord), any(HoodieSchema.class), + eq(HoodieRecord.RECORD_KEY_METADATA_FIELD)); + handle.writeDeleteRecord(inputRecord); + verify(inputRecord).clearNewLocation(); + verify(writer).appendDeleteRecord(eq(inputRecord), any(HoodieSchema.class), + eq(HoodieRecord.RECORD_KEY_METADATA_FIELD)); + + handle.flushWriter(); + verify(writer).flushAppend(any()); + assertEquals("/tmp/native.log.parquet", handle.logFilePath().toString()); + assertTrue(handle.canWrite(inputRecord)); + handle.closeWriter(); + verify(writer).close(); + } + } + + @Test + public void testFlushesBeforeWriterRolloverAndTracksPerFileCounts() throws Exception { + HoodieWriteConfig config = config(); + HoodieTable table = table(config); + HoodieRecord inputRecord = mock(HoodieRecord.class); + HoodieRecord populatedRecord = mock(HoodieRecord.class); + when(inputRecord.prependMetaFields(any(HoodieSchema.class), any(HoodieSchema.class), any(), any())) + .thenReturn(populatedRecord); + + try (MockedConstruction<HoodieNativeLogFormatWriter> writers = mockConstruction( + HoodieNativeLogFormatWriter.class, (writer, context) -> { + when(writer.canWriteDataFile()).thenReturn(false); + when(writer.hasPendingWrites()).thenReturn(false); Review Comment: `canWriteDataFile()` forces `writeInsertAndUpdate()` to call `flushAppend()`, but this stub makes `flushAppend()` return immediately; the assertion below even verifies that no physical flush occurred. As a result, this does not cover the “flush before rollover” behavior or the counter reset performed by `processAppendResults()`. Please return `true` here, provide `getLastAppendResults()`, and verify the flush happens before the next append (and that the next file starts with reset counts). ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeCDCLoggers.java: ########## @@ -0,0 +1,197 @@ +/* + * 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.io.cdc; + +import org.apache.hudi.common.avro.AvroRecordContext; +import org.apache.hudi.common.engine.TaskContextSupplier; +import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaUtils; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.cdc.HoodieCDCOperation; +import org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode; +import org.apache.hudi.common.table.log.LogFileCreationCallback; +import org.apache.hudi.common.table.read.BufferedRecord; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.generic.IndexedRecord; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.mockito.ArgumentCaptor; +import org.mockito.MockedConstruction; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.apache.hudi.common.table.cdc.HoodieCDCUtils.CDC_AFTER_IMAGE; +import static org.apache.hudi.common.table.cdc.HoodieCDCUtils.CDC_BEFORE_IMAGE; +import static org.apache.hudi.common.table.cdc.HoodieCDCUtils.CDC_COMMIT_TIMESTAMP; +import static org.apache.hudi.common.table.cdc.HoodieCDCUtils.CDC_OPERATION_TYPE; +import static org.apache.hudi.common.table.cdc.HoodieCDCUtils.CDC_RECORD_KEY; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class TestHoodieNativeCDCLoggers { + + private static final String COMMIT_TIME = "100"; + private static final HoodieSchema DATA_SCHEMA = HoodieSchema.fromAvroSchema(new Schema.Parser().parse( + "{\"type\":\"record\",\"name\":\"trip\",\"namespace\":\"org.apache.hudi\",\"fields\":[" + + "{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"name\",\"type\":\"string\"}]}")); + private static final HoodieSchema SCHEMA_WITH_METADATA = HoodieSchemaUtils.addMetadataFields(DATA_SCHEMA); + + @ParameterizedTest + @EnumSource(HoodieCDCSupplementalLoggingMode.class) + public void testAvroLoggerWritesAllOperationsAndSupportsRetraction( + HoodieCDCSupplementalLoggingMode mode) throws Exception { + try (MockedConstruction<HoodieNativeCDCFileWriter> mockedWriters = mockConstruction(HoodieNativeCDCFileWriter.class)) { + HoodieAvroNativeCDCLogger logger = new HoodieAvroNativeCDCLogger( + COMMIT_TIME, mock(HoodieWriteConfig.class), tableConfig(mode), "partition", + mock(HoodieStorage.class), SCHEMA_WITH_METADATA, new StoragePath("/tmp/partition"), + "file-1", "1-0-1", new LogFileCreationCallback() { + }, mock(TaskContextSupplier.class)); + HoodieNativeCDCFileWriter<IndexedRecord> writer = mockedWriters.constructed().get(0); + Map<String, Long> expectedStats = Collections.singletonMap("partition/file.cdc.parquet", 10L); + when(writer.getCDCWriteStats()).thenReturn(expectedStats); + + GenericRecord oldRecord = record("id-1", "old"); + GenericRecord newRecord = record("id-1", "new"); + logger.put("id-1", oldRecord, Option.of(newRecord)); + verify(writer, never()).write(anyString(), any()); + logger.put("id-2", null, Option.of(record("id-2", "insert"))); + logger.put("id-3", oldRecord, Option.empty()); + logger.put("id-4", null, Option.of(record("id-4", "retracted"))); + logger.remove("different-key"); + logger.remove("id-4"); + logger.close(); + + assertSame(expectedStats, logger.getCDCWriteStats()); + verify(writer, times(3)).write(anyString(), any()); + verify(writer).close(); + + ArgumentCaptor<IndexedRecord> recordCaptor = ArgumentCaptor.forClass(IndexedRecord.class); + verify(writer, times(3)).write(anyString(), recordCaptor.capture()); + List<IndexedRecord> cdcRecords = recordCaptor.getAllValues(); + assertCDCRecord((GenericRecord) cdcRecords.get(0), mode, HoodieCDCOperation.UPDATE, "id-1"); Review Comment: These assertions validate only the operation/key and removal of Hudi metadata fields. In `DATA_BEFORE` / `DATA_BEFORE_AFTER`, a transformer that swaps the old/new images, emits the wrong `name`, or leaves an image null can still pass. Since this test is the coverage for all supplemental logging modes, please assert the actual before/after image values (`old` and `new`) for update, plus the expected null/non-null images for insert/delete. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
