This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit baab138d7784d44d2a49ec83b861c444b3705587 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Sat Jul 18 15:06:43 2026 +0200 [FLINK-40182][table] `ArrayToArrayCastRule` and `MapToMapAndMultisetToMultisetCastRule` should check for null values in runtime This closes #28777. --- .../functions/casting/ArrayToArrayCastRule.java | 24 +++-- .../casting/MapAndMultisetToStringCastRule.java | 72 ++++++------- .../MapToMapAndMultisetToMultisetCastRule.java | 115 ++++++++++++--------- .../planner/functions/casting/CastRulesTest.java | 8 +- 4 files changed, 124 insertions(+), 95 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ArrayToArrayCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ArrayToArrayCastRule.java index df687fdb919..d3877160cb7 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ArrayToArrayCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ArrayToArrayCastRule.java @@ -55,7 +55,6 @@ class ArrayToArrayCastRule extends AbstractNullAwareCodeGeneratorCastRule<ArrayD return CastRuleProvider.resolve(innerInputType, innerTargetType) != null; } - @SuppressWarnings("rawtypes") @Override protected String generateCodeBlockInternal( CodeGeneratorCastRule.Context context, @@ -67,16 +66,17 @@ class ArrayToArrayCastRule extends AbstractNullAwareCodeGeneratorCastRule<ArrayD final LogicalType innerTargetType = ((ArrayType) targetLogicalType).getElementType(); final String innerTargetTypeTerm = arrayElementType(innerTargetType); - final String arraySize = methodCall(inputTerm, "size"); + final String arraySizeTerm = newName(context.getCodeGeneratorContext(), "arraySize"); final String objArrayTerm = newName(context.getCodeGeneratorContext(), "objArray"); return new CastRuleUtils.CodeWriter() + .declStmt(int.class, arraySizeTerm, methodCall(inputTerm, "size")) .declStmt( innerTargetTypeTerm + "[]", objArrayTerm, - newArray(innerTargetTypeTerm, arraySize)) + newArray(innerTargetTypeTerm, arraySizeTerm)) .forStmt( - arraySize, + arraySizeTerm, (index, loopWriter) -> { CastCodeBlock codeBlock = // Null check is done at the array access level @@ -97,10 +97,18 @@ class ArrayToArrayCastRule extends AbstractNullAwareCodeGeneratorCastRule<ArrayD index, codeBlock.getReturnTerm())); } else { - loopWriter - .append(codeBlock) - .assignArrayStmt( - objArrayTerm, index, codeBlock.getReturnTerm()); + loopWriter.ifStmt( + "!" + methodCall(inputTerm, "isNullAt", index), + thenWriter -> + thenWriter + .append(codeBlock) + .assignArrayStmt( + objArrayTerm, + index, + codeBlock.getReturnTerm()), + elseWriter -> + elseWriter.throwStmt( + "new org.apache.flink.table.api.TableRuntimeException(\"Target is not nullable but a NULL was found.\")")); } }, context.getCodeGeneratorContext()) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapAndMultisetToStringCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapAndMultisetToStringCastRule.java index a315fe2bbbb..2e9819b4e9f 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapAndMultisetToStringCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapAndMultisetToStringCastRule.java @@ -73,65 +73,63 @@ class MapAndMultisetToStringCastRule isNull$0 = _myInputIsNull; if (!isNull$0) { - org.apache.flink.table.data.ArrayData keys$2 = _myInput.keyArray(); - org.apache.flink.table.data.ArrayData values$3 = _myInput.valueArray(); - builder$1.setLength(0); - builder$1.append("{"); - for (int i$5 = 0; i$5 < _myInput.size(); i$5++) { - if (builder$1.length() > 12) { + int size$4 = _myInput.size(); + org.apache.flink.table.data.ArrayData keys$1 = _myInput.keyArray(); + org.apache.flink.table.data.ArrayData values$2 = _myInput.valueArray(); + builder$0.setLength(0); + builder$0.append("{"); + for (int i$5 = 0; i$5 < size$4; i$5++) { + if (builder$0.length() > 12) { break; } if (i$5 != 0) { - builder$1.append(", "); + builder$0.append(", "); } org.apache.flink.table.data.binary.BinaryStringData key$6 = org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8; - boolean keyIsNull$7 = keys$2.isNullAt(i$5); + boolean keyIsNull$7 = keys$1.isNullAt(i$5); int value$8 = -1; - boolean valueIsNull$9 = values$3.isNullAt(i$5); + boolean valueIsNull$9 = values$2.isNullAt(i$5); if (!keyIsNull$7) { - key$6 = ((org.apache.flink.table.data.binary.BinaryStringData) keys$2.getString(i$5)); - builder$1.append(key$6); + key$6 = ((org.apache.flink.table.data.binary.BinaryStringData) keys$1.getString(i$5)); + builder$0.append(key$6); } else { - builder$1.append("NULL"); + builder$0.append("NULL"); } - builder$1.append("="); + builder$0.append("="); if (!valueIsNull$9) { - value$8 = values$3.getInt(i$5); - isNull$2 = valueIsNull$9; + value$8 = values$2.getInt(i$5); + isNull$2 = false; if (!isNull$2) { - result$3 = org.apache.flink.table.data.binary.BinaryStringData.fromString("" + value$8); + result$3 = org.apache.flink.table.data.binary.BinaryStringData.fromString(org.apache.flink.table.utils.DateTimeUtils.formatIntervalYearMonth(value$8)); isNull$2 = result$3 == null; } else { result$3 = org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8; } - builder$1.append(result$3); - } else { - builder$1.append("NULL"); + builder$0.append(result$3); + } else { + builder$0.append("NULL"); + } } - } - builder$1.append("}"); - java.lang.String resultString$4; - resultString$4 = builder$1.toString(); - if (builder$1.length() > 12) { - resultString$4 = builder$1.substring(0, java.lang.Math.min(builder$1.length(), 12)); - } else { - if (resultString$.length() < 12) { + builder$0.append("}"); + java.lang.String resultString$3; + if (builder$0.length() > 12) { + resultString$3 = builder$0.substring(0, 12); + } else { + resultString$3 = builder$0.toString(); + if (builder$0.length() < 12) { int padLength$10; - padLength$10 = 12 - resultString$.length(); - java.lang.StringBuilder sbPadding$11; - sbPadding$11 = new java.lang.StringBuilder(); - for (int i$12 = 0; i$12 < padLength$10; i$12++) { - sbPadding$11.append(" "); - } - resultString$4 = resultString$4 + sbPadding$11.toString(); + padLength$10 = 12 - builder$0.length(); + resultString$3 = resultString$3 + " ".repeat(padLength$10); } } - result$1 = org.apache.flink.table.data.binary.BinaryStringData.fromString(resultString$4); + result$1 = org.apache.flink.table.data.binary.BinaryStringData.fromString(resultString$3); isNull$0 = result$1 == null; } else { result$1 = org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8; } + returnTerm = result$1 + isNullTerm = isNull$0 */ @Override protected String generateCodeBlockInternal( @@ -157,10 +155,12 @@ class MapAndMultisetToStringCastRule final String valueArrayTerm = newName(codeGeneratorContext, "values"); final String resultStringTerm = newName(codeGeneratorContext, "resultString"); + final String sizeTerm = newName(codeGeneratorContext, "size"); final int length = LogicalTypeChecks.getLength(targetLogicalType); CastRuleUtils.CodeWriter writer = new CastRuleUtils.CodeWriter() + .declStmt(int.class, sizeTerm, methodCall(inputTerm, "size")) .declStmt(ArrayData.class, keyArrayTerm, methodCall(inputTerm, "keyArray")) .declStmt( ArrayData.class, @@ -169,7 +169,7 @@ class MapAndMultisetToStringCastRule .stmt(methodCall(builderTerm, "setLength", 0)) .stmt(methodCall(builderTerm, "append", strLiteral("{"))) .forStmt( - methodCall(inputTerm, "size"), + sizeTerm, (indexTerm, loopBodyWriter) -> { String keyTerm = newName(codeGeneratorContext, "key"); String keyIsNullTerm = diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java index e4a15d17634..b0795760a55 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java @@ -18,6 +18,7 @@ package org.apache.flink.table.planner.functions.casting; +import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.GenericMapData; import org.apache.flink.table.data.MapData; import org.apache.flink.table.planner.codegen.CodeGeneratorContext; @@ -85,23 +86,27 @@ class MapToMapAndMultisetToMultisetCastRule float result$2; isNull$0 = _myInputIsNull; if (!isNull$0) { - java.util.Map map$838 = new java.util.HashMap(); - for (int i$841 = 0; i$841 < _myInput.size(); i$841++) { - java.lang.Float key$839 = null; - java.lang.Integer value$840 = null; - if (!_myInput.keyArray().isNullAt(i$841)) { - result$2 = ((float)(_myInput.keyArray().getInt(i$841))); - key$839 = result$2; + int size$2 = _myInput.size(); + org.apache.flink.table.data.ArrayData keyArray$0 = _myInput.keyArray(); + org.apache.flink.table.data.ArrayData valueArray$1 = _myInput.valueArray(); + java.util.Map map$3 = new java.util.HashMap(size$2); + for (int i$6 = 0; i$6 < size$2; i$6++) { + java.lang.Float key$4 = null; + java.lang.Integer value$5 = null; + if (!keyArray$0.isNullAt(i$6)) { + result$2 = ((float)(keyArray$0.getInt(i$6))); + key$4 = result$2; } - value$840 = _myInput.valueArray().getInt(i$841); - map$838.put(key$839, value$840); + if (!valueArray$1.isNullAt(i$6)) { + value$5 = valueArray$1.getInt(i$6); + } + map$3.put(key$4, value$5); } - result$1 = new org.apache.flink.table.data.GenericMapData(map$838); + result$1 = new org.apache.flink.table.data.GenericMapData(map$3); isNull$0 = result$1 == null; } else { result$1 = null; } - return result$1; */ @Override @@ -132,23 +137,25 @@ class MapToMapAndMultisetToMultisetCastRule final String innerTargetKeyTypeTerm = boxedTypeTermForType(innerTargetKeyType); final String innerTargetValueTypeTerm = boxedTypeTermForType(innerTargetValueType); - final String keyArrayTerm = methodCall(inputTerm, "keyArray"); - final String valueArrayTerm = methodCall(inputTerm, "valueArray"); - final String size = methodCall(inputTerm, "size"); + final String keyArray = newName(codeGeneratorContext, "keyArray"); + final String valueArray = newName(codeGeneratorContext, "valueArray"); + final String size = newName(codeGeneratorContext, "size"); final String map = newName(codeGeneratorContext, "map"); final String key = newName(codeGeneratorContext, "key"); final String value = newName(codeGeneratorContext, "value"); return new CastRuleUtils.CodeWriter() - .declStmt(className(Map.class), map, constructorCall(HashMap.class)) + .declStmt(int.class, size, methodCall(inputTerm, "size")) + .declStmt(ArrayData.class, keyArray, methodCall(inputTerm, "keyArray")) + .declStmt(ArrayData.class, valueArray, methodCall(inputTerm, "valueArray")) + .declStmt(className(Map.class), map, constructorCall(HashMap.class, size)) .forStmt( size, (index, codeWriter) -> { final CastCodeBlock keyCodeBlock = CastRuleProvider.generateAlwaysNonNullCodeBlock( context, - rowFieldReadAccess( - index, keyArrayTerm, innerInputKeyType), + rowFieldReadAccess(index, keyArray, innerInputKeyType), innerInputKeyType, innerTargetKeyType); assert keyCodeBlock != null; @@ -157,7 +164,7 @@ class MapToMapAndMultisetToMultisetCastRule CastRuleProvider.generateAlwaysNonNullCodeBlock( context, rowFieldReadAccess( - index, valueArrayTerm, innerInputValueType), + index, valueArray, innerInputValueType), innerInputValueType, innerTargetValueType); assert valueCodeBlock != null; @@ -165,39 +172,53 @@ class MapToMapAndMultisetToMultisetCastRule codeWriter .declStmt(innerTargetKeyTypeTerm, key, null) .declStmt(innerTargetValueTypeTerm, value, null); - if (innerTargetKeyType.isNullable()) { - codeWriter.ifStmt( - "!" + methodCall(keyArrayTerm, "isNullAt", index), - thenWriter -> - thenWriter - .append(keyCodeBlock) - .assignStmt( - key, keyCodeBlock.getReturnTerm())); - } else { - codeWriter - .append(keyCodeBlock) - .assignStmt(key, keyCodeBlock.getReturnTerm()); - } - - if (inputLogicalType.is(LogicalTypeRoot.MAP) - && innerTargetValueType.isNullable()) { - codeWriter.ifStmt( - "!" + methodCall(valueArrayTerm, "isNullAt", index), - thenWriter -> - thenWriter - .append(valueCodeBlock) - .assignStmt( - value, - valueCodeBlock.getReturnTerm())); - } else { - codeWriter - .append(valueCodeBlock) - .assignStmt(value, valueCodeBlock.getReturnTerm()); - } + iterateOverElements( + index, + codeWriter, + keyArray, + keyCodeBlock, + key, + !innerTargetKeyType.isNullable()); + + iterateOverElements( + index, + codeWriter, + valueArray, + valueCodeBlock, + value, + !inputLogicalType.is(LogicalTypeRoot.MAP) + || !innerTargetValueType.isNullable()); codeWriter.stmt(methodCall(map, "put", key, value)); }, codeGeneratorContext) .assignStmt(returnVariable, constructorCall(GenericMapData.class, map)) .toString(); } + + private static void iterateOverElements( + String index, + CastRuleUtils.CodeWriter codeWriter, + String keyArray, + CastCodeBlock keyCodeBlock, + String key, + boolean throwIfNull) { + if (throwIfNull) { + codeWriter.ifStmt( + "!" + methodCall(keyArray, "isNullAt", index), + thenWriter -> + thenWriter + .append(keyCodeBlock) + .assignStmt(key, keyCodeBlock.getReturnTerm()), + elseWriter -> + elseWriter.throwStmt( + "new org.apache.flink.table.api.TableRuntimeException(\"Target is not nullable but a NULL was found.\")")); + } else { + codeWriter.ifStmt( + "!" + methodCall(keyArray, "isNullAt", index), + thenWriter -> + thenWriter + .append(keyCodeBlock) + .assignStmt(key, keyCodeBlock.getReturnTerm())); + } + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index 8459a804163..aca973fa3ca 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -1370,7 +1370,7 @@ class CastRulesTest { new GenericArrayData(new Integer[] {1, 2, null}), new GenericArrayData(new Integer[] {3}) }), - NullPointerException.class) + TableRuntimeException.class) .fromCase( ARRAY(ARRAY(INT().nullable())), new GenericArrayData( @@ -1430,12 +1430,12 @@ class CastRulesTest { .fail( MAP(INT().nullable(), DOUBLE().nullable()), mapData(entry(null, 1d)), - NullPointerException.class), + TableRuntimeException.class), CastTestSpecBuilder.testCastTo(MAP(STRING().notNull(), STRING().notNull())) .fail( MAP(INT().nullable(), DOUBLE().nullable()), mapData(entry(123, null)), - NullPointerException.class), + TableRuntimeException.class), CastTestSpecBuilder.testCastTo(MULTISET(DOUBLE().notNull())) .fromCase( MULTISET(INT().nullable()), @@ -1455,7 +1455,7 @@ class CastRulesTest { .fail( MULTISET(INT().nullable()), mapData(entry(null, 1)), - NullPointerException.class), + TableRuntimeException.class), CastTestSpecBuilder.testCastTo( ROW(BIGINT().notNull(), BIGINT(), STRING(), ARRAY(STRING()))) .fromCase(
