This is an automated email from the ASF dual-hosted git repository.
bamaer 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 4d34b3713f Fix#8308 Add a Group By option to exclude rows from
aggregation based on a boolean field (#8309)
4d34b3713f is described below
commit 4d34b3713ffb9962e66df0360c57b457cc01d489
Author: Lance <[email protected]>
AuthorDate: Fri Sep 11 19:21:42 2026 +0800
Fix#8308 Add a Group By option to exclude rows from aggregation based on a
boolean field (#8309)
* Fix Add a Group By option to exclude rows from aggregation based on a
boolean field
Signed-off-by: lance <[email protected]>
* Fix Add a Group By option to exclude rows from aggregation based on a
boolean field
Signed-off-by: lance <[email protected]>
---------
Signed-off-by: lance <[email protected]>
---
.../ROOT/pages/pipeline/transforms/groupby.adoc | 5 +
.../transforms/0006-groupby-ignore-aggregate.hpl | 283 +++++++++++++++
.../datasets/golden-groupby-ignore-aggregate.csv | 3 +
integration-tests/transforms/main-0006-groupby.hwf | 9 +-
.../dataset/golden-groupby-ignore-aggregate.json | 32 ++
.../0006-groupby-ignore-aggregate UNIT.json | 36 ++
.../pipeline/transforms/groupby/Aggregation.java | 115 +-----
.../hop/pipeline/transforms/groupby/GroupBy.java | 113 +++++-
.../pipeline/transforms/groupby/GroupByData.java | 9 +
.../pipeline/transforms/groupby/GroupByDialog.java | 83 ++++-
.../pipeline/transforms/groupby/GroupByMeta.java | 191 +++-------
.../pipeline/transforms/groupby/GroupingField.java | 32 +-
.../groupby/messages/messages_en_US.properties | 12 +
.../groupby/messages/messages_zh_CN.properties | 12 +
.../transforms/groupby/AggregationTest.java | 107 ++++++
.../transforms/{ => groupby}/GroupByMetaTest.java | 126 ++++++-
.../pipeline/transforms/groupby/GroupByTest.java | 344 ++++++++++++++++++
.../transforms/groupby/GroupingFieldTest.java | 51 +++
.../transforms/groupby/IgnoreAggregateTest.java | 385 +++++++++++++++++++++
19 files changed, 1643 insertions(+), 305 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 b7e2e83b59..5555e9c096 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
@@ -67,6 +67,11 @@ You must specify the name of the flag field in that case
(the type is boolean).
|Always give back a row|If you enable this option, the Group By transform will
always give back a result row, even if there is no input row.
This can be useful if you want to count the number of rows.
Without this option you would never get a count of zero (0).
+|Ignore rows for aggregation|Enable to exclude rows from aggregation when the
selected boolean field is `true`.
+Ignored rows still participate in group detection.
+When *Include all rows* is enabled, ignored rows are still written to the
output and receive the group aggregate values, but they do not contribute to
cumulative sum, cumulative average, or moving average.
+|Ignore field name|Name of the boolean field that marks rows to exclude from
aggregation.
+This field is required when *Ignore rows for aggregation* is enabled.
|Group fields table|Specify the fields over which you want to group.
Click Get Fields to add all fields from the input stream(s).
|Aggregates table a|Specify the fields that must be aggregated, the method and
the name of the resulting new field.
diff --git a/integration-tests/transforms/0006-groupby-ignore-aggregate.hpl
b/integration-tests/transforms/0006-groupby-ignore-aggregate.hpl
new file mode 100644
index 0000000000..e2ba91e63e
--- /dev/null
+++ b/integration-tests/transforms/0006-groupby-ignore-aggregate.hpl
@@ -0,0 +1,283 @@
+<?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>
+ <name>0006-groupby-ignore-aggregate</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Verify Group By ignore_aggregate / field_ignore excludes rows
from aggregation</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <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>
+ <created_user>-</created_user>
+ <created_date>2026/03/25 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/03/25 00:00:00.000</modified_date>
+ <key_for_session_key>H4sIAAAAAAAAAAMAAAAAAAAAAAA=</key_for_session_key>
+ <is_key_private>N</is_key_private>
+ </info>
+ <notepads>
+ <notepad>
+ <note>skip=true rows are excluded from SUM/COUNT.
+Expect sum_amount=60, cnt=3 (not 160/4).</note>
+ <xloc>48</xloc>
+ <yloc>16</yloc>
+ <width>360</width>
+ <heigth>58</heigth>
+ <fontname>Segoe UI</fontname>
+ <fontsize>9</fontsize>
+ <fontbold>N</fontbold>
+ <fontitalic>N</fontitalic>
+ <fontcolorred>14</fontcolorred>
+ <fontcolorgreen>58</fontcolorgreen>
+ <fontcolorblue>90</fontcolorblue>
+ <backgroundcolorred>201</backgroundcolorred>
+ <backgroundcolorgreen>232</backgroundcolorgreen>
+ <backgroundcolorblue>251</backgroundcolorblue>
+ <bordercolorred>14</bordercolorred>
+ <bordercolorgreen>58</bordercolorgreen>
+ <bordercolorblue>90</bordercolorblue>
+ </notepad>
+ </notepads>
+ <order>
+ <hop>
+ <from>input</from>
+ <to>Group by</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Group by</from>
+ <to>numbers to string</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>numbers to string</from>
+ <to>Validate</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>input</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>grp</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <name>amount</name>
+ <type>Number</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <name>skip</name>
+ <type>Boolean</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <data>
+ <line>
+ <item>A</item>
+ <item>10</item>
+ <item>false</item>
+ </line>
+ <line>
+ <item>A</item>
+ <item>20</item>
+ <item>false</item>
+ </line>
+ <line>
+ <item>A</item>
+ <item>30</item>
+ <item>false</item>
+ </line>
+ <line>
+ <item>A</item>
+ <item>100</item>
+ <item>true</item>
+ </line>
+ <line>
+ <item>B</item>
+ <item>5</item>
+ <item>false</item>
+ </line>
+ <line>
+ <item>B</item>
+ <item>50</item>
+ <item>true</item>
+ </line>
+ </data>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Group by</name>
+ <type>GroupBy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <all_rows>N</all_rows>
+ <ignore_aggregate>Y</ignore_aggregate>
+ <field_ignore>skip</field_ignore>
+ <directory>${java.io.tmpdir}</directory>
+ <prefix>grp</prefix>
+ <add_linenr>N</add_linenr>
+ <linenr_fieldname/>
+ <give_back_row>N</give_back_row>
+ <group>
+ <field>
+ <name>grp</name>
+ </field>
+ </group>
+ <fields>
+ <field>
+ <aggregate>sum_amount</aggregate>
+ <subject>amount</subject>
+ <type>SUM</type>
+ <valuefield/>
+ </field>
+ <field>
+ <aggregate>cnt</aggregate>
+ <subject>amount</subject>
+ <type>COUNT_ALL</type>
+ <valuefield/>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>numbers to string</name>
+ <type>SelectValues</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field_rename/>
+ <select_unspecified>N</select_unspecified>
+ <meta>
+ <name>sum_amount</name>
+ <rename>sum_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>cnt</name>
+ <rename>cnt</rename>
+ <type>String</type>
+ <length>-2</length>
+ <precision>-2</precision>
+ <storage_type/>
+ <conversion_mask>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>
+ <attributes/>
+ <GUI>
+ <xloc>512</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Validate</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>704</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git
a/integration-tests/transforms/datasets/golden-groupby-ignore-aggregate.csv
b/integration-tests/transforms/datasets/golden-groupby-ignore-aggregate.csv
new file mode 100644
index 0000000000..e2937b95c0
--- /dev/null
+++ b/integration-tests/transforms/datasets/golden-groupby-ignore-aggregate.csv
@@ -0,0 +1,3 @@
+grp,sum_amount,cnt
+A,60.0,3
+B,5.0,1
diff --git a/integration-tests/transforms/main-0006-groupby.hwf
b/integration-tests/transforms/main-0006-groupby.hwf
index b933d71d93..138c2606e1 100644
--- a/integration-tests/transforms/main-0006-groupby.hwf
+++ b/integration-tests/transforms/main-0006-groupby.hwf
@@ -26,6 +26,8 @@ limitations under the License.
<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>
+ <created_hop_version/>
+ <modified_hop_version>2.20.0-SNAPSHOT</modified_hop_version>
<workflow_version/>
<parameters/>
<actions>
@@ -65,12 +67,15 @@ limitations under the License.
<test_name>
<name>0006-groupby-moving-average UNIT</name>
</test_name>
+ <test_name>
+ <name>0006-groupby-ignore-aggregate UNIT</name>
+ </test_name>
</test_names>
<name>Run Group By tests</name>
<description/>
<type>RunPipelineTests</type>
<attributes/>
- <xloc>768</xloc>
+ <xloc>864</xloc>
<yloc>80</yloc>
<parallel>N</parallel>
<attributes_hac/>
@@ -126,7 +131,7 @@ limitations under the License.
<description/>
<type>PIPELINE</type>
<attributes/>
- <xloc>544</xloc>
+ <xloc>592</xloc>
<yloc>80</yloc>
<parallel>N</parallel>
<attributes_hac/>
diff --git
a/integration-tests/transforms/metadata/dataset/golden-groupby-ignore-aggregate.json
b/integration-tests/transforms/metadata/dataset/golden-groupby-ignore-aggregate.json
new file mode 100644
index 0000000000..eede402023
--- /dev/null
+++
b/integration-tests/transforms/metadata/dataset/golden-groupby-ignore-aggregate.json
@@ -0,0 +1,32 @@
+{
+ "base_filename": "golden-groupby-ignore-aggregate.csv",
+ "name": "golden-groupby-ignore-aggregate",
+ "description": "Expected aggregates when ignore_aggregate skips
boolean-flagged rows",
+ "dataset_fields": [
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "grp"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "sum_amount"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_format": "",
+ "field_name": "cnt"
+ }
+ ],
+ "folder_name": ""
+}
diff --git
a/integration-tests/transforms/metadata/unit-test/0006-groupby-ignore-aggregate
UNIT.json
b/integration-tests/transforms/metadata/unit-test/0006-groupby-ignore-aggregate
UNIT.json
new file mode 100644
index 0000000000..33d9f309d7
--- /dev/null
+++
b/integration-tests/transforms/metadata/unit-test/0006-groupby-ignore-aggregate
UNIT.json
@@ -0,0 +1,36 @@
+{
+ "variableValues": [],
+ "database_replacements": [],
+ "autoOpening": true,
+ "basePath": "",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "grp",
+ "data_set_field": "grp"
+ },
+ {
+ "transform_field": "sum_amount",
+ "data_set_field": "sum_amount"
+ },
+ {
+ "transform_field": "cnt",
+ "data_set_field": "cnt"
+ }
+ ],
+ "field_order": [
+ "grp"
+ ],
+ "transform_name": "Validate",
+ "data_set_name": "golden-groupby-ignore-aggregate"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0006-groupby-ignore-aggregate UNIT",
+ "description": "Group By ignores rows where the configured boolean field is
true",
+ "trans_test_tweaks": [],
+ "persist_filename": "",
+ "pipeline_filename": "./0006-groupby-ignore-aggregate.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 ac8e7f2c0b..3e2cfc557f 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
@@ -13,15 +13,21 @@
* 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 java.util.Objects;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.metadata.api.HopMetadataProperty;
+@Getter
+@Setter
+@NoArgsConstructor
+@EqualsAndHashCode
public class Aggregation implements Cloneable {
private static final Class<?> PKG = Aggregation.class;
@@ -160,8 +166,6 @@ public class Aggregation implements Cloneable {
injectionKeyDescription = "GroupByMeta.Injection.AGG_ORDER_FIELD")
private String orderField;
- public Aggregation() {}
-
public Aggregation(String field, String subject, String typeDesc, String
value) {
this.field = field;
this.subject = subject;
@@ -186,113 +190,12 @@ public class Aggregation implements Cloneable {
orderField);
}
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- Aggregation that = (Aggregation) o;
- return type == that.type
- && Objects.equals(field, that.field)
- && Objects.equals(subject, that.subject)
- && Objects.equals(typeLabel, that.typeLabel)
- && Objects.equals(value, that.value)
- && Objects.equals(orderField, that.orderField);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(field, subject, typeLabel, type, value, orderField);
- }
-
- /**
- * Gets field
- *
- * @return value of field
- */
- public String getField() {
- return field;
- }
-
- /**
- * @param field The field to set
- */
- public void setField(String field) {
- this.field = field;
- }
-
- /**
- * Gets subject
- *
- * @return value of subject
- */
- public String getSubject() {
- return subject;
- }
-
- /**
- * @param subject The subject to set
- */
- public void setSubject(String subject) {
- this.subject = subject;
- }
-
- /**
- * Gets type
- *
- * @return value of type
- */
- public String getTypeLabel() {
- return typeLabel;
- }
-
+ /** Keep typeLabel and the numeric type code in sync. */
public void setTypeLabel(String typeCode) {
this.typeLabel = typeCode;
this.type = getTypeCodeFromLabel(typeCode);
}
- /**
- * Gets value
- *
- * @return value of value
- */
- public String getValue() {
- return value;
- }
-
- /**
- * @param value The value to set
- */
- public void setValue(String value) {
- 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;
- }
-
- /**
- * @param type The type to set
- */
- public void setType(int type) {
- this.type = type;
- }
-
public static final int getTypeCodeFromLongDesc(String desc) {
for (int i = 0; i < typeGroupLongDesc.length; i++) {
if (typeGroupLongDesc[i].equalsIgnoreCase(desc)) {
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 02fa2b2046..59ea72612c 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
@@ -206,6 +206,35 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
}
}
+ data.aggregateIgnoredFieldIndex = -1;
+ if (meta.isAggregateIgnored()) {
+ if (Utils.isEmpty(meta.getAggregateIgnoredField())) {
+ logError(BaseMessages.getString(PKG,
"GroupBy.Log.AggregateIgnoredFieldMissing"));
+ setErrors(1);
+ stopAll();
+ return false;
+ }
+ String ignoreField = resolve(meta.getAggregateIgnoredField());
+ data.aggregateIgnoredFieldIndex =
data.inputRowMeta.indexOfValue(ignoreField);
+ if (data.aggregateIgnoredFieldIndex < 0) {
+ logError(
+ BaseMessages.getString(
+ PKG, "GroupBy.Log.AggregateIgnoredFieldCouldNotFound",
ignoreField));
+ setErrors(1);
+ stopAll();
+ return false;
+ }
+ IValueMeta ignoreMeta =
data.inputRowMeta.getValueMeta(data.aggregateIgnoredFieldIndex);
+ if (!ignoreMeta.isBoolean()) {
+ logError(
+ BaseMessages.getString(
+ PKG, "GroupBy.Log.AggregateIgnoredFieldNotBoolean",
ignoreField));
+ setErrors(1);
+ stopAll();
+ return false;
+ }
+ }
+
// Create a metadata value for the counter Integers
//
data.valueMetaInteger = new ValueMetaInteger("count");
@@ -242,7 +271,9 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
data.previous = data.inputRowMeta.cloneRow(r); // copy the row to
previous
} else {
- calcAggregate(data.previous);
+ if (!isRowAggregateIgnored(data.previous)) {
+ calcAggregate(data.previous);
+ }
if (meta.isPassAllRows()) {
addToBuffer(data.previous);
@@ -302,7 +333,9 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
// ALL ROWS
if (data.previous != null) {
- calcAggregate(data.previous);
+ if (!isRowAggregateIgnored(data.previous)) {
+ calcAggregate(data.previous);
+ }
addToBuffer(data.previous);
}
data.groupResult = getAggregateResult();
@@ -334,7 +367,9 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
// Don't forget the last set of rows...
if (data.previous != null) {
- calcAggregate(data.previous);
+ if (!isRowAggregateIgnored(data.previous)) {
+ calcAggregate(data.previous);
+ }
}
Object[] result = buildResult(data.previous);
if (result != null) {
@@ -344,6 +379,12 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
}
private void addCumulativeSums(Object[] row) throws HopValueException {
+ if (isRowAggregateIgnored(row)) {
+ for (int i = 0; i < data.cumulativeSumSourceIndexes.size(); i++) {
+ row[data.cumulativeSumTargetIndexes.get(i)] = data.previousSums[i];
+ }
+ return;
+ }
// We need to adjust this row with cumulative averages?
//
@@ -377,6 +418,25 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
}
private void addCumulativeAverages(Object[] row) throws HopValueException {
+ if (isRowAggregateIgnored(row)) {
+ for (int i = 0; i < data.cumulativeAvgSourceIndexes.size(); i++) {
+ int targetIndex = data.cumulativeAvgTargetIndexes.get(i);
+ IValueMeta targetMeta = data.outputRowMeta.getValueMeta(targetIndex);
+ Object sum = data.previousAvgSum[i];
+ if (sum == null || data.previousAvgCount[i] == 0L) {
+ row[targetIndex] = null;
+ } else if (data.inputRowMeta
+ .getValueMeta(data.cumulativeAvgSourceIndexes.get(i))
+ .isInteger()) {
+ row[targetIndex] = ((Long) sum).doubleValue() /
data.previousAvgCount[i];
+ } else {
+ row[targetIndex] =
+ ValueDataUtil.divide(
+ targetMeta, sum, data.valueMetaInteger,
data.previousAvgCount[i]);
+ }
+ }
+ return;
+ }
// We need to adjust this row with cumulative sums
//
@@ -442,6 +502,20 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
int windowSize = data.movingAvgWidths.get(i);
int aggIndex = data.movingAvgIndexes.get(i);
+ if (isRowAggregateIgnored(row)) {
+ java.util.ArrayDeque<Double> window = data.movingAvgWindows[aggIndex];
+ if (window != null && window.size() == windowSize) {
+ double sum = 0.0;
+ for (double val : window) {
+ sum += val;
+ }
+ row[targetIndex] = sum / windowSize;
+ } else {
+ row[targetIndex] = null;
+ }
+ continue;
+ }
+
Object sourceValue = row[sourceIndex];
IValueMeta sourceMeta = data.inputRowMeta.getValueMeta(sourceIndex);
@@ -471,6 +545,18 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
return data.inputRowMeta.compare(previous, r, data.groupnrs) == 0;
}
+ /**
+ * Returns true when the row should be excluded from aggregation because
{@code ignore_aggregate}
+ * is enabled and the configured boolean field is true.
+ */
+ boolean isRowAggregateIgnored(Object[] row) throws HopValueException {
+ if (data.aggregateIgnoredFieldIndex < 0 || row == null) {
+ return false;
+ }
+ Boolean ignore = data.inputRowMeta.getBoolean(row,
data.aggregateIgnoredFieldIndex);
+ return ignore != null && ignore;
+ }
+
/**
* used for junits in GroupByAggregationNullsTest
*
@@ -593,10 +679,13 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
}
break;
case Aggregation.TYPE_GROUP_FIRST_INCL_NULL:
- // This is on purpose. The calculation of the
- // first field is done when setting up a new group
- // This is just the field of the first row
- // if (linesWritten==0) value.setValue(subj)
+ // First non-ignored row of the group, including a null subject
value.
+ if (data.firstInclNullSet == null || !data.firstInclNullSet[i]) {
+ data.agg[i] = subj;
+ if (data.firstInclNullSet != null) {
+ data.firstInclNullSet[i] = true;
+ }
+ }
break;
case Aggregation.TYPE_GROUP_LAST_INCL_NULL:
data.agg[i] = subj;
@@ -661,6 +750,9 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
data.agg = new Object[data.subjectnrs.length];
data.mean = new double[data.subjectnrs.length]; // sets all doubles to 0.0
data.aggMeta = new RowMeta();
+ data.firstInclNullSet = new boolean[data.subjectnrs.length];
+
+ boolean seedFromRow = r != null && !isRowAggregateIgnored(r);
for (int i = 0; i < data.subjectnrs.length; i++) {
Aggregation aggregation = meta.getAggregations().get(i);
@@ -708,7 +800,12 @@ public class GroupBy extends BaseTransform<GroupByMeta,
GroupByData> {
Aggregation.TYPE_GROUP_MAX:
vMeta = subjMeta.clone();
vMeta.setName(fieldName);
- v = r == null ? null : r[data.subjectnrs[i]];
+ // Do not seed from an ignored row; leave null so calcAggregate
initializes
+ // from the first non-ignored row of the group.
+ v = seedFromRow ? r[data.subjectnrs[i]] : null;
+ if (aggType == Aggregation.TYPE_GROUP_FIRST_INCL_NULL) {
+ data.firstInclNullSet[i] = seedFromRow;
+ }
break;
case Aggregation.TYPE_GROUP_CONCAT_STRING_CRLF,
Aggregation.TYPE_GROUP_CONCAT_COMMA:
vMeta = new ValueMetaString(fieldName);
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 f800d681bc..fc1461b81a 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
@@ -43,6 +43,15 @@ public class GroupByData extends BaseTransformData
implements ITransformData {
public IRowMeta groupAggMeta; // for speed: groupMeta+aggMeta
public int[] groupnrs;
+ /** Index of the boolean field that marks rows to exclude from aggregation,
or -1 when unused. */
+ public int aggregateIgnoredFieldIndex = -1;
+
+ /**
+ * Per-aggregation flag for {@code FIRST_INCL_NULL}: true once the first
non-ignored row of the
+ * current group has been applied.
+ */
+ public boolean[] firstInclNullSet;
+
/**
* array, length is equal to aggMeta value meta list size and metadata
subject fields length.
* Values corresponds to input values used to calculate target results.
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 79755122ec..a66991c48f 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
@@ -34,6 +34,7 @@ import org.apache.hop.ui.core.dialog.BaseDialog;
import org.apache.hop.ui.core.dialog.ErrorDialog;
import org.apache.hop.ui.core.dialog.MessageDialogWithToggle;
import org.apache.hop.ui.core.widget.ColumnInfo;
+import org.apache.hop.ui.core.widget.ComboVar;
import org.apache.hop.ui.core.widget.NamingSchemeTypes;
import org.apache.hop.ui.core.widget.TableView;
import org.apache.hop.ui.core.widget.TextVar;
@@ -85,6 +86,14 @@ public class GroupByDialog extends BaseTransformDialog {
private Button wAlwaysAddResult;
+ private Label wlIgnoreAggregate;
+
+ private Button wIgnoreAggregate;
+
+ private Label wlIgnoreAggregateField;
+
+ private ComboVar wIgnoreAggregateField;
+
private GroupByMeta input;
private boolean backupAllRows;
@@ -261,12 +270,64 @@ public class GroupByDialog extends BaseTransformDialog {
}
});
+ // Ignore rows for aggregation when a boolean field is true
+ //
+ wlIgnoreAggregate = new Label(shell, SWT.RIGHT);
+ wlIgnoreAggregate.setText(BaseMessages.getString(PKG,
"GroupByDialog.IgnoreAggregate.Label"));
+ wlIgnoreAggregate.setToolTipText(
+ BaseMessages.getString(PKG, "GroupByDialog.IgnoreAggregate.ToolTip"));
+ PropsUi.setLook(wlIgnoreAggregate);
+ FormData fdlIgnoreAggregate = new FormData();
+ fdlIgnoreAggregate.left = new FormAttachment(0, 0);
+ fdlIgnoreAggregate.top = new FormAttachment(wAlwaysAddResult, margin);
+ fdlIgnoreAggregate.right = new FormAttachment(middle, -margin);
+ wlIgnoreAggregate.setLayoutData(fdlIgnoreAggregate);
+ wIgnoreAggregate = new Button(shell, SWT.CHECK);
+ wIgnoreAggregate.setToolTipText(
+ BaseMessages.getString(PKG, "GroupByDialog.IgnoreAggregate.ToolTip"));
+ PropsUi.setLook(wIgnoreAggregate);
+ FormData fdIgnoreAggregate = new FormData();
+ fdIgnoreAggregate.left = new FormAttachment(middle, 0);
+ fdIgnoreAggregate.top = new FormAttachment(wlIgnoreAggregate, 0,
SWT.CENTER);
+ fdIgnoreAggregate.right = new FormAttachment(100, 0);
+ wIgnoreAggregate.setLayoutData(fdIgnoreAggregate);
+ wIgnoreAggregate.addSelectionListener(
+ new SelectionAdapter() {
+ @Override
+ public void widgetSelected(SelectionEvent e) {
+ input.setChanged();
+ setFlags();
+ }
+ });
+
+ wlIgnoreAggregateField = new Label(shell, SWT.RIGHT);
+ wlIgnoreAggregateField.setText(
+ BaseMessages.getString(PKG,
"GroupByDialog.IgnoreAggregateField.Label"));
+ wlIgnoreAggregateField.setToolTipText(
+ BaseMessages.getString(PKG,
"GroupByDialog.IgnoreAggregateField.ToolTip"));
+ PropsUi.setLook(wlIgnoreAggregateField);
+ FormData fdlIgnoreAggregateField = new FormData();
+ fdlIgnoreAggregateField.left = new FormAttachment(0, 0);
+ fdlIgnoreAggregateField.top = new FormAttachment(wIgnoreAggregate, margin);
+ fdlIgnoreAggregateField.right = new FormAttachment(middle, -margin);
+ wlIgnoreAggregateField.setLayoutData(fdlIgnoreAggregateField);
+ wIgnoreAggregateField = new ComboVar(variables, shell, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
+ wIgnoreAggregateField.setToolTipText(
+ BaseMessages.getString(PKG,
"GroupByDialog.IgnoreAggregateField.ToolTip"));
+ PropsUi.setLook(wIgnoreAggregateField);
+ wIgnoreAggregateField.addModifyListener(lsMod);
+ FormData fdIgnoreAggregateField = new FormData();
+ fdIgnoreAggregateField.left = new FormAttachment(middle, 0);
+ fdIgnoreAggregateField.top = new FormAttachment(wIgnoreAggregate, margin);
+ fdIgnoreAggregateField.right = new FormAttachment(100, 0);
+ wIgnoreAggregateField.setLayoutData(fdIgnoreAggregateField);
+
Label wlGroup = new Label(shell, SWT.NONE);
wlGroup.setText(BaseMessages.getString(PKG, "GroupByDialog.Group.Label"));
PropsUi.setLook(wlGroup);
FormData fdlGroup = new FormData();
fdlGroup.left = new FormAttachment(0, 0);
- fdlGroup.top = new FormAttachment(wAlwaysAddResult, margin);
+ fdlGroup.top = new FormAttachment(wIgnoreAggregateField, margin);
wlGroup.setLayoutData(fdlGroup);
int nrKeyCols = 1;
@@ -387,7 +448,15 @@ public class GroupByDialog extends BaseTransformDialog {
for (int i = 0; i < row.size(); i++) {
inputFields.add(row.getValueMeta(i).getName());
}
- setComboBoxes();
+ // ComboVar.setItems must run on the UI thread
+ shell
+ .getDisplay()
+ .asyncExec(
+ () -> {
+ if (!shell.isDisposed()) {
+ setComboBoxes();
+ }
+ });
} catch (HopException e) {
logError(BaseMessages.getString(PKG,
"System.Dialog.GetFieldsFailed.Message"));
}
@@ -421,6 +490,7 @@ public class GroupByDialog extends BaseTransformDialog {
ciKey[0].setComboValues(fieldNames);
ciReturn[1].setComboValues(fieldNames);
ciReturn[4].setComboValues(fieldNames);
+ wIgnoreAggregateField.setItems(fieldNames);
}
public void setFlags() {
@@ -434,6 +504,9 @@ public class GroupByDialog extends BaseTransformDialog {
wlLineNrField.setEnabled(wAllRows.getSelection() &&
wAddLineNr.getSelection());
wLineNrField.setEnabled(wAllRows.getSelection() &&
wAddLineNr.getSelection());
+
+ wlIgnoreAggregateField.setEnabled(wIgnoreAggregate.getSelection());
+ wIgnoreAggregateField.setEnabled(wIgnoreAggregate.getSelection());
}
/** Copy information from the meta-data input to the dialog fields. */
@@ -453,6 +526,10 @@ public class GroupByDialog extends BaseTransformDialog {
wLineNrField.setText(input.getLineNrInGroupField());
}
wAlwaysAddResult.setSelection(input.isAlwaysGivingBackOneRow());
+ wIgnoreAggregate.setSelection(input.isAggregateIgnored());
+ if (input.getAggregateIgnoredField() != null) {
+ wIgnoreAggregateField.setText(input.getAggregateIgnoredField());
+ }
if (input.getGroupingFields() != null) {
for (int i = 0; i < input.getGroupingFields().size(); i++) {
@@ -502,6 +579,8 @@ public class GroupByDialog extends BaseTransformDialog {
input.setLineNrInGroupField(wLineNrField.getText());
input.setAlwaysGivingBackOneRow(wAlwaysAddResult.getSelection());
input.setPassAllRows(wAllRows.getSelection());
+ input.setAggregateIgnored(wIgnoreAggregate.getSelection());
+ input.setAggregateIgnoredField(wIgnoreAggregateField.getText());
input.getGroupingFields().clear();
for (int i = 0; i < sizegroup; i++) {
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 c337443325..81b2b8d8f2 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
@@ -19,6 +19,8 @@ package org.apache.hop.pipeline.transforms.groupby;
import java.util.ArrayList;
import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
import org.apache.hop.core.CheckResult;
import org.apache.hop.core.ICheckResult;
import org.apache.hop.core.annotations.Transform;
@@ -38,6 +40,8 @@ import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.BaseTransformMeta;
import org.apache.hop.pipeline.transform.TransformMeta;
+@Getter
+@Setter
@Transform(
id = "GroupBy",
image = "groupby.svg",
@@ -72,15 +76,18 @@ public class GroupByMeta extends BaseTransformMeta<GroupBy,
GroupByData> {
injectionKeyDescription = "GroupByMeta.Injection.TEMP_FILE_PREFIX")
private String prefix;
- /** Indicate that some rows don't need to be considered : TODO: make work in
GUI & worker */
- @HopMetadataProperty(key = "ignore_aggregate")
+ /** When true, rows where the ignore field is true are excluded from
aggregation. */
+ @HopMetadataProperty(
+ key = "ignore_aggregate",
+ injectionKey = "IGNORE_AGGREGATE",
+ injectionKeyDescription = "GroupByMeta.Injection.IGNORE_AGGREGATE")
private boolean aggregateIgnored;
- /**
- * name of the boolean field that indicates we need to ignore the row :
TODO: make work in GUI &
- * worker
- */
- @HopMetadataProperty(key = "field_ignore")
+ /** Boolean field that marks rows to exclude from aggregation when
ignore_aggregate is enabled. */
+ @HopMetadataProperty(
+ key = "field_ignore",
+ injectionKey = "IGNORE_AGGREGATE_FIELD",
+ injectionKeyDescription = "GroupByMeta.Injection.IGNORE_AGGREGATE_FIELD")
private String aggregateIgnoredField;
/** Fields to group over */
@@ -125,62 +132,6 @@ public class GroupByMeta extends
BaseTransformMeta<GroupBy, GroupByData> {
aggregations = new ArrayList<>();
}
- /**
- * @return Returns the aggregateIgnored.
- */
- public boolean isAggregateIgnored() {
- return aggregateIgnored;
- }
-
- /**
- * @param aggregateIgnored The aggregateIgnored to set.
- */
- public void setAggregateIgnored(boolean aggregateIgnored) {
- this.aggregateIgnored = aggregateIgnored;
- }
-
- /**
- * @return Returns the aggregateIgnoredField.
- */
- public String getAggregateIgnoredField() {
- return aggregateIgnoredField;
- }
-
- /**
- * @param aggregateIgnoredField The aggregateIgnoredField to set.
- */
- public void setAggregateIgnoredField(String aggregateIgnoredField) {
- this.aggregateIgnoredField = aggregateIgnoredField;
- }
-
- /**
- * @return Returns the groupField.
- */
- public List<GroupingField> getGroupingFields() {
- return groupingFields;
- }
-
- /**
- * @param groupingFields The groupField to set.
- */
- public void setGroupingFields(List<GroupingField> groupingFields) {
- this.groupingFields = groupingFields;
- }
-
- /**
- * @return Returns the passAllRows.
- */
- public boolean isPassAllRows() {
- return passAllRows;
- }
-
- /**
- * @param passAllRows The passAllRows to set.
- */
- public void setPassAllRows(boolean passAllRows) {
- this.passAllRows = passAllRows;
- }
-
@Override
public void setDefault() {
directory = "${java.io.tmpdir}";
@@ -189,8 +140,6 @@ public class GroupByMeta extends BaseTransformMeta<GroupBy,
GroupByData> {
passAllRows = false;
aggregateIgnored = false;
aggregateIgnoredField = null;
-
- int sizeGroup = 0;
}
@Override
@@ -354,76 +303,34 @@ public class GroupByMeta extends
BaseTransformMeta<GroupBy, GroupByData> {
transformMeta);
remarks.add(cr);
}
- }
- /**
- * @return Returns the directory.
- */
- public String getDirectory() {
- return directory;
- }
-
- /**
- * @param directory The directory to set.
- */
- public void setDirectory(String directory) {
- this.directory = directory;
- }
-
- /**
- * @return Returns the prefix.
- */
- public String getPrefix() {
- return prefix;
- }
-
- /**
- * @param prefix The prefix to set.
- */
- public void setPrefix(String prefix) {
- this.prefix = prefix;
- }
-
- /**
- * @return the addingLineNrInGroup
- */
- public boolean isAddingLineNrInGroup() {
- return addingLineNrInGroup;
- }
-
- /**
- * @param addingLineNrInGroup the addingLineNrInGroup to set
- */
- public void setAddingLineNrInGroup(boolean addingLineNrInGroup) {
- this.addingLineNrInGroup = addingLineNrInGroup;
- }
-
- /**
- * @return the lineNrInGroupField
- */
- public String getLineNrInGroupField() {
- return lineNrInGroupField;
- }
-
- /**
- * @param lineNrInGroupField the lineNrInGroupField to set
- */
- public void setLineNrInGroupField(String lineNrInGroupField) {
- this.lineNrInGroupField = lineNrInGroupField;
- }
-
- /**
- * @return the alwaysGivingBackOneRow
- */
- public boolean isAlwaysGivingBackOneRow() {
- return alwaysGivingBackOneRow;
- }
-
- /**
- * @param alwaysGivingBackOneRow the alwaysGivingBackOneRow to set
- */
- public void setAlwaysGivingBackOneRow(boolean alwaysGivingBackOneRow) {
- this.alwaysGivingBackOneRow = alwaysGivingBackOneRow;
+ if (aggregateIgnored) {
+ if (Utils.isEmpty(aggregateIgnoredField)) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"GroupByMeta.CheckResult.IgnoreAggregateFieldMissing"),
+ transformMeta));
+ } else if (prev != null && !prev.isEmpty()) {
+ String ignoreField = variables.resolve(aggregateIgnoredField);
+ int idx = prev.indexOfValue(ignoreField);
+ if (idx < 0) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG,
"GroupByMeta.CheckResult.IgnoreAggregateFieldNotFound", ignoreField),
+ transformMeta));
+ } else if (!prev.getValueMeta(idx).isBoolean()) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG,
"GroupByMeta.CheckResult.IgnoreAggregateFieldNotBoolean", ignoreField),
+ transformMeta));
+ }
+ }
+ }
}
@Override
@@ -431,22 +338,6 @@ public class GroupByMeta extends
BaseTransformMeta<GroupBy, GroupByData> {
return new PipelineMeta.PipelineType[] {PipelineMeta.PipelineType.Normal};
}
- /**
- * Gets aggregations
- *
- * @return value of aggregations
- */
- public List<Aggregation> getAggregations() {
- return aggregations;
- }
-
- /**
- * @param aggregations The aggregations to set
- */
- public void setAggregations(List<Aggregation> aggregations) {
- this.aggregations = aggregations;
- }
-
@Override
public boolean supportsMultiCopyExecution() {
return false;
diff --git
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupingField.java
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupingField.java
index e94460737d..beaa78bef9 100644
---
a/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupingField.java
+++
b/plugins/transforms/groupby/src/main/java/org/apache/hop/pipeline/transforms/groupby/GroupingField.java
@@ -17,9 +17,16 @@
package org.apache.hop.pipeline.transforms.groupby;
-import java.util.Objects;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
import org.apache.hop.metadata.api.HopMetadataProperty;
+@Getter
+@Setter
+@NoArgsConstructor
+@EqualsAndHashCode
public class GroupingField implements Cloneable {
@HopMetadataProperty(
@@ -27,33 +34,10 @@ public class GroupingField implements Cloneable {
injectionKeyDescription = "GroupByMeta.Injection.GROUP_FIELD")
private String name;
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- public GroupingField() {}
-
public GroupingField(String name) {
this.name = name;
}
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- GroupingField that = (GroupingField) o;
- return Objects.equals(name, that.name);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(name);
- }
-
@Override
public GroupingField clone() {
return new GroupingField(name);
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 64ba278c6c..32c8a5032a 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
@@ -22,6 +22,9 @@ GroupBy.Exception.UnableToDeleteTemporaryFile=Unable to
delete temporary file: {
GroupBy.Exception.UnableToReadBackRowFromTemporaryFile=Unable to read back row
from temporary file\!
GroupBy.LineNumber=Linenr
GroupBy.Log.AggregateSubjectFieldCouldNotFound=Aggregate subject field [{0}]
couldn''t be found\!
+GroupBy.Log.AggregateIgnoredFieldCouldNotFound=Aggregate ignore field [{0}]
couldn''t be found\!
+GroupBy.Log.AggregateIgnoredFieldMissing=Ignore rows for aggregation is
enabled but no ignore field name was specified\!
+GroupBy.Log.AggregateIgnoredFieldNotBoolean=Aggregate ignore field [{0}] must
be a Boolean type\!
GroupBy.Log.GroupFieldCouldNotFound=Grouping field [{0}] couldn''t be found\!
GroupBy.Name=Group by
GroupByDialog.AddLineNr.Label=Add line number, restart in each group
@@ -29,6 +32,10 @@ GroupByDialog.Aggregates.Label=Aggregates\:
GroupByDialog.AllRows.Label=Include all rows
GroupByDialog.AlwaysAddResult.Label=Always give back a result row
GroupByDialog.AlwaysAddResult.ToolTip=To make sure we always output a correct
count aggregation we always output at least one row, even if there were no
input rows.\nThis makes the behavior consistent with the aggregation in an SQL
GROUP BY.
+GroupByDialog.IgnoreAggregate.Label=Ignore rows for aggregation
+GroupByDialog.IgnoreAggregate.ToolTip=When enabled, rows where the selected
boolean field is true are excluded from aggregation calculations.\nThey still
participate in group detection. With ''Include all rows'', ignored rows are
still written to the output with the group aggregates.
+GroupByDialog.IgnoreAggregateField.Label=Ignore field name
+GroupByDialog.IgnoreAggregateField.ToolTip=Boolean field that marks rows to
exclude from aggregation when the ignore option is enabled.
GroupByDialog.Browse.Button=&Browse...
GroupByDialog.ColumnInfo.GroupField=Group field
GroupByDialog.ColumnInfo.Name=Name
@@ -55,6 +62,9 @@ GroupByDialog.TempDir.Label=Temporary files directory
GroupByDialog.TransformName.Label=Transform name
GroupByMeta.CheckResult.NoInputError=No input received from other transforms\!
GroupByMeta.CheckResult.ReceivingInfoOK=Transform is receiving info from other
transforms.
+GroupByMeta.CheckResult.IgnoreAggregateFieldMissing=Ignore rows for
aggregation is enabled but no ignore field name was specified.
+GroupByMeta.CheckResult.IgnoreAggregateFieldNotFound=Aggregate ignore field
[{0}] couldn''t be found in the input stream.
+GroupByMeta.CheckResult.IgnoreAggregateFieldNotBoolean=Aggregate ignore field
[{0}] must be a Boolean type.
GroupByMeta.Exception.UnableToLoadTransformMetaFromXML=Unable to load
transform info from XML
GroupByMeta.Injection.ADD_GROUP_LINENR=Add group line number?
GroupByMeta.Injection.ADD_GROUP_LINENR_FIELD=Group line number field
@@ -67,6 +77,8 @@ GroupByMeta.Injection.AGGREGATIONS=Aggregations
GroupByMeta.Injection.ALWAYS_GIVE_ROW=Always give back row?
GroupByMeta.Injection.GROUP_FIELD=Group field
GroupByMeta.Injection.GROUPS=Groups
+GroupByMeta.Injection.IGNORE_AGGREGATE=Ignore rows for aggregation?
+GroupByMeta.Injection.IGNORE_AGGREGATE_FIELD=Ignore aggregate field
GroupByMeta.Injection.PASS_ALL_ROWS=Pass all rows?
GroupByMeta.Injection.TEMP_DIRECTORY=Temporary directory
GroupByMeta.Injection.TEMP_FILE_PREFIX=Temporary file prefix
diff --git
a/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_zh_CN.properties
b/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_zh_CN.properties
index a0b08a9c0a..41bbf7f26d 100644
---
a/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_zh_CN.properties
+++
b/plugins/transforms/groupby/src/main/resources/org/apache/hop/pipeline/transforms/groupby/messages/messages_zh_CN.properties
@@ -24,6 +24,9 @@
GroupBy.Exception.UnableToDeleteTemporaryFile=\u65E0\u6CD5\u5220\u9664\u4E34\u67
GroupBy.Exception.UnableToReadBackRowFromTemporaryFile=Unable to read back row
from temporary file\!
GroupBy.LineNumber=\u884C\u53F7
GroupBy.Log.AggregateSubjectFieldCouldNotFound=Aggregate subject field [{0}]
couldn''t be found\!
+GroupBy.Log.AggregateIgnoredFieldCouldNotFound=\u805A\u5408\u5FFD\u7565\u5B57\u6BB5
[{0}] \u672A\u627E\u5230\!
+GroupBy.Log.AggregateIgnoredFieldMissing=\u5DF2\u542F\u7528\u201C\u805A\u5408\u65F6\u5FFD\u7565\u884C\u201D\uFF0C\u4F46\u672A\u6307\u5B9A\u5FFD\u7565\u5B57\u6BB5\u540D\!
+GroupBy.Log.AggregateIgnoredFieldNotBoolean=\u805A\u5408\u5FFD\u7565\u5B57\u6BB5
[{0}] \u5FC5\u987B\u662F\u5E03\u5C14\u7C7B\u578B\!
GroupBy.Log.GroupFieldCouldNotFound=Grouping field [{0}] couldn''t be found\!
GroupBy.Name=\u5206\u7EC4
GroupByDialog.AddLineNr.Label=\u589E\u52A0\u884C\u53F7,\u6BCF\u7EC4\u91CD\u65B0\u5F00\u59CB\:
@@ -31,6 +34,10 @@ GroupByDialog.Aggregates.Label=\u805A\u5408 :
GroupByDialog.AllRows.Label=\u5305\u62EC\u6240\u6709\u7684\u884C:
GroupByDialog.AlwaysAddResult.Label=\u603B\u8FD4\u56DE\u4E00\u4E2A\u7ED3\u679C\u884C\:
GroupByDialog.AlwaysAddResult.ToolTip=\u5373\u4F7F\u5728\u6CA1\u6709\u8F93\u5165\u884C\u7684\u60C5\u51B5\u4E0B,\u4E5F\u81F3\u5C11\u751F\u6210\u4E00\u884C,\u8FD9\u6837\u5728\u8BA1\u6570\u805A\u96C6\u7684\u60C5\u51B5\u4E0B,\u53EF\u4EE5\u83B7\u5F97\u6B63\u786E\u7ED3\u679C.\r\n\u8BE5\u65B9\u5F0F\u548C
SQL \u8BED\u53E5\u7684 Group by \u7ED3\u679C\u4E00\u81F4.\r\n
+GroupByDialog.IgnoreAggregate.Label=\u805A\u5408\u65F6\u5FFD\u7565\u884C\:
+GroupByDialog.IgnoreAggregate.ToolTip=\u542F\u7528\u540E\uFF0C\u9009\u5B9A\u5E03\u5C14\u5B57\u6BB5\u4E3A
true
\u7684\u884C\u4E0D\u53C2\u4E0E\u805A\u5408\u8BA1\u7B97\u3002\u8BE5\u884C\u4ECD\u4F1A\u53C2\u4E0E\u5206\u7EC4\u5224\u65AD\u3002\u82E5\u542F\u7528\u201C\u5305\u62EC\u6240\u6709\u7684\u884C\u201D\uFF0C\u88AB\u5FFD\u7565\u7684\u884C\u4ECD\u4F1A\u8F93\u51FA\u5E76\u9644\u5E26\u7EC4\u805A\u5408\u7ED3\u679C\u3002
+GroupByDialog.IgnoreAggregateField.Label=\u5FFD\u7565\u5B57\u6BB5\u540D\:
+GroupByDialog.IgnoreAggregateField.ToolTip=\u7528\u4E8E\u6807\u8BB0\u4E0D\u53C2\u4E0E\u805A\u5408\u7684\u884C\u7684\u5E03\u5C14\u5B57\u6BB5\u3002
GroupByDialog.Browse.Button=\u6D4F\u89C8(&B)...
GroupByDialog.ColumnInfo.GroupField=\u5206\u7EC4\u5B57\u6BB5
GroupByDialog.ColumnInfo.Name=\u540D\u79F0
@@ -55,6 +62,11 @@ GroupByDialog.TempDir.Label=\u6392\u5E8F\u76EE\u5F55\:
GroupByDialog.TransformName.Label=Transform \u540D\u79F0:
GroupByMeta.CheckResult.NoInputError=No input received from other transforms\!
GroupByMeta.CheckResult.ReceivingInfoOK=Transform is receiving info from other
transforms.
+GroupByMeta.CheckResult.IgnoreAggregateFieldMissing=\u5DF2\u542F\u7528\u201C\u805A\u5408\u65F6\u5FFD\u7565\u884C\u201D\uFF0C\u4F46\u672A\u6307\u5B9A\u5FFD\u7565\u5B57\u6BB5\u540D\u3002
+GroupByMeta.CheckResult.IgnoreAggregateFieldNotFound=\u5728\u8F93\u5165\u6D41\u4E2D\u672A\u627E\u5230\u805A\u5408\u5FFD\u7565\u5B57\u6BB5
[{0}]\u3002
+GroupByMeta.CheckResult.IgnoreAggregateFieldNotBoolean=\u805A\u5408\u5FFD\u7565\u5B57\u6BB5
[{0}] \u5FC5\u987B\u662F\u5E03\u5C14\u7C7B\u578B\u3002
+GroupByMeta.Injection.IGNORE_AGGREGATE=\u805A\u5408\u65F6\u662F\u5426\u5FFD\u7565\u884C\uFF1F
+GroupByMeta.Injection.IGNORE_AGGREGATE_FIELD=\u805A\u5408\u5FFD\u7565\u5B57\u6BB5
GroupByMeta.Exception.UnableToLoadTransformMetaFromXML=Unable to load
transform info from XML
GroupByMeta.keyword=groupby
GroupByMeta.TypeGroupLongDesc.AVERAGE=\u5E73\u5747
diff --git
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/AggregationTest.java
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/AggregationTest.java
new file mode 100644
index 0000000000..a4965be68d
--- /dev/null
+++
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/AggregationTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+
+import org.apache.hop.core.HopEnvironment;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class AggregationTest {
+
+ @BeforeAll
+ static void setUpClass() throws Exception {
+ HopEnvironment.init();
+ }
+
+ @Test
+ void setTypeLabelKeepsNumericTypeInSync() {
+ Aggregation aggregation = new Aggregation();
+ aggregation.setTypeLabel("SUM");
+
+ assertEquals("SUM", aggregation.getTypeLabel());
+ assertEquals(Aggregation.TYPE_GROUP_SUM, aggregation.getType());
+
+ aggregation.setTypeLabel("COUNT_ALL");
+ assertEquals("COUNT_ALL", aggregation.getTypeLabel());
+ assertEquals(Aggregation.TYPE_GROUP_COUNT_ALL, aggregation.getType());
+ }
+
+ @Test
+ void constructorResolvesTypeFromLongDescription() {
+ Aggregation aggregation =
+ new Aggregation(
+ "sum_amount",
+ "amount",
+ Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null);
+
+ assertEquals("sum_amount", aggregation.getField());
+ assertEquals("amount", aggregation.getSubject());
+ assertEquals(Aggregation.TYPE_GROUP_SUM, aggregation.getType());
+ assertEquals("SUM", aggregation.getTypeLabel());
+ }
+
+ @Test
+ void cloneCreatesIndependentCopy() {
+ Aggregation original =
+ new Aggregation(
+ "moving_avg",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_MOVING_AVERAGE),
+ "3",
+ "ts");
+
+ Aggregation copy = original.clone();
+
+ assertNotSame(original, copy);
+ assertEquals(original, copy);
+
+ copy.setField("other");
+ assertNotEquals(original.getField(), copy.getField());
+ }
+
+ @Test
+ void equalsAndHashCodeConsiderAllFields() {
+ Aggregation left =
+ new Aggregation(
+ "sum_amount",
+ "amount",
+ Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null);
+ Aggregation right =
+ new Aggregation(
+ "sum_amount",
+ "amount",
+ Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null);
+ Aggregation different =
+ new Aggregation(
+ "cnt",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_COUNT_ALL),
+ null);
+
+ assertEquals(left, right);
+ assertEquals(left.hashCode(), right.hashCode());
+ assertNotEquals(left, different);
+ }
+}
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/groupby/GroupByMetaTest.java
similarity index 53%
rename from
plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/GroupByMetaTest.java
rename to
plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupByMetaTest.java
index 5c7ef9c666..98bf9e6128 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/groupby/GroupByMetaTest.java
@@ -13,26 +13,40 @@
* 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;
+package org.apache.hop.pipeline.transforms.groupby;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBoolean;
+import org.apache.hop.core.row.value.ValueMetaNumber;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.Variables;
import org.apache.hop.core.xml.XmlHandler;
import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
import org.apache.hop.metadata.serializer.xml.XmlMetadataUtil;
-import org.apache.hop.pipeline.transforms.groupby.Aggregation;
-import org.apache.hop.pipeline.transforms.groupby.GroupByMeta;
-import org.apache.hop.pipeline.transforms.groupby.GroupingField;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
class GroupByMetaTest {
+ @BeforeAll
+ static void setUpClass() throws Exception {
+ HopEnvironment.init();
+ }
+
@Test
- void testClone() throws Exception {
+ void testClone() {
GroupByMeta meta1 = generateTestMeta();
GroupByMeta meta2 = (GroupByMeta) meta1.clone();
@@ -53,24 +67,100 @@ class GroupByMetaTest {
compareMetas(meta1, meta2);
}
- public void compareMetas(GroupByMeta meta1, GroupByMeta meta2) {
+ @Test
+ void testSetDefault() {
+ GroupByMeta meta = new GroupByMeta();
+ meta.setDefault();
+
+ assertEquals("${java.io.tmpdir}", meta.getDirectory());
+ assertEquals("grp", meta.getPrefix());
+ assertFalse(meta.isPassAllRows());
+ assertFalse(meta.isAggregateIgnored());
+ assertNull(meta.getAggregateIgnoredField());
+ assertNotNull(meta.getGroupingFields());
+ assertNotNull(meta.getAggregations());
+ }
+
+ @Test
+ void testGetFieldsWithoutPassAllRows() {
+ GroupByMeta meta = new GroupByMeta();
+ meta.setDefault();
+ meta.getGroupingFields().add(new GroupingField("grp"));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "cnt",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_COUNT_ALL),
+ null));
+
+ IRowMeta rowMeta = inputRowMeta();
+ meta.getFields(rowMeta, "Group by", null, null, new Variables(), null);
+
+ assertEquals(3, rowMeta.size());
+ assertEquals("grp", rowMeta.getValueMeta(0).getName());
+ assertEquals("sum_amount", rowMeta.getValueMeta(1).getName());
+ assertEquals(IValueMeta.TYPE_NUMBER, rowMeta.getValueMeta(1).getType());
+ assertEquals("cnt", rowMeta.getValueMeta(2).getName());
+ assertEquals(IValueMeta.TYPE_INTEGER, rowMeta.getValueMeta(2).getType());
+ }
+
+ @Test
+ void testGetFieldsWithPassAllRowsAndLineNr() {
+ GroupByMeta meta = new GroupByMeta();
+ meta.setDefault();
+ meta.setPassAllRows(true);
+ meta.setAddingLineNrInGroup(true);
+ meta.setLineNrInGroupField("linenr");
+ meta.getGroupingFields().add(new GroupingField("grp"));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+
+ IRowMeta rowMeta = inputRowMeta();
+ meta.getFields(rowMeta, "Group by", null, null, new Variables(), null);
+
+ assertTrue(rowMeta.indexOfValue("grp") >= 0);
+ assertTrue(rowMeta.indexOfValue("amount") >= 0);
+ assertTrue(rowMeta.indexOfValue("skip") >= 0);
+ assertTrue(rowMeta.indexOfValue("sum_amount") >= 0);
+ assertTrue(rowMeta.indexOfValue("linenr") >= 0);
+ assertEquals(IValueMeta.TYPE_INTEGER,
rowMeta.searchValueMeta("linenr").getType());
+ }
+
+ @Test
+ void testSupportsMultiCopyExecution() {
+ assertFalse(new GroupByMeta().supportsMultiCopyExecution());
+ }
+
+ private void compareMetas(GroupByMeta meta1, GroupByMeta meta2) {
assertEquals(meta1.getGroupingFields().size(),
meta2.getGroupingFields().size());
for (int i = 0; i < meta1.getGroupingFields().size(); i++) {
- GroupingField field1 = meta1.getGroupingFields().get(i);
- GroupingField field2 = meta2.getGroupingFields().get(i);
- assertEquals(field1, field2);
+ assertEquals(meta1.getGroupingFields().get(i),
meta2.getGroupingFields().get(i));
}
assertEquals(meta1.getAggregations().size(),
meta2.getAggregations().size());
for (int i = 0; i < meta1.getAggregations().size(); i++) {
- Aggregation agg1 = meta1.getAggregations().get(i);
- Aggregation agg2 = meta2.getAggregations().get(i);
- assertEquals(agg1, agg2);
+ assertEquals(meta1.getAggregations().get(i),
meta2.getAggregations().get(i));
}
assertEquals(meta1.isPassAllRows(), meta2.isPassAllRows());
assertEquals(meta1.isAddingLineNrInGroup(), meta2.isAddingLineNrInGroup());
assertEquals(meta1.getLineNrInGroupField(), meta2.getLineNrInGroupField());
assertEquals(meta1.getDirectory(), meta2.getDirectory());
assertEquals(meta1.getPrefix(), meta2.getPrefix());
+ assertEquals(meta1.isAlwaysGivingBackOneRow(),
meta2.isAlwaysGivingBackOneRow());
+ assertEquals(meta1.isAggregateIgnored(), meta2.isAggregateIgnored());
+ assertEquals(meta1.getAggregateIgnoredField(),
meta2.getAggregateIgnoredField());
}
private GroupByMeta generateTestMeta() {
@@ -108,9 +198,19 @@ class GroupByMetaTest {
meta.setPrefix("prefix");
meta.setAddingLineNrInGroup(true);
meta.setLineNrInGroupField("lineNr");
+ meta.setAggregateIgnored(true);
+ meta.setAggregateIgnoredField("skip");
return meta;
}
+ private IRowMeta inputRowMeta() {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("grp"));
+ rowMeta.addValueMeta(new ValueMetaNumber("amount"));
+ rowMeta.addValueMeta(new ValueMetaBoolean("skip"));
+ return rowMeta;
+ }
+
private String getDesc(String label) {
int type = Aggregation.getTypeCodeFromLabel(label);
return Aggregation.getTypeDescLongFromCode(type);
diff --git
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupByTest.java
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupByTest.java
new file mode 100644
index 0000000000..7110ce1463
--- /dev/null
+++
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupByTest.java
@@ -0,0 +1,344 @@
+/*
+ * 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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.QueueRowSet;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILoggingObject;
+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.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for the {@link GroupBy} transform worker. */
+class GroupByTest {
+
+ private TransformMockHelper<GroupByMeta, GroupByData> mockHelper;
+ private GroupByMeta meta;
+ private GroupByData data;
+ private GroupBy groupBy;
+ private IRowMeta inputRowMeta;
+
+ @BeforeAll
+ static void setUpClass() throws HopException {
+ HopEnvironment.init();
+ }
+
+ @BeforeEach
+ void setUp() throws HopException {
+ mockHelper = new TransformMockHelper<>("Group By", GroupByMeta.class,
GroupByData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+ when(mockHelper.pipeline.isRunning()).thenReturn(true);
+
+ meta = new GroupByMeta();
+ meta.setDefault();
+ meta.getGroupingFields().add(new GroupingField("grp"));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "cnt",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_COUNT_ALL),
+ null));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "avg_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_AVERAGE),
+ null));
+
+ data = new GroupByData();
+ inputRowMeta = new RowMeta();
+ inputRowMeta.addValueMeta(new ValueMetaString("grp"));
+ inputRowMeta.addValueMeta(new ValueMetaNumber("amount"));
+
+ groupBy = createGroupBy(meta, data);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ @Test
+ void initPreparesBuffer() {
+ assertTrue(groupBy.init());
+ assertNotNullBuffer();
+ }
+
+ @Test
+ void processRowAggregatesSingleGroup() throws Exception {
+ List<Object[]> rows =
+ runPipeline(new Object[] {"A", 10.0}, new Object[] {"A", 20.0}, new
Object[] {"A", 30.0});
+
+ assertEquals(1, rows.size());
+ Object[] result = rows.getFirst();
+ assertEquals("A", result[0]);
+ assertEquals(60.0, ((Number) result[1]).doubleValue(), 1e-9);
+ assertEquals(3L, ((Number) result[2]).longValue());
+ assertEquals(20.0, ((Number) result[3]).doubleValue(), 1e-9);
+ }
+
+ @Test
+ void processRowAggregatesMultipleGroups() throws Exception {
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0},
+ new Object[] {"A", 20.0},
+ new Object[] {"B", 5.0},
+ new Object[] {"B", 15.0},
+ new Object[] {"C", 100.0});
+
+ assertEquals(3, rows.size());
+
+ assertEquals("A", rows.getFirst()[0]);
+ assertEquals(30.0, ((Number) rows.get(0)[1]).doubleValue(), 1e-9);
+ assertEquals(2L, ((Number) rows.get(0)[2]).longValue());
+
+ assertEquals("B", rows.get(1)[0]);
+ assertEquals(20.0, ((Number) rows.get(1)[1]).doubleValue(), 1e-9);
+ assertEquals(2L, ((Number) rows.get(1)[2]).longValue());
+
+ assertEquals("C", rows.get(2)[0]);
+ assertEquals(100.0, ((Number) rows.get(2)[1]).doubleValue(), 1e-9);
+ assertEquals(1L, ((Number) rows.get(2)[2]).longValue());
+ }
+
+ @Test
+ void processRowWithNoInputAndAlwaysGiveBackRow() throws Exception {
+ meta.setAlwaysGivingBackOneRow(true);
+ meta.getGroupingFields().clear();
+ meta.getAggregations().clear();
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "cnt",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_COUNT_ANY),
+ null));
+
+ data = new GroupByData();
+ groupBy = createGroupBy(meta, data);
+
+ QueueRowSet input = new QueueRowSet();
+ input.setDone();
+ groupBy.addRowSetToInputRowSets(input);
+ groupBy.setInputRowMeta(inputRowMeta);
+
+ QueueRowSet output = new QueueRowSet();
+ groupBy.addRowSetToOutputRowSets(output);
+
+ assertFalse(groupBy.processRow());
+
+ List<Object[]> rows = drain(output);
+ assertEquals(1, rows.size());
+ assertEquals(0L, ((Number) rows.getFirst()[0]).longValue());
+ }
+
+ @Test
+ void processRowWithNoInputAndNoAlwaysGiveBack() throws Exception {
+ meta.setAlwaysGivingBackOneRow(false);
+
+ QueueRowSet input = new QueueRowSet();
+ input.setDone();
+ groupBy.addRowSetToInputRowSets(input);
+
+ QueueRowSet output = new QueueRowSet();
+ groupBy.addRowSetToOutputRowSets(output);
+
+ assertFalse(groupBy.processRow());
+ assertTrue(drain(output).isEmpty());
+ }
+
+ @Test
+ void processRowPassAllRowsKeepsInputAndAddsAggregates() throws Exception {
+ meta.setPassAllRows(true);
+ meta.getAggregations().clear();
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+
+ data = new GroupByData();
+ groupBy = createGroupBy(meta, data);
+
+ List<Object[]> rows =
+ runPipeline(new Object[] {"A", 10.0}, new Object[] {"A", 20.0}, new
Object[] {"B", 5.0});
+
+ assertEquals(3, rows.size());
+
+ // Input fields are preserved; aggregate is appended.
+ assertEquals("A", rows.getFirst()[0]);
+ assertEquals(10.0, ((Number) rows.get(0)[1]).doubleValue(), 1e-9);
+ assertEquals(30.0, ((Number) rows.get(0)[2]).doubleValue(), 1e-9);
+
+ assertEquals("A", rows.get(1)[0]);
+ assertEquals(20.0, ((Number) rows.get(1)[1]).doubleValue(), 1e-9);
+ assertEquals(30.0, ((Number) rows.get(1)[2]).doubleValue(), 1e-9);
+
+ assertEquals("B", rows.get(2)[0]);
+ assertEquals(5.0, ((Number) rows.get(2)[1]).doubleValue(), 1e-9);
+ assertEquals(5.0, ((Number) rows.get(2)[2]).doubleValue(), 1e-9);
+ }
+
+ @Test
+ void sameGroupComparesConfiguredGroupFields() throws Exception {
+ data.inputRowMeta = inputRowMeta;
+ data.groupnrs = new int[] {0};
+
+ assertTrue(groupBy.sameGroup(new Object[] {"A", 1.0}, new Object[] {"A",
2.0}));
+ assertFalse(groupBy.sameGroup(new Object[] {"A", 1.0}, new Object[] {"B",
1.0}));
+ }
+
+ @Test
+ void calcAggregateUpdatesSumAndCount() throws Exception {
+ data.inputRowMeta = inputRowMeta;
+ data.subjectnrs = new int[] {1, 1};
+ data.counts = new long[2];
+ data.mean = new double[2];
+ data.previousSums = new Object[0];
+ data.previousAvgSum = new Object[0];
+ data.previousAvgCount = new long[0];
+ data.movingAvgWindows = new ArrayDeque[2];
+
+ meta.getAggregations().clear();
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "cnt",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_COUNT_ALL),
+ null));
+
+ groupBy.newAggregate(new Object[] {"A", 10.0});
+ groupBy.calcAggregate(new Object[] {"A", 10.0});
+ groupBy.calcAggregate(new Object[] {"A", 20.0});
+
+ Object[] result = groupBy.getAggregateResult();
+ assertEquals(30.0, ((Number) result[0]).doubleValue(), 1e-9);
+ assertEquals(2L, ((Number) result[1]).longValue());
+ }
+
+ @Test
+ void batchCompleteFlushesCurrentGroup() throws Exception {
+ QueueRowSet input = new QueueRowSet();
+ input.putRow(inputRowMeta, new Object[] {"A", 10.0});
+ input.putRow(inputRowMeta, new Object[] {"A", 20.0});
+ // Intentionally not done yet: batchComplete should flush the open group.
+ groupBy.addRowSetToInputRowSets(input);
+
+ QueueRowSet output = new QueueRowSet();
+ groupBy.addRowSetToOutputRowSets(output);
+
+ assertTrue(groupBy.processRow());
+ assertTrue(groupBy.processRow());
+ assertTrue(drain(output).isEmpty());
+
+ groupBy.batchComplete();
+
+ List<Object[]> rows = drain(output);
+ assertEquals(1, rows.size());
+ assertEquals("A", rows.getFirst()[0]);
+ assertEquals(30.0, ((Number) rows.getFirst()[1]).doubleValue(), 1e-9);
+ assertEquals(2L, ((Number) rows.getFirst()[2]).longValue());
+ assertTrue(data.newBatch);
+ }
+
+ private GroupBy createGroupBy(GroupByMeta transformMeta, GroupByData
transformData) {
+ GroupBy transform =
+ new GroupBy(
+ mockHelper.transformMeta,
+ transformMeta,
+ transformData,
+ 0,
+ mockHelper.pipelineMeta,
+ mockHelper.pipeline) {
+ @Override
+ public String resolve(String str) {
+ return str;
+ }
+ };
+ transform.init();
+ return transform;
+ }
+
+ private List<Object[]> runPipeline(Object[]... inputRows) throws
HopException {
+ QueueRowSet input = new QueueRowSet();
+ for (Object[] inputRow : inputRows) {
+ input.putRow(inputRowMeta, inputRow);
+ }
+ input.setDone();
+ groupBy.addRowSetToInputRowSets(input);
+
+ QueueRowSet output = new QueueRowSet();
+ groupBy.addRowSetToOutputRowSets(output);
+
+ while (groupBy.processRow()) {
+ // drain all input
+ }
+ return drain(output);
+ }
+
+ private List<Object[]> drain(QueueRowSet output) {
+ List<Object[]> rows = new ArrayList<>();
+ Object[] row;
+ while ((row = output.getRowImmediate()) != null) {
+ rows.add(row);
+ }
+ return rows;
+ }
+
+ private void assertNotNullBuffer() {
+ assertNotNull(data.bufferList);
+ }
+}
diff --git
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupingFieldTest.java
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupingFieldTest.java
new file mode 100644
index 0000000000..2cfc41393b
--- /dev/null
+++
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/GroupingFieldTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+
+import org.junit.jupiter.api.Test;
+
+class GroupingFieldTest {
+
+ @Test
+ void equalsAndHashCodeUseName() {
+ GroupingField left = new GroupingField("grp");
+ GroupingField right = new GroupingField("grp");
+ GroupingField other = new GroupingField("other");
+
+ assertEquals(left, right);
+ assertEquals(left.hashCode(), right.hashCode());
+ assertNotEquals(left, other);
+ }
+
+ @Test
+ void cloneCreatesIndependentCopy() {
+ GroupingField original = new GroupingField("grp");
+ GroupingField copy = original.clone();
+
+ assertNotSame(original, copy);
+ assertEquals(original, copy);
+
+ copy.setName("other");
+ assertEquals("grp", original.getName());
+ assertEquals("other", copy.getName());
+ }
+}
diff --git
a/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/IgnoreAggregateTest.java
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/IgnoreAggregateTest.java
new file mode 100644
index 0000000000..d8794be483
--- /dev/null
+++
b/plugins/transforms/groupby/src/test/java/org/apache/hop/pipeline/transforms/groupby/IgnoreAggregateTest.java
@@ -0,0 +1,385 @@
+/*
+ * 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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.QueueRowSet;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBoolean;
+import org.apache.hop.core.row.value.ValueMetaNumber;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for Group By ignore_aggregate / field_ignore. */
+class IgnoreAggregateTest {
+
+ private TransformMockHelper<GroupByMeta, GroupByData> mockHelper;
+ private GroupByMeta meta;
+ private GroupByData data;
+ private GroupBy groupBy;
+ private IRowMeta inputRowMeta;
+
+ @BeforeAll
+ static void setUpClass() throws HopException {
+ HopEnvironment.init();
+ }
+
+ @BeforeEach
+ void setUp() throws HopException {
+ mockHelper = new TransformMockHelper<>("Group By", GroupByMeta.class,
GroupByData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+ when(mockHelper.pipeline.isRunning()).thenReturn(true);
+
+ meta = new GroupByMeta();
+ meta.setDefault();
+ meta.setAggregateIgnored(true);
+ meta.setAggregateIgnoredField("skip");
+ meta.getGroupingFields().add(new GroupingField("grp"));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "cnt",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_COUNT_ALL),
+ null));
+
+ data = new GroupByData();
+ inputRowMeta = new RowMeta();
+ inputRowMeta.addValueMeta(new ValueMetaString("grp"));
+ inputRowMeta.addValueMeta(new ValueMetaNumber("amount"));
+ inputRowMeta.addValueMeta(new ValueMetaBoolean("skip"));
+
+ groupBy =
+ new GroupBy(
+ mockHelper.transformMeta, meta, data, 0, mockHelper.pipelineMeta,
mockHelper.pipeline) {
+ @Override
+ public String resolve(String str) {
+ return str;
+ }
+ };
+ groupBy.init();
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ @Test
+ void isRowAggregateIgnoredRespectsBooleanField() throws Exception {
+ data.inputRowMeta = inputRowMeta;
+ data.aggregateIgnoredFieldIndex = 2;
+
+ assertFalse(groupBy.isRowAggregateIgnored(new Object[] {"A", 10.0,
Boolean.FALSE}));
+ assertTrue(groupBy.isRowAggregateIgnored(new Object[] {"A", 100.0,
Boolean.TRUE}));
+ assertFalse(groupBy.isRowAggregateIgnored(new Object[] {"A", 10.0, null}));
+ }
+
+ @Test
+ void isRowAggregateIgnoredDisabledWhenIndexUnset() throws Exception {
+ data.inputRowMeta = inputRowMeta;
+ data.aggregateIgnoredFieldIndex = -1;
+
+ assertFalse(groupBy.isRowAggregateIgnored(new Object[] {"A", 100.0,
Boolean.TRUE}));
+ }
+
+ @Test
+ void processRowSkipsIgnoredRowsInAggregation() throws Exception {
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 20.0, Boolean.FALSE},
+ new Object[] {"A", 30.0, Boolean.FALSE},
+ new Object[] {"A", 100.0, Boolean.TRUE});
+
+ assertEquals(1, rows.size());
+ Object[] result = rows.getFirst();
+ assertEquals("A", result[0]);
+ assertEquals(60.0, ((Number) result[1]).doubleValue(), 1e-9);
+ assertEquals(3L, ((Number) result[2]).longValue());
+ }
+
+ @Test
+ void processRowIncludesAllRowsWhenIgnoreDisabled() throws Exception {
+ meta.setAggregateIgnored(false);
+ meta.setAggregateIgnoredField(null);
+
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 20.0, Boolean.FALSE},
+ new Object[] {"A", 30.0, Boolean.FALSE},
+ new Object[] {"A", 100.0, Boolean.TRUE});
+
+ assertEquals(1, rows.size());
+ Object[] result = rows.getFirst();
+ assertEquals(160.0, ((Number) result[1]).doubleValue(), 1e-9);
+ assertEquals(4L, ((Number) result[2]).longValue());
+ }
+
+ @Test
+ void processRowSkipsIgnoredRowsPerGroup() throws Exception {
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 20.0, Boolean.FALSE},
+ new Object[] {"A", 30.0, Boolean.FALSE},
+ new Object[] {"A", 100.0, Boolean.TRUE},
+ new Object[] {"B", 5.0, Boolean.FALSE},
+ new Object[] {"B", 50.0, Boolean.TRUE});
+
+ assertEquals(2, rows.size());
+
+ Object[] groupA = rows.getFirst();
+ assertEquals("A", groupA[0]);
+ assertEquals(60.0, ((Number) groupA[1]).doubleValue(), 1e-9);
+ assertEquals(3L, ((Number) groupA[2]).longValue());
+
+ Object[] groupB = rows.get(1);
+ assertEquals("B", groupB[0]);
+ assertEquals(5.0, ((Number) groupB[1]).doubleValue(), 1e-9);
+ assertEquals(1L, ((Number) groupB[2]).longValue());
+ }
+
+ @Test
+ void processRowIgnoresLastRowWhenFlagged() throws Exception {
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 20.0, Boolean.FALSE},
+ new Object[] {"A", 100.0, Boolean.TRUE});
+
+ assertEquals(1, rows.size());
+ Object[] result = rows.getFirst();
+ assertEquals(30.0, ((Number) result[1]).doubleValue(), 1e-9);
+ assertEquals(2L, ((Number) result[2]).longValue());
+ }
+
+ @Test
+ void processRowWithOnlyIgnoredRowsStillEmitsGroup() throws Exception {
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 100.0, Boolean.TRUE}, new Object[] {"A", 200.0,
Boolean.TRUE});
+
+ assertEquals(1, rows.size());
+ Object[] result = rows.getFirst();
+ assertEquals("A", result[0]);
+ assertNull(result[1]); // sum stays null when every row is ignored
+ assertEquals(0L, ((Number) result[2]).longValue());
+ }
+
+ @Test
+ void processRowSkipsFirstRowWhenFlaggedForMinMaxFirstLast() throws Exception
{
+ configureAggregations(
+ Aggregation.TYPE_GROUP_MIN,
+ Aggregation.TYPE_GROUP_MAX,
+ Aggregation.TYPE_GROUP_FIRST,
+ Aggregation.TYPE_GROUP_LAST,
+ Aggregation.TYPE_GROUP_FIRST_INCL_NULL,
+ Aggregation.TYPE_GROUP_LAST_INCL_NULL);
+
+ // Group A: ignored first row holds extreme values; valid rows are 10 and
20
+ // Group B: ignored first row holds extremes; valid rows are 10 and 20
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 999.0, Boolean.TRUE},
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 20.0, Boolean.FALSE},
+ new Object[] {"B", 1.0, Boolean.TRUE},
+ new Object[] {"B", 10.0, Boolean.FALSE},
+ new Object[] {"B", 20.0, Boolean.FALSE});
+
+ assertEquals(2, rows.size());
+
+ Object[] groupA = rows.getFirst();
+ assertEquals("A", groupA[0]);
+ assertEquals(10.0, ((Number) groupA[1]).doubleValue(), 1e-9); // min
+ assertEquals(20.0, ((Number) groupA[2]).doubleValue(), 1e-9); // max
+ assertEquals(10.0, ((Number) groupA[3]).doubleValue(), 1e-9); // first
+ assertEquals(20.0, ((Number) groupA[4]).doubleValue(), 1e-9); // last
+ assertEquals(10.0, ((Number) groupA[5]).doubleValue(), 1e-9); // first
incl null
+ assertEquals(20.0, ((Number) groupA[6]).doubleValue(), 1e-9); // last incl
null
+
+ Object[] groupB = rows.get(1);
+ assertEquals("B", groupB[0]);
+ assertEquals(10.0, ((Number) groupB[1]).doubleValue(), 1e-9); // min (not
1.0)
+ assertEquals(20.0, ((Number) groupB[2]).doubleValue(), 1e-9); // max
+ assertEquals(10.0, ((Number) groupB[3]).doubleValue(), 1e-9); // first
(not 1.0)
+ assertEquals(20.0, ((Number) groupB[4]).doubleValue(), 1e-9); // last
+ assertEquals(10.0, ((Number) groupB[5]).doubleValue(), 1e-9); // first
incl null
+ assertEquals(20.0, ((Number) groupB[6]).doubleValue(), 1e-9); // last incl
null
+ }
+
+ @Test
+ void processRowAllIgnoredLeavesMinMaxFirstNull() throws Exception {
+ configureAggregations(
+ Aggregation.TYPE_GROUP_MIN,
+ Aggregation.TYPE_GROUP_MAX,
+ Aggregation.TYPE_GROUP_FIRST,
+ Aggregation.TYPE_GROUP_FIRST_INCL_NULL,
+ Aggregation.TYPE_GROUP_COUNT_ALL);
+
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 100.0, Boolean.TRUE}, new Object[] {"A", 200.0,
Boolean.TRUE});
+
+ assertEquals(1, rows.size());
+ Object[] result = rows.getFirst();
+ assertEquals("A", result[0]);
+ assertNull(result[1]); // min
+ assertNull(result[2]); // max
+ assertNull(result[3]); // first
+ assertNull(result[4]); // first incl null
+ assertEquals(0L, ((Number) result[5]).longValue()); // count
+ }
+
+ @Test
+ void processRowPassAllRowsSkipsIgnoredInCumulativeSum() throws Exception {
+ meta.setPassAllRows(true);
+ meta.getAggregations().clear();
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "cum_sum",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_CUMULATIVE_SUM),
+ null));
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "sum_amount",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_SUM),
+ null));
+
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 100.0, Boolean.TRUE},
+ new Object[] {"A", 20.0, Boolean.FALSE});
+
+ assertEquals(3, rows.size());
+
+ // output: grp, amount, skip, cum_sum, sum_amount
+ assertEquals(10.0, ((Number) rows.get(0)[3]).doubleValue(), 1e-9);
+ assertEquals(10.0, ((Number) rows.get(1)[3]).doubleValue(), 1e-9); //
ignored: carry previous
+ assertEquals(30.0, ((Number) rows.get(2)[3]).doubleValue(), 1e-9);
+
+ assertEquals(30.0, ((Number) rows.get(0)[4]).doubleValue(), 1e-9); //
group sum on every row
+ assertEquals(30.0, ((Number) rows.get(1)[4]).doubleValue(), 1e-9);
+ assertEquals(30.0, ((Number) rows.get(2)[4]).doubleValue(), 1e-9);
+ }
+
+ @Test
+ void processRowPassAllRowsSkipsIgnoredInMovingAverage() throws Exception {
+ meta.setPassAllRows(true);
+ meta.getAggregations().clear();
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "mov_avg",
+ "amount",
+
Aggregation.getTypeDescLongFromCode(Aggregation.TYPE_GROUP_MOVING_AVERAGE),
+ "2",
+ null));
+
+ List<Object[]> rows =
+ runPipeline(
+ new Object[] {"A", 10.0, Boolean.FALSE},
+ new Object[] {"A", 100.0, Boolean.TRUE},
+ new Object[] {"A", 20.0, Boolean.FALSE});
+
+ assertEquals(3, rows.size());
+ // window size 2: after first valid row incomplete; ignored does not enter
window;
+ // after second valid row window is [10, 20] -> 15
+ assertNull(rows.get(0)[3]);
+ assertNull(rows.get(1)[3]); // ignored: still incomplete window
+ assertEquals(15.0, ((Number) rows.get(2)[3]).doubleValue(), 1e-9);
+ }
+
+ @Test
+ void processRowFailsWhenIgnoreEnabledButFieldBlank() throws Exception {
+ meta.setAggregateIgnoredField("");
+
+ QueueRowSet input = new QueueRowSet();
+ input.putRow(inputRowMeta, new Object[] {"A", 10.0, Boolean.FALSE});
+ input.setDone();
+ groupBy.addRowSetToInputRowSets(input);
+ groupBy.addRowSetToOutputRowSets(new QueueRowSet());
+
+ assertFalse(groupBy.processRow());
+ assertTrue(groupBy.getErrors() > 0);
+ }
+
+ private void configureAggregations(int... types) {
+ meta.getAggregations().clear();
+ for (int i = 0; i < types.length; i++) {
+ meta.getAggregations()
+ .add(
+ new Aggregation(
+ "agg" + i, "amount",
Aggregation.getTypeDescLongFromCode(types[i]), null));
+ }
+ }
+
+ private List<Object[]> runPipeline(Object[]... inputRows) throws
HopException {
+ QueueRowSet input = new QueueRowSet();
+ for (Object[] inputRow : inputRows) {
+ input.putRow(inputRowMeta, inputRow);
+ }
+ input.setDone();
+ groupBy.addRowSetToInputRowSets(input);
+
+ QueueRowSet output = new QueueRowSet();
+ groupBy.addRowSetToOutputRowSets(output);
+
+ while (groupBy.processRow()) {
+ // drain all input
+ }
+
+ List<Object[]> rows = new ArrayList<>();
+ Object[] row;
+ while ((row = output.getRowImmediate()) != null) {
+ rows.add(row);
+ }
+ return rows;
+ }
+}
