nateab commented on code in PR #27505: URL: https://github.com/apache/flink/pull/27505#discussion_r2763375020
########## 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. + assertThat(output).hasSize(2); + + // Verify the output contains results for both "bbb" and "ccc" + assertThat(output.stream().anyMatch(row -> row.getString(0).toString().equals("bbb"))) + .isTrue(); + assertThat(output.stream().anyMatch(row -> row.getString(0).toString().equals("ccc"))) + .isTrue(); + } + + /** Test collector that stores output rows. */ + private static class TestCollector implements Collector<RowData> { + private final List<RowData> output; + + TestCollector(List<RowData> output) { + this.output = output; + } + + @Override + public void collect(RowData record) { + // Make a copy since the record might be reused + GenericRowData copy = GenericRowData.of(record.getString(0), record.getLong(1)); + copy.setRowKind(record.getRowKind()); + output.add(copy); + } + + @Override + public void close() {} + } + + /** + * A test version of MiniBatchGroupAggFunction that doesn't require state infrastructure. It + * simulates the behavior of the real function but uses in-memory maps instead of Flink state. + */ + private static class TestMiniBatchGroupAggFunction { + private final Map<RowData, RowData> accumulators = new LinkedHashMap<>(); + private RowData currentKey; + + public void finishBundle(Map<RowData, List<RowData>> buffer, Collector<RowData> out) + throws Exception { + for (Map.Entry<RowData, List<RowData>> entry : buffer.entrySet()) { + RowData currentKey = entry.getKey(); + List<RowData> inputRows = entry.getValue(); + + boolean firstRow = false; + + // Simulate: set current key and get accumulator from state + this.currentKey = currentKey; + RowData acc = accumulators.get(currentKey); + + if (acc == null) { + // Don't create a new accumulator for a retraction message. + java.util.Iterator<RowData> inputIter = inputRows.iterator(); + while (inputIter.hasNext()) { + RowData current = inputIter.next(); + if (isRetractMsg(current)) { + inputIter.remove(); + } else { + break; + } + } + if (inputRows.isEmpty()) { + // Use 'continue' instead of 'return' to avoid exiting the entire method + continue; + } + acc = createAccumulators(); + firstRow = true; + } + + // Simulate accumulation + long sum = acc.getLong(0); + for (RowData input : inputRows) { + if (isAccumulateMsg(input)) { + sum += input.getLong(1); + } else { + sum -= input.getLong(1); + } + } + acc = GenericRowData.of(sum); + + // Simulate: store accumulator + accumulators.put(currentKey, acc); + + // Output result + GenericRowData result = GenericRowData.of(currentKey.getString(0), sum); + result.setRowKind(firstRow ? RowKind.INSERT : RowKind.UPDATE_AFTER); + out.collect(result); + } + } + + private RowData createAccumulators() { + return GenericRowData.of(0L); + } + + private boolean isRetractMsg(RowData row) { + return row.getRowKind() == RowKind.UPDATE_BEFORE || row.getRowKind() == RowKind.DELETE; + } + + private boolean isAccumulateMsg(RowData row) { + return row.getRowKind() == RowKind.INSERT || row.getRowKind() == RowKind.UPDATE_AFTER; + } + } Review Comment: replaced with real instance of MiniBatchGroupAggFunction production code ########## 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. + assertThat(output).hasSize(2); + + // Verify the output contains results for both "bbb" and "ccc" + assertThat(output.stream().anyMatch(row -> row.getString(0).toString().equals("bbb"))) + .isTrue(); + assertThat(output.stream().anyMatch(row -> row.getString(0).toString().equals("ccc"))) + .isTrue(); + } + + /** Test collector that stores output rows. */ + private static class TestCollector implements Collector<RowData> { + private final List<RowData> output; + + TestCollector(List<RowData> output) { + this.output = output; + } + + @Override + public void collect(RowData record) { + // Make a copy since the record might be reused + GenericRowData copy = GenericRowData.of(record.getString(0), record.getLong(1)); + copy.setRowKind(record.getRowKind()); + output.add(copy); + } + + @Override + public void close() {} + } Review Comment: removed -- 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]
