public static void checkPhysicalLogicalTypeCompatible(
            LogicalType physicalFieldType,
            LogicalType logicalFieldType,
            String physicalFieldName,
            String logicalFieldName,
            boolean isSource) {
        if (isSource) {
            checkIfCompatible(
                    physicalFieldType,
                    logicalFieldType,
                    (cause) ->
                            new ValidationException(
                                    String.format(
                                            "Type %s of table field '%s' does not match with "
                                                    + "the physical type %s of the '%s' field of the TableSource return type.",
                                            logicalFieldType,
                                            logicalFieldName,
                                            physicalFieldType,
                                            physicalFieldName),
                                    cause));
        } else {
            checkIfCompatible(
                    logicalFieldType,
                    physicalFieldType,
                    (cause) ->
                            new ValidationException(
                                    String.format(
                                            "Type %s of table field '%s' does not match with "
                                                    + "the physical type %s of the '%s' field of the TableSink consumed type.",
                                            logicalFieldType,
                                            logicalFieldName,
                                            physicalFieldType,
                                            physicalFieldName),
                                    cause));
        }
    }
@Test
    void testCheckPhysicalLogicalTypeCompatible() {
        TableSchema tableSchema =
                TableSchema.builder()
                        .field("a", DataTypes.VARCHAR(2))
                        .field("b", DataTypes.DECIMAL(20, 2))
                        .build();
        TableSink tableSink = new TestTableSink(tableSchema);
        LegacyTypeInformationType legacyDataType =
                (LegacyTypeInformationType) tableSink.getConsumedDataType().getLogicalType();
        TypeInformation legacyTypeInfo =
                ((TupleTypeInfo) legacyDataType.getTypeInformation()).getTypeAt(1);
        DataType physicalType = TypeConversions.fromLegacyInfoToDataType(legacyTypeInfo);
        ResolvedSchema physicSchema = DataTypeUtils.expandCompositeTypeToSchema(physicalType);
        DataType[] logicalDataTypes = tableSchema.getFieldDataTypes();
        List<DataType> physicalDataTypes = physicSchema.getColumnDataTypes();
        for (int i = 0; i < logicalDataTypes.length; i++) {
            TypeMappingUtils.checkPhysicalLogicalTypeCompatible(
                    physicalDataTypes.get(i).getLogicalType(),
                    logicalDataTypes[i].getLogicalType(),
                    "physicalField",
                    "logicalField",
                    false);
        }
    }