When and how latest state of source data is updated in a rule ?

[younes199511]

Hello,
I've recently created a rule utilizing the latest analytic function within ekuiper, and I've encountered some behaviors that I'm struggling to comprehend. Specifically, I'm seeking clarity on the process and timing of how the latest state of the source data is updated within a rule.

I have the following graph rule:

{
    "id": "1",
    "name": "Test",
    "graph": {
        "nodes": {
            "demo": {
                "type": "source",
                "nodeType": "edgex",
                "props": {}
            },
            "renamePick": {
                "type": "operator",
                "nodeType": "pick",
                "props": {
                    "fields": [
                        "case when meta(`RoomTemperature`->deviceName)=\"Modbus-2\" then RoomTemperature end as var1",
                        "case when meta(`Temperature`->deviceName)=\"Modbus-2\" then Temperature end as var2"
                    ]
                }
            },
            "inputFilter": {
                "type": "operator",
                "nodeType": "filter",
                "props": {
                    "expr": "var1 != nil OR var2 != nil"
                }
            },
            "compFunction": {
                "type": "operator",
                "nodeType": "function",
                "props": {
                    "expr": "cast(latest(var1) != latest(var2), \"boolean\") as comp_result"
                }
            },
            "outputFilter": {
                "type": "operator",
                "nodeType": "filter",
                "props": {
                    "expr": "latest(var1) != nil AND latest(var2) != nil"
                }
            },
            "pick": {
                "type": "operator",
                "nodeType": "pick",
                "props": {
                    "fields": [
                        "comp_result",
                        "latest(var1,var1) as latestVar1",
                        "latest(var2,var2) as latestVar2"
                    ]
                }
            },
            "logOut": {
                "type": "sink",
                "nodeType": "log",
                "props": {}
            }
        },
        "topo": {
            "sources": [
                "demo"
            ],
            "edges": {
                "demo": [
                    "renamePick"
                ],
                "renamePick": [
                    "inputFilter"
                ],
                "inputFilter": [
                    "compFunction"
                ],
                "compFunction": [
                    "outputFilter"
                ],
                "outputFilter": [
                    "pick"
                ],
                "pick": [
                    "logOut"
                ]
            }
        }
    }
}

In this rule, I am retrieving data from the Edgex source, where multiple auto events with different polling periods are present. I am utilizing filters to manage the rule execution, with the intention of having the rule (compFunction node) executed only when the source input has readings for var1 or var2, and to ignore other events. Subsequently, I aim to publish the result with the sink only when the compFunction node is calculated with not nil data for latest(va1) and latest(var2)

The observed behavior is as follows:

• If var1 is published first:

  • 1st execution: var1 is published with a value of 20, and I receive no output.
  • 2nd execution: var2 is published with a value of 20, and I receive [{"comp_result":false,"latestVar2":20}].
  • 3rd execution: var1 is published with a value of 20, and I receive [{"comp_result":false,"latestVar1":20,"latestVar2":20}].
    In this scenario, I am puzzled as to why I do not have the latestVar1 in the second execution, even though comp_result is correct and rule passes the outputFilter.

• If var2 is published first:

  • 1st execution: var2 is published with a value of 20, and I receive no output.
  • 2nd execution: var1 is published with a value of 20, and I receive no output.
  • 3rd execution: var2 is published with a value of 20, and I receive [{"comp_result":false,"latestVar2":20}].
  • 4th execution: var1 is published with a value of 20, and I receive [{"comp_result":false,"latestVar1":20,"latestVar2":20}].
    In this scenario, I am puzzled as to why I do not receive any output in the second execution, even though I have already published data for both variables.

I have attempted other options such as filtering the latest non-nil data in the input filter or adding a switch node instead of an output filter, but I consistently encounter the issue of my rule never publishing a result or exhibiting the same behavior as described above.

If anyone could provide insights or share their experiences, it would be incredibly helpful. Thank you all in advance for your assistance.

[ngjaying]

@younes199511 When using SQL, analytic function will be planned to run for each input. However, in graph API, there is no planning. Thus, you'll need to put it in appropriate position and refer to the result with alias value. Like source -> latest(var) as lvar -> other operators, refer to the latest value by lvar.

[younes199511]

I understand better now. Thank you