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 8b24a40ea6 issue #7488 : Sorted Schema Merge transform (#7489)
8b24a40ea6 is described below
commit 8b24a40ea6917fd8c798d42900b1b0f759a7e705
Author: Matt Casters <[email protected]>
AuthorDate: Mon Jul 13 16:21:16 2026 +0200
issue #7488 : Sorted Schema Merge transform (#7489)
Port the Sorted Schema Merge transform from hop-data-vault into Hop as a
Flow transform that merges pre-sorted input streams with different schemas
into one globally sorted sparse output stream. Includes unit tests, docs,
and an integration test with a golden data set.
---
assemblies/plugins/pom.xml | 6 +
.../images/transforms/icons/sortedschemamerge.svg | 23 ++
docs/hop-user-manual/modules/ROOT/nav.adoc | 1 +
.../modules/ROOT/pages/pipeline/transforms.adoc | 1 +
.../pipeline/transforms/sortedschemamerge.adoc | 63 ++++
.../transforms/0106-sorted-schema-merge.hpl | 355 +++++++++++++++++++++
.../datasets/golden-sorted-schema-merge.csv | 11 +
.../transforms/main-0106-sorted-schema-merge.hwf | 80 +++++
.../dataset/golden-sorted-schema-merge.json | 64 ++++
.../unit-test/0106-sorted-schema-merge UNIT.json | 53 +++
plugins/transforms/pom.xml | 1 +
plugins/transforms/sortedschemamerge/pom.xml | 32 ++
.../sortedschemamerge/src/assembly/assembly.xml | 46 +++
.../sortedschemamerge/SortedSchemaMerge.java | 162 ++++++++++
.../sortedschemamerge/SortedSchemaMergeData.java | 40 +++
.../sortedschemamerge/SortedSchemaMergeDialog.java | 206 ++++++++++++
.../sortedschemamerge/SortedSchemaMergeInput.java | 44 +++
.../sortedschemamerge/SortedSchemaMergeLogic.java | 159 +++++++++
.../sortedschemamerge/SortedSchemaMergeMeta.java | 281 ++++++++++++++++
.../SortedSchemaMergeMetaFactory.java | 62 ++++
.../sortedschemamerge/SortedSchemaMergeRow.java | 54 ++++
.../SortedSchemaMergeSortKey.java | 49 +++
.../messages/messages_en_US.properties | 44 +++
.../src/main/resources/sortedschemamerge.svg | 23 ++
.../src/main/resources/version.xml | 20 ++
.../SortedSchemaMergeLogicTest.java | 158 +++++++++
26 files changed, 2038 insertions(+)
diff --git a/assemblies/plugins/pom.xml b/assemblies/plugins/pom.xml
index ef42a43597..6fc89979ce 100644
--- a/assemblies/plugins/pom.xml
+++ b/assemblies/plugins/pom.xml
@@ -1459,6 +1459,12 @@
<version>${project.version}</version>
<type>zip</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-transform-sortedschemamerge</artifactId>
+ <version>${project.version}</version>
+ <type>zip</type>
+ </dependency>
<dependency>
<groupId>org.apache.hop</groupId>
<artifactId>hop-transform-splitfieldtorows</artifactId>
diff --git
a/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/sortedschemamerge.svg
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/sortedschemamerge.svg
new file mode 100644
index 0000000000..9beaf28777
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/sortedschemamerge.svg
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN"
"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd";>
+<svg version="1.1" xmlns="http://www.w3.org/2000/svg"; x="0px" y="0px"
+ width="42px" height="42px" viewBox="0 0 42 42" enable-background="new 0 0
42 42">
+ <!-- Two source schemas (different column layouts) -->
+ <path fill="#0E3A5A" d="M18.5,3.5H2.5v12h16V3.5z M16.5,13.5h-3v-8h3V13.5z
M12.5,13.5h-3v-8h3V13.5z M8.5,13.5h-4v-8h4V13.5z"/>
+ <rect x="4.5" y="5.5" fill="#C9E8FB" width="4" height="8"/>
+ <rect x="9.5" y="5.5" fill="#C9E8FB" width="3" height="8"/>
+ <path fill="#0E3A5A" d="M18.5,26.5H2.5v12h16V26.5z M16.5,36.5h-4v-8h4V36.5z
M11.5,36.5h-3v-8h3V36.5z M7.5,36.5h-3v-8h3V36.5z"/>
+ <rect x="4.5" y="28.5" fill="#C9E8FB" width="3" height="8"/>
+ <rect x="8.5" y="28.5" fill="#C9E8FB" width="3" height="8"/>
+ <rect x="12.5" y="28.5" fill="#C9E8FB" width="4" height="8"/>
+ <!-- Merge funnel into sparse sorted output -->
+ <path fill="#0E3A5A" d="M20,9.5h6.5v-3h8v3H39v7.5h-4.5v3H25.5v-3H20V9.5z"/>
+ <rect x="22" y="11.5" fill="#FF9C04" width="5" height="1.5"/>
+ <rect x="22" y="13.7" fill="#FF9C04" width="3.5" height="1.5"/>
+ <path fill="#5CC0C4"
d="M20,32.5h6.5v3h8v-3H39v-7.5h-4.5v-3H25.5v3H20V32.5z"/>
+ <!-- Output sparse row bar -->
+ <path fill="#0E3A5A" d="M39.5,18.5h-14v5h14V18.5z"/>
+ <rect x="26.5" y="19.7" fill="#FFFFFF" width="3" height="2.6"/>
+ <rect x="30.5" y="19.7" fill="#FF9C04" width="3" height="2.6"/>
+ <rect x="34.5" y="19.7" fill="#C9E8FB" width="3.5" height="2.6"/>
+</svg>
diff --git a/docs/hop-user-manual/modules/ROOT/nav.adoc
b/docs/hop-user-manual/modules/ROOT/nav.adoc
index 422e970bae..43c7a9a476 100644
--- a/docs/hop-user-manual/modules/ROOT/nav.adoc
+++ b/docs/hop-user-manual/modules/ROOT/nav.adoc
@@ -251,6 +251,7 @@ under the License.
*** xref:pipeline/transforms/snowflakebulkloader.adoc[Snowflake Bulk Loader]
*** xref:pipeline/transforms/sort.adoc[Sort Rows]
*** xref:pipeline/transforms/sortedmerge.adoc[Sorted Merge]
+*** xref:pipeline/transforms/sortedschemamerge.adoc[Sorted Schema Merge]
*** xref:pipeline/transforms/splitfields.adoc[Split Fields]
*** xref:pipeline/transforms/splitfieldtorows.adoc[Split fields to rows]
*** xref:pipeline/transforms/splunkinput.adoc[Splunk Input]
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
index 40e598f0b1..bbb11f3fa9 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
@@ -207,6 +207,7 @@ The pages nested under this topic contain information on
how to use the transfor
* xref:pipeline/transforms/snowflakebulkloader.adoc[Snowflake Bulk Loader]
* xref:pipeline/transforms/sort.adoc[Sort Rows]
* xref:pipeline/transforms/sortedmerge.adoc[Sorted Merge]
+* xref:pipeline/transforms/sortedschemamerge.adoc[Sorted Schema Merge]
* xref:pipeline/transforms/splitfields.adoc[Split Fields]
* xref:pipeline/transforms/splitfieldtorows.adoc[Split fields to rows]
* xref:pipeline/transforms/splunkinput.adoc[Splunk Input]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/sortedschemamerge.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/sortedschemamerge.adoc
new file mode 100644
index 0000000000..bfbd09c0e2
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/sortedschemamerge.adoc
@@ -0,0 +1,63 @@
+////
+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.
+////
+:documentationPath: /pipeline/transforms/
+:language: en_US
+:description: The Sorted Schema Merge transform merges pre-sorted input
streams with different schemas into one sorted stream with a unified sparse row
layout.
+
+= image:transforms/icons/sortedschemamerge.svg[Sorted Schema Merge transform
Icon, role="image-doc-icon"] Sorted Schema Merge
+
+[%noheader,cols="3a,1a", role="table-no-borders" ]
+|===
+|
+== Description
+
+The Sorted Schema Merge transform combines the behavior of
xref:pipeline/transforms/sortedmerge.adoc[Sorted Merge] and
xref:pipeline/transforms/streamschemamerge.adoc[Stream Schema Merge]:
+
+* It merges multiple *pre-sorted* input streams into one *globally sorted*
output stream.
+* Input streams may have *different field layouts*. The transform builds a
unified output schema that is the union of all input fields.
+* Fields that are not present on a given input stream are left *null* (sparse
row layout) on rows that come from that stream.
+
+A typical use case is combining data from multiple Data Vault satellites that
share business keys and load timestamps, then using
xref:pipeline/transforms/repeatfields.adoc[Repeat Fields] to fill previously
known attribute values when building an SCD2 table.
+
+IMPORTANT: Each input stream must already be sorted on the configured sort
keys (for example with `ORDER BY` in
xref:pipeline/transforms/tableinput.adoc[Table Input], or with
xref:pipeline/transforms/sort.adoc[Sort Rows]). This transform does *not*
re-sort within each stream; it only merges already-sorted streams.
+
+|
+== Supported Engines
+[%noheader,cols="2,1a",frame=none, role="table-supported-engines"]
+!===
+!Hop Engine! image:check_mark.svg[Supported, 24]
+!Spark! image:cross.svg[Not Supported, 24]
+!Flink! image:cross.svg[Not Supported, 24]
+!Dataflow! image:cross.svg[Not Supported, 24]
+!===
+|===
+
+== Options
+
+[options="header"]
+|===
+|Option|Description
+|Transform name|Name of the transform; must be unique in a single pipeline.
+|Sort keys|The key fields used to compare rows across input streams, including
ascending or descending direction. Every sort key must exist on every input
stream.
+|Input transforms|The list of upstream transforms that feed this transform.
Use *Get* to fill the list from connected hops. At least two input streams are
required.
+|===
+
+== Notes
+
+* When the same field name appears on more than one stream with different data
types, the merged field type is coerced to `String` (same behavior as Stream
Schema Merge).
+* Rows with equal sort keys keep a stable order based on the input stream
index.
+* Safe mode row layout checks are not applied across streams with different
schemas (`excludeFromRowLayoutVerification`).
diff --git a/integration-tests/transforms/0106-sorted-schema-merge.hpl
b/integration-tests/transforms/0106-sorted-schema-merge.hpl
new file mode 100644
index 0000000000..3bc607416d
--- /dev/null
+++ b/integration-tests/transforms/0106-sorted-schema-merge.hpl
@@ -0,0 +1,355 @@
+<?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>0106-sorted-schema-merge</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Merge four pre-sorted Data Grid streams with different
schemas into one sparse sorted stream.</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/07/10 22:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/10 22:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>source-a</from>
+ <to>Sorted Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source-b</from>
+ <to>Sorted Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source-c</from>
+ <to>Sorted Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source-d</from>
+ <to>Sorted Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Sorted Schema Merge</from>
+ <to>validate</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>source-a</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>1</item>
+ <item>apple</item>
+ <item>A</item>
+ </line>
+ <line>
+ <item>5</item>
+ <item>elderberry</item>
+ <item>A</item>
+ </line>
+ <line>
+ <item>9</item>
+ <item>grape</item>
+ <item>A</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <format>0</format>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>attr_a</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>src</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>48</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source-b</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>2</item>
+ <item>banana</item>
+ <item>B</item>
+ </line>
+ <line>
+ <item>6</item>
+ <item>fig</item>
+ <item>B</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <format>0</format>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>attr_b</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>src</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source-c</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>3</item>
+ <item>cherry</item>
+ <item>C</item>
+ </line>
+ <line>
+ <item>7</item>
+ <item>guava</item>
+ <item>C</item>
+ </line>
+ <line>
+ <item>10</item>
+ <item>honeydew</item>
+ <item>C</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <format>0</format>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>attr_c</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>src</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>240</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source-d</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>4</item>
+ <item>date</item>
+ <item>Y</item>
+ <item>D</item>
+ </line>
+ <line>
+ <item>8</item>
+ <item>kiwi</item>
+ <item>N</item>
+ <item>D</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <format>0</format>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>attr_d</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>flag</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>src</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>336</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Sorted Schema Merge</name>
+ <type>SortedSchemaMerge</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <sort_keys>
+ <sort_key>
+ <name>id</name>
+ <ascending>Y</ascending>
+ </sort_key>
+ </sort_keys>
+ <inputs>
+ <input>
+ <name>source-a</name>
+ </input>
+ <input>
+ <name>source-b</name>
+ </input>
+ <input>
+ <name>source-c</name>
+ </input>
+ <input>
+ <name>source-d</name>
+ </input>
+ </inputs>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>192</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>528</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git
a/integration-tests/transforms/datasets/golden-sorted-schema-merge.csv
b/integration-tests/transforms/datasets/golden-sorted-schema-merge.csv
new file mode 100644
index 0000000000..11e920ea29
--- /dev/null
+++ b/integration-tests/transforms/datasets/golden-sorted-schema-merge.csv
@@ -0,0 +1,11 @@
+id,attr_a,attr_b,attr_c,attr_d,flag,src
+1,apple,,,,,A
+2,,banana,,,,B
+3,,,cherry,,,C
+4,,,,date,Y,D
+5,elderberry,,,,,A
+6,,fig,,,,B
+7,,,guava,,,C
+8,,,,kiwi,N,D
+9,grape,,,,,A
+10,,,honeydew,,,C
diff --git a/integration-tests/transforms/main-0106-sorted-schema-merge.hwf
b/integration-tests/transforms/main-0106-sorted-schema-merge.hwf
new file mode 100644
index 0000000000..57af5a588d
--- /dev/null
+++ b/integration-tests/transforms/main-0106-sorted-schema-merge.hwf
@@ -0,0 +1,80 @@
+<?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.
+
+-->
+<workflow>
+ <name>main-0106-sorted-schema-merge</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/07/10 22:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/10 22:00:00.000</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <doNotWaitOnFirstExecution>N</doNotWaitOnFirstExecution>
+ <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>96</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0106-sorted-schema-merge UNIT</name>
+ <description/>
+ <type>RunPipelineTests</type>
+ <attributes/>
+ <test_names>
+ <test_name>
+ <name>0106-sorted-schema-merge UNIT</name>
+ </test_name>
+ </test_names>
+ <parallel>N</parallel>
+ <xloc>304</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>0106-sorted-schema-merge UNIT</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/integration-tests/transforms/metadata/dataset/golden-sorted-schema-merge.json
b/integration-tests/transforms/metadata/dataset/golden-sorted-schema-merge.json
new file mode 100644
index 0000000000..50abf039af
--- /dev/null
+++
b/integration-tests/transforms/metadata/dataset/golden-sorted-schema-merge.json
@@ -0,0 +1,64 @@
+{
+ "base_filename": "golden-sorted-schema-merge.csv",
+ "name": "golden-sorted-schema-merge",
+ "description": "Golden sparse sorted output for Sorted Schema Merge with
four input streams",
+ "dataset_fields": [
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 5,
+ "field_precision": 0,
+ "field_format": "####0;-####0",
+ "field_name": "id"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "attr_a",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "attr_b",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "attr_c",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "attr_d",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "flag",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "src",
+ "field_format": ""
+ }
+ ],
+ "folder_name": ""
+}
diff --git
a/integration-tests/transforms/metadata/unit-test/0106-sorted-schema-merge
UNIT.json
b/integration-tests/transforms/metadata/unit-test/0106-sorted-schema-merge
UNIT.json
new file mode 100644
index 0000000000..9c36dfd586
--- /dev/null
+++ b/integration-tests/transforms/metadata/unit-test/0106-sorted-schema-merge
UNIT.json
@@ -0,0 +1,53 @@
+{
+ "database_replacements": [],
+ "autoOpening": true,
+ "description": "Validate Sorted Schema Merge produces a globally sorted
sparse union of four Data Grid streams",
+ "persist_filename": "",
+ "test_type": "UNIT_TEST",
+ "variableValues": [],
+ "basePath": "${HOP_UNIT_TESTS_FOLDER}",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "id",
+ "data_set_field": "id"
+ },
+ {
+ "transform_field": "attr_a",
+ "data_set_field": "attr_a"
+ },
+ {
+ "transform_field": "attr_b",
+ "data_set_field": "attr_b"
+ },
+ {
+ "transform_field": "attr_c",
+ "data_set_field": "attr_c"
+ },
+ {
+ "transform_field": "attr_d",
+ "data_set_field": "attr_d"
+ },
+ {
+ "transform_field": "flag",
+ "data_set_field": "flag"
+ },
+ {
+ "transform_field": "src",
+ "data_set_field": "src"
+ }
+ ],
+ "field_order": [
+ "id",
+ "src"
+ ],
+ "data_set_name": "golden-sorted-schema-merge",
+ "transform_name": "validate"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0106-sorted-schema-merge UNIT",
+ "trans_test_tweaks": [],
+ "pipeline_filename": "./0106-sorted-schema-merge.hpl"
+}
diff --git a/plugins/transforms/pom.xml b/plugins/transforms/pom.xml
index f2f3e86993..ec796c1588 100644
--- a/plugins/transforms/pom.xml
+++ b/plugins/transforms/pom.xml
@@ -138,6 +138,7 @@
<module>setvariable</module>
<module>sort</module>
<module>sortedmerge</module>
+ <module>sortedschemamerge</module>
<module>splitfieldtorows</module>
<module>splunk</module>
<module>sql</module>
diff --git a/plugins/transforms/sortedschemamerge/pom.xml
b/plugins/transforms/sortedschemamerge/pom.xml
new file mode 100644
index 0000000000..564196fea1
--- /dev/null
+++ b/plugins/transforms/sortedschemamerge/pom.xml
@@ -0,0 +1,32 @@
+<?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.
+ ~
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-plugins-transforms</artifactId>
+ <version>2.19.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>hop-transform-sortedschemamerge</artifactId>
+ <packaging>jar</packaging>
+ <name>Hop Plugins Transforms Sorted Schema Merge</name>
+
+</project>
diff --git a/plugins/transforms/sortedschemamerge/src/assembly/assembly.xml
b/plugins/transforms/sortedschemamerge/src/assembly/assembly.xml
new file mode 100644
index 0000000000..df50dedff4
--- /dev/null
+++ b/plugins/transforms/sortedschemamerge/src/assembly/assembly.xml
@@ -0,0 +1,46 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.2.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.2.0
http://maven.apache.org/xsd/assembly-2.2.0.xsd";>
+ <id>hop-transform-sortedschemamerge</id>
+ <formats>
+ <format>zip</format>
+ </formats>
+ <baseDirectory>.</baseDirectory>
+ <files>
+ <file>
+ <source>${project.basedir}/src/main/resources/version.xml</source>
+ <outputDirectory>${hop.plugin.libdir}</outputDirectory>
+ <filtered>true</filtered>
+ </file>
+ </files>
+
+ <fileSets>
+ <fileSet>
+ <directory>${project.basedir}/src/main/samples</directory>
+ <outputDirectory>config/projects/samples/</outputDirectory>
+ </fileSet>
+ </fileSets>
+
+
+ <componentDescriptors>
+
<componentDescriptor>${maven.multiModuleProjectDirectory}/assemblies/shared/hop-plugin-libs.xml</componentDescriptor>
+ </componentDescriptors>
+</assembly>
\ No newline at end of file
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMerge.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMerge.java
new file mode 100644
index 0000000000..b4e0073496
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMerge.java
@@ -0,0 +1,162 @@
+/*
+ * 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.sortedschemamerge;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+
+/**
+ * Merges multiple pre-sorted input streams with different schemas into one
globally sorted stream
+ * with a unified sparse row layout.
+ */
+public class SortedSchemaMerge extends BaseTransform<SortedSchemaMergeMeta,
SortedSchemaMergeData> {
+
+ private static final Class<?> PKG = SortedSchemaMergeMeta.class;
+
+ public SortedSchemaMerge(
+ TransformMeta transformMeta,
+ SortedSchemaMergeMeta meta,
+ SortedSchemaMergeData data,
+ int copyNr,
+ PipelineMeta pipelineMeta,
+ Pipeline pipeline) {
+ super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
+ }
+
+ @Override
+ public boolean processRow() throws HopException {
+ Object[] row = getRowSorted();
+ if (row == null) {
+ setOutputDone();
+ return false;
+ }
+
+ putRow(data.outputRowMeta, row);
+ if (checkFeedback(getLinesRead()) && isBasic()) {
+ logBasic(BaseMessages.getString(PKG, "SortedSchemaMerge.Log.LineNumber",
getLinesRead()));
+ }
+ return true;
+ }
+
+ @Override
+ public boolean init() {
+ if (!super.init()) {
+ return false;
+ }
+ if (meta.getSortKeys() == null || meta.getSortKeys().isEmpty()) {
+ logError(BaseMessages.getString(PKG,
"SortedSchemaMerge.Error.NoSortKeys"));
+ return false;
+ }
+ return true;
+ }
+
+ private synchronized Object[] getRowSorted() throws HopException {
+ if (first) {
+ first = false;
+ initializeSortedMerge();
+ }
+
+ if (data.sortedBuffer == null || data.sortedBuffer.isEmpty()) {
+ return null;
+ }
+
+ SortedSchemaMergeRow smallestRow = data.sortedBuffer.remove(0);
+ Object[] outputRow =
+ data.schemaMapping.mapRow(smallestRow.getStreamIndex(),
smallestRow.getRowData());
+
+ Object[] extraRow = getRowFrom(smallestRow.getRowSet());
+ if (extraRow != null) {
+ SortedSchemaMergeRow add =
+ new SortedSchemaMergeRow(
+ smallestRow.getStreamIndex(),
+ smallestRow.getRowSet(),
+ smallestRow.getRowSet().getRowMeta(),
+ extraRow);
+ int index = Collections.binarySearch(data.sortedBuffer, add,
data.comparator);
+ if (index < 0) {
+ data.sortedBuffer.add(-index - 1, add);
+ } else {
+ data.sortedBuffer.add(index, add);
+ }
+ }
+
+ if (getPipeline().isSafeModeEnabled()) {
+ safeModeChecking(smallestRow.getRowMeta());
+ }
+
+ return outputRow;
+ }
+
+ private void initializeSortedMerge() throws HopException {
+ List<IRowSet> inputRowSets = getInputRowSets();
+ if (inputRowSets == null || inputRowSets.size() < 2) {
+ throw new HopTransformException(
+ BaseMessages.getString(PKG, "SortedSchemaMerge.Error.TooFewInputs"));
+ }
+
+ data.sortedBuffer = new ArrayList<>();
+ List<IRowMeta> inputLayouts = new ArrayList<>();
+ for (int i = inputRowSets.size() - 1; i >= 0 && !isStopped(); i--) {
+ IRowSet rowSet = inputRowSets.get(i);
+ Object[] row = getRowFrom(rowSet);
+ if (row != null) {
+ int streamIndex = inputLayouts.size();
+ inputLayouts.add(rowSet.getRowMeta());
+ data.sortedBuffer.add(
+ new SortedSchemaMergeRow(streamIndex, rowSet, rowSet.getRowMeta(),
row));
+ }
+ }
+
+ if (data.sortedBuffer.isEmpty()) {
+ return;
+ }
+
+ IRowMeta[] inputRowMetas = inputLayouts.toArray(new IRowMeta[0]);
+ data.schemaMapping =
SortedSchemaMergeLogic.buildSchemaMapping(inputRowMetas);
+ data.outputRowMeta = data.schemaMapping.getOutputRowMeta().clone();
+ for (int i = 0; i < data.outputRowMeta.size(); i++) {
+ data.outputRowMeta.getValueMeta(i).setOrigin(getTransformName());
+ }
+
+ data.sortKeyIndices =
+ SortedSchemaMergeLogic.resolveSortKeyIndices(inputRowMetas,
meta.getSortKeys());
+ data.comparator =
+ (left, right) -> {
+ try {
+ return SortedSchemaMergeLogic.compareRows(
+ left, right, data.sortKeyIndices, meta.getSortKeys());
+ } catch (HopValueException e) {
+ throw new IllegalStateException("Error comparing sorted rows", e);
+ }
+ };
+
+ data.sortedBuffer.sort(data.comparator);
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeData.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeData.java
new file mode 100644
index 0000000000..ce5d707c0b
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeData.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.pipeline.transforms.sortedschemamerge;
+
+import java.util.Comparator;
+import java.util.List;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.pipeline.transform.BaseTransformData;
+import org.apache.hop.pipeline.transform.ITransformData;
+import
org.apache.hop.pipeline.transforms.sortedschemamerge.SortedSchemaMergeLogic.SchemaMapping;
+
+@SuppressWarnings("java:S1104")
+public class SortedSchemaMergeData extends BaseTransformData implements
ITransformData {
+
+ public SchemaMapping schemaMapping;
+ public IRowMeta outputRowMeta;
+ public int[][] sortKeyIndices;
+ public List<SortedSchemaMergeRow> sortedBuffer;
+ public Comparator<SortedSchemaMergeRow> comparator;
+
+ public SortedSchemaMergeData() {
+ super();
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeDialog.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeDialog.java
new file mode 100644
index 0000000000..da7ec299d9
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeDialog.java
@@ -0,0 +1,206 @@
+/*
+ * 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.sortedschemamerge;
+
+import org.apache.hop.core.Const;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.dialog.BaseDialog;
+import org.apache.hop.ui.core.widget.ColumnInfo;
+import org.apache.hop.ui.core.widget.TableView;
+import org.apache.hop.ui.pipeline.transform.BaseTransformDialog;
+import org.eclipse.swt.SWT;
+import org.eclipse.swt.events.ModifyListener;
+import org.eclipse.swt.layout.FormAttachment;
+import org.eclipse.swt.layout.FormData;
+import org.eclipse.swt.widgets.Label;
+import org.eclipse.swt.widgets.Shell;
+import org.eclipse.swt.widgets.Table;
+import org.eclipse.swt.widgets.TableItem;
+
+public class SortedSchemaMergeDialog extends BaseTransformDialog {
+
+ private static final Class<?> PKG = SortedSchemaMergeMeta.class;
+
+ private final SortedSchemaMergeMeta input;
+ private String[] previousTransforms;
+ private TableView wInputs;
+ private TableView wSortKeys;
+
+ public SortedSchemaMergeDialog(
+ Shell parent,
+ IVariables variables,
+ SortedSchemaMergeMeta transformMeta,
+ PipelineMeta pipelineMeta) {
+ super(parent, variables, transformMeta, pipelineMeta);
+ input = transformMeta;
+ }
+
+ @Override
+ public String open() {
+ createShell(BaseMessages.getString(PKG, "SortedSchemaMergeDialog.Title"));
+
+ buildButtonBar()
+ .ok(e -> ok())
+ .get(e -> getSourceTransformNames())
+ .cancel(e -> cancel())
+ .build();
+
+ changed = input.hasChanged();
+ ModifyListener lsMod = e -> input.setChanged();
+ previousTransforms = pipelineMeta.getPrevTransformNames(transformName);
+
+ Label wlSortKeys = new Label(shell, SWT.NONE);
+ wlSortKeys.setText(BaseMessages.getString(PKG,
"SortedSchemaMergeDialog.SortKeys.Label"));
+ PropsUi.setLook(wlSortKeys);
+ FormData fdlSortKeys = new FormData();
+ fdlSortKeys.left = new FormAttachment(0, 0);
+ fdlSortKeys.top = new FormAttachment(wSpacer, margin);
+ wlSortKeys.setLayoutData(fdlSortKeys);
+
+ ColumnInfo[] sortKeyColumns =
+ new ColumnInfo[] {
+ new ColumnInfo(
+ BaseMessages.getString(PKG,
"SortedSchemaMergeDialog.SortKeys.Column.Field"),
+ ColumnInfo.COLUMN_TYPE_TEXT,
+ false),
+ new ColumnInfo(
+ BaseMessages.getString(PKG,
"SortedSchemaMergeDialog.SortKeys.Column.Ascending"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ new String[] {"Y", "N"},
+ true)
+ };
+
+ wSortKeys =
+ new TableView(
+ variables,
+ shell,
+ SWT.BORDER | SWT.FULL_SELECTION | SWT.MULTI,
+ sortKeyColumns,
+ 1,
+ lsMod,
+ props);
+ FormData fdSortKeys = new FormData();
+ fdSortKeys.left = new FormAttachment(0, 0);
+ fdSortKeys.top = new FormAttachment(wlSortKeys, margin);
+ fdSortKeys.right = new FormAttachment(100, 0);
+ fdSortKeys.bottom = new FormAttachment(55, 0);
+ wSortKeys.setLayoutData(fdSortKeys);
+
+ Label wlNote = new Label(shell, SWT.WRAP);
+ wlNote.setText(BaseMessages.getString(PKG,
"SortedSchemaMergeDialog.SortedInputNote.Label"));
+ PropsUi.setLook(wlNote);
+ FormData fdlNote = new FormData();
+ fdlNote.left = new FormAttachment(0, 0);
+ fdlNote.right = new FormAttachment(100, 0);
+ fdlNote.bottom = new FormAttachment(wOk, -margin);
+ wlNote.setLayoutData(fdlNote);
+
+ Label wlInputs = new Label(shell, SWT.NONE);
+ wlInputs.setText(BaseMessages.getString(PKG,
"SortedSchemaMergeDialog.Inputs.Label"));
+ PropsUi.setLook(wlInputs);
+ FormData fdlInputs = new FormData();
+ fdlInputs.left = new FormAttachment(0, 0);
+ fdlInputs.top = new FormAttachment(wSortKeys, margin);
+ wlInputs.setLayoutData(fdlInputs);
+
+ ColumnInfo[] inputColumns =
+ new ColumnInfo[] {
+ new ColumnInfo(
+ BaseMessages.getString(PKG,
"SortedSchemaMergeDialog.Inputs.Column.Transform"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ previousTransforms,
+ false)
+ };
+
+ wInputs =
+ new TableView(
+ variables,
+ shell,
+ SWT.BORDER | SWT.FULL_SELECTION | SWT.MULTI,
+ inputColumns,
+ 1,
+ lsMod,
+ props);
+ FormData fdInputs = new FormData();
+ fdInputs.left = new FormAttachment(0, 0);
+ fdInputs.top = new FormAttachment(wlInputs, margin);
+ fdInputs.right = new FormAttachment(100, 0);
+ fdInputs.bottom = new FormAttachment(wlNote, -margin);
+ wInputs.setLayoutData(fdInputs);
+
+ populateDialog();
+ input.setChanged(changed);
+ focusTransformName();
+ BaseDialog.defaultShellHandling(shell, c -> ok(), c -> cancel());
+ return transformName;
+ }
+
+ private void populateDialog() {
+ for (SortedSchemaMergeSortKey sortKey : input.getSortKeys()) {
+ TableItem ti = new TableItem(wSortKeys.table, SWT.NONE);
+ ti.setText(1, Const.NVL(sortKey.getFieldName(), ""));
+ ti.setText(2, sortKey.isAscending() ? "Y" : "N");
+ }
+ wSortKeys.optimizeTableView();
+
+ for (SortedSchemaMergeInput mergeInput : input.getInputs()) {
+ TableItem ti = new TableItem(wInputs.table, SWT.NONE);
+ ti.setText(1, Const.NVL(mergeInput.getTransformName(), ""));
+ }
+ wInputs.optimizeTableView();
+ }
+
+ private void getSourceTransformNames() {
+ wInputs.removeAll();
+ Table table = wInputs.table;
+ for (String previousTransform : previousTransforms) {
+ TableItem ti = new TableItem(table, SWT.NONE);
+ ti.setText(1, previousTransform);
+ }
+ wInputs.optimizeTableView();
+ }
+
+ private void cancel() {
+ transformName = null;
+ input.setChanged(changed);
+ dispose();
+ }
+
+ private void ok() {
+ transformName = wTransformName.getText();
+
+ input.getSortKeys().clear();
+ for (TableItem item : wSortKeys.getNonEmptyItems()) {
+ SortedSchemaMergeSortKey sortKey = new SortedSchemaMergeSortKey();
+ sortKey.setFieldName(item.getText(1));
+ sortKey.setAscending(!"N".equalsIgnoreCase(item.getText(2)));
+ input.getSortKeys().add(sortKey);
+ }
+
+ input.getInputs().clear();
+ for (TableItem item : wInputs.getNonEmptyItems()) {
+ input.getInputs().add(new SortedSchemaMergeInput(item.getText(1)));
+ }
+
+ dispose();
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeInput.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeInput.java
new file mode 100644
index 0000000000..e1560ec807
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeInput.java
@@ -0,0 +1,44 @@
+/*
+ * 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.sortedschemamerge;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+
+/** Reference to an upstream transform that feeds a sorted input stream. */
+@Getter
+@Setter
+@NoArgsConstructor
+public class SortedSchemaMergeInput {
+
+ @HopMetadataProperty(key = "name")
+ private String transformName;
+
+ public SortedSchemaMergeInput(String transformName) {
+ this.transformName = transformName;
+ }
+
+ public SortedSchemaMergeInput(SortedSchemaMergeInput other) {
+ if (other != null) {
+ transformName = other.transformName;
+ }
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeLogic.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeLogic.java
new file mode 100644
index 0000000000..40d4534e89
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeLogic.java
@@ -0,0 +1,159 @@
+/*
+ * 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.sortedschemamerge;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import org.apache.hop.core.exception.HopPluginException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.value.ValueMetaFactory;
+import org.apache.hop.core.util.Utils;
+
+/** Schema union and row mapping helpers for {@link SortedSchemaMerge}. */
+public final class SortedSchemaMergeLogic {
+
+ private SortedSchemaMergeLogic() {}
+
+ /** Union of all input row layouts plus per-stream field index mappings. */
+ public static final class SchemaMapping {
+ private final IRowMeta outputRowMeta;
+ private final int[][] streamMappings;
+
+ private SchemaMapping(IRowMeta outputRowMeta, int[][] streamMappings) {
+ this.outputRowMeta = outputRowMeta;
+ this.streamMappings = streamMappings;
+ }
+
+ public IRowMeta getOutputRowMeta() {
+ return outputRowMeta;
+ }
+
+ public int[][] getStreamMappings() {
+ return streamMappings;
+ }
+
+ public Object[] mapRow(int streamIndex, Object[] sourceRow) {
+ Object[] outputRow = new Object[outputRowMeta.size()];
+ if (sourceRow == null || streamIndex < 0 || streamIndex >=
streamMappings.length) {
+ return outputRow;
+ }
+ int[] mapping = streamMappings[streamIndex];
+ for (int sourceIndex = 0; sourceIndex < mapping.length; sourceIndex++) {
+ int outputIndex = mapping[sourceIndex];
+ if (outputIndex >= 0 && outputIndex < outputRow.length) {
+ outputRow[outputIndex] = sourceRow[sourceIndex];
+ }
+ }
+ return outputRow;
+ }
+ }
+
+ public static SchemaMapping buildSchemaMapping(IRowMeta[] inputRowMetas)
+ throws HopPluginException {
+ if (inputRowMetas == null || inputRowMetas.length == 0) {
+ throw new HopPluginException("At least one input row layout is
required");
+ }
+
+ int[][] mappings = new int[inputRowMetas.length][];
+ IRowMeta outputRowMeta = inputRowMetas[0].clone();
+ HashSet<String> fieldNames = new HashSet<>();
+ Collections.addAll(fieldNames, outputRowMeta.getFieldNames());
+
+ for (int streamIndex = 0; streamIndex < inputRowMetas.length;
streamIndex++) {
+ IRowMeta inputRowMeta = inputRowMetas[streamIndex];
+ int[] streamMapping = new int[inputRowMeta.size()];
+ for (int fieldIndex = 0; fieldIndex < inputRowMeta.size(); fieldIndex++)
{
+ IValueMeta field = inputRowMeta.getValueMeta(fieldIndex);
+ String name = field.getName();
+ if (!fieldNames.contains(name)) {
+ outputRowMeta.addValueMeta(field);
+ fieldNames.add(name);
+ }
+ int outputIndex = outputRowMeta.indexOfValue(name);
+ streamMapping[fieldIndex] = outputIndex;
+ IValueMeta outputField = outputRowMeta.getValueMeta(outputIndex);
+ if (outputField.getType() != field.getType()) {
+ IValueMeta updatedField =
+ ValueMetaFactory.cloneValueMeta(outputField,
IValueMeta.TYPE_STRING);
+ outputRowMeta.setValueMeta(outputIndex, updatedField);
+ }
+ }
+ mappings[streamIndex] = streamMapping;
+ }
+
+ return new SchemaMapping(outputRowMeta, mappings);
+ }
+
+ public static int[][] resolveSortKeyIndices(
+ IRowMeta[] inputRowMetas, List<SortedSchemaMergeSortKey> sortKeys)
+ throws HopTransformException {
+ if (sortKeys == null || sortKeys.isEmpty()) {
+ throw new HopTransformException("At least one sort key is required");
+ }
+
+ int[][] indices = new int[inputRowMetas.length][];
+ for (int streamIndex = 0; streamIndex < inputRowMetas.length;
streamIndex++) {
+ IRowMeta rowMeta = inputRowMetas[streamIndex];
+ int[] streamIndices = new int[sortKeys.size()];
+ for (int keyIndex = 0; keyIndex < sortKeys.size(); keyIndex++) {
+ SortedSchemaMergeSortKey sortKey = sortKeys.get(keyIndex);
+ if (sortKey == null || Utils.isEmpty(sortKey.getFieldName())) {
+ throw new HopTransformException("Sort key name is required for all
streams");
+ }
+ int fieldIndex = rowMeta.indexOfValue(sortKey.getFieldName());
+ if (fieldIndex < 0) {
+ throw new HopTransformException(
+ "Sort key [" + sortKey.getFieldName() + "] not found in input
stream " + streamIndex);
+ }
+ streamIndices[keyIndex] = fieldIndex;
+ }
+ indices[streamIndex] = streamIndices;
+ }
+ return indices;
+ }
+
+ public static int compareRows(
+ SortedSchemaMergeRow left,
+ SortedSchemaMergeRow right,
+ int[][] sortKeyIndices,
+ List<SortedSchemaMergeSortKey> sortKeys)
+ throws HopValueException {
+ for (int keyIndex = 0; keyIndex < sortKeys.size(); keyIndex++) {
+ SortedSchemaMergeSortKey sortKey = sortKeys.get(keyIndex);
+ int leftFieldIndex = sortKeyIndices[left.getStreamIndex()][keyIndex];
+ int rightFieldIndex = sortKeyIndices[right.getStreamIndex()][keyIndex];
+ IValueMeta leftMeta = left.getRowMeta().getValueMeta(leftFieldIndex);
+ IValueMeta rightMeta = right.getRowMeta().getValueMeta(rightFieldIndex);
+ Object leftValue = left.getRowData()[leftFieldIndex];
+ Object rightValue = right.getRowData()[rightFieldIndex];
+ int compare = leftMeta.compare(leftValue, rightMeta, rightValue);
+ if (!sortKey.isAscending()) {
+ compare = -compare;
+ }
+ if (compare != 0) {
+ return compare;
+ }
+ }
+ return Integer.compare(left.getStreamIndex(), right.getStreamIndex());
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeMeta.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeMeta.java
new file mode 100644
index 0000000000..44ec30c660
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeMeta.java
@@ -0,0 +1,281 @@
+/*
+ * 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.sortedschemamerge;
+
+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.Const;
+import org.apache.hop.core.ICheckResult;
+import org.apache.hop.core.annotations.Transform;
+import org.apache.hop.core.exception.HopPluginException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.PipelineMeta.PipelineType;
+import org.apache.hop.pipeline.transform.BaseTransformMeta;
+import org.apache.hop.pipeline.transform.TransformMeta;
+
+@Transform(
+ id = "SortedSchemaMerge",
+ image = "sortedschemamerge.svg",
+ name = "i18n::SortedSchemaMerge.Name",
+ description = "i18n::SortedSchemaMerge.Description",
+ categoryDescription =
"i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Flow",
+ keywords = "i18n::SortedSchemaMerge.keyword",
+ documentationUrl = "/pipeline/transforms/sortedschemamerge.html")
+@Getter
+@Setter
+public class SortedSchemaMergeMeta
+ extends BaseTransformMeta<SortedSchemaMerge, SortedSchemaMergeData> {
+
+ private static final Class<?> PKG = SortedSchemaMergeMeta.class;
+
+ @HopMetadataProperty(key = "input", groupKey = "inputs")
+ private List<SortedSchemaMergeInput> inputs = new ArrayList<>();
+
+ @HopMetadataProperty(key = "sort_key", groupKey = "sort_keys")
+ private List<SortedSchemaMergeSortKey> sortKeys = new ArrayList<>();
+
+ public SortedSchemaMergeMeta() {}
+
+ public SortedSchemaMergeMeta(SortedSchemaMergeMeta other) {
+ if (other == null) {
+ return;
+ }
+ for (SortedSchemaMergeInput input : other.inputs) {
+ inputs.add(new SortedSchemaMergeInput(input));
+ }
+ for (SortedSchemaMergeSortKey sortKey : other.sortKeys) {
+ sortKeys.add(new SortedSchemaMergeSortKey(sortKey));
+ }
+ }
+
+ @Override
+ public SortedSchemaMergeMeta clone() {
+ return new SortedSchemaMergeMeta(this);
+ }
+
+ @Override
+ public boolean excludeFromRowLayoutVerification() {
+ return true;
+ }
+
+ @Override
+ public void getFields(
+ PipelineMeta pipelineMeta,
+ IRowMeta inputRowMeta,
+ String name,
+ IRowMeta[] info,
+ TransformMeta nextTransform,
+ IVariables variables,
+ IHopMetadataProvider metadataProvider)
+ throws HopTransformException {
+ IRowMeta[] inputLayouts = resolveInputLayouts(pipelineMeta, name, info,
variables);
+ if (inputLayouts == null || inputLayouts.length == 0) {
+ return;
+ }
+ mergeOutputFields(inputRowMeta, name, inputLayouts);
+ }
+
+ @Override
+ public void getFields(
+ IRowMeta inputRowMeta,
+ String name,
+ IRowMeta[] info,
+ TransformMeta nextTransform,
+ IVariables variables,
+ IHopMetadataProvider metadataProvider)
+ throws HopTransformException {
+ IRowMeta[] inputLayouts = resolveInputLayouts(null, name, info, variables);
+ if (inputLayouts == null || inputLayouts.length == 0) {
+ return;
+ }
+ mergeOutputFields(inputRowMeta, name, inputLayouts);
+ }
+
+ private static IRowMeta[] resolveInputLayouts(
+ PipelineMeta pipelineMeta, String transformName, IRowMeta[] info,
IVariables variables)
+ throws HopTransformException {
+ if (info != null && info.length > 0 && !isEmptyLayout(info)) {
+ return info;
+ }
+ if (pipelineMeta == null) {
+ return info;
+ }
+ TransformMeta transformMeta = pipelineMeta.findTransform(transformName);
+ if (transformMeta == null) {
+ return info;
+ }
+ String[] inputNames = pipelineMeta.getPrevTransformNames(transformMeta);
+ if (inputNames == null || inputNames.length == 0) {
+ return info;
+ }
+ List<IRowMeta> layouts = new ArrayList<>();
+ for (String inputName : inputNames) {
+ if (Utils.isEmpty(inputName)) {
+ continue;
+ }
+ IRowMeta layout = pipelineMeta.getTransformFields(variables, inputName);
+ if (layout != null && !layout.isEmpty()) {
+ layouts.add(layout);
+ }
+ }
+ return layouts.isEmpty() ? info : layouts.toArray(new IRowMeta[0]);
+ }
+
+ private static boolean isEmptyLayout(IRowMeta[] layouts) {
+ for (IRowMeta layout : layouts) {
+ if (layout != null && !layout.isEmpty()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static void mergeOutputFields(IRowMeta inputRowMeta, String name,
IRowMeta[] inputLayouts)
+ throws HopTransformException {
+ try {
+ SortedSchemaMergeLogic.SchemaMapping mapping =
+ SortedSchemaMergeLogic.buildSchemaMapping(inputLayouts);
+ IRowMeta outputRowMeta = mapping.getOutputRowMeta();
+ for (int i = 0; i < outputRowMeta.size(); i++) {
+ outputRowMeta.getValueMeta(i).setOrigin(name);
+ }
+ inputRowMeta.mergeRowMeta(outputRowMeta);
+ } catch (HopPluginException e) {
+ throw new HopTransformException("Unable to resolve merged output
fields", e);
+ }
+ }
+
+ @Override
+ public void check(
+ List<ICheckResult> remarks,
+ PipelineMeta pipelineMeta,
+ TransformMeta transformMeta,
+ IRowMeta prev,
+ String[] input,
+ String[] output,
+ IRowMeta info,
+ IVariables variables,
+ IHopMetadataProvider metadataProvider) {
+ if (inputs == null || inputs.size() < 2) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"SortedSchemaMergeMeta.CheckResult.TooFewInputs"),
+ transformMeta));
+ } else {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(PKG,
"SortedSchemaMergeMeta.CheckResult.InputCountOk"),
+ transformMeta));
+ }
+
+ if (sortKeys == null || sortKeys.isEmpty()) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"SortedSchemaMergeMeta.CheckResult.NoSortKeys"),
+ transformMeta));
+ return;
+ }
+
+ List<IRowMeta> inputLayouts = resolveCheckInputLayouts(pipelineMeta,
input, prev, variables);
+ if (!inputLayouts.isEmpty()) {
+ StringBuilder missing = new StringBuilder();
+ for (SortedSchemaMergeSortKey sortKey : sortKeys) {
+ if (sortKey == null || Utils.isEmpty(sortKey.getFieldName())) {
+ continue;
+ }
+ for (IRowMeta rowMeta : inputLayouts) {
+ if (rowMeta != null && rowMeta.indexOfValue(sortKey.getFieldName())
< 0) {
+
missing.append("\t\t").append(sortKey.getFieldName()).append(Const.CR);
+ break;
+ }
+ }
+ }
+ if (!missing.isEmpty()) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG, "SortedSchemaMergeMeta.CheckResult.SortKeysMissing",
missing.toString()),
+ transformMeta));
+ } else {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(PKG,
"SortedSchemaMergeMeta.CheckResult.SortKeysOk"),
+ transformMeta));
+ }
+ }
+
+ if (input != null && input.length > 0) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(PKG,
"SortedSchemaMergeMeta.CheckResult.ReceivingRowsOk"),
+ transformMeta));
+ } else {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"SortedSchemaMergeMeta.CheckResult.ReceivingRowsError"),
+ transformMeta));
+ }
+ }
+
+ @Override
+ public PipelineType[] getSupportedPipelineTypes() {
+ return new PipelineType[] {PipelineType.Normal};
+ }
+
+ private static List<IRowMeta> resolveCheckInputLayouts(
+ PipelineMeta pipelineMeta, String[] input, IRowMeta prev, IVariables
variables) {
+ List<IRowMeta> layouts = new ArrayList<>();
+ if (input != null) {
+ for (String inputName : input) {
+ if (Utils.isEmpty(inputName)) {
+ continue;
+ }
+ try {
+ IRowMeta layout = pipelineMeta.getTransformFields(variables,
inputName);
+ if (layout != null && !layout.isEmpty()) {
+ layouts.add(layout);
+ }
+ } catch (HopTransformException ignored) {
+ // Pipeline metadata may be incomplete while the dialog is being
edited.
+ }
+ }
+ }
+ if (layouts.isEmpty() && prev != null && !prev.isEmpty()) {
+ layouts.add(prev);
+ }
+ return layouts;
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeMetaFactory.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeMetaFactory.java
new file mode 100644
index 0000000000..b6eb6f2a0f
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeMetaFactory.java
@@ -0,0 +1,62 @@
+/*
+ * 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.sortedschemamerge;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.util.Utils;
+
+/** Builds {@link SortedSchemaMergeMeta} instances for generated pipelines. */
+public final class SortedSchemaMergeMetaFactory {
+
+ private SortedSchemaMergeMetaFactory() {}
+
+ public static SortedSchemaMergeMeta create(
+ List<String> inputTransformNames, List<SortedSchemaMergeSortKey>
sortKeys) {
+ SortedSchemaMergeMeta meta = new SortedSchemaMergeMeta();
+ if (inputTransformNames != null) {
+ for (String transformName : inputTransformNames) {
+ if (!Utils.isEmpty(transformName)) {
+ meta.getInputs().add(new SortedSchemaMergeInput(transformName));
+ }
+ }
+ }
+ if (sortKeys != null) {
+ for (SortedSchemaMergeSortKey sortKey : sortKeys) {
+ if (sortKey != null) {
+ meta.getSortKeys().add(new SortedSchemaMergeSortKey(sortKey));
+ }
+ }
+ }
+ return meta;
+ }
+
+ public static List<SortedSchemaMergeSortKey> sortKeys(String... fieldNames) {
+ List<SortedSchemaMergeSortKey> keys = new ArrayList<>();
+ if (fieldNames == null) {
+ return keys;
+ }
+ for (String fieldName : fieldNames) {
+ if (!Utils.isEmpty(fieldName)) {
+ keys.add(new SortedSchemaMergeSortKey(fieldName, true));
+ }
+ }
+ return keys;
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeRow.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeRow.java
new file mode 100644
index 0000000000..d0fdd8a19e
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeRow.java
@@ -0,0 +1,54 @@
+/*
+ * 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.sortedschemamerge;
+
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.row.IRowMeta;
+
+/** Buffered row from one sorted input stream during multi-way merge. */
+public final class SortedSchemaMergeRow {
+
+ private final int streamIndex;
+ private final IRowSet rowSet;
+ private final IRowMeta rowMeta;
+ private final Object[] rowData;
+
+ public SortedSchemaMergeRow(int streamIndex, IRowSet rowSet, IRowMeta
rowMeta, Object[] rowData) {
+ this.streamIndex = streamIndex;
+ this.rowSet = rowSet;
+ this.rowMeta = rowMeta;
+ this.rowData = rowData;
+ }
+
+ public int getStreamIndex() {
+ return streamIndex;
+ }
+
+ public IRowSet getRowSet() {
+ return rowSet;
+ }
+
+ public IRowMeta getRowMeta() {
+ return rowMeta;
+ }
+
+ public Object[] getRowData() {
+ return rowData;
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeSortKey.java
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeSortKey.java
new file mode 100644
index 0000000000..ea87958c26
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeSortKey.java
@@ -0,0 +1,49 @@
+/*
+ * 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.sortedschemamerge;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+
+/** Sort key used when merging pre-sorted input streams. */
+@Getter
+@Setter
+@NoArgsConstructor
+public class SortedSchemaMergeSortKey {
+
+ @HopMetadataProperty(key = "name")
+ private String fieldName;
+
+ @HopMetadataProperty(key = "ascending")
+ private boolean ascending = true;
+
+ public SortedSchemaMergeSortKey(String fieldName, boolean ascending) {
+ this.fieldName = fieldName;
+ this.ascending = ascending;
+ }
+
+ public SortedSchemaMergeSortKey(SortedSchemaMergeSortKey other) {
+ if (other != null) {
+ fieldName = other.fieldName;
+ ascending = other.ascending;
+ }
+ }
+}
diff --git
a/plugins/transforms/sortedschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedschemamerge/messages/messages_en_US.properties
b/plugins/transforms/sortedschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedschemamerge/messages/messages_en_US.properties
new file mode 100644
index 0000000000..84cd66c0be
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedschemamerge/messages/messages_en_US.properties
@@ -0,0 +1,44 @@
+#
+#
+# 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.
+#
+#
+
+SortedSchemaMerge.Name = Sorted Schema Merge
+SortedSchemaMerge.Description = Merge pre-sorted input streams with different
schemas into one sorted sparse stream
+SortedSchemaMerge.keyword = sorted schema merge stream union sparse
+
+SortedSchemaMerge.Log.LineNumber = Line number {0}
+SortedSchemaMerge.Error.TooFewInputs = At least two sorted input streams are
required
+SortedSchemaMerge.Error.NoSortKeys = At least one sort key is required
+SortedSchemaMerge.Error.NoInputRowSets = No input row sets were found
+
+SortedSchemaMergeDialog.Title = Sorted Schema Merge
+SortedSchemaMergeDialog.TransformName.Label = Transform name
+SortedSchemaMergeDialog.Inputs.Label = Input transforms
+SortedSchemaMergeDialog.Inputs.Column.Transform = Transform
+SortedSchemaMergeDialog.SortKeys.Label = Sort keys
+SortedSchemaMergeDialog.SortKeys.Column.Field = Field
+SortedSchemaMergeDialog.SortKeys.Column.Ascending = Ascending
+SortedSchemaMergeDialog.SortedInputNote.Label = Each input stream must already
be sorted on the sort keys (for example via ORDER BY in Table Input).
+
+SortedSchemaMergeMeta.CheckResult.TooFewInputs = Configure at least two input
transforms
+SortedSchemaMergeMeta.CheckResult.InputCountOk = At least two input transforms
are configured
+SortedSchemaMergeMeta.CheckResult.NoSortKeys = No sort keys are defined
+SortedSchemaMergeMeta.CheckResult.SortKeysMissing = The following sort keys
were not found in the input streams:{0}
+SortedSchemaMergeMeta.CheckResult.SortKeysOk = All sort keys were found in the
input streams
+SortedSchemaMergeMeta.CheckResult.ReceivingRowsOk = Receiving rows from
upstream transforms
+SortedSchemaMergeMeta.CheckResult.ReceivingRowsError = Not receiving rows from
upstream transforms. Connect at least two sorted input hops.
diff --git
a/plugins/transforms/sortedschemamerge/src/main/resources/sortedschemamerge.svg
b/plugins/transforms/sortedschemamerge/src/main/resources/sortedschemamerge.svg
new file mode 100644
index 0000000000..9beaf28777
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/main/resources/sortedschemamerge.svg
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN"
"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd";>
+<svg version="1.1" xmlns="http://www.w3.org/2000/svg"; x="0px" y="0px"
+ width="42px" height="42px" viewBox="0 0 42 42" enable-background="new 0 0
42 42">
+ <!-- Two source schemas (different column layouts) -->
+ <path fill="#0E3A5A" d="M18.5,3.5H2.5v12h16V3.5z M16.5,13.5h-3v-8h3V13.5z
M12.5,13.5h-3v-8h3V13.5z M8.5,13.5h-4v-8h4V13.5z"/>
+ <rect x="4.5" y="5.5" fill="#C9E8FB" width="4" height="8"/>
+ <rect x="9.5" y="5.5" fill="#C9E8FB" width="3" height="8"/>
+ <path fill="#0E3A5A" d="M18.5,26.5H2.5v12h16V26.5z M16.5,36.5h-4v-8h4V36.5z
M11.5,36.5h-3v-8h3V36.5z M7.5,36.5h-3v-8h3V36.5z"/>
+ <rect x="4.5" y="28.5" fill="#C9E8FB" width="3" height="8"/>
+ <rect x="8.5" y="28.5" fill="#C9E8FB" width="3" height="8"/>
+ <rect x="12.5" y="28.5" fill="#C9E8FB" width="4" height="8"/>
+ <!-- Merge funnel into sparse sorted output -->
+ <path fill="#0E3A5A" d="M20,9.5h6.5v-3h8v3H39v7.5h-4.5v3H25.5v-3H20V9.5z"/>
+ <rect x="22" y="11.5" fill="#FF9C04" width="5" height="1.5"/>
+ <rect x="22" y="13.7" fill="#FF9C04" width="3.5" height="1.5"/>
+ <path fill="#5CC0C4"
d="M20,32.5h6.5v3h8v-3H39v-7.5h-4.5v-3H25.5v3H20V32.5z"/>
+ <!-- Output sparse row bar -->
+ <path fill="#0E3A5A" d="M39.5,18.5h-14v5h14V18.5z"/>
+ <rect x="26.5" y="19.7" fill="#FFFFFF" width="3" height="2.6"/>
+ <rect x="30.5" y="19.7" fill="#FF9C04" width="3" height="2.6"/>
+ <rect x="34.5" y="19.7" fill="#C9E8FB" width="3.5" height="2.6"/>
+</svg>
diff --git
a/plugins/transforms/sortedschemamerge/src/main/resources/version.xml
b/plugins/transforms/sortedschemamerge/src/main/resources/version.xml
new file mode 100644
index 0000000000..6be576acae
--- /dev/null
+++ b/plugins/transforms/sortedschemamerge/src/main/resources/version.xml
@@ -0,0 +1,20 @@
+<?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.
+ ~
+ -->
+
+<version>${project.version}</version>
\ No newline at end of file
diff --git
a/plugins/transforms/sortedschemamerge/src/test/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeLogicTest.java
b/plugins/transforms/sortedschemamerge/src/test/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeLogicTest.java
new file mode 100644
index 0000000000..9a1305bb15
--- /dev/null
+++
b/plugins/transforms/sortedschemamerge/src/test/java/org/apache/hop/pipeline/transforms/sortedschemamerge/SortedSchemaMergeLogicTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.sortedschemamerge;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopPluginException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class SortedSchemaMergeLogicTest {
+
+ @BeforeAll
+ static void initHop() throws HopException {
+ HopEnvironment.init();
+ }
+
+ @Test
+ void buildSchemaMappingUnionsFieldsAndMapsSparseRows() throws
HopPluginException {
+ IRowMeta streamA = new RowMeta();
+ streamA.addValueMeta(new ValueMetaString("hub_hk"));
+ streamA.addValueMeta(new ValueMetaString("attr_a"));
+
+ IRowMeta streamB = new RowMeta();
+ streamB.addValueMeta(new ValueMetaString("hub_hk"));
+ streamB.addValueMeta(new ValueMetaString("attr_b"));
+
+ SortedSchemaMergeLogic.SchemaMapping mapping =
+ SortedSchemaMergeLogic.buildSchemaMapping(new IRowMeta[] {streamA,
streamB});
+
+ assertEquals(3, mapping.getOutputRowMeta().size());
+ assertTrue(mapping.getOutputRowMeta().indexOfValue("hub_hk") >= 0);
+ assertTrue(mapping.getOutputRowMeta().indexOfValue("attr_a") >= 0);
+ assertTrue(mapping.getOutputRowMeta().indexOfValue("attr_b") >= 0);
+
+ Object[] mappedA = mapping.mapRow(0, new Object[] {"hk-1", "value-a"});
+ Object[] mappedB = mapping.mapRow(1, new Object[] {"hk-1", "value-b"});
+
+ assertEquals("hk-1",
mappedA[mapping.getOutputRowMeta().indexOfValue("hub_hk")]);
+ assertEquals("value-a",
mappedA[mapping.getOutputRowMeta().indexOfValue("attr_a")]);
+ assertEquals(null,
mappedA[mapping.getOutputRowMeta().indexOfValue("attr_b")]);
+
+ assertEquals("hk-1",
mappedB[mapping.getOutputRowMeta().indexOfValue("hub_hk")]);
+ assertEquals(null,
mappedB[mapping.getOutputRowMeta().indexOfValue("attr_a")]);
+ assertEquals("value-b",
mappedB[mapping.getOutputRowMeta().indexOfValue("attr_b")]);
+ }
+
+ @Test
+ void buildSchemaMappingCoercesConflictingTypesToString() throws
HopPluginException {
+ IRowMeta streamA = new RowMeta();
+ streamA.addValueMeta(new ValueMetaString("shared"));
+
+ IRowMeta streamB = new RowMeta();
+ streamB.addValueMeta(new ValueMetaInteger("shared"));
+
+ SortedSchemaMergeLogic.SchemaMapping mapping =
+ SortedSchemaMergeLogic.buildSchemaMapping(new IRowMeta[] {streamA,
streamB});
+
+ assertEquals(
+ ValueMetaString.TYPE_STRING,
+ mapping
+ .getOutputRowMeta()
+ .getValueMeta(mapping.getOutputRowMeta().indexOfValue("shared"))
+ .getType());
+ }
+
+ @Test
+ void resolveSortKeyIndicesRequiresFieldOnEveryStream() {
+ IRowMeta streamA = row("hub_hk", "load_dts");
+ IRowMeta streamB = row("hub_hk");
+
+ List<SortedSchemaMergeSortKey> sortKeys =
+ SortedSchemaMergeMetaFactory.sortKeys("hub_hk", "load_dts");
+
+ assertThrows(
+ HopTransformException.class,
+ () ->
+ SortedSchemaMergeLogic.resolveSortKeyIndices(
+ new IRowMeta[] {streamA, streamB}, sortKeys));
+ }
+
+ @Test
+ void compareRowsUsesSortKeysAndStreamIndexTieBreak() throws Exception {
+ IRowMeta streamA = row("hub_hk", "load_dts");
+ IRowMeta streamB = row("hub_hk", "load_dts");
+
+ List<SortedSchemaMergeSortKey> sortKeys =
+ SortedSchemaMergeMetaFactory.sortKeys("hub_hk", "load_dts");
+ int[][] sortKeyIndices =
+ SortedSchemaMergeLogic.resolveSortKeyIndices(new IRowMeta[] {streamA,
streamB}, sortKeys);
+
+ SortedSchemaMergeRow left =
+ new SortedSchemaMergeRow(0, null, streamA, new Object[] {"hk-a",
"2024-01-01"});
+ SortedSchemaMergeRow right =
+ new SortedSchemaMergeRow(1, null, streamB, new Object[] {"hk-b",
"2024-01-02"});
+
+ assertTrue(SortedSchemaMergeLogic.compareRows(left, right, sortKeyIndices,
sortKeys) < 0);
+
+ SortedSchemaMergeRow tieLeft =
+ new SortedSchemaMergeRow(0, null, streamA, new Object[] {"hk-same",
"2024-01-01"});
+ SortedSchemaMergeRow tieRight =
+ new SortedSchemaMergeRow(1, null, streamB, new Object[] {"hk-same",
"2024-01-01"});
+
+ assertTrue(SortedSchemaMergeLogic.compareRows(tieLeft, tieRight,
sortKeyIndices, sortKeys) < 0);
+ }
+
+ @Test
+ void compareRowsHonorsDescendingSortKey() throws Exception {
+ IRowMeta layout = row("seq");
+
+ List<SortedSchemaMergeSortKey> sortKeys = List.of(new
SortedSchemaMergeSortKey("seq", false));
+ int[][] sortKeyIndices =
+ SortedSchemaMergeLogic.resolveSortKeyIndices(new IRowMeta[] {layout},
sortKeys);
+
+ SortedSchemaMergeRow low = new SortedSchemaMergeRow(0, null, layout, new
Object[] {1L});
+ SortedSchemaMergeRow high = new SortedSchemaMergeRow(0, null, layout, new
Object[] {2L});
+
+ assertTrue(SortedSchemaMergeLogic.compareRows(high, low, sortKeyIndices,
sortKeys) < 0);
+ }
+
+ private static IRowMeta row(String... names) {
+ IRowMeta rowMeta = new RowMeta();
+ for (String name : names) {
+ if ("seq".equals(name)) {
+ rowMeta.addValueMeta(new ValueMetaInteger(name));
+ } else {
+ rowMeta.addValueMeta(new ValueMetaString(name));
+ }
+ }
+ return rowMeta;
+ }
+}
