Siddhi (wso2sp) -在rdbs存储上执行多个插入操作时遇到DuplicateDefinitionException

Question

我有一个siddhi应用程序(wso2sp 4.4.0)，一旦在一个商店中添加第二个插入操作，它就无法部署。有一个更新或插入，然后在同一存储上更新第二个流没有问题，但我需要它是一个更新或插入以及。我做错了什么？

上下文:我希望有多个kafka主题源，每个都根据不同的有效负载更新同一个表上的不同字段(因此使用'set')，但是由于它们中的任何一个都可能首先发生，所以它们都需要更新或插入到操作中，而不仅仅是更新。更新不会导致错误，但是让insert into确实会导致错误。

[2022-01-21 14:46:02,582] ERROR {org.wso2.carbon.stream.processor.core.internal.StreamProcessorDeployer} - org.wso2.siddhi.query.api.exception.DuplicateDefinitionException: Error on 'HumanTasks' @ Line: 37. 
Position: 35, near 'update or insert into humanTaskEvents
set humanTasksTable.taskCompletionDate = taskCompletionDate, humanTasksTable.taskCompletionDateEpochValue = taskCompletionDateEpochValue, humanTasksTable.taskCompletionAction = taskCompletionAction
on humanTaskEvents.taskId == taskId'. Error between @ Line: 35. Position: 0 and @ Line: 37. Position: 35. Different definition same as output 'define stream humanTaskEvents (taskId string, taskCompletionDate string, taskCompletionDateEpochValue string, taskCompletionAction string)' already exist as '@Store( type = "rdbms", datasource = "eventProcessRDS")define table humanTaskEvents (taskId string, taskName string, correlationId string, identifierName string, identifierValue string, groupName string, readOnlyHumanTask string, task string, timeoutTask string, taskCreationDate string, taskCompletionDate string, taskCreationDateEpochValue string, taskCompletionDateEpochValue string, taskCompletionAction string)' org.wso2.carbon.stream.processor.core.internal.exception.SiddhiAppDeploymentException: org.wso2.siddhi.query.api.exception.DuplicateDefinitionException: Error on 'HumanTasks' @ Line: 37. Position: 35, near 'update or insert into humanTaskEvents
set humanTasksTable.taskCompletionDate = taskCompletionDate, humanTasksTable.taskCompletionDateEpochValue = taskCompletionDateEpochValue, humanTasksTable.taskCompletionAction = taskCompletionAction
on humanTaskEvents.taskId == taskId'. Error between @ Line: 35. Position: 0 and @ Line: 37. Position: 35. Different definition same as output 'define stream humanTaskEvents (taskId string, taskCompletionDate string, taskCompletionDateEpochValue string, taskCompletionAction string)' already exist as '@Store( type = "rdbms", datasource = "eventProcessRDS")define table humanTaskEvents (taskId string, taskName string, correlationId string, identifierName string, identifierValue string, groupName string, readOnlyHumanTask string, task string, timeoutTask string, taskCreationDate string, taskCompletionDate string, taskCreationDateEpochValue string, taskCompletionDateEpochValue string, taskCompletionAction string)'
        at org.wso2.carbon.stream.processor.core.internal.StreamProcessorDeployer.deploySiddhiQLFile(StreamProcessorDeployer.java:106)
        at org.wso2.carbon.stream.processor.core.internal.StreamProcessorDeployer.deploy(StreamProcessorDeployer.java:330)
        at org.wso2.carbon.deployment.engine.internal.DeploymentEngine.lambda$deployArtifacts$0(DeploymentEngine.java:291)
        at java.util.ArrayList.forEach(ArrayList.java:1257)
        at org.wso2.carbon.deployment.engine.internal.DeploymentEngine.deployArtifacts(DeploymentEngine.java:282)
        at org.wso2.carbon.deployment.engine.internal.RepositoryScanner.sweep(RepositoryScanner.java:112)
        at org.wso2.carbon.deployment.engine.internal.RepositoryScanner.scan(RepositoryScanner.java:68)
        at org.wso2.carbon.deployment.engine.internal.DeploymentEngine.start(DeploymentEngine.java:121)
        at org.wso2.carbon.deployment.engine.internal.DeploymentEngineListenerComponent.onAllRequiredCapabilitiesAvailable(DeploymentEngineListenerComponent.java:216)
        at org.wso2.carbon.kernel.internal.startupresolver.StartupComponentManager.lambda$notifySatisfiableComponents$7(StartupComponentManager.java:266)
        at java.util.ArrayList.forEach(ArrayList.java:1257)
        at org.wso2.carbon.kernel.internal.startupresolver.StartupComponentManager.notifySatisfiableComponents(StartupComponentManager.java:252)
        at org.wso2.carbon.kernel.internal.startupresolver.StartupOrderResolver$1.run(StartupOrderResolver.java:204)
        at java.util.TimerThread.mainLoop(Timer.java:555)
        at java.util.TimerThread.run(Timer.java:505)
Caused by: org.wso2.siddhi.query.api.exception.DuplicateDefinitionException: Error on 'HumanTasks' @ Line: 37. Position: 35, near 'update or insert into humanTaskEvents
set humanTasksTable.taskCompletionDate = taskCompletionDate, humanTasksTable.taskCompletionDateEpochValue = taskCompletionDateEpochValue, humanTasksTable.taskCompletionAction = taskCompletionAction
on humanTaskEvents.taskId == taskId'. Error between @ Line: 35. Position: 0 and @ Line: 37. Position: 35. Different definition same as output 'define stream humanTaskEvents (taskId string, taskCompletionDate string, taskCompletionDateEpochValue string, taskCompletionAction string)' already exist as '@Store( type = "rdbms", datasource = "eventProcessRDS")define table humanTaskEvents (taskId string, taskName string, correlationId string, identifierName string, identifierValue string, groupName string, readOnlyHumanTask string, task string, timeoutTask string, taskCreationDate string, taskCompletionDate string, taskCreationDateEpochValue string, taskCompletionDateEpochValue string, taskCompletionAction string)'
        at org.wso2.siddhi.core.util.parser.QueryParser.parse(QueryParser.java:225)
        at org.wso2.siddhi.core.util.parser.SiddhiAppParser.parse(SiddhiAppParser.java:245)
        at org.wso2.siddhi.core.SiddhiManager.createSiddhiAppRuntime(SiddhiManager.java:65)
        at org.wso2.siddhi.core.SiddhiManager.createSiddhiAppRuntime(SiddhiManager.java:74)
        at org.wso2.carbon.stream.processor.core.internal.StreamProcessorService.deploySiddhiApp(StreamProcessorService.java:100)
        at org.wso2.carbon.stream.processor.core.internal.StreamProcessorDeployer.deploySiddhiQLFile(StreamProcessorDeployer.java:94)
        ... 14 more
Caused by: org.wso2.siddhi.query.api.exception.DuplicateDefinitionException: Error between @ Line: 35. Position: 0 and @ Line: 37. Position: 35. Different definition same as output 'define stream humanTaskEvents (taskId string, taskCompletionDate string, taskCompletionDateEpochValue string, taskCompletionAction string)' already exist as '@Store( type = "rdbms", datasource = "eventProcessRDS")define table humanTaskEvents (taskId string, taskName string, correlationId string, identifierName string, identifierValue string, groupName string, readOnlyHumanTask string, task string, timeoutTask string, taskCreationDate string, taskCompletionDate string, taskCreationDateEpochValue string, taskCompletionDateEpochValue string, taskCompletionAction string)'
        at org.wso2.siddhi.core.util.parser.helper.DefinitionParserHelper.validateOutputStream(DefinitionParserHelper.java:156)
        at org.wso2.siddhi.core.util.parser.OutputParser.constructOutputCallback(OutputParser.java:199)
        at org.wso2.siddhi.core.util.parser.QueryParser.parse(QueryParser.java:196)
        ... 19 more

和代码：

@App:name("HumanTasks")
@App:description("testing")


@Store(type='rdbms', datasource='eventProcessRDS')
define table humanTaskEvents (taskId String, taskName String, correlationId String, identifierName String, identifierValue String, groupName String, readOnlyHumanTask String, task String, timeoutTask String, taskCreationDate String, taskCompletionDate String, taskCreationDateEpochValue String, taskCompletionDateEpochValue String, taskCompletionAction String);

@source(ref='sourceKafkaHumanTaskCreationEvent',
    @map(type = 'json', fail.on.missing.attribute= "false",
        @attributes(taskId = "taskId", taskName = "taskName", identifierName = "identifierName", correlationId = "correlationId", identifierValue = "identifierValue", groupName = "groupName", readOnlyHumanTask = "readOnlyUserTask", task = "task", timeoutTask = "timeoutTask", taskCreationDate = "taskCreationDate", taskCompletionDate = "taskCompletionDate", taskCreationDateEpochValue = "taskCreationDateEpochValue", taskCompletionDateEpochValue = "taskCompletionDateEpochValue", taskCompletionAction = "taskCompletionAction")))
define stream humanTaskCreation (taskId String, taskName String, correlationId String, identifierName String, identifierValue String, groupName String, readOnlyHumanTask String, task String, timeoutTask String, taskCreationDate String, taskCompletionDate String, taskCreationDateEpochValue String, taskCompletionDateEpochValue String, taskCompletionAction String);

@source(ref='sourceKafkaHumanTaskCompletionEvent',
    @map(type = 'json', fail.on.missing.attribute= "false",
        @attributes(taskId = "taskId", taskCompletionDate = "taskCompletionDate", taskCompletionDateEpochValue = "taskCompletionDateEpochValue", taskCompletionAction = "taskCompletionAction")))
define stream humanTaskCompletion (taskId String, taskCompletionDate String, taskCompletionDateEpochValue String, taskCompletionAction String);


------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------


from humanTaskCreation
select *
update or insert into humanTaskEvents
set humanTasksTable.taskName = taskName, humanTasksTable.correlationId = correlationId, humanTasksTable.identifierName = identifierName, humanTasksTable.identifierValue = identifierValue, humanTasksTable.groupName = groupName, humanTasksTable.readOnlyHumanTask = readOnlyHumanTask, humanTasksTable.task = task, humanTasksTable.timeoutTask = timeoutTask, humanTasksTable.taskCreationDate = taskCreationDate, humanTasksTable.taskCreationDateEpochValue = taskCreationDateEpochValue
on humanTaskEvents.taskId == taskId;

from humanTaskCompletion
select taskId, taskCompletionDate, taskCompletionDateEpochValue, taskCompletionAction
update or insert into humanTaskEvents
set humanTasksTable.taskCompletionDate = taskCompletionDate, humanTasksTable.taskCompletionDateEpochValue = taskCompletionDateEpochValue, humanTasksTable.taskCompletionAction = taskCompletionAction
on humanTaskEvents.taskId == taskId;

Answer 1

如果要执行更新或插入，则必须为humanTaskEvents表中的所有列提供一个值。因为如果更新失败，它将尝试将该事件添加到表中，如果没有设置所有列，siddhi将无法向表中插入部分事件。