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 642b6d88a1 improve error handling for oracle on table output and SAM,
fixes #5758 (#8076)
642b6d88a1 is described below
commit 642b6d88a11ff3fa3bf0a1816b046c154ce0ee01
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Tue Aug 25 08:51:20 2026 +0200
improve error handling for oracle on table output and SAM, fixes #5758
(#8076)
---
.../oracle/0006-batch-error-accounting.hpl | 256 +++++++++++++
.../oracle/0007-sync-batch-error-accounting.hpl | 265 ++++++++++++++
.../oracle/main-0006-batch-error-accounting.hwf | 247 +++++++++++++
.../main-0007-sync-batch-error-accounting.hwf | 230 ++++++++++++
.../SynchronizeAfterMerge.java | 105 ++++--
.../messages/messages_en_US.properties | 1 +
.../SynchronizeAfterMergeBatchErrorTest.java | 307 ++++++++++++++++
.../transforms/tableoutput/TableOutput.java | 210 +++++++++--
.../transforms/tableoutput/TableOutputData.java | 10 +
.../tableoutput/messages/messages_en_US.properties | 1 +
.../tableoutput/TableOutputBatchErrorTest.java | 405 +++++++++++++++++++++
11 files changed, 1989 insertions(+), 48 deletions(-)
diff --git a/integration-tests/oracle/0006-batch-error-accounting.hpl
b/integration-tests/oracle/0006-batch-error-accounting.hpl
new file mode 100644
index 0000000000..3de6aa5fd3
--- /dev/null
+++ b/integration-tests/oracle/0006-batch-error-accounting.hpl
@@ -0,0 +1,256 @@
+<?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-batch-error-accounting</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Ten rows into Oracle in one batch, two of them violating the
primary key, with batch mode and transform error handling both on. Oracle stops
at the first failing statement, so the update counts it hands back cover only
the rows before it; the rest were never sent. Every input row still has to come
out on one of the two streams.</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/08/24 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/24 12:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Ten rows, two duplicates</from>
+ <to>Batch out</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Batch out</from>
+ <to>Rejects</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Batch out</from>
+ <to>Written</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Ten rows, two duplicates</name>
+ <type>DataGrid</type>
+ <description>Rows at position 3 and 6 reuse the IDs seeded by the
workflow, so they violate the primary key. A duplicate key is the one failure
every database enforces.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line><item>0</item><item>r0</item></line>
+ <line><item>1</item><item>r1</item></line>
+ <line><item>2</item><item>r2</item></line>
+ <line><item>9000</item><item>r3</item></line>
+ <line><item>4</item><item>r4</item></line>
+ <line><item>5</item><item>r5</item></line>
+ <line><item>9001</item><item>r6</item></line>
+ <line><item>7</item><item>r7</item></line>
+ <line><item>8</item><item>r8</item></line>
+ <line><item>9</item><item>r9</item></line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>ID</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>NAME</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>112</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Batch out</name>
+ <type>TableOutput</type>
+ <description>Commit size 10 with ten input rows puts the whole set in a
single batch, so one executeBatch call carries both bad rows.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <commit>10</commit>
+ <connection>oracle-service-name</connection>
+ <fields>
+ <field>
+ <column_name>ID</column_name>
+ <stream_name>ID</stream_name>
+ </field>
+ <field>
+ <column_name>NAME</column_name>
+ <stream_name>NAME</stream_name>
+ </field>
+ </fields>
+ <ignore_errors>N</ignore_errors>
+ <only_when_have_rows>N</only_when_have_rows>
+ <partitioning_daily>N</partitioning_daily>
+ <partitioning_enabled>N</partitioning_enabled>
+ <partitioning_field/>
+ <partitioning_monthly>Y</partitioning_monthly>
+ <return_field/>
+ <return_keys>N</return_keys>
+ <schema/>
+ <specify_fields>Y</specify_fields>
+ <table>HOP_BATCH_ERR_TARGET</table>
+ <tablename_field/>
+ <tablename_in_field>N</tablename_in_field>
+ <tablename_in_table>Y</tablename_in_table>
+ <truncate>N</truncate>
+ <use_batch>Y</use_batch>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Written</name>
+ <type>TableOutput</type>
+ <description>The rows Table Output reported as written. Kept so the check
can compare what the transform claimed against what Oracle actually
holds.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <commit>1000</commit>
+ <connection>oracle-service-name</connection>
+ <fields>
+ <field>
+ <column_name>ID</column_name>
+ <stream_name>ID</stream_name>
+ </field>
+ <field>
+ <column_name>NAME</column_name>
+ <stream_name>NAME</stream_name>
+ </field>
+ </fields>
+ <ignore_errors>N</ignore_errors>
+ <only_when_have_rows>N</only_when_have_rows>
+ <partitioning_daily>N</partitioning_daily>
+ <partitioning_enabled>N</partitioning_enabled>
+ <partitioning_field/>
+ <partitioning_monthly>Y</partitioning_monthly>
+ <return_field/>
+ <return_keys>N</return_keys>
+ <schema/>
+ <specify_fields>Y</specify_fields>
+ <table>HOP_BATCH_ERR_WRITTEN</table>
+ <tablename_field/>
+ <tablename_in_field>N</tablename_in_field>
+ <tablename_in_table>Y</tablename_in_table>
+ <truncate>N</truncate>
+ <use_batch>N</use_batch>
+ <attributes/>
+ <GUI>
+ <xloc>544</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Rejects</name>
+ <type>TableOutput</type>
+ <description>The error stream. Before the fix for issue 5758 this stayed
empty on Oracle while seven rows disappeared entirely.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <commit>1000</commit>
+ <connection>oracle-service-name</connection>
+ <fields>
+ <field>
+ <column_name>ID</column_name>
+ <stream_name>ID</stream_name>
+ </field>
+ <field>
+ <column_name>NAME</column_name>
+ <stream_name>NAME</stream_name>
+ </field>
+ <field>
+ <column_name>ERR_DESC</column_name>
+ <stream_name>ERR_DESC</stream_name>
+ </field>
+ </fields>
+ <ignore_errors>N</ignore_errors>
+ <only_when_have_rows>N</only_when_have_rows>
+ <partitioning_daily>N</partitioning_daily>
+ <partitioning_enabled>N</partitioning_enabled>
+ <partitioning_field/>
+ <partitioning_monthly>Y</partitioning_monthly>
+ <return_field/>
+ <return_keys>N</return_keys>
+ <schema/>
+ <specify_fields>Y</specify_fields>
+ <table>HOP_BATCH_ERR_REJECTS</table>
+ <tablename_field/>
+ <tablename_in_field>N</tablename_in_field>
+ <tablename_in_table>Y</tablename_in_table>
+ <truncate>N</truncate>
+ <use_batch>N</use_batch>
+ <attributes/>
+ <GUI>
+ <xloc>544</xloc>
+ <yloc>208</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ <error>
+ <source_transform>Batch out</source_transform>
+ <target_transform>Rejects</target_transform>
+ <is_enabled>Y</is_enabled>
+ <nr_valuename>ERR_COUNT</nr_valuename>
+ <descriptions_valuename>ERR_DESC</descriptions_valuename>
+ <fields_valuename>ERR_FIELDS</fields_valuename>
+ <codes_valuename>ERR_CODES</codes_valuename>
+ <max_errors/>
+ <max_pct_errors/>
+ <min_pct_rows/>
+ </error>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/oracle/0007-sync-batch-error-accounting.hpl
b/integration-tests/oracle/0007-sync-batch-error-accounting.hpl
new file mode 100644
index 0000000000..5089797df0
--- /dev/null
+++ b/integration-tests/oracle/0007-sync-batch-error-accounting.hpl
@@ -0,0 +1,265 @@
+<?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>0007-sync-batch-error-accounting</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Ten inserts through Synchronize after merge in one batch, two
of them violating the primary key, with batch mode and transform error handling
both on. Oracle stops at the first failing statement, so the update counts
cover only the rows before it. Every input row still has to come out on one of
the two streams.</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/08/24 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/24 12:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Ten inserts, two duplicates</from>
+ <to>Sync out</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Sync out</from>
+ <to>Rejects</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Sync out</from>
+ <to>Written</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Ten inserts, two duplicates</name>
+ <type>DataGrid</type>
+ <description>Rows at position 3 and 6 reuse the IDs seeded by the
workflow, so they violate the primary key.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line><item>0</item><item>r0</item><item>new</item></line>
+ <line><item>1</item><item>r1</item><item>new</item></line>
+ <line><item>2</item><item>r2</item><item>new</item></line>
+ <line><item>9000</item><item>r3</item><item>new</item></line>
+ <line><item>4</item><item>r4</item><item>new</item></line>
+ <line><item>5</item><item>r5</item><item>new</item></line>
+ <line><item>9001</item><item>r6</item><item>new</item></line>
+ <line><item>7</item><item>r7</item><item>new</item></line>
+ <line><item>8</item><item>r8</item><item>new</item></line>
+ <line><item>9</item><item>r9</item><item>new</item></line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>ID</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>NAME</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>OP</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>112</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Sync out</name>
+ <type>SynchronizeAfterMerge</type>
+ <description>Commit size 10 with ten input rows puts the whole set in a
single batch, so one executeBatch call carries both bad rows.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <connection>oracle-service-name</connection>
+ <commit>10</commit>
+ <tablename_in_field>N</tablename_in_field>
+ <tablename_field/>
+ <use_batch>Y</use_batch>
+ <perform_lookup>N</perform_lookup>
+ <operation_order_field>OP</operation_order_field>
+ <order_insert>new</order_insert>
+ <order_update>changed</order_update>
+ <order_delete>deleted</order_delete>
+ <lookup>
+ <schema/>
+ <table>HOP_SYNC_ERR_TARGET</table>
+ <key>
+ <name>ID</name>
+ <field>ID</field>
+ <condition>=</condition>
+ <name2/>
+ </key>
+ <value>
+ <name>ID</name>
+ <rename>ID</rename>
+ <update>Y</update>
+ </value>
+ <value>
+ <name>NAME</name>
+ <rename>NAME</rename>
+ <update>Y</update>
+ </value>
+ </lookup>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>144</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Written</name>
+ <type>TableOutput</type>
+ <description>The rows Synchronize after merge reported as written, so the
check can compare what the transform claimed against what Oracle actually
holds.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <commit>1000</commit>
+ <connection>oracle-service-name</connection>
+ <fields>
+ <field>
+ <column_name>ID</column_name>
+ <stream_name>ID</stream_name>
+ </field>
+ <field>
+ <column_name>NAME</column_name>
+ <stream_name>NAME</stream_name>
+ </field>
+ </fields>
+ <ignore_errors>N</ignore_errors>
+ <only_when_have_rows>N</only_when_have_rows>
+ <partitioning_daily>N</partitioning_daily>
+ <partitioning_enabled>N</partitioning_enabled>
+ <partitioning_field/>
+ <partitioning_monthly>Y</partitioning_monthly>
+ <return_field/>
+ <return_keys>N</return_keys>
+ <schema/>
+ <specify_fields>Y</specify_fields>
+ <table>HOP_SYNC_ERR_WRITTEN</table>
+ <tablename_field/>
+ <tablename_in_field>N</tablename_in_field>
+ <tablename_in_table>Y</tablename_in_table>
+ <truncate>N</truncate>
+ <use_batch>N</use_batch>
+ <attributes/>
+ <GUI>
+ <xloc>544</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Rejects</name>
+ <type>TableOutput</type>
+ <description>The error stream. Before the fix this stayed empty on Oracle
while seven rows disappeared entirely.</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <commit>1000</commit>
+ <connection>oracle-service-name</connection>
+ <fields>
+ <field>
+ <column_name>ID</column_name>
+ <stream_name>ID</stream_name>
+ </field>
+ <field>
+ <column_name>NAME</column_name>
+ <stream_name>NAME</stream_name>
+ </field>
+ <field>
+ <column_name>ERR_DESC</column_name>
+ <stream_name>ERR_DESC</stream_name>
+ </field>
+ </fields>
+ <ignore_errors>N</ignore_errors>
+ <only_when_have_rows>N</only_when_have_rows>
+ <partitioning_daily>N</partitioning_daily>
+ <partitioning_enabled>N</partitioning_enabled>
+ <partitioning_field/>
+ <partitioning_monthly>Y</partitioning_monthly>
+ <return_field/>
+ <return_keys>N</return_keys>
+ <schema/>
+ <specify_fields>Y</specify_fields>
+ <table>HOP_SYNC_ERR_REJECTS</table>
+ <tablename_field/>
+ <tablename_in_field>N</tablename_in_field>
+ <tablename_in_table>Y</tablename_in_table>
+ <truncate>N</truncate>
+ <use_batch>N</use_batch>
+ <attributes/>
+ <GUI>
+ <xloc>544</xloc>
+ <yloc>208</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ <error>
+ <source_transform>Sync out</source_transform>
+ <target_transform>Rejects</target_transform>
+ <is_enabled>Y</is_enabled>
+ <nr_valuename>ERR_COUNT</nr_valuename>
+ <descriptions_valuename>ERR_DESC</descriptions_valuename>
+ <fields_valuename>ERR_FIELDS</fields_valuename>
+ <codes_valuename>ERR_CODES</codes_valuename>
+ <max_errors/>
+ <max_pct_errors/>
+ <min_pct_rows/>
+ </error>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/oracle/main-0006-batch-error-accounting.hwf
b/integration-tests/oracle/main-0006-batch-error-accounting.hwf
new file mode 100644
index 0000000000..ca0565362b
--- /dev/null
+++ b/integration-tests/oracle/main-0006-batch-error-accounting.hwf
@@ -0,0 +1,247 @@
+<?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-0006-batch-error-accounting</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Regression test for issue 5758. Oracle stops a batch at the
first failing statement and returns update counts only for the rows before it,
so the rows behind the failure were never sent to the database. Table Output
used to walk only the counts array and then clear its buffer, which dropped
those rows silently: no output row, no error row, no log line, and a green
pipeline. This asserts that every input row is accounted for on exactly one
stream, and that the good rows beh [...]
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/08/24 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/24 12:00:00.000</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>50</xloc>
+ <yloc>50</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Init tables</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <sql>DECLARE
+ table_missing EXCEPTION;
+ PRAGMA EXCEPTION_INIT(table_missing, -942);
+
+ PROCEDURE drop_if_present(table_name VARCHAR2) IS
+ BEGIN
+ EXECUTE IMMEDIATE 'DROP TABLE ' || table_name;
+ EXCEPTION
+ WHEN table_missing THEN NULL;
+ END;
+BEGIN
+ drop_if_present('HOP_BATCH_ERR_TARGET');
+ drop_if_present('HOP_BATCH_ERR_REJECTS');
+ drop_if_present('HOP_BATCH_ERR_WRITTEN');
+
+ EXECUTE IMMEDIATE 'CREATE TABLE HOP_BATCH_ERR_TARGET (
+ ID NUMBER(10) PRIMARY KEY,
+ NAME VARCHAR2(10))';
+
+ EXECUTE IMMEDIATE 'CREATE TABLE HOP_BATCH_ERR_WRITTEN (
+ ID NUMBER(10),
+ NAME VARCHAR2(10))';
+
+ EXECUTE IMMEDIATE 'CREATE TABLE HOP_BATCH_ERR_REJECTS (
+ ID NUMBER(10),
+ NAME VARCHAR2(10),
+ ERR_DESC VARCHAR2(2000))';
+
+ -- The rows the batch will collide with. Seeded and committed up front so
the collision is with
+ -- existing data and does not depend on how the driver orders the statements
within the batch.
+ EXECUTE IMMEDIATE 'INSERT INTO HOP_BATCH_ERR_TARGET VALUES (9000, ''seed'')';
+ EXECUTE IMMEDIATE 'INSERT INTO HOP_BATCH_ERR_TARGET VALUES (9001, ''seed'')';
+ COMMIT;
+END;</sql>
+ <useVariableSubstitution>F</useVariableSubstitution>
+ <sqlfromfile>F</sqlfromfile>
+ <sqlfilename/>
+ <sendOneStatement>T</sendOneStatement>
+ <connection>oracle-service-name</connection>
+ <parallel>N</parallel>
+ <xloc>224</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0006-batch-error-accounting.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+ <filename>${PROJECT_HOME}/0006-batch-error-accounting.hpl</filename>
+ <logext/>
+ <logfile/>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>432</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>no row went missing</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <sql>DECLARE
+ written NUMBER;
+ rejected NUMBER;
+ in_table NUMBER;
+ untried NUMBER;
+BEGIN
+ SELECT COUNT(*) INTO written FROM HOP_BATCH_ERR_WRITTEN;
+ SELECT COUNT(*) INTO rejected FROM HOP_BATCH_ERR_REJECTS;
+
+ -- The whole point of issue 5758. Ten rows went in; ten rows have to come
out somewhere. Before
+ -- the fix this was 3 + 0 on Oracle, with seven rows gone without a trace.
+ IF written + rejected <> 10 THEN
+ RAISE_APPLICATION_ERROR(-20060,
+ 'rows went missing: ' || written || ' written + ' || rejected ||
+ ' rejected, expected 10 in total');
+ END IF;
+
+ -- Only the two rows Oracle actually refused are rejects. Oracle abandons
the rest of the batch at
+ -- the first failure, so the six good rows behind it were never sent and
have to be driven again -
+ -- otherwise this pipeline would write eight rows on PostgreSQL and three on
Oracle.
+ IF written <> 8 THEN
+ RAISE_APPLICATION_ERROR(-20061,
+ 'expected the 8 good rows to be written, got ' || written);
+ END IF;
+
+ -- What the transform claimed matches what Oracle actually holds: the 2 seed
rows plus those 8.
+ SELECT COUNT(*) INTO in_table FROM HOP_BATCH_ERR_TARGET;
+ IF in_table <> 10 THEN
+ RAISE_APPLICATION_ERROR(-20062,
+ 'the target table holds ' || in_table || ' rows, expected 2 seeded + 8
written');
+ END IF;
+
+ -- No row was reported as written while it is absent from the table, and
none is present while
+ -- being reported as a reject.
+ SELECT COUNT(*) INTO in_table
+ FROM HOP_BATCH_ERR_WRITTEN w
+ WHERE NOT EXISTS (SELECT 1 FROM HOP_BATCH_ERR_TARGET t WHERE t.ID = w.ID);
+ IF in_table <> 0 THEN
+ RAISE_APPLICATION_ERROR(-20063,
+ in_table || ' rows were reported as written but are not in the table');
+ END IF;
+
+ SELECT COUNT(*) INTO in_table
+ FROM HOP_BATCH_ERR_REJECTS r
+ WHERE EXISTS (SELECT 1 FROM HOP_BATCH_ERR_TARGET t WHERE t.ID = r.ID AND
t.NAME = r.NAME);
+ IF in_table <> 0 THEN
+ RAISE_APPLICATION_ERROR(-20064,
+ in_table || ' rows were rejected but are in the table anyway');
+ END IF;
+
+ -- Every row is written exactly once. A retry that re-sent a row the
database had already applied
+ -- would show up here as a duplicate.
+ SELECT COUNT(*) INTO in_table FROM (
+ SELECT ID FROM HOP_BATCH_ERR_WRITTEN GROUP BY ID HAVING COUNT(*) > 1);
+ IF in_table <> 0 THEN
+ RAISE_APPLICATION_ERROR(-20065,
+ in_table || ' rows were written more than once by the batch retry');
+ END IF;
+
+ -- Both rejects are rows Oracle itself refused, so they carry its message.
Nothing is left saying
+ -- it was never attempted: those rows were retried instead of being reported.
+ SELECT COUNT(*) INTO untried
+ FROM HOP_BATCH_ERR_REJECTS
+ WHERE ERR_DESC LIKE '%not sent to the database%';
+ IF untried <> 0 THEN
+ RAISE_APPLICATION_ERROR(-20066,
+ untried || ' rows were reported as never attempted instead of being
retried');
+ END IF;
+
+ SELECT COUNT(*) INTO untried
+ FROM HOP_BATCH_ERR_REJECTS
+ WHERE ID IN (9000, 9001) AND ERR_DESC LIKE '%ORA-00001%';
+ IF untried <> 2 THEN
+ RAISE_APPLICATION_ERROR(-20067,
+ 'both rejected rows should carry the ORA-00001 unique constraint
message, got ' || untried);
+ END IF;
+END;</sql>
+ <useVariableSubstitution>F</useVariableSubstitution>
+ <sqlfromfile>F</sqlfromfile>
+ <sqlfilename/>
+ <sendOneStatement>T</sendOneStatement>
+ <connection>oracle-service-name</connection>
+ <parallel>N</parallel>
+ <xloc>688</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>Init tables</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>Init tables</from>
+ <to>0006-batch-error-accounting.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0006-batch-error-accounting.hpl</from>
+ <to>no row went missing</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git a/integration-tests/oracle/main-0007-sync-batch-error-accounting.hwf
b/integration-tests/oracle/main-0007-sync-batch-error-accounting.hwf
new file mode 100644
index 0000000000..6f524819eb
--- /dev/null
+++ b/integration-tests/oracle/main-0007-sync-batch-error-accounting.hwf
@@ -0,0 +1,230 @@
+<?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-0007-sync-batch-error-accounting</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Synchronize after merge carries the same batch-error handling
Table Output does, and carried the same defect as issue 5758: against a driver
that abandons the rest of the batch at the first failure, the rows behind it
reached neither stream. This guards that every row is accounted for.
+
+Note the difference from main-0006: this transform buffers rows for its
insert, update and delete statements in one interleaved list, so the rows the
database never attempted cannot be re-bound and driven again. They are reported
as rejects instead. That is why the expected counts here are 3 written and 7
rejected where Table Output manages 8 and 2 - it is the current limitation, not
the desired end state.</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/08/24 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/24 12:00:00.000</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>50</xloc>
+ <yloc>50</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Init tables</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <sql>DECLARE
+ table_missing EXCEPTION;
+ PRAGMA EXCEPTION_INIT(table_missing, -942);
+
+ PROCEDURE drop_if_present(table_name VARCHAR2) IS
+ BEGIN
+ EXECUTE IMMEDIATE 'DROP TABLE ' || table_name;
+ EXCEPTION
+ WHEN table_missing THEN NULL;
+ END;
+BEGIN
+ drop_if_present('HOP_SYNC_ERR_TARGET');
+ drop_if_present('HOP_SYNC_ERR_REJECTS');
+ drop_if_present('HOP_SYNC_ERR_WRITTEN');
+
+ EXECUTE IMMEDIATE 'CREATE TABLE HOP_SYNC_ERR_TARGET (
+ ID NUMBER(10) PRIMARY KEY,
+ NAME VARCHAR2(10))';
+
+ EXECUTE IMMEDIATE 'CREATE TABLE HOP_SYNC_ERR_WRITTEN (
+ ID NUMBER(10),
+ NAME VARCHAR2(10))';
+
+ EXECUTE IMMEDIATE 'CREATE TABLE HOP_SYNC_ERR_REJECTS (
+ ID NUMBER(10),
+ NAME VARCHAR2(10),
+ ERR_DESC VARCHAR2(2000))';
+
+ -- The rows the batch will collide with. Seeded and committed up front so
the collision is with
+ -- existing data and does not depend on how the driver orders the statements
within the batch.
+ EXECUTE IMMEDIATE 'INSERT INTO HOP_SYNC_ERR_TARGET VALUES (9000, ''seed'')';
+ EXECUTE IMMEDIATE 'INSERT INTO HOP_SYNC_ERR_TARGET VALUES (9001, ''seed'')';
+ COMMIT;
+END;</sql>
+ <useVariableSubstitution>F</useVariableSubstitution>
+ <sqlfromfile>F</sqlfromfile>
+ <sqlfilename/>
+ <sendOneStatement>T</sendOneStatement>
+ <connection>oracle-service-name</connection>
+ <parallel>N</parallel>
+ <xloc>224</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0007-sync-batch-error-accounting.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+ <filename>${PROJECT_HOME}/0007-sync-batch-error-accounting.hpl</filename>
+ <logext/>
+ <logfile/>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>432</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>no row went missing</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <sql>DECLARE
+ written NUMBER;
+ rejected NUMBER;
+ in_table NUMBER;
+BEGIN
+ SELECT COUNT(*) INTO written FROM HOP_SYNC_ERR_WRITTEN;
+ SELECT COUNT(*) INTO rejected FROM HOP_SYNC_ERR_REJECTS;
+
+ -- The invariant that was broken. Ten rows went in; ten rows have to come
out somewhere. Before
+ -- the fix this was 3 + 0 on Oracle, with seven rows gone without a trace.
+ IF written + rejected <> 10 THEN
+ RAISE_APPLICATION_ERROR(-20070,
+ 'rows went missing: ' || written || ' written + ' || rejected ||
+ ' rejected, expected 10 in total');
+ END IF;
+
+ -- Oracle stops at the fourth row of the batch, so only the three before it
were applied. This
+ -- transform cannot re-drive the six good rows behind the failure, so they
are rejected rather
+ -- than written - see the note on this workflow. Table Output writes eight
for the same input.
+ IF written <> 3 THEN
+ RAISE_APPLICATION_ERROR(-20071,
+ 'expected the 3 rows before the first failure to be reported as written,
got ' || written);
+ END IF;
+
+ -- What the transform claimed matches what Oracle actually holds: the 2 seed
rows plus those 3.
+ SELECT COUNT(*) INTO in_table FROM HOP_SYNC_ERR_TARGET;
+ IF in_table <> 5 THEN
+ RAISE_APPLICATION_ERROR(-20072,
+ 'the target table holds ' || in_table || ' rows, expected 2 seeded + 3
written');
+ END IF;
+
+ -- No row was reported as written while it is absent from the table.
+ SELECT COUNT(*) INTO in_table
+ FROM HOP_SYNC_ERR_WRITTEN w
+ WHERE NOT EXISTS (SELECT 1 FROM HOP_SYNC_ERR_TARGET t WHERE t.ID = w.ID);
+ IF in_table <> 0 THEN
+ RAISE_APPLICATION_ERROR(-20073,
+ in_table || ' rows were reported as written but are not in the table');
+ END IF;
+
+ -- The row Oracle actually refused keeps the database's own message; the six
behind it say they
+ -- were never attempted, so a user reading the rejects can tell which row
was the bad one.
+ SELECT COUNT(*) INTO in_table
+ FROM HOP_SYNC_ERR_REJECTS
+ WHERE ERR_DESC LIKE '%not sent to the database%';
+ IF in_table <> 6 THEN
+ RAISE_APPLICATION_ERROR(-20074,
+ 'expected 6 rows marked as never attempted, got ' || in_table);
+ END IF;
+
+ SELECT COUNT(*) INTO in_table
+ FROM HOP_SYNC_ERR_REJECTS
+ WHERE ID = 9000 AND ERR_DESC LIKE '%ORA-00001%';
+ IF in_table <> 1 THEN
+ RAISE_APPLICATION_ERROR(-20075,
+ 'the row Oracle rejected should carry the ORA-00001 unique constraint
message');
+ END IF;
+END;</sql>
+ <useVariableSubstitution>F</useVariableSubstitution>
+ <sqlfromfile>F</sqlfromfile>
+ <sqlfilename/>
+ <sendOneStatement>T</sendOneStatement>
+ <connection>oracle-service-name</connection>
+ <parallel>N</parallel>
+ <xloc>688</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>Init tables</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>Init tables</from>
+ <to>0007-sync-batch-error-accounting.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0007-sync-batch-error-accounting.hpl</from>
+ <to>no row went missing</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/plugins/transforms/synchronizeaftermerge/src/main/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMerge.java
b/plugins/transforms/synchronizeaftermerge/src/main/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMerge.java
index bba2613f9e..70e1a17b68 100644
---
a/plugins/transforms/synchronizeaftermerge/src/main/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMerge.java
+++
b/plugins/transforms/synchronizeaftermerge/src/main/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMerge.java
@@ -19,6 +19,7 @@ package
org.apache.hop.pipeline.transforms.synchronizeaftermerge;
import java.sql.PreparedStatement;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.Arrays;
import java.util.List;
import org.apache.hop.core.Const;
@@ -552,36 +553,88 @@ public class SynchronizeAfterMerge
}
}
- private void processBatchException(
+ /**
+ * Route every row of a failed batch to the regular output or to the error
stream.
+ *
+ * <p>Drivers do not agree on what {@link
java.sql.BatchUpdateException#getUpdateCounts()} holds
+ * after a failure. Some report one count per row, marking the failures in
place. Some stop at the
+ * first failure, so the array is <em>shorter</em> than the batch: the row
at {@code
+ * updateCounts.length} is the one that failed and the rows behind it were
never sent at all
+ * (Oracle and Derby; a failure on the first row gives a zero-length array,
not a null one). Some
+ * report nothing usable.
+ *
+ * <p>Whatever comes back, every buffered row has to leave here on one of
the two streams. Walking
+ * only the counts array and then clearing the buffer is what used to make
rows disappear without
+ * an error row or a log line - see <a
href="https://github.com/apache/hop/issues/5758";>issue
+ * 5758</a>, which reported it against Table Output.
+ *
+ * <p>Unlike Table Output, this transform does not re-drive the rows the
database never attempted.
+ * It buffers rows for its insert, update and delete statements in one
interleaved list, so a
+ * buffer position does not identify which statement the row belongs to and
the values to re-bind
+ * cannot be worked out. Those rows are reported as rejects instead, which
keeps them out of the
+ * silent-loss category but does mean this transform writes fewer rows than
Table Output would for
+ * the same input on a driver that abandons the rest of the batch.
+ *
+ * <p>Package-private so the accounting can be exercised directly against
each driver shape,
+ * without a database.
+ */
+ void processBatchException(
String errorMessage, int[] updateCounts, List<Exception> exceptionsList)
throws HopException {
- // There was an error with the commit
- // We should put all the failing rows out there...
+
+ List<Exception> exceptions = exceptionsList == null ? List.of() :
exceptionsList;
+ int bufferSize = data.batchBuffer.size();
+
+ // More counts than rows means the counts belong to a different batch than
the one buffered, so
+ // no row can be matched to a count. Reporting the wrong rows on the wrong
stream would be worse
+ // than failing here, where the mismatch is still visible.
//
- if (updateCounts != null) {
- int errNr = 0;
- for (int i = 0; i < updateCounts.length; i++) {
- Object[] row = data.batchBuffer.get(i);
- if (updateCounts[i] > 0) {
- // send the error forward
- putRow(data.outputRowMeta, row);
- incrementLinesOutput();
- } else {
- String exMessage = errorMessage;
- if (errNr < exceptionsList.size()) {
- SQLException se = (SQLException) exceptionsList.get(errNr);
- errNr++;
- exMessage = se.toString();
- }
- putError(data.outputRowMeta, row, 1L, exMessage, null, "SUYNC002");
+ if (updateCounts != null && updateCounts.length > bufferSize) {
+ throw new HopException(
+ "Unable to attribute batch errors to rows: the database returned "
+ + updateCounts.length
+ + " update counts for a batch of "
+ + bufferSize
+ + " buffered rows.");
+ }
+
+ int counted = updateCounts == null ? 0 : updateCounts.length;
+ int errNr = 0;
+
+ for (int i = 0; i < counted; i++) {
+ Object[] row = data.batchBuffer.get(i);
+ if (updateCounts[i] != Statement.EXECUTE_FAILED) {
+ // Anything that isn't EXECUTE_FAILED is a success. That includes
SUCCESS_NO_INFO, and it
+ // includes a count of zero: this transform issues updates and
deletes, and a statement that
+ // matched no rows still ran.
+ //
+ putRow(data.outputRowMeta, row);
+ incrementLinesOutput();
+ } else {
+ String exMessage = errorMessage;
+ if (errNr < exceptions.size()) {
+ exMessage = exceptions.get(errNr).toString();
+ errNr++;
}
+ putError(data.outputRowMeta, row, 1L, exMessage, null, "SUYNC002");
}
- } else {
- // If we don't have update counts, it probably means the DB doesn't
support it.
- // In this case we don't have a choice but to consider all inserted rows
to be error rows.
- //
- for (int i = 0; i < data.batchBuffer.size(); i++) {
- Object[] row = data.batchBuffer.get(i);
- putError(data.outputRowMeta, row, 1L, errorMessage, null, "SUYNC003");
+ }
+
+ // Whatever the driver did not account for. With a short array the first
of these is the row it
+ // actually refused, so it keeps the database's own message; the rest were
never attempted.
+ //
+ for (int i = counted; i < bufferSize; i++) {
+ Object[] row = data.batchBuffer.get(i);
+ if (i == counted && updateCounts != null) {
+ putError(data.outputRowMeta, row, 1L, errorMessage, null, "SUYNC002");
+ } else {
+ putError(
+ data.outputRowMeta,
+ row,
+ 1L,
+ BaseMessages.getString(
+ PKG, "SynchronizeAfterMerge.Error.RowNotAttempted",
Const.NVL(errorMessage, "")),
+ null,
+ "SUYNC003");
}
}
diff --git
a/plugins/transforms/synchronizeaftermerge/src/main/resources/org/apache/hop/pipeline/transforms/synchronizeaftermerge/messages/messages_en_US.properties
b/plugins/transforms/synchronizeaftermerge/src/main/resources/org/apache/hop/pipeline/transforms/synchronizeaftermerge/messages/messages_en_US.properties
index fed8dc03a5..5a76c77386 100644
---
a/plugins/transforms/synchronizeaftermerge/src/main/resources/org/apache/hop/pipeline/transforms/synchronizeaftermerge/messages/messages_en_US.properties
+++
b/plugins/transforms/synchronizeaftermerge/src/main/resources/org/apache/hop/pipeline/transforms/synchronizeaftermerge/messages/messages_en_US.properties
@@ -45,6 +45,7 @@ SynchronizeAfterMerge.Injection.UPDATE_FIELD=The field in the
table that can ins
SynchronizeAfterMerge.Injection.UPDATE_TABLE_FIELD=The field to update after
lookup.
SynchronizeAfterMerge.Injection.USE_BATCH_UPDATE=Set this flag to indicate if
batch updates should be used.
SynchronizeAfterMerge.InsertRow=Insert row\: \!
+SynchronizeAfterMerge.Error.RowNotAttempted=This row was not sent to the
database\: an earlier row in the same batch failed and the driver stopped
processing the batch. Original error\: {0}
SynchronizeAfterMerge.Log.BatchModeDisabled=Batch mode is disabled\!
SynchronizeAfterMerge.Log.CheckingRow=Checking row\:
SynchronizeAfterMerge.Log.Error.TableFieldnameEmpty=Tablename field is empty\!
diff --git
a/plugins/transforms/synchronizeaftermerge/src/test/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMergeBatchErrorTest.java
b/plugins/transforms/synchronizeaftermerge/src/test/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMergeBatchErrorTest.java
new file mode 100644
index 0000000000..027f316aab
--- /dev/null
+++
b/plugins/transforms/synchronizeaftermerge/src/test/java/org/apache/hop/pipeline/transforms/synchronizeaftermerge/SynchronizeAfterMergeBatchErrorTest.java
@@ -0,0 +1,307 @@
+/*
+ * 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.synchronizeaftermerge;
+
+import static java.sql.Statement.EXECUTE_FAILED;
+import static java.sql.Statement.SUCCESS_NO_INFO;
+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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+import java.sql.BatchUpdateException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transform.TransformPartitioningMeta;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Accounting for the rows of a failed batch.
+ *
+ * <p>This transform carries the same batch-error handling Table Output does,
and carried the same
+ * defect: the row-splitting loop was bounded by {@code updateCounts.length}
and the buffer was
+ * cleared unconditionally afterwards, so against a driver that stops at the
first failure - Oracle
+ * and Derby - every row from the failure onward reached neither stream. See <a
+ * href="https://github.com/apache/hop/issues/5758";>issue 5758</a>, reported
against Table Output.
+ *
+ * <p>The invariant here is the one that was broken: <em>every buffered row
leaves on exactly one
+ * stream</em>, whatever shape the driver's update counts arrive in.
+ *
+ * <p>Unlike Table Output this transform does not re-drive the rows the
database never attempted -
+ * its three statements share one interleaved buffer, so the values to re-bind
cannot be worked out.
+ * The tests below pin that as the current behaviour rather than as the
desirable one.
+ */
+class SynchronizeAfterMergeBatchErrorTest {
+
+ private static final int BATCH = 10;
+
+ private SynchronizeAfterMerge transform;
+ private SynchronizeAfterMergeData data;
+
+ private List<Object[]> emitted;
+ private List<String> rejected;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ SynchronizeAfterMergeMeta meta = mock(SynchronizeAfterMergeMeta.class);
+ TransformMeta transformMeta = mock(TransformMeta.class);
+ doReturn("transform").when(transformMeta).getName();
+ doReturn(mock(TransformPartitioningMeta.class))
+ .when(transformMeta)
+ .getTargetTransformPartitioningMeta();
+ doReturn(meta).when(transformMeta).getTransform();
+
+ PipelineMeta pipelineMeta = mock(PipelineMeta.class);
+ doReturn(transformMeta).when(pipelineMeta).findTransform(anyString());
+
+ data = new SynchronizeAfterMergeData();
+ data.outputRowMeta = new RowMeta();
+ data.outputRowMeta.addValueMeta(new ValueMetaString("name"));
+ data.batchBuffer = new ArrayList<>();
+
+ transform =
+ spy(
+ new SynchronizeAfterMerge(
+ transformMeta, meta, data, 1, pipelineMeta, spy(new
LocalPipelineEngine())));
+ doReturn(transformMeta).when(transform).getTransformMeta();
+ doReturn(false).when(transform).isRowLevel();
+ doNothing().when(transform).logDetailed(anyString());
+
+ emitted = new ArrayList<>();
+ rejected = new ArrayList<>();
+
+ doAnswer(
+ inv -> {
+ emitted.add(inv.getArgument(1));
+ return null;
+ })
+ .when(transform)
+ .putRow(any(IRowMeta.class), any());
+ doAnswer(
+ inv -> {
+ rejected.add(inv.getArgument(3));
+ return null;
+ })
+ .when(transform)
+ .putError(
+ any(IRowMeta.class),
+ any(),
+ anyLong(),
+ anyString(),
+ nullable(String.class),
+ anyString());
+ }
+
+ private void bufferRows(int count) {
+ for (int i = 0; i < count; i++) {
+ data.batchBuffer.add(new Object[] {"row" + i});
+ }
+ }
+
+ /** Every buffered row has to come out of exactly one of the two streams. */
+ private void assertAllRowsAccountedFor() {
+ assertEquals(
+ BATCH,
+ emitted.size() + rejected.size(),
+ "every buffered row must be emitted exactly once, on one stream or the
other");
+ assertTrue(data.batchBuffer.isEmpty(), "the buffer must be drained");
+ }
+
+ @Test
+ void marksFailedRows_splitsOnTheCountsArray() throws HopException {
+ // SQL Server, MySQL, H2: a full-length array with EXECUTE_FAILED at the
failing rows.
+ bufferRows(BATCH);
+ int[] counts = {1, 1, 1, EXECUTE_FAILED, 1, 1, EXECUTE_FAILED, 1, 1, 1};
+
+ transform.processBatchException("batch failed", counts, List.of());
+
+ assertEquals(8, emitted.size());
+ assertEquals(2, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ /** The regression guard: a short counts array must not take the tail down
with it. */
+ @Test
+ void keepsPrefix_shortCountsArrayDoesNotDropTheTail() throws HopException {
+ // Oracle and Derby stop at the first failure, so a batch of ten that
fails at index three comes
+ // back with three counts. The old loop ended there and the buffer was
cleared: seven rows gone.
+ bufferRows(BATCH);
+
+ transform.processBatchException(
+ "ORA-00001: unique constraint violated", new int[] {1, 1, 1},
List.of());
+
+ assertEquals(3, emitted.size(), "only the rows the driver confirmed may go
downstream");
+ assertEquals(
+ 7, rejected.size(), "the failing row and the six never attempted must
be rejected");
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void keepsPrefix_failureOnTheFirstRowGivesAnEmptyArray() throws HopException
{
+ // A first-row failure yields a zero-length array, not a null one, so the
old code took the
+ // counts branch, ran its loop zero times, and discarded the entire commit
window.
+ bufferRows(BATCH);
+
+ transform.processBatchException("ORA-12899: value too large", new int[0],
List.of());
+
+ assertEquals(0, emitted.size());
+ assertEquals(BATCH, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void keepsPrefix_theFailingRowIsDistinguishedFromTheUntriedOnes() throws
HopException {
+ bufferRows(BATCH);
+
+ transform.processBatchException(
+ "ORA-00001: unique constraint violated", new int[] {1, 1, 1},
List.of());
+
+ assertEquals(
+ "ORA-00001: unique constraint violated",
+ rejected.get(0),
+ "the row the database actually rejected keeps the database's own
message");
+ for (String message : rejected.subList(1, rejected.size())) {
+ assertTrue(
+ message.contains("not sent to the database"),
+ "rows the driver never attempted must say so, not repeat the first
row's error");
+ }
+ }
+
+ @Test
+ void losesWholeBatch_everyRowIsRejected() throws HopException {
+ // PostgreSQL and DuckDB abort the transaction: a full-length array of
EXECUTE_FAILED, and
+ // nothing durable.
+ bufferRows(BATCH);
+ int[] counts = new int[BATCH];
+ Arrays.fill(counts, EXECUTE_FAILED);
+
+ transform.processBatchException("current transaction is aborted", counts,
List.of());
+
+ assertEquals(0, emitted.size());
+ assertEquals(BATCH, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void noCountsAtAll_everyRowIsRejected() throws HopException {
+ // A plain SQLException rather than a BatchUpdateException - SQLite,
DuckDB, and SQL Server's
+ // batch-aborting error classes.
+ bufferRows(BATCH);
+
+ transform.processBatchException("driver reported no update counts", null,
List.of());
+
+ assertEquals(0, emitted.size());
+ assertEquals(BATCH, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void successNoInfoCountsAsSuccess() throws HopException {
+ // JDBC says success is anything that is not EXECUTE_FAILED. The old "> 0"
test read
+ // SUCCESS_NO_INFO as a failure and pushed rows the database had applied
onto the error stream.
+ bufferRows(BATCH);
+ int[] counts = {
+ SUCCESS_NO_INFO, SUCCESS_NO_INFO, SUCCESS_NO_INFO, EXECUTE_FAILED,
SUCCESS_NO_INFO,
+ SUCCESS_NO_INFO, SUCCESS_NO_INFO, SUCCESS_NO_INFO, SUCCESS_NO_INFO,
SUCCESS_NO_INFO
+ };
+
+ transform.processBatchException("batch failed", counts, List.of());
+
+ assertEquals(9, emitted.size(), "SUCCESS_NO_INFO must not be read as a
failure");
+ assertEquals(1, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ /**
+ * This transform issues updates and deletes, not just inserts, so a
statement that matched no
+ * rows is entirely ordinary - and the old {@code > 0} test reported every
one of them as an
+ * error.
+ */
+ @Test
+ void zeroRowCountIsNotAFailure() throws HopException {
+ bufferRows(BATCH);
+ int[] counts = {1, 0, 1, 0, 0, 1, 1, 0, 1, 1};
+
+ transform.processBatchException("batch failed", counts, List.of());
+
+ assertEquals(
+ BATCH, emitted.size(), "an update or delete that matched no rows still
ran successfully");
+ assertEquals(0, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void perRowMessagesAreUsedWhenTheDriverChainsThem() throws HopException {
+ bufferRows(BATCH);
+ int[] counts = {1, EXECUTE_FAILED, 1, EXECUTE_FAILED, 1, 1, 1, 1, 1, 1};
+ List<Exception> chained =
+ List.of(
+ new BatchUpdateException("first row problem", counts),
+ new BatchUpdateException("second row problem", counts));
+
+ transform.processBatchException("generic batch message", counts, chained);
+
+ assertEquals(2, rejected.size());
+ assertTrue(rejected.get(0).contains("first row problem"));
+ assertTrue(rejected.get(1).contains("second row problem"));
+ }
+
+ @Test
+ void moreCountsThanBufferedRowsFailsLoudly() {
+ // The counts belong to a different batch than the one buffered - which
this transform can
+ // produce on its own, because it runs a commit schedule alongside the one
inside Database. No
+ // row can be matched to a count, and reporting the wrong rows would be
worse than stopping.
+ bufferRows(3);
+
+ HopException e =
+ assertThrows(
+ HopException.class,
+ () ->
+ transform.processBatchException("mismatch", new int[] {1, 1,
1, 1, 1}, List.of()));
+ assertTrue(e.getMessage().contains("Unable to attribute batch errors to
rows"));
+ }
+
+ @Test
+ void nullExceptionListIsTolerated() throws HopException {
+ bufferRows(BATCH);
+ int[] counts = {1, 1, 1, EXECUTE_FAILED, 1, 1, 1, 1, 1, 1};
+
+ transform.processBatchException("batch failed", counts, null);
+
+ assertEquals(9, emitted.size());
+ assertEquals(1, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+}
diff --git
a/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutput.java
b/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutput.java
index abbcb92a8b..78a7bd03be 100644
---
a/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutput.java
+++
b/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutput.java
@@ -20,6 +20,7 @@ package org.apache.hop.pipeline.transforms.tableoutput;
import com.google.common.annotations.VisibleForTesting;
import java.sql.PreparedStatement;
import java.sql.SQLException;
+import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
@@ -474,9 +475,11 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
if (sendToErrorRow) {
if (batchProblem) {
data.batchBuffer.add(outputRowData);
+ data.batchBindBuffer.add(insertRowData);
outputRowData = null;
- processBatchException(errorMessage, updateCounts, exceptionsList);
+ processBatchException(
+ insertStatement, tableName, errorMessage, updateCounts,
exceptionsList);
} else {
// Simply add this row to the error row
putError(rowMeta, r, 1L, errorMessage, null, "TOP001");
@@ -484,6 +487,7 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
}
} else {
data.batchBuffer.add(outputRowData);
+ data.batchBindBuffer.add(insertRowData);
outputRowData = null;
if (rowIsSafe) { // A commit was done and the rows are all safe (no
error)
@@ -494,6 +498,7 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
}
// Clear the buffer
data.batchBuffer.clear();
+ data.batchBindBuffer.clear();
}
}
} else {
@@ -506,41 +511,186 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
return outputRowData;
}
- private void processBatchException(
- String errorMessage, int[] updateCounts, List<Exception> exceptionsList)
throws HopException {
- // There was an error with the commit
- // We should put all the failing rows out there...
+ /**
+ * Route every row of a failed batch to the regular output or to the error
stream.
+ *
+ * <p>JDBC lets a driver react to a failing statement in more than one way,
and the update counts
+ * we get back mean different things as a result:
+ *
+ * <ul>
+ * <li>Some drivers execute the whole batch and hand back one count per
row, with {@link
+ * Statement#EXECUTE_FAILED} at the rows that failed. SQL Server,
MySQL and H2 do this.
+ * <li>Some stop at the first failure. The array is then <em>shorter</em>
than the batch: it
+ * holds only the rows that succeeded before the failure, the row at
{@code
+ * updateCounts.length} is the one that failed, and the rows behind it
were never sent to
+ * the database at all. Oracle and Derby do this, and a failure on the
very first row of the
+ * batch gives us a zero-length array rather than a null one.
+ * <li>Some report nothing usable and we get no counts at all.
+ * </ul>
+ *
+ * <p>Whichever we get, every row in the buffer has to leave here on exactly
one of the two
+ * streams. Rows the counts array does not account for are not in the
database, so they are
+ * rejects - silently dropping them is what used to make rows disappear on
Oracle.
+ */
+ @VisibleForTesting
+ void processBatchException(
+ PreparedStatement insertStatement,
+ String tableName,
+ String errorMessage,
+ int[] updateCounts,
+ List<Exception> exceptionsList)
+ throws HopException {
+
+ int bufferSize = data.batchBuffer.size();
+
+ // More counts than rows means the counts belong to a different batch than
the one we buffered,
+ // so the row a count refers to can't be worked out. Reporting the wrong
rows on the wrong
+ // stream is worse than failing here, where the mismatch is still visible.
+ //
+ if (updateCounts != null && updateCounts.length > bufferSize) {
+ throw new HopException(
+ "Unable to attribute batch errors to rows: the database returned "
+ + updateCounts.length
+ + " update counts for a batch of "
+ + bufferSize
+ + " buffered rows.");
+ }
+
+ // Re-driving the untried rows means binding them again, which is only
unambiguous when the
+ // whole buffer belongs to the one statement that just failed. With the
table name in a field or
+ // date partitioning the buffer interleaves rows for several tables, so
the tail is reported
+ // rather than retried.
//
- if (updateCounts != null) {
+ boolean canRetryTail =
+ insertStatement != null
+ && !meta.isTableNameInField()
+ && !meta.isPartitioningEnabled()
+ && data.batchBindBuffer.size() == bufferSize;
+
+ int from = 0;
+ int[] counts = updateCounts;
+ String message = errorMessage;
+ List<Exception> exceptions = exceptionsList == null ? List.of() :
exceptionsList;
+
+ while (from < bufferSize) {
+
+ // Rows the database told us about, relative to where this attempt
started.
+ //
+ int counted = counts == null ? 0 : counts.length;
int errNr = 0;
- for (int i = 0; i < updateCounts.length; i++) {
- Object[] row = data.batchBuffer.get(i);
- if (updateCounts[i] > 0) {
- // send the error foward
+ for (int i = 0; counts != null && i < counted; i++) {
+ Object[] row = data.batchBuffer.get(from + i);
+ if (counts[i] != Statement.EXECUTE_FAILED) {
+ // Anything that isn't EXECUTE_FAILED is a success: a row count, or
SUCCESS_NO_INFO from a
+ // driver that applied the row without saying how many rows it
touched.
+ //
putRow(data.outputRowMeta, row);
incrementLinesOutput();
} else {
- String exMessage = errorMessage;
- if (errNr < exceptionsList.size()) {
- SQLException se = (SQLException) exceptionsList.get(errNr);
+ String exMessage = message;
+ if (errNr < exceptions.size()) {
+ exMessage = exceptions.get(errNr).toString();
errNr++;
- exMessage = se.toString();
}
putError(data.outputRowMeta, row, 1L, exMessage, null, "TOP0002");
}
}
- } else {
- // If we don't have update counts, it probably means the DB doesn't
support it.
- // In this case we don't have a choice but to consider all inserted rows
to be error rows.
+ from += counted;
+
+ if (from >= bufferSize) {
+ break;
+ }
+
+ if (counts == null) {
+ // The driver gave us nothing to go on: it never said which rows it
applied, so retrying
+ // could write a row twice. Everything still unaccounted for is
reported instead.
+ //
+ for (int i = from; i < bufferSize; i++) {
+ putError(data.outputRowMeta, data.batchBuffer.get(i), 1L, message,
null, "TOP0003");
+ }
+ break;
+ }
+
+ // A short counts array means the driver stopped here. This row is the
one it refused, and it
+ // gets the database's own message.
//
- for (int i = 0; i < data.batchBuffer.size(); i++) {
- Object[] row = data.batchBuffer.get(i);
- putError(data.outputRowMeta, row, 1L, errorMessage, null, "TOP0003");
+ putError(data.outputRowMeta, data.batchBuffer.get(from), 1L, message,
null, "TOP0002");
+ from++;
+
+ if (from >= bufferSize) {
+ break;
}
+
+ if (!canRetryTail) {
+ for (int i = from; i < bufferSize; i++) {
+ putError(
+ data.outputRowMeta,
+ data.batchBuffer.get(i),
+ 1L,
+ notAttemptedMessage(message),
+ null,
+ "TOP0003");
+ }
+ break;
+ }
+
+ // The rows behind the failure were never sent, so they are still owed a
write. Re-drive them
+ // as one batch: a batch that fails again comes back round this loop,
which costs one extra
+ // round trip per bad row rather than a round trip per row.
+ //
+ try {
+ counts = retryBatch(insertStatement, from, bufferSize);
+ message = errorMessage;
+ exceptions = List.of();
+ } catch (HopDatabaseBatchException be) {
+ counts = be.getUpdateCounts();
+ message = be.toString();
+ exceptions = be.getExceptionsList() == null ? List.of() :
be.getExceptionsList();
+ }
+ }
+
+ if (isDetailed()) {
+ logDetailed(
+ "Recovered a failed batch of " + bufferSize + " rows on table [" +
tableName + "]");
}
- // Clear the buffer afterwards...
+ // Clear the buffers afterwards...
data.batchBuffer.clear();
+ data.batchBindBuffer.clear();
+ }
+
+ /**
+ * Re-binds and re-executes rows {@code [from, to)} of the batch buffer on
the same prepared
+ * statement, and commits them.
+ *
+ * @return one update count per row submitted, so the caller can attribute
them from {@code from}
+ * @throws HopDatabaseBatchException if this attempt fails in its turn,
carrying whatever the
+ * driver reported about it
+ */
+ private int[] retryBatch(PreparedStatement insertStatement, int from, int to)
+ throws HopException {
+ int[] counts;
+ try {
+ for (int i = from; i < to; i++) {
+ data.db.setValues(data.insertRowMeta, data.batchBindBuffer.get(i),
insertStatement);
+ insertStatement.addBatch();
+ }
+ counts = insertStatement.executeBatch();
+ data.db.commit();
+ insertStatement.clearBatch();
+ } catch (SQLException ex) {
+ data.db.clearBatch(insertStatement);
+ data.db.commit(true);
+ throw Database.createHopDatabaseBatchException("Error updating batch",
ex);
+ }
+ // executeBatch may return fewer counts than rows submitted; the caller
handles that the same
+ // way it handles the original short array.
+ return counts;
+ }
+
+ private String notAttemptedMessage(String errorMessage) {
+ return BaseMessages.getString(
+ PKG, "TableOutput.Error.RowNotAttempted", Const.NVL(errorMessage, ""));
}
@Override
@@ -1232,6 +1382,7 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
data.db = null;
data.preparedStatements = null;
data.batchBuffer = null;
+ data.batchBindBuffer = null;
data.commitCounterMap = null;
data.outputRowMeta = null;
}
@@ -1249,6 +1400,10 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
}
private void emptyAndCommitBatchBuffers(boolean dispose) {
+ // Which statement was being flushed when it threw, so a failure can be
recovered against the
+ // statement that actually produced it.
+ PreparedStatement flushingStatement = null;
+ String flushingTable = null;
try {
for (String schemaTable : data.preparedStatements.keySet()) {
// Get a commit counter per prepared statement to keep track of
separate tables, etc.
@@ -1259,6 +1414,8 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
}
PreparedStatement insertStatement =
data.preparedStatements.get(schemaTable);
+ flushingStatement = insertStatement;
+ flushingTable = schemaTable;
data.db.emptyAndCommit(insertStatement, data.batchMode, batchCounter,
dispose);
data.commitCounterMap.put(schemaTable, 0);
}
@@ -1269,12 +1426,21 @@ public class TableOutput extends
BaseTransform<TableOutputMeta, TableOutputData>
}
// Clear the buffer
data.batchBuffer.clear();
+ data.batchBindBuffer.clear();
} catch (HopDatabaseBatchException be) {
if (getTransformMeta().isDoingErrorHandling()) {
// Right at the back we are experiencing a batch commit problem...
// OK, we have the numbers...
try {
- processBatchException(be.toString(), be.getUpdateCounts(),
be.getExceptionsList());
+ // The statement is still open here: Database.emptyAndCommit only
closes it after a
+ // successful executeBatch, so a failure leaves it usable and
dispose()'s own finally
+ // closes it afterwards. The last partial batch gets the same
recovery as any other.
+ processBatchException(
+ flushingStatement,
+ flushingTable,
+ be.toString(),
+ be.getUpdateCounts(),
+ be.getExceptionsList());
} catch (HopException e) {
logError("Unexpected error processing batch error", e);
setErrors(1);
diff --git
a/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputData.java
b/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputData.java
index eaae4c60bf..ad5ffd27b2 100644
---
a/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputData.java
+++
b/plugins/transforms/tableoutput/src/main/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputData.java
@@ -68,6 +68,15 @@ public class TableOutputData extends BaseTransformData
implements ITransformData
public boolean dynamicLineageTruncated;
public List<Object[]> batchBuffer;
+
+ /**
+ * The values bound to the prepared statement for each row in {@link
#batchBuffer}, in the same
+ * order. It is not the same array: with "specify database fields" the bound
row is a projection
+ * of the stream row, and with the table name in a field it is a copy with
that column removed. A
+ * batch that has to be re-driven after a failure needs the values that were
actually bound.
+ */
+ public List<Object[]> batchBindBuffer;
+
public boolean sendToErrorRow;
public IRowMeta outputRowMeta;
public IRowMeta insertRowMeta;
@@ -94,6 +103,7 @@ public class TableOutputData extends BaseTransformData
implements ITransformData
indexOfTableNameField = -1;
batchBuffer = new ArrayList<>();
+ batchBindBuffer = new ArrayList<>();
commitCounterMap = new HashMap<>();
releaseSavepoint = true;
diff --git
a/plugins/transforms/tableoutput/src/main/resources/org/apache/hop/pipeline/transforms/tableoutput/messages/messages_en_US.properties
b/plugins/transforms/tableoutput/src/main/resources/org/apache/hop/pipeline/transforms/tableoutput/messages/messages_en_US.properties
index cfc0c44f59..a414d743b3 100644
---
a/plugins/transforms/tableoutput/src/main/resources/org/apache/hop/pipeline/transforms/tableoutput/messages/messages_en_US.properties
+++
b/plugins/transforms/tableoutput/src/main/resources/org/apache/hop/pipeline/transforms/tableoutput/messages/messages_en_US.properties
@@ -22,6 +22,7 @@ TableOutput.Init.ConnectionMissing=Database connection is
missing for transform
TableOutput.Name=Table output
TableOutput.UnsupportedConnection.DialogTitle=Unable to use database connection
TableOutput.Warning=Warning!
+TableOutput.Error.RowNotAttempted=This row was not sent to the database\: an
earlier row in the same batch failed and the driver stopped processing the
batch. Original error\: {0}
TableOutput.Warning.ErrorHandlingIsNotFullySupportedWithBatchProcessing=WARNING\!
Error handling in combination with batch processing is not fully supported on
the used database because of driver limitations. Proceed with caution at your
own risk.
TableOutputDialog.AvailableSchemas.Message=Please select a schema name
TableOutputDialog.AvailableSchemas.Title=Available schemas
diff --git
a/plugins/transforms/tableoutput/src/test/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputBatchErrorTest.java
b/plugins/transforms/tableoutput/src/test/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputBatchErrorTest.java
new file mode 100644
index 0000000000..c3957d2acf
--- /dev/null
+++
b/plugins/transforms/tableoutput/src/test/java/org/apache/hop/pipeline/transforms/tableoutput/TableOutputBatchErrorTest.java
@@ -0,0 +1,405 @@
+/*
+ * 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.tableoutput;
+
+import static java.sql.Statement.EXECUTE_FAILED;
+import static java.sql.Statement.SUCCESS_NO_INFO;
+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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+import java.sql.BatchUpdateException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hop.core.database.Database;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transform.TransformPartitioningMeta;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Accounting for the rows of a failed batch.
+ *
+ * <p>JDBC drivers disagree about what {@link
BatchUpdateException#getUpdateCounts()} contains, and
+ * the transform used to assume the one shape SQL Server produces. Against a
driver that stops at
+ * the first failure - Oracle and Derby do - the array is shorter than the
batch, and the rows it
+ * does not mention used to be dropped on the floor: no output row, no error
row, no log line. That
+ * is <a href="https://github.com/apache/hop/issues/5758";>issue #5758</a>.
+ *
+ * <p>The invariant every test here checks is the same one: <em>every buffered
row leaves on exactly
+ * one stream</em>, whatever the driver reported.
+ */
+class TableOutputBatchErrorTest {
+
+ private static final int BATCH = 10;
+
+ private TableOutput transform;
+ private TableOutputData data;
+ private TableOutputMeta meta;
+
+ private List<Object[]> written;
+ private List<String> rejected;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ meta = mock(TableOutputMeta.class);
+ TransformMeta transformMeta = mock(TransformMeta.class);
+ doReturn("transform").when(transformMeta).getName();
+ doReturn(mock(TransformPartitioningMeta.class))
+ .when(transformMeta)
+ .getTargetTransformPartitioningMeta();
+ doReturn(meta).when(transformMeta).getTransform();
+
+ PipelineMeta pipelineMeta = mock(PipelineMeta.class);
+ doReturn(transformMeta).when(pipelineMeta).findTransform(anyString());
+
+ data = new TableOutputData();
+ data.outputRowMeta = new RowMeta();
+ data.outputRowMeta.addValueMeta(new ValueMetaString("name"));
+ data.insertRowMeta = data.outputRowMeta;
+ data.batchBuffer = new ArrayList<>();
+ data.batchBindBuffer = new ArrayList<>();
+ data.db = mock(Database.class);
+
+ transform =
+ spy(
+ new TableOutput(
+ transformMeta, meta, data, 1, pipelineMeta, spy(new
LocalPipelineEngine())));
+ doReturn(transformMeta).when(transform).getTransformMeta();
+ doReturn(false).when(transform).isRowLevel();
+ doNothing().when(transform).logDetailed(anyString());
+
+ written = new ArrayList<>();
+ rejected = new ArrayList<>();
+
+ doAnswer(
+ inv -> {
+ written.add(inv.getArgument(1));
+ return null;
+ })
+ .when(transform)
+ .putRow(any(IRowMeta.class), any());
+ doAnswer(
+ inv -> {
+ rejected.add(inv.getArgument(3));
+ return null;
+ })
+ .when(transform)
+ .putError(
+ any(IRowMeta.class),
+ any(),
+ anyLong(),
+ anyString(),
+ nullable(String.class),
+ anyString());
+ }
+
+ private void bufferRows(int count) {
+ for (int i = 0; i < count; i++) {
+ data.batchBuffer.add(new Object[] {"row" + i});
+ }
+ }
+
+ /** Every buffered row has to come out of exactly one of the two streams. */
+ private void assertAllRowsAccountedFor() {
+ assertEquals(
+ BATCH,
+ written.size() + rejected.size(),
+ "every buffered row must be emitted exactly once, on one stream or the
other");
+ assertTrue(data.batchBuffer.isEmpty(), "the buffer must be drained");
+ }
+
+ @Test
+ void marksFailedRows_splitsOnTheCountsArray() throws HopException {
+ // SQL Server, MySQL, H2: a full-length array with EXECUTE_FAILED at the
failing rows.
+ bufferRows(BATCH);
+ int[] counts = {1, 1, 1, EXECUTE_FAILED, 1, 1, EXECUTE_FAILED, 1, 1, 1};
+
+ transform.processBatchException(null, "T", "batch failed", counts,
List.of());
+
+ assertEquals(8, written.size());
+ assertEquals(2, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ /** The regression test for issue #5758. */
+ @Test
+ void keepsPrefix_shortCountsArrayDoesNotDropTheTail() throws HopException {
+ // Oracle and Derby stop at the first failure, so a batch of 10 that fails
at index 3 comes back
+ // with only three counts. Rows 3..9 are not in the database and must be
rejected, not dropped.
+ bufferRows(BATCH);
+ int[] counts = {1, 1, 1};
+
+ transform.processBatchException(
+ null, "T", "ORA-00001: unique constraint violated", counts, List.of());
+
+ assertEquals(3, written.size(), "only the rows the driver confirmed may go
downstream");
+ assertEquals(
+ 7, rejected.size(), "the failing row and the six never attempted must
be rejected");
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void keepsPrefix_failureOnTheFirstRowGivesAnEmptyArray() throws HopException
{
+ // Oracle returns a zero-length array - not null - when the very first row
of the batch fails.
+ // The old code's loop simply never ran, and the whole commit window
disappeared.
+ bufferRows(BATCH);
+
+ transform.processBatchException(null, "T", "ORA-12899: value too large",
new int[0], List.of());
+
+ assertEquals(0, written.size());
+ assertEquals(BATCH, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void keepsPrefix_theFailingRowIsDistinguishedFromTheUntriedOnes() throws
HopException {
+ bufferRows(BATCH);
+
+ transform.processBatchException(
+ null, "T", "ORA-00001: unique constraint violated", new int[] {1, 1,
1}, List.of());
+
+ assertEquals(
+ "ORA-00001: unique constraint violated",
+ rejected.get(0),
+ "the row the database actually rejected keeps the database's own
message");
+ for (String message : rejected.subList(1, rejected.size())) {
+ assertTrue(
+ message.contains("not sent to the database"),
+ "rows the driver never attempted must say so, not repeat the first
row's error");
+ }
+ }
+
+ @Test
+ void losesWholeBatch_everyRowIsRejected() throws HopException {
+ // PostgreSQL and DuckDB abort the transaction: a full-length array of
EXECUTE_FAILED, nothing
+ // durable.
+ bufferRows(BATCH);
+ int[] counts = new int[BATCH];
+ Arrays.fill(counts, EXECUTE_FAILED);
+
+ transform.processBatchException(null, "T", "current transaction is
aborted", counts, List.of());
+
+ assertEquals(0, written.size());
+ assertEquals(BATCH, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void noCountsAtAll_everyRowIsRejected() throws HopException {
+ // A plain SQLException rather than a BatchUpdateException - SQLite,
DuckDB, and SQL Server's
+ // batch-aborting error classes.
+ bufferRows(BATCH);
+
+ transform.processBatchException(null, "T", "driver reported no update
counts", null, List.of());
+
+ assertEquals(0, written.size());
+ assertEquals(BATCH, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void successNoInfoCountsAsSuccess() throws HopException {
+ // JDBC says success is anything that is not EXECUTE_FAILED.
SUCCESS_NO_INFO (-2) means the row
+ // was applied without the driver saying how many rows it touched; the old
"> 0" test read that
+ // as a failure and pushed rows that ARE in the table onto the error
stream.
+ bufferRows(BATCH);
+ int[] counts = {
+ SUCCESS_NO_INFO, SUCCESS_NO_INFO, SUCCESS_NO_INFO, EXECUTE_FAILED,
SUCCESS_NO_INFO,
+ SUCCESS_NO_INFO, SUCCESS_NO_INFO, SUCCESS_NO_INFO, SUCCESS_NO_INFO,
SUCCESS_NO_INFO
+ };
+
+ transform.processBatchException(null, "T", "batch failed", counts,
List.of());
+
+ assertEquals(9, written.size(), "SUCCESS_NO_INFO must not be read as a
failure");
+ assertEquals(1, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void zeroRowCountIsNotAFailure() throws HopException {
+ // A statement that matched no rows is still a statement that ran.
+ bufferRows(BATCH);
+ int[] counts = {1, 0, 1, 0, 1, 1, 1, 1, 1, 1};
+
+ transform.processBatchException(null, "T", "batch failed", counts,
List.of());
+
+ assertEquals(BATCH, written.size());
+ assertEquals(0, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void perRowMessagesAreUsedWhenTheDriverChainsThem() throws HopException {
+ bufferRows(BATCH);
+ int[] counts = {1, EXECUTE_FAILED, 1, EXECUTE_FAILED, 1, 1, 1, 1, 1, 1};
+ List<Exception> chained =
+ List.of(
+ new BatchUpdateException("first row problem", counts),
+ new BatchUpdateException("second row problem", counts));
+
+ transform.processBatchException(null, "T", "generic batch message",
counts, chained);
+
+ assertEquals(2, rejected.size());
+ assertTrue(rejected.get(0).contains("first row problem"));
+ assertTrue(rejected.get(1).contains("second row problem"));
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // Re-driving the untried tail. A driver that stops at the first failure
never sent the rows
+ // behind
+ // it, so they are still owed a write: rejecting them would leave Oracle
writing fewer rows than
+ // PostgreSQL or SQL Server for the same input.
+ //
---------------------------------------------------------------------------------------------
+
+ /** A prepared statement that fails the batch at a given offset, the way
Oracle does. */
+ private PreparedStatement statementFailingAt(int... failingOffsets) throws
Exception {
+ PreparedStatement ps = mock(PreparedStatement.class);
+ List<Integer> remaining = new ArrayList<>();
+ for (int f : failingOffsets) {
+ remaining.add(f);
+ }
+ int[] submitted = {0};
+ doAnswer(inv -> submitted[0]++).when(ps).addBatch();
+ doAnswer(
+ inv -> {
+ int n = submitted[0];
+ submitted[0] = 0;
+ if (!remaining.isEmpty()) {
+ int stopAt = remaining.remove(0);
+ int[] prefix = new int[stopAt];
+ Arrays.fill(prefix, 1);
+ throw new BatchUpdateException("ORA-00001: unique constraint
violated", prefix);
+ }
+ int[] all = new int[n];
+ Arrays.fill(all, 1);
+ return all;
+ })
+ .when(ps)
+ .executeBatch();
+ return ps;
+ }
+
+ @Test
+ void keepsPrefix_theUntriedTailIsRewrittenNotRejected() throws Exception {
+ // Ten rows, bad ones at 3 and 6. Oracle stops at 3; the retry of rows
4..9 stops at 6; the
+ // retry
+ // of rows 7..9 succeeds. Eight rows end up written, exactly as on a
driver that marks failures
+ // in place - which is the whole point.
+ bufferRows(BATCH);
+ for (int i = 0; i < BATCH; i++) {
+ data.batchBindBuffer.add(new Object[] {"row" + i});
+ }
+ PreparedStatement ps = statementFailingAt(3, 2);
+
+ transform.processBatchException(
+ ps, "T", "ORA-00001: unique constraint violated", new int[] {1, 1, 1},
List.of());
+
+ assertEquals(8, written.size(), "the six good rows behind the failures
must be written");
+ assertEquals(2, rejected.size(), "only the two rows the database actually
refused are rejects");
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void keepsPrefix_retryStopsWhenTheDriverStopsReportingCounts() throws
Exception {
+ // If a retry comes back with no counts at all we can no longer tell what
was applied, so the
+ // remainder is reported rather than driven again - retrying blind could
write a row twice.
+ bufferRows(BATCH);
+ for (int i = 0; i < BATCH; i++) {
+ data.batchBindBuffer.add(new Object[] {"row" + i});
+ }
+ PreparedStatement ps = mock(PreparedStatement.class);
+ doAnswer(inv -> null).when(ps).addBatch();
+ doAnswer(
+ inv -> {
+ throw new SQLException("connection went away");
+ })
+ .when(ps)
+ .executeBatch();
+
+ transform.processBatchException(ps, "T", "ORA-00001", new int[] {1, 1, 1},
List.of());
+
+ assertEquals(3, written.size());
+ assertEquals(7, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void tailIsNotRetriedWhenTheBufferSpansSeveralTables() throws Exception {
+ // With the table name in a field the buffer interleaves rows for
different statements, so the
+ // tail cannot be bound unambiguously. It is reported instead of retried.
+ bufferRows(BATCH);
+ for (int i = 0; i < BATCH; i++) {
+ data.batchBindBuffer.add(new Object[] {"row" + i});
+ }
+ doReturn(true).when(meta).isTableNameInField();
+ PreparedStatement ps = statementFailingAt();
+
+ transform.processBatchException(ps, "T", "ORA-00001", new int[] {1, 1, 1},
List.of());
+
+ assertEquals(3, written.size());
+ assertEquals(7, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+
+ @Test
+ void moreCountsThanBufferedRowsFailsLoudly() {
+ // The counts belong to a different batch than the one we buffered, so no
row can be matched to
+ // a
+ // count. Reporting the wrong rows on the wrong stream would be worse than
stopping here.
+ bufferRows(3);
+
+ HopException e =
+ assertThrows(
+ HopException.class,
+ () ->
+ transform.processBatchException(
+ null, "T", "mismatch", new int[] {1, 1, 1, 1, 1},
List.of()));
+ assertTrue(e.getMessage().contains("Unable to attribute batch errors to
rows"));
+ }
+
+ @Test
+ void nullExceptionListIsTolerated() throws HopException {
+ bufferRows(BATCH);
+ int[] counts = {1, 1, 1, EXECUTE_FAILED, 1, 1, 1, 1, 1, 1};
+
+ transform.processBatchException(null, "T", "batch failed", counts, null);
+
+ assertEquals(9, written.size());
+ assertEquals(1, rejected.size());
+ assertAllRowsAccountedFor();
+ }
+}
