Repository: apache/flink
File: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/TypeInferenceOperandInference.java (Java)

Code before:
            TypeInference typeInference) {
        this.dataTypeFactory = dataTypeFactory;
        this.definition = definition;
        this.typeInference = typeInference;
    }

    @Override
    public void inferOperandTypes(
            SqlCallBinding callBinding, RelDataType returnType, RelDataType[] operandTypes) {
        final CallContext callContext =
                new CallBindingCallContext(
                        dataTypeFactory,
                        definition,
                        callBinding,
                        returnType,
                        typeInference.getStaticArguments().orElse(null));
        try {
            if (TypeInferenceUtil.validateArgumentCount(
                    typeInference.getInputTypeStrategy().getArgumentCount(),
                    callContext.getArgumentDataTypes().size(),
                    false)) {
                inferOperandTypesOrError(unwrapTypeFactory(callBinding), callContext, operandTypes);
            }
        } catch (ValidationException e) {
            // let operand checker fail
        } catch (Throwable t) {
            throw createUnexpectedException(callContext, t);
        }
    }

    // --------------------------------------------------------------------------------------------

    private void inferOperandTypesOrError(
            FlinkTypeFactory typeFactory, CallContext callContext, RelDataType[] operandTypes) {
        final List<DataType> expectedDataTypes;

        // Static arguments have the highest priority
        final List<StaticArgument> staticArgs = typeInference.getStaticArguments().orElse(null);
        if (staticArgs != null) {
            expectedDataTypes =
                    staticArgs.stream()
                            .map(staticArg -> staticArg.getDataType().orElse(null))
                            .collect(Collectors.toList());
        } else {
            expectedDataTypes =
                    typeInference
                            .getInputTypeStrategy()
                            .inferInputTypes(callContext, false)
                            .orElse(null);
        }


Review comment (bug):
rethrowing here as `CalciteContextException` wrapped into `ValidationException` will not help for complex cases and will lead again to `StackOverflow`: first it rethrows as `ValidationException` and then swallows.

For that reason there is `testNestedCoalesceOnInvalidField` highlighting this case with  a query
```sql
SELECT coalesce(SELECT coalesce(SELECT coalesce(invalid)))
```

Code after:
            FunctionDefinition definition,
            TypeInference typeInference) {
        this.dataTypeFactory = dataTypeFactory;
        this.definition = definition;
        this.typeInference = typeInference;
    }

    @Override
    public void inferOperandTypes(
            SqlCallBinding callBinding, RelDataType returnType, RelDataType[] operandTypes) {
        final CallContext callContext =
                new CallBindingCallContext(
                        dataTypeFactory,
                        definition,
                        callBinding,
                        returnType,
                        typeInference.getStaticArguments().orElse(null));
        try {
            if (TypeInferenceUtil.validateArgumentCount(
                    typeInference.getInputTypeStrategy().getArgumentCount(),
                    callContext.getArgumentDataTypes().size(),
                    false)) {
                inferOperandTypesOrError(unwrapTypeFactory(callBinding), callContext, operandTypes);
            }
        } catch (ValidationException e) {
            // let operand checker fail
        } catch (CalciteContextException e) {
            throw e;
        } catch (Throwable t) {
            throw createUnexpectedException(callContext, t);
        }
    }

    // --------------------------------------------------------------------------------------------

    private void inferOperandTypesOrError(
            FlinkTypeFactory typeFactory, CallContext callContext, RelDataType[] operandTypes) {
        final List<DataType> expectedDataTypes;

        // Static arguments have the highest priority
        final List<StaticArgument> staticArgs = typeInference.getStaticArguments().orElse(null);
        if (staticArgs != null) {
            expectedDataTypes =
                    staticArgs.stream()
                            .map(staticArg -> staticArg.getDataType().orElse(null))
                            .collect(Collectors.toList());
        } else {
            expectedDataTypes =
                    typeInference
                            .getInputTypeStrategy()
                            .inferInputTypes(callContext, false)