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 427f12ed5c harden and fix vertica bulkloader, fixes #4394 (#8050)
427f12ed5c is described below
commit 427f12ed5ce5ec9c5570e8624500012f05556094
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Aug 21 16:14:19 2026 +0200
harden and fix vertica bulkloader, fixes #4394 (#8050)
---
.../integration-tests-vertica.yaml | 36 +-
.../resource/vertica/vertica-entrypoint.sh | 108 +++++
.../pipeline/transforms/verticabulkloader.adoc | 6 +-
.../vertica/0004-vertica-bulk-load-no-fields.hpl | 153 +++++++
.../vertica/0005-vertica-bulk-load-partial-row.hpl | 118 +++++
integration-tests/vertica/disabled.txt | 17 -
integration-tests/vertica/main-0000-wait.hwf | 76 ----
.../main-0004-vertica-bulk-load-no-fields.hwf | 154 +++++++
.../main-0005-vertica-bulk-load-partial-row.hwf | 152 +++++++
.../vertica/bulkloader/VerticaBulkLoader.java | 269 ++++++------
.../bulkloader/VerticaBulkLoaderDialog.java | 18 +-
.../vertica/bulkloader/VerticaBulkLoaderMeta.java | 32 +-
.../bulkloader/messages/messages_en_US.properties | 4 +
.../bulkloader/VerticaBulkLoaderMetaTest.java | 75 ++++
.../vertica/bulkloader/VerticaBulkLoaderTest.java | 475 +++++++++++++++++++++
.../bulkloader/nativebinary/StreamEncoderTest.java | 158 +++++++
.../resources/vertica-bulkloader-transform.xml | 56 +++
17 files changed, 1650 insertions(+), 257 deletions(-)
diff --git a/docker/integration-tests/integration-tests-vertica.yaml
b/docker/integration-tests/integration-tests-vertica.yaml
index ac9020b5ea..5cfacbd566 100644
--- a/docker/integration-tests/integration-tests-vertica.yaml
+++ b/docker/integration-tests/integration-tests-vertica.yaml
@@ -15,6 +15,16 @@
# specific language governing permissions and limitations
# under the License.
+# Vertica tests.
+#
+# vertica/vertica-ce, the image these tests used to run against, has been
withdrawn from Docker
+# Hub. opentext/vertica-k8s is what OpenText still publishes, but it is built
for the VerticaDB
+# Kubernetes operator: no entry point, no admintools, and no database of its
own. The entry point
+# script below does the three things the operator would otherwise do. See it
for the details.
+#
+# The version is pinned rather than tracking latest for two reasons: the
Community Edition licence
+# these tests rely on was dropped in Vertica 26.1, and the -multiarch tags run
natively on both
+# arm64 and amd64 where the old image was amd64 only.
services:
integration_test_vertica:
extends:
@@ -22,12 +32,32 @@ services:
service: integration_test
environment:
- HOP_DRIVERS_DOWNLOAD=vertica
+ depends_on:
+ vertica:
+ condition: service_healthy
links:
- vertica
vertica:
- image: vertica/vertica-ce:latest
- hostname: vertica
+ image: opentext/vertica-k8s:25.4.0-0-multiarch
+ hostname: vertica
+ # The image runs as the unprivileged "daemon" user and carries no passwd
entry for the account
+ # that owns /opt/vertica. The entry point needs root to create it before
dropping down to it.
+ user: root
+ entrypoint: /vertica-entrypoint.sh
+ volumes:
+ - ./resource/vertica/vertica-entrypoint.sh:/vertica-entrypoint.sh:ro
ports:
- "5433"
- - "5444"
+ healthcheck:
+ # Creating the database takes appreciably longer than opening the port,
so ready means
+ # "answers a query".
+ test:
+ [
+ "CMD-SHELL",
+ 'su dbadmin -c "/opt/vertica/bin/vsql -U dbadmin -d vmart -tAc
\"SELECT 1\"" || exit 1',
+ ]
+ interval: 10s
+ timeout: 10s
+ retries: 30
+ start_period: 60s
diff --git a/docker/integration-tests/resource/vertica/vertica-entrypoint.sh
b/docker/integration-tests/resource/vertica/vertica-entrypoint.sh
new file mode 100755
index 0000000000..65f6253f88
--- /dev/null
+++ b/docker/integration-tests/resource/vertica/vertica-entrypoint.sh
@@ -0,0 +1,108 @@
+#!/bin/bash
+#
+# 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.
+#
+# Boots a single node Vertica database inside opentext/vertica-k8s.
+#
+# That image is built for the VerticaDB Kubernetes operator: it has no entry
point and
+# no admintools, and it does not create a database on its own. The operator
normally
+# generates the TLS material, starts the node management agent (NMA) and then
calls
+# "vcluster create_db" over its REST API. This script does the same three
things so the
+# image can be used from plain docker compose.
+
+set -euo pipefail
+
+DB_NAME=${VERTICA_DB_NAME:-vmart}
+DB_USER=${VERTICA_DB_USER:-dbadmin}
+DB_PASSWORD=${VERTICA_DB_PASSWORD:-}
+DATA_PATH=${VERTICA_DATA_PATH:-/data}
+CERT_DIR=/opt/vertica/config/https_certs
+
+# The image ships the Vertica files owned by uid 997 but carries no matching
passwd entry,
+# because the operator supplies one through the pod security context.
+if ! getent passwd "${DB_USER}" > /dev/null; then
+ groupadd -g 995 verticadba 2> /dev/null || true
+ useradd -u 997 -g 995 -m -d "/home/${DB_USER}" -s /bin/bash "${DB_USER}" 2>
/dev/null || true
+fi
+
+mkdir -p "${DATA_PATH}" /opt/vertica/log /vertica/tmp
+chown -R 997:995 "${DATA_PATH}" /opt/vertica/log /opt/vertica/config /vertica
"/home/${DB_USER}"
+
+as_dbadmin() {
+ su "${DB_USER}" -c "$1"
+}
+
+# The NMA refuses to start without TLS material. The image ships the generator
the
+# operator uses; it writes into the current directory under its own names,
which then
+# have to be linked to the names the NMA and vcluster look for.
+if [ ! -f "${CERT_DIR}/rootca.pem" ]; then
+ echo "Generating TLS certificates for the node management agent"
+ as_dbadmin "cd ${CERT_DIR} && /opt/vertica/bin/gen_httpstls_json.sh" >
/tmp/gen_certs.log 2>&1 ||
+ { echo "Certificate generation failed:"; cat /tmp/gen_certs.log; exit 1; }
+ cd "${CERT_DIR}"
+ cp -f rootca_cert.pem rootca.pem # trusted CA, used by the NMA
and by vcluster
+ cp -f nma_cert.pem vertica_https.pem # NMA server certificate
+ cp -f nma_key.pem vertica_https.key
+ cp -f "${DB_USER}_cert.pem" "${DB_USER}.pem" # vcluster client certificate
+ cp -f "${DB_USER}_key.pem" "${DB_USER}.key"
+ chown 997:995 rootca.pem vertica_https.pem vertica_https.key
"${DB_USER}.pem" "${DB_USER}.key"
+ cd /
+fi
+
+echo "Starting the node management agent"
+rm -f /opt/vertica/config/node_management_agent.pid
+as_dbadmin "nohup /opt/vertica/bin/node_management_agent >
/opt/vertica/log/nma.log 2>&1 & disown"
+
+for _ in $(seq 1 60); do
+ if (exec 3<> /dev/tcp/127.0.0.1/5554) 2> /dev/null; then
+ break
+ fi
+ sleep 1
+done
+if ! (exec 3<> /dev/tcp/127.0.0.1/5554) 2> /dev/null; then
+ echo "The node management agent did not come up:"
+ cat /opt/vertica/log/nma.log
+ exit 1
+fi
+
+# vcluster addresses the node by IP, and only IPv4 is supported.
+HOST_IP=$(grep -oE '^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+' /etc/hosts | grep -v
'^127\.' | head -1)
+
+if [ ! -d "${DATA_PATH}/${DB_NAME}" ]; then
+ echo "Creating database ${DB_NAME} on ${HOST_IP}"
+ as_dbadmin "/opt/vertica/bin/vcluster create_db \
+ --db-name ${DB_NAME} \
+ --hosts ${HOST_IP} \
+ --catalog-path ${DATA_PATH} \
+ --data-path ${DATA_PATH} \
+ --password '${DB_PASSWORD}' \
+ --skip-package-install \
+ --force-cleanup-on-failure \
+ --config-param HttpServerConf=${CERT_DIR}/httpstls.json"
+else
+ echo "Restarting existing database ${DB_NAME} on ${HOST_IP}"
+ as_dbadmin "/opt/vertica/bin/vcluster start_db --db-name ${DB_NAME} --hosts
${HOST_IP} --password '${DB_PASSWORD}'"
+fi
+
+echo "Vertica is ready on port 5433, database ${DB_NAME}"
+
+# create_db leaves the server running in the background, so hold the container
open and
+# forward a stop signal to the database rather than letting the node be killed
outright.
+trap "as_dbadmin \"/opt/vertica/bin/vcluster stop_db --db-name ${DB_NAME}
--password '${DB_PASSWORD}'\" || true; exit 0" TERM INT
+tail -f /opt/vertica/log/nma.log &
+wait $!
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/verticabulkloader.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/verticabulkloader.adoc
index 0cc6829be2..66f83a05cf 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/verticabulkloader.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/verticabulkloader.adoc
@@ -57,7 +57,7 @@ This is typically significantly faster than loading data
through e.g. a Table Ou
|Target table|Name of the target table.
|Truncate table|Truncate this table before loading data
|Truncate on first row|Only truncate this table if there is data (when the
first row is received)
-|Specify database fields|Enable this option to specify the fields in the
Database fields tab. Otherwise all fields are taken into account by default.
+|Specify database fields|Enable this option to specify the fields in the
Database fields tab. Otherwise all fields are taken into account by default and
every incoming field is loaded into the table column of the same name (case
insensitive). A field without a matching column makes the transform fail;
columns that are not on the stream keep their default value.
|===
=== Main options tab
@@ -73,4 +73,6 @@ This is typically significantly faster than loading data
through e.g. a Table Ou
=== Database fields tab
-Map table columns to stream fields using 'Get Fields' and/or 'Enter Field
Mapping'.
\ No newline at end of file
+Map table columns to stream fields using 'Get Fields' and/or 'Enter Field
Mapping'.
+
+This tab is only used when *Specify database fields* is enabled. Without it
the incoming fields are matched to the table columns by name, so the order in
which they arrive does not matter.
\ No newline at end of file
diff --git a/integration-tests/vertica/0004-vertica-bulk-load-no-fields.hpl
b/integration-tests/vertica/0004-vertica-bulk-load-no-fields.hpl
new file mode 100644
index 0000000000..c96189fc33
--- /dev/null
+++ b/integration-tests/vertica/0004-vertica-bulk-load-no-fields.hpl
@@ -0,0 +1,153 @@
+<?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>0004-vertica-bulk-load-no-fields</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Bulk loads without "Specify database fields". The fields
arrive in a different
+ order than the columns of the table, so the loader has to match them on
name.</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/21 10:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/21 10:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>invoices</from>
+ <to>Vertica bulk loader</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>invoices</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>EUR</item>
+ <item>first invoice</item>
+ <item>1</item>
+ <item>Y</item>
+ <item>2024/03/15</item>
+ </line>
+ <line>
+ <item>USD</item>
+ <item>second invoice</item>
+ <item>2</item>
+ <item>N</item>
+ <item>2024/04/01</item>
+ </line>
+ <line>
+ <item>GBP</item>
+ <item>third invoice</item>
+ <item>3</item>
+ <item>Y</item>
+ <item>2024/12/31</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>cost_currency</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>invoice_note</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>invoice_id</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>is_paid</name>
+ <type>Boolean</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>invoice_receipt_date</name>
+ <format>yyyy/MM/dd</format>
+ <type>Date</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Vertica bulk loader</name>
+ <type>VerticaBulkLoader</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <abort_on_error>Y</abort_on_error>
+ <connection>vertica</connection>
+ <direct>Y</direct>
+ <fields>
+ </fields>
+ <specify_fields>N</specify_fields>
+ <table>invoices</table>
+ <attributes/>
+ <GUI>
+ <xloc>336</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/vertica/0005-vertica-bulk-load-partial-row.hpl
b/integration-tests/vertica/0005-vertica-bulk-load-partial-row.hpl
new file mode 100644
index 0000000000..461a5100e8
--- /dev/null
+++ b/integration-tests/vertica/0005-vertica-bulk-load-partial-row.hpl
@@ -0,0 +1,118 @@
+<?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>0005-vertica-bulk-load-partial-row</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Bulk loads without "Specify database fields" while the stream
only covers part of
+ the table. The columns that are not on the stream keep their
default.</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/21 10:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/21 10:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>partial invoices</from>
+ <to>Vertica bulk loader</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>partial invoices</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>EUR</item>
+ <item>10</item>
+ </line>
+ <line>
+ <item>USD</item>
+ <item>20</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>cost_currency</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>invoice_id</name>
+ <type>Integer</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Vertica bulk loader</name>
+ <type>VerticaBulkLoader</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <abort_on_error>Y</abort_on_error>
+ <connection>vertica</connection>
+ <direct>Y</direct>
+ <fields>
+ </fields>
+ <specify_fields>N</specify_fields>
+ <table>partial_invoices</table>
+ <attributes/>
+ <GUI>
+ <xloc>336</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/vertica/disabled.txt
b/integration-tests/vertica/disabled.txt
deleted file mode 100644
index 7560e5b3bc..0000000000
--- a/integration-tests/vertica/disabled.txt
+++ /dev/null
@@ -1,17 +0,0 @@
-#
-# 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.
-#
-#
\ No newline at end of file
diff --git a/integration-tests/vertica/main-0000-wait.hwf
b/integration-tests/vertica/main-0000-wait.hwf
deleted file mode 100644
index 4da445be00..0000000000
--- a/integration-tests/vertica/main-0000-wait.hwf
+++ /dev/null
@@ -1,76 +0,0 @@
-<?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-0000-wait</name>
- <name_sync_with_filename>Y</name_sync_with_filename>
- <description/>
- <extended_description/>
- <workflow_version/>
- <created_user>-</created_user>
- <created_date>2022/12/21 12:51:47.205</created_date>
- <modified_user>-</modified_user>
- <modified_date>2022/12/21 12:51:47.205</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>wait 30s</name>
- <description/>
- <type>DELAY</type>
- <attributes/>
- <maximumTimeout>30</maximumTimeout>
- <scaletime>0</scaletime>
- <parallel>N</parallel>
- <xloc>192</xloc>
- <yloc>48</yloc>
- <attributes_hac/>
- </action>
- </actions>
- <hops>
- <hop>
- <from>Start</from>
- <to>wait 30s</to>
- <enabled>Y</enabled>
- <evaluation>Y</evaluation>
- <unconditional>Y</unconditional>
- </hop>
- </hops>
- <notepads>
- </notepads>
- <attributes/>
-</workflow>
diff --git
a/integration-tests/vertica/main-0004-vertica-bulk-load-no-fields.hwf
b/integration-tests/vertica/main-0004-vertica-bulk-load-no-fields.hwf
new file mode 100644
index 0000000000..63cfd6b0e8
--- /dev/null
+++ b/integration-tests/vertica/main-0004-vertica-bulk-load-no-fields.hwf
@@ -0,0 +1,154 @@
+<?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-0004-vertica-bulk-load-no-fields</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Regression test for issue 4394: the bulk loader has to work
without the
+ "Specify database fields" option, matching stream fields and table columns
on name.</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/08/21 10:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/21 10: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>invoices DDL</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <sql>DROP TABLE IF EXISTS invoices;
+CREATE TABLE IF NOT EXISTS invoices (
+ invoice_id INT
+, invoice_receipt_date DATE
+, cost_currency VARCHAR(3)
+, is_paid BOOLEAN
+, invoice_note VARCHAR(150)
+)
+</sql>
+ <useVariableSubstitution>F</useVariableSubstitution>
+ <sqlfromfile>F</sqlfromfile>
+ <sqlfilename/>
+ <sendOneStatement>F</sendOneStatement>
+ <connection>vertica</connection>
+ <parallel>N</parallel>
+ <xloc>176</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0004-vertica-bulk-load-no-fields.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <filename>${PROJECT_HOME}/0004-vertica-bulk-load-no-fields.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <set_logfile>N</set_logfile>
+ <logfile/>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <loglevel>Basic</loglevel>
+ <set_append_logfile>N</set_append_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <create_parent_folder>N</create_parent_folder>
+ <run_configuration>local</run_configuration>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <parallel>N</parallel>
+ <xloc>352</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Every value landed in its own column</name>
+ <description/>
+ <type>EVAL_TABLE_CONTENT</type>
+ <attributes/>
+ <connection>vertica</connection>
+ <schemaname/>
+ <tablename>invoices</tablename>
+ <success_condition>rows_count_equal</success_condition>
+ <limit>3</limit>
+ <is_custom_sql>Y</is_custom_sql>
+ <is_usevars>N</is_usevars>
+ <custom_sql>SELECT invoice_id FROM invoices
+WHERE (invoice_id = 1 AND invoice_receipt_date = '2024-03-15' AND
cost_currency = 'EUR' AND is_paid = true AND invoice_note = 'first invoice')
+ OR (invoice_id = 2 AND invoice_receipt_date = '2024-04-01' AND
cost_currency = 'USD' AND is_paid = false AND invoice_note = 'second invoice')
+ OR (invoice_id = 3 AND invoice_receipt_date = '2024-12-31' AND
cost_currency = 'GBP' AND is_paid = true AND invoice_note = 'third
invoice')</custom_sql>
+ <add_rows_result>N</add_rows_result>
+ <clear_result_rows>Y</clear_result_rows>
+ <parallel>N</parallel>
+ <xloc>592</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>invoices DDL</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>invoices DDL</from>
+ <to>0004-vertica-bulk-load-no-fields.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0004-vertica-bulk-load-no-fields.hpl</from>
+ <to>Every value landed in its own column</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/integration-tests/vertica/main-0005-vertica-bulk-load-partial-row.hwf
b/integration-tests/vertica/main-0005-vertica-bulk-load-partial-row.hwf
new file mode 100644
index 0000000000..ffdd63e8ec
--- /dev/null
+++ b/integration-tests/vertica/main-0005-vertica-bulk-load-partial-row.hwf
@@ -0,0 +1,152 @@
+<?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-0005-vertica-bulk-load-partial-row</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Without "Specify database fields" the stream does not have to
cover the whole
+ table: the columns that are missing keep their default.</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/08/21 10:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/21 10: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>partial_invoices DDL</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <sql>DROP TABLE IF EXISTS partial_invoices;
+CREATE TABLE IF NOT EXISTS partial_invoices (
+ invoice_id INT
+, invoice_receipt_date DATE
+, cost_currency VARCHAR(3)
+, invoice_note VARCHAR(150)
+)
+</sql>
+ <useVariableSubstitution>F</useVariableSubstitution>
+ <sqlfromfile>F</sqlfromfile>
+ <sqlfilename/>
+ <sendOneStatement>F</sendOneStatement>
+ <connection>vertica</connection>
+ <parallel>N</parallel>
+ <xloc>176</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0005-vertica-bulk-load-partial-row.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+
<filename>${PROJECT_HOME}/0005-vertica-bulk-load-partial-row.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <set_logfile>N</set_logfile>
+ <logfile/>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <loglevel>Basic</loglevel>
+ <set_append_logfile>N</set_append_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <create_parent_folder>N</create_parent_folder>
+ <run_configuration>local</run_configuration>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <parallel>N</parallel>
+ <xloc>352</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Only the streamed columns are filled in</name>
+ <description/>
+ <type>EVAL_TABLE_CONTENT</type>
+ <attributes/>
+ <connection>vertica</connection>
+ <schemaname/>
+ <tablename>partial_invoices</tablename>
+ <success_condition>rows_count_equal</success_condition>
+ <limit>2</limit>
+ <is_custom_sql>Y</is_custom_sql>
+ <is_usevars>N</is_usevars>
+ <custom_sql>SELECT invoice_id FROM partial_invoices
+WHERE invoice_receipt_date IS NULL AND invoice_note IS NULL
+ AND ((invoice_id = 10 AND cost_currency = 'EUR') OR (invoice_id = 20 AND
cost_currency = 'USD'))</custom_sql>
+ <add_rows_result>N</add_rows_result>
+ <clear_result_rows>Y</clear_result_rows>
+ <parallel>N</parallel>
+ <xloc>592</xloc>
+ <yloc>48</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>partial_invoices DDL</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>partial_invoices DDL</from>
+ <to>0005-vertica-bulk-load-partial-row.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0005-vertica-bulk-load-partial-row.hpl</from>
+ <to>Only the streamed columns are filled in</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoader.java
b/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoader.java
index 5e0dbe933f..42f9fc1226 100644
---
a/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoader.java
+++
b/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoader.java
@@ -35,6 +35,7 @@ import java.util.List;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import org.apache.commons.dbcp2.DelegatingConnection;
+import org.apache.hop.core.Const;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.exception.HopDatabaseException;
@@ -61,9 +62,6 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
private static final SimpleDateFormat SIMPLE_DATE_FORMAT =
new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
- public static final String CONST_FIELD = "Field ";
- public static final String
CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN =
- " must be a Date compatible type to match target column ";
private FileOutputStream exceptionLog;
private FileOutputStream rejectedLog;
@@ -107,64 +105,7 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
IRowMeta tableMeta = meta.getRequiredFields(variables);
- if (!meta.specifyFields()) {
-
- // Just take the whole input row
- data.insertRowMeta = getInputRowMeta().clone();
- data.selectedRowFieldIndices = new int[data.insertRowMeta.size()];
-
- data.colSpecs = new ArrayList<>(data.insertRowMeta.size());
-
- for (int insertFieldIdx = 0; insertFieldIdx <
data.insertRowMeta.size(); insertFieldIdx++) {
- data.selectedRowFieldIndices[insertFieldIdx] = insertFieldIdx;
- IValueMeta inputValueMeta =
data.insertRowMeta.getValueMeta(insertFieldIdx);
- IValueMeta insertValueMeta = inputValueMeta.clone();
- IValueMeta targetValueMeta = tableMeta.getValueMeta(insertFieldIdx);
- insertValueMeta.setName(targetValueMeta.getName());
- data.insertRowMeta.setValueMeta(insertFieldIdx, insertValueMeta);
- ColumnSpec cs = getColumnSpecFromField(inputValueMeta,
insertValueMeta, targetValueMeta);
- data.colSpecs.add(insertFieldIdx, cs);
- }
-
- } else {
-
- int numberOfInsertFields = meta.getFields().size();
- data.insertRowMeta = new RowMeta();
- data.colSpecs = new ArrayList<>(numberOfInsertFields);
-
- // Cache the position of the selected fields in the row array
- data.selectedRowFieldIndices = new int[numberOfInsertFields];
- for (int insertFieldIdx = 0; insertFieldIdx < numberOfInsertFields;
insertFieldIdx++) {
- VerticaBulkLoaderField vbf = meta.getFields().get(insertFieldIdx);
- String inputFieldName = vbf.getFieldStream();
- int inputFieldIdx = getInputRowMeta().indexOfValue(inputFieldName);
- if (inputFieldIdx < 0) {
- throw new HopTransformException(
- BaseMessages.getString(
- PKG,
- "VerticaBulkLoader.Exception.FieldRequired",
- inputFieldName)); //$NON-NLS-1$
- }
- data.selectedRowFieldIndices[insertFieldIdx] = inputFieldIdx;
-
- String insertFieldName = vbf.getFieldDatabase();
- IValueMeta inputValueMeta =
getInputRowMeta().getValueMeta(inputFieldIdx);
- if (inputValueMeta == null) {
- throw new HopTransformException(
- BaseMessages.getString(
- PKG,
- "VerticaBulkLoader.Exception.FailedToFindField",
- vbf.getFieldStream())); // $NON-NLS-1$
- }
- IValueMeta insertValueMeta = inputValueMeta.clone();
- insertValueMeta.setName(insertFieldName);
- data.insertRowMeta.addValueMeta(insertValueMeta);
-
- IValueMeta targetValueMeta =
tableMeta.searchValueMeta(insertFieldName);
- ColumnSpec cs = getColumnSpecFromField(inputValueMeta,
insertValueMeta, targetValueMeta);
- data.colSpecs.add(insertFieldIdx, cs);
- }
- }
+ prepareFieldMapping(getInputRowMeta(), tableMeta);
try {
data.pipedInputStream = new PipedInputStream();
@@ -292,8 +233,103 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
}
}
+ /**
+ * Works out which incoming field feeds which column of the target table and
how every value has
+ * to be encoded in Vertica's native binary format.
+ *
+ * <p>When the database fields are not specified explicitly the complete
input row is loaded and
+ * the target column is looked up <b>by name</b>. The order in which the
columns happen to be
+ * defined in the table is unrelated to the order of the fields on the
stream, so matching them by
+ * position (as this transform used to do) silently loads values into the
wrong columns or fails
+ * with a confusing type error.
+ *
+ * @param inputRowMeta the layout of the rows arriving at this transform
+ * @param tableMeta the layout of the target table
+ */
+ @VisibleForTesting
+ void prepareFieldMapping(IRowMeta inputRowMeta, IRowMeta tableMeta) throws
HopException {
+ if (!meta.specifyFields()) {
+
+ // Just take the whole input row and match the columns on name.
+ data.insertRowMeta = inputRowMeta.clone();
+ data.selectedRowFieldIndices = new int[data.insertRowMeta.size()];
+
+ data.colSpecs = new ArrayList<>(data.insertRowMeta.size());
+
+ for (int insertFieldIdx = 0; insertFieldIdx < data.insertRowMeta.size();
insertFieldIdx++) {
+ data.selectedRowFieldIndices[insertFieldIdx] = insertFieldIdx;
+ IValueMeta inputValueMeta =
data.insertRowMeta.getValueMeta(insertFieldIdx);
+ IValueMeta targetValueMeta = findTargetColumn(tableMeta,
inputValueMeta.getName());
+ IValueMeta insertValueMeta = inputValueMeta.clone();
+ // Take over the name the way the database spells it: the COPY
statement is built from it.
+ insertValueMeta.setName(targetValueMeta.getName());
+ data.insertRowMeta.setValueMeta(insertFieldIdx, insertValueMeta);
+ ColumnSpec cs = getColumnSpecFromField(inputValueMeta,
insertValueMeta, targetValueMeta);
+ data.colSpecs.add(insertFieldIdx, cs);
+ }
+
+ } else {
+
+ int numberOfInsertFields = meta.getFields().size();
+ data.insertRowMeta = new RowMeta();
+ data.colSpecs = new ArrayList<>(numberOfInsertFields);
+
+ // Cache the position of the selected fields in the row array
+ data.selectedRowFieldIndices = new int[numberOfInsertFields];
+ for (int insertFieldIdx = 0; insertFieldIdx < numberOfInsertFields;
insertFieldIdx++) {
+ VerticaBulkLoaderField vbf = meta.getFields().get(insertFieldIdx);
+ String inputFieldName = vbf.getFieldStream();
+ int inputFieldIdx = inputRowMeta.indexOfValue(inputFieldName);
+ if (inputFieldIdx < 0) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG, "VerticaBulkLoader.Exception.FieldRequired",
inputFieldName)); // $NON-NLS-1$
+ }
+ data.selectedRowFieldIndices[insertFieldIdx] = inputFieldIdx;
+
+ String insertFieldName = vbf.getFieldDatabase();
+ IValueMeta inputValueMeta = inputRowMeta.getValueMeta(inputFieldIdx);
+ if (inputValueMeta == null) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "VerticaBulkLoader.Exception.FailedToFindField",
+ vbf.getFieldStream())); // $NON-NLS-1$
+ }
+
+ IValueMeta targetValueMeta = findTargetColumn(tableMeta,
insertFieldName);
+ IValueMeta insertValueMeta = inputValueMeta.clone();
+ insertValueMeta.setName(targetValueMeta.getName());
+ data.insertRowMeta.addValueMeta(insertValueMeta);
+
+ ColumnSpec cs = getColumnSpecFromField(inputValueMeta,
insertValueMeta, targetValueMeta);
+ data.colSpecs.add(insertFieldIdx, cs);
+ }
+ }
+ }
+
+ /**
+ * Looks up a column in the target table, case insensitively, and reports a
usable error when it
+ * simply isn't there.
+ */
+ private IValueMeta findTargetColumn(IRowMeta tableMeta, String columnName)
+ throws HopTransformException {
+ IValueMeta targetValueMeta = tableMeta.searchValueMeta(columnName);
+ if (targetValueMeta == null) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "VerticaBulkLoader.Exception.ColumnNotFoundInTable",
+ Const.NVL(columnName, ""),
+ resolve(meta.getTableName()),
+ String.join(", ", tableMeta.getFieldNames())));
+ }
+ return targetValueMeta;
+ }
+
private ColumnSpec getColumnSpecFromField(
- IValueMeta inputValueMeta, IValueMeta insertValueMeta, IValueMeta
targetValueMeta) {
+ IValueMeta inputValueMeta, IValueMeta insertValueMeta, IValueMeta
targetValueMeta)
+ throws HopTransformException {
if (isBasic()) {
logBasic(
"Mapping input field "
@@ -308,9 +344,18 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
+ ") ");
}
+ if (targetValueMeta.getOriginalColumnTypeName() == null) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "VerticaBulkLoader.Exception.UnknownColumnType",
+ insertValueMeta.getName())); // $NON-NLS-1$
+ }
+
String targetColumnTypeName =
targetValueMeta.getOriginalColumnTypeName().toUpperCase();
switch (targetColumnTypeName) {
+ // Vertica reports every integer column as "Integer"; BIGINT is a SQL
standard synonym.
case "INTEGER", "BIGINT" -> {
return new ColumnSpec(ColumnSpec.ConstantWidthType.INTEGER_64);
}
@@ -323,79 +368,39 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
case "CHAR" -> {
return new ColumnSpec(ColumnSpec.UserDefinedWidthType.CHAR,
targetValueMeta.getLength());
}
- case "VARCHAR", "CHARACTER VARYING" -> {
+ case "VARCHAR", "CHARACTER VARYING", "LONG VARCHAR" -> {
return new ColumnSpec(ColumnSpec.VariableWidthType.VARCHAR,
targetValueMeta.getLength());
}
case "DATE" -> {
- if (!inputValueMeta.isDate()) {
- throw new IllegalArgumentException(
- CONST_FIELD
- + inputValueMeta.getName()
- + CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN
- + insertValueMeta.getName());
- } else {
- return new ColumnSpec(ColumnSpec.ConstantWidthType.DATE);
- }
+ requireDateCompatibleInput(inputValueMeta, insertValueMeta,
targetColumnTypeName);
+ return new ColumnSpec(ColumnSpec.ConstantWidthType.DATE);
}
case "TIME" -> {
- if (!inputValueMeta.isDate()) {
- throw new IllegalArgumentException(
- CONST_FIELD
- + inputValueMeta.getName()
- + CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN
- + insertValueMeta.getName());
- } else {
- return new ColumnSpec(ColumnSpec.ConstantWidthType.TIME);
- }
+ requireDateCompatibleInput(inputValueMeta, insertValueMeta,
targetColumnTypeName);
+ return new ColumnSpec(ColumnSpec.ConstantWidthType.TIME);
}
case "TIMETZ" -> {
- if (!inputValueMeta.isDate()) {
- throw new IllegalArgumentException(
- CONST_FIELD
- + inputValueMeta.getName()
- + CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN
- + insertValueMeta.getName());
- } else {
- return new ColumnSpec(ColumnSpec.ConstantWidthType.TIMETZ);
- }
+ requireDateCompatibleInput(inputValueMeta, insertValueMeta,
targetColumnTypeName);
+ return new ColumnSpec(ColumnSpec.ConstantWidthType.TIMETZ);
}
case "TIMESTAMP" -> {
- if (!inputValueMeta.isDate()) {
- throw new IllegalArgumentException(
- CONST_FIELD
- + inputValueMeta.getName()
- + CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN
- + insertValueMeta.getName());
- } else {
- return new ColumnSpec(ColumnSpec.ConstantWidthType.TIMESTAMP);
- }
+ requireDateCompatibleInput(inputValueMeta, insertValueMeta,
targetColumnTypeName);
+ return new ColumnSpec(ColumnSpec.ConstantWidthType.TIMESTAMP);
}
case "TIMESTAMPTZ" -> {
- if (!inputValueMeta.isDate()) {
- throw new IllegalArgumentException(
- CONST_FIELD
- + inputValueMeta.getName()
- + CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN
- + insertValueMeta.getName());
- } else {
- return new ColumnSpec(ColumnSpec.ConstantWidthType.TIMESTAMPTZ);
- }
+ requireDateCompatibleInput(inputValueMeta, insertValueMeta,
targetColumnTypeName);
+ return new ColumnSpec(ColumnSpec.ConstantWidthType.TIMESTAMPTZ);
}
case "INTERVAL", "INTERVAL DAY TO SECOND" -> {
- if (!inputValueMeta.isDate()) {
- throw new IllegalArgumentException(
- CONST_FIELD
- + inputValueMeta.getName()
- + CONST_MUST_BE_A_DATE_COMPATIBLE_TYPE_TO_MATCH_TARGET_COLUMN
- + insertValueMeta.getName());
- } else {
- return new ColumnSpec(ColumnSpec.ConstantWidthType.INTERVAL);
- }
+ requireDateCompatibleInput(inputValueMeta, insertValueMeta,
targetColumnTypeName);
+ return new ColumnSpec(ColumnSpec.ConstantWidthType.INTERVAL);
}
+ // BINARY is a fixed width type: the value is padded with zeroes up to
the column width,
+ // unlike VARBINARY which is written with a length prefix.
case "BINARY" -> {
- return new ColumnSpec(ColumnSpec.VariableWidthType.VARBINARY,
targetValueMeta.getLength());
+ return new ColumnSpec(ColumnSpec.UserDefinedWidthType.BINARY,
targetValueMeta.getLength());
}
- case "VARBINARY" -> {
+ case "VARBINARY", "LONG VARBINARY" -> {
return new ColumnSpec(ColumnSpec.VariableWidthType.VARBINARY,
targetValueMeta.getLength());
}
case "NUMERIC" -> {
@@ -404,9 +409,28 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
targetValueMeta.getLength(),
targetValueMeta.getPrecision());
}
+ default ->
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "VerticaBulkLoader.Exception.ColumnTypeNotSupported",
+ targetColumnTypeName,
+ insertValueMeta.getName())); // $NON-NLS-1$
+ }
+ }
+
+ private void requireDateCompatibleInput(
+ IValueMeta inputValueMeta, IValueMeta insertValueMeta, String
targetColumnTypeName)
+ throws HopTransformException {
+ if (!inputValueMeta.isDate()) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "VerticaBulkLoader.Exception.FieldMustBeDateCompatible",
+ inputValueMeta.getName(),
+ insertValueMeta.getName(),
+ targetColumnTypeName)); // $NON-NLS-1$
}
- throw new IllegalArgumentException(
- "Column type " + targetColumnTypeName + " not supported."); //
$NON-NLS-1$
}
private void initializeWorker() {
@@ -448,7 +472,8 @@ public class VerticaBulkLoader extends
BaseTransform<VerticaBulkLoaderMeta, Vert
data.workerThread.start();
}
- private String buildCopyStatementSqlString() {
+ @VisibleForTesting
+ String buildCopyStatementSqlString() {
final DatabaseMeta databaseMeta = data.db.getDatabaseMeta();
StringBuilder sb = new StringBuilder(150);
diff --git
a/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderDialog.java
b/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderDialog.java
index 02302b79cd..a5394af1f4 100644
---
a/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderDialog.java
+++
b/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderDialog.java
@@ -360,7 +360,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
PropsUi.setLook(wDirect);
fdDirect = new FormData();
fdDirect.left = new FormAttachment(middle, 0);
- fdDirect.top = new FormAttachment(0, margin);
+ fdDirect.top = new FormAttachment(wlDirect, 0, SWT.CENTER);
fdDirect.right = new FormAttachment(100, 0);
wDirect.setLayoutData(fdDirect);
wDirect.addSelectionListener(lsSelMod);
@@ -375,7 +375,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
PropsUi.setLook(wlAbortOnError);
fdlAbortOnError = new FormData();
fdlAbortOnError.left = new FormAttachment(0, 0);
- fdlAbortOnError.top = new FormAttachment(wlDirect, margin);
+ fdlAbortOnError.top = new FormAttachment(wDirect, margin);
fdlAbortOnError.right = new FormAttachment(middle, -margin);
wlAbortOnError.setLayoutData(fdlAbortOnError);
wAbortOnError = new Button(wMainComp, SWT.CHECK);
@@ -384,7 +384,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
PropsUi.setLook(wAbortOnError);
fdAbortOnError = new FormData();
fdAbortOnError.left = new FormAttachment(middle, 0);
- fdAbortOnError.top = new FormAttachment(wlDirect, margin);
+ fdAbortOnError.top = new FormAttachment(wlAbortOnError, 0, SWT.CENTER);
fdAbortOnError.right = new FormAttachment(100, 0);
wAbortOnError.setLayoutData(fdAbortOnError);
wAbortOnError.addSelectionListener(lsSelMod);
@@ -402,7 +402,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
fdlExceptionsLogFile = new FormData();
fdlExceptionsLogFile.left = new FormAttachment(0, 0);
fdlExceptionsLogFile.right = new FormAttachment(middle, -margin);
- fdlExceptionsLogFile.top = new FormAttachment(wlAbortOnError, margin);
+ fdlExceptionsLogFile.top = new FormAttachment(wAbortOnError, margin);
wlExceptionsLogFile.setLayoutData(fdlExceptionsLogFile);
wExceptionsLogFile = new TextVar(variables, wMainComp, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
@@ -414,7 +414,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
wExceptionsLogFile.addFocusListener(lsFocusLost);
fdExceptionsLogFile = new FormData();
fdExceptionsLogFile.left = new FormAttachment(middle, 0);
- fdExceptionsLogFile.top = new FormAttachment(wlAbortOnError, margin);
+ fdExceptionsLogFile.top = new FormAttachment(wAbortOnError, margin);
fdExceptionsLogFile.right = new FormAttachment(100, 0);
wExceptionsLogFile.setLayoutData(fdExceptionsLogFile);
@@ -430,7 +430,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
fdlRejectedDataLogFile = new FormData();
fdlRejectedDataLogFile.left = new FormAttachment(0, 0);
fdlRejectedDataLogFile.right = new FormAttachment(middle, -margin);
- fdlRejectedDataLogFile.top = new FormAttachment(wlExceptionsLogFile,
margin);
+ fdlRejectedDataLogFile.top = new FormAttachment(wExceptionsLogFile,
margin);
wlRejectedDataLogFile.setLayoutData(fdlRejectedDataLogFile);
wRejectedDataLogFile = new TextVar(variables, wMainComp, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
@@ -442,7 +442,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
wRejectedDataLogFile.addFocusListener(lsFocusLost);
fdRejectedDataLogFile = new FormData();
fdRejectedDataLogFile.left = new FormAttachment(middle, 0);
- fdRejectedDataLogFile.top = new FormAttachment(wlExceptionsLogFile,
margin);
+ fdRejectedDataLogFile.top = new FormAttachment(wExceptionsLogFile, margin);
fdRejectedDataLogFile.right = new FormAttachment(100, 0);
wRejectedDataLogFile.setLayoutData(fdRejectedDataLogFile);
@@ -456,7 +456,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
FormData fdlStreamName = new FormData();
fdlStreamName.left = new FormAttachment(0, 0);
fdlStreamName.right = new FormAttachment(middle, -margin);
- fdlStreamName.top = new FormAttachment(wlRejectedDataLogFile, margin);
+ fdlStreamName.top = new FormAttachment(wRejectedDataLogFile, margin);
wlStreamName.setLayoutData(fdlStreamName);
wStreamName = new TextVar(variables, wMainComp, SWT.SINGLE | SWT.LEFT |
SWT.BORDER);
@@ -467,7 +467,7 @@ public class VerticaBulkLoaderDialog extends
BaseTransformDialog {
wStreamName.addFocusListener(lsFocusLost);
fdStreamName = new FormData();
fdStreamName.left = new FormAttachment(middle, 0);
- fdStreamName.top = new FormAttachment(wlRejectedDataLogFile, margin);
+ fdStreamName.top = new FormAttachment(wRejectedDataLogFile, margin);
fdStreamName.right = new FormAttachment(100, 0);
wStreamName.setLayoutData(fdStreamName);
diff --git
a/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMeta.java
b/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMeta.java
index e7778711f4..3a94da41a2 100644
---
a/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMeta.java
+++
b/plugins/databases/vertica/src/main/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMeta.java
@@ -148,16 +148,6 @@ public class VerticaBulkLoaderMeta
this.fields = fields;
}
- @HopMetadataProperty(
- groupKey = "fields",
- key = "field",
- injectionGroupKey = "FIELDS",
- injectionGroupDescription = "VerticaBulkLoader.Injection.FIELDS",
- injectionKey = "FIELDDATABASE",
- injectionKeyDescription = "VerticaBulkLoader.Injection.FIELDDATABASE")
- /** Fields in the table to insert */
- private String[] fieldDatabase;
-
public VerticaBulkLoaderMeta() {
super(); // allocate BaseTransformMeta
@@ -428,10 +418,10 @@ public class VerticaBulkLoaderMeta
}
} else {
// Specifying the column names explicitly
- for (int i = 0; i < getFieldDatabase().length; i++) {
- int idx = r.indexOfValue(getFieldDatabase()[i]);
+ for (VerticaBulkLoaderField vbf : fields) {
+ int idx = r.indexOfValue(vbf.getFieldDatabase());
if (idx < 0) {
- error_message += "\t\t" + getFieldDatabase()[i] +
Const.CR;
+ error_message += "\t\t" + vbf.getFieldDatabase() +
Const.CR;
error_found = true;
}
}
@@ -460,7 +450,7 @@ public class VerticaBulkLoaderMeta
error_message = "";
if (!specifyFields()) {
// Starting from table fields in r...
- for (int i = 0; i < getFieldDatabase().length; i++) {
+ for (int i = 0; i < r.size(); i++) {
IValueMeta rv = r.getValueMeta(i);
int idx = prev.indexOfValue(rv.getName());
if (idx < 0) {
@@ -733,20 +723,6 @@ public class VerticaBulkLoaderMeta
}
}
- /**
- * @return Fields containing the fieldnames in the database insert.
- */
- public String[] getFieldDatabase() {
- return fieldDatabase;
- }
-
- /**
- * @param fieldDatabase The fields containing the names of the fields to
insert.
- */
- public void setFieldDatabase(String[] fieldDatabase) {
- this.fieldDatabase = fieldDatabase;
- }
-
/**
* Returns the schema name.
*
diff --git
a/plugins/databases/vertica/src/main/resources/org/apache/hop/pipeline/transforms/vertica/bulkloader/messages/messages_en_US.properties
b/plugins/databases/vertica/src/main/resources/org/apache/hop/pipeline/transforms/vertica/bulkloader/messages/messages_en_US.properties
index 222e6a2c7d..9e02c1d9d8 100644
---
a/plugins/databases/vertica/src/main/resources/org/apache/hop/pipeline/transforms/vertica/bulkloader/messages/messages_en_US.properties
+++
b/plugins/databases/vertica/src/main/resources/org/apache/hop/pipeline/transforms/vertica/bulkloader/messages/messages_en_US.properties
@@ -19,9 +19,13 @@
BaseTransform.TypeLongDesc.VerticaBulkLoaderMessage=Vertica bulk loader
BaseTransform.TypeTooltipDesc.VerticaBulkLoaderMessage=Bulk load data into a
Vertica database table
VerticaBulkLoader.Exception.ClosingLogError=Unable to close Log Files
+VerticaBulkLoader.Exception.ColumnNotFoundInTable=Field [{0}] has no matching
column in table [{1}]. Available columns\: {2}
+VerticaBulkLoader.Exception.ColumnTypeNotSupported=Column type [{0}] of target
column [{1}] is not supported by the Vertica bulk loader
VerticaBulkLoader.Exception.FailedToFindField=Could not find field {0} in
stream
+VerticaBulkLoader.Exception.FieldMustBeDateCompatible=Field [{0}] must be of a
date compatible type to match target column [{1}] of type [{2}]
VerticaBulkLoader.Exception.FieldRequired=Field [{0}] is required and
couldn''t be found\!
VerticaBulkLoader.Exception.RowRejected=Row Rejected\: {0}
+VerticaBulkLoader.Exception.UnknownColumnType=The database did not report a
column type for target column [{0}]
VerticaBulkLoader.Inject.OnlyWhenHaveRows.Field=Truncate on first row
VerticaBulkLoader.Injection.ABORTONERROR=Set this option to abort and rollback
data loading upon an error.
VerticaBulkLoader.Injection.CONNECTIONNAME=The name of the database connection
to get table names from.
diff --git
a/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMetaTest.java
b/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMetaTest.java
new file mode 100644
index 0000000000..2f11fef5c6
--- /dev/null
+++
b/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderMetaTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.vertica.bulkloader;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import org.apache.hop.core.injection.bean.BeanInjectionInfo;
+import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
+import org.junit.jupiter.api.Test;
+
+class VerticaBulkLoaderMetaTest {
+
+ @Test
+ void readsAllOptionsFromXml() throws Exception {
+ VerticaBulkLoaderMeta meta =
+ TransformSerializationTestUtil.testSerialization(
+ "/vertica-bulkloader-transform.xml", VerticaBulkLoaderMeta.class);
+
+ assertEquals("vertica", meta.getConnection());
+ assertEquals("staging", meta.getSchemaName());
+ assertEquals("INVOICE", meta.getTableName());
+ assertTrue(meta.isTruncateTable());
+ assertTrue(meta.isOnlyWhenHaveRows());
+ assertTrue(meta.isDirect());
+ assertFalse(meta.isAbortOnError());
+ assertEquals("/tmp/exceptions.log", meta.getExceptionsFileName());
+ assertEquals("/tmp/rejected.log", meta.getRejectedDataFileName());
+ assertEquals("nightly load", meta.getStreamName());
+ assertTrue(meta.isSpecifyFields());
+ assertEquals(
+ List.of(
+ new VerticaBulkLoaderField("INVOICE_ID", "id"),
+ new VerticaBulkLoaderField("COST_CURRENCY", "currency")),
+ meta.getFields());
+ }
+
+ @Test
+ void defaultsToLoadingTheWholeInputRow() {
+ VerticaBulkLoaderMeta meta = new VerticaBulkLoaderMeta();
+ meta.setDefault();
+
+ assertFalse(meta.specifyFields());
+ assertTrue(meta.getFields().isEmpty());
+ }
+
+ @Test
+ void exposesTheStreamAndColumnNamesToMetadataInjection() {
+ // The column names used to be declared twice: once on the field list and
once on a String[]
+ // that was never filled in, which left the target column unreachable for
injection.
+ BeanInjectionInfo<VerticaBulkLoaderMeta> info =
+ new BeanInjectionInfo<>(VerticaBulkLoaderMeta.class);
+
+ assertTrue(info.getProperties().containsKey("STREAM_FIELDNAME"));
+ assertTrue(info.getProperties().containsKey("DATABASE_FIELDNAME"));
+ assertEquals("FIELDS",
info.getProperties().get("DATABASE_FIELDNAME").getGroupKey());
+ }
+}
diff --git
a/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderTest.java
b/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderTest.java
new file mode 100644
index 0000000000..4f57e2f746
--- /dev/null
+++
b/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/VerticaBulkLoaderTest.java
@@ -0,0 +1,475 @@
+/*
+ * 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.vertica.bulkloader;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+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.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import org.apache.hop.core.database.Database;
+import org.apache.hop.core.database.DatabaseMeta;
+import org.apache.hop.core.database.DatabasePluginType;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.logging.HopLogStore;
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.core.plugins.PluginRegistry;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBigNumber;
+import org.apache.hop.core.row.value.ValueMetaBinary;
+import org.apache.hop.core.row.value.ValueMetaDate;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaPluginType;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import
org.apache.hop.pipeline.transforms.vertica.bulkloader.nativebinary.ColumnSpec;
+import
org.apache.hop.pipeline.transforms.vertica.bulkloader.nativebinary.ColumnType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * Covers the way the Vertica bulk loader lines up the fields on the stream
with the columns of the
+ * target table, both with and without the "Specify database fields" option.
+ */
+class VerticaBulkLoaderTest {
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
+ private TransformMockHelper<VerticaBulkLoaderMeta, VerticaBulkLoaderData>
mockHelper;
+ private VerticaBulkLoaderMeta meta;
+ private VerticaBulkLoaderData data;
+ private VerticaBulkLoader transform;
+
+ @BeforeAll
+ static void setUpBeforeClass() throws HopException {
+ PluginRegistry.addPluginType(ValueMetaPluginType.getInstance());
+ PluginRegistry.addPluginType(DatabasePluginType.getInstance());
+ PluginRegistry.init();
+ HopLogStore.init();
+ }
+
+ @BeforeEach
+ void setUp() {
+ mockHelper =
+ new TransformMockHelper<>(
+ "Vertica bulk loader", VerticaBulkLoaderMeta.class,
VerticaBulkLoaderData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+
when(mockHelper.logChannelFactory.create(any())).thenReturn(mockHelper.iLogChannel);
+
+ meta = new VerticaBulkLoaderMeta();
+ meta.setConnection("vertica");
+ meta.setTableName("INVOICE");
+ data = new VerticaBulkLoaderData();
+
+ transform =
+ new VerticaBulkLoader(
+ mockHelper.transformMeta, meta, data, 0, mockHelper.pipelineMeta,
mockHelper.pipeline);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ /**
+ * The table as the database reports it. Note that the column order
deliberately differs from the
+ * order of the fields on the stream.
+ */
+ private static IRowMeta invoiceTable() {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaInteger("INVOICE_ID"), "INTEGER",
0, 0));
+ table.addValueMeta(column(new ValueMetaDate("INVOICE_RECEIPT_DATE"),
"DATE", 0, 0));
+ table.addValueMeta(column(new ValueMetaString("COST_CURRENCY"), "VARCHAR",
3, 0));
+ return table;
+ }
+
+ private static IValueMeta column(
+ IValueMeta valueMeta, String columnTypeName, int length, int precision) {
+ valueMeta.setOriginalColumnTypeName(columnTypeName);
+ valueMeta.setLength(length);
+ valueMeta.setPrecision(precision);
+ return valueMeta;
+ }
+
+ //
----------------------------------------------------------------------------------------------
+ // Without "Specify database fields": the whole input row is loaded, matched
on name.
+ //
----------------------------------------------------------------------------------------------
+
+ @Test
+ void matchesColumnsByNameAndNotByPositionWhenFieldsAreNotSpecified() throws
Exception {
+ // The exact scenario of issue #4394: a string field sits on the position
that the table uses
+ // for a date column. Matching on position made this fail with "Field
COST_CURRENCY must be a
+ // Date compatible type to match target column INVOICE_RECEIPT_DATE".
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaInteger("INVOICE_ID"));
+ input.addValueMeta(new ValueMetaString("COST_CURRENCY"));
+ input.addValueMeta(new ValueMetaDate("INVOICE_RECEIPT_DATE"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+
+ assertArrayEquals(new int[] {0, 1, 2}, data.selectedRowFieldIndices);
+ assertArrayEquals(
+ new String[] {"INVOICE_ID", "COST_CURRENCY", "INVOICE_RECEIPT_DATE"},
+ data.insertRowMeta.getFieldNames());
+ assertEquals(
+ List.of(ColumnType.INTEGER, ColumnType.VARCHAR, ColumnType.DATE),
+ data.colSpecs.stream().map(cs -> cs.type).toList());
+ }
+
+ @Test
+ void takesOverTheColumnNameAsTheDatabaseSpellsIt() throws Exception {
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("cost_currency"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+
+ // The COPY statement is generated from these names, so they have to match
the table.
+ assertArrayEquals(new String[] {"COST_CURRENCY"},
data.insertRowMeta.getFieldNames());
+ }
+
+ @Test
+ void loadsAnInputRowThatCoversOnlyPartOfTheTable() throws Exception {
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("COST_CURRENCY"));
+ input.addValueMeta(new ValueMetaInteger("INVOICE_ID"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+
+ assertEquals(2, data.colSpecs.size());
+ assertArrayEquals(
+ new String[] {"COST_CURRENCY", "INVOICE_ID"},
data.insertRowMeta.getFieldNames());
+ }
+
+ @Test
+ void reportsAnInputFieldWithoutAMatchingColumn() {
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("INVOICE_ID"));
+ input.addValueMeta(new ValueMetaString("NOT_A_COLUMN"));
+
+ HopException e =
+ assertThrows(
+ HopTransformException.class,
+ () -> transform.prepareFieldMapping(input, invoiceTable()));
+
+ // Naming the field, the table and the available columns beats an out of
bounds exception.
+ assertTrue(e.getMessage().contains("NOT_A_COLUMN"), e.getMessage());
+ assertTrue(e.getMessage().contains("INVOICE"), e.getMessage());
+ assertTrue(e.getMessage().contains("INVOICE_RECEIPT_DATE"),
e.getMessage());
+ }
+
+ @Test
+ void reportsAnInputRowThatIsWiderThanTheTable() {
+ // Before the fix this walked off the end of the table row meta with a
NullPointerException.
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaInteger("INVOICE_ID"));
+ input.addValueMeta(new ValueMetaDate("INVOICE_RECEIPT_DATE"));
+ input.addValueMeta(new ValueMetaString("COST_CURRENCY"));
+ input.addValueMeta(new ValueMetaString("ONE_TOO_MANY"));
+
+ HopException e =
+ assertThrows(
+ HopTransformException.class,
+ () -> transform.prepareFieldMapping(input, invoiceTable()));
+ assertTrue(e.getMessage().contains("ONE_TOO_MANY"), e.getMessage());
+ }
+
+ //
----------------------------------------------------------------------------------------------
+ // With "Specify database fields".
+ //
----------------------------------------------------------------------------------------------
+
+ @Test
+ void mapsStreamFieldsOntoTheirConfiguredColumns() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(
+ List.of(
+ new VerticaBulkLoaderField("COST_CURRENCY", "currency"),
+ new VerticaBulkLoaderField("INVOICE_ID", "id")));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaInteger("id"));
+ input.addValueMeta(new ValueMetaDate("received"));
+ input.addValueMeta(new ValueMetaString("currency"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+
+ assertArrayEquals(new int[] {2, 0}, data.selectedRowFieldIndices);
+ assertArrayEquals(
+ new String[] {"COST_CURRENCY", "INVOICE_ID"},
data.insertRowMeta.getFieldNames());
+ assertEquals(
+ List.of(ColumnType.VARCHAR, ColumnType.INTEGER),
+ data.colSpecs.stream().map(cs -> cs.type).toList());
+ }
+
+ @Test
+ void reportsAConfiguredColumnThatIsMissingFromTheTable() {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new VerticaBulkLoaderField("NOT_A_COLUMN", "id")));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaInteger("id"));
+
+ // This used to be a NullPointerException on the unknown target column.
+ HopException e =
+ assertThrows(
+ HopTransformException.class,
+ () -> transform.prepareFieldMapping(input, invoiceTable()));
+ assertTrue(e.getMessage().contains("NOT_A_COLUMN"), e.getMessage());
+ }
+
+ @Test
+ void reportsAConfiguredStreamFieldThatIsMissingFromTheInput() {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new VerticaBulkLoaderField("INVOICE_ID",
"not_on_the_stream")));
+
+ HopException e =
+ assertThrows(
+ HopTransformException.class,
+ () -> transform.prepareFieldMapping(new RowMeta(),
invoiceTable()));
+ assertTrue(e.getMessage().contains("not_on_the_stream"), e.getMessage());
+ }
+
+ @Test
+ void usesTheColumnNameOfTheTableRatherThanTheConfiguredCasing() throws
Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new VerticaBulkLoaderField("invoice_id", "id")));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaInteger("id"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+
+ assertArrayEquals(new String[] {"INVOICE_ID"},
data.insertRowMeta.getFieldNames());
+ }
+
+ //
----------------------------------------------------------------------------------------------
+ // Column types.
+ //
----------------------------------------------------------------------------------------------
+
+ @ParameterizedTest
+ @CsvSource({
+ // The names on the left are what the Vertica driver returns from
getColumnTypeName(),
+ // upper cased, plus the SQL standard synonyms this transform has always
accepted.
+ // column type name, expected native binary type, expected width in the
file header
+ "INTEGER, INTEGER, 8",
+ "BIGINT, INTEGER, 8",
+ "BOOLEAN, BOOLEAN, 1",
+ "FLOAT, FLOAT, 8",
+ "DOUBLE PRECISION, FLOAT, 8",
+ "CHAR, CHAR, 12",
+ "VARCHAR, VARCHAR, -1",
+ "CHARACTER VARYING, VARCHAR, -1",
+ "LONG VARCHAR, VARCHAR, -1",
+ "BINARY, BINARY, 12",
+ "VARBINARY, VARBINARY, -1",
+ "LONG VARBINARY, VARBINARY, -1",
+ "NUMERIC, NUMERIC, -1",
+ })
+ void mapsColumnTypesOntoTheNativeBinaryFormat(String columnTypeName, String
type, int bytes)
+ throws Exception {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaString("VALUE"), columnTypeName,
12, 2));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("VALUE"));
+
+ transform.prepareFieldMapping(input, table);
+
+ ColumnSpec spec = data.colSpecs.get(0);
+ assertEquals(ColumnType.valueOf(type), spec.type);
+ assertEquals(bytes, spec.bytes);
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {"DATE", "TIME", "TIMETZ", "TIMESTAMP", "TIMESTAMPTZ",
"INTERVAL DAY TO SECOND"})
+ void mapsTheDateTimeColumnTypes(String columnTypeName) throws Exception {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaDate("MOMENT"), columnTypeName, 0,
0));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaDate("MOMENT"));
+
+ transform.prepareFieldMapping(input, table);
+
+ assertEquals(8, data.colSpecs.get(0).bytes);
+ }
+
+ @Test
+ void keepsThePrecisionAndScaleOfNumericColumns() throws Exception {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaBigNumber("AMOUNT"), "NUMERIC", 18,
4));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaBigNumber("AMOUNT"));
+
+ transform.prepareFieldMapping(input, table);
+
+ ColumnSpec spec = data.colSpecs.get(0);
+ assertEquals(18, spec.getMaxLength());
+ assertEquals(4, spec.scale);
+ }
+
+ @Test
+ void rejectsANonDateFieldForADateColumn() {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaDate("INVOICE_RECEIPT_DATE"),
"DATE", 0, 0));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("INVOICE_RECEIPT_DATE"));
+
+ HopException e =
+ assertThrows(
+ HopTransformException.class, () ->
transform.prepareFieldMapping(input, table));
+ assertTrue(e.getMessage().contains("INVOICE_RECEIPT_DATE"),
e.getMessage());
+ assertTrue(e.getMessage().contains("DATE"), e.getMessage());
+ }
+
+ @Test
+ void rejectsAColumnTypeItCannotEncode() {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaString("KEY"), "UUID", 16, 0));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("KEY"));
+
+ HopException e =
+ assertThrows(
+ HopTransformException.class, () ->
transform.prepareFieldMapping(input, table));
+ assertTrue(e.getMessage().contains("UUID"), e.getMessage());
+ }
+
+ @Test
+ void rejectsAColumnWithoutATypeName() {
+ // A driver that does not report a column type name used to cause a
NullPointerException.
+ RowMeta table = new RowMeta();
+ table.addValueMeta(new ValueMetaString("KEY"));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("KEY"));
+
+ HopException e =
+ assertThrows(
+ HopTransformException.class, () ->
transform.prepareFieldMapping(input, table));
+ assertTrue(e.getMessage().contains("KEY"), e.getMessage());
+ }
+
+ @Test
+ void binaryColumnsAreFixedWidthAndVarbinaryColumnsAreNot() throws Exception {
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaBinary("FIXED"), "BINARY", 10, 0));
+ table.addValueMeta(column(new ValueMetaBinary("VARIABLE"), "VARBINARY",
10, 0));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaBinary("FIXED"));
+ input.addValueMeta(new ValueMetaBinary("VARIABLE"));
+
+ transform.prepareFieldMapping(input, table);
+
+ // Vertica pads a BINARY column up to its declared width, a VARBINARY
carries a length prefix.
+ assertEquals(ColumnType.BINARY, data.colSpecs.get(0).type);
+ assertEquals(10, data.colSpecs.get(0).bytes);
+ assertEquals(ColumnType.VARBINARY, data.colSpecs.get(1).type);
+ assertEquals(-1, data.colSpecs.get(1).bytes);
+ }
+
+ //
----------------------------------------------------------------------------------------------
+ // The generated COPY statement.
+ //
----------------------------------------------------------------------------------------------
+
+ @Test
+ void copyStatementListsTheColumnsInTheOrderOfTheStream() throws Exception {
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaString("COST_CURRENCY"));
+ input.addValueMeta(new ValueMetaInteger("INVOICE_ID"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+ data.db = verticaDatabase();
+
+ String sql = transform.buildCopyStatementSqlString();
+
+ assertTrue(
+ sql.startsWith("COPY INVOICE (\"COST_CURRENCY\", \"INVOICE_ID\") FROM
STDIN NATIVE "), sql);
+ assertTrue(sql.contains("ENFORCELENGTH"), sql);
+ }
+
+ @Test
+ void copyStatementCastsNumericColumnsThroughAFillerColumn() throws Exception
{
+ RowMeta table = new RowMeta();
+ table.addValueMeta(column(new ValueMetaBigNumber("AMOUNT"), "NUMERIC", 18,
4));
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaBigNumber("AMOUNT"));
+
+ transform.prepareFieldMapping(input, table);
+ data.db = verticaDatabase();
+
+ String sql = transform.buildCopyStatementSqlString();
+
+ assertTrue(
+ sql.contains(
+ "(TMPFILLERCOL0 FILLER VARCHAR(1000), \"AMOUNT\" AS
CAST(TMPFILLERCOL0 AS NUMERIC))"),
+ sql);
+ }
+
+ @Test
+ void copyStatementHonoursTheTransformOptions() throws Exception {
+ meta.setSchemaName("staging");
+ meta.setDirect(true);
+ meta.setAbortOnError(true);
+ meta.setStreamName("nightly load");
+
+ RowMeta input = new RowMeta();
+ input.addValueMeta(new ValueMetaInteger("INVOICE_ID"));
+
+ transform.prepareFieldMapping(input, invoiceTable());
+ data.db = verticaDatabase();
+
+ String sql = transform.buildCopyStatementSqlString();
+
+ assertTrue(sql.startsWith("COPY staging.INVOICE "), sql);
+ assertTrue(sql.contains("ABORT ON ERROR "), sql);
+ assertTrue(sql.contains("DIRECT "), sql);
+ assertTrue(sql.contains("STREAM NAME E'nightly load' "), sql);
+ }
+
+ private Database verticaDatabase() {
+ DatabaseMeta databaseMeta =
+ new DatabaseMeta("vertica", "VERTICA", "Native", "localhost", "db",
"5433", "user", "pass");
+ Database db = mock(Database.class);
+ when(db.getDatabaseMeta()).thenReturn(databaseMeta);
+ when(db.resolve(anyString())).thenAnswer(invocation ->
invocation.getArgument(0));
+ return db;
+ }
+}
diff --git
a/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/nativebinary/StreamEncoderTest.java
b/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/nativebinary/StreamEncoderTest.java
new file mode 100644
index 0000000000..95742f9382
--- /dev/null
+++
b/plugins/databases/vertica/src/test/java/org/apache/hop/pipeline/transforms/vertica/bulkloader/nativebinary/StreamEncoderTest.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.vertica.bulkloader.nativebinary;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.IOException;
+import java.io.PipedInputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBinary;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Checks the bytes the encoder puts on the wire against Vertica's native
binary format, in
+ * particular the column widths in the file header: a fixed width type has to
declare its width, a
+ * variable width type declares -1.
+ */
+class StreamEncoderTest {
+
+ private static final byte[] SIGNATURE = {
+ 'N', 'A', 'T', 'I', 'V', 'E', 0x0A, (byte) 0xFF, 0x0D, 0x0A, 0x00
+ };
+
+ @Test
+ void headerDeclaresTheWidthOfEveryColumn() throws IOException {
+ byte[] header =
+ encode(
+ List.of(
+ new ColumnSpec(ColumnSpec.ConstantWidthType.INTEGER_64),
+ new ColumnSpec(ColumnSpec.ConstantWidthType.DATE),
+ new ColumnSpec(ColumnSpec.UserDefinedWidthType.CHAR, 8),
+ new ColumnSpec(ColumnSpec.UserDefinedWidthType.BINARY, 10),
+ new ColumnSpec(ColumnSpec.VariableWidthType.VARCHAR, 255),
+ new ColumnSpec(ColumnSpec.VariableWidthType.VARBINARY, 255),
+ new ColumnSpec(ColumnSpec.PrecisionScaleWidthType.NUMERIC, 18,
4)));
+
+ ByteBuffer buffer = ByteBuffer.wrap(header).order(ByteOrder.LITTLE_ENDIAN);
+
+ byte[] signature = new byte[SIGNATURE.length];
+ buffer.get(signature);
+ assertArrayEquals(SIGNATURE, signature);
+
+ assertEquals(5 + 4 * 7, buffer.getInt(), "header area length");
+ assertEquals(1, buffer.getShort(), "format version");
+ assertEquals(0, buffer.get(), "filler");
+ assertEquals(7, buffer.getShort(), "number of columns");
+
+ int[] widths = new int[7];
+ for (int i = 0; i < widths.length; i++) {
+ widths[i] = buffer.getInt();
+ }
+ // BINARY is fixed width and NUMERIC travels as a VARCHAR filler column,
hence -1.
+ assertArrayEquals(new int[] {8, 8, 8, 10, -1, -1, -1}, widths);
+ }
+
+ @Test
+ void padsFixedWidthBinaryValuesUpToTheColumnWidth() throws Exception {
+ ColumnSpec binary = new ColumnSpec(ColumnSpec.UserDefinedWidthType.BINARY,
4);
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaBinary("PAYLOAD"));
+
+ byte[] written = encodeRow(List.of(binary), rowMeta, new Object[] {new
byte[] {1, 2}});
+
+ ByteBuffer buffer =
ByteBuffer.wrap(written).order(ByteOrder.LITTLE_ENDIAN);
+ assertEquals(4, buffer.getInt(), "row data size");
+ assertEquals(0, buffer.get(), "null bitmap");
+ byte[] payload = new byte[4];
+ buffer.get(payload);
+ assertArrayEquals(new byte[] {1, 2, 0, 0}, payload);
+ }
+
+ @Test
+ void marksMissingValuesInTheNullBitmap() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("A"));
+ rowMeta.addValueMeta(new ValueMetaInteger("B"));
+
+ byte[] written =
+ encodeRow(
+ List.of(
+ new ColumnSpec(ColumnSpec.ConstantWidthType.INTEGER_64),
+ new ColumnSpec(ColumnSpec.ConstantWidthType.INTEGER_64)),
+ rowMeta,
+ new Object[] {null, 7L});
+
+ ByteBuffer buffer =
ByteBuffer.wrap(written).order(ByteOrder.LITTLE_ENDIAN);
+ assertEquals(8, buffer.getInt(), "only the second column carries data");
+ // The first column (index 0) is the most significant bit.
+ assertEquals((byte) 0b1000_0000, buffer.get());
+ assertEquals(7L, buffer.getLong());
+ }
+
+ @Test
+ void refusesARowThatIsNarrowerThanTheColumnSpec() throws IOException {
+ PipedInputStream pipedInputStream = new PipedInputStream();
+ StreamEncoder encoder =
+ new StreamEncoder(
+ List.of(
+ new ColumnSpec(ColumnSpec.ConstantWidthType.INTEGER_64),
+ new ColumnSpec(ColumnSpec.ConstantWidthType.INTEGER_64)),
+ pipedInputStream);
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("A"));
+
+ assertThrows(
+ IllegalArgumentException.class, () -> encoder.writeRow(rowMeta, new
Object[] {1L}));
+ }
+
+ @Test
+ void writesTheHeaderInUtf8() throws IOException {
+ byte[] header = encode(List.of(new
ColumnSpec(ColumnSpec.ConstantWidthType.BOOLEAN)));
+ assertEquals("NATIVE", new String(header, 0, 6, StandardCharsets.UTF_8));
+ }
+
+ /** Writes the header for the given columns and returns the bytes handed to
Vertica. */
+ private static byte[] encode(List<ColumnSpec> columns) throws IOException {
+ PipedInputStream pipedInputStream = new PipedInputStream();
+ StreamEncoder encoder = new StreamEncoder(columns, pipedInputStream);
+ encoder.writeHeader();
+ encoder.close();
+ return pipedInputStream.readAllBytes();
+ }
+
+ /** Writes a single row (without the header) and returns the bytes handed to
Vertica. */
+ private static byte[] encodeRow(List<ColumnSpec> columns, IRowMeta rowMeta,
Object[] row)
+ throws Exception {
+ PipedInputStream pipedInputStream = new PipedInputStream();
+ StreamEncoder encoder = new StreamEncoder(columns, pipedInputStream);
+ encoder.writeRow(rowMeta, row);
+ encoder.close();
+ return pipedInputStream.readAllBytes();
+ }
+}
diff --git
a/plugins/databases/vertica/src/test/resources/vertica-bulkloader-transform.xml
b/plugins/databases/vertica/src/test/resources/vertica-bulkloader-transform.xml
new file mode 100644
index 0000000000..e32aed2ca0
--- /dev/null
+++
b/plugins/databases/vertica/src/test/resources/vertica-bulkloader-transform.xml
@@ -0,0 +1,56 @@
+<?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.
+ ~
+ -->
+<transform>
+ <name>Vertica bulk loader</name>
+ <type>VerticaBulkLoader</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <connection>vertica</connection>
+ <schema>staging</schema>
+ <table>INVOICE</table>
+ <truncate>Y</truncate>
+ <only_when_have_rows>Y</only_when_have_rows>
+ <direct>Y</direct>
+ <abort_on_error>N</abort_on_error>
+ <exceptions_filename>/tmp/exceptions.log</exceptions_filename>
+ <rejected_data_filename>/tmp/rejected.log</rejected_data_filename>
+ <stream_name>nightly load</stream_name>
+ <specify_fields>Y</specify_fields>
+ <fields>
+ <field>
+ <stream_name>id</stream_name>
+ <column_name>INVOICE_ID</column_name>
+ </field>
+ <field>
+ <stream_name>currency</stream_name>
+ <column_name>COST_CURRENCY</column_name>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>464</xloc>
+ <yloc>160</yloc>
+ </GUI>
+</transform>
