mattyb149 commented on a change in pull request #3977: NIFI-7007 Add update
functionality to the PutCassandraRecord processor.
URL: https://github.com/apache/nifi/pull/3977#discussion_r383290090
##########
File path:
nifi-nar-bundles/nifi-cassandra-bundle/nifi-cassandra-processors/src/main/java/org/apache/nifi/processors/cassandra/PutCassandraRecord.java
##########
@@ -193,6 +319,106 @@ public void onTrigger(ProcessContext context,
ProcessSession session) throws Pro
}
+ protected Statement generateUpdate(String cassandraTable, RecordSchema
schema, String updateKeys, String updateMethod, Map<String, Object>
recordContentMap) {
+ Update updateQuery;
+
+ // Split up the update key names separated by a comma, should not be
empty
+ final Set<String> updateKeyNames;
+ updateKeyNames = Arrays.stream(updateKeys.split(","))
+ .map(String::trim)
+ .filter(StringUtils::isNotEmpty)
+ .collect(Collectors.toSet());
+ if (updateKeyNames.isEmpty()) {
+ throw new IllegalArgumentException("No Update Keys were
specified");
+ }
+
+ // Verify if all update keys are present in the record
+ for (String updateKey : updateKeyNames) {
+ if (!schema.getFieldNames().contains(updateKey)) {
+ throw new IllegalArgumentException("Update key '" + updateKey
+ "' is not present in the record schema");
+ }
+ }
+
+ // Prepare keyspace/table names
+ if (cassandraTable.contains(".")) {
+ String[] keyspaceAndTable = cassandraTable.split("\\.");
+ updateQuery = QueryBuilder.update(keyspaceAndTable[0],
keyspaceAndTable[1]);
+ } else {
+ updateQuery = QueryBuilder.update(cassandraTable);
+ }
+
+ // Loop through the field names, setting those that are not in the
update key set, and using those
+ // in the update key set as conditions.
+ for (String fieldName : schema.getFieldNames()) {
+ Object fieldValue = recordContentMap.get(fieldName);
+
+ if (updateKeyNames.contains(fieldName)) {
+ updateQuery.where(QueryBuilder.eq(fieldName, fieldValue));
+ } else {
+ Assignment assignment;
+ if (SET_TYPE.getValue().equalsIgnoreCase(updateMethod)) {
+ assignment = QueryBuilder.set(fieldName, fieldValue);
+ } else if
(INCR_TYPE.getValue().equalsIgnoreCase(updateMethod)) {
+ assignment = QueryBuilder.incr(fieldName,
convertFieldObjectToLong(fieldName, fieldValue));
+ } else if
(DECR_TYPE.getValue().equalsIgnoreCase(updateMethod)) {
+ assignment = QueryBuilder.decr(fieldName,
convertFieldObjectToLong(fieldName, fieldValue));
+ } else {
+ throw new IllegalArgumentException("Update Method '" +
updateMethod + "' is not valid.");
+ }
+ updateQuery.with(assignment);
+ }
+ }
+ return updateQuery;
+ }
+
+ private Long convertFieldObjectToLong(String name, Object value) {
+ if (!(value instanceof Number)) {
+ throw new IllegalArgumentException("Field '" + name + "' is not of
type Number");
+ }
+ return ((Number) value).longValue();
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext
validationContext) {
+ Set<ValidationResult> results = (Set<ValidationResult>)
super.customValidate(validationContext);
+
+ String statementType =
validationContext.getProperty(STATEMENT_TYPE).getValue();
+
+ if (UPDATE_TYPE.getValue().equalsIgnoreCase(statementType)) {
+ // Check that update keys are set
+ String updateKeys =
validationContext.getProperty(UPDATE_KEYS).getValue();
+ if (StringUtils.isEmpty(updateKeys)) {
+ results.add(new ValidationResult.Builder().subject("Update
statement configuration").valid(false).explanation(
+ "if the Statement Type is set to Update, then the
Update Keys must be specified as well").build());
+ }
+
+ // Check that if the update method is set to increment or
decrement that the batch statement type is set to
+ // unlogged or counter (or USE_ATTR_TYPE, which we cannot check at
this point).
+ String batchStatementType =
validationContext.getProperty(BATCH_STATEMENT_TYPE).getValue();
+ if (!Sets.newHashSet(COUNTER_TYPE.getValue(),
UNLOGGED_TYPE.getValue(),
BATCH_STATEMENT_TYPE_USE_ATTR_TYPE.getValue()).contains(batchStatementType)) {
Review comment:
There's a logic error here. If my Batch Statement Type is LOGGED but my
Update Type is "Set", the processor is invalid due to the line below. Probably
should just have some explicit if-checks rather than a Set.contains() here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
