cshuo commented on code in PR #19353: URL: https://github.com/apache/hudi/pull/19353#discussion_r3642808189
########## 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: Fixed in 5d8636b4995. The tests now validate before/after image IDs, names, and nullability for update, insert, and delete across every supplemental logging mode while retaining metadata projection checks. ########## 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: Fixed in 5d8636b4995. The test now supplies pending append results, verifies the flush occurs before the rollover append, and checks that the next data file starts with reset counts. -- 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]
