This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 61a49ad9386 [improve](extension) improve kettle plugin and add some
testcase (#44324)
61a49ad9386 is described below
commit 61a49ad9386b2f3d6dd89363657d6dc06d768a13
Author: wudi <[email protected]>
AuthorDate: Wed Nov 20 15:13:30 2024 +0800
[improve](extension) improve kettle plugin and add some testcase (#44324)
### What problem does this PR solve?
improve kettle plugin and add some testcase
---
extension/kettle/impl/pom.xml | 6 +
.../steps/dorisstreamloader/DorisStreamLoader.java | 6 +
.../dorisstreamloader/DorisStreamLoaderMeta.java | 18 +-
.../load/DorisBatchStreamLoad.java | 10 +-
.../DorisBatchStreamLoadTest.java | 2 +-
.../DorisRecordSerializerTest.java | 82 +++++
.../DorisStreamLoaderMetaTest.java | 114 ++++++
.../dorisstreamloader/DorisStreamLoaderTest.java | 389 +++++++++++++++++++++
8 files changed, 623 insertions(+), 4 deletions(-)
diff --git a/extension/kettle/impl/pom.xml b/extension/kettle/impl/pom.xml
index 59b3f42934b..64eb91076dd 100644
--- a/extension/kettle/impl/pom.xml
+++ b/extension/kettle/impl/pom.xml
@@ -66,6 +66,12 @@ under the License.
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>8.0.26</version>
+ <scope>test</scope>
+ </dependency>
<!--log-->
<dependency>
diff --git
a/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoader.java
b/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoader.java
index 85dc097be4e..91b546baf21 100644
---
a/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoader.java
+++
b/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoader.java
@@ -17,6 +17,7 @@
package org.pentaho.di.trans.steps.dorisstreamloader;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang.StringUtils;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.ValueMetaInterface;
@@ -172,4 +173,9 @@ public class DorisStreamLoader extends BaseStep implements
StepInterface {
super.dispose( smi, sdi );
}
+
+ @VisibleForTesting
+ public DorisBatchStreamLoad getStreamLoad(){
+ return streamLoad;
+ }
}
diff --git
a/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMeta.java
b/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMeta.java
index 12d72d838af..5e111b1bf38 100644
---
a/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMeta.java
+++
b/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMeta.java
@@ -24,9 +24,11 @@ import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.exception.KettleXMLException;
+import org.pentaho.di.core.injection.AfterInjection;
import org.pentaho.di.core.injection.Injection;
import org.pentaho.di.core.injection.InjectionSupported;
import org.pentaho.di.core.row.RowMetaInterface;
+import org.pentaho.di.core.util.Utils;
import org.pentaho.di.core.variables.VariableSpace;
import org.pentaho.di.core.xml.XMLHandler;
import org.pentaho.di.i18n.BaseMessages;
@@ -341,7 +343,7 @@ public class DorisStreamLoaderMeta extends BaseStepMeta
implements StepMetaInter
this.deletable = deletable;
}
- public String[] getFieldTable() {
+ public String[] getFieldTable() {
return fieldTable;
}
@@ -379,4 +381,18 @@ public class DorisStreamLoaderMeta extends BaseStepMeta
implements StepMetaInter
", fieldStream=" + Arrays.toString(fieldStream) +
'}';
}
+
+ /**
+ * If we use injection we can have different arrays lengths.
+ * We need synchronize them for consistency behavior with UI
+ */
+ @AfterInjection
+ public void afterInjectionSynchronization() {
+ int nrFields = (fieldTable == null) ? -1 : fieldTable.length;
+ if (nrFields <= 0) {
+ return;
+ }
+ String[][] rtnStrings = Utils.normalizeArrays(nrFields, fieldStream);
+ fieldStream = rtnStrings[0];
+ }
}
diff --git
a/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/load/DorisBatchStreamLoad.java
b/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/load/DorisBatchStreamLoad.java
index a73725add00..226f5f61e73 100644
---
a/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/load/DorisBatchStreamLoad.java
+++
b/extension/kettle/impl/src/main/java/org/pentaho/di/trans/steps/dorisstreamloader/load/DorisBatchStreamLoad.java
@@ -259,6 +259,10 @@ public class DorisBatchStreamLoad implements Serializable {
}
private synchronized boolean flush(String bufferKey, boolean waitUtilDone)
{
+ if (bufferMap.isEmpty()) {
+ // bufferMap may have been flushed by other threads
+ return false;
+ }
if (null == bufferKey) {
boolean flush = false;
for (String key : bufferMap.keySet()) {
@@ -275,7 +279,7 @@ public class DorisBatchStreamLoad implements Serializable {
} else if (bufferMap.containsKey(bufferKey)) {
flushBuffer(bufferKey);
} else {
- throw new DorisRuntimeException("buffer not found for key: " +
bufferKey);
+ log.logDetailed("buffer not found for key: {}, may be already
flushed.", bufferKey);
}
if (waitUtilDone) {
waitAsyncLoadFinish();
@@ -311,7 +315,9 @@ public class DorisBatchStreamLoad implements Serializable {
}
private void waitAsyncLoadFinish() {
- for (int i = 0; i < FLUSH_QUEUE_SIZE + 1; i++) {
+ // Because the flush thread will drainTo once after polling is
completed
+ // if queue_size is 2, at least 4 empty queues must be consumed to
ensure that flush has been completed
+ for (int i = 0; i < FLUSH_QUEUE_SIZE * 2; i++) {
BatchRecordBuffer empty = new BatchRecordBuffer();
putRecordToFlushQueue(empty);
}
diff --git
a/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisBatchStreamLoadTest.java
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisBatchStreamLoadTest.java
index c8d79156c18..556d3737bf4 100644
---
a/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisBatchStreamLoadTest.java
+++
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisBatchStreamLoadTest.java
@@ -32,7 +32,7 @@ public class DorisBatchStreamLoadTest {
@Ignore
public void testStreamLoad() throws Exception {
DorisOptions options = DorisOptions.builder()
- .withFenodes("10.16.10.6:28737")
+ .withFenodes("127.0.0.1:8030")
.withDatabase("test")
.withTable("test_flink_c")
.withUsername("root")
diff --git
a/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisRecordSerializerTest.java
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisRecordSerializerTest.java
new file mode 100644
index 00000000000..19b3ba6463c
--- /dev/null
+++
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisRecordSerializerTest.java
@@ -0,0 +1,82 @@
+// 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.pentaho.di.trans.steps.dorisstreamloader;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.pentaho.di.core.logging.LogChannel;
+import org.pentaho.di.core.row.ValueMetaInterface;
+import org.pentaho.di.core.row.value.ValueMetaBigNumber;
+import org.pentaho.di.core.row.value.ValueMetaBinary;
+import org.pentaho.di.core.row.value.ValueMetaBoolean;
+import org.pentaho.di.core.row.value.ValueMetaDate;
+import org.pentaho.di.core.row.value.ValueMetaInteger;
+import org.pentaho.di.core.row.value.ValueMetaInternetAddress;
+import org.pentaho.di.core.row.value.ValueMetaNumber;
+import org.pentaho.di.core.row.value.ValueMetaString;
+import org.pentaho.di.core.row.value.ValueMetaTimestamp;
+import
org.pentaho.di.trans.steps.dorisstreamloader.serializer.DorisRecordSerializer;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
+import javax.mail.internet.InternetAddress;
+
+import static
org.pentaho.di.trans.steps.dorisstreamloader.load.LoadConstants.CSV;
+
+public class DorisRecordSerializerTest {
+
+ @Test
+ public void testSerialize() throws Exception {
+ ValueMetaInterface[] formatMeta = new ValueMetaInterface[]{
+ new ValueMetaBoolean("boolean"),
+ new ValueMetaInteger("integer"),
+ new ValueMetaNumber("number"),
+ new ValueMetaBigNumber("bignumber"),
+ new ValueMetaDate("date"),
+ new ValueMetaTimestamp("timestamp"),
+ new ValueMetaBinary("binary"),
+ new ValueMetaString("string"),
+ new ValueMetaInternetAddress("address"),
+ };
+
+ DorisRecordSerializer serializer = DorisRecordSerializer.builder()
+ .setType(CSV)
+ .setFieldNames(new String[]{"c_boolean", "c_integer", "c_number",
"c_bignumber", "c_date", "c_timestamp", "c_binary", "c_string",
"c_internetAddress"})
+ .setFormatMeta(formatMeta)
+ .setFieldDelimiter(",")
+ .setLogChannelInterface(new LogChannel())
+ .setDeletable(false)
+ .build();
+
+ SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
+ Object[] data = new Object[]{
+ true,
+ 10L,
+ 123.23,
+ new BigDecimal("123456789.1234"),
+ dateFormat.parse("2024-01-01"),
+ Timestamp.valueOf("2024-01-01 10:11:22.123"),
+ "binary",
+ "string",
+ new InternetAddress("127.0.0.1")};
+ String actual = serializer.buildCSVString(data, formatMeta.length);
+ String except = "true,10,123.23,123456789.1234,2024-01-01,2024-01-01
10:11:22.123,binary,string,127.0.0.1";
+ Assert.assertEquals(except, actual);
+ }
+}
diff --git
a/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMetaTest.java
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMetaTest.java
new file mode 100644
index 00000000000..4a9f0f36c30
--- /dev/null
+++
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderMetaTest.java
@@ -0,0 +1,114 @@
+// 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.pentaho.di.trans.steps.dorisstreamloader;
+
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.pentaho.di.core.exception.KettleException;
+import org.pentaho.di.junit.rules.RestorePDIEngineEnvironment;
+import org.pentaho.di.trans.steps.loadsave.LoadSaveTester;
+import org.pentaho.di.trans.steps.loadsave.validator.ArrayLoadSaveValidator;
+import org.pentaho.di.trans.steps.loadsave.validator.FieldLoadSaveValidator;
+import org.pentaho.di.trans.steps.loadsave.validator.IntLoadSaveValidator;
+import org.pentaho.di.trans.steps.loadsave.validator.LongLoadSaveValidator;
+import org.pentaho.di.trans.steps.loadsave.validator.StringLoadSaveValidator;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class DorisStreamLoaderMetaTest {
+
+ @ClassRule
+ public static RestorePDIEngineEnvironment env = new
RestorePDIEngineEnvironment();
+
+ @Test
+ public void testRoundTrip() throws KettleException {
+ List<String> attributes =
+ Arrays.asList( "fenodes", "database", "table", "username",
"password",
+ "streamLoadProp", "bufferFlushMaxRows", "bufferFlushMaxBytes",
"maxRetries",
+ "deletable", "stream_name", "field_name");
+
+ Map<String, String> getterMap = new HashMap<>();
+ getterMap.put( "fenodes", "getFenodes" );
+ getterMap.put( "database", "getDatabase" );
+ getterMap.put( "table", "getTable" );
+ getterMap.put( "username", "getUsername" );
+ getterMap.put( "password", "getPassword" );
+ getterMap.put( "streamLoadProp", "getStreamLoadProp" );
+ getterMap.put( "bufferFlushMaxRows", "getBufferFlushMaxRows" );
+ getterMap.put( "bufferFlushMaxBytes", "getBufferFlushMaxBytes" );
+ getterMap.put( "maxRetries", "getMaxRetries" );
+ getterMap.put( "deletable", "isDeletable" );
+ getterMap.put( "stream_name", "getFieldTable" );
+ getterMap.put( "field_name", "getFieldStream" );
+
+ Map<String, String> setterMap = new HashMap<>();
+ setterMap.put( "fenodes", "setFenodes" );
+ setterMap.put( "database", "setDatabase" );
+ setterMap.put( "table", "setTable" );
+ setterMap.put( "username", "setUsername" );
+ setterMap.put( "password", "setPassword" );
+ setterMap.put( "streamLoadProp", "setStreamLoadProp" );
+ setterMap.put( "bufferFlushMaxRows", "setBufferFlushMaxRows" );
+ setterMap.put( "bufferFlushMaxBytes", "setBufferFlushMaxBytes" );
+ setterMap.put( "maxRetries", "setMaxRetries" );
+ setterMap.put( "deletable", "setDeletable" );
+ setterMap.put( "stream_name", "setFieldTable" );
+ setterMap.put( "field_name", "setFieldStream" );
+
+ Map<String, FieldLoadSaveValidator<?>>
fieldLoadSaveValidatorAttributeMap = new HashMap<>();
+ FieldLoadSaveValidator<String[]> stringArrayLoadSaveValidator = new
ArrayLoadSaveValidator<>( new StringLoadSaveValidator(), 25 );
+
+ fieldLoadSaveValidatorAttributeMap.put("maxRetries", new
IntLoadSaveValidator( Integer.MAX_VALUE ));
+ fieldLoadSaveValidatorAttributeMap.put("bufferFlushMaxRows", new
LongLoadSaveValidator());
+ fieldLoadSaveValidatorAttributeMap.put("bufferFlushMaxBytes", new
LongLoadSaveValidator());
+ fieldLoadSaveValidatorAttributeMap.put("streamLoadProp", new
StringLoadSaveValidator());
+ fieldLoadSaveValidatorAttributeMap.put("stream_name",
stringArrayLoadSaveValidator );
+ fieldLoadSaveValidatorAttributeMap.put("field_name",
stringArrayLoadSaveValidator );
+
+ LoadSaveTester loadSaveTester =
+ new LoadSaveTester( DorisStreamLoaderMeta.class, attributes,
getterMap, setterMap,
+ fieldLoadSaveValidatorAttributeMap, new HashMap<String,
FieldLoadSaveValidator<?>>() );
+
+ loadSaveTester.testSerialization();
+ }
+
+ @Test
+ public void testPDI16559() throws Exception {
+ DorisStreamLoaderMeta streamLoader = new DorisStreamLoaderMeta();
+ streamLoader.setFieldTable( new String[] { "table1", "table2",
"table3" } );
+ streamLoader.setFieldStream( new String[] { "stream1" } );
+ streamLoader.setTable( "test_table" );
+
+ try {
+ String badXml = streamLoader.getXML();
+ Assert.fail( "Before calling afterInjectionSynchronization, should
have thrown an ArrayIndexOOB" );
+ } catch ( Exception expected ) {
+ // Do Nothing
+ }
+ streamLoader.afterInjectionSynchronization();
+ //run without a exception
+ String ktrXml = streamLoader.getXML();
+
+ int targetSz = streamLoader.getFieldTable().length;
+ Assert.assertEquals( targetSz, streamLoader.getFieldStream().length );
+ }
+}
diff --git
a/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderTest.java
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderTest.java
new file mode 100644
index 00000000000..45cadf62f78
--- /dev/null
+++
b/extension/kettle/impl/src/test/java/org/pentaho/di/trans/steps/dorisstreamloader/DorisStreamLoaderTest.java
@@ -0,0 +1,389 @@
+// 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.pentaho.di.trans.steps.dorisstreamloader;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang.StringUtils;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.pentaho.di.core.KettleEnvironment;
+import org.pentaho.di.core.exception.KettleException;
+import org.pentaho.di.core.plugins.PluginRegistry;
+import org.pentaho.di.core.plugins.StepPluginType;
+import org.pentaho.di.core.row.RowMeta;
+import org.pentaho.di.core.row.value.ValueMetaBigNumber;
+import org.pentaho.di.core.row.value.ValueMetaBoolean;
+import org.pentaho.di.core.row.value.ValueMetaDate;
+import org.pentaho.di.core.row.value.ValueMetaInteger;
+import org.pentaho.di.core.row.value.ValueMetaNumber;
+import org.pentaho.di.core.row.value.ValueMetaString;
+import org.pentaho.di.core.row.value.ValueMetaTimestamp;
+import org.pentaho.di.junit.rules.RestorePDIEngineEnvironment;
+import org.pentaho.di.trans.Trans;
+import org.pentaho.di.trans.TransMeta;
+import org.pentaho.di.trans.step.StepMeta;
+import org.pentaho.di.trans.steps.dorisstreamloader.load.DorisRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+
+/**
+ * Test the deletion and writing process of kettle,
+ * temporarily using the doris environment passed in by the user.
+ * mvn clean package -Ddoris_host=127.0.0.1 -Ddoris_http_port=8030
-Ddoris_passwd= -Ddoris_query_port=9030 -Ddoris_user=root
+ */
+public class DorisStreamLoaderTest {
+
+ @ClassRule
+ public static RestorePDIEngineEnvironment env = new
RestorePDIEngineEnvironment();
+ private static final Logger LOG =
LoggerFactory.getLogger(DorisStreamLoaderTest.class);
+ DorisStreamLoaderMeta lmeta;
+ DorisStreamLoaderData ldata;
+ DorisStreamLoader lder;
+ StepMeta smeta;
+
+ private static final String JDBC_URL = "jdbc:mysql://%s:%s";
+ private static final String DATABASE = "test_kettle";
+ private static final String TABLE_INSERT = "tbl_insert";
+ private static final String TABLE_DELETE = "tbl_delete";
+ private static String HOST;
+ private static String QUERY_PORT;
+ private static String HTTP_PORT;
+ private static String USER;
+ private static String PASSWD;
+
+ @BeforeClass
+ public static void initEnvironment() throws Exception {
+ checkParams();
+ KettleEnvironment.init();
+ }
+
+ private static void checkParams() {
+ Preconditions.checkArgument(
+ System.getProperty("doris_host") != null, "doris_host is
required.");
+ Preconditions.checkArgument(
+ System.getProperty("doris_query_port") != null, "doris_query_port
is required.");
+ Preconditions.checkArgument(
+ System.getProperty("doris_http_port") != null, "doris_http_port is
required.");
+ Preconditions.checkArgument(
+ System.getProperty("doris_user") != null, "doris_user is
required.");
+ Preconditions.checkArgument(
+ System.getProperty("doris_passwd") != null, "doris_passwd is
required.");
+ HOST = System.getProperty("doris_host");
+ QUERY_PORT = System.getProperty("doris_query_port");
+ HTTP_PORT = System.getProperty("doris_http_port");
+ USER = System.getProperty("doris_user");
+ PASSWD = System.getProperty("doris_passwd");
+ }
+
+ @Test
+ public void testWrite() throws KettleException, ParseException,
InterruptedException {
+ TransMeta transMeta = new TransMeta();
+ transMeta.setName( "DorisStreamLoader" );
+
+ Map<String, String> vars = new HashMap<String, String>();
+ vars.put( "fenodes", HOST + ":" + HTTP_PORT);
+ vars.put( "database", DATABASE );
+ vars.put( "table", TABLE_INSERT );
+ vars.put( "username", USER );
+ vars.put( "password", PASSWD );
+ vars.put( "bufferFlushMaxRows", "10000");
+ vars.put( "bufferFlushMaxBytes", 10 * 1024 * 1024 + "");
+ vars.put( "streamLoadProp", "format:json;read_json_by_line:true" );
+ vars.put( "maxRetries", "3" );
+ vars.put( "deletable", "false" );
+ transMeta.injectVariables(vars);
+
+ lmeta = new DorisStreamLoaderMeta();
+ lmeta.setFenodes(transMeta.environmentSubstitute("${fenodes}"));
+ lmeta.setDatabase(transMeta.environmentSubstitute("${database}"));
+ lmeta.setTable(transMeta.environmentSubstitute("${table}"));
+ lmeta.setUsername(transMeta.environmentSubstitute("${username}"));
+ lmeta.setPassword(transMeta.environmentSubstitute("${password}"));
+
lmeta.setBufferFlushMaxRows(Long.parseLong(transMeta.environmentSubstitute("${bufferFlushMaxRows}")));
+
lmeta.setBufferFlushMaxBytes(Long.parseLong(transMeta.environmentSubstitute("${bufferFlushMaxBytes}")));
+
lmeta.setMaxRetries(Integer.parseInt(transMeta.environmentSubstitute("${maxRetries}")));
+
lmeta.setStreamLoadProp(transMeta.environmentSubstitute("${streamLoadProp}"));
+
lmeta.setDeletable(Boolean.parseBoolean(transMeta.environmentSubstitute("${deletable}")));
+ lmeta.setFieldStream(new String[]{"id", "c1", "c2",
"c3","c4","c5","c6"});
+ lmeta.setFieldTable(new String[]{"id", "c1", "c2",
"c3","c4","c5","c6"});
+
+ ldata = new DorisStreamLoaderData();
+ PluginRegistry plugReg = PluginRegistry.getInstance();
+ String mblPid = plugReg.getPluginId( StepPluginType.class, lmeta );
+ smeta = new StepMeta( mblPid, "DorisStreamLoader", lmeta );
+ Trans trans = new Trans( transMeta );
+ transMeta.addStep( smeta );
+ lder = Mockito.spy(new DorisStreamLoader( smeta, ldata, 1, transMeta,
trans ));
+
+ RowMeta rm = new RowMeta();
+ rm.addValueMeta(new ValueMetaInteger("id"));
+ rm.addValueMeta(new ValueMetaBoolean("c1"));
+ rm.addValueMeta(new ValueMetaBigNumber("c2"));
+ rm.addValueMeta(new ValueMetaNumber("c3"));
+ rm.addValueMeta(new ValueMetaDate("c4"));
+ rm.addValueMeta(new ValueMetaTimestamp("c5"));
+ rm.addValueMeta(new ValueMetaString("c6"));
+ lder.setInputRowMeta(rm);
+
+ lder.copyVariablesFrom( transMeta );
+ initializeTable(TABLE_INSERT);
+ lder.init(lmeta, ldata);
+
+ Object[] data = new Object[]{
+ 1L,
+ true,
+ BigDecimal.valueOf(123456789L),
+ 12345.6789,
+ new SimpleDateFormat("yyyy-MM-dd").parse("2024-11-18"),
+ Timestamp.valueOf("2024-11-18 15:30:45"),
+ "First Row"};
+ doReturn(data).when(lder).getRow();
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws
Throwable {
+ lder.dispose(lmeta, ldata);
+ return null;
+ }
+ }).when(lder).putRow(any(), any());
+ lder.processRow(lmeta, ldata);
+ lder.dispose(lmeta, ldata);
+
+ while (lder.getStreamLoad() != null &&
lder.getStreamLoad().isLoadThreadAlive()){
+ Thread.sleep(1000);
+ LOG.info("Waiting for load thread to finish.");
+ }
+ List<String> expected =
Arrays.asList("1,true,123456789,12345.6789,2024-11-18,2024-11-18T15:30:45,First
Row");
+ String query =
+ String.format("select * from %s.%s order by 1", DATABASE,
TABLE_INSERT);
+ checkResult(getQueryConnection(), LOG, expected, query, 7);
+ }
+
+
+ @Test
+ public void testDelete() throws KettleException, ParseException,
InterruptedException {
+ TransMeta transMeta = new TransMeta();
+ transMeta.setName( "DorisStreamLoaderDelete" );
+
+ Map<String, String> vars = new HashMap<String, String>();
+ vars.put( "fenodes", HOST + ":" + HTTP_PORT);
+ vars.put( "database", DATABASE );
+ vars.put( "table", TABLE_DELETE );
+ vars.put( "username", USER );
+ vars.put( "password", PASSWD );
+ vars.put( "bufferFlushMaxRows", "10000");
+ vars.put( "bufferFlushMaxBytes", 10 * 1024 * 1024 + "");
+ vars.put( "streamLoadProp", "format:json;read_json_by_line:true" );
+ vars.put( "maxRetries", "3" );
+ vars.put( "deletable", "true" );
+ transMeta.injectVariables(vars);
+
+ lmeta = new DorisStreamLoaderMeta();
+ lmeta.setFenodes(transMeta.environmentSubstitute("${fenodes}"));
+ lmeta.setDatabase(transMeta.environmentSubstitute("${database}"));
+ lmeta.setTable(transMeta.environmentSubstitute("${table}"));
+ lmeta.setUsername(transMeta.environmentSubstitute("${username}"));
+ lmeta.setPassword(transMeta.environmentSubstitute("${password}"));
+
lmeta.setBufferFlushMaxRows(Long.parseLong(transMeta.environmentSubstitute("${bufferFlushMaxRows}")));
+
lmeta.setBufferFlushMaxBytes(Long.parseLong(transMeta.environmentSubstitute("${bufferFlushMaxBytes}")));
+
lmeta.setMaxRetries(Integer.parseInt(transMeta.environmentSubstitute("${maxRetries}")));
+
lmeta.setStreamLoadProp(transMeta.environmentSubstitute("${streamLoadProp}"));
+
lmeta.setDeletable(Boolean.parseBoolean(transMeta.environmentSubstitute("${deletable}")));
+ lmeta.setFieldStream(new String[]{"id", "c1", "c2",
"c3","c4","c5","c6"});
+ lmeta.setFieldTable(new String[]{"id", "c1", "c2",
"c3","c4","c5","c6"});
+
+ ldata = new DorisStreamLoaderData();
+ PluginRegistry plugReg = PluginRegistry.getInstance();
+ String mblPid = plugReg.getPluginId( StepPluginType.class, lmeta );
+ smeta = new StepMeta( mblPid, "DorisStreamLoaderDelete", lmeta );
+ Trans trans = new Trans( transMeta );
+ transMeta.addStep( smeta );
+ lder = Mockito.spy(new DorisStreamLoader( smeta, ldata, 1, transMeta,
trans ));
+
+ RowMeta rm = new RowMeta();
+ rm.addValueMeta(new ValueMetaInteger("id"));
+ rm.addValueMeta(new ValueMetaBoolean("c1"));
+ rm.addValueMeta(new ValueMetaBigNumber("c2"));
+ rm.addValueMeta(new ValueMetaNumber("c3"));
+ rm.addValueMeta(new ValueMetaDate("c4"));
+ rm.addValueMeta(new ValueMetaTimestamp("c5"));
+ rm.addValueMeta(new ValueMetaString("c6"));
+ lder.setInputRowMeta(rm);
+
+ lder.copyVariablesFrom( transMeta );
+ initializeTable(TABLE_DELETE);
+ mockDorisData(TABLE_DELETE);
+ lder.init(lmeta, ldata);
+
+ Object[] data = new Object[]{
+ 1L,
+ true,
+ BigDecimal.valueOf(123456789L),
+ 12345.6789,
+ new SimpleDateFormat("yyyy-MM-dd").parse("2024-11-18"),
+ Timestamp.valueOf("2024-11-18 15:30:45"),
+ "First Row"};
+ doReturn(data).when(lder).getRow();
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws
Throwable {
+ lder.dispose(lmeta, ldata);
+ return null;
+ }
+ }).when(lder).putRow(any(), any());
+ lder.processRow(lmeta, ldata);
+ lder.dispose(lmeta, ldata);
+
+ while (lder.getStreamLoad() != null &&
lder.getStreamLoad().isLoadThreadAlive()){
+ Thread.sleep(1000);
+ LOG.info("Waiting for load thread to finish.");
+ }
+ List<String> expected =
Arrays.asList("2,false,987654321,98765.4321,2024-11-17,2024-11-17T08:15:30,Second
Row");
+ String query =
+ String.format("select * from %s.%s order by 1", DATABASE,
TABLE_DELETE);
+ checkResult(getQueryConnection(), LOG, expected, query, 7);
+ }
+
+
+ private void initializeTable(String table) {
+ executeSQLStatement(
+ getQueryConnection(),
+ LOG,
+ String.format("CREATE DATABASE IF NOT EXISTS %s", DATABASE),
+ String.format("DROP TABLE IF EXISTS %s.%s", DATABASE, table),
+ String.format(
+ "CREATE TABLE %s.%s (\n" +
+ " `id` int, \n" +
+ " `c1` boolean, \n" +
+ " `c2` bigint,\n" +
+ " `c3` decimal(12,4),\n" +
+ " `c4` date,\n" +
+ " `c5` datetime,\n" +
+ " `c6` string\n" +
+ ") ENGINE=OLAP\n" +
+ "UNIQUE KEY(`id`)\n" +
+ "COMMENT 'OLAP'\n" +
+ "DISTRIBUTED BY HASH(`id`) BUCKETS 1\n" +
+ "PROPERTIES (\n" +
+ "\"replication_allocation\" = \"tag.location.default:
1\"\n" +
+ "); \n",
+ DATABASE, table));
+ }
+
+ public void mockDorisData(String table){
+ executeSQLStatement(
+ getQueryConnection(),
+ LOG,
+ String.format("INSERT INTO %s.%s VALUES \n" +
+ "(1, TRUE, 123456789, 12345.6789, '2024-11-18', '2024-11-18
15:30:45', 'First Row')," +
+ "(2, FALSE, 987654321, 98765.4321, '2024-11-17', '2024-11-17
08:15:30', 'Second Row')",
+ DATABASE, table));
+ }
+
+
+ public Connection getQueryConnection() {
+ LOG.info("Try to get query connection from doris.");
+ String jdbcUrl =
+ String.format(
+ JDBC_URL,
+ System.getProperty("doris_host"),
+ System.getProperty("doris_query_port"));
+ try {
+ return DriverManager.getConnection(jdbcUrl, USER, PASSWD);
+ } catch (SQLException e) {
+ LOG.info("Failed to get doris query connection. jdbcUrl={}",
jdbcUrl, e);
+ throw new DorisRuntimeException(e);
+ }
+ }
+
+ public static void executeSQLStatement(Connection connection, Logger
logger, String... sql) {
+ if (Objects.isNull(sql) || sql.length == 0) {
+ return;
+ }
+ try (Statement statement = connection.createStatement()) {
+ for (String s : sql) {
+ if (StringUtils.isNotEmpty(s)) {
+ logger.info("start to execute sql={}", s);
+ statement.execute(s);
+ }
+ }
+ } catch (SQLException e) {
+ throw new DorisRuntimeException(e);
+ }
+ }
+
+ public static void checkResult(
+ Connection connection,
+ Logger logger,
+ List<String> expected,
+ String query,
+ int columnSize) {
+ List<String> actual = new ArrayList<>();
+ try (Statement statement = connection.createStatement()) {
+ ResultSet sinkResultSet = statement.executeQuery(query);
+ while (sinkResultSet.next()) {
+ List<String> row = new ArrayList<>();
+ for (int i = 1; i <= columnSize; i++) {
+ Object value = sinkResultSet.getObject(i);
+ if (value == null) {
+ row.add("null");
+ } else {
+ row.add(value.toString());
+ }
+ }
+ actual.add(StringUtils.join(row, ","));
+ }
+ } catch (SQLException e) {
+ logger.info(
+ "Failed to check query result. expected={}, actual={}",
+ String.join(",", expected),
+ String.join(",", actual),
+ e);
+ throw new DorisRuntimeException(e);
+ }
+ logger.info(
+ "checking test result. expected={}, actual={}",
+ String.join(",", expected),
+ String.join(",", actual));
+ Assert.assertArrayEquals(expected.toArray(), actual.toArray());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
