NIFI-15474 Support timestamp truncation in RecordPath DSL

This closes #10796.

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

Author: r1b

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Divide.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.functions;
+
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPathEvaluationContext;
+import org.apache.nifi.record.path.math.MathBinaryEvaluator;
+import org.apache.nifi.record.path.paths.RecordPathSegment;
+
+import java.util.stream.Stream;
+
+public class Divide extends RecordPathSegment {
+    private final RecordPathSegment lhsPath;
+    private final RecordPathSegment rhsPath;
+    private final MathBinaryEvaluator divide = MathBinaryEvaluator.divide();
+
+    public Divide(final RecordPathSegment lhsPath, final RecordPathSegment rhsPath, final boolean absolute) {
+        super("divide", null, absolute);
+        this.lhsPath = lhsPath;
+        this.rhsPath = rhsPath;
+    }
+
+    @Override
+    public Stream<FieldValue> evaluate(RecordPathEvaluationContext context) {
+        final FieldValue lhs = lhsPath.evaluate(context).findFirst().orElseThrow(() -> new IllegalArgumentException("divide function requires a left-hand operand"));
+        final FieldValue rhs = rhsPath.evaluate(context).findFirst().orElseThrow(() -> new IllegalArgumentException("divide function requires a right-hand operand"));
+
+        if (lhs.getValue() == null || rhs.getValue() == null) {
+            return Stream.of();
+        }
+
+        return Stream.of(divide.evaluate(lhs, rhs));
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Multiply.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.functions;
+
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPathEvaluationContext;
+import org.apache.nifi.record.path.math.MathBinaryEvaluator;
+import org.apache.nifi.record.path.paths.RecordPathSegment;
+
+import java.util.stream.Stream;
+
+public class Multiply extends RecordPathSegment {
+    private final RecordPathSegment lhsPath;
+    private final RecordPathSegment rhsPath;
+    private final MathBinaryEvaluator multiply = MathBinaryEvaluator.multiply();
+
+    public Multiply(final RecordPathSegment lhsPath, final RecordPathSegment rhsPath, final boolean absolute) {
+        super("multiply", null, absolute);
+        this.lhsPath = lhsPath;
+        this.rhsPath = rhsPath;
+    }
+
+    @Override
+    public Stream<FieldValue> evaluate(RecordPathEvaluationContext context) {
+        final FieldValue lhs = lhsPath.evaluate(context).findFirst().orElseThrow(() -> new IllegalArgumentException("multiply function requires a left-hand operand"));
+        final FieldValue rhs = rhsPath.evaluate(context).findFirst().orElseThrow(() -> new IllegalArgumentException("multiply function requires a right-hand operand"));
+
+        if (lhs.getValue() == null || rhs.getValue() == null) {
+            return Stream.of();
+        }
+
+        return Stream.of(multiply.evaluate(lhs, rhs));
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/ToNumber.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.functions;
+
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPathEvaluationContext;
+import org.apache.nifi.record.path.math.MathTypeUtils;
+import org.apache.nifi.record.path.paths.RecordPathSegment;
+
+import java.util.stream.Stream;
+
+public class ToNumber extends RecordPathSegment {
+    private final RecordPathSegment valuePath;
+
+    public ToNumber(final RecordPathSegment valuePath, final boolean absolute) {
+        super("toNumber", null, absolute);
+        this.valuePath = valuePath;
+    }
+
+    @Override
+    public Stream<FieldValue> evaluate(RecordPathEvaluationContext context) {
+        final FieldValue fieldValue = valuePath.evaluate(context).findFirst().orElseThrow(() -> new IllegalArgumentException("toNumber function requires an operand"));
+        final Object value = fieldValue.getValue();
+
+        if (value == null || value instanceof Number) {
+            return Stream.of(fieldValue);
+        } else {
+            return Stream.of(MathTypeUtils.toNumber(fieldValue));
+        }
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/math/MathBinaryEvaluator.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.math;
+
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.StandardFieldValue;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.RecordFieldType;
+
+public class MathBinaryEvaluator extends MathEvaluator<MathBinaryOperator> {
+    public MathBinaryEvaluator(MathBinaryOperator op) {
+        super(op);
+    }
+
+    public static MathBinaryEvaluator divide() {
+        return new MathBinaryEvaluator(new MathDivideOperator());
+    }
+
+    public static MathBinaryEvaluator multiply() {
+        return new MathBinaryEvaluator(new MathMultiplyOperator());
+    }
+
+    public FieldValue evaluate(FieldValue lhs, FieldValue rhs) {
+        final Number lhsValue = MathTypeUtils.coerceNumber(lhs);
+        final Number rhsValue = MathTypeUtils.coerceNumber(rhs);
+
+        Number result;
+        DataType resultType;
+
+        if (MathTypeUtils.isLongCompatible(lhsValue) && MathTypeUtils.isLongCompatible(rhsValue)) {
+            result = op.operate(lhsValue.longValue(), rhsValue.longValue());
+            resultType = RecordFieldType.LONG.getDataType();
+        } else {
+            result = op.operate(lhsValue.doubleValue(), rhsValue.doubleValue());
+            resultType = RecordFieldType.DOUBLE.getDataType();
+        }
+
+        return new StandardFieldValue(result, new RecordField(op.getFieldName(), resultType), null);
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/math/MathBinaryOperator.java

@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.math;
+
+public interface MathBinaryOperator extends MathOperator {
+    Long operate(Long n, Long m);
+    Double operate(Double n, Double m);
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/math/MathDivideOperator.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.math;
+
+public class MathDivideOperator implements MathBinaryOperator {
+    @Override
+    public Long operate(Long n, Long m) {
+        if (m == 0L) {
+            throw new ArithmeticException("Division by zero in RecordPath divide function");
+        }
+        return n / m;
+    }
+
+    @Override
+    public Double operate(Double n, Double m) {
+        if (m == 0.0) {
+            throw new ArithmeticException("Division by zero in RecordPath divide function");
+        }
+        return n / m;
+    }
+
+    @Override
+    public String getFieldName() {
+        return "divide";
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/math/MathEvaluator.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.math;
+
+public class MathEvaluator<T extends MathOperator> {
+    protected final T op;
+
+    public MathEvaluator(T op) {
+        this.op = op;
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/math/MathMultiplyOperator.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.math;
+
+public class MathMultiplyOperator implements MathBinaryOperator {
+    @Override
+    public Long operate(Long n, Long m) {
+        return n * m;
+    }
+
+    @Override
+    public Double operate(Double n, Double m) {
+        return n * m;
+    }
+
+    @Override
+    public String getFieldName() {
+        return "multiply";
+    }
+}

nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/math/MathOperator.java

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.record.path.math;
+
+public interface MathOperator {
+    String getFieldName();
+}