This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 98aaee2939 Issue 7023 : group by moving average (#7383)
98aaee2939 is described below
commit 98aaee2939a2c0f077ed222a6213ce8370546f36
Author: Matt Casters <[email protected]>
AuthorDate: Wed Jul 1 16:27:29 2026 +0200
Issue 7023 : group by moving average (#7383)
* issue #7118 : field selection for table view export to spreadsheet
* issue #7023 : Moving average for the group by transform
* issue #7023 : Moving average for the group by transform (IT fix)
---
.../ROOT/pages/pipeline/transforms/groupby.adoc | 1 +
.../transforms/0006-groupby-moving-average.hpl | 268 +++++++++++++++++++++
.../datasets/golden-groupby-moving-average.csv | 11 +
integration-tests/transforms/main-0006-groupby.hwf | 86 +++----
.../dataset/golden-groupby-moving-average.json | 40 +++
.../0006-groupby-moving-average UNIT.json | 41 ++++
.../pipeline/transforms/groupby/Aggregation.java | 45 +++-
.../hop/pipeline/transforms/groupby/GroupBy.java | 79 ++++++
.../pipeline/transforms/groupby/GroupByData.java | 14 ++
.../pipeline/transforms/groupby/GroupByDialog.java | 20 +-
.../pipeline/transforms/groupby/GroupByMeta.java | 1 +
.../groupby/messages/messages_en_US.properties | 6 +-
.../hop/pipeline/transforms/GroupByMetaTest.java | 3 +-
.../groupby/MovingAverageAggregationTest.java | 193 +++++++++++++++
14 files changed, 757 insertions(+), 51 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/groupby.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/groupby.adoc
index 6ba7f824da..2b139e181b 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/groupby.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/groupby.adoc
@@ -92,4 +92,5 @@ Here are the available aggregation methods :
- Percentile (nearest-rank method)
- Concatenate string separated by new line (CRLF)
- Concatenate distinct values separated by <Value>: specify the separator in
the Value column (This supports
xref::variables.adoc#_hexadecimal_values[hexadecimals])
+- Moving average (last N rows) (all rows option only!): computes the rolling
average over the last N events. Specify the window size N in the Value column
(e.g. 5) and the sequencing field in the Order field column (input data must be
pre-sorted).
|===
\ No newline at end of file
diff --git a/integration-tests/transforms/0006-groupby-moving-average.hpl
b/integration-tests/transforms/0006-groupby-moving-average.hpl
new file mode 100644
index 0000000000..ea81bd642f
--- /dev/null
+++ b/integration-tests/transforms/0006-groupby-moving-average.hpl
@@ -0,0 +1,268 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <pipeline_version/>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <pipeline_type>Normal</pipeline_type>
+ <pipeline_status>0</pipeline_status>
+ <parameters/>
+ <name>0006-groupby-moving-average</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <created_user>-</created_user>
+ <modified_user>-</modified_user>
+ <created_date>2026/06/30 20:50:00.000</created_date>
+ <modified_date>2026/06/30 20:50:00.000</modified_date>
+ </info>
+ <transform>
+ <type>DataGrid</type>
+ <name>Sample Data</name>
+ <fields>
+ <field>
+ <currency/>
+ <decimal/>
+ <group/>
+ <name>customer_id</name>
+ <type>String</type>
+ <format/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <currency/>
+ <decimal/>
+ <group/>
+ <name>timestamp</name>
+ <type>Integer</type>
+ <format/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <currency/>
+ <decimal/>
+ <group/>
+ <name>amount</name>
+ <type>Number</type>
+ <format/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <data>
+ <line>
+ <item>c1</item>
+ <item>1</item>
+ <item>10.0</item>
+ </line>
+ <line>
+ <item>c1</item>
+ <item>2</item>
+ <item>20.0</item>
+ </line>
+ <line>
+ <item>c1</item>
+ <item>3</item>
+ <item>30.0</item>
+ </line>
+ <line>
+ <item>c1</item>
+ <item>4</item>
+ <item>40.0</item>
+ </line>
+ <line>
+ <item>c1</item>
+ <item>5</item>
+ <item>50.0</item>
+ </line>
+ <line>
+ <item>c2</item>
+ <item>1</item>
+ <item>100.0</item>
+ </line>
+ <line>
+ <item>c2</item>
+ <item>2</item>
+ <item>200.0</item>
+ </line>
+ <line>
+ <item>c2</item>
+ <item>3</item>
+ <item/>
+ </line>
+ <line>
+ <item>c2</item>
+ <item>4</item>
+ <item>300.0</item>
+ </line>
+ <line>
+ <item>c2</item>
+ <item>5</item>
+ <item>400.0</item>
+ </line>
+ </data>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>GroupBy</type>
+ <name>Group by customer</name>
+ <all_rows>Y</all_rows>
+ <directory>${java.io.tmpdir}</directory>
+ <prefix>grp</prefix>
+ <ignore_aggregate>N</ignore_aggregate>
+ <field_ignore/>
+ <group>
+ <field>
+ <name>customer_id</name>
+ </field>
+ </group>
+ <fields>
+ <field>
+ <aggregate>moving_avg</aggregate>
+ <subject>amount</subject>
+ <type>MOVING_AVG</type>
+ <valuefield>3</valuefield>
+ <orderfield>timestamp</orderfield>
+ </field>
+ </fields>
+ <add_linenr>N</add_linenr>
+ <linenr_fieldname/>
+ <give_back_row>N</give_back_row>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>240</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>SelectValues</type>
+ <name>numbers to string</name>
+ <fields>
+ <select_unspecified>N</select_unspecified>
+ <meta>
+ <name>amount</name>
+ <rename>amount</rename>
+ <type>String</type>
+ <length>-2</length>
+ <precision>-2</precision>
+ <storage_type/>
+ <conversion_mask>0.0;-0.0</conversion_mask>
+ <date_format_lenient>N</date_format_lenient>
+ <date_format_locale/>
+ <date_format_timezone/>
+ <lenient_string_to_number>N</lenient_string_to_number>
+ <decimal_symbol>.</decimal_symbol>
+ <grouping_symbol/>
+ <currency_symbol/>
+ <encoding/>
+ </meta>
+ <meta>
+ <name>moving_avg</name>
+ <rename>moving_avg</rename>
+ <type>String</type>
+ <length>-2</length>
+ <precision>-2</precision>
+ <storage_type/>
+ <conversion_mask>0.0;-0.0</conversion_mask>
+ <date_format_lenient>N</date_format_lenient>
+ <date_format_locale/>
+ <date_format_timezone/>
+ <lenient_string_to_number>N</lenient_string_to_number>
+ <decimal_symbol>.</decimal_symbol>
+ <grouping_symbol/>
+ <currency_symbol/>
+ <encoding/>
+ </meta>
+ </fields>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>400</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>Dummy</type>
+ <name>Validate</name>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>560</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <order>
+ <hop>
+ <from>Sample Data</from>
+ <to>Group by customer</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Group by customer</from>
+ <to>numbers to string</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>numbers to string</from>
+ <to>Validate</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <notepads/>
+ <attributes/>
+ <transform_error_handling/>
+</pipeline>
diff --git
a/integration-tests/transforms/datasets/golden-groupby-moving-average.csv
b/integration-tests/transforms/datasets/golden-groupby-moving-average.csv
new file mode 100644
index 0000000000..8610d23782
--- /dev/null
+++ b/integration-tests/transforms/datasets/golden-groupby-moving-average.csv
@@ -0,0 +1,11 @@
+customer_id,timestamp,amount,moving_avg
+c1,1,10.0,
+c1,2,20.0,
+c1,3,30.0,20.0
+c1,4,40.0,30.0
+c1,5,50.0,40.0
+c2,1,100.0,
+c2,2,200.0,
+c2,3,,
+c2,4,300.0,200.0
+c2,5,400.0,300.0
diff --git a/integration-tests/transforms/main-0006-groupby.hwf
b/integration-tests/transforms/main-0006-groupby.hwf
index 8ba8f1a928..b933d71d93 100644
--- a/integration-tests/transforms/main-0006-groupby.hwf
+++ b/integration-tests/transforms/main-0006-groupby.hwf
@@ -22,37 +22,33 @@ limitations under the License.
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
- <workflow_version/>
<created_user>-</created_user>
- <created_date>2021/04/19 12:05:51.497</created_date>
<modified_user>-</modified_user>
+ <created_date>2021/04/19 12:05:51.497</created_date>
<modified_date>2021/04/19 12:05:51.497</modified_date>
- <parameters>
- </parameters>
+ <workflow_version/>
+ <parameters/>
<actions>
<action>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <intervalSeconds>0</intervalSeconds>
+ <intervalMinutes>60</intervalMinutes>
+ <DayOfMonth>1</DayOfMonth>
+ <weekDay>1</weekDay>
+ <minutes>0</minutes>
+ <hour>12</hour>
+ <doNotWaitOnFirstExecution>N</doNotWaitOnFirstExecution>
<name>Start</name>
<description/>
<type>SPECIAL</type>
<attributes/>
- <DayOfMonth>1</DayOfMonth>
- <hour>12</hour>
- <intervalMinutes>60</intervalMinutes>
- <intervalSeconds>0</intervalSeconds>
- <minutes>0</minutes>
- <repeat>N</repeat>
- <schedulerType>0</schedulerType>
- <weekDay>1</weekDay>
- <parallel>N</parallel>
<xloc>80</xloc>
<yloc>80</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>Run Group By tests</name>
- <description/>
- <type>RunPipelineTests</type>
- <attributes/>
<test_names>
<test_name>
<name>0006-groupby-basics UNIT</name>
@@ -66,66 +62,73 @@ limitations under the License.
<test_name>
<name>0006-groupby-concat UNIT</name>
</test_name>
+ <test_name>
+ <name>0006-groupby-moving-average UNIT</name>
+ </test_name>
</test_names>
- <parallel>N</parallel>
- <xloc>608</xloc>
+ <name>Run Group By tests</name>
+ <description/>
+ <type>RunPipelineTests</type>
+ <attributes/>
+ <xloc>768</xloc>
<yloc>80</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>0006-groupby-alwaysgiveback-result.hpl</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
<filename>${PROJECT_HOME}/0006-groupby-alwaysgiveback-result.hpl</filename>
<params_from_previous>N</params_from_previous>
<exec_per_row>N</exec_per_row>
<clear_rows>N</clear_rows>
<clear_files>N</clear_files>
+ <create_parent_folder>N</create_parent_folder>
<set_logfile>N</set_logfile>
+ <set_append_logfile>N</set_append_logfile>
<logfile/>
<logext/>
<add_date>N</add_date>
<add_time>N</add_time>
<loglevel>Basic</loglevel>
- <set_append_logfile>N</set_append_logfile>
<wait_until_finished>Y</wait_until_finished>
- <create_parent_folder>N</create_parent_folder>
- <run_configuration>local</run_configuration>
<parameters>
<pass_all_parameters>Y</pass_all_parameters>
</parameters>
- <parallel>N</parallel>
- <xloc>240</xloc>
+ <run_configuration>local</run_configuration>
+ <name>0006-groupby-alwaysgiveback-result.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>272</xloc>
<yloc>80</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>0006-groupby-alwaysgiveback-result.hpl 2</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
<filename>${PROJECT_HOME}/0006-groupby-alwaysgiveback-result.hpl</filename>
<params_from_previous>N</params_from_previous>
<exec_per_row>N</exec_per_row>
<clear_rows>N</clear_rows>
<clear_files>N</clear_files>
+ <create_parent_folder>N</create_parent_folder>
<set_logfile>N</set_logfile>
+ <set_append_logfile>N</set_append_logfile>
<logfile/>
<logext/>
<add_date>N</add_date>
<add_time>N</add_time>
<loglevel>Basic</loglevel>
- <set_append_logfile>N</set_append_logfile>
<wait_until_finished>Y</wait_until_finished>
- <create_parent_folder>N</create_parent_folder>
- <run_configuration>local</run_configuration>
<parameters>
<pass_all_parameters>Y</pass_all_parameters>
</parameters>
- <parallel>N</parallel>
- <xloc>448</xloc>
+ <run_configuration>local</run_configuration>
+ <name>0006-groupby-alwaysgiveback-result.hpl 2</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>544</xloc>
<yloc>80</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
</actions>
@@ -133,26 +136,25 @@ limitations under the License.
<hop>
<from>Start</from>
<to>0006-groupby-alwaysgiveback-result.hpl</to>
- <enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
<hop>
<from>0006-groupby-alwaysgiveback-result.hpl</from>
<to>0006-groupby-alwaysgiveback-result.hpl 2</to>
- <enabled>Y</enabled>
<evaluation>N</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
<hop>
<from>0006-groupby-alwaysgiveback-result.hpl 2</from>
<to>Run Group By tests</to>
- <enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
</hops>
- <notepads>
- </notepads>
+ <notepads/>
<attributes/>
</workflow>
diff --git
a/integration-tests/transforms/metadata/dataset/golden-groupby-moving-average.json
b/integration-tests/transforms/metadata/dataset/golden-groupby-moving-average.json
new file mode 100644
index 0000000000..7fa62b3282
--- /dev/null
+++
b/integration-tests/transforms/metadata/dataset/golden-groupby-moving-average.json
@@ -0,0 +1,40 @@
+{
+ "base_filename": "golden-groupby-moving-average.csv",
+ "name": "golden-groupby-moving-average",
+ "description": "",
+ "dataset_fields": [
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "customer_id"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 5,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "timestamp"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "amount"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "moving_avg"
+ }
+ ],
+ "folder_name": ""
+}
diff --git
a/integration-tests/transforms/metadata/unit-test/0006-groupby-moving-average
UNIT.json
b/integration-tests/transforms/metadata/unit-test/0006-groupby-moving-average
UNIT.json
new file mode 100644
index 0000000000..2aff6f64a0
--- /dev/null
+++
b/integration-tests/transforms/metadata/unit-test/0006-groupby-moving-average
UNIT.json
@@ -0,0 +1,41 @@
+{
+ "variableValues": [],
+ "database_replacements": [],
+ "autoOpening": true,
+ "basePath": "",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "customer_id",
+ "data_set_field": "customer_id"
+ },
+ {
+ "transform_field": "timestamp",
+ "data_set_field": "timestamp"
+ },
+ {
+ "transform_field": "amount",
+ "data_set_field": "amount"
+ },
+ {
+ "transform_field": "moving_avg",
+ "data_set_field": "moving_avg"
+ }
+ ],
+ "field_order": [
+ "customer_id",
+ "timestamp"
+ ],
+ "transform_name": "Validate",
+ "data_set_name": "golden-groupby-moving-average"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0006-groupby-moving-average UNIT",
+ "description": "",
+ "trans_test_tweaks": [],
+ "persist_filename": "",
+ "pipeline_filename": "./0006-groupby-moving-average.hpl",
+ "test_type": "UNIT_TEST"
+}
diff --git
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/Aggregation.java
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/Aggregation.java
index 7d318fffe7..ac8e7f2c0b 100644
---
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/Aggregation.java
+++
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/Aggregation.java
@@ -72,6 +72,8 @@ public class Aggregation implements Cloneable {
public static final int TYPE_GROUP_CONCAT_DISTINCT = 22;
+ public static final int TYPE_GROUP_MOVING_AVERAGE = 23;
+
public static final String[]
typeGroupLabel = /* WARNING: DO NOT TRANSLATE THIS. WE ARE SERIOUS,
DON'T TRANSLATE! */ {
"-",
@@ -97,6 +99,7 @@ public class Aggregation implements Cloneable {
"PERCENTILE_NEAREST_RANK",
"CONCAT_STRING_CRLF",
"CONCAT_DISTINCT",
+ "MOVING_AVG",
};
public static final String[] typeGroupLongDesc = {
@@ -122,7 +125,8 @@ public class Aggregation implements Cloneable {
BaseMessages.getString(PKG,
"GroupByMeta.TypeGroupLongDesc.STANDARD_DEVIATION_SAMPLE"),
BaseMessages.getString(PKG,
"GroupByMeta.TypeGroupLongDesc.PERCENTILE_NEAREST_RANK"),
BaseMessages.getString(PKG,
"GroupByMeta.TypeGroupLongDesc.CONCAT_STRING_CRLF"),
- BaseMessages.getString(PKG,
"GroupByMeta.TypeGroupLongDesc.CONCAT_DISTINCT")
+ BaseMessages.getString(PKG,
"GroupByMeta.TypeGroupLongDesc.CONCAT_DISTINCT"),
+ BaseMessages.getString(PKG, "GroupByMeta.TypeGroupLongDesc.MOVING_AVERAGE")
};
@HopMetadataProperty(
@@ -150,6 +154,12 @@ public class Aggregation implements Cloneable {
injectionKeyDescription = "GroupByMeta.Injection.AGG_VALUE")
private String value;
+ @HopMetadataProperty(
+ key = "orderfield",
+ injectionKey = "AGG_ORDER_FIELD",
+ injectionKeyDescription = "GroupByMeta.Injection.AGG_ORDER_FIELD")
+ private String orderField;
+
public Aggregation() {}
public Aggregation(String field, String subject, String typeDesc, String
value) {
@@ -160,10 +170,20 @@ public class Aggregation implements Cloneable {
this.value = value;
}
+ public Aggregation(
+ String field, String subject, String typeDesc, String value, String
orderField) {
+ this(field, subject, typeDesc, value);
+ this.orderField = orderField;
+ }
+
@Override
public Aggregation clone() {
return new Aggregation(
- field, subject,
getTypeDescLongFromCode(getTypeCodeFromLabel(typeLabel)), value);
+ field,
+ subject,
+ getTypeDescLongFromCode(getTypeCodeFromLabel(typeLabel)),
+ value,
+ orderField);
}
@Override
@@ -175,12 +195,13 @@ public class Aggregation implements Cloneable {
&& Objects.equals(field, that.field)
&& Objects.equals(subject, that.subject)
&& Objects.equals(typeLabel, that.typeLabel)
- && Objects.equals(value, that.value);
+ && Objects.equals(value, that.value)
+ && Objects.equals(orderField, that.orderField);
}
@Override
public int hashCode() {
- return Objects.hash(field, subject, typeLabel, type, value);
+ return Objects.hash(field, subject, typeLabel, type, value, orderField);
}
/**
@@ -245,6 +266,22 @@ public class Aggregation implements Cloneable {
this.value = value;
}
+ /**
+ * Gets orderField
+ *
+ * @return value of orderField
+ */
+ public String getOrderField() {
+ return orderField;
+ }
+
+ /**
+ * @param orderField The orderField to set
+ */
+ public void setOrderField(String orderField) {
+ this.orderField = orderField;
+ }
+
public int getType() {
return type;
}
diff --git
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupBy.java
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupBy.java
index bca8b14a6d..02fa2b2046 100644
---
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupBy.java
+++
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupBy.java
@@ -122,6 +122,11 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
data.cumulativeAvgSourceIndexes = new ArrayList<>();
data.cumulativeAvgTargetIndexes = new ArrayList<>();
+ data.movingAvgSourceIndexes = new ArrayList<>();
+ data.movingAvgTargetIndexes = new ArrayList<>();
+ data.movingAvgWidths = new ArrayList<>();
+ data.movingAvgIndexes = new ArrayList<>();
+
for (int i = 0; i < meta.getAggregations().size(); i++) {
Aggregation aggregation = meta.getAggregations().get(i);
if (aggregation.getType() == Aggregation.TYPE_GROUP_COUNT_ANY) {
@@ -152,6 +157,20 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
//
data.cumulativeAvgTargetIndexes.add(data.inputRowMeta.size() + i);
}
+ if (aggregation.getType() == Aggregation.TYPE_GROUP_MOVING_AVERAGE) {
+ data.movingAvgSourceIndexes.add(data.subjectnrs[i]);
+ data.movingAvgTargetIndexes.add(data.inputRowMeta.size() + i);
+ data.movingAvgIndexes.add(i);
+ int windowSize = 1;
+ if (!Utils.isEmpty(aggregation.getValue())) {
+ try {
+ windowSize = Math.max(1,
Integer.parseInt(resolve(aggregation.getValue())));
+ } catch (NumberFormatException e) {
+ // default to 1
+ }
+ }
+ data.movingAvgWidths.add(windowSize);
+ }
}
data.previousSums = new Object[data.cumulativeSumTargetIndexes.size()];
@@ -159,6 +178,18 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
data.previousAvgSum = new Object[data.cumulativeAvgTargetIndexes.size()];
data.previousAvgCount = new long[data.cumulativeAvgTargetIndexes.size()];
+ // Initialise per-aggregation sliding windows for MOVING_AVERAGE
+ //
+ @SuppressWarnings("unchecked")
+ java.util.ArrayDeque<Double>[] windows =
+ new java.util.ArrayDeque[meta.getAggregations().size()];
+ data.movingAvgWindows = windows;
+ for (int i = 0; i < meta.getAggregations().size(); i++) {
+ if (meta.getAggregations().get(i).getType() ==
Aggregation.TYPE_GROUP_MOVING_AVERAGE) {
+ data.movingAvgWindows[i] = new java.util.ArrayDeque<>();
+ }
+ }
+
data.groupnrs = new int[meta.getGroupingFields().size()];
for (int i = 0; i < meta.getGroupingFields().size(); i++) {
data.groupnrs[i] =
@@ -242,6 +273,7 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
addCumulativeSums(row);
addCumulativeAverages(row);
+ addMovingAverages(row);
putRow(data.outputRowMeta, row);
row = getRowFromBuffer();
@@ -291,6 +323,7 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
addCumulativeSums(row);
addCumulativeAverages(row);
+ addMovingAverages(row);
putRow(data.outputRowMeta, row);
row = getRowFromBuffer();
@@ -402,6 +435,37 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
}
}
+ void addMovingAverages(Object[] row) throws HopValueException {
+ for (int i = 0; i < data.movingAvgSourceIndexes.size(); i++) {
+ int sourceIndex = data.movingAvgSourceIndexes.get(i);
+ int targetIndex = data.movingAvgTargetIndexes.get(i);
+ int windowSize = data.movingAvgWidths.get(i);
+ int aggIndex = data.movingAvgIndexes.get(i);
+
+ Object sourceValue = row[sourceIndex];
+ IValueMeta sourceMeta = data.inputRowMeta.getValueMeta(sourceIndex);
+
+ if (!sourceMeta.isNull(sourceValue)) {
+ java.util.ArrayDeque<Double> window = data.movingAvgWindows[aggIndex];
+ window.addLast(sourceMeta.getNumber(sourceValue));
+ while (window.size() > windowSize) {
+ window.pollFirst();
+ }
+ if (window.size() == windowSize) {
+ double sum = 0.0;
+ for (double val : window) {
+ sum += val;
+ }
+ row[targetIndex] = sum / windowSize;
+ } else {
+ row[targetIndex] = null;
+ }
+ } else {
+ row[targetIndex] = null;
+ }
+ }
+ }
+
// Is the row r of the same group as previous?
boolean sameGroup(Object[] previous, Object[] r) throws HopValueException {
return data.inputRowMeta.compare(previous, r, data.groupnrs) == 0;
@@ -574,6 +638,9 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
SortedSet<Object> set = (SortedSet<Object>) value;
set.add(subj);
}
+ break;
+ case Aggregation.TYPE_GROUP_MOVING_AVERAGE:
+ break;
default:
break;
}
@@ -655,6 +722,15 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
vMeta = new ValueMetaString(fieldName);
v = new TreeSet<>();
break;
+ case Aggregation.TYPE_GROUP_MOVING_AVERAGE:
+ vMeta = new ValueMetaNumber(fieldName);
+ // agg[i] is null until the first full window is seen; set by
calcAggregate per row
+ v = null;
+ // Reset (clear) the sliding window for this aggregation within the
new group
+ if (data.movingAvgWindows != null && data.movingAvgWindows[i] !=
null) {
+ data.movingAvgWindows[i].clear();
+ }
+ break;
default:
// TODO raise an error here because we cannot continue successfully
maybe the UI should
// validate this
@@ -790,6 +866,9 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
Aggregation.TYPE_GROUP_CONCAT_STRING_CRLF:
ag = ((StringBuilder) ag).toString();
break;
+ case Aggregation.TYPE_GROUP_MOVING_AVERAGE:
+ // The result is already computed per-row in calcAggregate; pass
through as-is.
+ break;
case Aggregation.TYPE_GROUP_CONCAT_DISTINCT:
IValueMeta subjMeta =
data.inputRowMeta.getValueMeta(data.subjectnrs[i]);
String separator = "";
diff --git
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByData.java
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByData.java
index 89e867bda8..f800d681bc 100644
---
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByData.java
+++
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByData.java
@@ -22,6 +22,7 @@ import java.io.DataOutputStream;
import java.io.File;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
@@ -80,6 +81,11 @@ public class GroupByData extends BaseTransformData
implements ITransformData {
public List<Integer> cumulativeAvgSourceIndexes;
public List<Integer> cumulativeAvgTargetIndexes;
+ public List<Integer> movingAvgSourceIndexes;
+ public List<Integer> movingAvgTargetIndexes;
+ public List<Integer> movingAvgWidths;
+ public List<Integer> movingAvgIndexes;
+
public Object[] previousSums;
public Object[] previousAvgSum;
@@ -93,6 +99,14 @@ public class GroupByData extends BaseTransformData
implements ITransformData {
public boolean newBatch;
+ /**
+ * Sliding window buffers for MOVING_AVERAGE aggregations. One ArrayDeque
per aggregation index;
+ * null for aggregations that are not MOVING_AVERAGE. Each deque holds the
last N non-null numeric
+ * values of the subject field.
+ */
+ @SuppressWarnings("java:S1104")
+ public ArrayDeque<Double>[] movingAvgWindows;
+
public GroupByData() {
super();
diff --git
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByDialog.java
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByDialog.java
index 31fe5629a0..21c9819dd2 100644
---
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByDialog.java
+++
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByDialog.java
@@ -55,6 +55,7 @@ public class GroupByDialog extends BaseTransformDialog {
public static final String STRING_SORT_WARNING_PARAMETER =
"GroupSortWarning";
private static final int AGGREGATION_TABLE_TYPE_INDEX = 3;
+ private static final int AGGREGATION_TABLE_ORDER_FIELD_INDEX = 5;
private TableView wGroup;
@@ -309,7 +310,7 @@ public class GroupByDialog extends BaseTransformDialog {
fdlAgg.top = new FormAttachment(wGroup, margin);
wlAgg.setLayoutData(fdlAgg);
- int nrCols = 4;
+ int nrCols = 5;
int nrRows = input.getAggregations().size();
ciReturn = new ColumnInfo[nrCols];
@@ -336,6 +337,14 @@ public class GroupByDialog extends BaseTransformDialog {
false);
ciReturn[3].setToolTip(BaseMessages.getString(PKG,
"GroupByDialog.ColumnInfo.Value.Tooltip"));
ciReturn[3].setUsingVariables(true);
+ ciReturn[4] =
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "GroupByDialog.ColumnInfo.OrderField"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ new String[] {""},
+ false);
+ ciReturn[4].setToolTip(
+ BaseMessages.getString(PKG,
"GroupByDialog.ColumnInfo.OrderField.Tooltip"));
wAgg =
new TableView(
@@ -408,6 +417,7 @@ public class GroupByDialog extends BaseTransformDialog {
String[] fieldNames = ConstUi.sortFieldNames(inputFields);
ciKey[0].setComboValues(fieldNames);
ciReturn[1].setComboValues(fieldNames);
+ ciReturn[4].setComboValues(fieldNames);
}
public void setFlags() {
@@ -457,6 +467,7 @@ public class GroupByDialog extends BaseTransformDialog {
item.setText(2, Const.NVL(aggregation.getSubject(), ""));
item.setText(3,
Const.NVL(Aggregation.getTypeDescLongFromCode(aggregation.getType()), ""));
item.setText(4, Const.NVL(aggregation.getValue(), ""));
+ item.setText(5, Const.NVL(aggregation.getOrderField(), ""));
}
wGroup.setRowNums();
@@ -502,7 +513,9 @@ public class GroupByDialog extends BaseTransformDialog {
String aggSubject = item.getText(2);
String aggTypeDesc = item.getText(3);
String aggValue = item.getText(4);
- Aggregation aggr = new Aggregation(aggField, aggSubject, aggTypeDesc,
aggValue);
+ String aggOrderField = item.getText(5);
+ Aggregation aggr =
+ new Aggregation(aggField, aggSubject, aggTypeDesc, aggValue,
aggOrderField);
input.getAggregations().add(aggr);
}
@@ -582,7 +595,8 @@ public class GroupByDialog extends BaseTransformDialog {
.anyMatch(
pred ->
pred == Aggregation.TYPE_GROUP_CUMULATIVE_SUM
- || pred == Aggregation.TYPE_GROUP_CUMULATIVE_AVERAGE);
+ || pred == Aggregation.TYPE_GROUP_CUMULATIVE_AVERAGE
+ || pred == Aggregation.TYPE_GROUP_MOVING_AVERAGE);
allRowsButton.setEnabled(!isCumulativeSelected);
diff --git
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByMeta.java
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByMeta.java
index 832e0ee7da..64b2983f96 100644
---
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByMeta.java
+++
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupByMeta.java
@@ -274,6 +274,7 @@ public class GroupByMeta extends BaseTransformMeta<GroupBy,
GroupByData> {
case Aggregation.TYPE_GROUP_STANDARD_DEVIATION_SAMPLE:
case Aggregation.TYPE_GROUP_PERCENTILE:
case Aggregation.TYPE_GROUP_PERCENTILE_NEAREST_RANK:
+ case Aggregation.TYPE_GROUP_MOVING_AVERAGE:
valueType = IValueMeta.TYPE_NUMBER;
break;
case Aggregation.TYPE_GROUP_CONCAT_STRING:
diff --git
a/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_en_US.properties
b/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_en_US.properties
index 2d95207d11..64ba278c6c 100644
---
a/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_en_US.properties
+++
b/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_en_US.properties
@@ -32,10 +32,12 @@ GroupByDialog.AlwaysAddResult.ToolTip=To make sure we
always output a correct co
GroupByDialog.Browse.Button=&Browse...
GroupByDialog.ColumnInfo.GroupField=Group field
GroupByDialog.ColumnInfo.Name=Name
+GroupByDialog.ColumnInfo.OrderField=Order field
GroupByDialog.ColumnInfo.Subject=Subject
GroupByDialog.ColumnInfo.Type=Type
GroupByDialog.ColumnInfo.Value=Value
-GroupByDialog.ColumnInfo.Value.Tooltip=Fill here if more value is
requested.\nFor example for ''Concatenate strings separated by'' option.
+GroupByDialog.ColumnInfo.Value.Tooltip=Fill here if more value is
requested.\nFor example for ''Concatenate strings separated by'' option.\nFor
Moving Average, enter the window size N (e.g. 5).
+GroupByDialog.ColumnInfo.OrderField.Tooltip=Specify the field that defines the
order of rows within a group.\nEnsure the input data is sorted by this field
before this transform (e.g. using a Sort transform).\nThis is informational
only\: Moving Average does not sort internally.
GroupByDialog.FailedToGetFields.DialogMessage=Unable to get fields from
previous transforms because of an error
GroupByDialog.FailedToGetFields.DialogTitle=Get fields failed
GroupByDialog.FilePrefix.Label=TMP-file prefix
@@ -57,6 +59,7 @@ GroupByMeta.Exception.UnableToLoadTransformMetaFromXML=Unable
to load transform
GroupByMeta.Injection.ADD_GROUP_LINENR=Add group line number?
GroupByMeta.Injection.ADD_GROUP_LINENR_FIELD=Group line number field
GroupByMeta.Injection.AGG_FIELD=Aggregation field
+GroupByMeta.Injection.AGG_ORDER_FIELD=Aggregation order field
GroupByMeta.Injection.AGG_SUBJECT=Aggregation subject
GroupByMeta.Injection.AGG_TYPE=Aggregation type
GroupByMeta.Injection.AGG_VALUE=Aggregation value
@@ -85,6 +88,7 @@ GroupByMeta.TypeGroupLongDesc.LAST_INCL_NULL=Last value
GroupByMeta.TypeGroupLongDesc.MAX=Maximum
GroupByMeta.TypeGroupLongDesc.MEDIAN=Median
GroupByMeta.TypeGroupLongDesc.MIN=Minimum
+GroupByMeta.TypeGroupLongDesc.MOVING_AVERAGE=Moving average (last N rows)
GroupByMeta.TypeGroupLongDesc.PERCENTILE=Percentile (linear interpolation)
GroupByMeta.TypeGroupLongDesc.PERCENTILE_NEAREST_RANK=Percentile (nearest-rank
method)
GroupByMeta.TypeGroupLongDesc.STANDARD_DEVIATION=Standard deviation
(population)
diff --git
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/GroupByMetaTest.java
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/GroupByMetaTest.java
index 65678c9bc0..d1b21f70e0 100644
---
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/GroupByMetaTest.java
+++
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/GroupByMetaTest.java
@@ -100,7 +100,8 @@ class GroupByMetaTest {
new Aggregation("field21", "subject21", getDesc("STD_DEV_SAMPLE"),
"value21"),
new Aggregation("field22", "subject22",
getDesc("PERCENTILE_NEAREST_RANK"), "value22"),
new Aggregation("field23", "subject23",
getDesc("CONCAT_STRING_CRLF"), null),
- new Aggregation("field24", "subject23",
getDesc("CONCAT_DISTINCT"), "value24")));
+ new Aggregation("field24", "subject23",
getDesc("CONCAT_DISTINCT"), "value24"),
+ new Aggregation("field25", "subject25", getDesc("MOVING_AVG"),
"5", "order_field_25")));
meta.setPassAllRows(true);
meta.setAlwaysGivingBackOneRow(true);
meta.setDirectory("directory");
diff --git
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/MovingAverageAggregationTest.java
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/MovingAverageAggregationTest.java
new file mode 100644
index 0000000000..08551fa561
--- /dev/null
+++
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/MovingAverageAggregationTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.hop.pipeline.transforms.groupby;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaNumber;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for the Moving Average (Last N Events) aggregation type in the
Group By transform.
+ *
+ * <p>Uses the same package as GroupBy so that package-private methods
(addMovingAverages,
+ * newAggregate, getAggregateResult, setAllNullsAreZero) are accessible.
+ */
+class MovingAverageAggregationTest {
+
+ @BeforeAll
+ static void setUpClass() throws HopException {
+ HopEnvironment.init();
+ }
+
+ private GroupByMeta meta;
+ private GroupByData data;
+ private GroupBy groupBy;
+
+ /** Builds a minimal GroupBy with one MOVING_AVERAGE aggregation on a Number
column (N=3). */
+ @BeforeEach
+ void setUp() throws HopException {
+ meta = new GroupByMeta();
+
+ // Minimal TransformMeta stub required by BaseTransform constructor
+ TransformMeta transformMeta = new TransformMeta();
+ transformMeta.setName("test-groupby");
+
+ // Input row: [amount (Number)]
+ IRowMeta inputRowMeta = new RowMeta();
+ inputRowMeta.addValueMeta(new ValueMetaNumber("amount"));
+
+ // One aggregation: MOVING_AVERAGE of "amount" with N=3
+ Aggregation agg =
+ new Aggregation(
+ "moving_avg_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_MOVING_AVERAGE),
+ "3",
+ "");
+ meta.getAggregations().add(agg);
+
+ data = new GroupByData();
+ data.inputRowMeta = inputRowMeta;
+ data.subjectnrs = new int[] {0}; // "amount" is at index 0
+ data.counts = new long[1];
+ data.mean = new double[1];
+ // Cumulative-sum/avg tracking arrays (required by newAggregate's teardown
section)
+ data.previousSums = new Object[0];
+ data.previousAvgSum = new Object[0];
+ data.previousAvgCount = new long[0];
+
+ // Initialise moving average metadata lists used by addMovingAverages
+ data.movingAvgSourceIndexes = List.of(0);
+ data.movingAvgTargetIndexes = List.of(1);
+ data.movingAvgWidths = new ArrayList<>(List.of(3));
+ data.movingAvgIndexes = List.of(0);
+
+ // Initialise the sliding window array
+ @SuppressWarnings("unchecked")
+ ArrayDeque<Double>[] windows = new ArrayDeque[1];
+ data.movingAvgWindows = windows;
+ data.movingAvgWindows[0] = new ArrayDeque<>();
+
+ // Minimal GroupBy instance - we only exercise
addMovingAverages/newAggregate/getAggregateResult
+ groupBy =
+ new GroupBy(transformMeta, meta, data, 0, null, null) {
+ // Override resolve so variable expressions are returned as-is
+ @Override
+ public String resolve(String str) {
+ return str;
+ }
+ };
+ groupBy.setAllNullsAreZero(false);
+
+ // Call newAggregate to initialise data.agg[] and data.aggMeta
+ groupBy.newAggregate(new Object[] {null});
+ }
+
+ // Helper: feed a value through addMovingAverages and return the computed
moving_avg value
+ private Object feedValue(Double value) throws Exception {
+ Object[] row = new Object[] {value, null};
+ groupBy.addMovingAverages(row);
+ return row[1];
+ }
+
+ /**
+ * With N=3, the first two rows should produce null (window not full), and
the third row should
+ * produce the average of the three values.
+ */
+ @Test
+ void testPartialWindowReturnsNull() throws Exception {
+ assertNull(feedValue(10.0), "1st row: window not full -> null");
+ assertNull(feedValue(20.0), "2nd row: window not full -> null");
+ assertEquals(20.0, (Double) feedValue(30.0), 1e-9, "3rd row: full window
-> avg(10,20,30)=20");
+ }
+
+ /** After the window fills, it slides correctly: removes oldest, adds
newest, recomputes. */
+ @Test
+ void testSlidingWindowMovesCorrectly() throws Exception {
+ feedValue(10.0); // null
+ feedValue(20.0); // null
+ feedValue(30.0); // avg(10,20,30)=20
+ // Now add 40: window = [20,30,40], avg = 30
+ assertEquals(30.0, (Double) feedValue(40.0), 1e-9, "avg(20,30,40)=30");
+ // Add 50: window = [30,40,50], avg = 40
+ assertEquals(40.0, (Double) feedValue(50.0), 1e-9, "avg(30,40,50)=40");
+ }
+
+ /** A null subject value must be ignored: the window does not grow and
result stays null. */
+ @Test
+ void testNullSubjectIsSkipped() throws Exception {
+ feedValue(10.0); // window=[10], size=1 -> null
+ feedValue(null); // null value skipped; window=[10], size=1 -> still null
+ assertNull(feedValue(20.0), "Window size 2 < N=3 -> null");
+ assertEquals(20.0, (Double) feedValue(30.0), 1e-9, "After null skip:
avg(10,20,30)=20");
+ }
+
+ /**
+ * Resetting the window (simulating a new group) via newAggregate must clear
the deque and reset
+ * to null.
+ */
+ @Test
+ void testWindowResetsOnNewGroup() throws Exception {
+ feedValue(10.0);
+ feedValue(20.0);
+ feedValue(30.0); // full window, avg=20
+
+ // Simulate start of a new group
+ groupBy.newAggregate(new Object[] {null});
+
+ // Window should be clear; first value produces null again
+ assertNull(feedValue(5.0), "After group reset, 1st value -> null");
+ assertNull(feedValue(15.0), "After group reset, 2nd value -> null");
+ assertEquals(
+ 15.0, (Double) feedValue(25.0), 1e-9, "After group reset, 3rd value
avg(5,15,25)=15");
+ }
+
+ /** With N=1, every non-null row should immediately return that row's own
value. */
+ @Test
+ void testWindowSizeOne() throws Exception {
+ // Reconfigure with N=1
+ meta.getAggregations().get(0).setValue("1");
+ data.movingAvgWidths.set(0, 1);
+ data.movingAvgWindows[0].clear();
+ groupBy.newAggregate(new Object[] {null});
+
+ assertEquals(42.0, (Double) feedValue(42.0), 1e-9, "N=1: returns current
value 42");
+ assertEquals(7.0, (Double) feedValue(7.0), 1e-9, "N=1: returns current
value 7");
+ assertEquals(100.0, (Double) feedValue(100.0), 1e-9, "N=1: returns current
value 100");
+ }
+
+ /** Verify that getAggregateResult passes through the rolling value stored
in agg[0]. */
+ @Test
+ void testGetAggregateResultPassesThrough() throws Exception {
+ data.agg[0] = 20.0;
+ Object[] result = groupBy.getAggregateResult();
+ assertEquals(20.0, (Double) result[0], 1e-9, "getAggregateResult passes
through moving avg");
+ }
+}
