nateab commented on code in PR #27505: URL: https://github.com/apache/flink/pull/27505#discussion_r2763370201
########## .gitignore: ########## @@ -10,6 +10,7 @@ scalastyle-output.xml .metals .bloop .cursor +.claude Review Comment: separated out into its own commit ########## flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/aggregate/MiniBatchGroupAggFunctionTest.java: ########## @@ -0,0 +1,196 @@ +/* + * 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.flink.table.runtime.operators.aggregate; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.types.RowKind; +import org.apache.flink.util.Collector; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link MiniBatchGroupAggFunction}. + * + * <p>This test covers the scenario where MiniBatchGroupAggFunction.finishBundle() encounters a key + * with only retraction messages and no state. The method must use 'continue' instead of 'return' to + * avoid silently dropping all subsequent keys in the bundle. + */ +class MiniBatchGroupAggFunctionTest { + + /** + * Verifies that when finishBundle processes a key with only retraction messages (which gets + * filtered out because there's no accumulator state), the method continues to process + * subsequent keys in the bundle instead of returning early. + * + * <p>This test uses a LinkedHashMap to ensure deterministic iteration order, placing the + * retraction-only key first to trigger the bug scenario. + */ + @Test + void testFinishBundleContinuesAfterEmptyInputRows() throws Exception { + // Create a mock buffer with deterministic order (LinkedHashMap) + // Key "aaa" has only a DELETE message (will be filtered out since no state) + // Key "bbb" has an INSERT message (should be processed) + Map<RowData, List<RowData>> buffer = new LinkedHashMap<>(); + + // Key "aaa" - only retraction, no existing state + GenericRowData keyA = GenericRowData.of(StringData.fromString("aaa")); + List<RowData> rowsA = new ArrayList<>(); + GenericRowData deleteA = GenericRowData.of(StringData.fromString("aaa"), 1L); + deleteA.setRowKind(RowKind.DELETE); + rowsA.add(deleteA); + buffer.put(keyA, rowsA); + + // Key "bbb" - normal insert + GenericRowData keyB = GenericRowData.of(StringData.fromString("bbb")); + List<RowData> rowsB = new ArrayList<>(); + GenericRowData insertB = GenericRowData.of(StringData.fromString("bbb"), 2L); + insertB.setRowKind(RowKind.INSERT); + rowsB.add(insertB); + buffer.put(keyB, rowsB); + + // Key "ccc" - normal insert + GenericRowData keyC = GenericRowData.of(StringData.fromString("ccc")); + List<RowData> rowsC = new ArrayList<>(); + GenericRowData insertC = GenericRowData.of(StringData.fromString("ccc"), 3L); + insertC.setRowKind(RowKind.INSERT); + rowsC.add(insertC); + buffer.put(keyC, rowsC); + + // Collect output + List<RowData> output = new ArrayList<>(); + TestCollector collector = new TestCollector(output); + + // Create a test instance with mocks + TestMiniBatchGroupAggFunction function = new TestMiniBatchGroupAggFunction(); + function.finishBundle(buffer, collector); + + // Verify that keys "bbb" and "ccc" were processed (2 outputs) + // Before the fix, only key "aaa" would be processed (and filtered out), + // then the method would return early, producing 0 outputs. Review Comment: dropped -- 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]
