Skip to content

Commit

Permalink
Merge pull request #1749 from ashendes/master
Browse files Browse the repository at this point in the history 
Primary key violation during persisted aggregation fixed.
  • Loading branch information
@dnwick
dnwick authored Oct 1, 2021
2 parents 1f519ae + 87dda7f commit 43565dd
Show file tree
Hide file tree
Showing 2 changed files with 20 additions and 8 deletions.
20 changes: 16 additions & 4 deletions modules/siddhi-core/src/main/java/io/siddhi/core/util/parser/AggregationParser.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1229,6 +1229,7 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
StringJoiner subSelectT2OnConditionBuilder = new StringJoiner(SQL_AND);

StringJoiner groupByQueryBuilder = new StringJoiner(", ");
StringJoiner groupByT3QueryBuilder = new StringJoiner(", ");
StringJoiner finalSelectQuery = new StringJoiner(" ");
StringJoiner completeQuery = new StringJoiner(" ");
StringJoiner insertIntoColumns = new StringJoiner(", ");
Expand Down Expand Up @@ -1257,6 +1258,7 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
if (isDistributed) {
filterQueryBuilder.append(" AND ").append(AGG_SHARD_ID_COL).append(" = '").append(shardID).append("' ");
groupByQueryBuilder.add(AGG_SHARD_ID_COL);
groupByT3QueryBuilder.add(INNER_SELECT_QUERY_REF_T3 + "." + AGG_SHARD_ID_COL);
innerSelectT2ColumnJoiner.add(AGG_SHARD_ID_COL);
subSelectT2OnConditionBuilder.add(parentAggregationTable.getTableDefinition().getId() + "." +
AGG_SHARD_ID_COL + EQUALS + INNER_SELECT_QUERY_REF_T3 + "." + AGG_SHARD_ID_COL);
Expand All @@ -1270,6 +1272,7 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
groupByVariableList.stream().forEach(variable -> {
groupByColumnNames.add(variable.getAttributeName());
groupByQueryBuilder.add(variable.getAttributeName());
groupByT3QueryBuilder.add(INNER_SELECT_QUERY_REF_T3 + "." + variable.getAttributeName());
onConditionBuilder.add(SUB_SELECT_QUERY_REF_T1 + "." + variable.getAttributeName() +
EQUALS + SUB_SELECT_QUERY_REF_T2 + "." + variable.getAttributeName());
subSelectT2OnConditionBuilder.add(parentAggregationTable.getTableDefinition().getId() + "." +
Expand Down Expand Up @@ -1298,7 +1301,10 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
subSelectT1ColumnJoiner.add(variableExpressionExecutor.getAttribute().getName());
innerSelectT2ColumnJoiner.add(variableExpressionExecutor.getAttribute().getName());
} else {
subSelectT2ColumnJoiner.add(variableExpressionExecutor.getAttribute().getName());
subSelectT2ColumnJoiner.add(dbAggregationSelectFunctionTemplates.getMaxFunction().
replace(PLACEHOLDER_COLUMN, variableExpressionExecutor.
getAttribute().getName()) + SQL_AS + variableExpressionExecutor.
getAttribute().getName());
outerSelectColumnJoiner.add(SUB_SELECT_QUERY_REF_T2 + "." +
variableExpressionExecutor.getAttribute().getName() +
SQL_AS + attributeList.get(i).getName());
Expand All @@ -1325,8 +1331,9 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
if (attributeList.get(i).getName().equals(AGG_LAST_TIMESTAMP_COL)) {
innerSelectT2ColumnJoiner.add(dbAggregationSelectFunctionTemplates.getMaxFunction().
replace(PLACEHOLDER_COLUMN, attributeList.get(i).getName()) + SQL_AS + attributeList.get(i).getName());
subSelectT2ColumnJoiner.add(INNER_SELECT_QUERY_REF_T3 + "." + attributeList.get(i).getName() +
SQL_AS + attributeList.get(i).getName());
subSelectT2ColumnJoiner.add(dbAggregationSelectFunctionTemplates.getMaxFunction().
replace(PLACEHOLDER_COLUMN, INNER_SELECT_QUERY_REF_T3 + "." +
attributeList.get(i).getName()) + SQL_AS + attributeList.get(i).getName());
outerSelectColumnJoiner.add(SUB_SELECT_QUERY_REF_T2 + "." + attributeList.get(i).getName() +
SQL_AS + attributeList.get(i).getName());
subSelectT2OnConditionBuilder.add(parentAggregationTable.getTableDefinition().getId() + "." +
Expand Down Expand Up @@ -1361,6 +1368,10 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
replace(PLACEHOLDER_COLUMN, AGG_EXTERNAL_TIMESTAMP_COL).replace(PLACEHOLDER_DURATION,
dbAggregationTimeConversionDurationMapping.getDurationMapping(duration)));

groupByT3QueryBuilder.add(dbAggregationSelectFunctionTemplates.getTimeConversionFunction().
replace(PLACEHOLDER_COLUMN, AGG_EXTERNAL_TIMESTAMP_COL).replace(PLACEHOLDER_DURATION,
dbAggregationTimeConversionDurationMapping.getDurationMapping(duration)));

groupByClause = dbAggregationSelectQueryTemplate.getGroupByClause().replace(PLACEHOLDER_COLUMNS,
groupByQueryBuilder.toString());

Expand All @@ -1376,7 +1387,8 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
replace(PLACEHOLDER_TABLE_NAME, parentAggregationTable.getTableDefinition().getId()).
replace(PLACEHOLDER_INNER_QUERY_2, innerT2Query.toString()).
replace(PLACEHOLDER_ON_CONDITION, subSelectT2OnConditionBuilder.toString()).
replace(PLACEHOLDER_CONDITION, innerT2WhereCondition));
replace(PLACEHOLDER_CONDITION, innerT2WhereCondition).
replace(PLACEHOLDER_COLUMNS, groupByT3QueryBuilder.toString()));


finalSelectQuery.add(dbAggregationSelectQueryTemplate.getJoinClause().
Expand Down
8 changes: 4 additions & 4 deletions modules/siddhi-core/src/main/resources/db-aggregation-config.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
<selectClause>SELECT {{SELECTORS}} FROM {{TABLE_NAME}}</selectClause>
<whereClause>WHERE {{CONDITION}}</whereClause>
<selectQueryWithInnerSelect>SELECT {{SELECTORS}} FROM {{TABLE_NAME}} LEFT JOIN ({{INNER_QUERY_2}}) AS t3 ON
{{ON_CONDITION}} WHERE {{CONDITION}}
{{ON_CONDITION}} WHERE {{CONDITION}} GROUP BY {{COLUMNS}}
</selectQueryWithInnerSelect>
<joinClause>SELECT {{SELECTORS}} FROM ({{FROM_CONDITION}}) AS t1 JOIN ({{INNER_QUERY_1}}) AS t2 ON
{{CONDITION}}
Expand Down Expand Up @@ -46,7 +46,7 @@
<selectClause>SELECT {{SELECTORS}} FROM {{TABLE_NAME}}</selectClause>
<whereClause>WHERE {{CONDITION}}</whereClause>
<selectQueryWithInnerSelect>SELECT {{SELECTORS}} FROM {{TABLE_NAME}} LEFT JOIN ({{INNER_QUERY_2}}) t3 ON
{{ON_CONDITION}} WHERE {{CONDITION}}
{{ON_CONDITION}} WHERE {{CONDITION}} GROUP BY {{COLUMNS}}
</selectQueryWithInnerSelect>
<joinClause>SELECT {{SELECTORS}} FROM ({{FROM_CONDITION}}) t1 JOIN ({{INNER_QUERY_1}}) t2 ON
{{CONDITION}}
Expand Down Expand Up @@ -88,7 +88,7 @@
<selectClause>SELECT {{SELECTORS}} FROM {{TABLE_NAME}}</selectClause>
<whereClause>WHERE {{CONDITION}}</whereClause>
<selectQueryWithInnerSelect>SELECT {{SELECTORS}} FROM {{TABLE_NAME}} LEFT JOIN ({{INNER_QUERY_2}}) AS t3 ON
{{ON_CONDITION}} WHERE {{CONDITION}}
{{ON_CONDITION}} WHERE {{CONDITION}} GROUP BY {{COLUMNS}}
</selectQueryWithInnerSelect>
<joinClause>SELECT {{SELECTORS}} FROM ({{FROM_CONDITION}}) AS t1 JOIN ({{INNER_QUERY_1}}) AS t2 ON
{{CONDITION}}
Expand Down Expand Up @@ -127,7 +127,7 @@
<selectClause>SELECT {{SELECTORS}} FROM {{TABLE_NAME}}</selectClause>
<whereClause>WHERE {{CONDITION}}</whereClause>
<selectQueryWithInnerSelect>SELECT {{SELECTORS}} FROM {{TABLE_NAME}} LEFT JOIN ({{INNER_QUERY_2}}) AS t3 ON
{{ON_CONDITION}} WHERE {{CONDITION}}
{{ON_CONDITION}} WHERE {{CONDITION}} GROUP BY {{COLUMNS}}
</selectQueryWithInnerSelect>
<joinClause>SELECT {{SELECTORS}} FROM ({{FROM_CONDITION}}) AS t1 JOIN ({{INNER_QUERY_1}}) AS t2 ON
{{CONDITION}}
Expand Down

0 comments on commit 43565dd

Please sign in to comment.