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 f21de78903 Issue #3245 : Add truncate table option to Snowflake Bulk
Loader (#7701)
f21de78903 is described below
commit f21de78903beb51d495f9394e6ce6973abfbd965
Author: Matt Casters <[email protected]>
AuthorDate: Fri Jul 31 09:04:11 2026 +0200
Issue #3245 : Add truncate table option to Snowflake Bulk Loader (#7701)
Allow clearing the target table before bulk load, with an optional
mode that only truncates when input rows are received (same pattern as
Table Output). Includes dialog, metadata, docs, and unit tests.
---
.../pipeline/transforms/snowflakebulkloader.adoc | 2 +
.../snowflake/bulkloader/SnowflakeBulkLoader.java | 45 +++--
.../bulkloader/SnowflakeBulkLoaderDialog.java | 80 ++++++++-
.../bulkloader/SnowflakeBulkLoaderMeta.java | 52 ++++++
.../bulkloader/messages/messages_en_US.properties | 6 +
.../bulkloader/SnowflakeBulkLoaderMetaTest.java | 57 ++++++
.../bulkloader/SnowflakeBulkLoaderTest.java | 191 +++++++++++++++++++++
7 files changed, 420 insertions(+), 13 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/snowflakebulkloader.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/snowflakebulkloader.adoc
index ca8c1404c4..ceec39aa6a 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/snowflakebulkloader.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/snowflakebulkloader.adoc
@@ -51,6 +51,8 @@ The Snowflake bulk loader transform utilizes the Snowflake
Copy command to load
- **Connection**: The database connection to use when bulk loading
- **Schema**: (Optional) The schema containing the table being loaded.
- **Table name**: The name of the table being loaded.
+- **Truncate table**: (Y/N) Empty the target table before loading. This is the
master switch for truncate.
+- **Only when rows are received**: (Y/N) Only used when **Truncate table** is
checked. If checked, truncate only when at least one input row arrives. If
unchecked, always truncate (including when the input is empty).
- **Staging location type**: The type of Snowflake stage to use to store the
files.
* **User Location**: Uses the user's home directory to store the files being
loaded.
* **Table Location**: Uses the table's internal stage to store the files being
loaded.
diff --git
a/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoader.java
b/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoader.java
index 250676674b..5c68bab04f 100644
---
a/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoader.java
+++
b/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoader.java
@@ -41,6 +41,7 @@ import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.value.ValueMetaBigNumber;
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.variables.IVariables;
import org.apache.hop.core.vfs.HopVfs;
import org.apache.hop.i18n.BaseMessages;
@@ -79,8 +80,27 @@ public class SnowflakeBulkLoader
Object[] row = getRow(); // This also waits for a row to be finished.
- if (row != null && first) {
+ if (row == null) {
+ // no more input to be expected...
+ if (first && meta.isTruncateTable() && !meta.isOnlyWhenHaveRows()) {
+ truncateTable();
+ }
+ // Only close/load when files were written; empty input has nothing to
bulk load
+ if (data.oneFileOpened) {
+ closeFile();
+ loadDatabase();
+ }
+ setOutputDone();
+ return false;
+ }
+
+ if (first) {
first = false;
+
+ if (meta.isTruncateTable()) {
+ truncateTable();
+ }
+
data.outputRowMeta = getInputRowMeta().clone();
meta.getFields(data.outputRowMeta, getTransformName(), null, null, this,
metadataProvider);
@@ -142,8 +162,7 @@ public class SnowflakeBulkLoader
}
// Create a new split?
- if ((row != null
- && data.outputCount > 0
+ if ((data.outputCount > 0
&& Const.toIntExpanded(resolve(meta.getSplitSize()), 0) > 0
&& (data.outputCount %
Const.toIntExpanded(resolve(meta.getSplitSize()), 0)) == 0)) {
@@ -154,14 +173,6 @@ public class SnowflakeBulkLoader
openNewFile(buildFilename());
}
- if (row == null) {
- // no more input to be expected...
- closeFile();
- loadDatabase();
- setOutputDone();
- return false;
- }
-
writeRowToFile(data.outputRowMeta, row);
putRow(data.outputRowMeta, row); // in case we want it to go further...
@@ -174,6 +185,18 @@ public class SnowflakeBulkLoader
return true;
}
+ /**
+ * Truncates the target table when the option is enabled. Only the first
transform copy (or a
+ * partitioned copy) performs the truncate to avoid races with concurrent
copies.
+ *
+ * @throws HopDatabaseException if the truncate fails
+ */
+ void truncateTable() throws HopDatabaseException {
+ if (meta.isTruncateTable() && ((getCopy() == 0) ||
!Utils.isEmpty(getPartitionId()))) {
+ data.db.truncateTable(resolve(meta.getTargetSchema()),
resolve(meta.getTargetTable()));
+ }
+ }
+
/**
* Runs a desc table to get the fields, and field types from the database.
Uses a desc table as
* opposed to the select * from table limit 0 that Hop normally uses to get
the fields and types,
diff --git
a/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderDialog.java
b/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderDialog.java
index b47e7d58b7..c7470396de 100644
---
a/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderDialog.java
+++
b/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderDialog.java
@@ -121,6 +121,14 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
private TextVar wTable;
+ // Truncate table line
+ private Label wlTruncate;
+ private Button wTruncate;
+
+ // Truncate only when have rows (only meaningful when truncate is enabled)
+ private Label wlOnlyWhenHaveRows;
+ private Button wOnlyWhenHaveRows;
+
// Location Type line
private CCombo wLocationType;
@@ -386,6 +394,61 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
}
});
+ // Truncate table (master switch). "Truncate on first row" only applies
when this is enabled.
+ wlTruncate = new Label(wLoaderComp, SWT.RIGHT);
+ wlTruncate.setText(
+ BaseMessages.getString(PKG,
"SnowflakeBulkLoader.Dialog.TruncateTable.Label"));
+ wlTruncate.setToolTipText(
+ BaseMessages.getString(PKG,
"SnowflakeBulkLoader.Dialog.TruncateTable.Tooltip"));
+ PropsUi.setLook(wlTruncate);
+ FormData fdlTruncate = new FormData();
+ fdlTruncate.left = new FormAttachment(0, 0);
+ fdlTruncate.top = new FormAttachment(wTable, margin);
+ fdlTruncate.right = new FormAttachment(middle, -margin);
+ wlTruncate.setLayoutData(fdlTruncate);
+
+ wTruncate = new Button(wLoaderComp, SWT.CHECK);
+ wTruncate.setToolTipText(
+ BaseMessages.getString(PKG,
"SnowflakeBulkLoader.Dialog.TruncateTable.Tooltip"));
+ PropsUi.setLook(wTruncate);
+ FormData fdTruncate = new FormData();
+ fdTruncate.left = new FormAttachment(middle, 0);
+ fdTruncate.top = new FormAttachment(wlTruncate, 0, SWT.CENTER);
+ fdTruncate.right = new FormAttachment(100, 0);
+ wTruncate.setLayoutData(fdTruncate);
+ wTruncate.addSelectionListener(bMod);
+ wTruncate.addSelectionListener(
+ new SelectionAdapter() {
+ @Override
+ public void widgetSelected(SelectionEvent e) {
+ setFlags();
+ }
+ });
+
+ // Truncate only when have rows (qualifier for Truncate table)
+ wlOnlyWhenHaveRows = new Label(wLoaderComp, SWT.RIGHT);
+ wlOnlyWhenHaveRows.setText(
+ BaseMessages.getString(PKG,
"SnowflakeBulkLoader.Dialog.OnlyWhenHaveRows.Label"));
+ wlOnlyWhenHaveRows.setToolTipText(
+ BaseMessages.getString(PKG,
"SnowflakeBulkLoader.Dialog.OnlyWhenHaveRows.Tooltip"));
+ PropsUi.setLook(wlOnlyWhenHaveRows);
+ FormData fdlOnlyWhenHaveRows = new FormData();
+ fdlOnlyWhenHaveRows.left = new FormAttachment(0, 0);
+ fdlOnlyWhenHaveRows.top = new FormAttachment(wlTruncate, margin);
+ fdlOnlyWhenHaveRows.right = new FormAttachment(middle, -margin);
+ wlOnlyWhenHaveRows.setLayoutData(fdlOnlyWhenHaveRows);
+
+ wOnlyWhenHaveRows = new Button(wLoaderComp, SWT.CHECK);
+ wOnlyWhenHaveRows.setToolTipText(
+ BaseMessages.getString(PKG,
"SnowflakeBulkLoader.Dialog.OnlyWhenHaveRows.Tooltip"));
+ PropsUi.setLook(wOnlyWhenHaveRows);
+ FormData fdOnlyWhenHaveRows = new FormData();
+ fdOnlyWhenHaveRows.left = new FormAttachment(middle, 0);
+ fdOnlyWhenHaveRows.top = new FormAttachment(wlOnlyWhenHaveRows, 0,
SWT.CENTER);
+ fdOnlyWhenHaveRows.right = new FormAttachment(100, 0);
+ wOnlyWhenHaveRows.setLayoutData(fdOnlyWhenHaveRows);
+ wOnlyWhenHaveRows.addSelectionListener(bMod);
+
// Location Type line
//
Label wlLocationType = new Label(wLoaderComp, SWT.RIGHT);
@@ -396,7 +459,7 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
PropsUi.setLook(wlLocationType);
FormData fdlLocationType = new FormData();
fdlLocationType.left = new FormAttachment(0, 0);
- fdlLocationType.top = new FormAttachment(wTable, margin);
+ fdlLocationType.top = new FormAttachment(wlOnlyWhenHaveRows, margin);
fdlLocationType.right = new FormAttachment(middle, -margin);
wlLocationType.setLayoutData(fdlLocationType);
@@ -407,7 +470,7 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
wLocationType.addSelectionListener(lsFlags);
FormData fdLocationType = new FormData();
fdLocationType.left = new FormAttachment(middle, 0);
- fdLocationType.top = new FormAttachment(wTable, margin);
+ fdLocationType.top = new FormAttachment(wlOnlyWhenHaveRows, margin);
fdLocationType.right = new FormAttachment(100, 0);
wLocationType.setLayoutData(fdLocationType);
for (String locationType : LOCATION_TYPE_COMBO) {
@@ -1123,6 +1186,9 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
wTable.setText(input.getTargetTable());
}
+ wTruncate.setSelection(input.isTruncateTable());
+ wOnlyWhenHaveRows.setSelection(input.isOnlyWhenHaveRows());
+
if (input.getLocationType() != null) {
wLocationType.setText(LOCATION_TYPE_COMBO[input.getLocationTypeId()]);
}
@@ -1208,6 +1274,8 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
sbl.setConnection(wConnection.getText());
sbl.setTargetSchema(wSchema.getText());
sbl.setTargetTable(wTable.getText());
+ sbl.setTruncateTable(wTruncate.getSelection());
+ sbl.setOnlyWhenHaveRows(wOnlyWhenHaveRows.getSelection());
sbl.setLocationTypeById(wLocationType.getSelectionIndex());
sbl.setStageName(wStageName.getText());
sbl.setWorkDirectory(wWorkDirectory.getText());
@@ -1542,6 +1610,14 @@ public class SnowflakeBulkLoaderDialog extends
BaseTransformDialog {
/** Enable and disable fields based on selection changes */
private void setFlags() {
+ /////////////////////////////////
+ // Truncate
+ ////////////////////////////////
+ // Second option only applies when truncate is enabled (same as Table
Output)
+ boolean truncateEnabled = wTruncate.getSelection();
+ wlOnlyWhenHaveRows.setEnabled(truncateEnabled);
+ wOnlyWhenHaveRows.setEnabled(truncateEnabled);
+
/////////////////////////////////
// On Error
////////////////////////////////
diff --git
a/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMeta.java
b/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMeta.java
index 93db50b298..3c9904af9c 100644
---
a/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMeta.java
+++
b/plugins/databases/snowflake/src/main/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMeta.java
@@ -123,6 +123,24 @@ public class SnowflakeBulkLoaderMeta
hopMetadataPropertyType = HopMetadataPropertyType.RDBMS_TABLE)
private String targetTable;
+ /** Truncate the target table before loading */
+ @HopMetadataProperty(
+ key = "truncate",
+ injectionKey = "TRUNCATE_TABLE",
+ injectionKeyDescription =
"SnowflakeBulkLoader.Injection.TruncateTable.Field",
+ hopMetadataPropertyType = HopMetadataPropertyType.RDBMS_TRUNCATE)
+ private boolean truncateTable;
+
+ /**
+ * When truncate is enabled, only truncate if at least one input row is
received. When false,
+ * truncate even when the input stream is empty.
+ */
+ @HopMetadataProperty(
+ key = "only_when_have_rows",
+ injectionKey = "ONLY_WHEN_HAVE_ROWS",
+ injectionKeyDescription =
"SnowflakeBulkLoader.Injection.OnlyWhenHaveRows.Field")
+ private boolean onlyWhenHaveRows;
+
/** The location type (user, table, internal_stage) */
@HopMetadataProperty(key = "location_type", injectionKeyDescription = "")
private String locationType;
@@ -266,6 +284,38 @@ public class SnowflakeBulkLoaderMeta
this.targetTable = targetTable;
}
+ /**
+ * @return true if the target table should be truncated before loading
+ */
+ public boolean isTruncateTable() {
+ return truncateTable;
+ }
+
+ /**
+ * Set whether the target table should be truncated before loading
+ *
+ * @param truncateTable true/false
+ */
+ public void setTruncateTable(boolean truncateTable) {
+ this.truncateTable = truncateTable;
+ }
+
+ /**
+ * @return true if truncate should only run when input rows are received
+ */
+ public boolean isOnlyWhenHaveRows() {
+ return onlyWhenHaveRows;
+ }
+
+ /**
+ * Set whether truncate should only run when input rows are received
+ *
+ * @param onlyWhenHaveRows true/false
+ */
+ public void setOnlyWhenHaveRows(boolean onlyWhenHaveRows) {
+ this.onlyWhenHaveRows = onlyWhenHaveRows;
+ }
+
/**
* @return The location type code for the files to load
*/
@@ -719,6 +769,8 @@ public class SnowflakeBulkLoaderMeta
workDirectory = "${java.io.tmpdir}";
onError = ON_ERROR_CODES[ON_ERROR_ABORT];
removeFiles = true;
+ truncateTable = false;
+ onlyWhenHaveRows = false;
dataType = DATA_TYPE_CODES[DATA_TYPE_CSV];
trimWhitespace = false;
diff --git
a/plugins/databases/snowflake/src/main/resources/org/apache/hop/pipeline/transforms/snowflake/bulkloader/messages/messages_en_US.properties
b/plugins/databases/snowflake/src/main/resources/org/apache/hop/pipeline/transforms/snowflake/bulkloader/messages/messages_en_US.properties
index 481b9a298a..7049a78c54 100644
---
a/plugins/databases/snowflake/src/main/resources/org/apache/hop/pipeline/transforms/snowflake/bulkloader/messages/messages_en_US.properties
+++
b/plugins/databases/snowflake/src/main/resources/org/apache/hop/pipeline/transforms/snowflake/bulkloader/messages/messages_en_US.properties
@@ -70,8 +70,12 @@ SnowflakeBulkLoader.Dialog.NullIf.Label=Null if
SnowflakeBulkLoader.Dialog.NullIf.Tooltip=Comma delimited list of strings to
convert to null if\nthey are found in a field.
SnowflakeBulkLoader.Dialog.OnError.Label=On error
SnowflakeBulkLoader.Dialog.OnError.Tooltip=The action to take when an error
occurs during a bulk load.
+SnowflakeBulkLoader.Dialog.OnlyWhenHaveRows.Label=Only when rows are received
+SnowflakeBulkLoader.Dialog.OnlyWhenHaveRows.Tooltip=Only used when Truncate
table is checked.\nIf checked, truncate only when at least one input row is
received.\nIf unchecked, always truncate the table (even when the input is
empty).
SnowflakeBulkLoader.Dialog.RemoveFiles.Label=Remove files after load
SnowflakeBulkLoader.Dialog.RemoveFiles.Tooltip=If checked, the files will be
removed from the Snowflake\nstaging location after the load.
+SnowflakeBulkLoader.Dialog.TruncateTable.Label=Truncate table
+SnowflakeBulkLoader.Dialog.TruncateTable.Tooltip=Empty the target table before
loading.\nUse "Only when rows are received" to skip truncate when there is no
input.
SnowflakeBulkLoader.Dialog.Schema.Label=Schema
SnowflakeBulkLoader.Dialog.Schema.Tooltip=The schema containing the table you
wish to load.\nIf blank loads the default schema.
SnowflakeBulkLoader.Dialog.SpecifyFields.Label=Specifying Fields
@@ -107,8 +111,10 @@ SnowflakeBulkLoader.Injection.JSON_FIELD=The field
containing the JSON to load.
SnowflakeBulkLoader.Injection.LOCATION_TYPE=(user, table, internal_stage) The
Snowflake location type to store data being loaded.
SnowflakeBulkLoader.Injection.NULL_IF=Comma delimited list of field values
that should be converted to null, if CSV data type.
SnowflakeBulkLoader.Injection.ON_ERROR=The action to take when an error occurs
(continue, skip_file, skip_file_percent, abort)
+SnowflakeBulkLoader.Injection.OnlyWhenHaveRows.Field=Truncate only when have
rows? (Y/N)
SnowflakeBulkLoader.Injection.OUTPUT_FIELDS=CSV Output fields
SnowflakeBulkLoader.Injection.REMOVE_FILES=(Y/N) Remove files from Snowflake
stage after load.
+SnowflakeBulkLoader.Injection.TruncateTable.Field=Truncate table? (Y/N)
SnowflakeBulkLoader.Injection.SPECIFY_FIELDS=(Y/N) Is the field m
SnowflakeBulkLoader.Injection.SPLIT_SIZE=Split load files every ... rows
SnowflakeBulkLoader.Injection.STAGE_NAME=The name of the Snowflake internal
stage to use when loading.
diff --git
a/plugins/databases/snowflake/src/test/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMetaTest.java
b/plugins/databases/snowflake/src/test/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMetaTest.java
new file mode 100644
index 0000000000..81870e9bdd
--- /dev/null
+++
b/plugins/databases/snowflake/src/test/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderMetaTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.snowflake.bulkloader;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+class SnowflakeBulkLoaderMetaTest {
+
+ @Test
+ void testDefaultTruncateFlagsAreOff() {
+ SnowflakeBulkLoaderMeta meta = new SnowflakeBulkLoaderMeta();
+ meta.setDefault();
+
+ assertFalse(meta.isTruncateTable());
+ assertFalse(meta.isOnlyWhenHaveRows());
+ }
+
+ @Test
+ void testTruncateTableAccessors() {
+ SnowflakeBulkLoaderMeta meta = new SnowflakeBulkLoaderMeta();
+
+ meta.setTruncateTable(true);
+ assertTrue(meta.isTruncateTable());
+
+ meta.setTruncateTable(false);
+ assertFalse(meta.isTruncateTable());
+ }
+
+ @Test
+ void testOnlyWhenHaveRowsAccessors() {
+ SnowflakeBulkLoaderMeta meta = new SnowflakeBulkLoaderMeta();
+
+ meta.setOnlyWhenHaveRows(true);
+ assertTrue(meta.isOnlyWhenHaveRows());
+
+ meta.setOnlyWhenHaveRows(false);
+ assertFalse(meta.isOnlyWhenHaveRows());
+ }
+}
diff --git
a/plugins/databases/snowflake/src/test/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderTest.java
b/plugins/databases/snowflake/src/test/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderTest.java
new file mode 100644
index 0000000000..d4d8f13e37
--- /dev/null
+++
b/plugins/databases/snowflake/src/test/java/org/apache/hop/pipeline/transforms/snowflake/bulkloader/SnowflakeBulkLoaderTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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.snowflake.bulkloader;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.nullable;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.sql.Connection;
+import org.apache.hop.core.database.Database;
+import org.apache.hop.core.database.DatabaseMeta;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transform.TransformPartitioningMeta;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class SnowflakeBulkLoaderTest {
+
+ private SnowflakeBulkLoader bulkLoaderSpy;
+ private SnowflakeBulkLoaderMeta meta;
+ private SnowflakeBulkLoaderData data;
+ private Database db;
+ private TransformMeta transformMeta;
+ private PipelineMeta pipelineMeta;
+
+ @BeforeEach
+ void setUp() {
+ meta = mock(SnowflakeBulkLoaderMeta.class);
+ doReturn("TestConnection").when(meta).getConnection();
+ doReturn("myschema").when(meta).getTargetSchema();
+ doReturn("mytable").when(meta).getTargetTable();
+
+ transformMeta = mock(TransformMeta.class);
+ doReturn("transform").when(transformMeta).getName();
+ doReturn(mock(TransformPartitioningMeta.class))
+ .when(transformMeta)
+ .getTargetTransformPartitioningMeta();
+ doReturn(meta).when(transformMeta).getTransform();
+
+ db = mock(Database.class);
+ doReturn(mock(Connection.class)).when(db).getConnection();
+
+ data = new SnowflakeBulkLoaderData();
+ data.db = db;
+ data.databaseMeta = mock(DatabaseMeta.class);
+
+ pipelineMeta = mock(PipelineMeta.class);
+ doReturn(transformMeta).when(pipelineMeta).findTransform(anyString());
+
+ SnowflakeBulkLoader bulkLoader =
+ new SnowflakeBulkLoader(
+ transformMeta, meta, data, 0, pipelineMeta, spy(new
LocalPipelineEngine()));
+ bulkLoaderSpy = spy(bulkLoader);
+ doReturn(transformMeta).when(bulkLoaderSpy).getTransformMeta();
+ doReturn(false).when(bulkLoaderSpy).isRowLevel();
+ doReturn(false).when(bulkLoaderSpy).isDebug();
+ doReturn("myschema").when(bulkLoaderSpy).resolve("myschema");
+ doReturn("mytable").when(bulkLoaderSpy).resolve("mytable");
+ }
+
+ @Test
+ void testTruncateTableOff() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(false);
+
+ bulkLoaderSpy.truncateTable();
+
+ verify(db, never()).truncateTable(anyString(), anyString());
+ }
+
+ @Test
+ void testTruncateTableOnCopyZero() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ when(bulkLoaderSpy.getCopy()).thenReturn(0);
+
+ bulkLoaderSpy.truncateTable();
+
+ verify(db).truncateTable(nullable(String.class), nullable(String.class));
+ }
+
+ @Test
+ void testTruncateTableOnWithPartitionId() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ when(bulkLoaderSpy.getCopy()).thenReturn(1);
+ when(bulkLoaderSpy.getPartitionId()).thenReturn("partition id");
+
+ bulkLoaderSpy.truncateTable();
+
+ verify(db).truncateTable(nullable(String.class), nullable(String.class));
+ }
+
+ @Test
+ void testTruncateTableSkippedOnNonZeroCopyWithoutPartition() throws
Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ when(bulkLoaderSpy.getCopy()).thenReturn(1);
+ when(bulkLoaderSpy.getPartitionId()).thenReturn(null);
+
+ bulkLoaderSpy.truncateTable();
+
+ verify(db, never()).truncateTable(anyString(), anyString());
+ }
+
+ @Test
+ void testProcessRowTruncatesIfNoRowsAvailable() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ when(meta.isOnlyWhenHaveRows()).thenReturn(false);
+ doReturn(null).when(bulkLoaderSpy).getRow();
+
+ boolean result = bulkLoaderSpy.processRow();
+
+ assertFalse(result);
+ verify(bulkLoaderSpy).truncateTable();
+ }
+
+ @Test
+ void testProcessRowDoesNotTruncateIfNoRowsAndOnlyWhenHaveRows() throws
Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ when(meta.isOnlyWhenHaveRows()).thenReturn(true);
+ doReturn(null).when(bulkLoaderSpy).getRow();
+
+ boolean result = bulkLoaderSpy.processRow();
+
+ assertFalse(result);
+ verify(bulkLoaderSpy, never()).truncateTable();
+ }
+
+ @Test
+ void testProcessRowDoesNotTruncateIfNoRowsAndTruncateOff() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(false);
+ doReturn(null).when(bulkLoaderSpy).getRow();
+
+ boolean result = bulkLoaderSpy.processRow();
+
+ assertFalse(result);
+ verify(bulkLoaderSpy, never()).truncateTable();
+ }
+
+ @Test
+ void testProcessRowTruncatesOnFirstRow() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ Object[] row = new Object[] {};
+ doReturn(row).when(bulkLoaderSpy).getRow();
+
+ try {
+ bulkLoaderSpy.processRow();
+ } catch (Exception e) {
+ // Not fully wired for a complete row write; truncate should still have
been attempted.
+ }
+
+ verify(bulkLoaderSpy, times(1)).truncateTable();
+ }
+
+ @Test
+ void testProcessRowDoesNotTruncateOnSubsequentRows() throws Exception {
+ when(meta.isTruncateTable()).thenReturn(true);
+ Object[] row = new Object[] {};
+ doReturn(row).when(bulkLoaderSpy).getRow();
+ bulkLoaderSpy.first = false;
+
+ try {
+ bulkLoaderSpy.processRow();
+ } catch (Exception e) {
+ // Not fully wired for a complete row write.
+ }
+
+ verify(bulkLoaderSpy, never()).truncateTable();
+ }
+}
