You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by ha...@apache.org on 2026/08/25 13:21:47 UTC
(hop) branch main updated: Improve RedShift Bulkloader and related CrateDB, fixes #3621 (#8083)
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 e552699516 Improve RedShift Bulkloader and related CrateDB, fixes #3621 (#8083)
e552699516 is described below
commit e55269951691a5eeffb65292509d831838323a1d
Author: Hans Van Akelyen <ha...@gmail.com>
AuthorDate: Tue Aug 25 15:21:41 2026 +0200
Improve RedShift Bulkloader and related CrateDB, fixes #3621 (#8083)
---
.../ROOT/pages/database/databases/redshift.adoc | 92 +++
.../pipeline/transforms/redshift-bulkloader.adoc | 41 +-
.../cratedb/0004-copy-from-filesystem-types.hpl | 183 +++++
.../cratedb/import/output-stream-0004.csv | 2 +
integration-tests/cratedb/import/output-stream.csv | 200 ++---
.../main-0004-copy-from-filesystem-types.hwf | 249 ++++++
lib-jdbc/pom.xml | 2 +-
lib/pom.xml | 26 +
.../cratedbbulkloader/CrateDBBulkLoader.java | 839 ++++++++++----------
.../cratedbbulkloader/CrateDBBulkLoaderData.java | 21 +-
.../cratedbbulkloader/CrateDBBulkLoaderDialog.java | 89 +--
.../cratedbbulkloader/CrateDBBulkLoaderMeta.java | 35 +-
.../messages/messages_en_US.properties | 4 +
.../cratedbbulkloader/CrateDBBulkLoaderTest.java | 388 ++++++++++
plugins/databases/redshift/pom.xml | 47 ++
.../redshift/RedshiftAuthenticationType.java | 58 ++
.../databases/redshift/RedshiftDatabaseMeta.java | 431 ++++++++++-
.../databases/redshift/RedshiftDeploymentType.java | 33 +
.../redshift/messages/messages_en_US.properties | 42 +
.../redshift/RedshiftAuthenticationTest.java | 405 ++++++++++
plugins/tech/aws/pom.xml | 22 +
.../redshift/bulkloader/RedshiftBulkLoader.java | 843 +++++++++++----------
.../bulkloader/RedshiftBulkLoaderData.java | 39 +-
.../bulkloader/RedshiftBulkLoaderDialog.java | 163 ++--
.../bulkloader/RedshiftBulkLoaderMeta.java | 58 +-
.../bulkloader/messages/messages_en_US.properties | 7 +
.../bulkloader/RedshiftBulkLoaderTest.java | 602 +++++++++++++++
27 files changed, 3738 insertions(+), 1183 deletions(-)
diff --git a/docs/hop-user-manual/modules/ROOT/pages/database/databases/redshift.adoc b/docs/hop-user-manual/modules/ROOT/pages/database/databases/redshift.adoc
index 2a91dc5289..b9f4fc7506 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/database/databases/redshift.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/database/databases/redshift.adoc
@@ -30,3 +30,95 @@ under the License.
|Documentation | https://docs.aws.amazon.com/redshift/latest/mgmt/configure-jdbc-connection.html[Documentation Link]
|JDBC Url | jdbc:redshift://endpoint:port/database
|===
+
+== Deployment
+
+Redshift comes in two shapes, and the *Deployment* option says which one you are connecting to.
+
+`PROVISIONED`:: A Redshift cluster. Enter its endpoint as the *Server hostname*, the way you always have. This is the default, so existing connections are unaffected.
++
+One extra field appears here when an IAM option is chosen:
++
+[options="header"]
+|===
+|Option|Description
+|Cluster identifier|The name of the cluster. Only needed when the driver cannot read it from the hostname — a custom domain name, or a load balancer in front of the cluster. Leave it empty for a normal cluster endpoint, where the driver works it out for itself.
+|===
+
+`SERVERLESS`:: A serverless workgroup. Its endpoint is entirely predictable, so rather than asking for a hostname that is easy to get subtly wrong, Hop builds it for you from three fields:
++
+[options="header"]
+|===
+|Option|Description
+|Workgroup name|The name of the workgroup. Note this is the *workgroup*, not the namespace — the two are easy to mix up, and a namespace name produces a hostname that does not resolve.
+|AWS account number|The 12 digit account number the workgroup belongs to.
+|AWS region|The region the workgroup runs in, e.g. `eu-west-1`.
+|===
++
+On a `PROVISIONED` cluster the region only appears for the IAM options, where it tells the driver
+which region's Redshift API to ask for temporary credentials. A cluster signing in with a database
+user and password has no use for it, and none is sent.
++
+Which yields `<workgroup>.<account-number>.<region>.redshift-serverless.amazonaws.com`. The console shows the same string on the workgroup's details page, under **General information**, if you would rather copy it.
++
+The default database of a serverless namespace is `dev`, *not* the namespace name.
++
+*Server hostname* stays available in this mode. Leave it empty and Hop builds the endpoint from
+the three fields above. Fill it in — with a custom domain name, or the address of a load balancer
+in front of the workgroup — and it is used as entered instead.
++
+Keep the workgroup, account number and region filled in either way. Behind a custom name the driver
+can no longer work out which workgroup it is talking to, so those three are passed to it as the
+`serverlessWorkGroup`, `serverlessAcctId` and `Region` options, which is precisely what AWS provides
+them for. Without them, IAM authentication against a custom endpoint fails.
++
+A provisioned cluster has the same problem and the same answer: fill in its *Cluster identifier*,
+which reaches the driver as `ClusterID`.
+
+[TIP]
+====
+A serverless workgroup lives inside a VPC. A `SocketTimeoutException: Connect timed out` when testing the connection means the endpoint could not be reached at all, which is a networking matter rather than anything about the connection settings.
+
+Reaching one from outside its VPC takes three things, and all three are easy to miss:
+
+. The workgroup must be publicly accessible (`aws redshift-serverless update-workgroup --publicly-accessible`).
+. Its security group must allow your address on the workgroup's port, 5439 by default.
+. Its subnets must route `0.0.0.0/0` to an *internet gateway*. A subnet whose default route is a NAT gateway is outbound-only: the endpoint still gets a public IP, but replies leave through the NAT and never reach you, so the connection times out exactly as if it were firewalled.
+
+Resolving the endpoint's hostname tells you which case you are in: a `10.x`, `172.16-31.x` or `192.168.x` answer means the workgroup is private.
+
+Without enhanced VPC routing a workgroup needs at least two subnets across two Availability Zones, so making one publicly reachable means having two internet-gateway-routed subnets, not one.
+
+Running Hop inside the VPC avoids all of this, and is usually the better answer for scheduled or automated runs.
+====
+
+== Authentication
+
+The *Authentication* option chooses how the driver proves who you are.
+
+[options="header"]
+|===
+|Option|Description
+|`DATABASE`|A Redshift database user and password, entered as the connection's username and password. This is the default and the historical behaviour.
+|`IAM_CREDENTIALS`|An AWS access key ID and secret access key, optionally with a session token for temporary credentials.
+|`IAM_PROFILE`|A named profile from your shared AWS credentials file, usually `~/.aws/credentials`. Nothing secret is stored in the Hop connection.
+|`IAM_DEFAULT_CHAIN`|Whatever the AWS default credentials chain finds: environment variables, the shared credentials file, an SSO session, an EC2 instance profile, an ECS task role. This is the one to use when Hop runs inside AWS.
+|===
+
+The three `IAM_` options do not sign in with a Redshift password at all. The driver calls AWS for temporary database credentials and connects with those, which is why the URL Hop builds for them starts with `jdbc:redshift:iam://` instead of `jdbc:redshift://`.
+
+They share three further options:
+
+[options="header"]
+|===
+|Option|Description
+|Database user|The Redshift user the temporary credentials are issued for. Leave empty to use the connection's username.
+|Database groups|Comma separated list of existing database groups the user joins for the session. Optional.
+|Create the user if missing|Create the database user when it does not exist yet, rather than failing to authenticate.
+|===
+
+NOTE: An AWS access key is not a Redshift user, so it does not go in the connection's username and password fields. Those are always the database credentials.
+
+== Custom domain names
+
+If a custom domain name or a load balancer sits in front of your cluster or workgroup, the driver cannot work out from the hostname which cluster it is talking to. Enter the JDBC URL yourself in the *Manual URL* field; anything set on the *Options* tab is still passed to the driver, so the authentication options above keep working.
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/redshift-bulkloader.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/redshift-bulkloader.adoc
index 8f0c7ff1d0..c61ad81ac8 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/redshift-bulkloader.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/redshift-bulkloader.adoc
@@ -54,7 +54,9 @@ IMPORTANT: The Redshift Bulk Loader is linked to the database type. It will fetc
|Connection|Name of the database connection on which the target table resides.
|Target schema|The name of the target schema to write data to.
|Target table|The name of the target table to write data to.
-|AWS Authentication a|choose which authentication method to use with the `COPY` command. Supported options are `AWS Credentials` and `IAM Role`. +
+|AWS Authentication a|choose which authentication method to use with the `COPY` command. Supported options are `Inherit from connection`, `Credentials` and `IAM Role`. +
+
+* `Inherit from connection` takes the AWS credentials from the Redshift connection selected above, so they only have to be configured in one place. The connection must use one of its IAM authentication options; one that signs in with a database user and password has no AWS credentials to lend. A connection pointing at a named profile or at the AWS default credentials chain is resolved when the pipeline runs, and a session token is passed along when the resulting credentials are temporary.
* check the https://docs.aws.amazon.com/redshift/latest/dg/copy-usage_notes-access-permissions.html#copy-usage_notes-access-key-based[Key-based access control] for more information on the `Credentials` option.
* check the https://docs.aws.amazon.com/redshift/latest/dg/copy-usage_notes-access-permissions.html#copy-usage_notes-access-role-based[IAM Role] docs for more information on the `IAM Role` option.
@@ -78,6 +80,43 @@ IMPORTANT: The Redshift Bulk Loader is linked to the database type. It will fetc
|Copy into Redshift from existing file|path to the file in S3 to `COPY` load the data from.
|===
+== The staging file
+
+When *Stream to S3 CSV* is enabled the transform writes the CSV named under
+*Copy into Redshift from file name/path* itself, and creates the folder it goes in when that does
+not exist yet. This matters on S3, where a prefix only exists for as long as an object sits under
+it: a brand new path would otherwise be refused before a single row was written.
+
+== JSON and SUPER columns
+
+A JSON field is written to the CSV file enclosed and escaped, the same as any text that could
+otherwise break the row -- JSON is full of commas and quotes, and Apache Hop pretty prints it
+across several lines by default.
+
+`COPY` parses the field into a real `SUPER` object, so `JSON_TYPEOF(payload)` returns `object`
+once the value has loaded.
+
+Two things surprise people when reading it back:
+
+* `payload::varchar` on a `SUPER` **object** is `NULL`. Use `JSON_SERIALIZE(payload)` to get the
+ JSON text.
+* Redshift folds unquoted `SUPER` attribute names to lower case unless
+ `enable_case_sensitive_super_attribute` is on, so `payload.myField` does not match a camelCase
+ key -- and quoting the attribute does not help while that setting is off.
+
+Addressing the value as text sidesteps both:
+
+[source,sql]
+----
+SELECT JSON_EXTRACT_PATH_TEXT(JSON_SERIALIZE(payload), 'myField') FROM my_table;
+----
+
+== Dates and timestamps
+
+Date and timestamp fields are written to the CSV file as ISO 8601 with milliseconds, `yyyy-MM-dd HH:mm:ss.SSS`, and the `COPY` statement declares `DATEFORMAT AS 'auto'` and `TIMEFORMAT AS 'auto'` to read them back. `auto` is used because the explicit `TIMEFORMAT` patterns have no token for fractional seconds.
+
+An Apache Hop `Date` carries a time of day just as a `Timestamp` does, so both are written whole and the target column decides what to keep: a `TIMESTAMP` column keeps the time, a `DATE` column truncates it.
+
== Database fields
Map the current stream fields to the Redshift table's columns.
diff --git a/integration-tests/cratedb/0004-copy-from-filesystem-types.hpl b/integration-tests/cratedb/0004-copy-from-filesystem-types.hpl
new file mode 100644
index 0000000000..0d82583cd9
--- /dev/null
+++ b/integration-tests/cratedb/0004-copy-from-filesystem-types.hpl
@@ -0,0 +1,183 @@
+<?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-copy-from-filesystem-types</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <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/25 09:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/25 09:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>rows</from>
+ <to>load</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>rows</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>1</item>
+ <item>1.5</item>
+ <item>Acme, "The Best" Inc.</item>
+ <item>2026-08-24 15:30:45.123</item>
+ </line>
+ <line>
+ <item>2</item>
+ <item>-0.25</item>
+ <item>plain</item>
+ <item>2026-01-01 00:00:00.000</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <format></format>
+ <group/>
+ <decimal/>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <set_empty_string>N</set_empty_string>
+ <name>ratio</name>
+ <format>0.####</format>
+ <group/>
+ <decimal/>
+ <type>Number</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <set_empty_string>N</set_empty_string>
+ <name>name</name>
+ <format></format>
+ <group/>
+ <decimal/>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <set_empty_string>N</set_empty_string>
+ <name>when_ts</name>
+ <format>yyyy-MM-dd HH:mm:ss.SSS</format>
+ <group/>
+ <decimal/>
+ <type>Timestamp</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>192</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>load</name>
+ <type>CrateDBBulkLoader</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <connection>cratedb-test</connection>
+ <schema>doc</schema>
+ <table>types_test</table>
+ <use_http_endpoint>N</use_http_endpoint>
+ <http_endpoint/>
+ <batch_size>1000</batch_size>
+ <http_login/>
+ <http_password>Encrypted </http_password>
+ <use_system_env_vars>N</use_system_env_vars>
+ <aws_access_key_id/>
+ <aws_secret_access_key>Encrypted </aws_secret_access_key>
+ <truncate>N</truncate>
+ <only_when_have_rows>N</only_when_have_rows>
+ <stream_to_s3>Y</stream_to_s3>
+ <specify_fields>Y</specify_fields>
+ <load_from_filename>file:///home/output-stream-0004.csv</load_from_filename>
+ <volume_mapping>${PROJECT_HOME}/import/</volume_mapping>
+ <trim_whitespace>N</trim_whitespace>
+ <null_if/>
+ <error_column_mismatch>N</error_column_mismatch>
+ <strip_null>N</strip_null>
+ <fields>
+ <field>
+ <stream_name>id</stream_name>
+ <column_name>id</column_name>
+ </field>
+ <field>
+ <stream_name>ratio</stream_name>
+ <column_name>ratio</column_name>
+ </field>
+ <field>
+ <stream_name>name</stream_name>
+ <column_name>name</column_name>
+ </field>
+ <field>
+ <stream_name>when_ts</stream_name>
+ <column_name>when_ts</column_name>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>480</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/cratedb/import/output-stream-0004.csv b/integration-tests/cratedb/import/output-stream-0004.csv
new file mode 100644
index 0000000000..1523e7f531
--- /dev/null
+++ b/integration-tests/cratedb/import/output-stream-0004.csv
@@ -0,0 +1,2 @@
+1,1.5,"Acme, ""The Best"" Inc.","2026-08-24 15:30:45.123"
+2,-0.25,"plain","2026-01-01 00:00:00.000"
diff --git a/integration-tests/cratedb/import/output-stream.csv b/integration-tests/cratedb/import/output-stream.csv
index 1a7f41f41b..49945a3bdb 100644
--- a/integration-tests/cratedb/import/output-stream.csv
+++ b/integration-tests/cratedb/import/output-stream.csv
@@ -1,100 +1,100 @@
-"Margie ","Koss ","1978-05-24","Clydeport ","Martinique ","1999-11-09"
-"Freddie ","Marks ","1993-06-15","Schaeferfort ","Madagascar ","1970-09-29"
-"Paul ","McLaughlin ","1977-09-16","Bayermouth ","Liechtenstein ","2001-11-10"
-"Mary ","Langosh ","2004-07-19","Austinstad ","Slovakia (Slovak Republic) ","1973-09-09"
-"Andrea ","Wunsch ","1962-11-21","Lake Renettaburgh","Democratic People's Republic of Korea","2006-01-22"
-"Leeanne ","Brekke ","2000-10-09","Williamsonside ","Pitcairn Islands ","1974-08-30"
-"Laverne ","Schinner ","1983-09-21","North Shantay ","Cape Verde ","1974-09-30"
-"Sabrina ","Hane ","1983-10-17","Carlyhaven ","Cote d'Ivoire ","1998-02-20"
-"Lettie ","Wintheiser ","1974-02-19","Hermistonchester ","Benin ","1962-11-25"
-"Mica ","Jacobs ","1978-05-22","East Fermin ","Turks and Caicos Islands ","1996-06-19"
-"Brynn ","Lueilwitz ","2005-11-20","Feeneybury ","Western Sahara ","1996-12-10"
-"Stanton ","Kirlin ","1999-05-09","Port Ahmad ","Romania ","1991-09-08"
-"Olimpia ","Parker ","1980-05-23","Corkeryville ","Gabon ","1965-08-09"
-"Jude ","Steuber ","2002-04-12","Wendieport ","Mongolia ","2006-03-09"
-"Adelle ","Willms ","1983-12-19","Hodkiewiczstad ","Pitcairn Islands ","1971-11-06"
-"Reanna ","O'Keefe ","1990-06-12","Nelleshire ","Congo ","1983-04-25"
-"Les ","Boyer ","1977-01-09","Wehnerstad ","Netherlands Antilles ","1986-06-08"
-"Jonna ","Ward ","1970-12-03","Tommyside ","Vanuatu ","1995-07-06"
-"Stefani ","Bernier ","1970-03-14","Treenaview ","Gibraltar ","1971-11-22"
-"Maryetta ","Cronin ","1984-03-11","East Melissaburgh","Svalbard & Jan Mayen Islands ","1985-11-23"
-"Sophie ","Mosciski ","1984-10-20","Lake Barrettland ","Reunion ","1959-12-27"
-"Oscar ","Braun ","1963-11-15","East Scottieside ","Cameroon ","1998-10-21"
-"Margorie ","Huel ","1963-02-20","New Ed ","Netherlands Antilles ","1981-02-17"
-"Lenora ","Rutherford ","2003-05-11","Reedbury ","Micronesia ","1996-09-17"
-"Tawana ","Satterfield","1964-05-20","Noelbury ","Northern Mariana Islands ","1968-04-07"
-"Roni ","Koelpin ","1965-05-19","South Ericside ","Serbia ","1995-09-29"
-"Artie ","Thiel ","1976-11-07","West Demetrice ","Grenada ","1978-04-08"
-"Eliseo ","Kshlerin ","1966-06-11","Lurlenefurt ","Venezuela ","2003-10-29"
-"Alissa ","Konopelski ","1991-04-03","Brandeeport ","Kazakhstan ","1961-10-11"
-"Toi ","Balistreri ","1976-08-11","Greenfelderhaven ","Saudi Arabia ","1973-10-17"
-"Katherina","Yost ","1998-10-23","West Ladonna ","Macedonia ","1979-05-02"
-"Harlan ","Murazik ","2004-06-24","New Tamicaburgh ","Brunei Darussalam ","1987-03-30"
-"Romaine ","Weimann ","1993-01-19","Port Dorismouth ","Mozambique ","1978-04-16"
-"Vanessa ","Stanton ","2002-04-10","Wehnerhaven ","Slovenia ","1966-08-23"
-"Clara ","Wolff ","1969-08-17","McDermottview ","Spain ","1965-03-12"
-"Carmel ","Wisoky ","1999-02-24","West Carenmouth ","French Southern Territories ","1972-09-04"
-"Jimmie ","Torp ","1991-06-05","East Winford ","French Southern Territories ","1985-05-10"
-"Livia ","Renner ","1972-07-22","Port Raymonport ","Vietnam ","1973-07-22"
-"Raul ","Roberts ","2002-07-07","Fernandehaven ","Macedonia ","1987-08-22"
-"Ariane ","Cummings ","1969-09-04","North Kitmouth ","French Guiana ","1983-05-24"
-"Janae ","Farrell ","2000-08-29","Ramirotown ","Indonesia ","1996-06-06"
-"Tena ","Streich ","1991-10-06","West Stephenview ","Bhutan ","1985-01-28"
-"Dalila ","Miller ","1983-11-02","Port Tanner ","Andorra ","1963-12-21"
-"Orval ","Williamson ","1998-02-19","Lake Jesusita ","Bahamas ","1995-10-14"
-"Loren ","Bednar ","1967-06-13","Leisaburgh ","Slovenia ","1994-08-30"
-"Delbert ","Sanford ","1997-02-18","Port Tobyview ","Suriname ","1977-09-28"
-"Bernadine","O'Reilly ","1963-01-20","Dudleyport ","Mauritius ","1989-10-27"
-"Jc ","Wuckert ","1989-10-06","Shaneville ","Jersey ","2005-02-23"
-"Logan ","Wolff ","1972-12-09","Croninstad ","Guinea-Bissau ","1977-05-01"
-"Erinn ","Grady ","1969-01-29","East Olga ","Spain ","1987-06-14"
-"Phung ","Will ","1975-04-22","Nicolasville ","Micronesia ","1998-08-27"
-"Kareem ","Kuvalis ","1995-06-27","Loviefort ","Cambodia ","1973-09-27"
-"Thad ","O'Keefe ","1972-02-27","New Latonia ","Palestinian Territory ","1995-05-02"
-"Andy ","Rath ","1968-10-22","Port Christianasi","Cambodia ","1963-11-25"
-"Jinny ","Brakus ","1966-06-06","New Gena ","Lesotho ","1963-05-08"
-"Celena ","Goyette ","1977-05-25","Cruickshankmouth ","Equatorial Guinea ","1985-08-23"
-"Tashina ","Fisher ","1995-08-30","Lake Vernon ","Romania ","1975-08-05"
-"Rhett ","Bradtke ","1976-06-29","East Erasmo ","Pakistan ","1991-03-06"
-"Myrtis ","Prohaska ","1978-07-07","Lake Arleenland ","Afghanistan ","1990-01-18"
-"Lamont ","Nitzsche ","1967-09-02","Marcoland ","Marshall Islands ","1977-06-26"
-"Alla ","Roob ","1996-07-31","New Dion ","Bolivia ","2004-09-20"
-"Lashawna ","Collier ","1987-08-18","Schimmelshire ","South Africa ","2000-12-06"
-"Art ","Herzog ","1968-11-27","West Tonyfort ","Australia ","1996-08-12"
-"Adam ","Smitham ","1960-07-16","Dareview ","Bosnia and Herzegovina ","1995-07-18"
-"Beatrice ","Mayer ","1996-06-14","Port Loren ","Jordan ","1966-12-02"
-"Gus ","Cremin ","1960-08-19","Alfredside ","Niger ","1964-02-19"
-"Micah ","Kulas ","2004-08-29","Janycemouth ","Gabon ","1982-08-19"
-"Avery ","Walsh ","1974-08-20","Chassidyton ","Saint Martin ","1971-03-18"
-"Rufus ","Runte ","1986-12-09","Randallburgh ","Greenland ","2004-01-19"
-"Curtis ","Nitzsche ","1988-04-16","Lake Rebecca ","Mali ","1964-08-08"
-"Dawn ","Hansen ","1967-11-08","North Romeo ","Tajikistan ","1968-09-06"
-"Chassidy ","Lynch ","1997-01-13","West Hassan ","China ","1990-12-03"
-"Laurena ","Graham ","1987-07-16","Gutmannview ","French Polynesia ","1978-06-30"
-"Sabine ","Rippin ","1999-08-24","North Chu ","Equatorial Guinea ","1985-10-09"
-"Shalonda ","Purdy ","2005-03-29","West Tamar ","Equatorial Guinea ","2005-02-08"
-"Trevor ","Kling ","1974-07-25","North Bellafort ","Iceland ","1985-11-05"
-"Warner ","Hintz ","1985-03-16","North Amos ","Dominica ","1962-07-13"
-"Leigh ","Willms ","1978-01-11","O'Konside ","India ","1970-04-21"
-"Charlyn ","Schmidt ","1963-12-16","Leland ","Tokelau ","2004-05-31"
-"Dessie ","Paucek ","1984-04-01","Bechtelarton ","Malaysia ","1980-02-07"
-"Janette ","Jacobson ","2005-07-03","Kumfurt ","French Guiana ","1992-10-09"
-"Young ","Langworth ","1980-04-09","North Geraldstad ","Latvia ","1997-04-14"
-"Sparkle ","Waters ","1984-11-27","Rathstad ","Gambia ","1964-08-04"
-"Yvonne ","Koelpin ","1988-09-13","South Ivorybury ","British Indian Ocean Territory (Chago","1993-01-25"
-"Burt ","Reynolds ","1962-04-29","North Jonas ","Jamaica ","1993-07-24"
-"Chase ","Murazik ","1963-11-25","Wizaborough ","Cyprus ","1987-09-18"
-"Oliver ","Ruecker ","1971-12-05","West Nolan ","Benin ","1979-09-03"
-"Rolande ","Funk ","1988-05-01","New Wendellland ","Venezuela ","1971-08-04"
-"Floyd ","Littel ","1985-09-12","North Leroyton ","Niger ","2004-12-12"
-"Sung ","Altenwerth ","1964-07-29","McGlynnberg ","British Indian Ocean Territory (Chago","2004-10-20"
-"Jacque ","Torp ","1988-10-05","Denesikshire ","Mozambique ","1985-06-07"
-"Kristyn ","Murazik ","1983-04-08","Brettmouth ","Congo ","2001-02-17"
-"Gerardo ","Kris ","1971-05-16","Lefflerside ","Zimbabwe ","2005-04-26"
-"Darron ","O'Connell ","1984-08-22","South Cary ","Malta ","1998-08-31"
-"Dominga ","Dickinson ","1963-10-22","East Paris ","Samoa ","1993-05-28"
-"Wilson ","Nolan ","1999-03-16","West Keenahaven ","Norfolk Island ","1986-06-21"
-"Morton ","Koelpin ","1965-04-17","Cummerataberg ","Denmark ","1988-04-14"
-"Sharan ","Rice ","1985-11-17","Renafurt ","Somalia ","1972-09-10"
-"Hershel ","O'Kon ","1980-04-17","Bethannmouth ","Slovakia (Slovak Republic) ","1983-02-27"
-"Sophie ","Ratke ","1966-04-13","Leschfurt ","Nigeria ","1981-02-18"
+"Margie","Koss","1978-05-24","Clydeport","Martinique","1999-11-09"
+"Freddie","Marks","1993-06-15","Schaeferfort","Madagascar","1970-09-29"
+"Paul","McLaughlin","1977-09-16","Bayermouth","Liechtenstein","2001-11-10"
+"Mary","Langosh","2004-07-19","Austinstad","Slovakia (Slovak Republic)","1973-09-09"
+"Andrea","Wunsch","1962-11-21","Lake Renettaburgh","Democratic People's Republic of Korea","2006-01-22"
+"Leeanne","Brekke","2000-10-09","Williamsonside","Pitcairn Islands","1974-08-30"
+"Laverne","Schinner","1983-09-21","North Shantay","Cape Verde","1974-09-30"
+"Sabrina","Hane","1983-10-17","Carlyhaven","Cote d'Ivoire","1998-02-20"
+"Lettie","Wintheiser","1974-02-19","Hermistonchester","Benin","1962-11-25"
+"Mica","Jacobs","1978-05-22","East Fermin","Turks and Caicos Islands","1996-06-19"
+"Brynn","Lueilwitz","2005-11-20","Feeneybury","Western Sahara","1996-12-10"
+"Stanton","Kirlin","1999-05-09","Port Ahmad","Romania","1991-09-08"
+"Olimpia","Parker","1980-05-23","Corkeryville","Gabon","1965-08-09"
+"Jude","Steuber","2002-04-12","Wendieport","Mongolia","2006-03-09"
+"Adelle","Willms","1983-12-19","Hodkiewiczstad","Pitcairn Islands","1971-11-06"
+"Reanna","O'Keefe","1990-06-12","Nelleshire","Congo","1983-04-25"
+"Les","Boyer","1977-01-09","Wehnerstad","Netherlands Antilles","1986-06-08"
+"Jonna","Ward","1970-12-03","Tommyside","Vanuatu","1995-07-06"
+"Stefani","Bernier","1970-03-14","Treenaview","Gibraltar","1971-11-22"
+"Maryetta","Cronin","1984-03-11","East Melissaburgh","Svalbard & Jan Mayen Islands","1985-11-23"
+"Sophie","Mosciski","1984-10-20","Lake Barrettland","Reunion","1959-12-27"
+"Oscar","Braun","1963-11-15","East Scottieside","Cameroon","1998-10-21"
+"Margorie","Huel","1963-02-20","New Ed","Netherlands Antilles","1981-02-17"
+"Lenora","Rutherford","2003-05-11","Reedbury","Micronesia","1996-09-17"
+"Tawana","Satterfield","1964-05-20","Noelbury","Northern Mariana Islands","1968-04-07"
+"Roni","Koelpin","1965-05-19","South Ericside","Serbia","1995-09-29"
+"Artie","Thiel","1976-11-07","West Demetrice","Grenada","1978-04-08"
+"Eliseo","Kshlerin","1966-06-11","Lurlenefurt","Venezuela","2003-10-29"
+"Alissa","Konopelski","1991-04-03","Brandeeport","Kazakhstan","1961-10-11"
+"Toi","Balistreri","1976-08-11","Greenfelderhaven","Saudi Arabia","1973-10-17"
+"Katherina","Yost","1998-10-23","West Ladonna","Macedonia","1979-05-02"
+"Harlan","Murazik","2004-06-24","New Tamicaburgh","Brunei Darussalam","1987-03-30"
+"Romaine","Weimann","1993-01-19","Port Dorismouth","Mozambique","1978-04-16"
+"Vanessa","Stanton","2002-04-10","Wehnerhaven","Slovenia","1966-08-23"
+"Clara","Wolff","1969-08-17","McDermottview","Spain","1965-03-12"
+"Carmel","Wisoky","1999-02-24","West Carenmouth","French Southern Territories","1972-09-04"
+"Jimmie","Torp","1991-06-05","East Winford","French Southern Territories","1985-05-10"
+"Livia","Renner","1972-07-22","Port Raymonport","Vietnam","1973-07-22"
+"Raul","Roberts","2002-07-07","Fernandehaven","Macedonia","1987-08-22"
+"Ariane","Cummings","1969-09-04","North Kitmouth","French Guiana","1983-05-24"
+"Janae","Farrell","2000-08-29","Ramirotown","Indonesia","1996-06-06"
+"Tena","Streich","1991-10-06","West Stephenview","Bhutan","1985-01-28"
+"Dalila","Miller","1983-11-02","Port Tanner","Andorra","1963-12-21"
+"Orval","Williamson","1998-02-19","Lake Jesusita","Bahamas","1995-10-14"
+"Loren","Bednar","1967-06-13","Leisaburgh","Slovenia","1994-08-30"
+"Delbert","Sanford","1997-02-18","Port Tobyview","Suriname","1977-09-28"
+"Bernadine","O'Reilly","1963-01-20","Dudleyport","Mauritius","1989-10-27"
+"Jc","Wuckert","1989-10-06","Shaneville","Jersey","2005-02-23"
+"Logan","Wolff","1972-12-09","Croninstad","Guinea-Bissau","1977-05-01"
+"Erinn","Grady","1969-01-29","East Olga","Spain","1987-06-14"
+"Phung","Will","1975-04-22","Nicolasville","Micronesia","1998-08-27"
+"Kareem","Kuvalis","1995-06-27","Loviefort","Cambodia","1973-09-27"
+"Thad","O'Keefe","1972-02-27","New Latonia","Palestinian Territory","1995-05-02"
+"Andy","Rath","1968-10-22","Port Christianaside","Cambodia","1963-11-25"
+"Jinny","Brakus","1966-06-06","New Gena","Lesotho","1963-05-08"
+"Celena","Goyette","1977-05-25","Cruickshankmouth","Equatorial Guinea","1985-08-23"
+"Tashina","Fisher","1995-08-30","Lake Vernon","Romania","1975-08-05"
+"Rhett","Bradtke","1976-06-29","East Erasmo","Pakistan","1991-03-06"
+"Myrtis","Prohaska","1978-07-07","Lake Arleenland","Afghanistan","1990-01-18"
+"Lamont","Nitzsche","1967-09-02","Marcoland","Marshall Islands","1977-06-26"
+"Alla","Roob","1996-07-31","New Dion","Bolivia","2004-09-20"
+"Lashawna","Collier","1987-08-18","Schimmelshire","South Africa","2000-12-06"
+"Art","Herzog","1968-11-27","West Tonyfort","Australia","1996-08-12"
+"Adam","Smitham","1960-07-16","Dareview","Bosnia and Herzegovina","1995-07-18"
+"Beatrice","Mayer","1996-06-14","Port Loren","Jordan","1966-12-02"
+"Gus","Cremin","1960-08-19","Alfredside","Niger","1964-02-19"
+"Micah","Kulas","2004-08-29","Janycemouth","Gabon","1982-08-19"
+"Avery","Walsh","1974-08-20","Chassidyton","Saint Martin","1971-03-18"
+"Rufus","Runte","1986-12-09","Randallburgh","Greenland","2004-01-19"
+"Curtis","Nitzsche","1988-04-16","Lake Rebecca","Mali","1964-08-08"
+"Dawn","Hansen","1967-11-08","North Romeo","Tajikistan","1968-09-06"
+"Chassidy","Lynch","1997-01-13","West Hassan","China","1990-12-03"
+"Laurena","Graham","1987-07-16","Gutmannview","French Polynesia","1978-06-30"
+"Sabine","Rippin","1999-08-24","North Chu","Equatorial Guinea","1985-10-09"
+"Shalonda","Purdy","2005-03-29","West Tamar","Equatorial Guinea","2005-02-08"
+"Trevor","Kling","1974-07-25","North Bellafort","Iceland","1985-11-05"
+"Warner","Hintz","1985-03-16","North Amos","Dominica","1962-07-13"
+"Leigh","Willms","1978-01-11","O'Konside","India","1970-04-21"
+"Charlyn","Schmidt","1963-12-16","Leland","Tokelau","2004-05-31"
+"Dessie","Paucek","1984-04-01","Bechtelarton","Malaysia","1980-02-07"
+"Janette","Jacobson","2005-07-03","Kumfurt","French Guiana","1992-10-09"
+"Young","Langworth","1980-04-09","North Geraldstad","Latvia","1997-04-14"
+"Sparkle","Waters","1984-11-27","Rathstad","Gambia","1964-08-04"
+"Yvonne","Koelpin","1988-09-13","South Ivorybury","British Indian Ocean Territory (Chagos Archipelago)","1993-01-25"
+"Burt","Reynolds","1962-04-29","North Jonas","Jamaica","1993-07-24"
+"Chase","Murazik","1963-11-25","Wizaborough","Cyprus","1987-09-18"
+"Oliver","Ruecker","1971-12-05","West Nolan","Benin","1979-09-03"
+"Rolande","Funk","1988-05-01","New Wendellland","Venezuela","1971-08-04"
+"Floyd","Littel","1985-09-12","North Leroyton","Niger","2004-12-12"
+"Sung","Altenwerth","1964-07-29","McGlynnberg","British Indian Ocean Territory (Chagos Archipelago)","2004-10-20"
+"Jacque","Torp","1988-10-05","Denesikshire","Mozambique","1985-06-07"
+"Kristyn","Murazik","1983-04-08","Brettmouth","Congo","2001-02-17"
+"Gerardo","Kris","1971-05-16","Lefflerside","Zimbabwe","2005-04-26"
+"Darron","O'Connell","1984-08-22","South Cary","Malta","1998-08-31"
+"Dominga","Dickinson","1963-10-22","East Paris","Samoa","1993-05-28"
+"Wilson","Nolan","1999-03-16","West Keenahaven","Norfolk Island","1986-06-21"
+"Morton","Koelpin","1965-04-17","Cummerataberg","Denmark","1988-04-14"
+"Sharan","Rice","1985-11-17","Renafurt","Somalia","1972-09-10"
+"Hershel","O'Kon","1980-04-17","Bethannmouth","Slovakia (Slovak Republic)","1983-02-27"
+"Sophie","Ratke","1966-04-13","Leschfurt","Nigeria","1981-02-18"
diff --git a/integration-tests/cratedb/main-0004-copy-from-filesystem-types.hwf b/integration-tests/cratedb/main-0004-copy-from-filesystem-types.hwf
new file mode 100644
index 0000000000..e5950b4a2b
--- /dev/null
+++ b/integration-tests/cratedb/main-0004-copy-from-filesystem-types.hwf
@@ -0,0 +1,249 @@
+<?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-copy-from-filesystem-types</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/08/25 09:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/25 09: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>96</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Create types table</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <connection>cratedb-test</connection>
+ <sendOneStatement>N</sendOneStatement>
+ <sql>DROP TABLE IF EXISTS doc.types_test;
+CREATE TABLE doc.types_test
+(
+ id BIGINT
+, ratio DOUBLE PRECISION
+, name VARCHAR(4000)
+, when_ts TIMESTAMP
+);</sql>
+ <sqlfromfile>N</sqlfromfile>
+ <useVariableSubstitution>N</useVariableSubstitution>
+ <parallel>N</parallel>
+ <xloc>288</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0004-copy-from-filesystem-types</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}/0004-copy-from-filesystem-types.hpl</filename>
+ <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>544</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Refresh types table</name>
+ <description/>
+ <type>SQL</type>
+ <attributes/>
+ <connection>cratedb-test</connection>
+ <sendOneStatement>N</sendOneStatement>
+ <sql>REFRESH TABLE doc.types_test;</sql>
+ <sqlfromfile>N</sqlfromfile>
+ <useVariableSubstitution>N</useVariableSubstitution>
+ <parallel>N</parallel>
+ <xloc>800</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>2 rows loaded</name>
+ <description/>
+ <type>EVAL_TABLE_CONTENT</type>
+ <attributes/>
+ <connection>cratedb-test</connection>
+ <schemaname/>
+ <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 id FROM doc.types_test</custom_sql>
+ <add_rows_result>N</add_rows_result>
+ <clear_result_rows>Y</clear_result_rows>
+ <parallel>N</parallel>
+ <xloc>1040</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>numbers survived the round trip</name>
+ <description/>
+ <type>EVAL_TABLE_CONTENT</type>
+ <attributes/>
+ <connection>cratedb-test</connection>
+ <schemaname/>
+ <tablename/>
+ <success_condition>rows_count_equal</success_condition>
+ <limit>0</limit>
+ <is_custom_sql>Y</is_custom_sql>
+ <is_usevars>N</is_usevars>
+ <custom_sql>SELECT id FROM doc.types_test WHERE NOT COALESCE((id = 1 AND ratio = 1.5) OR (id = 2 AND ratio = -0.25), FALSE)</custom_sql>
+ <add_rows_result>N</add_rows_result>
+ <clear_result_rows>Y</clear_result_rows>
+ <parallel>N</parallel>
+ <xloc>1280</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>commas and quotes kept in the text column</name>
+ <description/>
+ <type>EVAL_TABLE_CONTENT</type>
+ <attributes/>
+ <connection>cratedb-test</connection>
+ <schemaname/>
+ <tablename/>
+ <success_condition>rows_count_equal</success_condition>
+ <limit>0</limit>
+ <is_custom_sql>Y</is_custom_sql>
+ <is_usevars>N</is_usevars>
+ <custom_sql>SELECT id FROM doc.types_test WHERE id = 1 AND name IS DISTINCT FROM 'Acme, "The Best" Inc.'</custom_sql>
+ <add_rows_result>N</add_rows_result>
+ <clear_result_rows>Y</clear_result_rows>
+ <parallel>N</parallel>
+ <xloc>1520</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>timestamps kept their milliseconds</name>
+ <description/>
+ <type>EVAL_TABLE_CONTENT</type>
+ <attributes/>
+ <connection>cratedb-test</connection>
+ <schemaname/>
+ <tablename/>
+ <success_condition>rows_count_equal</success_condition>
+ <limit>0</limit>
+ <is_custom_sql>Y</is_custom_sql>
+ <is_usevars>N</is_usevars>
+ <custom_sql>SELECT id FROM doc.types_test WHERE id = 1 AND when_ts IS DISTINCT FROM '2026-08-24T15:30:45.123'::timestamp</custom_sql>
+ <add_rows_result>N</add_rows_result>
+ <clear_result_rows>Y</clear_result_rows>
+ <parallel>N</parallel>
+ <xloc>1760</xloc>
+ <yloc>144</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>Create types table</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>Create types table</from>
+ <to>0004-copy-from-filesystem-types</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0004-copy-from-filesystem-types</from>
+ <to>Refresh types table</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>Refresh types table</from>
+ <to>2 rows loaded</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>2 rows loaded</from>
+ <to>numbers survived the round trip</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>numbers survived the round trip</from>
+ <to>commas and quotes kept in the text column</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>commas and quotes kept in the text column</from>
+ <to>timestamps kept their milliseconds</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git a/lib-jdbc/pom.xml b/lib-jdbc/pom.xml
index 50db4afb82..6afc28dc53 100644
--- a/lib-jdbc/pom.xml
+++ b/lib-jdbc/pom.xml
@@ -39,7 +39,7 @@
<monetdb.version>12.2</monetdb.version>
<mssqlnative.version>13.4.0.jre11</mssqlnative.version>
<postgresql.version>42.7.12</postgresql.version>
- <redshift.version>2.2.7</redshift.version>
+ <redshift.version>2.2.8</redshift.version>
<snowflake.jdbc.version>4.3.0</snowflake.jdbc.version>
<sqlite.version>3.53.2.0</sqlite.version>
<vertica.jdbc.version>25.3.0-0</vertica.jdbc.version>
diff --git a/lib/pom.xml b/lib/pom.xml
index d0014185c1..d6f61f044e 100644
--- a/lib/pom.xml
+++ b/lib/pom.xml
@@ -259,6 +259,32 @@
<artifactId>secretsmanager</artifactId>
<version>${aws-java-sdk.version}</version>
</dependency>
+ <!--
+ The Redshift JDBC driver declares these as optional dependencies, so Maven never
+ brings them in by itself. They are only needed for IAM authentication, where the
+ driver calls AWS for temporary database credentials rather than signing in with a
+ Redshift password.
+ -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>redshift</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>redshiftserverless</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sts</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>auth</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>sns</artifactId>
diff --git a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoader.java b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoader.java
index 927c097c3c..db9b6687db 100644
--- a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoader.java
+++ b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoader.java
@@ -18,19 +18,19 @@
package org.apache.hop.pipeline.transforms.cratedbbulkloader;
import com.fasterxml.jackson.core.JsonProcessingException;
+import java.io.BufferedOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
-import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.vfs2.FileObject;
import org.apache.hop.core.Const;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
@@ -60,8 +60,17 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
public static final String DATE_CONVERSION_MASK = "yyyy-MM-dd";
public static final String NUMBER_CONVERSION_MASK = "#############0.##############";
- private final BulkImportClient bulkImportClient =
- new BulkImportClient(meta.getHttpEndpoint(), meta.getHttpLogin(), meta.getHttpPassword());
+ /** Size of the buffer we put in front of the (remote) output stream. */
+ private static final int OUTPUT_BUFFER_SIZE = 128 * 1024;
+
+ /** Placeholder written instead of the AWS credentials when the COPY statement is logged. */
+ private static final String CREDENTIALS_MASK = "<credentials hidden>";
+
+ /**
+ * Only built when the HTTP endpoint is actually used, and from resolved values: at construction
+ * time the transform's variables are not available yet.
+ */
+ private BulkImportClient bulkImportClient;
public CrateDBBulkLoader(
TransformMeta transformMeta,
@@ -78,26 +87,37 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
if (super.init()) {
try {
- // Validating that the connection has been defined.
+ // Validating that the connection and the load settings have been defined.
verifyDatabaseConnection();
+ verifyLoadSettings();
data.databaseMeta = this.getPipelineMeta().findDatabase(meta.getConnection(), variables);
- if (meta.isStreamToS3Csv() && !meta.isUseHttpEndpoint()) {
+ if (meta.isUseHttpEndpoint()) {
+ bulkImportClient =
+ new BulkImportClient(
+ resolve(meta.getHttpEndpoint()),
+ resolve(meta.getHttpLogin()),
+ resolve(meta.getHttpPassword()));
+ data.maxBatchSize = resolveBatchSize();
+ } else if (meta.isStreamToS3Csv()) {
String readFromFilename = resolve(meta.getReadFromFilename());
String localPath = resolve(meta.getVolumeMapping());
- if (!StringUtils.isEmpty(localPath) && isURIOfScheme(readFromFilename, Scheme.FILE)) {
- data.writer =
- HopVfs.getOutputStream(
- FilenameUtils.concat(localPath, extractFilename(readFromFilename)), false);
- } else {
- data.writer = HopVfs.getOutputStream(readFromFilename, false);
- }
- // get the file output stream to write to S3
+ String target =
+ !StringUtils.isEmpty(localPath) && isURIOfScheme(readFromFilename, Scheme.FILE)
+ ? FilenameUtils.concat(localPath, extractFilename(readFromFilename))
+ : readFromFilename;
+ ensureParentFolderExists(target);
+ // Every field is written as a separate chunk of bytes, so a buffer in front of the
+ // (remote) stream matters a lot here.
+ data.writer =
+ new BufferedOutputStream(
+ HopVfs.getOutputStream(target, false, variables), OUTPUT_BUFFER_SIZE);
}
data.db = new Database(this, this, data.databaseMeta);
data.db.connect();
getDbFields();
+ verifyTableFields();
if (isBasic()) {
logBasic(
@@ -145,245 +165,205 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
Object[] r = getRow(); // this also waits for a previous transform to be finished.
if (r == null) { // no more input to be expected...
- if (first && meta.isTruncateTable() && !meta.isOnlyWhenHaveRows()) {
- truncateTable();
- }
-
- if (!first) {
- try {
- data.close();
- closeFile();
- if (meta.isUseHttpEndpoint()) {
- String[] columns =
- meta.getFields().stream()
- .map(CrateDBBulkLoaderField::getDatabaseField)
- .toArray(String[]::new);
- if (isBasic()) {
- data.outputRowMeta.getValueMetaList().forEach(v -> logBasic(v.toString()));
- }
- String schema = meta.getSchemaName();
- String table = meta.getTableName();
- writeBatchToCrateDB(schema, table, columns);
- } else {
- String copyStmt = buildCopyStatementSqlString();
- Connection conn = data.db.getConnection();
- int errorCount = 0;
- try (Statement stmt = conn.createStatement();
- ResultSet resultSet = stmt.executeQuery(copyStmt)) {
- while (resultSet.next()) {
- String node = resultSet.getString("node");
- String uri = resultSet.getString("uri");
- int successCount = resultSet.getInt("success_count");
- errorCount = resultSet.getInt("error_count");
- String errors = resultSet.getString("errors");
- logError(
- "Node: "
- + node
- + " URI: "
- + uri
- + " Success Count: "
- + successCount
- + " Error Count: "
- + errorCount
- + " Errors: "
- + errors);
- incrementLinesOutput(successCount);
- incrementLinesRejected(errorCount);
- }
- conn.commit();
- }
- conn.close();
- if (errorCount > 0) {
- throw new HopException(
- "Failed to COPY FROM CSV file to CrateDB: " + errorCount + " rows failed");
- }
- }
-
- } catch (SQLException sqle) {
- setErrors(1);
- stopAll();
- setOutputDone(); // signal end to receiver(s)
- throw new HopDatabaseException("Error executing COPY statements", sqle);
- } catch (IOException ioe) {
- setErrors(1);
- stopAll();
- setOutputDone(); // signal end to receiver(s)
- throw new HopTransformException("Error releasing resources", ioe);
- } catch (CrateDBHopException e) {
- throw new HopException(e);
- }
- }
-
+ endOfStream();
return false;
}
if (first) {
first = false;
+ data.rowsReceived = true;
- if (meta.isStreamToS3Csv()) {
-
- data.fieldnrs = new HashMap<>();
-
- meta.getFields(data.insertRowMeta, getTransformName(), null, null, this, metadataProvider);
-
- if (!meta.specifyFields()) {
- // write all fields in the stream to CrateDB
- // Just take the whole input row
- data.insertRowMeta = getInputRowMeta().clone();
- data.selectedRowFieldIndices = new int[data.insertRowMeta.size()];
- // TODO Serasoft
- // Is the statement below really needed??
- try {
- getDbFields();
- } catch (HopException e) {
- logError("Error getting database fields", e);
- setErrors(1);
- stopAll();
- setOutputDone(); // signal end to receiver(s)
- return false;
- }
-
- defineAllFieldsMetadataList();
- } else {
- defineSelectedFieldsMetadataList();
- }
- } else {
- if (meta.isUseHttpEndpoint()) {
- data.fieldnrs = new HashMap<>();
- if (meta.specifyFields()) {
- defineSelectedFieldsMetadataList();
- } else {
- defineAllFieldsMetadataList();
- }
- }
+ if (meta.isTruncateTable()) {
+ truncateTable();
}
- }
- data.outputRowMeta = getInputRowMeta().clone();
+ prepareRowMapping();
+ }
- if (!meta.isUseHttpEndpoint()) {
- if (meta.isStreamToS3Csv()) {
- writeRowToFile(data.outputRowMeta, r);
- }
- } else {
- appendRowAsJsonLine(data.outputRowMeta, r);
+ if (meta.isUseHttpEndpoint()) {
+ appendRowAsJsonLine(r);
try {
writeIfBatchSizeRecordsAreReached();
} catch (IOException | CrateDBHopException e) {
throw new HopException(e);
}
+ } else if (meta.isStreamToS3Csv()) {
+ writeRowToFile(r);
}
- putRow(getInputRowMeta().clone(), r);
+ putRow(data.outputRowMeta, r);
return true;
}
- private void incrementLinesRejected(int count) {
- for (int i = 0; i < count; i++) {
- incrementLinesRejected();
+ /**
+ * Flush whatever is left: close the CSV file and fire the COPY statement, or send the last HTTP
+ * batch.
+ *
+ * @throws HopException in case the load failed or resources could not be released
+ */
+ private void endOfStream() throws HopException {
+ if (!data.rowsReceived) {
+ if (meta.isTruncateTable() && !meta.isOnlyWhenHaveRows()) {
+ truncateTable();
+ }
+ return;
}
- }
- private void incrementLinesOutput(int count) {
- for (int i = 0; i < count; i++) {
- incrementLinesOutput();
+ if (meta.isUseHttpEndpoint()) {
+ try {
+ writeBatchToCrateDB();
+ } catch (IOException | CrateDBHopException e) {
+ setErrors(1);
+ stopAll();
+ setOutputDone(); // signal end to receiver(s)
+ throw new HopException(e);
+ }
+ return;
}
- }
- private void defineAllFieldsMetadataList() throws HopException {
- data.insertRowMeta = new RowMeta();
- for (int i = 0; i < meta.getFields().size(); i++) {
- int streamFieldLocation =
- data.insertRowMeta.indexOfValue(meta.getFields().get(i).getStreamField());
- if (streamFieldLocation < 0) {
- throw new HopTransformException(
- "Field ["
- + meta.getFields().get(i).getStreamField()
- + "] couldn't be found in the input stream!");
- }
-
- data.selectedRowFieldIndices[i] = streamFieldLocation;
-
- int dbFieldLocation = -1;
- for (int e = 0; e < data.dbFields.size(); e++) {
- String[] field = data.dbFields.get(e);
- if (field[0].equalsIgnoreCase(meta.getFields().get(i).getDatabaseField())) {
- dbFieldLocation = e;
- break;
+ // The file has to be complete before CrateDB reads it.
+ if (!closeFile()) {
+ setErrors(1);
+ stopAll();
+ setOutputDone(); // signal end to receiver(s)
+ throw new HopTransformException("Error releasing resources");
+ }
+
+ try {
+ String copyStmt = buildCopyStatementSqlString(false);
+ if (isDetailed()) {
+ logDetailed("Copy stmt: " + buildCopyStatementSqlString(true));
+ }
+ int errorCount = 0;
+ try (Statement stmt = data.db.getConnection().createStatement();
+ ResultSet resultSet = stmt.executeQuery(copyStmt)) {
+ while (resultSet.next()) {
+ String node = resultSet.getString("node");
+ String uri = resultSet.getString("uri");
+ int successCount = resultSet.getInt("success_count");
+ errorCount = resultSet.getInt("error_count");
+ String errors = resultSet.getString("errors");
+ logError(
+ "Node: "
+ + node
+ + " URI: "
+ + uri
+ + " Success Count: "
+ + successCount
+ + " Error Count: "
+ + errorCount
+ + " Errors: "
+ + errors);
+ incrementLinesOutput(successCount);
+ incrementLinesRejected(errorCount);
}
}
- if (dbFieldLocation < 0) {
+ data.db.commit();
+ if (errorCount > 0) {
throw new HopException(
- "Field ["
- + meta.getFields().get(i).getDatabaseField()
- + "] couldn't be found in the table!");
+ "Failed to COPY FROM CSV file to CrateDB: " + errorCount + " rows failed");
}
- IValueMeta inputValueMeta = getInputRowMeta().getValueMeta(streamFieldLocation);
+ } catch (SQLException sqle) {
+ setErrors(1);
+ stopAll();
+ setOutputDone(); // signal end to receiver(s)
+ throw new HopDatabaseException("Error executing COPY statements", sqle);
+ }
+ }
- IValueMeta insertValueMeta = inputValueMeta.clone();
- insertValueMeta.setName(data.dbFields.get(dbFieldLocation)[0]);
+ /**
+ * Resolve, once, which field of the input row ends up in which column, together with the value
+ * meta used to render it. Doing this per row is what used to make this transform slow: it cloned
+ * the whole input row meta twice for every row.
+ *
+ * @throws HopException in case a configured field is not present on the input stream
+ */
+ void prepareRowMapping() throws HopException {
+ IRowMeta inputRowMeta = getInputRowMeta();
+ data.outputRowMeta = inputRowMeta.clone();
+ data.insertRowMeta = new RowMeta();
- data.insertRowMeta.addValueMeta(insertValueMeta);
+ if (meta.isSpecifyFields()) {
+ List<CrateDBBulkLoaderField> fields = meta.getFields();
+ data.selectedRowFieldIndices = new int[fields.size()];
+ for (int i = 0; i < fields.size(); i++) {
+ CrateDBBulkLoaderField field = fields.get(i);
+ int index = inputRowMeta.indexOfValue(field.getStreamField());
+ if (index < 0) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG, "CrateDBBulkLoader.Exception.FieldRequired", field.getStreamField()));
+ }
+ data.selectedRowFieldIndices[i] = index;
- data.fieldnrs.put(
- meta.getFields().get(i).getDatabaseField().toUpperCase(), streamFieldLocation);
+ IValueMeta insertValueMeta = inputRowMeta.getValueMeta(index).clone();
+ insertValueMeta.setName(field.getDatabaseField());
+ data.insertRowMeta.addValueMeta(insertValueMeta);
+ }
+ } else {
+ // Take the whole input row, the columns are named after the fields on the stream.
+ data.selectedRowFieldIndices = new int[inputRowMeta.size()];
+ for (int i = 0; i < inputRowMeta.size(); i++) {
+ data.selectedRowFieldIndices[i] = i;
+ data.insertRowMeta.addValueMeta(inputRowMeta.getValueMeta(i).clone());
+ }
}
+
+ // Both the COPY statement and the HTTP bulk insert name these columns.
+ data.columnNames = data.insertRowMeta.getFieldNames();
}
- private void defineSelectedFieldsMetadataList() throws HopTransformException {
- // use the columns/fields mapping.
- int numberOfInsertFields = meta.getFields().size();
- data.insertRowMeta = new RowMeta();
+ private void incrementLinesRejected(int count) {
+ for (int i = 0; i < count; i++) {
+ incrementLinesRejected();
+ }
+ }
- // Cache the position of the selected fields in the row array
- data.selectedRowFieldIndices = new int[numberOfInsertFields];
- for (int i = 0; i < meta.getFields().size(); i++) {
- CrateDBBulkLoaderField vbf = meta.getFields().get(i);
- String inputFieldName = vbf.getStreamField();
- int inputFieldIdx = getInputRowMeta().indexOfValue(inputFieldName);
- if (inputFieldIdx < 0) {
- throw new HopTransformException(
- BaseMessages.getString(
- PKG, "CrateDBBulkLoader.Exception.FieldRequired", inputFieldName)); // $NON-NLS-1$
- }
- data.selectedRowFieldIndices[i] = inputFieldIdx;
- String insertFieldName = vbf.getDatabaseField();
- IValueMeta inputValueMeta = getInputRowMeta().getValueMeta(inputFieldIdx);
- if (inputValueMeta == null) {
- throw new HopTransformException(
- BaseMessages.getString(
- PKG,
- "CrateDBBulkLoader.Exception.FailedToFindField",
- vbf.getStreamField())); // $NON-NLS-1$
- }
- IValueMeta insertValueMeta = inputValueMeta.clone();
- insertValueMeta.setName(insertFieldName);
- data.insertRowMeta.addValueMeta(insertValueMeta);
- data.fieldnrs.put(meta.getFields().get(i).getDatabaseField().toUpperCase(), inputFieldIdx);
+ private void incrementLinesOutput(int count) {
+ for (int i = 0; i < count; i++) {
+ incrementLinesOutput();
}
}
private void writeIfBatchSizeRecordsAreReached()
throws HopException, CrateDBHopException, IOException {
- String batchSize = meta.getBatchSize();
+ if (data.httpBulkArgs.size() >= data.maxBatchSize) {
+ writeBatchToCrateDB();
+ }
+ }
+
+ /**
+ * Resolve the batch size once, at init time, instead of parsing it again for every row.
+ *
+ * @return the number of rows to send in one HTTP bulk request
+ * @throws HopException when the configured batch size is not a positive number
+ */
+ int resolveBatchSize() throws HopException {
+ String batchSize = resolve(meta.getBatchSize());
String expandedBatchSize = Const.expandIntegerString(batchSize);
- int maxBatchSize = Integer.parseInt(expandedBatchSize != null ? expandedBatchSize : batchSize);
- if (data.httpBulkArgs.size() >= maxBatchSize) {
- String[] columns =
- meta.getFields().stream()
- .map(CrateDBBulkLoaderField::getDatabaseField)
- .toArray(String[]::new);
- String schema = meta.getSchemaName();
- String table = meta.getTableName();
- writeBatchToCrateDB(schema, table, columns);
+ try {
+ int size = Integer.parseInt(expandedBatchSize != null ? expandedBatchSize : batchSize);
+ if (size <= 0) {
+ throw new NumberFormatException(batchSize);
+ }
+ return size;
+ } catch (NumberFormatException | NullPointerException e) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "CrateDBBulkLoaderMeta.Error.InvalidBatchSize", batchSize));
}
}
- private void writeBatchToCrateDB(String schema, String table, String[] columns)
- throws HopException, CrateDBHopException, IOException {
+ private void writeBatchToCrateDB() throws HopException, CrateDBHopException, IOException {
+ if (data.httpBulkArgs.isEmpty()) {
+ return;
+ }
try {
final HttpBulkImportResponse httpResponse =
- bulkImportClient.batchInsert(schema, table, columns, data.httpBulkArgs);
+ bulkImportClient.batchInsert(
+ resolve(meta.getSchemaName()),
+ resolve(meta.getTableName()),
+ data.columnNames,
+ data.httpBulkArgs);
for (int i = 0; i < httpResponse.outputRows(); i++) {
incrementLinesOutput();
@@ -410,8 +390,8 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
}
}
- private void appendRowAsJsonLine(IRowMeta rowMeta, Object[] row) throws HopTransformException {
- Object[] args = new Object[rowMeta.size()];
+ private void appendRowAsJsonLine(Object[] row) throws HopTransformException {
+ Object[] args = new Object[data.insertRowMeta.size()];
try {
for (int i = 0; i < data.insertRowMeta.size(); i++) {
IValueMeta v = data.insertRowMeta.getValueMeta(i);
@@ -435,30 +415,32 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
data.convertedRowMeta = data.insertRowMeta.clone();
}
+ // Whatever comes out of here is text, so the row meta the value is written with has to say so
+ // as well. Leaving the original meta in place made writing a number fail outright: by then the
+ // value was a String and the meta still claimed an Integer.
+ vc = new ValueMetaString(v.getName());
+
if (rowItem != null) {
switch (v.getType()) {
case IValueMeta.TYPE_STRING:
- convertedValue = (String) rowItem;
+ // The value can be stored as a binary string, so never cast it straight to String.
+ convertedValue = v.getString(rowItem);
break;
case IValueMeta.TYPE_INTEGER, IValueMeta.TYPE_NUMBER, IValueMeta.TYPE_BIGNUMBER:
convertedValue = String.valueOf(rowItem);
break;
case IValueMeta.TYPE_TIMESTAMP:
- vc = new ValueMetaString();
- vc.setName(v.getName());
v.setConversionMask(TIMESTAMP_CONVERSION_MASK);
vc.setConversionMask(TIMESTAMP_CONVERSION_MASK);
convertedValue = (String) vc.convertData(v, rowItem);
break;
case IValueMeta.TYPE_DATE:
- vc = new ValueMetaString();
- vc.setName(v.getName());
v.setConversionMask(DATE_CONVERSION_MASK);
vc.setConversionMask(DATE_CONVERSION_MASK);
convertedValue = (String) vc.convertData(v, rowItem);
break;
default:
- convertedValue = String.valueOf(rowItem);
+ convertedValue = v.getString(rowItem);
break;
}
}
@@ -466,7 +448,9 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
logDetailed("Field: " + v.getName() + " - Converted Value: " + convertedValue);
}
- if (vc != null && !data.convertedRowMetaReady) data.convertedRowMeta.setValueMeta(pos, vc);
+ if (!data.convertedRowMetaReady) {
+ data.convertedRowMeta.setValueMeta(pos, vc);
+ }
return convertedValue;
}
@@ -498,7 +482,14 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
return returnValue;
}
- private String buildCopyStatementSqlString() {
+ /**
+ * Build the CrateDB COPY statement for the CSV file.
+ *
+ * @param maskCredentials when true the AWS credentials are replaced by a placeholder, so the
+ * statement can safely be written to the log
+ * @return the COPY statement
+ */
+ String buildCopyStatementSqlString(boolean maskCredentials) {
final DatabaseMeta databaseMeta = data.db.getDatabaseMeta();
StringBuilder sb = new StringBuilder(150);
@@ -510,56 +501,81 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
data.db.resolve(meta.getSchemaName()),
data.db.resolve(meta.getTableName())));
- sb.append(" (");
- List<CrateDBBulkLoaderField> fieldList = meta.getFields();
- for (int i = 0; i < fieldList.size(); i++) {
- CrateDBBulkLoaderField field = fieldList.get(i);
- if (i > 0) {
- sb.append(", ").append(field.getDatabaseField());
- } else {
- sb.append(field.getDatabaseField());
+ // The CSV file has no header, so CrateDB needs to be told which columns it holds.
+ String[] columns = copyColumnNames();
+ if (columns.length > 0) {
+ sb.append(" (");
+ for (int i = 0; i < columns.length; i++) {
+ if (i > 0) {
+ sb.append(", ");
+ }
+ sb.append(columns[i]);
}
+ sb.append(")");
}
+
+ sb.append(" FROM '").append(buildCopyFromUri(maskCredentials)).append("'");
+ sb.append(" WITH (format='csv', wait_for_completion=true");
+ sb.append(", header=false");
+ sb.append(", delimiter='" + CrateDBBulkLoaderMeta.DEFAULT_CSV_DELIMITER + "'");
sb.append(")");
+ sb.append(" RETURN SUMMARY");
- String awsAccessKeyId = "";
- String awsSecretAccessKey = "";
- if (meta.isUseSystemEnvVars()) {
- awsAccessKeyId = System.getenv("AWS_ACCESS_KEY_ID");
- awsSecretAccessKey = System.getenv("AWS_SECRET_ACCESS_KEY");
- } else {
- awsAccessKeyId = resolve(meta.getAwsAccessKeyId());
- awsSecretAccessKey = resolve(meta.getAwsSecretAccessKey());
- }
+ return sb.toString();
+ }
- String filename = resolve(meta.getReadFromFilename());
+ /**
+ * CrateDB reads S3 credentials from the URI itself, so they have to be spliced into the file
+ * name.
+ *
+ * @param maskCredentials when true the credentials are replaced by a placeholder
+ * @return the URI the COPY statement reads from
+ */
+ String[] copyColumnNames() {
+ if (meta.isSpecifyFields()) {
+ return meta.getFields().stream()
+ .map(CrateDBBulkLoaderField::getDatabaseField)
+ .toArray(String[]::new);
+ }
+ if (meta.isStreamToS3Csv() && data.columnNames != null) {
+ return data.columnNames;
+ }
+ return new String[0];
+ }
- String[] parts = filename.split("://", 2);
- String uriLeft = parts[0];
- String uriRight = parts[1];
- String awsSec = awsAccessKeyId + ":" + awsSecretAccessKey;
+ private String buildCopyFromUri(boolean maskCredentials) {
+ String filename = Const.NVL(resolve(meta.getReadFromFilename()), "");
- if ("s3".equals(uriLeft)) {
- filename =
- ":".equals(awsSec)
- ? uriLeft + "://" + uriRight
- : uriLeft + "://" + awsSec + "@" + uriRight;
- } else {
- filename = uriLeft + "://" + uriRight;
+ int schemeEnd = filename.indexOf("://");
+ if (schemeEnd < 0) {
+ // Not a URI at all: nothing to splice credentials into.
+ return filename;
}
+ String scheme = filename.substring(0, schemeEnd);
+ String rest = filename.substring(schemeEnd + 3);
- sb.append(" FROM '" + filename + "'");
- sb.append(" WITH (format='csv', wait_for_completion=true");
- sb.append(", header=false");
- sb.append(", delimiter='" + CrateDBBulkLoaderMeta.DEFAULT_CSV_DELIMITER + "'");
- sb.append(")");
- sb.append(" RETURN SUMMARY");
+ if (!"s3".equals(scheme)) {
+ return filename;
+ }
+ if (maskCredentials) {
+ return scheme + "://" + CREDENTIALS_MASK + "@" + rest;
+ }
- if (isDetailed()) {
- logDetailed("Copy stmt: " + sb);
+ String awsAccessKeyId;
+ String awsSecretAccessKey;
+ if (meta.isUseSystemEnvVars()) {
+ awsAccessKeyId = System.getenv("AWS_ACCESS_KEY_ID");
+ awsSecretAccessKey = System.getenv("AWS_SECRET_ACCESS_KEY");
+ } else {
+ awsAccessKeyId = resolve(meta.getAwsAccessKeyId());
+ awsSecretAccessKey = resolve(meta.getAwsSecretAccessKey());
}
+ String credentials = Const.NVL(awsAccessKeyId, "") + ":" + Const.NVL(awsSecretAccessKey, "");
- return sb.toString();
+ // No credentials configured: let CrateDB fall back to its own configuration.
+ return ":".equals(credentials)
+ ? scheme + "://" + rest
+ : scheme + "://" + credentials + "@" + rest;
}
/**
@@ -573,15 +589,15 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
private void getDbFields() throws HopException {
data.dbFields = new ArrayList<>();
- IRowMeta rowMeta = null;
+ String schemaName = resolve(meta.getSchemaName());
+ String tableName = resolve(meta.getTableName());
- if (!StringUtils.isEmpty(resolve(meta.getSchemaName()))) {
- rowMeta = data.db.getTableFields(meta.getSchemaName() + "." + meta.getTableName());
- } else {
- rowMeta = data.db.getTableFields(meta.getTableName());
- }
+ IRowMeta rowMeta =
+ StringUtils.isEmpty(schemaName)
+ ? data.db.getTableFields(tableName)
+ : data.db.getTableFields(schemaName + "." + tableName);
try {
- if (rowMeta.isEmpty()) {
+ if (rowMeta == null || rowMeta.isEmpty()) {
throw new HopException("No fields found in table");
}
@@ -604,12 +620,82 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
}
}
+ /**
+ * Make sure the folder the staging file goes in exists. This file is the transform's own
+ * business, so there is nothing to ask the user about: on S3 a prefix only exists while an object
+ * sits under it, and a brand new path would otherwise be refused with "Parent directory ... does
+ * not exist".
+ *
+ * @param filename the file about to be written
+ * @throws HopException when the folder is missing and cannot be created
+ */
+ void ensureParentFolderExists(String filename) throws HopException {
+ try {
+ FileObject parent = HopVfs.getFileObject(filename, variables).getParent();
+ if (parent == null || parent.exists()) {
+ return;
+ }
+ parent.createFolder();
+ if (isDetailed()) {
+ logDetailed("Created the parent folder " + HopVfs.getFriendlyURI(parent));
+ }
+ } catch (Exception e) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "CrateDBBulkLoaderMeta.Error.CannotCreateParentFolder", filename),
+ e);
+ }
+ }
+
+ /**
+ * The HTTP endpoint needs a URL and the COPY statement needs a file, so fail before any row is
+ * read when either is missing.
+ *
+ * @throws HopException when the load settings are incomplete
+ */
+ protected void verifyLoadSettings() throws HopException {
+ if (meta.isUseHttpEndpoint()) {
+ if (StringUtils.isEmpty(resolve(meta.getHttpEndpoint()))) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "CrateDBBulkLoaderMeta.Error.NoHttpEndpoint"));
+ }
+ } else if (StringUtils.isEmpty(resolve(meta.getReadFromFilename()))) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "CrateDBBulkLoaderMeta.Error.NoReadFromFilename"));
+ }
+ }
+
+ /**
+ * Fail before any row is read when a column was mapped to a table column that does not exist. The
+ * COPY statement would reject it anyway, but only after the whole file was written.
+ *
+ * @throws HopException when a selected column is not a column of the target table
+ */
+ protected void verifyTableFields() throws HopException {
+ if (!meta.isSpecifyFields()) {
+ return;
+ }
+ for (CrateDBBulkLoaderField field : meta.getFields()) {
+ boolean found = false;
+ for (String[] dbField : data.dbFields) {
+ if (dbField[0].equalsIgnoreCase(field.getDatabaseField())) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ throw new HopException(
+ "Field [" + field.getDatabaseField() + "] couldn't be found in the table!");
+ }
+ }
+ }
+
/**
* Initialize the binary values of delimiters, enclosures, and escape characters
*
* @throws HopException
*/
- private void initBinaryDataFields() throws HopException {
+ void initBinaryDataFields() throws HopException {
try {
data.binarySeparator = new byte[] {};
data.binaryEnclosure = new byte[] {};
@@ -638,21 +724,19 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
* @param row The input row
* @throws HopTransformException
*/
- private void writeRowToFile(IRowMeta rowMeta, Object[] row) throws HopTransformException {
+ void writeRowToFile(Object[] row) throws HopTransformException {
try {
+ byte[] nullString = meta.isSpecifyFields() ? data.binaryNullValue : null;
for (int i = 0; i < data.insertRowMeta.size(); i++) {
- if (i > 0 && data.binarySeparator.length > 0) {
+ if (i > 0) {
data.writer.write(data.binarySeparator);
}
- Object convertedValue = null;
IValueMeta v = data.insertRowMeta.getValueMeta(i);
- convertedValue = convertDatatypeIfNeeded(v, row[data.selectedRowFieldIndices[i]], i);
+ Object convertedValue = convertDatatypeIfNeeded(v, row[data.selectedRowFieldIndices[i]], i);
writeField(
- data.convertedRowMeta.getValueMeta(i),
- convertedValue,
- (!meta.isSpecifyFields() ? null : data.binaryNullValue));
+ data.convertedRowMeta.getValueMeta(i), convertedValue, nullString, enclosedByType(v));
}
data.convertedRowMetaReady = true;
data.writer.write(data.binaryNewline);
@@ -669,7 +753,16 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
* @param nullString The bytes to put in the temp file if the value is null
* @throws HopTransformException
*/
- private void writeField(IValueMeta v, Object valueData, byte[] nullString)
+ /**
+ * @param v the value meta of the field on the stream, before it was rendered as text
+ * @return true if a value of this type is enclosed whatever it holds
+ */
+ private boolean enclosedByType(IValueMeta v) {
+ // A number or a boolean can never hold a separator, a quote or a line break.
+ return !v.isNumeric() && v.getType() != IValueMeta.TYPE_BOOLEAN;
+ }
+
+ private void writeField(IValueMeta v, Object valueData, byte[] nullString, boolean enclosedByType)
throws HopTransformException {
try {
byte[] str;
@@ -684,42 +777,15 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
}
if (str != null && str.length > 0) {
- List<Integer> enclosures = null;
- boolean writeEnclosures = false;
-
- if (v.isString()) {
- writeEnclosures = true;
-
- if (containsSeparatorOrEnclosure(
- str, data.binarySeparator, data.binaryEnclosure, data.escapeCharacters)) {
- writeEnclosures = true;
- }
- }
-
- if (writeEnclosures) {
+ // Strings are always enclosed, the COPY statement reads the same quote character.
+ // Anything else is enclosed only when its own content would otherwise break the row --
+ // JSON above all, which Hop does not treat as a string yet is full of commas and quotes.
+ if (enclosedByType || needsEnclosure(str)) {
data.writer.write(data.binaryEnclosure);
- enclosures = getEnclosurePositions(str);
- }
-
- if (enclosures == null) {
- data.writer.write(str);
- } else {
- // Skip the enclosures, escape them instead...
- int from = 0;
- for (Integer enclosure : enclosures) {
- // Minus one to write the escape before the enclosure
- int position = enclosure;
- data.writer.write(str, from, position - from);
- data.writer.write(data.escapeCharacters); // write enclosure a second time
- from = position;
- }
- if (from < str.length) {
- data.writer.write(str, from, str.length - from);
- }
- }
-
- if (writeEnclosures) {
+ writeEscaped(str);
data.writer.write(data.binaryEnclosure);
+ } else {
+ data.writer.write(str);
}
}
} catch (Exception e) {
@@ -807,137 +873,58 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
}
/**
- * Check if a string contains separators or enclosures. Can be used to determine if the string
- * needs enclosures around it or not.
+ * Whether a value has to be enclosed to survive the trip through the CSV file: it is only safe to
+ * write bare when it holds no separator, no quote and no line break.
*
- * @param source The string to check
- * @param separator The separator character(s)
- * @param enclosure The enclosure character(s)
- * @param escape The escape character(s)
- * @return True if the string contains separators or enclosures
+ * @param str The bytes of the value
+ * @return true if the value must be enclosed
*/
- @SuppressWarnings("Duplicates")
- private boolean containsSeparatorOrEnclosure(
- byte[] source, byte[] separator, byte[] enclosure, byte[] escape) {
- boolean result = false;
-
- boolean enclosureExists = enclosure != null && enclosure.length > 0;
- boolean separatorExists = separator != null && separator.length > 0;
- boolean escapeExists = escape != null && escape.length > 0;
-
- // Skip entire test if neither separator nor enclosure exist
- if (separatorExists || enclosureExists || escapeExists) {
-
- // Search for the first occurrence of the separator or enclosure
- for (int index = 0; !result && index < source.length; index++) {
- if (enclosureExists && source[index] == enclosure[0]) {
-
- // Potential match found, make sure there are enough bytes to support a full match
- if (index + enclosure.length <= source.length) {
- // First byte of enclosure found
- result = true; // Assume match
- for (int i = 1; i < enclosure.length; i++) {
- if (source[index + i] != enclosure[i]) {
- // Enclosure match is proven false
- result = false;
- break;
- }
- }
- }
-
- } else if (separatorExists && source[index] == separator[0]) {
-
- // Potential match found, make sure there are enough bytes to support a full match
- if (index + separator.length <= source.length) {
- // First byte of separator found
- result = true; // Assume match
- for (int i = 1; i < separator.length; i++) {
- if (source[index + i] != separator[i]) {
- // Separator match is proven false
- result = false;
- break;
- }
- }
- }
+ private boolean needsEnclosure(byte[] str) {
+ byte separator = data.binarySeparator.length > 0 ? data.binarySeparator[0] : 0;
+ byte enclosure = data.binaryEnclosure.length > 0 ? data.binaryEnclosure[0] : 0;
- } else if (escapeExists
- && source[index] == escape[0]
- && index + escape.length <= source.length) {
- // Potential match found, make sure there are enough bytes to support a full match
- // First byte of separator found
- result = true; // Assume match
- for (int i = 1; i < escape.length; i++) {
- if (source[index + i] != escape[i]) {
- // Separator match is proven false
- result = false;
- break;
- }
- }
- }
+ for (byte b : str) {
+ if (b == separator || b == enclosure || b == '\n' || b == '\r') {
+ return true;
}
}
- return result;
+ return false;
}
/**
- * Gets the positions of any double quotes or backslashes in the string
+ * Write the value, doubling every occurrence of the enclosure character so the CSV file stays
+ * readable for the COPY statement. Only bytes that actually need escaping are copied separately,
+ * a value without enclosures is written in one go.
*
- * @param str The string to check
- * @return The positions within the string of double quotes and backslashes.
+ * @param str The bytes of the value
+ * @throws IOException in case the value could not be written
*/
- private List<Integer> getEnclosurePositions(byte[] str) {
- List<Integer> positions = null;
- // +1 because otherwise we will not find it at the end
- for (int i = 0, len = str.length; i < len; i++) {
- // verify if on position i there is an enclosure
- //
- boolean found = true;
- for (int x = 0; found && x < data.binaryEnclosure.length; x++) {
- if (str[i + x] != data.binaryEnclosure[x]) {
- found = false;
- }
- }
-
- if (!found) {
- found = true;
- for (int x = 0; found && x < data.escapeCharacters.length; x++) {
- if (str[i + x] != data.escapeCharacters[x]) {
- found = false;
- }
- }
- }
+ private void writeEscaped(byte[] str) throws IOException {
+ if (data.binaryEnclosure.length == 0) {
+ data.writer.write(str);
+ return;
+ }
+ byte enclosure = data.binaryEnclosure[0];
- if (found) {
- if (positions == null) {
- positions = new ArrayList<>();
- }
- positions.add(i);
+ int from = 0;
+ for (int i = 0; i < str.length; i++) {
+ if (str[i] == enclosure) {
+ data.writer.write(str, from, i - from);
+ data.writer.write(data.escapeCharacters); // write the enclosure a second time
+ from = i;
}
}
- return positions;
+ data.writer.write(str, from, str.length - from);
}
@Override
public void stopRunning() throws HopException {
setStopped(true);
- if (data.workerThread != null) {
- synchronized (data.workerThread) {
- if (data.workerThread.isAlive() && !data.workerThread.isInterrupted()) {
- try {
- data.workerThread.interrupt();
- data.workerThread.join();
- } catch (InterruptedException e) { // Checkstyle:OFF:
- }
- // Checkstyle:ONN:
- }
- }
- }
-
super.stopRunning();
}
void truncateTable() throws HopDatabaseException {
- if (meta.isTruncateTable() && ((getCopy() == 0) || !Utils.isEmpty(getPartitionId()))) {
+ if ((getCopy() == 0) || !Utils.isEmpty(getPartitionId())) {
data.db.truncateTable(resolve(meta.getSchemaName()), resolve(meta.getTableName()));
}
}
@@ -947,23 +934,17 @@ public class CrateDBBulkLoader extends BaseTransform<CrateDBBulkLoaderMeta, Crat
setOutputDone();
- try {
- if (getErrors() > 0) {
- data.db.rollback();
- }
- } catch (HopDatabaseException e) {
- logError("Unexpected error rolling back the database connection.", e);
- }
+ // The file is normally closed before the COPY statement runs, this catches the error paths.
+ closeFile();
- if (data.workerThread != null) {
+ if (data.db != null) {
try {
- data.workerThread.join();
- } catch (InterruptedException e) { // Checkstyle:OFF:
+ if (getErrors() > 0) {
+ data.db.rollback();
+ }
+ } catch (HopDatabaseException e) {
+ logError("Unexpected error rolling back the database connection.", e);
}
- // Checkstyle:ONN:
- }
-
- if (data.db != null) {
data.db.disconnect();
}
super.dispose();
diff --git a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderData.java b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderData.java
index c65d9e794e..fca6073c51 100644
--- a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderData.java
+++ b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderData.java
@@ -17,12 +17,9 @@
package org.apache.hop.pipeline.transforms.cratedbbulkloader;
-import java.io.IOException;
import java.io.OutputStream;
-import java.io.PipedInputStream;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.row.IRowMeta;
@@ -47,8 +44,14 @@ public class CrateDBBulkLoaderData extends BaseTransformData implements ITransfo
// data type
public ArrayList<String[]> dbFields;
- // Maps table fields to the location of the corresponding field on the input stream.
- public Map<String, Integer> fieldnrs;
+ /** The names of the columns we load, in the order the CSV file and the bulk insert use them. */
+ protected String[] columnNames;
+
+ /** Rows per HTTP bulk request, resolved once at init time. */
+ protected int maxBatchSize;
+
+ /** Set as soon as a first row is read, so we know whether the stream was empty. */
+ protected boolean rowsReceived;
protected OutputStream writer;
@@ -60,10 +63,6 @@ public class CrateDBBulkLoaderData extends BaseTransformData implements ITransfo
public byte[] binaryNewline;
public byte[] binaryNullValue;
- protected PipedInputStream pipedInputStream;
-
- protected volatile Thread workerThread;
-
public CrateDBBulkLoaderData() {
super();
@@ -77,8 +76,4 @@ public class CrateDBBulkLoaderData extends BaseTransformData implements ITransfo
public void setDatabaseMeta(DatabaseMeta databaseMeta) {
this.databaseMeta = databaseMeta;
}
-
- public void close() throws IOException {
- // Do nothing
- }
}
diff --git a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderDialog.java b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderDialog.java
index c41ef87858..e6d53889bc 100644
--- a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderDialog.java
+++ b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderDialog.java
@@ -431,6 +431,7 @@ public class CrateDBBulkLoaderDialog extends BaseTransformDialog {
new SelectionAdapter() {
@Override
public void widgetSelected(SelectionEvent e) {
+ input.setChanged();
toggleKeysSelection();
}
});
@@ -1020,43 +1021,23 @@ public class CrateDBBulkLoaderDialog extends BaseTransformDialog {
/** Copy information from the meta-data input to the dialog fields. */
public void getData() {
- if (!StringUtils.isEmpty(input.getConnection())) {
- wConnection.setText(input.getConnection());
- }
- if (!StringUtils.isEmpty(input.getSchemaName())) {
- wSchema.setText(input.getSchemaName());
- }
- if (!StringUtils.isEmpty(input.getTableName())) {
- wTable.setText(input.getTableName());
- }
+ wConnection.setText(Const.NVL(input.getConnection(), ""));
+ wSchema.setText(Const.NVL(input.getSchemaName(), ""));
+ wTable.setText(Const.NVL(input.getTableName(), ""));
wUseHTTPEndpoint.setSelection(input.isUseHttpEndpoint());
-
- if (input.isUseHttpEndpoint()) {
- wHttpEndpoint.setText(input.getHttpEndpoint());
- wBatchSize.setText(Const.NVL(input.getBatchSize(), ""));
- wHttpLogin.setText(Const.NVL(input.getHttpLogin(), ""));
- wHttpPassword.setText(Const.NVL(input.getHttpPassword(), ""));
- }
+ wHttpEndpoint.setText(Const.NVL(input.getHttpEndpoint(), ""));
+ wBatchSize.setText(Const.NVL(input.getBatchSize(), ""));
+ wHttpLogin.setText(Const.NVL(input.getHttpLogin(), ""));
+ wHttpPassword.setText(Const.NVL(input.getHttpPassword(), ""));
wUseSystemVars.setSelection(input.isUseSystemEnvVars());
- if (!input.isUseSystemEnvVars()) {
- if (!StringUtil.isEmpty(input.getAwsAccessKeyId())) {
- wAccessKeyId.setText(input.getAwsAccessKeyId());
- }
- if (!StringUtils.isEmpty(input.getAwsSecretAccessKey())) {
- wSecretAccessKey.setText(input.getAwsSecretAccessKey());
- }
- }
+ wAccessKeyId.setText(Const.NVL(input.getAwsAccessKeyId(), ""));
+ wSecretAccessKey.setText(Const.NVL(input.getAwsSecretAccessKey(), ""));
wStreamToRemoteCsv.setSelection(input.isStreamToS3Csv());
- if (!StringUtils.isEmpty(input.getReadFromFilename())) {
- wReadFromFilename.setText(input.getReadFromFilename());
- }
-
- if (!StringUtils.isEmpty(input.getVolumeMapping())) {
- wLocalRemoteVolumeMapping.setText(input.getVolumeMapping());
- }
+ wReadFromFilename.setText(Const.NVL(input.getReadFromFilename(), ""));
+ wLocalRemoteVolumeMapping.setText(Const.NVL(input.getVolumeMapping(), ""));
// wTruncate.setSelection(input.isTruncateTable());
// wOnlyWhenHaveRows.setSelection(input.isOnlyWhenHaveRows());
@@ -1083,44 +1064,22 @@ public class CrateDBBulkLoaderDialog extends BaseTransformDialog {
private void getInfo(CrateDBBulkLoaderMeta info) {
- if (!StringUtils.isEmpty(wConnection.getText())) {
- info.setConnection(wConnection.getText());
- }
-
- if (!StringUtils.isEmpty(wSchema.getText())) {
- info.setSchemaName(wSchema.getText());
- }
-
- if (!StringUtils.isEmpty(wTable.getText())) {
- info.setTablename(wTable.getText());
- }
+ info.setConnection(wConnection.getText());
+ info.setSchemaName(wSchema.getText());
+ info.setTablename(wTable.getText());
info.setUseHttpEndpoint(wUseHTTPEndpoint.getSelection());
- if (wUseHTTPEndpoint.getSelection()) {
- info.setHttpEndpoint(wHttpEndpoint.getText());
- info.setBatchSize(wBatchSize.getText());
- info.setHttpLogin(wHttpLogin.getText());
- info.setHttpPassword(wHttpPassword.getText());
- }
+ info.setHttpEndpoint(wHttpEndpoint.getText());
+ info.setBatchSize(wBatchSize.getText());
+ info.setHttpLogin(wHttpLogin.getText());
+ info.setHttpPassword(wHttpPassword.getText());
- if (wUseSystemVars.getSelection()) {
- info.setUseSystemEnvVars(wUseSystemVars.getSelection());
- } else {
- info.setUseSystemEnvVars(wUseSystemVars.getSelection());
- if (!StringUtils.isEmpty(wAccessKeyId.getText())) {
- info.setAwsAccessKeyId(wAccessKeyId.getText());
- }
- if (!StringUtil.isEmpty(wSecretAccessKey.getText())) {
- info.setAwsSecretAccessKey(wSecretAccessKey.getText());
- }
- }
- // info.setTruncateTable(wTruncate.getSelection());
- // info.setOnlyWhenHaveRows(wOnlyWhenHaveRows.getSelection());
- info.setStreamToS3Csv(wStreamToRemoteCsv.getSelection());
+ info.setUseSystemEnvVars(wUseSystemVars.getSelection());
+ info.setAwsAccessKeyId(wAccessKeyId.getText());
+ info.setAwsSecretAccessKey(wSecretAccessKey.getText());
- if (!StringUtils.isEmpty(wReadFromFilename.getText())) {
- info.setReadFromFilename(wReadFromFilename.getText());
- }
+ info.setStreamToS3Csv(wStreamToRemoteCsv.getSelection());
+ info.setReadFromFilename(wReadFromFilename.getText());
info.setSpecifyFields(wSpecifyFields.getSelection());
diff --git a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderMeta.java b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderMeta.java
index 093fae2bac..9fb4402057 100644
--- a/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderMeta.java
+++ b/plugins/databases/cratedb/src/main/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderMeta.java
@@ -117,7 +117,8 @@ public class CrateDBBulkLoaderMeta
@HopMetadataProperty(
key = "aws_secret_access_key",
injectionKey = "AWS_SECRET_ACCESS_KEY",
- injectionKeyDescription = "")
+ injectionKeyDescription = "",
+ password = true)
private String awsSecretAccessKey;
@HopMetadataProperty(
@@ -190,16 +191,6 @@ public class CrateDBBulkLoaderMeta
/** Fields containing the values in the input stream to insert */
private List<CrateDBBulkLoaderField> fields;
- @HopMetadataProperty(
- groupKey = "fields",
- key = "field",
- injectionGroupKey = "FIELDS",
- injectionGroupDescription = "CrateDBBulkLoader.Injection.FIELDS",
- injectionKey = "FIELDDATABASE",
- injectionKeyDescription = "CrateDBBulkLoader.Injection.FIELDDATABASE")
- /** Fields in the table to insert */
- private String[] fieldDatabase;
-
public CrateDBBulkLoaderMeta() {
super(); // allocate BaseTransformMeta
@@ -589,10 +580,10 @@ public class CrateDBBulkLoaderMeta
}
} else {
// Specifying the column names explicitly
- for (int i = 0; i < getFieldDatabase().length; i++) {
- int idx = r.indexOfValue(getFieldDatabase()[i]);
+ for (CrateDBBulkLoaderField vbf : fields) {
+ int idx = r.indexOfValue(vbf.getDatabaseField());
if (idx < 0) {
- error_message += "\t\t" + getFieldDatabase()[i] + Const.CR;
+ error_message += "\t\t" + vbf.getDatabaseField() + Const.CR;
error_found = true;
}
}
@@ -621,7 +612,7 @@ public class CrateDBBulkLoaderMeta
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) {
@@ -894,20 +885,6 @@ public class CrateDBBulkLoaderMeta
}
}
- /**
- * @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;
- }
-
/**
* @return the schemaName
*/
diff --git a/plugins/databases/cratedb/src/main/resources/org/apache/hop/pipeline/transforms/cratedbbulkloader/messages/messages_en_US.properties b/plugins/databases/cratedb/src/main/resources/org/apache/hop/pipeline/transforms/cratedbbulkloader/messages/messages_en_US.properties
index 6d5e7dc1fb..b36250424a 100644
--- a/plugins/databases/cratedb/src/main/resources/org/apache/hop/pipeline/transforms/cratedbbulkloader/messages/messages_en_US.properties
+++ b/plugins/databases/cratedb/src/main/resources/org/apache/hop/pipeline/transforms/cratedbbulkloader/messages/messages_en_US.properties
@@ -121,6 +121,10 @@ CrateDBBulkLoaderMeta.CheckResult.TableOk=Table [{0}] is readeable and we got th
CrateDBBulkLoaderMeta.CheckResult.UndefinedError=An error occurred\: {0}
CrateDBBulkLoaderMeta.Error.ErrorConnecting=I was unable to connect to the database to verify the status of the table\: {0}
CrateDBBulkLoaderMeta.Error.NoConnection=There is no connection defined in this transform.
+CrateDBBulkLoaderMeta.Error.CannotCreateParentFolder=Unable to create the folder for the file ''{0}''.
+CrateDBBulkLoaderMeta.Error.NoHttpEndpoint=There is no HTTP endpoint defined in this transform. The bulk import needs a URL to post to.
+CrateDBBulkLoaderMeta.Error.NoReadFromFilename=There is no file name defined in this transform. The COPY statement needs a file to load from.
+CrateDBBulkLoaderMeta.Error.InvalidBatchSize=The batch size ''{0}'' is not a positive number.
CrateDBBulkLoaderMeta.Error.NoInput=Not receiving any fields from previous transforms. Check the previous transforms for errors & the connecting hops.
CrateDBBulkLoaderMeta.Error.NoTable=No table is defined on this connection.
CrateDBBulkLoaderMeta.Exception.ConnectionNotDefined=Unable to determine the required fields because the database connection wasn''t defined.
diff --git a/plugins/databases/cratedb/src/test/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderTest.java b/plugins/databases/cratedb/src/test/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderTest.java
new file mode 100644
index 0000000000..26d5e568ec
--- /dev/null
+++ b/plugins/databases/cratedb/src/test/java/org/apache/hop/pipeline/transforms/cratedbbulkloader/CrateDBBulkLoaderTest.java
@@ -0,0 +1,388 @@
+/*
+ * 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.cratedbbulkloader;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+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 com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.List;
+import java.util.TimeZone;
+import org.apache.hop.core.database.Database;
+import org.apache.hop.core.database.DatabaseMeta;
+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.ValueMetaDate;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaJson;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Unit tests around the COPY statement and the field mapping of the CrateDB bulk loader. */
+class CrateDBBulkLoaderTest {
+
+ private TransformMockHelper<CrateDBBulkLoaderMeta, CrateDBBulkLoaderData> helper;
+ private CrateDBBulkLoaderMeta meta;
+ private CrateDBBulkLoaderData data;
+ private CrateDBBulkLoader transform;
+
+ @BeforeEach
+ void setUp() {
+ helper =
+ new TransformMockHelper<>(
+ "CrateDB bulk loader", CrateDBBulkLoaderMeta.class, CrateDBBulkLoaderData.class);
+ when(helper.logChannelFactory.create(any(), any())).thenReturn(helper.iLogChannel);
+ when(helper.logChannelFactory.create(any())).thenReturn(helper.iLogChannel);
+
+ meta = new CrateDBBulkLoaderMeta();
+ meta.setConnection("cratedb");
+ meta.setSchemaName("doc");
+ meta.setTablename("orders");
+ meta.setReadFromFilename("s3://bucket/orders.csv");
+
+ data = new CrateDBBulkLoaderData();
+
+ DatabaseMeta databaseMeta = mock(DatabaseMeta.class);
+ when(databaseMeta.getQuotedSchemaTableCombination(any(), any(), any()))
+ .thenReturn("\"doc\".\"orders\"");
+
+ data.db = mock(Database.class);
+ when(data.db.getDatabaseMeta()).thenReturn(databaseMeta);
+ when(data.db.resolve(anyString())).thenAnswer(invocation -> invocation.getArgument(0));
+
+ transform =
+ new CrateDBBulkLoader(
+ helper.transformMeta, meta, data, 0, helper.pipelineMeta, helper.pipeline);
+ }
+
+ @AfterEach
+ void tearDown() {
+ helper.cleanUp();
+ }
+
+ /** A local path has no "://" in it, which used to blow up on split()[1]. */
+ @Test
+ void buildsTheCopyStatementForAFileNameThatIsNotAUri() {
+ meta.setReadFromFilename("/data/orders.csv");
+ prepare(rowMeta("name"));
+
+ String sql = assertDoesNotThrow(() -> transform.buildCopyStatementSqlString(false));
+
+ assertTrue(sql.contains(" FROM '/data/orders.csv'"), sql);
+ }
+
+ /** And an unset file name should not NPE either. */
+ @Test
+ void refusesToStartWithoutAFileName() {
+ meta.setReadFromFilename(null);
+
+ assertThrows(HopException.class, () -> transform.verifyLoadSettings());
+ }
+
+ @Test
+ void refusesToStartWithoutAnHttpEndpoint() {
+ meta.setUseHttpEndpoint(true);
+ meta.setHttpEndpoint("");
+
+ assertThrows(HopException.class, () -> transform.verifyLoadSettings());
+ }
+
+ /** CrateDB reads the S3 credentials from the URI, so they have to end up in it, resolved. */
+ @Test
+ void resolvesVariablesInTheCredentials() {
+ transform.setVariable("AWS_KEY", "AKIAEXAMPLE");
+ transform.setVariable("AWS_SECRET", "s3cr3t");
+ meta.setAwsAccessKeyId("${AWS_KEY}");
+ meta.setAwsSecretAccessKey("${AWS_SECRET}");
+ prepare(rowMeta("name"));
+
+ assertTrue(
+ transform
+ .buildCopyStatementSqlString(false)
+ .contains(" FROM 's3://AKIAEXAMPLE:s3cr3t@bucket/orders.csv'"),
+ transform.buildCopyStatementSqlString(false));
+ }
+
+ /** Without credentials the URI stays clean so CrateDB can use its own configuration. */
+ @Test
+ void leavesTheUriAloneWithoutCredentials() {
+ prepare(rowMeta("name"));
+
+ assertTrue(
+ transform.buildCopyStatementSqlString(false).contains(" FROM 's3://bucket/orders.csv'"),
+ transform.buildCopyStatementSqlString(false));
+ }
+
+ /** The statement ends up in the log on detailed level, the credentials should not. */
+ @Test
+ void masksTheCredentialsInTheLoggedStatement() {
+ meta.setAwsAccessKeyId("AKIAEXAMPLE");
+ meta.setAwsSecretAccessKey("s3cr3t");
+ prepare(rowMeta("name"));
+
+ String logged = transform.buildCopyStatementSqlString(true);
+
+ assertFalse(logged.contains("AKIAEXAMPLE"), logged);
+ assertFalse(logged.contains("s3cr3t"), logged);
+ }
+
+ /**
+ * The CSV file carries no header, so the COPY statement has to name the columns. It used to emit
+ * an empty "()" whenever the fields were not specified explicitly.
+ */
+ @Test
+ void namesTheColumnsOfTheWholeInputRowWhenNoFieldsAreSpecified() {
+ prepare(rowMeta("name", "amount"));
+
+ assertTrue(
+ transform.buildCopyStatementSqlString(false).contains("\"doc\".\"orders\" (name, amount)"),
+ transform.buildCopyStatementSqlString(false));
+ }
+
+ @Test
+ void namesTheConfiguredColumnsWhenFieldsAreSpecified() {
+ meta.setSpecifyFields(true);
+ meta.setFields(
+ List.of(
+ new CrateDBBulkLoaderField("order_amount", "amount"),
+ new CrateDBBulkLoaderField("order_name", "name")));
+ prepare(rowMeta("name", "amount"));
+
+ assertTrue(
+ transform
+ .buildCopyStatementSqlString(false)
+ .contains("\"doc\".\"orders\" (order_amount, order_name)"),
+ transform.buildCopyStatementSqlString(false));
+ }
+
+ /** The mapping picks the fields out of the row in the configured order. */
+ @Test
+ void mapsTheSelectedFieldsToTheirPositionOnTheStream() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(
+ List.of(
+ new CrateDBBulkLoaderField("order_amount", "amount"),
+ new CrateDBBulkLoaderField("order_name", "name")));
+ prepare(rowMeta("name", "amount"));
+
+ assertArrayEquals(new int[] {1, 0}, data.selectedRowFieldIndices);
+ assertArrayEquals(new String[] {"order_amount", "order_name"}, data.columnNames);
+ }
+
+ /** Without an explicit mapping the whole row is loaded, in stream order. */
+ @Test
+ void mapsTheWholeRowWhenNoFieldsAreSpecified() {
+ prepare(rowMeta("name", "amount"));
+
+ assertArrayEquals(new int[] {0, 1}, data.selectedRowFieldIndices);
+ assertArrayEquals(new String[] {"name", "amount"}, data.columnNames);
+ assertEquals(2, data.insertRowMeta.size());
+ }
+
+ /** A configured field that is not on the stream is an error, not an empty column. */
+ @Test
+ void failsOnAFieldThatIsNotOnTheStream() {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new CrateDBBulkLoaderField("order_note", "note")));
+ transform.setInputRowMeta(rowMeta("name"));
+
+ assertThrows(HopException.class, () -> transform.prepareRowMapping());
+ }
+
+ /** A column that does not exist in the table has to be caught before the file is written. */
+ @Test
+ void refusesToStartWhenAColumnIsNotInTheTable() {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new CrateDBBulkLoaderField("order_totl", "amount")));
+ data.dbFields = new ArrayList<>();
+ data.dbFields.add(new String[] {"ORDER_ID", "INTEGER"});
+
+ HopException e = assertThrows(HopException.class, () -> transform.verifyTableFields());
+ assertTrue(e.getMessage().contains("order_totl"), e.getMessage());
+ }
+
+ // ------------------------------------------------------------ the CSV write path
+
+ /** The plain case: values separated, strings enclosed. */
+ @Test
+ void writesTheRowAsCsv() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+ rowMeta.addValueMeta(new ValueMetaInteger("amount"));
+
+ assertEquals("\"Acme\",42\n", writeRow(rowMeta, new Object[] {"Acme", 42L}));
+ }
+
+ /** A value carrying the separator has to be enclosed or it becomes two columns. */
+ @Test
+ void enclosesValuesHoldingTheSeparator() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+
+ assertEquals("\"Acme, Inc\"\n", writeRow(rowMeta, new Object[] {"Acme, Inc"}));
+ }
+
+ /** Quotes inside a value are doubled, the way the COPY statement expects. */
+ @Test
+ void doublesQuotesInsideAValue() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+
+ assertEquals("\"a\"\"b\"\n", writeRow(rowMeta, new Object[] {"a\"b"}));
+ }
+
+ /**
+ * JSON is not a string as far as Hop is concerned, so it used to be written bare: its commas
+ * split it into extra columns. Hop pretty prints it across lines by default too.
+ */
+ @Test
+ void enclosesJsonSoItSurvivesTheCsvRow() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaJson("payload"));
+ JsonNode json = new ObjectMapper().readTree("{\"a\":\"x\",\"b\":{\"c\":[1,2]}}");
+
+ String written = writeRow(rowMeta, new Object[] {json});
+
+ assertTrue(written.startsWith("\""), written);
+ assertTrue(written.endsWith("\"\n"), written);
+ assertTrue(written.contains("\"\"a\"\""), written);
+ }
+
+ /** Dates are rendered with the mask the COPY statement reads them back with. */
+ @Test
+ void writesDatesInTheDeclaredFormat() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaDate("event_date"));
+
+ Calendar calendar = Calendar.getInstance(TimeZone.getDefault());
+ calendar.clear();
+ calendar.set(2026, Calendar.AUGUST, 24, 15, 30, 45);
+
+ assertEquals("\"2026-08-24\"\n", writeRow(rowMeta, new Object[] {calendar.getTime()}));
+ }
+
+ /** Numbers stay bare, so the common case pays nothing for the enclosure logic. */
+ @Test
+ void doesNotEncloseValuesThatCannotBreakTheRow() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("amount"));
+
+ assertEquals("42\n", writeRow(rowMeta, new Object[] {42L}));
+ }
+
+ // ------------------------------------------------------------ batch size
+
+ @Test
+ void resolvesTheBatchSize() throws Exception {
+ meta.setBatchSize("500");
+ assertEquals(500, transform.resolveBatchSize());
+ }
+
+ @Test
+ void resolvesTheBatchSizeFromAVariable() throws Exception {
+ transform.setVariable("BATCH", "250");
+ meta.setBatchSize("${BATCH}");
+ assertEquals(250, transform.resolveBatchSize());
+ }
+
+ /** The docs say the batch size has no default, so a missing or silly one must be an error. */
+ @Test
+ void refusesABatchSizeThatIsNotAPositiveNumber() {
+ meta.setBatchSize("");
+ assertThrows(HopException.class, () -> transform.resolveBatchSize());
+ meta.setBatchSize("0");
+ assertThrows(HopException.class, () -> transform.resolveBatchSize());
+ meta.setBatchSize("not a number");
+ assertThrows(HopException.class, () -> transform.resolveBatchSize());
+ }
+
+ // ------------------------------------------------------------ existing files
+
+ /**
+ * A file somebody else wrote is the one case we cannot speak for: the trigger stream says nothing
+ * about its columns, so the statement has to stay silent.
+ */
+ @Test
+ void namesNoColumnsForAnExistingFileWithoutAMapping() throws Exception {
+ meta.setStreamToS3Csv(false);
+ meta.setSpecifyFields(false);
+ prepare(rowMeta("trigger"));
+
+ assertEquals(0, transform.copyColumnNames().length);
+ }
+
+ // ------------------------------------------------------------ staging folder
+
+ @Test
+ void createsTheStagingFolderWhenItDoesNotExistYet(@TempDir Path tempDir) throws Exception {
+ Path missing = tempDir.resolve("not-there-yet/nested");
+ assertFalse(Files.exists(missing));
+
+ transform.ensureParentFolderExists(missing.resolve("stage.csv").toString());
+
+ assertTrue(Files.isDirectory(missing));
+ }
+
+ private String writeRow(IRowMeta rowMeta, Object[] row) throws Exception {
+ meta.setStreamToS3Csv(true);
+ meta.setSpecifyFields(false);
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ data.writer = out;
+ transform.initBinaryDataFields();
+ transform.setInputRowMeta(rowMeta);
+ transform.prepareRowMapping();
+ transform.writeRowToFile(row);
+
+ return out.toString(StandardCharsets.UTF_8);
+ }
+
+ private IRowMeta rowMeta(String... names) {
+ IRowMeta rowMeta = new RowMeta();
+ for (String name : names) {
+ rowMeta.addValueMeta(
+ "amount".equals(name) ? new ValueMetaInteger(name) : new ValueMetaString(name));
+ }
+ return rowMeta;
+ }
+
+ private void prepare(IRowMeta rowMeta) {
+ transform.setInputRowMeta(rowMeta);
+ assertDoesNotThrow(() -> transform.prepareRowMapping());
+ }
+}
diff --git a/plugins/databases/redshift/pom.xml b/plugins/databases/redshift/pom.xml
index e1b782ae48..00aec22d1f 100644
--- a/plugins/databases/redshift/pom.xml
+++ b/plugins/databases/redshift/pom.xml
@@ -35,6 +35,13 @@
<dependencyManagement>
<dependencies>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-libs</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hop</groupId>
<artifactId>hop-libs-jdbc</artifactId>
@@ -56,6 +63,46 @@
<artifactId>redshift-jdbc42</artifactId>
<scope>provided</scope>
</dependency>
+ <!--
+ IAM authentication makes the driver ask AWS for temporary database credentials instead
+ of signing in with a Redshift password. It reaches for these to do so, but declares them
+ optional, so without them the connection fails with a NoClassDefFoundError on
+ software.amazon.awssdk.services.redshift.RedshiftClient. The rest of the AWS SDK they
+ need is already shared in lib/core.
+ -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>redshift</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>redshiftserverless</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sts</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
</project>
diff --git a/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftAuthenticationType.java b/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftAuthenticationType.java
new file mode 100644
index 0000000000..83fb91ba5e
--- /dev/null
+++ b/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftAuthenticationType.java
@@ -0,0 +1,58 @@
+/*
+ * 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.databases.redshift;
+
+/**
+ * How the Redshift JDBC driver is asked to authenticate. Everything other than {@link #DATABASE}
+ * makes the driver fetch temporary database credentials from AWS, which it only does when the URL
+ * carries the {@code iam} marker: {@code jdbc:redshift:iam://...}.
+ */
+public enum RedshiftAuthenticationType {
+ /**
+ * A Redshift database user and password, the way this plugin has always worked. Kept as the
+ * default so existing connections keep connecting exactly as before.
+ */
+ DATABASE,
+
+ /**
+ * An AWS access key and secret access key, optionally with a session token for temporary
+ * credentials. Passed to the driver as {@code AccessKeyID} / {@code SecretAccessKey} / {@code
+ * SessionToken}.
+ */
+ IAM_CREDENTIALS,
+
+ /**
+ * A named profile from the shared AWS credentials file, passed as {@code Profile}. Nothing secret
+ * ends up in the Hop metadata, which makes this the friendlier option for a shared project.
+ */
+ IAM_PROFILE,
+
+ /**
+ * Whatever the AWS default credentials chain finds: environment variables, the shared credentials
+ * file, an SSO session, an EC2 instance profile, an ECS task role, ... Nothing is passed to the
+ * driver, which is exactly what makes it fall back to that chain. This is the one to use when Hop
+ * runs inside AWS.
+ */
+ IAM_DEFAULT_CHAIN;
+
+ /**
+ * @return true if this type makes the driver ask AWS for temporary database credentials
+ */
+ public boolean isIam() {
+ return this != DATABASE;
+ }
+}
diff --git a/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDatabaseMeta.java b/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDatabaseMeta.java
index a1d9a2223d..02f091a152 100644
--- a/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDatabaseMeta.java
+++ b/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDatabaseMeta.java
@@ -16,15 +16,32 @@
*/
package org.apache.hop.databases.redshift;
+import java.util.HashSet;
import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.Const;
+import org.apache.hop.core.database.BaseDatabaseMeta;
import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.database.DatabaseMetaPlugin;
import org.apache.hop.core.database.DriverDownload;
import org.apache.hop.core.database.types.DatabaseTypes;
import org.apache.hop.core.database.types.IDatabaseTypeRule;
+import org.apache.hop.core.encryption.Encr;
+import org.apache.hop.core.gui.plugin.GuiElementType;
import org.apache.hop.core.gui.plugin.GuiPlugin;
+import org.apache.hop.core.gui.plugin.GuiWidgetElement;
import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.variables.IVariables;
import org.apache.hop.databases.postgresql.PostgreSqlDatabaseMeta;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.ui.core.gui.GuiCompositeWidgets;
+import org.apache.hop.ui.core.gui.IGuiPluginCompositeWidgetsListener;
+import org.eclipse.swt.widgets.Combo;
+import org.eclipse.swt.widgets.Control;
@DatabaseMetaPlugin(
type = "REDSHIFT",
@@ -33,7 +50,40 @@ import org.apache.hop.databases.postgresql.PostgreSqlDatabaseMeta;
documentationUrl = "/database/databases/redshift.html",
classLoaderGroup = "redshift-db")
@GuiPlugin(id = "GUI-RedshiftDatabaseMeta")
-public class RedshiftDatabaseMeta extends PostgreSqlDatabaseMeta {
+public class RedshiftDatabaseMeta extends PostgreSqlDatabaseMeta
+ implements IGuiPluginCompositeWidgetsListener {
+
+ /** The host name of every serverless workgroup ends in this. */
+ public static final String SERVERLESS_HOST_SUFFIX = ".redshift-serverless.amazonaws.com";
+
+ public static final String ID_DEPLOYMENT_TYPE = "deploymentType";
+ public static final String ID_WORKGROUP = "workgroup";
+ public static final String ID_ACCOUNT_ID = "accountId";
+ public static final String ID_CLUSTER_ID = "clusterId";
+ public static final String ID_AWS_REGION = "awsRegion";
+ public static final String ID_AUTHENTICATION_TYPE = "authenticationType";
+ public static final String ID_AWS_ACCESS_KEY_ID = "awsAccessKeyId";
+ public static final String ID_AWS_SECRET_ACCESS_KEY = "awsSecretAccessKey";
+ public static final String ID_AWS_SESSION_TOKEN = "awsSessionToken";
+ public static final String ID_AWS_PROFILE = "awsProfile";
+ public static final String ID_DB_USER = "dbUser";
+ public static final String ID_DB_GROUPS = "dbGroups";
+ public static final String ID_AUTO_CREATE = "autoCreate";
+
+ // Driver options, spelled the way https://docs.aws.amazon.com/redshift/latest/mgmt/
+ // jdbc20-configuration-options.html spells them. The driver itself is not case sensitive.
+ private static final String PROPERTY_ACCESS_KEY_ID = "AccessKeyID";
+ private static final String PROPERTY_SECRET_ACCESS_KEY = "SecretAccessKey";
+ private static final String PROPERTY_SESSION_TOKEN = "SessionToken";
+ private static final String PROPERTY_PROFILE = "Profile";
+ private static final String PROPERTY_DB_USER = "DbUser";
+ private static final String PROPERTY_DB_GROUPS = "DbGroups";
+ private static final String PROPERTY_AUTO_CREATE = "AutoCreate";
+ private static final String PROPERTY_REGION = "Region";
+ private static final String PROPERTY_CLUSTER_ID = "ClusterID";
+ private static final String PROPERTY_IS_SERVERLESS = "isServerless";
+ private static final String PROPERTY_SERVERLESS_WORK_GROUP = "serverlessWorkGroup";
+ private static final String PROPERTY_SERVERLESS_ACCT_ID = "serverlessAcctId";
private static final List<IDatabaseTypeRule> TYPE_RULES =
DatabaseTypes.rules()
@@ -43,6 +93,166 @@ public class RedshiftDatabaseMeta extends PostgreSqlDatabaseMeta {
.as("SUPER")
.build();
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_DEPLOYMENT_TYPE,
+ order = "04",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.COMBO,
+ variables = false,
+ label = "i18n::RedshiftDatabaseMeta.label.DeploymentType",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.DeploymentType")
+ @HopMetadataProperty(enumNameWhenNotFound = "PROVISIONED")
+ private RedshiftDeploymentType deploymentType = RedshiftDeploymentType.PROVISIONED;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_WORKGROUP,
+ order = "05",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.Workgroup",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.Workgroup")
+ @HopMetadataProperty
+ private String workgroup;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_ACCOUNT_ID,
+ order = "06",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.AccountId",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AccountId")
+ @HopMetadataProperty
+ private String accountId;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_CLUSTER_ID,
+ order = "07",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.ClusterId",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.ClusterId")
+ @HopMetadataProperty
+ private String clusterId;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AWS_REGION,
+ order = "08",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.AwsRegion",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AwsRegion")
+ @HopMetadataProperty
+ private String awsRegion;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AUTHENTICATION_TYPE,
+ order = "09",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.COMBO,
+ variables = false,
+ label = "i18n::RedshiftDatabaseMeta.label.AuthenticationType",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AuthenticationType")
+ @HopMetadataProperty(enumNameWhenNotFound = "DATABASE")
+ private RedshiftAuthenticationType authenticationType = RedshiftAuthenticationType.DATABASE;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AWS_ACCESS_KEY_ID,
+ order = "10",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.AwsAccessKeyId",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AwsAccessKeyId")
+ @HopMetadataProperty
+ private String awsAccessKeyId;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AWS_SECRET_ACCESS_KEY,
+ order = "11",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ password = true,
+ label = "i18n::RedshiftDatabaseMeta.label.AwsSecretAccessKey",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AwsSecretAccessKey")
+ @HopMetadataProperty(password = true)
+ private String awsSecretAccessKey;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AWS_SESSION_TOKEN,
+ order = "12",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ password = true,
+ label = "i18n::RedshiftDatabaseMeta.label.AwsSessionToken",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AwsSessionToken")
+ @HopMetadataProperty(password = true)
+ private String awsSessionToken;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AWS_PROFILE,
+ order = "13",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.AwsProfile",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AwsProfile")
+ @HopMetadataProperty
+ private String awsProfile;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_DB_USER,
+ order = "14",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.DbUser",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.DbUser")
+ @HopMetadataProperty
+ private String dbUser;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_DB_GROUPS,
+ order = "15",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::RedshiftDatabaseMeta.label.DbGroups",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.DbGroups")
+ @HopMetadataProperty
+ private String dbGroups;
+
+ @Getter
+ @Setter
+ @GuiWidgetElement(
+ id = ID_AUTO_CREATE,
+ order = "16",
+ parentId = DatabaseMeta.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.CHECKBOX,
+ label = "i18n::RedshiftDatabaseMeta.label.AutoCreate",
+ toolTip = "i18n::RedshiftDatabaseMeta.toolTip.AutoCreate")
+ @HopMetadataProperty
+ private boolean autoCreate;
+
@Override
public List<IDatabaseTypeRule> getTypeRules() {
return TYPE_RULES;
@@ -69,7 +279,7 @@ public class RedshiftDatabaseMeta extends PostgreSqlDatabaseMeta {
public DriverDownload getDriverDownload() {
return DriverDownload.builder()
.mavenCoordinate("com.amazon.redshift:redshift-jdbc42")
- .defaultVersion("2.2.7")
+ .defaultVersion("2.2.8")
.licenseCategory("A")
.licenseName("Apache-2.0")
.licenseUrl("https://github.com/aws/amazon-redshift-jdbc-driver/blob/master/LICENSE")
@@ -78,9 +288,140 @@ public class RedshiftDatabaseMeta extends PostgreSqlDatabaseMeta {
.build();
}
+ /**
+ * The driver only asks AWS for temporary credentials when the URL says {@code iam}, and it only
+ * finds a serverless workgroup when the host name is the one AWS publishes for it. Both of those
+ * are decisions the connection already carries, so neither has to be typed by hand.
+ *
+ * <p>Variables are deliberately left alone here. A {@link BaseDatabaseMeta} subclass has no
+ * variables to resolve with, but {@code DatabaseMeta.getURL(IVariables)} resolves whatever this
+ * returns, so a workgroup or a region can perfectly well be a variable.
+ */
@Override
public String getURL(String hostname, String port, String databaseName) {
- return "jdbc:redshift://" + hostname + ":" + port + "/" + databaseName;
+ String scheme = authenticationType.isIam() ? "jdbc:redshift:iam://" : "jdbc:redshift://";
+ return scheme + serverHostname(hostname) + ":" + port + "/" + databaseName;
+ }
+
+ /**
+ * The host name to connect to: the one that was entered, or the one a serverless workgroup is
+ * guaranteed to have.
+ *
+ * @param hostname the host name on the connection, used for a provisioned cluster
+ * @return the host name the URL should carry
+ */
+ private String serverHostname(String hostname) {
+ // A serverless endpoint is built only when nothing was entered. Anyone sitting behind a custom
+ // domain name or a load balancer enters that instead, and still names the workgroup below so
+ // the driver knows which one it is talking to -- it can no longer tell from the host name.
+ if (deploymentType == RedshiftDeploymentType.SERVERLESS && StringUtils.isEmpty(hostname)) {
+ return buildServerlessHostname(workgroup, accountId, awsRegion);
+ }
+ return hostname;
+ }
+
+ /**
+ * Build the endpoint AWS publishes for a serverless workgroup:
+ *
+ * <pre><workgroup>.<account-id>.<region>.redshift-serverless.amazonaws.com
+ * </pre>
+ *
+ * <p>Note this is the <em>workgroup</em> name, not the namespace name -- they are easy to mix up
+ * and a namespace name gives a host that does not resolve.
+ *
+ * @param workgroup the workgroup name
+ * @param accountId the 12 digit AWS account number
+ * @param region the AWS region the workgroup runs in, for example eu-west-1
+ * @return the endpoint host name
+ */
+ public static String buildServerlessHostname(String workgroup, String accountId, String region) {
+ return Const.NVL(workgroup, "")
+ + "."
+ + Const.NVL(accountId, "")
+ + "."
+ + Const.NVL(region, "")
+ + SERVERLESS_HOST_SUFFIX;
+ }
+
+ /**
+ * Everything the driver needs beyond the URL. These deliberately do not go in the URL: the
+ * secrets among them would end up in log lines and in the connection string shown on the dialog,
+ * and AWS asks for them to be URL encoded there, which is a trap nobody needs.
+ *
+ * <p>Passing them as properties also makes them work for a manually entered URL, which is how
+ * anyone with a custom domain name in front of their cluster has to connect.
+ */
+ @Override
+ public Properties getConnectionProperties(IVariables variables) {
+ Properties properties = new Properties();
+
+ if (deploymentType == RedshiftDeploymentType.SERVERLESS) {
+ // The driver works these out from the host name, but not when something like a load
+ // balancer or a custom domain name sits in front of it. Saying so costs nothing.
+ properties.put(PROPERTY_IS_SERVERLESS, "true");
+ putIfFilled(properties, PROPERTY_SERVERLESS_WORK_GROUP, variables.resolve(workgroup));
+ putIfFilled(properties, PROPERTY_SERVERLESS_ACCT_ID, variables.resolve(accountId));
+ } else if (authenticationType.isIam()) {
+ // The same problem on a cluster: the driver reads the cluster name out of the host name,
+ // and a load balancer or a custom domain name leaves it nothing to read.
+ putIfFilled(properties, PROPERTY_CLUSTER_ID, variables.resolve(clusterId));
+ }
+ // The region builds a serverless endpoint and tells IAM which AWS to ask. It says nothing to
+ // a plain user and password connection, so a value left behind by another choice stays here.
+ if (deploymentType == RedshiftDeploymentType.SERVERLESS || authenticationType.isIam()) {
+ putIfFilled(properties, PROPERTY_REGION, variables.resolve(awsRegion));
+ }
+
+ if (!authenticationType.isIam()) {
+ return properties;
+ }
+
+ switch (authenticationType) {
+ case IAM_CREDENTIALS -> {
+ putIfFilled(properties, PROPERTY_ACCESS_KEY_ID, variables.resolve(awsAccessKeyId));
+ putIfFilled(properties, PROPERTY_SECRET_ACCESS_KEY, decrypt(variables, awsSecretAccessKey));
+ putIfFilled(properties, PROPERTY_SESSION_TOKEN, decrypt(variables, awsSessionToken));
+ }
+ case IAM_PROFILE -> putIfFilled(properties, PROPERTY_PROFILE, variables.resolve(awsProfile));
+ default -> {
+ // IAM_DEFAULT_CHAIN: saying nothing is what makes the driver go and look for itself.
+ }
+ }
+
+ // A serverless workgroup issues credentials for the IAM identity itself: its GetCredentials
+ // call takes a database and a workgroup and nothing else. Naming a database user, its groups
+ // or asking for it to be created are all GetClusterCredentials ideas, and only a provisioned
+ // cluster has any use for them.
+ if (deploymentType == RedshiftDeploymentType.SERVERLESS) {
+ return properties;
+ }
+
+ // Which database user the temporary credentials are for. Falling back to the user name on the
+ // connection means the obvious place to have typed it is also the right one.
+ String user = StringUtils.isNotEmpty(dbUser) ? dbUser : getUsername();
+ putIfFilled(properties, PROPERTY_DB_USER, variables.resolve(user));
+ putIfFilled(properties, PROPERTY_DB_GROUPS, variables.resolve(dbGroups));
+ if (autoCreate) {
+ properties.put(PROPERTY_AUTO_CREATE, "true");
+ }
+
+ return properties;
+ }
+
+ /**
+ * Add a driver option, trimmed. None of what goes through here -- a key, a region, a workgroup
+ * name -- can meaningfully start or end in whitespace, and a stray space picked up while copying
+ * an access key out of the AWS console otherwise comes back as an authentication failure with
+ * nothing on screen to explain it.
+ */
+ private void putIfFilled(Properties properties, String name, String value) {
+ if (StringUtils.isNotBlank(value)) {
+ properties.put(name, value.trim());
+ }
+ }
+
+ private String decrypt(IVariables variables, String password) {
+ return Encr.decryptPasswordOptionallyEncrypted(variables.resolve(password));
}
@Override
@@ -112,4 +453,88 @@ public class RedshiftDatabaseMeta extends PostgreSqlDatabaseMeta {
public boolean isSupportsSetMaxRows() {
return false;
}
+
+ @Override
+ public void widgetsCreated(GuiCompositeWidgets compositeWidgets) {
+ // Nothing to do, the values aren't set yet at this point.
+ }
+
+ @Override
+ public void widgetsPopulated(GuiCompositeWidgets compositeWidgets) {
+ hideFieldsThatDoNotApply(compositeWidgets);
+ }
+
+ @Override
+ public void widgetModified(
+ GuiCompositeWidgets compositeWidgets, Control changedWidget, String widgetId) {
+ hideFieldsThatDoNotApply(compositeWidgets);
+ }
+
+ @Override
+ public void persistContents(GuiCompositeWidgets compositeWidgets) {
+ // Not needed, the dialog reads the widgets back itself.
+ }
+
+ /**
+ * Keeps the dialog down to the fields that can actually do something for the connection being
+ * edited. Between two deployments and four ways to authenticate, showing everything at once would
+ * leave most of the tab permanently empty and the fields that matter hard to pick out.
+ */
+ private void hideFieldsThatDoNotApply(GuiCompositeWidgets compositeWidgets) {
+ RedshiftDeploymentType deployment = readDeploymentType(compositeWidgets);
+ RedshiftAuthenticationType authentication = readAuthenticationType(compositeWidgets);
+ boolean serverless = deployment == RedshiftDeploymentType.SERVERLESS;
+
+ Set<String> hidden = new HashSet<>();
+ // The host name stays on offer for a serverless workgroup: left empty Hop builds the endpoint,
+ // filled in it wins, which is how a custom domain name or a load balancer is reached.
+ hideUnless(hidden, serverless, ID_WORKGROUP, ID_ACCOUNT_ID);
+ // A cluster only has to name itself when IAM has to find it.
+ hideUnless(hidden, !serverless && authentication.isIam(), ID_CLUSTER_ID);
+ // The region builds the host name, and IAM needs it to know which AWS to ask.
+ hideUnless(hidden, serverless || authentication.isIam(), ID_AWS_REGION);
+
+ hideUnless(
+ hidden,
+ authentication == RedshiftAuthenticationType.IAM_CREDENTIALS,
+ ID_AWS_ACCESS_KEY_ID,
+ ID_AWS_SECRET_ACCESS_KEY,
+ ID_AWS_SESSION_TOKEN);
+ hideUnless(hidden, authentication == RedshiftAuthenticationType.IAM_PROFILE, ID_AWS_PROFILE);
+ // Only a provisioned cluster maps IAM onto a named database user.
+ hideUnless(
+ hidden, authentication.isIam() && !serverless, ID_DB_USER, ID_DB_GROUPS, ID_AUTO_CREATE);
+
+ compositeWidgets.setWidgetsHidden(this, hidden);
+ }
+
+ private void hideUnless(Set<String> hidden, boolean applies, String... widgetIds) {
+ if (!applies) {
+ hidden.addAll(List.of(widgetIds));
+ }
+ }
+
+ private RedshiftDeploymentType readDeploymentType(GuiCompositeWidgets compositeWidgets) {
+ Control control = compositeWidgets.getWidgetsMap().get(ID_DEPLOYMENT_TYPE);
+ if (control instanceof Combo combo) {
+ try {
+ return RedshiftDeploymentType.valueOf(combo.getText());
+ } catch (IllegalArgumentException e) {
+ // Nothing selected yet, the metadata is the better answer.
+ }
+ }
+ return getDeploymentType();
+ }
+
+ private RedshiftAuthenticationType readAuthenticationType(GuiCompositeWidgets compositeWidgets) {
+ Control control = compositeWidgets.getWidgetsMap().get(ID_AUTHENTICATION_TYPE);
+ if (control instanceof Combo combo) {
+ try {
+ return RedshiftAuthenticationType.valueOf(combo.getText());
+ } catch (IllegalArgumentException e) {
+ // Nothing selected yet, the metadata is the better answer.
+ }
+ }
+ return getAuthenticationType();
+ }
}
diff --git a/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDeploymentType.java b/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDeploymentType.java
new file mode 100644
index 0000000000..ee1c25b106
--- /dev/null
+++ b/plugins/databases/redshift/src/main/java/org/apache/hop/databases/redshift/RedshiftDeploymentType.java
@@ -0,0 +1,33 @@
+/*
+ * 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.databases.redshift;
+
+/** Which flavour of Redshift is on the other end, which decides how the host name is arrived at. */
+public enum RedshiftDeploymentType {
+ /**
+ * A provisioned cluster. The server host name is entered as is, the way this plugin has always
+ * worked. Kept as the default so existing connections keep connecting exactly as before.
+ */
+ PROVISIONED,
+
+ /**
+ * A serverless workgroup. Its endpoint is entirely predictable from the workgroup name, the AWS
+ * account number and the region, so Hop builds it rather than asking for a host name that is easy
+ * to get subtly wrong.
+ */
+ SERVERLESS
+}
diff --git a/plugins/databases/redshift/src/main/resources/org/apache/hop/databases/redshift/messages/messages_en_US.properties b/plugins/databases/redshift/src/main/resources/org/apache/hop/databases/redshift/messages/messages_en_US.properties
new file mode 100644
index 0000000000..76f29c0f29
--- /dev/null
+++ b/plugins/databases/redshift/src/main/resources/org/apache/hop/databases/redshift/messages/messages_en_US.properties
@@ -0,0 +1,42 @@
+#
+# 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.
+#
+RedshiftDatabaseMeta.label.DeploymentType = Deployment
+RedshiftDatabaseMeta.toolTip.DeploymentType = PROVISIONED is a Redshift cluster, whose endpoint you enter as the server hostname. SERVERLESS is a workgroup, whose endpoint Hop builds from the workgroup name, the account number and the region.
+RedshiftDatabaseMeta.label.Workgroup = Workgroup name
+RedshiftDatabaseMeta.toolTip.Workgroup = The name of the serverless workgroup. This is the workgroup, not the namespace — they are easy to mix up, and a namespace name gives a hostname that does not resolve. Leave the server hostname empty to have Hop build the endpoint from this; fill the hostname in to reach the workgroup through a custom domain name or a load balancer, in which case this still tells the driver which workgroup it is.
+RedshiftDatabaseMeta.label.AccountId = AWS account number
+RedshiftDatabaseMeta.toolTip.AccountId = The 12 digit AWS account number the workgroup belongs to. It is the second label of the endpoint the Redshift console shows for the workgroup.
+RedshiftDatabaseMeta.label.ClusterId = Cluster identifier
+RedshiftDatabaseMeta.toolTip.ClusterId = The name of the cluster, needed only when the driver cannot read it from the server hostname — a custom domain name or a load balancer in front of the cluster. Leave empty for a normal cluster endpoint.
+RedshiftDatabaseMeta.label.AwsRegion = AWS region
+RedshiftDatabaseMeta.toolTip.AwsRegion = The region the workgroup or cluster runs in, for example eu-west-1. Used to build a serverless endpoint, and to tell the driver which AWS to ask for temporary credentials.
+RedshiftDatabaseMeta.label.AuthenticationType = Authentication
+RedshiftDatabaseMeta.toolTip.AuthenticationType = DATABASE signs in with a Redshift user and password. The IAM options make the driver ask AWS for temporary database credentials instead: IAM_CREDENTIALS uses an access key you enter here, IAM_PROFILE a named profile from your AWS credentials file, and IAM_DEFAULT_CHAIN whatever the AWS default credentials chain finds — environment variables, an SSO session, an EC2 instance profile or an ECS task role.
+RedshiftDatabaseMeta.label.AwsAccessKeyId = AWS access key ID
+RedshiftDatabaseMeta.toolTip.AwsAccessKeyId = The AWS access key ID to authenticate with. This is not a Redshift user: it identifies an IAM user or role that is allowed to fetch database credentials.
+RedshiftDatabaseMeta.label.AwsSecretAccessKey = AWS secret access key
+RedshiftDatabaseMeta.toolTip.AwsSecretAccessKey = The secret access key belonging to the access key ID above.
+RedshiftDatabaseMeta.label.AwsSessionToken = AWS session token
+RedshiftDatabaseMeta.toolTip.AwsSessionToken = Only needed for temporary credentials, such as those from an assumed role. Leave empty for a long lived access key.
+RedshiftDatabaseMeta.label.AwsProfile = AWS profile
+RedshiftDatabaseMeta.toolTip.AwsProfile = The name of a profile in your shared AWS credentials file, usually ~/.aws/credentials. Nothing secret is stored in the Hop connection this way.
+RedshiftDatabaseMeta.label.DbUser = Database user
+RedshiftDatabaseMeta.toolTip.DbUser = The Redshift user the temporary credentials are issued for. Leave empty to use the username entered above.
+RedshiftDatabaseMeta.label.DbGroups = Database groups
+RedshiftDatabaseMeta.toolTip.DbGroups = A comma separated list of existing database groups the user joins for the session. Optional.
+RedshiftDatabaseMeta.label.AutoCreate = Create the user if missing
+RedshiftDatabaseMeta.toolTip.AutoCreate = Create the database user named above when it does not exist yet, rather than failing to authenticate.
diff --git a/plugins/databases/redshift/src/test/java/org/apache/hop/databases/redshift/RedshiftAuthenticationTest.java b/plugins/databases/redshift/src/test/java/org/apache/hop/databases/redshift/RedshiftAuthenticationTest.java
new file mode 100644
index 0000000000..6f6b16831f
--- /dev/null
+++ b/plugins/databases/redshift/src/test/java/org/apache/hop/databases/redshift/RedshiftAuthenticationTest.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.databases.redshift;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Properties;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.database.DatabaseMeta;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.core.variables.Variables;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** The authentication modes and the serverless endpoint helper of the Redshift connection. */
+class RedshiftAuthenticationTest {
+
+ private static final String SERVERLESS_HOST =
+ "workgroup.123456789012.eu-west-1.redshift-serverless.amazonaws.com";
+
+ private RedshiftDatabaseMeta dbMeta;
+ private IVariables variables;
+
+ @BeforeAll
+ static void initHop() throws Exception {
+ // The secret access key goes through Encr, which needs the password encoder plugin.
+ HopClientEnvironment.init();
+ }
+
+ @BeforeEach
+ void setUp() {
+ dbMeta = new RedshiftDatabaseMeta();
+ dbMeta.setAccessType(DatabaseMeta.TYPE_ACCESS_NATIVE);
+ variables = new Variables();
+ }
+
+ // ------------------------------------------------------------------ defaults
+
+ /**
+ * The whole point of the defaults: a connection saved before any of this existed has to build the
+ * exact same URL and add no properties of its own.
+ */
+ @Test
+ void aConnectionThatSaysNothingBehavesExactlyAsBefore() {
+ assertEquals(RedshiftDeploymentType.PROVISIONED, dbMeta.getDeploymentType());
+ assertEquals(RedshiftAuthenticationType.DATABASE, dbMeta.getAuthenticationType());
+
+ assertEquals(
+ "jdbc:redshift://my-cluster.abc123.eu-west-1.redshift.amazonaws.com:5439/dev",
+ dbMeta.getURL("my-cluster.abc123.eu-west-1.redshift.amazonaws.com", "5439", "dev"));
+ assertTrue(dbMeta.getConnectionProperties(variables).isEmpty());
+ }
+
+ // ------------------------------------------------------------------ the URL
+
+ @Test
+ void databaseAuthenticationKeepsThePlainScheme() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.DATABASE);
+
+ assertTrue(dbMeta.getURL("host", "5439", "dev").startsWith("jdbc:redshift://"));
+ }
+
+ /** The driver only goes and fetches credentials when the URL says so. */
+ @Test
+ void everyIamModeSwitchesTheUrlToTheIamScheme() {
+ for (RedshiftAuthenticationType type : RedshiftAuthenticationType.values()) {
+ dbMeta.setAuthenticationType(type);
+ String url = dbMeta.getURL("host", "5439", "dev");
+ if (type == RedshiftAuthenticationType.DATABASE) {
+ assertFalse(url.startsWith("jdbc:redshift:iam://"), type.name());
+ } else {
+ assertTrue(url.startsWith("jdbc:redshift:iam://"), type.name());
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------ serverless endpoint
+
+ @Test
+ void buildsTheServerlessEndpointFromWorkgroupAccountAndRegion() {
+ assertEquals(
+ SERVERLESS_HOST,
+ RedshiftDatabaseMeta.buildServerlessHostname("workgroup", "123456789012", "eu-west-1"));
+ }
+
+ /** With no host name entered, the endpoint is built from the workgroup. */
+ @Test
+ void aServerlessConnectionBuildsItsOwnHostnameWhenNoneWasEntered() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setWorkgroup("workgroup");
+ dbMeta.setAccountId("123456789012");
+ dbMeta.setAwsRegion("eu-west-1");
+
+ assertEquals(
+ "jdbc:redshift://" + SERVERLESS_HOST + ":5439/dev", dbMeta.getURL("", "5439", "dev"));
+ }
+
+ /**
+ * getURL() has no variables to resolve with, but DatabaseMeta resolves what it returns, so the
+ * pieces have to survive into the URL untouched.
+ */
+ @Test
+ void leavesVariablesInTheEndpointForTheCallerToResolve() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setWorkgroup("${WORKGROUP}");
+ dbMeta.setAccountId("${ACCOUNT}");
+ dbMeta.setAwsRegion("${REGION}");
+
+ assertEquals(
+ "jdbc:redshift://${WORKGROUP}.${ACCOUNT}.${REGION}.redshift-serverless.amazonaws.com:5439/dev",
+ dbMeta.getURL("", "5439", "dev"));
+ }
+
+ /**
+ * A custom domain name or a load balancer in front of the workgroup: the entered host wins over
+ * the built one, which is the whole point of still offering the field.
+ */
+ @Test
+ void aServerlessHostnameThatWasEnteredWinsOverTheBuiltOne() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setWorkgroup("workgroup");
+ dbMeta.setAccountId("123456789012");
+ dbMeta.setAwsRegion("eu-west-1");
+
+ assertEquals(
+ "jdbc:redshift://redshift.example.com:5439/dev",
+ dbMeta.getURL("redshift.example.com", "5439", "dev"));
+ }
+
+ /**
+ * And the workgroup still has to reach the driver, because behind a custom name it can no longer
+ * work out which workgroup it is talking to.
+ */
+ @Test
+ void stillNamesTheWorkgroupBehindACustomHostname() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_DEFAULT_CHAIN);
+ dbMeta.setWorkgroup("workgroup");
+ dbMeta.setAccountId("123456789012");
+ dbMeta.setAwsRegion("eu-west-1");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("true", properties.get("isServerless"));
+ assertEquals("workgroup", properties.get("serverlessWorkGroup"));
+ assertEquals("eu-west-1", properties.get("Region"));
+ }
+
+ /**
+ * The cluster's counterpart of serverlessWorkGroup: behind a custom domain name or a load
+ * balancer the driver cannot read the cluster name out of the host name either.
+ */
+ @Test
+ void namesTheClusterForProvisionedIam() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.PROVISIONED);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_DEFAULT_CHAIN);
+ dbMeta.setClusterId("my-cluster");
+
+ assertEquals("my-cluster", dbMeta.getConnectionProperties(variables).get("ClusterID"));
+ }
+
+ /** A cluster on user and password never asks AWS anything, so it has no cluster to name. */
+ @Test
+ void namesNoClusterForAProvisionedPasswordConnection() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.PROVISIONED);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.DATABASE);
+ dbMeta.setClusterId("my-cluster");
+
+ assertNull(dbMeta.getConnectionProperties(variables).get("ClusterID"));
+ }
+
+ /** A workgroup is not a cluster: the two identities must never both go out. */
+ @Test
+ void namesNoClusterForAServerlessConnection() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_DEFAULT_CHAIN);
+ dbMeta.setWorkgroup("workgroup");
+ // Left behind by an earlier choice.
+ dbMeta.setClusterId("my-cluster");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertNull(properties.get("ClusterID"));
+ assertEquals("workgroup", properties.get("serverlessWorkGroup"));
+ }
+
+ /** The region says nothing to a plain user and password connection on a cluster. */
+ @Test
+ void sendsNoRegionForAProvisionedPasswordConnection() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.PROVISIONED);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.DATABASE);
+ // Left behind by an earlier choice: it must not travel with the connection.
+ dbMeta.setAwsRegion("eu-west-1");
+
+ assertNull(dbMeta.getConnectionProperties(variables).get("Region"));
+ }
+
+ /** It does matter for IAM on a cluster: it says which AWS to ask for credentials. */
+ @Test
+ void sendsTheRegionForProvisionedIamSoCredentialsAreFetchedFromTheRightPlace() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.PROVISIONED);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+ dbMeta.setAwsRegion("eu-west-1");
+
+ assertEquals("eu-west-1", dbMeta.getConnectionProperties(variables).get("Region"));
+ }
+
+ @Test
+ void tellsTheDriverItIsServerlessSoItWorksBehindALoadBalancer() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setWorkgroup("workgroup");
+ dbMeta.setAccountId("123456789012");
+ dbMeta.setAwsRegion("eu-west-1");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("true", properties.get("isServerless"));
+ assertEquals("workgroup", properties.get("serverlessWorkGroup"));
+ assertEquals("123456789012", properties.get("serverlessAcctId"));
+ assertEquals("eu-west-1", properties.get("Region"));
+ }
+
+ @Test
+ void saysNothingAboutServerlessForAProvisionedCluster() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.PROVISIONED);
+
+ assertNull(dbMeta.getConnectionProperties(variables).get("isServerless"));
+ }
+
+ // ------------------------------------------------------------------ credentials
+
+ @Test
+ void passesAnAccessKeyAsTheDriverPropertiesTheDriverExpects() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ dbMeta.setAwsAccessKeyId("AKIAEXAMPLE");
+ dbMeta.setAwsSecretAccessKey("s3cr3t");
+ dbMeta.setAwsSessionToken("token");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("AKIAEXAMPLE", properties.get("AccessKeyID"));
+ assertEquals("s3cr3t", properties.get("SecretAccessKey"));
+ assertEquals("token", properties.get("SessionToken"));
+ assertNull(properties.get("Profile"));
+ }
+
+ /** Copying an access key out of the console very easily brings whitespace along with it. */
+ @Test
+ void trimsWhitespaceAroundTheCredentials() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ dbMeta.setAwsAccessKeyId(" AKIAEXAMPLE ");
+ dbMeta.setAwsSecretAccessKey("s3cr3t\n");
+ dbMeta.setAwsRegion(" eu-west-1 ");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("AKIAEXAMPLE", properties.get("AccessKeyID"));
+ assertEquals("s3cr3t", properties.get("SecretAccessKey"));
+ assertEquals("eu-west-1", properties.get("Region"));
+ }
+
+ /** A field holding nothing but spaces is not a credential. */
+ @Test
+ void treatsAWhitespaceOnlyValueAsUnset() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ dbMeta.setAwsAccessKeyId("AKIAEXAMPLE");
+ dbMeta.setAwsSecretAccessKey("s3cr3t");
+ dbMeta.setAwsSessionToken(" ");
+
+ assertNull(dbMeta.getConnectionProperties(variables).get("SessionToken"));
+ }
+
+ @Test
+ void resolvesVariablesInTheCredentials() {
+ variables.setVariable("KEY", "AKIAEXAMPLE");
+ variables.setVariable("SECRET", "s3cr3t");
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ dbMeta.setAwsAccessKeyId("${KEY}");
+ dbMeta.setAwsSecretAccessKey("${SECRET}");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("AKIAEXAMPLE", properties.get("AccessKeyID"));
+ assertEquals("s3cr3t", properties.get("SecretAccessKey"));
+ }
+
+ @Test
+ void passesAProfileByName() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+ dbMeta.setAwsProfile("integration-test");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("integration-test", properties.get("Profile"));
+ assertNull(properties.get("AccessKeyID"));
+ }
+
+ /** Saying nothing is precisely what makes the driver fall back to the AWS default chain. */
+ @Test
+ void passesNoCredentialsAtAllForTheDefaultChain() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_DEFAULT_CHAIN);
+ // Left over from a previous choice: it must not leak into the connection.
+ dbMeta.setAwsAccessKeyId("AKIAEXAMPLE");
+ dbMeta.setAwsProfile("integration-test");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertNull(properties.get("AccessKeyID"));
+ assertNull(properties.get("Profile"));
+ }
+
+ /** A stale access key must not travel with a plain database connection either. */
+ @Test
+ void passesNoCredentialsAtAllForDatabaseAuthentication() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.DATABASE);
+ dbMeta.setAwsAccessKeyId("AKIAEXAMPLE");
+ dbMeta.setAwsProfile("integration-test");
+ dbMeta.setDbUser("analyst");
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertNull(properties.get("AccessKeyID"));
+ assertNull(properties.get("Profile"));
+ assertNull(properties.get("DbUser"));
+ }
+
+ // ------------------------------------------------------------------ database user
+
+ @Test
+ void fallsBackToTheConnectionUsernameForTheDatabaseUser() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+ dbMeta.setUsername("analyst");
+
+ assertEquals("analyst", dbMeta.getConnectionProperties(variables).get("DbUser"));
+ }
+
+ @Test
+ void anExplicitDatabaseUserWinsOverTheConnectionUsername() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+ dbMeta.setUsername("analyst");
+ dbMeta.setDbUser("loader");
+
+ assertEquals("loader", dbMeta.getConnectionProperties(variables).get("DbUser"));
+ }
+
+ /**
+ * The serverless GetCredentials call takes a database and a workgroup, nothing else. Sending it a
+ * database user only invites confusion about which identity the session actually runs as.
+ */
+ @Test
+ void sendsNoDatabaseUserForAServerlessWorkgroup() {
+ dbMeta.setDeploymentType(RedshiftDeploymentType.SERVERLESS);
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+ dbMeta.setAwsProfile("integration-test");
+ dbMeta.setUsername("admin");
+ dbMeta.setDbUser("loader");
+ dbMeta.setDbGroups("analysts");
+ dbMeta.setAutoCreate(true);
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertNull(properties.get("DbUser"));
+ assertNull(properties.get("DbGroups"));
+ assertNull(properties.get("AutoCreate"));
+ // The credentials themselves still have to get through.
+ assertEquals("integration-test", properties.get("Profile"));
+ }
+
+ @Test
+ void passesTheGroupsAndTheAutoCreateFlag() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+ dbMeta.setDbGroups("analysts,loaders");
+ dbMeta.setAutoCreate(true);
+
+ Properties properties = dbMeta.getConnectionProperties(variables);
+
+ assertEquals("analysts,loaders", properties.get("DbGroups"));
+ assertEquals("true", properties.get("AutoCreate"));
+ }
+
+ @Test
+ void leavesTheAutoCreateFlagOutWhenItIsOff() {
+ dbMeta.setAuthenticationType(RedshiftAuthenticationType.IAM_PROFILE);
+
+ assertNull(dbMeta.getConnectionProperties(variables).get("AutoCreate"));
+ }
+}
diff --git a/plugins/tech/aws/pom.xml b/plugins/tech/aws/pom.xml
index b5465a68d5..e3ff15c3d4 100644
--- a/plugins/tech/aws/pom.xml
+++ b/plugins/tech/aws/pom.xml
@@ -64,6 +64,28 @@
</exclusion>
</exclusions>
</dependency>
+ <!--
+ The bulk loader can inherit the AWS credentials configured on the Redshift connection.
+ Reading them needs the dialect on the classpath, and the transform shares its
+ classLoaderGroup so the two see the same classes at runtime.
+ -->
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-databases-redshift</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>auth</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3</artifactId>
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoader.java b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoader.java
index 84dcd6ff84..49613ebc59 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoader.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoader.java
@@ -17,18 +17,19 @@
package org.apache.hop.pipeline.transforms.redshift.bulkloader;
+import java.io.BufferedOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
-import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.vfs2.FileObject;
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.database.IDatabase;
+import org.apache.hop.core.encryption.Encr;
import org.apache.hop.core.exception.HopDatabaseException;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.exception.HopTransformException;
@@ -37,19 +38,39 @@ 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.ValueMetaDate;
+import org.apache.hop.core.row.value.ValueMetaString;
import org.apache.hop.core.util.Utils;
import org.apache.hop.core.vfs.HopVfs;
+import org.apache.hop.databases.redshift.RedshiftDatabaseMeta;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMeta;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
public class RedshiftBulkLoader
extends BaseTransform<RedshiftBulkLoaderMeta, RedshiftBulkLoaderData> {
private static final Class<?> PKG =
RedshiftBulkLoader.class; // for i18n purposes, needed by Translator2!!
+ /** Size of the buffer we put in front of the (remote) output stream. */
+ private static final int OUTPUT_BUFFER_SIZE = 128 * 1024;
+
+ /** Placeholder written instead of the AWS credentials when the COPY statement is logged. */
+ private static final String CREDENTIALS_MASK = "<credentials hidden>";
+
+ /**
+ * ISO 8601, with milliseconds. Redshift coerces this into either a DATE or a TIMESTAMP column --
+ * a DATE truncates the time, a TIMESTAMP keeps it -- so both Hop types can be written with full
+ * precision and let the target column decide what to keep.
+ */
+ private static final String TIMESTAMP_CONVERSION_MASK = "yyyy-MM-dd HH:mm:ss.SSS";
+
public RedshiftBulkLoader(
TransformMeta transformMeta,
RedshiftBulkLoaderMeta meta,
@@ -65,19 +86,25 @@ public class RedshiftBulkLoader
if (super.init()) {
try {
- // Validating that the connection has been defined.
+ // Validating that the connection and the S3 file have been defined.
verifyDatabaseConnection();
+ verifyFileSettings();
data.databaseMeta = this.getPipelineMeta().findDatabase(meta.getConnection(), variables);
if (meta.isStreamToS3Csv()) {
- // get the file output stream to write to S3
+ String target = resolve(meta.getCopyFromFilename());
+ ensureParentFolderExists(target);
+ // Get the file output stream to write to S3. Every field is written as a separate
+ // chunk of bytes, so a buffer in front of the (remote) stream matters a lot here.
data.writer =
- HopVfs.getOutputStream(resolve(meta.getCopyFromFilename()), false, variables);
+ new BufferedOutputStream(
+ HopVfs.getOutputStream(target, false, variables), OUTPUT_BUFFER_SIZE);
}
data.db = new Database(this, this, data.databaseMeta);
data.db.connect();
getDbFields();
+ verifyTableFields();
if (isBasic()) {
logBasic(
@@ -104,144 +131,148 @@ public class RedshiftBulkLoader
Object[] r = getRow(); // this also waits for a previous transform to be finished.
if (r == null) { // no more input to be expected...
- if (first && meta.isTruncateTable() && !meta.isOnlyWhenHaveRows()) {
- truncateTable();
- }
-
- if (!first) {
- try {
- data.close();
- closeFile();
- String copyStmt = buildCopyStatementSqlString();
- Connection conn = data.db.getConnection();
- try (Statement stmt = conn.createStatement()) {
- stmt.executeUpdate(copyStmt);
- conn.commit();
- }
- conn.close();
- } catch (SQLException sqle) {
- setErrors(1);
- stopAll();
- setOutputDone(); // signal end to receiver(s)
- throw new HopDatabaseException("Error executing COPY statements", sqle);
- } catch (IOException ioe) {
- setErrors(1);
- stopAll();
- setOutputDone(); // signal end to receiver(s)
- throw new HopTransformException("Error releasing resources", ioe);
- }
- }
-
+ endOfStream();
return false;
}
- if (first && meta.isStreamToS3Csv()) {
-
+ if (first) {
first = false;
- data.fieldnrs = new HashMap<>();
+ data.rowsReceived = true;
if (meta.isTruncateTable()) {
truncateTable();
}
- data.outputRowMeta = getInputRowMeta().clone();
- meta.getFields(data.insertRowMeta, getTransformName(), null, null, this, metadataProvider);
-
if (meta.isStreamToS3Csv()) {
- // Do noting
+ prepareRowMapping();
}
+ }
- // write all fields in the stream to Redshift
- if (!meta.specifyFields()) {
-
- // Just take the whole input row
- data.insertRowMeta = getInputRowMeta().clone();
- data.selectedRowFieldIndices = new int[data.insertRowMeta.size()];
-
- try {
- getDbFields();
- } catch (HopException e) {
- logError("Error getting database fields", e);
- setErrors(1);
- stopAll();
- setOutputDone(); // signal end to receiver(s)
- return false;
- }
+ if (meta.isStreamToS3Csv()) {
+ writeRowToFile(data.outputRowMeta, r);
+ putRow(data.outputRowMeta, r);
+ } else {
+ // We are loading a file that already exists on S3, the stream itself is only a trigger.
+ putRow(getInputRowMeta(), r);
+ }
- for (int i = 0; i < meta.getFields().size(); i++) {
- int streamFieldLocation =
- data.insertRowMeta.indexOfValue(meta.getFields().get(i).getStreamField());
- if (streamFieldLocation < 0) {
- throw new HopTransformException(
- "Field ["
- + meta.getFields().get(i).getStreamField()
- + "] couldn't be found in the input stream!");
- }
-
- int dbFieldLocation = -1;
- for (int e = 0; e < data.dbFields.size(); e++) {
- String[] field = data.dbFields.get(e);
- if (field[0].equalsIgnoreCase(meta.getFields().get(i).getDatabaseField())) {
- dbFieldLocation = e;
- break;
- }
- }
- if (dbFieldLocation < 0) {
- throw new HopException(
- "Field ["
- + meta.getFields().get(i).getDatabaseField()
- + "] couldn't be found in the table!");
- }
-
- data.fieldnrs.put(
- meta.getFields().get(i).getDatabaseField().toUpperCase(), streamFieldLocation);
- }
+ return true;
+ }
- } else {
+ /**
+ * Close the file we streamed to S3 (if any) and fire the COPY statement.
+ *
+ * @throws HopException in case the statement failed or resources could not be released
+ */
+ private void endOfStream() throws HopException {
+ if (!data.rowsReceived && meta.isTruncateTable() && !meta.isOnlyWhenHaveRows()) {
+ truncateTable();
+ }
- // use the columns/fields mapping.
- int numberOfInsertFields = meta.getFields().size();
- data.insertRowMeta = new RowMeta();
-
- // Cache the position of the selected fields in the row array
- data.selectedRowFieldIndices = new int[numberOfInsertFields];
- for (int i = 0; i < meta.getFields().size(); i++) {
- RedshiftBulkLoaderField vbf = meta.getFields().get(i);
- String inputFieldName = vbf.getStreamField();
- int inputFieldIdx = i;
- if (inputFieldIdx < 0) {
- throw new HopTransformException(
- BaseMessages.getString(
- PKG,
- "RedshiftBulkLoader.Exception.FieldRequired",
- inputFieldName)); //$NON-NLS-1$
- }
- data.selectedRowFieldIndices[i] = inputFieldIdx;
-
- String insertFieldName = vbf.getDatabaseField();
- IValueMeta inputValueMeta = getInputRowMeta().getValueMeta(inputFieldIdx);
- if (inputValueMeta == null) {
- throw new HopTransformException(
- BaseMessages.getString(
- PKG,
- "RedshiftBulkLoader.Exception.FailedToFindField",
- vbf.getStreamField())); // $NON-NLS-1$
- }
- IValueMeta insertValueMeta = inputValueMeta.clone();
- insertValueMeta.setName(insertFieldName);
- data.insertRowMeta.addValueMeta(insertValueMeta);
- data.fieldnrs.put(
- meta.getFields().get(i).getDatabaseField().toUpperCase(), inputFieldIdx);
- }
+ if (!shouldExecuteCopy()) {
+ return;
+ }
+
+ // The file has to be complete on S3 before Redshift reads it.
+ if (!closeFile()) {
+ setErrors(1);
+ stopAll();
+ setOutputDone(); // signal end to receiver(s)
+ throw new HopTransformException("Error releasing resources");
+ }
+
+ try {
+ String copyStmt = buildCopyStatementSqlString(false);
+ if (isDebug()) {
+ logDebug("copy stmt: " + buildCopyStatementSqlString(true));
+ }
+ try (Statement stmt = data.db.getConnection().createStatement()) {
+ stmt.executeUpdate(copyStmt);
}
+ data.db.commit();
+ } catch (SQLException sqle) {
+ setErrors(1);
+ stopAll();
+ setOutputDone(); // signal end to receiver(s)
+ throw new HopDatabaseException("Error executing COPY statements", sqle);
}
+ }
+ /**
+ * When we stream the rows to S3 ourselves there is nothing to load if the stream was empty. When
+ * we load a file that is already on S3 the load only depends on the "only when we have rows"
+ * option.
+ *
+ * @return true if the COPY statement has to be executed
+ */
+ boolean shouldExecuteCopy() {
if (meta.isStreamToS3Csv()) {
- writeRowToFile(data.outputRowMeta, r);
- putRow(data.outputRowMeta, r);
+ return data.rowsReceived;
}
+ return data.rowsReceived || !meta.isOnlyWhenHaveRows();
+ }
- return true;
+ /**
+ * Resolve, once, which field of the input row ends up in which column of the CSV file, together
+ * with the value meta used to render it. Doing this per row is what used to make this transform
+ * slow: it allocated and scanned the list of field names for every single field of every row.
+ *
+ * @throws HopException in case a configured field is not present on the input stream
+ */
+ void prepareRowMapping() throws HopException {
+ IRowMeta inputRowMeta = getInputRowMeta();
+ data.outputRowMeta = inputRowMeta.clone();
+
+ // Both modes build the same mapping: without an explicit field list every field of the row is
+ // written, in the order it arrives. Keeping one shape means date handling and everything else
+ // below applies either way -- when the two paths were separate, only the explicit one ever
+ // converted its dates.
+ int count = meta.isSpecifyFields() ? meta.getFields().size() : inputRowMeta.size();
+ data.insertRowMeta = new RowMeta();
+ data.streamFieldIndexes = new int[count];
+ data.writeValueMeta = new IValueMeta[count];
+ data.sourceValueMeta = new IValueMeta[count];
+
+ for (int i = 0; i < count; i++) {
+ String streamField =
+ meta.isSpecifyFields()
+ ? meta.getFields().get(i).getStreamField()
+ : inputRowMeta.getValueMeta(i).getName();
+ String databaseField =
+ meta.isSpecifyFields() ? meta.getFields().get(i).getDatabaseField() : streamField;
+
+ int index = meta.isSpecifyFields() ? inputRowMeta.indexOfValue(streamField) : i;
+ if (index < 0 && meta.isErrorColumnMismatch()) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG, "RedshiftBulkLoader.Exception.FailedToFindField", streamField));
+ }
+ data.streamFieldIndexes[i] = index;
+
+ IValueMeta inputValueMeta =
+ index >= 0 ? inputRowMeta.getValueMeta(index) : new ValueMetaString(streamField);
+
+ // Dates and timestamps are written in the format the COPY statement declares. A Hop Date
+ // carries a time of day just as a Timestamp does, so both are written whole and the target
+ // column decides whether to keep the time.
+ IValueMeta writeValueMeta = inputValueMeta;
+ IValueMeta sourceValueMeta = null;
+ if (inputValueMeta.getType() == IValueMeta.TYPE_TIMESTAMP
+ || inputValueMeta.getType() == IValueMeta.TYPE_DATE) {
+ writeValueMeta = new ValueMetaDate();
+ writeValueMeta.setConversionMask(TIMESTAMP_CONVERSION_MASK);
+ sourceValueMeta = inputValueMeta;
+ }
+ data.writeValueMeta[i] = writeValueMeta;
+ data.sourceValueMeta[i] = sourceValueMeta;
+
+ IValueMeta insertValueMeta = inputValueMeta.clone();
+ insertValueMeta.setName(databaseField);
+ data.insertRowMeta.addValueMeta(insertValueMeta);
+ }
+
+ // What the COPY statement has to name, since the file holds these columns and no others.
+ data.columnNames = data.insertRowMeta.getFieldNames();
}
/**
@@ -271,8 +302,16 @@ public class RedshiftBulkLoader
return returnValue;
}
- private String buildCopyStatementSqlString() {
+ /**
+ * Build the Redshift COPY statement for the file on S3.
+ *
+ * @param maskCredentials when true the AWS credentials are replaced by a placeholder, so the
+ * statement can safely be written to the log
+ * @return the COPY statement
+ */
+ String buildCopyStatementSqlString(boolean maskCredentials) throws HopException {
final DatabaseMeta databaseMeta = data.db.getDatabaseMeta();
+ boolean csv = isCsvFormat();
StringBuilder sb = new StringBuilder(150);
sb.append("COPY ");
@@ -283,74 +322,167 @@ public class RedshiftBulkLoader
data.db.resolve(meta.getSchemaName()),
data.db.resolve(meta.getTableName())));
- if (meta.isStreamToS3Csv() || meta.getLoadFromExistingFileFormat().equals("CSV")) {
+ // Name the columns the file holds. Without this Redshift expects a value for every column of
+ // the table, in table order, and a narrower file fails with "Delimiter not found".
+ String[] columns = copyColumnNames();
+ if (csv && columns.length > 0) {
sb.append(" (");
- List<RedshiftBulkLoaderField> fieldList = meta.getFields();
- for (int i = 0; i < fieldList.size(); i++) {
- RedshiftBulkLoaderField field = fieldList.get(i);
+ for (int i = 0; i < columns.length; i++) {
if (i > 0) {
- sb.append(", ").append(field.getDatabaseField());
- } else {
- sb.append(field.getDatabaseField());
+ sb.append(", ");
}
+ sb.append(columns[i]);
}
sb.append(")");
}
- sb.append(" FROM '" + resolve(meta.getCopyFromFilename()) + "'");
- if (meta.isStreamToS3Csv() || meta.getLoadFromExistingFileFormat().equals("CSV")) {
+ sb.append(" FROM '").append(resolve(meta.getCopyFromFilename())).append("'");
+ if (csv) {
sb.append(" DELIMITER ',' ");
sb.append(" CSV QUOTE AS '\"'");
sb.append(" NULL '' ");
sb.append(" EMPTYASNULL ");
- sb.append("DATEFORMAT AS 'YYYY/MM/DD' ");
- sb.append("TIMEFORMAT AS 'YYYY/MM/DD HH:MI:SS'");
+ // 'auto' is the only setting that accepts fractional seconds; the explicit TIMEFORMAT
+ // patterns have no token for them, so a value carrying milliseconds is rejected.
+ sb.append("DATEFORMAT AS 'auto' ");
+ sb.append("TIMEFORMAT AS 'auto'");
}
if (meta.isUseAwsIamRole()) {
- sb.append(" iam_role '" + meta.getAwsIamRole() + "'");
- } else if (meta.isUseCredentials()) {
- String awsAccessKeyId = "";
- String awsSecretAccessKey = "";
- if (meta.isUseSystemEnvVars()) {
- awsAccessKeyId = System.getenv("AWS_ACCESS_KEY_ID");
- awsSecretAccessKey = System.getenv("AWS_SECRET_ACCESS_KEY");
+ sb.append(" iam_role '")
+ .append(maskCredentials ? CREDENTIALS_MASK : resolve(meta.getAwsIamRole()))
+ .append("'");
+ } else if (meta.isUseConnectionCredentials() || meta.isUseCredentials()) {
+ if (maskCredentials) {
+ sb.append(" CREDENTIALS '").append(CREDENTIALS_MASK).append("'");
} else {
- awsAccessKeyId = resolve(meta.getAwsAccessKeyId());
- awsSecretAccessKey = resolve(meta.getAwsSecretAccessKey());
+ sb.append(" CREDENTIALS '").append(buildCredentialsClause()).append("'");
}
- sb.append(
- " CREDENTIALS 'aws_access_key_id="
- + awsAccessKeyId
- + ";aws_secret_access_key="
- + awsSecretAccessKey
- + "'");
- }
- if (!StringUtils.isEmpty(meta.getLoadFromExistingFileFormat())
- && meta.getLoadFromExistingFileFormat().equals("Parquet")) {
- sb.append(" FORMAT AS PARQUET;");
}
-
- if (isDebug()) {
- logDebug("copy stmt: " + sb);
+ if (RedshiftBulkLoaderMeta.FILE_FORMAT_PARQUET.equals(meta.getLoadFromExistingFileFormat())) {
+ sb.append(" FORMAT AS PARQUET;");
}
return sb.toString();
}
- private Object[] writeToOutputStream(Object[] r) {
- assert (r != null);
+ /**
+ * The body of the COPY statement's CREDENTIALS clause, from wherever this transform is configured
+ * to get its AWS credentials.
+ *
+ * @return the credential key/value pairs, without the surrounding quotes
+ * @throws HopException when the credentials cannot be worked out
+ */
+ private String buildCredentialsClause() throws HopException {
+ String accessKeyId;
+ String secretAccessKey;
+ String sessionToken = null;
+
+ if (meta.isUseConnectionCredentials()) {
+ AwsCredentials credentials = resolveConnectionCredentials();
+ accessKeyId = credentials.accessKeyId();
+ secretAccessKey = credentials.secretAccessKey();
+ if (credentials instanceof AwsSessionCredentials session) {
+ sessionToken = session.sessionToken();
+ }
+ } else if (meta.isUseSystemEnvVars()) {
+ accessKeyId = System.getenv("AWS_ACCESS_KEY_ID");
+ secretAccessKey = System.getenv("AWS_SECRET_ACCESS_KEY");
+ sessionToken = System.getenv("AWS_SESSION_TOKEN");
+ } else {
+ accessKeyId = resolve(meta.getAwsAccessKeyId());
+ secretAccessKey = resolve(meta.getAwsSecretAccessKey());
+ }
+
+ StringBuilder clause = new StringBuilder();
+ clause
+ .append("aws_access_key_id=")
+ .append(Const.NVL(accessKeyId, ""))
+ .append(";aws_secret_access_key=")
+ .append(Const.NVL(secretAccessKey, ""));
+ if (StringUtils.isNotEmpty(sessionToken)) {
+ clause.append(";token=").append(sessionToken);
+ }
+ return clause.toString();
+ }
- Object[] insertRowData = r;
- Object[] outputRowData = r;
+ /**
+ * Take the credentials the Redshift connection is configured with. A connection holding an access
+ * key hands it straight over; one pointing at a profile or leaving it to the AWS default chain is
+ * resolved here, so the COPY statement gets a concrete key -- including a session token when the
+ * credentials are temporary.
+ *
+ * @return the credentials to put in the COPY statement
+ * @throws HopException when the connection has no AWS credentials to give
+ */
+ private AwsCredentials resolveConnectionCredentials() throws HopException {
+ IDatabase database = data.databaseMeta == null ? null : data.databaseMeta.getIDatabase();
+ if (!(database instanceof RedshiftDatabaseMeta redshift)) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "RedshiftBulkLoaderMeta.Error.NotARedshiftConnection"));
+ }
- if (meta.specifyFields()) {
- insertRowData = new Object[data.selectedRowFieldIndices.length];
- for (int idx = 0; idx < data.selectedRowFieldIndices.length; idx++) {
- insertRowData[idx] = r[data.selectedRowFieldIndices[idx]];
+ try {
+ switch (redshift.getAuthenticationType()) {
+ case IAM_CREDENTIALS:
+ String key = Const.NVL(resolve(redshift.getAwsAccessKeyId()), "");
+ String secret = Const.NVL(decrypt(redshift.getAwsSecretAccessKey()), "");
+ String token = decrypt(redshift.getAwsSessionToken());
+ return StringUtils.isEmpty(token)
+ ? AwsBasicCredentials.create(key, secret)
+ : AwsSessionCredentials.create(key, secret, token);
+ case IAM_PROFILE:
+ return ProfileCredentialsProvider.builder()
+ .profileName(resolve(redshift.getAwsProfile()))
+ .build()
+ .resolveCredentials();
+ case IAM_DEFAULT_CHAIN:
+ return DefaultCredentialsProvider.create().resolveCredentials();
+ default:
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "RedshiftBulkLoaderMeta.Error.ConnectionHasNoAwsCredentials"));
}
+ } catch (HopException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "RedshiftBulkLoaderMeta.Error.CredentialsNotResolved"), e);
+ }
+ }
+
+ private String decrypt(String value) {
+ return Encr.decryptPasswordOptionallyEncrypted(resolve(value));
+ }
+
+ /**
+ * The columns the COPY statement should name.
+ *
+ * <p>An explicit mapping says it outright. Without one, the columns are known only for a file
+ * this transform wrote itself, where they are the fields of the stream. A pre-existing file with
+ * no mapping is the one case we cannot speak for, so the statement stays silent and Redshift
+ * matches the file against the table positionally.
+ *
+ * @return the column names, empty when they cannot be established
+ */
+ private String[] copyColumnNames() {
+ if (meta.isSpecifyFields()) {
+ return meta.getFields().stream()
+ .map(RedshiftBulkLoaderField::getDatabaseField)
+ .toArray(String[]::new);
+ }
+ if (meta.isStreamToS3Csv() && data.columnNames != null) {
+ return data.columnNames;
}
+ return new String[0];
+ }
- return outputRowData;
+ /**
+ * @return true when the file we load is a CSV file: either one we streamed to S3 ourselves, or an
+ * existing one the user declared as CSV
+ */
+ private boolean isCsvFormat() {
+ return meta.isStreamToS3Csv()
+ || RedshiftBulkLoaderMeta.FILE_FORMAT_CSV.equals(meta.getLoadFromExistingFileFormat());
}
/**
@@ -364,15 +496,15 @@ public class RedshiftBulkLoader
private void getDbFields() throws HopException {
data.dbFields = new ArrayList<>();
- IRowMeta rowMeta = null;
+ String schemaName = resolve(meta.getSchemaName());
+ String tableName = resolve(meta.getTableName());
- if (!StringUtils.isEmpty(resolve(meta.getSchemaName()))) {
- rowMeta = data.db.getTableFields(meta.getSchemaName() + "." + meta.getTableName());
- } else {
- rowMeta = data.db.getTableFields(meta.getTableName());
- }
+ IRowMeta rowMeta =
+ StringUtils.isEmpty(schemaName)
+ ? data.db.getTableFields(tableName)
+ : data.db.getTableFields(schemaName + "." + tableName);
try {
- if (rowMeta.isEmpty()) {
+ if (rowMeta == null || rowMeta.isEmpty()) {
throw new HopException("No fields found in table");
}
@@ -395,12 +527,83 @@ public class RedshiftBulkLoader
}
}
+ /**
+ * Make sure the folder the staging file goes in exists. This file is the transform's own
+ * business, so there is nothing to ask the user about: on S3 a prefix only exists while an object
+ * sits under it, and a brand new path would otherwise be refused with "Parent directory ... does
+ * not exist".
+ *
+ * @param filename the file about to be written
+ * @throws HopException when the folder is missing and cannot be created
+ */
+ void ensureParentFolderExists(String filename) throws HopException {
+ try {
+ FileObject parent = HopVfs.getFileObject(filename, variables).getParent();
+ if (parent == null || parent.exists()) {
+ return;
+ }
+ parent.createFolder();
+ if (isDetailed()) {
+ logDetailed(
+ BaseMessages.getString(
+ PKG, "RedshiftBulkLoader.Log.ParentFolderCreated", HopVfs.getFriendlyURI(parent)));
+ }
+ } catch (Exception e) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "RedshiftBulkLoaderMeta.Error.CannotCreateParentFolder", filename),
+ e);
+ }
+ }
+
+ /**
+ * The COPY statement always reads a file from S3, so we need to know which one. When we do not
+ * write that file ourselves we also need to know its format.
+ *
+ * @throws HopException when the file settings are incomplete
+ */
+ /**
+ * Fail before any row is read when a column was mapped to a table column that does not exist. The
+ * COPY statement would reject it anyway, but only after the whole file was written to S3.
+ *
+ * @throws HopException when a selected column is not a column of the target table
+ */
+ protected void verifyTableFields() throws HopException {
+ if (!meta.isSpecifyFields()) {
+ return;
+ }
+ for (RedshiftBulkLoaderField field : meta.getFields()) {
+ boolean found = false;
+ for (String[] dbField : data.dbFields) {
+ if (dbField[0].equalsIgnoreCase(field.getDatabaseField())) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ throw new HopException(
+ "Field [" + field.getDatabaseField() + "] couldn't be found in the table!");
+ }
+ }
+ }
+
+ protected void verifyFileSettings() throws HopException {
+ if (StringUtils.isEmpty(resolve(meta.getCopyFromFilename()))) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "RedshiftBulkLoaderMeta.Error.NoCopyFromFilename"));
+ }
+ if (!meta.isStreamToS3Csv() && StringUtils.isEmpty(meta.getLoadFromExistingFileFormat())) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "RedshiftBulkLoaderMeta.Error.NoFileFormat"));
+ }
+ }
+
/**
* Initialize the binary values of delimiters, enclosures, and escape characters
*
* @throws HopException
*/
- private void initBinaryDataFields() throws HopException {
+ void initBinaryDataFields() throws HopException {
try {
data.binarySeparator = new byte[] {};
data.binaryEnclosure = new byte[] {};
@@ -429,84 +632,24 @@ public class RedshiftBulkLoader
* @param row The input row
* @throws HopTransformException
*/
- private void writeRowToFile(IRowMeta rowMeta, Object[] row) throws HopTransformException {
+ void writeRowToFile(IRowMeta rowMeta, Object[] row) throws HopTransformException {
try {
-
- if (meta.isStreamToS3Csv() && !meta.isSpecifyFields()) {
- /*
- * Write all values in stream to text file.
- */
- for (int i = 0; i < rowMeta.size(); i++) {
- if (i > 0 && data.binarySeparator.length > 0) {
- data.writer.write(data.binarySeparator);
- }
- IValueMeta v = rowMeta.getValueMeta(i);
- Object valueData = row[i];
-
- // no special null value default was specified since no fields are specified at all
- // As such, we pass null
- //
- writeField(v, valueData, null);
+ // The columns, in the order the COPY statement names them.
+ for (int i = 0; i < data.streamFieldIndexes.length; i++) {
+ if (i > 0) {
+ data.writer.write(data.binarySeparator);
}
- data.writer.write(data.binaryNewline);
- } else if (meta.isStreamToS3Csv() && meta.isSpecifyFields()) {
- /*
- * Only write the fields specified!
- */
- for (int i = 0; i < meta.getFields().size(); i++) {
- if (meta.getFields().get(i).getDatabaseField() != null) {
- if (i > 0 && data.binarySeparator.length > 0) {
- data.writer.write(data.binarySeparator);
- }
-
- IValueMeta v = null;
- String streamFieldName = meta.getFields().get(i).getStreamField();
- String[] rowFields = data.outputRowMeta.getFieldNames();
- String streamFieldType = "";
- int streamIndex = -1;
- for (int j = 0; j < rowFields.length; j++) {
- if (streamFieldName.equals(rowFields[j])) {
- v = rowMeta.getValueMeta(j);
- streamIndex = j;
- }
- }
-
- boolean needConversion = false;
- if (v.getType() == IValueMeta.TYPE_TIMESTAMP) {
- v = new ValueMetaDate();
- v.setConversionMask("yyyy/MM/dd HH:mm:ss.SSS");
- needConversion = true;
- } else if (v.getType() == IValueMeta.TYPE_DATE) {
- v = new ValueMetaDate();
- v.setConversionMask("yyyy/MM/dd");
- needConversion = true;
- }
-
- Object valueData = null;
- if (streamIndex >= 0) {
- if (needConversion) {
- IValueMeta valueMeta = rowMeta.getValueMeta(streamIndex);
- Object obj = row[streamIndex];
- valueData = v.convertData(valueMeta, obj);
- } else {
- valueData = row[streamIndex];
- }
- } else if (meta.isErrorColumnMismatch()) {
- throw new HopException(
- "Error column mismatch: Database streamField "
- + meta.getFields().get(i).getStreamField()
- + " not found on stream.");
- }
- writeField(v, valueData, data.binaryNullValue);
- }
+ int index = data.streamFieldIndexes[i];
+ IValueMeta valueMeta = data.writeValueMeta[i];
+ Object valueData = null;
+ if (index >= 0) {
+ IValueMeta sourceMeta = data.sourceValueMeta[i];
+ valueData =
+ sourceMeta == null ? row[index] : valueMeta.convertData(sourceMeta, row[index]);
}
- data.writer.write(data.binaryNewline);
- } else {
- int jsonField = data.fieldnrs.get("json");
- data.writer.write(
- data.insertRowMeta.getString(row, jsonField).getBytes(StandardCharsets.UTF_8));
- data.writer.write(data.binaryNewline);
+ writeField(valueMeta, valueData, data.binaryNullValue);
}
+ data.writer.write(data.binaryNewline);
} catch (Exception e) {
throw new HopTransformException("Error writing line", e);
}
@@ -535,42 +678,18 @@ public class RedshiftBulkLoader
}
if (str != null && str.length > 0) {
- List<Integer> enclosures = null;
- boolean writeEnclosures = false;
-
- if (v.isString()) {
- writeEnclosures = true;
-
- if (containsSeparatorOrEnclosure(
- str, data.binarySeparator, data.binaryEnclosure, data.escapeCharacters)) {
- writeEnclosures = true;
- }
- }
+ // Strings are always enclosed, the COPY statement declares the same quote character.
+ // Anything else is enclosed only when its own content would otherwise break the row.
+ // JSON is the case that matters: Hop does not consider it a string, yet it is full of
+ // commas and quotes and is often pretty printed across several lines.
+ boolean writeEnclosures = v.isString() || needsEnclosure(str);
if (writeEnclosures) {
data.writer.write(data.binaryEnclosure);
- enclosures = getEnclosurePositions(str);
- }
-
- if (enclosures == null) {
- data.writer.write(str);
- } else {
- // Skip the enclosures, escape them instead...
- int from = 0;
- for (Integer enclosure : enclosures) {
- // Minus one to write the escape before the enclosure
- int position = enclosure;
- data.writer.write(str, from, position - from);
- data.writer.write(data.escapeCharacters); // write enclosure a second time
- from = position;
- }
- if (from < str.length) {
- data.writer.write(str, from, str.length - from);
- }
- }
-
- if (writeEnclosures) {
+ writeEscaped(str);
data.writer.write(data.binaryEnclosure);
+ } else {
+ data.writer.write(str);
}
}
} catch (Exception e) {
@@ -633,136 +752,58 @@ public class RedshiftBulkLoader
}
/**
- * Check if a string contains separators or enclosures. Can be used to determine if the string
- * needs enclosures around it or not.
+ * Whether a value has to be enclosed to survive the trip through the CSV file: it is only safe to
+ * write bare when it holds no separator, no quote and no line break.
*
- * @param source The string to check
- * @param separator The separator character(s)
- * @param enclosure The enclosure character(s)
- * @param escape The escape character(s)
- * @return True if the string contains separators or enclosures
+ * @param str The bytes of the value
+ * @return true if the value must be enclosed
*/
- private boolean containsSeparatorOrEnclosure(
- byte[] source, byte[] separator, byte[] enclosure, byte[] escape) {
- boolean result = false;
-
- boolean enclosureExists = enclosure != null && enclosure.length > 0;
- boolean separatorExists = separator != null && separator.length > 0;
- boolean escapeExists = escape != null && escape.length > 0;
-
- // Skip entire test if neither separator nor enclosure exist
- if (separatorExists || enclosureExists || escapeExists) {
-
- // Search for the first occurrence of the separator or enclosure
- for (int index = 0; !result && index < source.length; index++) {
- if (enclosureExists && source[index] == enclosure[0]) {
-
- // Potential match found, make sure there are enough bytes to support a full match
- if (index + enclosure.length <= source.length) {
- // First byte of enclosure found
- result = true; // Assume match
- for (int i = 1; i < enclosure.length; i++) {
- if (source[index + i] != enclosure[i]) {
- // Enclosure match is proven false
- result = false;
- break;
- }
- }
- }
-
- } else if (separatorExists && source[index] == separator[0]) {
-
- // Potential match found, make sure there are enough bytes to support a full match
- if (index + separator.length <= source.length) {
- // First byte of separator found
- result = true; // Assume match
- for (int i = 1; i < separator.length; i++) {
- if (source[index + i] != separator[i]) {
- // Separator match is proven false
- result = false;
- break;
- }
- }
- }
-
- } else if (escapeExists
- && source[index] == escape[0]
- && index + escape.length <= source.length) {
- // Potential match found, make sure there are enough bytes to support a full match
- // First byte of separator found
- result = true; // Assume match
- for (int i = 1; i < escape.length; i++) {
- if (source[index + i] != escape[i]) {
- // Separator match is proven false
- result = false;
- break;
- }
- }
- }
+ private boolean needsEnclosure(byte[] str) {
+ byte separator = data.binarySeparator.length > 0 ? data.binarySeparator[0] : 0;
+ byte enclosure = data.binaryEnclosure.length > 0 ? data.binaryEnclosure[0] : 0;
+
+ for (byte b : str) {
+ if (b == separator || b == enclosure || b == '\n' || b == '\r') {
+ return true;
}
}
- return result;
+ return false;
}
/**
- * Gets the positions of any double quotes or backslashes in the string
+ * Write the value, doubling every occurrence of the enclosure character so the CSV file stays
+ * readable for the COPY statement. Only bytes that actually need escaping are copied separately,
+ * a value without enclosures is written in one go.
*
- * @param str The string to check
- * @return The positions within the string of double quotes and backslashes.
+ * @param str The bytes of the value
+ * @throws IOException in case the value could not be written
*/
- private List<Integer> getEnclosurePositions(byte[] str) {
- List<Integer> positions = null;
- // +1 because otherwise we will not find it at the end
- for (int i = 0, len = str.length; i < len; i++) {
- // verify if on position i there is an enclosure
- //
- boolean found = true;
- for (int x = 0; found && x < data.binaryEnclosure.length; x++) {
- if (str[i + x] != data.binaryEnclosure[x]) {
- found = false;
- }
- }
-
- if (!found) {
- found = true;
- for (int x = 0; found && x < data.escapeCharacters.length; x++) {
- if (str[i + x] != data.escapeCharacters[x]) {
- found = false;
- }
- }
- }
-
- if (found) {
- if (positions == null) {
- positions = new ArrayList<>();
- }
- positions.add(i);
+ private void writeEscaped(byte[] str) throws IOException {
+ if (data.binaryEnclosure.length == 0) {
+ data.writer.write(str);
+ return;
+ }
+ byte enclosure = data.binaryEnclosure[0];
+
+ int from = 0;
+ for (int i = 0; i < str.length; i++) {
+ if (str[i] == enclosure) {
+ data.writer.write(str, from, i - from);
+ data.writer.write(data.escapeCharacters); // write the enclosure a second time
+ from = i;
}
}
- return positions;
+ data.writer.write(str, from, str.length - from);
}
@Override
public void stopRunning() throws HopException {
setStopped(true);
- if (data.workerThread != null) {
- synchronized (data.workerThread) {
- if (data.workerThread.isAlive() && !data.workerThread.isInterrupted()) {
- try {
- data.workerThread.interrupt();
- data.workerThread.join();
- } catch (InterruptedException e) {
- // Do nothing
- }
- }
- }
- }
-
super.stopRunning();
}
void truncateTable() throws HopDatabaseException {
- if (meta.isTruncateTable() && ((getCopy() == 0) || !Utils.isEmpty(getPartitionId()))) {
+ if ((getCopy() == 0) || !Utils.isEmpty(getPartitionId())) {
data.db.truncateTable(resolve(meta.getSchemaName()), resolve(meta.getTableName()));
}
}
@@ -772,23 +813,17 @@ public class RedshiftBulkLoader
setOutputDone();
- try {
- if (getErrors() > 0) {
- data.db.rollback();
- }
- } catch (HopDatabaseException e) {
- logError("Unexpected error rolling back the database connection.", e);
- }
+ // The stream is normally closed before the COPY statement runs, this catches the error paths.
+ closeFile();
- if (data.workerThread != null) {
+ if (data.db != null) {
try {
- data.workerThread.join();
- } catch (InterruptedException e) {
- // Do nothing
+ if (getErrors() > 0) {
+ data.db.rollback();
+ }
+ } catch (HopDatabaseException e) {
+ logError("Unexpected error rolling back the database connection.", e);
}
- }
-
- if (data.db != null) {
data.db.disconnect();
}
super.dispose();
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderData.java b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderData.java
index 6eb40a10ac..ebd7693b97 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderData.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderData.java
@@ -17,14 +17,12 @@
package org.apache.hop.pipeline.transforms.redshift.bulkloader;
-import java.io.IOException;
import java.io.OutputStream;
-import java.io.PipedInputStream;
import java.util.ArrayList;
-import java.util.Map;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.pipeline.transform.BaseTransformData;
import org.apache.hop.pipeline.transform.ITransformData;
@@ -33,8 +31,6 @@ public class RedshiftBulkLoaderData extends BaseTransformData implements ITransf
protected Database db;
protected DatabaseMeta databaseMeta;
- protected int[] selectedRowFieldIndices;
-
protected IRowMeta outputRowMeta;
protected IRowMeta insertRowMeta;
@@ -43,8 +39,29 @@ public class RedshiftBulkLoaderData extends BaseTransformData implements ITransf
// data type
public ArrayList<String[]> dbFields;
- // Maps table fields to the location of the corresponding field on the input stream.
- public Map<String, Integer> fieldnrs;
+ /**
+ * Per selected column: the index of the corresponding field on the input row, or -1 when the
+ * field is not present on the stream. Resolved once, when the first row comes in.
+ */
+ protected int[] streamFieldIndexes;
+
+ /** Per selected column: the value meta used to render the value in the CSV file. */
+ protected IValueMeta[] writeValueMeta;
+
+ /**
+ * Per selected column: the value meta of the field on the input row, but only when the value
+ * needs to be converted before it is written. {@code null} means the value can be written as is.
+ */
+ protected IValueMeta[] sourceValueMeta;
+
+ /**
+ * The names of the columns the CSV file actually holds, in the order they were written. Only set
+ * when this transform wrote the file itself.
+ */
+ protected String[] columnNames;
+
+ /** Set as soon as a first row is read, so we know whether the stream was empty. */
+ protected boolean rowsReceived;
protected OutputStream writer;
// Byte arrays for constant characters put into output files.
@@ -54,10 +71,6 @@ public class RedshiftBulkLoaderData extends BaseTransformData implements ITransf
public byte[] binaryNewline;
public byte[] binaryNullValue;
- protected PipedInputStream pipedInputStream;
-
- protected volatile Thread workerThread;
-
public RedshiftBulkLoaderData() {
super();
@@ -67,8 +80,4 @@ public class RedshiftBulkLoaderData extends BaseTransformData implements ITransf
public IRowMeta getInsertRowMeta() {
return insertRowMeta;
}
-
- public void close() throws IOException {
- // Do nothing
- }
}
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderDialog.java b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderDialog.java
index 2bf26de1ac..e05d462606 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderDialog.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderDialog.java
@@ -112,9 +112,10 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
private ColumnInfo[] ciFields;
+ private static final String AWS_INHERIT = "Inherit from connection";
private static final String AWS_CREDENTIALS = "Credentials";
private static final String AWS_IAM_ROLE = "IAM Role";
- private String[] awsAuthOptions = new String[] {AWS_CREDENTIALS, AWS_IAM_ROLE};
+ private String[] awsAuthOptions = new String[] {AWS_INHERIT, AWS_CREDENTIALS, AWS_IAM_ROLE};
private Label wlAwsAuthType;
private ComboVar wAwsAuthType;
@@ -257,7 +258,9 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
input.setChanged(backupChanged);
toggleSpecifyFieldsFlags();
+ toggleAuthSelection();
toggleKeysSelection();
+ toggleStreamToS3Flags();
focusTransformName();
BaseDialog.defaultShellHandling(shell, c -> ok(), c -> cancel());
return transformName;
@@ -399,6 +402,7 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
lastControl = wlStreamToS3Csv;
+ wStreamToS3Csv.addSelectionListener(lsSelMod);
wStreamToS3Csv.addSelectionListener(
new SelectionAdapter() {
@Override
@@ -406,7 +410,7 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
if (wStreamToS3Csv.getSelection()) {
wLoadFromExistingFileFormat.setText("");
}
- wLoadFromExistingFileFormat.setEnabled(!wStreamToS3Csv.getSelection());
+ toggleStreamToS3Flags();
}
});
@@ -429,8 +433,11 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
fdLoadFromExistingFile.left = new FormAttachment(middle, 0);
fdLoadFromExistingFile.right = new FormAttachment(100, 0);
wLoadFromExistingFileFormat.setLayoutData(fdLoadFromExistingFile);
- String[] fileFormats = {"CSV", "Parquet"};
+ String[] fileFormats = {
+ RedshiftBulkLoaderMeta.FILE_FORMAT_CSV, RedshiftBulkLoaderMeta.FILE_FORMAT_PARQUET
+ };
wLoadFromExistingFileFormat.setItems(fileFormats);
+ wLoadFromExistingFileFormat.addModifyListener(lsMod);
lastControl = wLoadFromExistingFileFormat;
Label wlCopyFromFile = new Label(wGeneralComp, SWT.RIGHT);
@@ -492,7 +499,7 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
wlAwsAuthType.setLayoutData(fdlAwsAuthType);
wAwsAuthType = new ComboVar(variables, wAwsAuthComp, SWT.BORDER | SWT.READ_ONLY);
wAwsAuthType.setItems(awsAuthOptions);
- wAwsAuthType.setText(awsAuthOptions[0]);
+ wAwsAuthType.setText(AWS_INHERIT);
PropsUi.setLook(wAwsAuthType);
FormData fdAwsAuthType = new FormData();
fdAwsAuthType.top = new FormAttachment(0, margin);
@@ -527,6 +534,7 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
new SelectionAdapter() {
@Override
public void widgetSelected(SelectionEvent e) {
+ input.setChanged();
toggleKeysSelection();
}
});
@@ -595,6 +603,7 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
new SelectionAdapter() {
@Override
public void widgetSelected(SelectionEvent e) {
+ input.setChanged();
toggleAuthSelection();
}
});
@@ -945,40 +954,25 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
/** Copy information from the meta-data input to the dialog fields. */
public void getData() {
- if (!StringUtils.isEmpty(input.getConnection())) {
- wConnection.setText(input.getConnection());
- }
- if (!StringUtils.isEmpty(input.getSchemaName())) {
- wSchema.setText(input.getSchemaName());
- }
- if (!StringUtils.isEmpty(input.getTableName())) {
- wTable.setText(input.getTableName());
- }
- if (input.isUseCredentials()) {
- wAwsAuthType.setText(awsAuthOptions[0]);
- wUseSystemVars.setSelection(input.isUseSystemEnvVars());
- if (!input.isUseSystemEnvVars()) {
- if (!StringUtil.isEmpty(input.getAwsAccessKeyId())) {
- wAccessKeyId.setText(input.getAwsAccessKeyId());
- }
- if (!StringUtils.isEmpty(input.getAwsSecretAccessKey())) {
- wAccessKeyId.setText(input.getAwsSecretAccessKey());
- }
- }
- } else if (input.isUseAwsIamRole()) {
- wAwsAuthType.setText(awsAuthOptions[1]);
- if (!StringUtils.isEmpty(input.getAwsIamRole())) {
- wAwsIamRole.setText(input.getAwsIamRole());
- }
+ wConnection.setText(Const.NVL(input.getConnection(), ""));
+ wSchema.setText(Const.NVL(input.getSchemaName(), ""));
+ wTable.setText(Const.NVL(input.getTableName(), ""));
+ if (input.isUseAwsIamRole()) {
+ wAwsAuthType.setText(AWS_IAM_ROLE);
+ } else if (input.isUseCredentials()) {
+ wAwsAuthType.setText(AWS_CREDENTIALS);
+ } else {
+ // Covers both an explicit inherit and a transform that never had the option.
+ wAwsAuthType.setText(AWS_INHERIT);
}
+ wUseSystemVars.setSelection(input.isUseSystemEnvVars());
+ wAccessKeyId.setText(Const.NVL(input.getAwsAccessKeyId(), ""));
+ wSecretAccessKey.setText(Const.NVL(input.getAwsSecretAccessKey(), ""));
+ wAwsIamRole.setText(Const.NVL(input.getAwsIamRole(), ""));
wStreamToS3Csv.setSelection(input.isStreamToS3Csv());
- if (!StringUtils.isEmpty(input.getLoadFromExistingFileFormat())) {
- wLoadFromExistingFileFormat.setText(input.getLoadFromExistingFileFormat());
- }
- if (!StringUtils.isEmpty(input.getCopyFromFilename())) {
- wCopyFromFilename.setText(input.getCopyFromFilename());
- }
+ wLoadFromExistingFileFormat.setText(Const.NVL(input.getLoadFromExistingFileFormat(), ""));
+ wCopyFromFilename.setText(Const.NVL(input.getCopyFromFilename(), ""));
wTruncate.setSelection(input.isTruncateTable());
wOnlyWhenHaveRows.setSelection(input.isOnlyWhenHaveRows());
@@ -1004,45 +998,25 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
}
private void getInfo(RedshiftBulkLoaderMeta info) {
- if (!StringUtils.isEmpty(wConnection.getText())) {
- info.setConnection(wConnection.getText());
- }
- if (!StringUtils.isEmpty(wSchema.getText())) {
- info.setSchemaName(wSchema.getText());
- }
- if (!StringUtils.isEmpty(wTable.getText())) {
- info.setTablename(wTable.getText());
- }
- if (wAwsAuthType.getText().equals(AWS_CREDENTIALS)) {
- info.setUseCredentials(true);
- info.setUseAwsIamRole(false);
- if (wUseSystemVars.getSelection()) {
- info.setUseSystemEnvVars(true);
- } else {
- info.setUseSystemEnvVars(false);
- if (!StringUtils.isEmpty(wAccessKeyId.getText())) {
- info.setAwsAccessKeyId(wAccessKeyId.getText());
- }
- if (!StringUtil.isEmpty(wSecretAccessKey.getText())) {
- info.setAwsSecretAccessKey(wSecretAccessKey.getText());
- }
- }
- } else if (wAwsAuthType.getText().equals(AWS_IAM_ROLE)) {
- info.setUseCredentials(false);
- info.setUseAwsIamRole(true);
- if (!StringUtils.isEmpty(wAwsIamRole.getText())) {
- info.setAwsIamRole(wAwsIamRole.getText());
- }
- }
+ info.setConnection(wConnection.getText());
+ info.setSchemaName(wSchema.getText());
+ info.setTablename(wTable.getText());
+
+ boolean useIamRole = AWS_IAM_ROLE.equals(wAwsAuthType.getText());
+ boolean inherit = AWS_INHERIT.equals(wAwsAuthType.getText());
+ info.setUseAwsIamRole(useIamRole);
+ info.setUseConnectionCredentials(inherit);
+ info.setUseCredentials(!useIamRole && !inherit);
+ info.setUseSystemEnvVars(!useIamRole && !inherit && wUseSystemVars.getSelection());
+ info.setAwsAccessKeyId(wAccessKeyId.getText());
+ info.setAwsSecretAccessKey(wSecretAccessKey.getText());
+ info.setAwsIamRole(wAwsIamRole.getText());
+
info.setTruncateTable(wTruncate.getSelection());
info.setOnlyWhenHaveRows(wOnlyWhenHaveRows.getSelection());
info.setStreamToS3Csv(wStreamToS3Csv.getSelection());
- if (!StringUtils.isEmpty(wLoadFromExistingFileFormat.getText())) {
- info.setLoadFromExistingFileFormat(wLoadFromExistingFileFormat.getText());
- }
- if (!StringUtils.isEmpty(wCopyFromFilename.getText())) {
- info.setCopyFromFilename(wCopyFromFilename.getText());
- }
+ info.setLoadFromExistingFileFormat(wLoadFromExistingFileFormat.getText());
+ info.setCopyFromFilename(wCopyFromFilename.getText());
info.setSpecifyFields(wSpecifyFields.getSelection());
@@ -1241,34 +1215,29 @@ public class RedshiftBulkLoaderDialog extends BaseTransformDialog {
}
public void toggleAuthSelection() {
- if (wAwsAuthType.getText().equals(AWS_CREDENTIALS)) {
- wlUseSystemVars.setEnabled(true);
- wUseSystemVars.setEnabled(true);
- wlAccessKeyId.setEnabled(true);
- wAccessKeyId.setEnabled(true);
- wlSecretAccessKey.setEnabled(true);
- wSecretAccessKey.setEnabled(true);
-
- wlAwsIamRole.setEnabled(false);
- wAwsIamRole.setEnabled(false);
- }
- if (wAwsAuthType.getText().equals(AWS_IAM_ROLE)) {
- wlUseSystemVars.setEnabled(false);
- wUseSystemVars.setEnabled(false);
- wlAccessKeyId.setEnabled(false);
- wAccessKeyId.setEnabled(false);
- wlSecretAccessKey.setEnabled(false);
- wSecretAccessKey.setEnabled(false);
-
- wlAwsIamRole.setEnabled(true);
- wAwsIamRole.setEnabled(true);
- }
+ boolean useIamRole = AWS_IAM_ROLE.equals(wAwsAuthType.getText());
+ boolean useCredentials = AWS_CREDENTIALS.equals(wAwsAuthType.getText());
+
+ // Inheriting takes everything from the connection, so nothing here applies.
+ wlUseSystemVars.setEnabled(useCredentials);
+ wUseSystemVars.setEnabled(useCredentials);
+ wlAwsIamRole.setEnabled(useIamRole);
+ wAwsIamRole.setEnabled(useIamRole);
+
+ toggleKeysSelection();
}
public void toggleKeysSelection() {
- wlAccessKeyId.setEnabled(!wUseSystemVars.getSelection());
- wAccessKeyId.setEnabled(!wUseSystemVars.getSelection());
- wlSecretAccessKey.setEnabled(!wUseSystemVars.getSelection());
- wSecretAccessKey.setEnabled(!wUseSystemVars.getSelection());
+ boolean useKeys =
+ AWS_CREDENTIALS.equals(wAwsAuthType.getText()) && !wUseSystemVars.getSelection();
+ // (inherit and IAM role both leave the key fields off)
+ wlAccessKeyId.setEnabled(useKeys);
+ wAccessKeyId.setEnabled(useKeys);
+ wlSecretAccessKey.setEnabled(useKeys);
+ wSecretAccessKey.setEnabled(useKeys);
+ }
+
+ private void toggleStreamToS3Flags() {
+ wLoadFromExistingFileFormat.setEnabled(!wStreamToS3Csv.getSelection());
}
}
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderMeta.java b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderMeta.java
index 524a4b93cf..b68e56a64b 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderMeta.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderMeta.java
@@ -55,7 +55,9 @@ import org.apache.hop.pipeline.transform.TransformMeta;
categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Bulk",
documentationUrl = "/pipeline/transforms/redshift-bulkloader.html",
isIncludeJdbcDrivers = true,
- classLoaderGroup = "redshift",
+ // Same group as the Redshift dialect: the loader reads its connection settings directly,
+ // and classes only match across plugins when they load in the same classloader.
+ classLoaderGroup = "redshift-db",
actionTransformTypes = {ActionTransformType.OUTPUT, ActionTransformType.RDBMS})
@RelationalLineage(operation = RelationalIoOperation.WRITE)
public class RedshiftBulkLoaderMeta
@@ -67,6 +69,11 @@ public class RedshiftBulkLoaderMeta
public static final String CSV_ESCAPE_CHAR = "\"";
public static final String ENCLOSURE = "\"";
+ /** Formats an existing file on S3 can have. */
+ public static final String FILE_FORMAT_CSV = "CSV";
+
+ public static final String FILE_FORMAT_PARQUET = "Parquet";
+
@HopMetadataProperty(
key = "connection",
injectionKey = "CONNECTIONNAME",
@@ -88,6 +95,16 @@ public class RedshiftBulkLoaderMeta
hopMetadataPropertyType = HopMetadataPropertyType.RDBMS_TABLE)
private String tablename;
+ /**
+ * Take the AWS credentials from the Redshift connection rather than repeating them here. Kept as
+ * a separate flag so a transform saved before this existed keeps whatever it was set to.
+ */
+ @HopMetadataProperty(
+ key = "use_connection_credentials",
+ injectionKey = "USE_CONNECTION_CREDENTIALS",
+ injectionKeyDescription = "")
+ private boolean useConnectionCredentials;
+
@HopMetadataProperty(
key = "use_credentials",
injectionKey = "USE_CREDENTIALS",
@@ -109,7 +126,8 @@ public class RedshiftBulkLoaderMeta
@HopMetadataProperty(
key = "aws_secret_access_key",
injectionKey = "AWS_SECRET_ACCESS_KEY",
- injectionKeyDescription = "")
+ injectionKeyDescription = "",
+ password = true)
private String awsSecretAccessKey;
@HopMetadataProperty(
@@ -189,12 +207,6 @@ public class RedshiftBulkLoaderMeta
hopMetadataPropertyType = HopMetadataPropertyType.FIELD_LIST)
private List<RedshiftBulkLoaderField> fields;
- @HopMetadataProperty(
- injectionKey = "FIELDDATABASE",
- injectionKeyDescription = "RedshiftBulkLoader.Injection.FIELDDATABASE")
- /** Fields in the table to insert */
- private String[] fieldDatabase;
-
public RedshiftBulkLoaderMeta() {
super(); // allocate BaseTransformMeta
@@ -399,6 +411,14 @@ public class RedshiftBulkLoaderMeta
return specifyFields;
}
+ public boolean isUseConnectionCredentials() {
+ return useConnectionCredentials;
+ }
+
+ public void setUseConnectionCredentials(boolean useConnectionCredentials) {
+ this.useConnectionCredentials = useConnectionCredentials;
+ }
+
public boolean isUseCredentials() {
return useCredentials;
}
@@ -568,10 +588,10 @@ public class RedshiftBulkLoaderMeta
}
} else {
// Specifying the column names explicitly
- for (int i = 0; i < getFieldDatabase().length; i++) {
- int idx = r.indexOfValue(getFieldDatabase()[i]);
+ for (RedshiftBulkLoaderField vbf : fields) {
+ int idx = r.indexOfValue(vbf.getDatabaseField());
if (idx < 0) {
- error_message += "\t\t" + getFieldDatabase()[i] + Const.CR;
+ error_message += "\t\t" + vbf.getDatabaseField() + Const.CR;
error_found = true;
}
}
@@ -600,7 +620,7 @@ public class RedshiftBulkLoaderMeta
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) {
@@ -874,20 +894,6 @@ public class RedshiftBulkLoaderMeta
}
}
- /**
- * @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;
- }
-
/**
* @return the schemaName
*/
diff --git a/plugins/tech/aws/src/main/resources/org/apache/hop/pipeline/transforms/redshift/bulkloader/messages/messages_en_US.properties b/plugins/tech/aws/src/main/resources/org/apache/hop/pipeline/transforms/redshift/bulkloader/messages/messages_en_US.properties
index 3673918380..1402790acf 100644
--- a/plugins/tech/aws/src/main/resources/org/apache/hop/pipeline/transforms/redshift/bulkloader/messages/messages_en_US.properties
+++ b/plugins/tech/aws/src/main/resources/org/apache/hop/pipeline/transforms/redshift/bulkloader/messages/messages_en_US.properties
@@ -127,6 +127,13 @@ RedshiftBulkLoaderMeta.CheckResult.TableOk=Table [{0}] is readeable and we got t
RedshiftBulkLoaderMeta.CheckResult.UndefinedError=An error occurred\: {0}
RedshiftBulkLoaderMeta.Error.ErrorConnecting=I was unable to connect to the database to verify the status of the table\: {0}
RedshiftBulkLoaderMeta.Error.NoConnection=There is no connection defined in this transform.
+RedshiftBulkLoaderMeta.Error.CannotCreateParentFolder=Unable to create the folder for the file ''{0}''.
+RedshiftBulkLoader.Log.ParentFolderCreated=Created the parent folder {0}
+RedshiftBulkLoaderMeta.Error.NotARedshiftConnection=The credentials cannot be inherited because the selected connection is not a Redshift connection.
+RedshiftBulkLoaderMeta.Error.ConnectionHasNoAwsCredentials=The selected Redshift connection signs in with a database user and password, so it has no AWS credentials to inherit. Set the connection to one of the IAM options, or specify credentials on this transform.
+RedshiftBulkLoaderMeta.Error.CredentialsNotResolved=Unable to resolve the AWS credentials configured on the Redshift connection.
+RedshiftBulkLoaderMeta.Error.NoCopyFromFilename=There is no S3 file name defined in this transform. The COPY statement needs a file to load from.
+RedshiftBulkLoaderMeta.Error.NoFileFormat=There is no file format defined in this transform. Pick a format for the existing file, or enable streaming to a CSV file on S3.
RedshiftBulkLoaderMeta.Error.NoInput=Not receiving any fields from previous transforms. Check the previous transforms for errors & the connecting hops.
RedshiftBulkLoaderMeta.Error.NoTable=No table is defined on this connection.
RedshiftBulkLoaderMeta.Exception.ConnectionNotDefined=Unable to determine the required fields because the database connection wasn''t defined.
diff --git a/plugins/tech/aws/src/test/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderTest.java b/plugins/tech/aws/src/test/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderTest.java
new file mode 100644
index 0000000000..5876d8f81b
--- /dev/null
+++ b/plugins/tech/aws/src/test/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderTest.java
@@ -0,0 +1,602 @@
+/*
+ * 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.redshift.bulkloader;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+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 com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.List;
+import java.util.TimeZone;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.database.Database;
+import org.apache.hop.core.database.DatabaseMeta;
+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.ValueMetaDate;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaJson;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.row.value.ValueMetaTimestamp;
+import org.apache.hop.databases.redshift.RedshiftAuthenticationType;
+import org.apache.hop.databases.redshift.RedshiftDatabaseMeta;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Unit tests around the COPY statement and the CSV the Redshift bulk loader streams to S3.
+ *
+ * @see <a href="https://github.com/apache/hop/issues/3621">issue 3621</a>
+ */
+class RedshiftBulkLoaderTest {
+
+ private TransformMockHelper<RedshiftBulkLoaderMeta, RedshiftBulkLoaderData> helper;
+ private RedshiftBulkLoaderMeta meta;
+ private RedshiftBulkLoaderData data;
+ private RedshiftBulkLoader transform;
+ private DatabaseMeta databaseMeta;
+
+ @BeforeAll
+ static void initHop() throws Exception {
+ // Inherited secrets go through Encr, which needs the password encoder plugin.
+ HopClientEnvironment.init();
+ }
+
+ @BeforeEach
+ void setUp() throws Exception {
+ helper =
+ new TransformMockHelper<>(
+ "Redshift bulk loader", RedshiftBulkLoaderMeta.class, RedshiftBulkLoaderData.class);
+ when(helper.logChannelFactory.create(any(), any())).thenReturn(helper.iLogChannel);
+ when(helper.logChannelFactory.create(any())).thenReturn(helper.iLogChannel);
+
+ meta = new RedshiftBulkLoaderMeta();
+ meta.setConnection("redshift");
+ meta.setSchemaName("staging");
+ meta.setTablename("orders");
+ meta.setCopyFromFilename("s3://bucket/orders.csv");
+
+ data = new RedshiftBulkLoaderData();
+
+ databaseMeta = mock(DatabaseMeta.class);
+ when(databaseMeta.getQuotedSchemaTableCombination(any(), any(), any()))
+ .thenReturn("\"staging\".\"orders\"");
+ data.databaseMeta = databaseMeta;
+
+ data.db = mock(Database.class);
+ when(data.db.getDatabaseMeta()).thenReturn(databaseMeta);
+ when(data.db.resolve(anyString())).thenAnswer(invocation -> invocation.getArgument(0));
+
+ transform =
+ new RedshiftBulkLoader(
+ helper.transformMeta, meta, data, 0, helper.pipelineMeta, helper.pipeline);
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ helper.cleanUp();
+ }
+
+ /**
+ * The reported NullPointerException: with "stream to S3" off nothing ever set a file format, and
+ * the statement builder called equals() straight on it.
+ */
+ @Test
+ void buildsTheCopyStatementWhenNoFileFormatWasEverPicked() throws Exception {
+ meta.setStreamToS3Csv(false);
+ meta.setUseCredentials(true);
+ meta.setAwsAccessKeyId("key");
+ meta.setAwsSecretAccessKey("secret");
+
+ String sql = assertDoesNotThrow(() -> transform.buildCopyStatementSqlString(false));
+
+ assertTrue(sql.startsWith("COPY \"staging\".\"orders\" FROM 's3://bucket/orders.csv'"), sql);
+ // Without a format there are no CSV options to add.
+ assertFalse(sql.contains("DELIMITER"), sql);
+ }
+
+ /** The same guard has to hold when the transform is asked for its configuration up front. */
+ @Test
+ void refusesToStartWithoutAFileFormatForAnExistingFile() throws Exception {
+ meta.setStreamToS3Csv(false);
+
+ HopException e = assertThrows(HopException.class, () -> transform.verifyFileSettings());
+ assertTrue(e.getMessage().contains("file format"), e.getMessage());
+ }
+
+ /** And a COPY statement without a file to read from is never going to work either. */
+ @Test
+ void refusesToStartWithoutAFileName() throws Exception {
+ meta.setCopyFromFilename(null);
+
+ assertThrows(HopException.class, () -> transform.verifyFileSettings());
+ }
+
+ /** Variables in the credentials have to reach the COPY statement resolved. */
+ @Test
+ void resolvesVariablesInTheCredentials() throws Exception {
+ transform.setVariable("AWS_KEY", "AKIAEXAMPLE");
+ transform.setVariable("AWS_SECRET", "s3cr3t");
+ meta.setUseCredentials(true);
+ meta.setAwsAccessKeyId("${AWS_KEY}");
+ meta.setAwsSecretAccessKey("${AWS_SECRET}");
+
+ String sql = transform.buildCopyStatementSqlString(false);
+
+ assertTrue(
+ sql.contains("CREDENTIALS 'aws_access_key_id=AKIAEXAMPLE;aws_secret_access_key=s3cr3t'"),
+ sql);
+ }
+
+ /** The statement ends up in the log on debug level, the credentials should not. */
+ @Test
+ void masksTheCredentialsInTheLoggedStatement() throws Exception {
+ meta.setUseCredentials(true);
+ meta.setAwsAccessKeyId("AKIAEXAMPLE");
+ meta.setAwsSecretAccessKey("s3cr3t");
+
+ String logged = transform.buildCopyStatementSqlString(true);
+
+ assertFalse(logged.contains("AKIAEXAMPLE"), logged);
+ assertFalse(logged.contains("s3cr3t"), logged);
+ }
+
+ /** An IAM role is a secret too. */
+ @Test
+ void masksTheIamRoleInTheLoggedStatement() throws Exception {
+ meta.setUseAwsIamRole(true);
+ meta.setAwsIamRole("arn:aws:iam::123456789012:role/loader");
+
+ assertTrue(transform.buildCopyStatementSqlString(false).contains("role/loader"));
+ assertFalse(transform.buildCopyStatementSqlString(true).contains("role/loader"));
+ }
+
+ /** An empty column list is not valid SQL, so it must never be emitted. */
+ @Test
+ void neverEmitsAnEmptyColumnList() throws Exception {
+ meta.setStreamToS3Csv(true);
+
+ assertFalse(transform.buildCopyStatementSqlString(false).contains("()"));
+ }
+
+ @Test
+ void namesTheColumnsTheMappingPicked() throws Exception {
+ meta.setStreamToS3Csv(true);
+ meta.setSpecifyFields(true);
+ meta.setFields(
+ List.of(
+ new RedshiftBulkLoaderField("order_id", "id"),
+ new RedshiftBulkLoaderField("order_date", "date")));
+
+ assertTrue(
+ transform.buildCopyStatementSqlString(false).contains("(order_id, order_date)"),
+ transform.buildCopyStatementSqlString(false));
+ }
+
+ /**
+ * Without a mapping the file still holds only the fields of the stream. Saying nothing makes
+ * Redshift expect a value for every column of the table and fail the load with "Delimiter not
+ * found" as soon as the table is wider than the stream.
+ */
+ @Test
+ void namesTheStreamFieldsWhenThereIsNoMapping() throws Exception {
+ meta.setStreamToS3Csv(true);
+ meta.setSpecifyFields(false);
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("id"));
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+ transform.setInputRowMeta(rowMeta);
+ transform.prepareRowMapping();
+
+ assertTrue(
+ transform.buildCopyStatementSqlString(false).contains("(id, name)"),
+ transform.buildCopyStatementSqlString(false));
+ }
+
+ /**
+ * A file somebody else wrote is the one case we cannot speak for: with no mapping there is
+ * nothing to go on, so the statement stays silent and Redshift matches it positionally.
+ */
+ @Test
+ void namesNoColumnsForAnExistingFileWithoutAMapping() throws Exception {
+ meta.setStreamToS3Csv(false);
+ meta.setLoadFromExistingFileFormat(RedshiftBulkLoaderMeta.FILE_FORMAT_CSV);
+ meta.setSpecifyFields(false);
+
+ String sql = transform.buildCopyStatementSqlString(false);
+
+ assertTrue(sql.startsWith("COPY \"staging\".\"orders\" FROM"), sql);
+ }
+
+ /** Parquet files are loaded as is, without any of the CSV options. */
+ @Test
+ void addsTheParquetFormatForParquetFiles() throws Exception {
+ meta.setStreamToS3Csv(false);
+ meta.setLoadFromExistingFileFormat(RedshiftBulkLoaderMeta.FILE_FORMAT_PARQUET);
+ meta.setCopyFromFilename("s3://bucket/orders.parquet");
+
+ String sql = transform.buildCopyStatementSqlString(false);
+
+ assertTrue(sql.endsWith(" FORMAT AS PARQUET;"), sql);
+ assertFalse(sql.contains("DELIMITER"), sql);
+ }
+
+ @Test
+ void writesEveryFieldOfTheRowWhenNoFieldsAreSpecified() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+ rowMeta.addValueMeta(new ValueMetaInteger("amount"));
+
+ assertEquals("\"Acme, Inc\",42\n", writeRow(rowMeta, new Object[] {"Acme, Inc", 42L}, false));
+ }
+
+ /** Quotes inside a value are doubled, the way the COPY statement expects them. */
+ @Test
+ void doublesTheQuotesInsideAValue() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+
+ assertEquals("\"a\"\"b\"\"\"\n", writeRow(rowMeta, new Object[] {"a\"b\""}, false));
+ }
+
+ /** The selected fields are written in the order the COPY statement names them. */
+ @Test
+ void writesTheSelectedFieldsInTheConfiguredOrder() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(
+ List.of(
+ new RedshiftBulkLoaderField("order_amount", "amount"),
+ new RedshiftBulkLoaderField("order_name", "name")));
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+ rowMeta.addValueMeta(new ValueMetaInteger("amount"));
+
+ assertEquals("42,\"Acme\"\n", writeRow(rowMeta, new Object[] {"Acme", 42L}, true));
+ }
+
+ /**
+ * Dates are written as ISO 8601 with milliseconds, which is what DATEFORMAT/TIMEFORMAT 'auto'
+ * reads. A Hop Date carries a time of day, so writing it whole lets a TIMESTAMP column keep it; a
+ * DATE column truncates it on Redshift's side.
+ */
+ @Test
+ void writesDatesAsIsoTimestampsSoNoTimeOfDayIsLost() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new RedshiftBulkLoaderField("order_date", "date")));
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaDate("date"));
+
+ assertEquals("2026-08-24 15:30:45.123\n", writeRow(rowMeta, new Object[] {dateTime()}, true));
+ }
+
+ /** Timestamps get the same treatment, fractional seconds included. */
+ @Test
+ void writesTimestampsAsIsoWithMilliseconds() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new RedshiftBulkLoaderField("order_ts", "ts")));
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaTimestamp("ts"));
+
+ assertEquals(
+ "2026-08-24 15:30:45.123\n",
+ writeRow(rowMeta, new Object[] {new Timestamp(dateTime().getTime())}, true));
+ }
+
+ /**
+ * The regression that made this visible: with "specify database fields" off the transform used a
+ * separate write path that never converted dates at all, so they went out in whatever format the
+ * incoming field happened to carry.
+ */
+ @Test
+ void convertsDatesEvenWhenNoFieldsAreSpecified() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+ rowMeta.addValueMeta(new ValueMetaDate("date"));
+
+ assertEquals(
+ "\"Acme\",2026-08-24 15:30:45.123\n",
+ writeRow(rowMeta, new Object[] {"Acme", dateTime()}, false));
+ }
+
+ /** The COPY statement has to declare the format the rows are actually written in. */
+ @Test
+ void declaresTheAutoDateFormatsTheRowsAreWrittenIn() throws Exception {
+ meta.setStreamToS3Csv(true);
+
+ String sql = transform.buildCopyStatementSqlString(false);
+
+ assertTrue(sql.contains("DATEFORMAT AS 'auto'"), sql);
+ assertTrue(sql.contains("TIMEFORMAT AS 'auto'"), sql);
+ }
+
+ private java.util.Date dateTime() {
+ Calendar calendar = Calendar.getInstance(TimeZone.getDefault());
+ calendar.clear();
+ calendar.set(2026, Calendar.AUGUST, 24, 15, 30, 45);
+ calendar.set(Calendar.MILLISECOND, 123);
+ return calendar.getTime();
+ }
+
+ /** A field that is not on the stream is written as a null value, not as a crash. */
+ @Test
+ void writesAnEmptyValueForAFieldThatIsNotOnTheStream() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(
+ List.of(
+ new RedshiftBulkLoaderField("order_name", "name"),
+ new RedshiftBulkLoaderField("order_note", "note")));
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+
+ assertEquals("\"Acme\",\n", writeRow(rowMeta, new Object[] {"Acme"}, true));
+ }
+
+ /** Unless the user asked for that to be an error. */
+ @Test
+ void failsOnAMissingStreamFieldWhenErrorColumnMismatchIsSet() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setErrorColumnMismatch(true);
+ meta.setFields(List.of(new RedshiftBulkLoaderField("order_note", "note")));
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("name"));
+ transform.setInputRowMeta(rowMeta);
+
+ assertThrows(HopException.class, () -> transform.prepareRowMapping());
+ }
+
+ /** A column that does not exist in the table has to be caught before the file is written. */
+ @Test
+ void refusesToStartWhenAColumnIsNotInTheTable() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new RedshiftBulkLoaderField("order_totl", "amount")));
+ data.dbFields = new ArrayList<>();
+ data.dbFields.add(new String[] {"ORDER_ID", "INTEGER"});
+
+ HopException e = assertThrows(HopException.class, () -> transform.verifyTableFields());
+ assertTrue(e.getMessage().contains("order_totl"), e.getMessage());
+ }
+
+ @Test
+ void acceptsAColumnThatDiffersOnlyInCase() throws Exception {
+ meta.setSpecifyFields(true);
+ meta.setFields(List.of(new RedshiftBulkLoaderField("Order_Id", "id")));
+ data.dbFields = new ArrayList<>();
+ data.dbFields.add(new String[] {"ORDER_ID", "INTEGER"});
+
+ assertDoesNotThrow(() -> transform.verifyTableFields());
+ }
+
+ /** Streaming an empty stream to S3 leaves nothing to load, so the COPY is skipped. */
+ @Test
+ void skipsTheCopyWhenNothingWasStreamedToS3() throws Exception {
+ meta.setStreamToS3Csv(true);
+ data.rowsReceived = false;
+
+ assertFalse(transform.shouldExecuteCopy());
+
+ data.rowsReceived = true;
+ assertTrue(transform.shouldExecuteCopy());
+ }
+
+ /**
+ * A file that is already on S3 is loaded even when the stream that triggers the transform is
+ * empty, unless the user asked for the load to depend on the rows.
+ */
+ @Test
+ void loadsAnExistingFileEvenOnAnEmptyStream() throws Exception {
+ meta.setStreamToS3Csv(false);
+ meta.setLoadFromExistingFileFormat(RedshiftBulkLoaderMeta.FILE_FORMAT_CSV);
+ data.rowsReceived = false;
+
+ assertTrue(transform.shouldExecuteCopy());
+
+ meta.setOnlyWhenHaveRows(true);
+ assertFalse(transform.shouldExecuteCopy());
+ }
+
+ /**
+ * A JSON value is not a string as far as Hop is concerned, so it used to be written bare. Its
+ * commas then split it into extra CSV columns and Redshift saw only a fragment, reporting
+ * "End-of-input inside object or array". Hop pretty prints JSON by default, so there are line
+ * breaks in there too.
+ */
+ @Test
+ void enclosesJsonSoItSurvivesTheCsvRow() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaJson("config"));
+ JsonNode json =
+ new ObjectMapper()
+ .readTree(
+ "{\"metadataBaseFolder\":\"${PROJECT_HOME}/metadata\","
+ + "\"enforcingExecutionInHome\":true,"
+ + "\"config\":{\"variables\":[]}}");
+
+ String written = writeRow(rowMeta, new Object[] {json}, false);
+
+ // One enclosed field: quoted at both ends, with every inner quote doubled.
+ assertTrue(written.startsWith("\""), written);
+ assertTrue(written.endsWith("\"\n"), written);
+ assertTrue(written.contains("\"\"metadataBaseFolder\"\""), written);
+ // The pretty printed line breaks have to sit inside the quotes, not end the row early.
+ assertEquals(1, written.split("\n").length - countNewlinesInsideQuotes(written), written);
+ }
+
+ /** Numbers must stay bare, so the common case pays nothing for the fix above. */
+ @Test
+ void doesNotEncloseValuesThatCannotBreakTheRow() throws Exception {
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("amount"));
+
+ assertEquals("42\n", writeRow(rowMeta, new Object[] {42L}, false));
+ }
+
+ private int countNewlinesInsideQuotes(String written) {
+ // everything up to the final closing quote is one field
+ int lastQuote = written.lastIndexOf('"');
+ return (int) written.substring(0, lastQuote).chars().filter(c -> c == '\n').count();
+ }
+
+ /**
+ * The staging file is the transform's own business, so a folder that is not there yet gets
+ * created rather than refused. On S3 this is the common case: a prefix only exists while an
+ * object sits under it, so any brand new path failed with "Parent directory does not exist".
+ */
+ @Test
+ void createsTheStagingFolderWhenItDoesNotExistYet(@TempDir Path tempDir) throws Exception {
+ Path missing = tempDir.resolve("not-there-yet/nested");
+ assertFalse(Files.exists(missing));
+
+ transform.ensureParentFolderExists(missing.resolve("stage.csv").toString());
+
+ assertTrue(Files.isDirectory(missing), "the staging folder should have been created");
+ }
+
+ /** An existing folder is left exactly as it is. */
+ @Test
+ void leavesAnExistingStagingFolderAlone(@TempDir Path tempDir) throws Exception {
+ Path existing = Files.createDirectories(tempDir.resolve("already-there"));
+
+ transform.ensureParentFolderExists(existing.resolve("stage.csv").toString());
+
+ assertTrue(Files.isDirectory(existing));
+ }
+
+ // ------------------------------------------------------------ inherited credentials
+
+ /** An access key on the connection is handed straight to the COPY statement. */
+ @Test
+ void inheritsAnAccessKeyFromTheConnection() throws Exception {
+ RedshiftDatabaseMeta connection = new RedshiftDatabaseMeta();
+ connection.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ connection.setAwsAccessKeyId("AKIAFROMCONNECTION");
+ connection.setAwsSecretAccessKey("connection-secret");
+ useConnection(connection);
+ meta.setUseConnectionCredentials(true);
+
+ String sql = assertDoesNotThrow(() -> transform.buildCopyStatementSqlString(false));
+
+ assertTrue(
+ sql.contains(
+ "CREDENTIALS 'aws_access_key_id=AKIAFROMCONNECTION"
+ + ";aws_secret_access_key=connection-secret'"),
+ sql);
+ }
+
+ /** Temporary credentials need their session token to travel with them. */
+ @Test
+ void inheritsASessionTokenWhenTheConnectionHasOne() throws Exception {
+ RedshiftDatabaseMeta connection = new RedshiftDatabaseMeta();
+ connection.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ connection.setAwsAccessKeyId("AKIAFROMCONNECTION");
+ connection.setAwsSecretAccessKey("connection-secret");
+ connection.setAwsSessionToken("temporary-token");
+ useConnection(connection);
+ meta.setUseConnectionCredentials(true);
+
+ assertTrue(
+ assertDoesNotThrow(() -> transform.buildCopyStatementSqlString(false))
+ .contains(";token=temporary-token'"));
+ }
+
+ /** Variables on the connection are resolved with the transform's own variables. */
+ @Test
+ void resolvesVariablesInTheInheritedCredentials() throws Exception {
+ transform.setVariable("CONN_KEY", "AKIARESOLVED");
+ RedshiftDatabaseMeta connection = new RedshiftDatabaseMeta();
+ connection.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ connection.setAwsAccessKeyId("${CONN_KEY}");
+ connection.setAwsSecretAccessKey("connection-secret");
+ useConnection(connection);
+ meta.setUseConnectionCredentials(true);
+
+ assertTrue(
+ assertDoesNotThrow(() -> transform.buildCopyStatementSqlString(false))
+ .contains("aws_access_key_id=AKIARESOLVED"));
+ }
+
+ /** A connection signing in with a database password has nothing to lend. */
+ @Test
+ void refusesToInheritFromAPasswordOnlyConnection() throws Exception {
+ RedshiftDatabaseMeta connection = new RedshiftDatabaseMeta();
+ connection.setAuthenticationType(RedshiftAuthenticationType.DATABASE);
+ useConnection(connection);
+ meta.setUseConnectionCredentials(true);
+
+ assertThrows(HopException.class, () -> transform.buildCopyStatementSqlString(false));
+ }
+
+ /** Inherited credentials are secrets like any other, so they stay out of the log. */
+ @Test
+ void masksInheritedCredentialsInTheLoggedStatement() throws Exception {
+ RedshiftDatabaseMeta connection = new RedshiftDatabaseMeta();
+ connection.setAuthenticationType(RedshiftAuthenticationType.IAM_CREDENTIALS);
+ connection.setAwsAccessKeyId("AKIAFROMCONNECTION");
+ connection.setAwsSecretAccessKey("connection-secret");
+ useConnection(connection);
+ meta.setUseConnectionCredentials(true);
+
+ String logged = assertDoesNotThrow(() -> transform.buildCopyStatementSqlString(true));
+
+ assertFalse(logged.contains("AKIAFROMCONNECTION"), logged);
+ assertFalse(logged.contains("connection-secret"), logged);
+ }
+
+ private void useConnection(RedshiftDatabaseMeta connection) {
+ when(databaseMeta.getIDatabase()).thenReturn(connection);
+ }
+
+ private String writeRow(IRowMeta rowMeta, Object[] row, boolean specifyFields) throws Exception {
+ meta.setStreamToS3Csv(true);
+ meta.setSpecifyFields(specifyFields);
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ data.writer = out;
+ transform.initBinaryDataFields();
+ transform.setInputRowMeta(rowMeta);
+ transform.prepareRowMapping();
+ transform.writeRowToFile(data.outputRowMeta, row);
+
+ return out.toString(StandardCharsets.UTF_8);
+ }
+}