feat(TBD-12334): add run submit mode to DBR (#6629) (#6659)

- Add migration task to set default value

i18n/plugins/org.talend.designer.core.nl/src/main/java/messages_en.properties

@@ -584,13 +584,11 @@ EParameterName.jdbcURL=JDBC URL
 EParameterName.driverJar=Driver jar
 EParameterName.className=Class name
 EParameterName.mappingFile=Mapping file
-SetupProcessDependenciesRoutinesAction.title=Setup routine dependencies
+SetupProcessDependenciesRoutinesAction.title=Setup Codes Dependencies
 SetupProcessDependenciesRoutinesDialog.systemRoutineLabel=System routines
-SetupProcessDependenciesRoutinesDialog.userRoutineLabel=User routines
 PerformancePreferencePage.addAllSystemRoutines=Add all system routines to job dependencies, when creating a new job
 PerformancePreferencePage.addAllUserRoutines=Add all user routines to job dependencies, when creating a new job
 ShowRoutineItemsDialog.systemTitle=Select Sytem Routines
-ShowRoutineItemsDialog.title=Select Routines
 AbstractMultiPageTalendEditor_pleaseWait=Saving Please Wait....
 DocumentationPreferencePage.use_css_template=Use CSS file as a template when export to HTML
 DocumentationPreferencePage.css_file=CSS File

i18n/plugins/org.talend.repository.view.di.nl/plugin_en.properties

@@ -1,5 +1,5 @@
 NavigatorContent.contexts=Contexts
-NavigatorContent.routines=Routines
+NavigatorContent.routines=Global Routines
 NavigatorContent.sqlTemplates=SQL Templates
 NavigatorContent.documentation=Documentation
 NavigatorContent.activation=di.fake.for.activation

main/features/org.talend.rcp.branding.tos.test.feature/feature.xml

@@ -6,7 +6,7 @@
   <license url="http://www.example.com/license">[Enter License Description here.]</license>  
   <requires>
     <import feature="org.eclipse.test" version="0.0.0" match="greaterOrEqual"/>
-    <import plugin="org.junit" version="0.0.0" match="greaterOrEqual"/>
+    <import plugin="org.junit" version="4.13.2" match="greaterOrEqual"/>
     <import plugin="org.talend.commons.runtime" version="0.0.0" match="greaterOrEqual"/>
     <import plugin="org.talend.commons.ui" version="0.0.0" match="greaterOrEqual"/>
     <import plugin="org.talend.core" version="0.0.0" match="greaterOrEqual"/>

main/features/org.talend.tos.libraries.feature/feature.xml

@@ -16,7 +16,6 @@
   </requires>  
   <plugin id="org.talend.libraries.apache" download-size="0" install-size="0" version="0.0.0"/>
   <plugin id="org.talend.libraries.apache.axis2" download-size="0" install-size="0" version="0.0.0"/>
-  <plugin id="org.talend.libraries.apache.batik" download-size="0" install-size="0" version="0.0.0"/>
   <plugin id="org.talend.libraries.apache.chemistry" download-size="0" install-size="0" version="0.0.0"/>
   <plugin id="org.talend.libraries.apache.common" download-size="0" install-size="0" version="0.0.0"/>
   <plugin id="org.talend.libraries.apache.cxf" download-size="0" install-size="0" version="0.0.0"/>
@@ -51,5 +50,4 @@
   <plugin id="org.talend.libraries.slf4j" download-size="0" install-size="0" version="0.0.0"/>
   <plugin id="org.talend.libraries.xml" download-size="0" install-size="0" version="0.0.0"/>
   <plugin id="org.talend.libraries.zmq" download-size="0" install-size="0" version="0.0.0"/>
-  <plugin id="org.talend.libraries.zookeeper" download-size="0" install-size="0" version="0.0.0"/>
 </feature>

main/plugins/org.talend.designer.codegen/jet_stub/component_part_header.javajet

@@ -412,7 +412,7 @@
                     
                     if(ignoredParamsNames.contains(name)) {
                         //do nothing
-                    } else if(org.talend.core.model.process.EParameterFieldType.PASSWORD.equals(ep.getFieldType())){
+                    } else if(org.talend.core.model.process.EParameterFieldType.PASSWORD.equals(ep.getFieldType()) || org.talend.core.model.process.EParameterFieldType.HIDDEN_TEXT.equals(ep.getFieldType())){
                         //not log password
                     }else{
                         String value = org.talend.core.model.utils.NodeUtil.getRuntimeParameterValue(node, ep);

main/plugins/org.talend.designer.codegen/jet_stub/footer.javajet

@@ -126,8 +126,16 @@
 
     boolean exist_tSCP = false;
     List<INode> scpComponentsList = (List<INode>)process.getNodesOfType("tSCPConnection");
-    if (scpComponentsList.size() > 0) {
+	String parameterNames = "";
+	int scpsize = scpComponentsList.size();
+    if (scpsize > 0) {
         exist_tSCP = true;
+        for (int i = 0; i < scpsize; i++) {
+            parameterNames += "\"conn_" + scpComponentsList.get(i).getUniqueName() + "\"";
+			if(i < scpsize-1){
+			    parameterNames += ",";
+			}
+        }
     }
 
     boolean exist_tCassandra = false;
@@ -647,7 +655,7 @@
         if (inOSGi) {
             java.util.Dictionary<String, Object> jobProperties = routines.system.BundleUtils.getJobProperties(jobName);
 
-            if (jobProperties != null) {
+            if (jobProperties != null && jobProperties.get("context") != null) {
                 contextStr = (String)jobProperties.get("context");
             }
         }
@@ -733,34 +741,39 @@
 <%
                         } else if(typeToGenerate.equals("java.util.Date")) {
 %>
-                        try{
                             String context_<%=ctxParam.getName()%>_value = context.getProperty("<%=ctxParam.getName()%>");
-                            if (context_<%=ctxParam.getName()%>_value == null){
-                                context_<%=ctxParam.getName()%>_value = "";
-                            }
-                            int context_<%=ctxParam.getName()%>_pos = context_<%=ctxParam.getName()%>_value.indexOf(";");
-                            String context_<%=ctxParam.getName()%>_pattern =  "yyyy-MM-dd HH:mm:ss";
-                            if(context_<%=ctxParam.getName()%>_pos > -1){
-                                context_<%=ctxParam.getName()%>_pattern = context_<%=ctxParam.getName()%>_value.substring(0, context_<%=ctxParam.getName()%>_pos);
-                                context_<%=ctxParam.getName()%>_value = context_<%=ctxParam.getName()%>_value.substring(context_<%=ctxParam.getName()%>_pos + 1);
-                            }
+                            try{
+                                if (context_<%=ctxParam.getName()%>_value == null){
+                                    context_<%=ctxParam.getName()%>_value = "";
+                                }
+                                int context_<%=ctxParam.getName()%>_pos = context_<%=ctxParam.getName()%>_value.indexOf(";");
+                                String context_<%=ctxParam.getName()%>_pattern =  "yyyy-MM-dd HH:mm:ss";
+                                if(context_<%=ctxParam.getName()%>_pos > -1){
+                                    context_<%=ctxParam.getName()%>_pattern = context_<%=ctxParam.getName()%>_value.substring(0, context_<%=ctxParam.getName()%>_pos);
+                                    context_<%=ctxParam.getName()%>_value = context_<%=ctxParam.getName()%>_value.substring(context_<%=ctxParam.getName()%>_pos + 1);
+                                }
 
-                            context.<%=ctxParam.getName()%>=(java.util.Date)(new java.text.SimpleDateFormat(context_<%=ctxParam.getName()%>_pattern).parse(context_<%=ctxParam.getName()%>_value));
+                                context.<%=ctxParam.getName()%>=(java.util.Date)(new java.text.SimpleDateFormat(context_<%=ctxParam.getName()%>_pattern).parse(context_<%=ctxParam.getName()%>_value));
 
-                        } catch(ParseException e) {
+                            } catch(ParseException e) {
+                                try { <% /*try to check if date passed as long also*/ %>
+                                    long context_<%=ctxParam.getName()%>_longValue = Long.valueOf(context_<%=ctxParam.getName()%>_value);
+                                    context.<%=ctxParam.getName()%> = new java.util.Date(context_<%=ctxParam.getName()%>_longValue);
+                                } catch (NumberFormatException cantParseToLongException) {
 <%
-                            if (isLog4jEnabled) {
+                                    if (isLog4jEnabled) {
 %>
-                                log.warn(String.format("<%=warningMessageFormat %>", "<%=ctxParam.getName() %>", e.getMessage()));
+                                        log.warn(String.format("<%=warningMessageFormat %>", "<%=ctxParam.getName() %>", "Can't parse date string: " + e.getMessage() + " and long: " + cantParseToLongException.getMessage()));
 <%
-                            } else {
+                                    } else {
 %>
-                                System.err.println(String.format("<%=warningMessageFormat %>", "<%=ctxParam.getName() %>", e.getMessage()));
+                                        System.err.println(String.format("<%=warningMessageFormat %>", "<%=ctxParam.getName() %>", "Can't parse date string: " + e.getMessage() + " and long: " + cantParseToLongException.getMessage()));
 <%
+                                    }
+%>
+                                    context.<%=ctxParam.getName()%>=null;
+                                }
                             }
-%>
-                            context.<%=ctxParam.getName()%>=null;
-                        }
 <%
                         } else if(typeToGenerate.equals("Object")||typeToGenerate.equals("String")||typeToGenerate.equals("java.lang.String")) {
 %>
@@ -1185,9 +1198,12 @@ this.globalResumeTicket = true;//to run tPostJob
     closeJmsConnections();
 <% } %>
 
-<% if (exist_tSCP) { %>
-    closeScpConnections();
-<% } %>
+<% if (exist_tSCP) { 
+%>
+    closeCloseableConnections(<%=parameterNames%>);
+<% 
+} 
+%>
 
 <%
     if (stats) {
@@ -1241,7 +1257,7 @@ if (execStat) {
     closeJmsConnections();
 <% } %>
 <% if(exist_tSCP) { %>
-    closeScpConnections();
+    closeCloseableConnections(<%=parameterNames%>);
 <% } %>
 <% if (exist_tSQLDB) { %>
     closeSqlDbConnections();
@@ -1309,22 +1325,17 @@ if (execStat) {
 <%
     if(exist_tSCP) {
 %>
-    private void closeScpConnections() {
-        try {
-            Object obj_conn;
-<%
-            for (INode scpNode : scpComponentsList) {
-%>
-            obj_conn = globalMap.remove("conn_<%=scpNode.getUniqueName() %>");
-            if (null != obj_conn) {
-                ((ch.ethz.ssh2.Connection) obj_conn).close();
+    private void closeCloseableConnections(String... names) {
+	        java.util.Arrays.stream(names).forEach(name-> {
+            try {
+			    Object obj_conn = globalMap.remove(name);
+				if(obj_conn != null){
+				    ((java.io.Closeable)obj_conn).close();
+				}
+            } catch (IOException ioException) {
             }
-<%
-            }
-%>
-        } catch (java.lang.Exception e) {
-        }
-    }
+        });
+	}
 <%
     }
 %>
@@ -1434,6 +1445,7 @@ if (execStat) {
             if ("sftp".equals(type)) { %>
                 ((com.jcraft.jsch.ChannelSftp) obj_conn).quit();
             <%} else { %>
+                ((org.apache.commons.net.ftp.FTPClient) obj_conn).logout();
                 ((org.apache.commons.net.ftp.FTPClient) obj_conn).disconnect();
             <%}%>
             }

main/plugins/org.talend.designer.codegen/jet_stub/generic/component_begin.javajet

@@ -56,11 +56,25 @@ if ((metadatas != null) && (metadatas.size() > 0)) { // metadata
 // Set up the component definition, and the properties for all types of
 // components.
 
+List<? extends IConnection> allInLineJobConns = NodeUtil.getFirstIncomingLineConnectionsOfType(node, "tRESTRequestIn");
+%>
+boolean doesNodeBelongToRequest_<%=cid%> = <%= allInLineJobConns.size() %> == 0;
+@SuppressWarnings("unchecked")
+java.util.Map<String, Object> restRequest_<%=cid%> = (java.util.Map<String, Object>)globalMap.get("restRequest");
+String currentTRestRequestOperation_<%=cid%> = (String)(restRequest_<%=cid%> != null ? restRequest_<%=cid%>.get("OPERATION") : null);
+<%
+for (IConnection inLineConn : allInLineJobConns) {
+    %>
+        if("<%= inLineConn.getName() %>".equals(currentTRestRequestOperation_<%=cid%>)) {
+            doesNodeBelongToRequest_<%=cid%> = true;
+        }
+	<%
+    }
 %>
 
 org.talend.components.api.component.ComponentDefinition def_<%=cid %> =
         new <%= def.getClass().getName()%>();
-        
+
 org.talend.components.api.component.runtime.Writer writer_<%=cid%> = null;
 org.talend.components.api.component.runtime.Reader reader_<%=cid%> = null;
 
@@ -149,7 +163,7 @@ globalMap.put("TALEND_COMPONENTS_VERSION", "<%=component.getVersion()%>");
 boolean isParallelize ="true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__PARALLELIZE__"));
 if (isParallelize) {
 %>
-final String buffersSizeKey_<%=cid%> = "buffersSizeKey_<%=cid%>_" + Thread.currentThread().getId(); 
+final String buffersSizeKey_<%=cid%> = "buffersSizeKey_<%=cid%>_" + Thread.currentThread().getId();
 <%
 }
 %>
@@ -219,9 +233,11 @@ if(componentRuntime_<%=cid%> instanceof org.talend.components.api.component.runt
 org.talend.components.api.component.runtime.SourceOrSink sourceOrSink_<%=cid%> = null;
 if(componentRuntime_<%=cid%> instanceof org.talend.components.api.component.runtime.SourceOrSink) {
 	sourceOrSink_<%=cid%> = (org.talend.components.api.component.runtime.SourceOrSink)componentRuntime_<%=cid%>;
-	org.talend.daikon.properties.ValidationResult vr_<%=cid%> = sourceOrSink_<%=cid%>.validate(container_<%=cid%>);
-	if (vr_<%=cid%>.getStatus() == org.talend.daikon.properties.ValidationResult.Result.ERROR ) {
-    	throw new RuntimeException(vr_<%=cid%>.getMessage());
+	if (doesNodeBelongToRequest_<%=cid%>) {
+        org.talend.daikon.properties.ValidationResult vr_<%=cid%> = sourceOrSink_<%=cid%>.validate(container_<%=cid%>);
+        if (vr_<%=cid%>.getStatus() == org.talend.daikon.properties.ValidationResult.Result.ERROR ) {
+            throw new RuntimeException(vr_<%=cid%>.getMessage());
+        }
 	}
 }
 
@@ -297,13 +313,13 @@ if (hasOutputOnly || asInputComponent) {
 
         for (; available_<%=cid%>; available_<%=cid%> = reader_<%=cid%>.advance()) {
 			nb_line_<%=cid %>++;
-	
+
 			<%if(hasDataOutput) {%>
 			if (multi_output_is_allowed_<%=cid%>) {
 				<%if(main!=null){%>
 					<%=main.getName()%> = null;
 				<%}%>
-	
+
 				<%if(reject!=null){%>
 					<%=reject.getName()%> = null;
 				<%}%>
@@ -315,11 +331,11 @@ if (hasOutputOnly || asInputComponent) {
 				<%
 				if (main != null) {
 					%>
-	
+
 					if(multi_output_is_allowed_<%=cid%>) {
 						<%=main.getName()%> = new <%=main.getName() %>Struct();
 					}
-	
+
 					<%
 					irToRow.generateConvertRecord("data_" + cid, main.getName(), main.getMetadataTable().getListColumns());
 				}
@@ -330,7 +346,7 @@ if (hasOutputOnly || asInputComponent) {
 				if (reject!=null) {
 					%>
 					Object data_<%=cid%> = info_<%=cid%>.get("talend_record");
-	
+
 					if (multi_output_is_allowed_<%=cid%>) {
 						<%=reject.getName()%> = new <%=reject.getName() %>Struct();
 					}
@@ -343,19 +359,19 @@ if (hasOutputOnly || asInputComponent) {
 					}
 					<%
 					Set<String> commonColumns = new HashSet<String>();
-	
+
 					for (IMetadataColumn column : columnList) {
 						commonColumns.add(column.getLabel());
 					}
-	
+
 					//pass error columns
 					List<IMetadataColumn> rejectColumns = reject.getMetadataTable().getListColumns();
 					for(IMetadataColumn column : rejectColumns) {
 						String columnName = column.getLabel();
-	
+
 						// JavaType javaType = JavaTypesManager.getJavaTypeFromId(column.getTalendType());
 						String typeToGenerate = JavaTypesManager.getTypeToGenerate(column.getTalendType(), column.isNullable());
-	
+
 						//error columns
 						if(!commonColumns.contains(columnName)) {
 							%>
@@ -385,7 +401,7 @@ if (hasOutputOnly || asInputComponent) {
 				<%
 				}
 			%>
-			} // end of catch 
+			} // end of catch
     <%
     // The for loop around the incoming records from the reader is left open.
 
@@ -397,9 +413,13 @@ if (hasOutputOnly || asInputComponent) {
     	org.talend.components.api.component.runtime.Sink sink_<%=cid%> =
             	(org.talend.components.api.component.runtime.Sink)sourceOrSink_<%=cid%>;
         org.talend.components.api.component.runtime.WriteOperation writeOperation_<%=cid%> = sink_<%=cid%>.createWriteOperation();
-        writeOperation_<%=cid%>.initialize(container_<%=cid%>);
+        if (doesNodeBelongToRequest_<%=cid%>) {
+            writeOperation_<%=cid%>.initialize(container_<%=cid%>);
+        }
         writer_<%=cid%> = writeOperation_<%=cid%>.createWriter(container_<%=cid%>);
-        writer_<%=cid%>.open("<%=cid%>");
+        if (doesNodeBelongToRequest_<%=cid%>) {
+            writer_<%=cid%>.open("<%=cid%>");
+        }
 
         resourceMap.put("writer_<%=cid%>", writer_<%=cid%>);
     } // end of "sourceOrSink_<%=cid%> instanceof ...Sink"
@@ -448,7 +468,7 @@ if (hasOutputOnly || asInputComponent) {
     }
 }
  %>
-    
+
                 java.lang.Iterable<?> outgoingMainRecordsList_<%=cid%> = new java.util.ArrayList<Object>();
                 java.util.Iterator outgoingMainRecordsIt_<%=cid%> = null;
 

main/plugins/org.talend.designer.codegen/jet_stub/generic/component_main.javajet

@@ -120,7 +120,8 @@ if(hasInput){
                             dm_<%=cid%>.getLogicalType(),
                             dm_<%=cid%>.getFormat(),
                             dm_<%=cid%>.getDescription(),
-                            dm_<%=cid%>.isNullable());
+                            dm_<%=cid%>.isNullable(),
+                            dm_<%=cid%>.isKey());
                 }
                 incomingEnforcer_<%=cid%>.createRuntimeSchema();
             }

main/plugins/org.talend.designer.codegen/jet_stub/header.javajet

@@ -73,6 +73,9 @@ import pigudf.<%=routine%>;
 import routines.<%=routine%>;
 <%  }
 }%>
+<%for (String codesJar : CodeGeneratorRoutine.getRequiredCodesJarName(process)) {%>
+import <%=codesJar%>;
+<%}%>
 import routines.system.*;
 import routines.system.api.*;
 import java.text.ParseException;

main/plugins/org.talend.designer.codegen/src/main/java/org/talend/designer/codegen/JavaRoutineSynchronizer.java

@@ -69,6 +69,15 @@ public class JavaRoutineSynchronizer extends AbstractRoutineSynchronizer {
         syncRoutineItems(getRoutines(true), true);
     }
 
+    @Override
+    public void syncAllInnerCodes() throws SystemException {
+        syncInnerCodeItems(false);
+    }
+
+    @Override
+    public void syncAllInnerCodesForLogOn() throws SystemException {
+        syncInnerCodeItems(true);
+    }
 
     private void syncRoutineItems(Collection<RoutineItem> routineObjects, boolean forceUpdate) throws SystemException {
         for (RoutineItem routineItem : routineObjects) {

main/plugins/org.talend.designer.components.exchange/META-INF/MANIFEST.MF

@@ -8,6 +8,7 @@ Require-Bundle: org.eclipse.core.runtime,
  org.eclipse.ui,
  org.apache.log4j,
  org.apache.commons.collections,
+ org.apache.commons.discovery,
  org.apache.commons.logging,
  org.apache.commons.beanutils,
  org.apache.commons.io,
@@ -25,7 +26,6 @@ Require-Bundle: org.eclipse.core.runtime,
  org.talend.repository,
  org.talend.core.repository,
  org.talend.updates.runtime,
- org.apache.axis,
  org.eclipse.ui.intro,
  org.eclipse.ui.forms,
  org.eclipse.jface.text

main/plugins/org.talend.designer.components.exchange/src/main/java/org/talend/designer/components/exchange/proxy/DefaultHTTPSTransportClientProperties.java

@@ -1,66 +0,0 @@
-// ============================================================================
-//
-// Copyright (C) 2006-2019 Talend Inc. - www.talend.com
-//
-// This source code is available under agreement available at
-// %InstallDIR%\features\org.talend.rcp.branding.%PRODUCTNAME%\%PRODUCTNAME%license.txt
-//
-// You should have received a copy of the agreement
-// along with this program; if not, write to Talend SA
-// 9 rue Pages 92150 Suresnes, France
-//
-// ============================================================================
-package org.talend.designer.components.exchange.proxy;
-
-import org.apache.commons.lang.StringUtils;
-
-/**
- *
- * DOC hcyi class global comment. Detailled comment
- */
-public class DefaultHTTPSTransportClientProperties extends DefaultHTTPTransportClientProperties {
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getProxyHost()
-     */
-    @Override
-    public String getProxyHost() {
-        return StringUtils.trimToEmpty(System.getProperty("https.proxyHost")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getNonProxyHosts()
-     */
-    @Override
-    public String getNonProxyHosts() {
-
-        return StringUtils.trimToEmpty(System.getProperty("https.nonProxyHosts")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getPort()
-     */
-    @Override
-    public String getProxyPort() {
-
-        return StringUtils.trimToEmpty(System.getProperty("https.proxyPort")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getUser()
-     */
-    @Override
-    public String getProxyUser() {
-
-        return StringUtils.trimToEmpty(System.getProperty("https.proxyUser")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getPassword()
-     */
-    @Override
-    public String getProxyPassword() {
-
-        return StringUtils.trimToEmpty(System.getProperty("https.proxyPassword")); //$NON-NLS-1$
-    }
-}

main/plugins/org.talend.designer.components.exchange/src/main/java/org/talend/designer/components/exchange/proxy/DefaultHTTPTransportClientProperties.java

@@ -1,58 +0,0 @@
-// ============================================================================
-//
-// Copyright (C) 2006-2019 Talend Inc. - www.talend.com
-//
-// This source code is available under agreement available at
-// %InstallDIR%\features\org.talend.rcp.branding.%PRODUCTNAME%\%PRODUCTNAME%license.txt
-//
-// You should have received a copy of the agreement
-// along with this program; if not, write to Talend SA
-// 9 rue Pages 92150 Suresnes, France
-//
-// ============================================================================
-package org.talend.designer.components.exchange.proxy;
-
-import org.apache.axis.components.net.TransportClientProperties;
-import org.apache.commons.lang.StringUtils;
-
-/**
- *
- * DOC hcyi class global comment. Detailled comment
- */
-public class DefaultHTTPTransportClientProperties implements TransportClientProperties {
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getProxyHost()
-     */
-    public String getProxyHost() {
-        return StringUtils.trimToEmpty(System.getProperty("http.proxyHost")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getNonProxyHosts()
-     */
-    public String getNonProxyHosts() {
-        return StringUtils.trimToEmpty(System.getProperty("http.nonProxyHosts")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getPort()
-     */
-    public String getProxyPort() {
-        return StringUtils.trimToEmpty(System.getProperty("http.proxyPort")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getProxyUser()
-     */
-    public String getProxyUser() {
-        return StringUtils.trimToEmpty(System.getProperty("http.proxyUser")); //$NON-NLS-1$
-    }
-
-    /**
-     * @see org.apache.axis.components.net.TransportClientProperties#getProxyPassword()
-     */
-    public String getProxyPassword() {
-        return StringUtils.trimToEmpty(System.getProperty("http.proxyPassword")); //$NON-NLS-1$
-    }
-}

main/plugins/org.talend.designer.components.exchange/src/main/java/org/talend/designer/components/exchange/util/ExchangeUtils.java

@@ -25,11 +25,9 @@ import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-
-import org.apache.axis.components.net.TransportClientProperties;
-import org.apache.axis.components.net.TransportClientPropertiesFactory;
 import org.apache.commons.beanutils.BeanUtils;
 import org.apache.commons.collections.map.MultiValueMap;
+import org.apache.commons.discovery.tools.ManagedProperties;
 import org.apache.commons.httpclient.HostConfiguration;
 import org.apache.commons.httpclient.HttpClient;
 import org.apache.commons.httpclient.NameValuePair;
@@ -151,14 +149,17 @@ public class ExchangeUtils {
     public static String sendGetRequest(String urlAddress) throws Exception {
         HttpClient httpclient = new HttpClient();
         GetMethod getMethod = new GetMethod(urlAddress);
-        TransportClientProperties tcp = TransportClientPropertiesFactory.create("http");
-        if (tcp.getProxyHost().length() != 0) {
+        String proxyUser = ManagedProperties.getProperty("http.proxyUser");
+        String proxyPassword = ManagedProperties.getProperty("http.proxyPassword");
+        String proxyHost = ManagedProperties.getProperty("http.proxyHost");
+        proxyHost = proxyHost != null ? proxyHost : "";
+        String proxyPort = ManagedProperties.getProperty("http.proxyPort");
+        if (proxyHost.length() != 0) {
             UsernamePasswordCredentials creds = new UsernamePasswordCredentials(
-                    tcp.getProxyUser() != null ? tcp.getProxyUser() : "",
-                    tcp.getProxyPassword() != null ? tcp.getProxyUser() : "");
+                    proxyUser != null ? proxyUser : "", proxyPassword != null ? proxyPassword : "");
             httpclient.getState().setProxyCredentials(AuthScope.ANY, creds);
             HostConfiguration hcf = new HostConfiguration();
-            hcf.setProxy(tcp.getProxyHost(), Integer.parseInt(tcp.getProxyPort()));
+            hcf.setProxy(proxyHost, Integer.parseInt(proxyPort));
             httpclient.executeMethod(hcf, getMethod);
         } else {
             httpclient.executeMethod(getMethod);

main/plugins/org.talend.designer.components.libs/libs_src/filecopy/pom.xml

@@ -4,7 +4,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>org.talend.components</groupId>
 	<artifactId>filecopy</artifactId>
-	<version>2.0.1</version>
+	<version>2.0.3</version>
 	<packaging>jar</packaging>
 
 	<name>talend-copy</name>
@@ -14,6 +14,7 @@
 		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>
 		<java.source.version>1.8</java.source.version>
 		<junit5.version>5.4.2</junit5.version>
+		<slf4j.version>1.7.28</slf4j.version>
 	</properties>
 
 	<distributionManagement>
@@ -52,7 +53,12 @@
 			<version>${junit5.version}</version>
 			<scope>test</scope>
 		</dependency>
-
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+			<version>${slf4j.version}</version>
+			<scope>provided</scope>
+		</dependency>
 	</dependencies>
 	<build>
 		<plugins>

main/plugins/org.talend.designer.components.libs/libs_src/filecopy/src/main/java/org/talend/FileCopy.java

@@ -20,11 +20,16 @@ import java.nio.file.Paths;
 import java.nio.file.StandardCopyOption;
 import java.nio.file.attribute.FileTime;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * DOC Administrator class global comment. Detailled comment
  */
 public class FileCopy {
 
+    static Logger  logger = LoggerFactory.getLogger(Object.class);
+
     /** Private constructor, only static methods */
     private FileCopy() {
     }
@@ -37,19 +42,57 @@ public class FileCopy {
      * @param delSrc : true if delete source.
      * @throws IOException : if IO pb.
      */
-    public static void copyFile(String srcFileName, String desFileName, boolean delSrc) throws IOException {
+    public static void copyFile(String srcFileName, String desFileName, boolean delSrc, boolean keepModified)
+            throws IOException {
         final Path source = Paths.get(srcFileName);
         final Path destination =  Paths.get(desFileName);
-
+        FileTime lastModifiedTime = null;
+        try {
+            lastModifiedTime = Files.getLastModifiedTime(source);
+        } catch (IOException e) {
+            logger.warn(e.getLocalizedMessage());
+        }
         if (delSrc) {
             // move : more efficient if in same FS and mustr delete existing file.
-            FileTime lastModifiedTime = Files.getLastModifiedTime(source);
             Files.move(source, destination, StandardCopyOption.REPLACE_EXISTING);
-            Files.setLastModifiedTime(destination,lastModifiedTime);
         } else {
             Files.copy(source, destination, StandardCopyOption.REPLACE_EXISTING);
-            Files.setLastModifiedTime(destination,Files.getLastModifiedTime(source));
+        }
+        if(keepModified){
+            try {
+                Files.setLastModifiedTime(destination,lastModifiedTime);
+            } catch (IOException e) {
+                logger.warn(e.getLocalizedMessage());
+            }
         }
     }
 
+    public static void copyFile(String srcFileName, String desFileName, boolean delSrc ) throws IOException {
+        copyFile(srcFileName,desFileName,delSrc,true);
+    }
+
+    /**
+     * Force Copy and Delete files.
+     *
+     * @param srcFileName : file name for source file.
+     * @param desFileName : file name for destination file.
+     * @throws IOException : if IO pb.
+     */
+    public static void forceCopyAndDelete(String srcFileName, String desFileName, boolean keepModified) throws IOException {
+        final Path source = Paths.get(srcFileName);
+        final Path destination =  Paths.get(desFileName);
+        final long lastModifiedTime = new File(srcFileName).lastModified();
+
+        Files.copy(source, destination, StandardCopyOption.REPLACE_EXISTING);
+        Files.delete(source);
+        if(keepModified){
+            destination.toFile().setLastModified(lastModifiedTime);
+        }
+
+    }
+
+    public static void forceCopyAndDelete(String srcFileName, String desFileName) throws IOException {
+        forceCopyAndDelete(srcFileName,desFileName,true);
+    }
+
 }

main/plugins/org.talend.designer.components.libs/libs_src/filecopy/src/test/java/org/talend/FileCopyTest.java

@@ -100,6 +100,26 @@ class FileCopyTest {
         Assertions.assertEquals(referenceSize, copy.length(), "Size error");
     }
 
+    @Test
+    void testForceCopyWithDelete() throws Exception {
+        final URL repCopy = Thread.currentThread().getContextClassLoader().getResource("copy");
+
+        File file = this.buildFile("fileToDelete.txt", 10L * 1024L);
+        file.deleteOnExit();
+        File copy = new File(repCopy.getPath(), "fileToDelete.txt");
+        long referenceSize = file.length();
+        if (!copy.exists()) {
+            copy.createNewFile();
+        }
+        copy.deleteOnExit();
+
+        FileCopy.forceCopyAndDelete(file.getPath(), copy.getPath());
+
+        Assertions.assertFalse(file.exists(), "file not delete");
+        Assertions.assertTrue(copy.exists(), "small file : original file deleted");
+        Assertions.assertEquals(referenceSize, copy.length(), "Size error");
+    }
+
     @Test
     void testLastModifiedTime() throws Exception {
         final URL repCopy = Thread.currentThread().getContextClassLoader().getResource("copy");
@@ -143,4 +163,22 @@ class FileCopyTest {
 
         return generatedFile;
     }
+
+    @Test
+    void testKeepLastModifiedTime() throws Exception {
+        final URL repCopy = Thread.currentThread().getContextClassLoader().getResource("copy");
+
+        File file = this.buildFile("fileLMT.txt", 10L * 1024L);
+        file.deleteOnExit();
+        long referencceTime = 324723894L;
+        file.setLastModified(referencceTime);
+
+        File copy = new File(repCopy.getPath(), "fileLMTDestination.txt");
+        if (copy.exists()) {
+            copy.delete();
+        }
+        copy.deleteOnExit();
+        FileCopy.copyFile(file.getPath(), copy.getPath(), true,true);
+        Assertions.assertEquals(referencceTime, copy.lastModified(), "modified time is not idential");
+    }
 }

main/plugins/org.talend.designer.components.libs/libs_src/mdm_services_tomcat/pom.xml

@@ -11,7 +11,7 @@
 
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-		<cxf.version>3.1.2</cxf.version>
+		<cxf.version>3.3.10</cxf.version>
 	</properties>
 
 	<build>

main/plugins/org.talend.designer.components.libs/libs_src/talend-codegen-utils/pom.xml

@@ -7,21 +7,21 @@
 	<groupId>org.talend.libraries</groupId>
 	<artifactId>talend-codegen-utils</artifactId>
 	<!-- release for revert version of library -->
-	<version>0.28.0</version>
+	<version>0.31.0</version>
 	<packaging>jar</packaging>
 	
 	<properties>
 		<avro.version>1.8.0</avro.version>
-		<components.version>0.25.0-SNAPSHOT</components.version>
-		<daikon.version>0.26.0-SNAPSHOT</daikon.version>
+		<components.version>0.30.0</components.version>
+		<daikon.version>0.31.11</daikon.version>
 		<hamcrest.version>1.3</hamcrest.version>
 		<junit.version>4.12</junit.version>
 		<java-formatter.plugin.version>0.1.0</java-formatter.plugin.version>
 		<formatter.plugin.version>1.6.0-SNAPSHOT</formatter.plugin.version>
 		<mockito.version>2.2.15</mockito.version>
 		<jacoco.plugin.version>0.7.8</jacoco.plugin.version>
-		<maven.compiler.source>1.7</maven.compiler.source>
-		<maven.compiler.target>1.7</maven.compiler.target>
+		<maven.compiler.source>1.8</maven.compiler.source>
+		<maven.compiler.target>1.8</maven.compiler.target>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>
 	</properties>

main/plugins/org.talend.designer.components.libs/libs_src/talend-codegen-utils/src/main/java/org/talend/codegen/enforcer/IncomingSchemaEnforcer.java

@@ -24,8 +24,10 @@ import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.TimeZone;
 
 import org.apache.avro.Schema;
@@ -33,9 +35,11 @@ import org.apache.avro.Schema.Field;
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.IndexedRecord;
+import org.apache.avro.SchemaParseException;
 import org.talend.codegen.DiSchemaConstants;
 import org.talend.daikon.avro.AvroUtils;
 import org.talend.daikon.avro.LogicalTypeUtils;
+import org.talend.daikon.avro.NameUtil;
 import org.talend.daikon.avro.SchemaConstants;
 
 /**
@@ -133,6 +137,7 @@ public class IncomingSchemaEnforcer {
         }
     }
 
+    //TODO remove this method as no place use it now in javajet
     /**
      * Take all of the parameters from the dynamic metadata and adapt it to a field for the runtime Schema.
      *
@@ -144,6 +149,62 @@ public class IncomingSchemaEnforcer {
         addDynamicField(name, type, null, format, description, isNullable);
     }
 
+    private Set<String> existNames;
+
+    private Map<String, String> unvalidName2ValidName;
+
+    private int index = 0;
+
+    /**
+     * Recreates dynamic field from parameters retrieved from DI dynamic metadata
+     *
+     * @param name dynamic field name
+     * @param diType di column type
+     * @param logicalType dynamic field logical type; could be null
+     * @param fieldPattern dynamic field date format
+     * @param description dynamic field description
+     * @param isNullable defines whether dynamic field may contain <code>null</code> value
+     * @param isKey defines whether dynamic field is key field
+     */
+    public void addDynamicField(String name, String diType, String logicalType, String fieldPattern, String description,
+            boolean isNullable, boolean isKey) {
+        if (!needsInitDynamicColumns())
+            return;
+        Schema fieldSchema = diToAvro(diType, logicalType);
+
+        if (isNullable) {
+            fieldSchema = SchemaBuilder.nullable().type(fieldSchema);
+        }
+
+        Schema.Field field;
+        try {
+            field = new Schema.Field(name, fieldSchema, description, (Object) null);
+        } catch (SchemaParseException e) {
+            //if the name contains special char which can't pass avro name check like $ and #, 
+            //but uniode like Japanese which can pass too though that is not expected
+            if (existNames == null) {
+                existNames = new HashSet<>();
+                unvalidName2ValidName = new HashMap<>();
+            }
+
+            String validName = NameUtil.correct(name, index++, existNames);
+            existNames.add(validName);
+            unvalidName2ValidName.put(name, validName);
+
+            field = new Schema.Field(validName, fieldSchema, description, (Object) null);
+            field.addProp(SchemaConstants.TALEND_COLUMN_DB_COLUMN_NAME, name);
+        }
+
+        // Set pattern for date type
+        if ("id_Date".equals(diType) && fieldPattern != null) {
+            field.addProp(SchemaConstants.TALEND_COLUMN_PATTERN, fieldPattern);
+        }
+        if (isKey) {
+            field.addProp(SchemaConstants.TALEND_COLUMN_IS_KEY, "true");
+        }
+        dynamicFields.add(field);
+    }
+
     /**
      * Recreates dynamic field from parameters retrieved from DI dynamic metadata
      *
@@ -154,21 +215,10 @@ public class IncomingSchemaEnforcer {
      * @param description dynamic field description
      * @param isNullable defines whether dynamic field may contain <code>null</code> value
      */
+    @Deprecated
     public void addDynamicField(String name, String diType, String logicalType, String fieldPattern, String description,
             boolean isNullable) {
-        if (!needsInitDynamicColumns())
-            return;
-        Schema fieldSchema = diToAvro(diType, logicalType);
-
-        if (isNullable) {
-            fieldSchema = SchemaBuilder.nullable().type(fieldSchema);
-        }
-        Schema.Field field = new Schema.Field(name, fieldSchema, description, (Object) null);
-        // Set pattern for date type
-        if ("id_Date".equals(diType) && fieldPattern != null) {
-            field.addProp(SchemaConstants.TALEND_COLUMN_PATTERN, fieldPattern);
-        }
-        dynamicFields.add(field);
+        addDynamicField(name, diType, logicalType, fieldPattern, description, isNullable, false);
     }
 
     public void addIncomingNodeField(String name, String className) {
@@ -250,6 +300,8 @@ public class IncomingSchemaEnforcer {
             fieldSchema = AvroUtils._decimal();
         } else if ("id_Date".equals(diType)) {
             fieldSchema = AvroUtils._date();
+        } else if ("id_byte[]".equals(diType)) {
+            fieldSchema = AvroUtils._bytes();
         } else {
             throw new UnsupportedOperationException("Unrecognized type " + diType);
         }
@@ -369,6 +421,9 @@ public class IncomingSchemaEnforcer {
         return designSchema;
     }
 
+    //here we do special process for dynamic input name, but in fact, 
+    //we have issue which support Japanese char or special char as label for basic talend column too,
+    //so not only dynamic columns may have special name, but also basic may have, but here, we don't consider that, that's TODO
     /**
      * Converts DI data value to Avro format and put it into record by field name
      *
@@ -376,9 +431,16 @@ public class IncomingSchemaEnforcer {
      * @param diValue data value
      */
     public void put(String name, Object diValue) {
+        if (unvalidName2ValidName != null) {
+            String validName = unvalidName2ValidName.get(name);
+            if (validName != null) {
+                name = validName;
+            }
+        }
         put(columnToFieldIndex.get(name), diValue);
     }
 
+    //TODO make it private, no place to call it except current class?
     /**
      * Converts DI data value to Avro format and put it into record by field index
      *

main/plugins/org.talend.designer.components.libs/libs_src/talend-codegen-utils/src/test/java/org/talend/codegen/enforcer/IncomingSchemaEnforcerTest.java

@@ -52,6 +52,8 @@ public class IncomingSchemaEnforcerTest {
      */
     private IndexedRecord componentRecord;
 
+    private IndexedRecord componentRecordWithSpecialName;
+
     @Rule
     public ExpectedException thrown = ExpectedException.none();
 
@@ -72,9 +74,29 @@ public class IncomingSchemaEnforcerTest {
         componentRecord.put(3, true);
         componentRecord.put(4, "Main Street");
         componentRecord.put(5, "This is a record with six columns.");
+
+        Schema componentSchemaWithSpecialName = SchemaBuilder.builder().record("Record").fields() //
+                .name("id").type().intType().noDefault() //
+                .name("name").type().stringType().noDefault() //
+                .name("age").type().intType().noDefault() //
+                .name("性别").type().booleanType().noDefault() //why this don't store the origin name, as it can pass the avro name check, it's a avro bug
+                .name("address_").prop(SchemaConstants.TALEND_COLUMN_DB_COLUMN_NAME, "address#").type().stringType().noDefault() //
+                .name("comment_").prop(SchemaConstants.TALEND_COLUMN_DB_COLUMN_NAME, "comment$").type().stringType().noDefault() //
+                .endRecord();
+        componentRecordWithSpecialName = new GenericData.Record(componentSchemaWithSpecialName);
+        componentRecordWithSpecialName.put(0, 1);
+        componentRecordWithSpecialName.put(1, "User");
+        componentRecordWithSpecialName.put(2, 100);
+        componentRecordWithSpecialName.put(3, true);
+        componentRecordWithSpecialName.put(4, "Main Street");
+        componentRecordWithSpecialName.put(5, "This is a record with six columns.");
     }
 
     private void checkEnforcerWithComponentRecordData(IncomingSchemaEnforcer enforcer) {
+        checkEnforcerWithComponentRecordData(enforcer, false);
+    }
+
+    private void checkEnforcerWithComponentRecordData(IncomingSchemaEnforcer enforcer, boolean specialName) {
         // The enforcer must be ready to receive values.
         assertThat(enforcer.needsInitDynamicColumns(), is(false));
 
@@ -88,15 +110,25 @@ public class IncomingSchemaEnforcerTest {
         IndexedRecord adapted = enforcer.createIndexedRecord();
 
         // Ensure that the result is the same as the expected component record.
-        assertThat(adapted, is(componentRecord));
+        if (specialName) {
+            assertThat(adapted, is(componentRecordWithSpecialName));
+        } else {
+            assertThat(adapted, is(componentRecord));
+        }
 
         // Ensure that we create a new instance when we give it another value.
         enforcer.put("id", 2);
         enforcer.put("name", "User2");
         enforcer.put("age", 200);
-        enforcer.put("valid", false);
-        enforcer.put("address", "2 Main Street");
-        enforcer.put("comment", "2 This is a record with six columns.");
+        if (specialName) {
+            enforcer.put("性别", false);
+            enforcer.put("address#", "2 Main Street");
+            enforcer.put("comment$", "2 This is a record with six columns.");
+        } else {
+            enforcer.put("valid", false);
+            enforcer.put("address", "2 Main Street");
+            enforcer.put("comment", "2 This is a record with six columns.");
+        }
         IndexedRecord adapted2 = enforcer.createIndexedRecord();
 
         // It should have the same schema, but not be the same instance.
@@ -392,6 +424,39 @@ public class IncomingSchemaEnforcerTest {
         checkEnforcerWithComponentRecordData(enforcer);
     }
 
+    @Test
+    public void testDynamicColumnWithSpecialName() {
+        Schema designSchema = SchemaBuilder.builder().record("Record") //
+                .prop(DiSchemaConstants.TALEND6_DYNAMIC_COLUMN_POSITION, "3") //
+                .prop(SchemaConstants.INCLUDE_ALL_FIELDS, "true") //
+                .fields() //
+                .name("id").type().intType().noDefault() //
+                .name("name").type().stringType().noDefault() //
+                .name("age").type().intType().noDefault() //
+                .endRecord();
+
+        IncomingSchemaEnforcer enforcer = new IncomingSchemaEnforcer(designSchema);
+
+        // The enforcer isn't usable yet.
+        assertThat(enforcer.getDesignSchema(), is(designSchema));
+        assertFalse(enforcer.areDynamicFieldsInitialized());
+        assertThat(enforcer.getRuntimeSchema(), nullValue());
+
+        enforcer.addDynamicField("性别", "id_Boolean", null, null, null, false, false);
+        enforcer.addDynamicField("address#", "id_String", null, null, null, false, false);
+        enforcer.addDynamicField("comment$", "id_String", null, null, null, false, false);
+        assertFalse(enforcer.areDynamicFieldsInitialized());
+        enforcer.createRuntimeSchema();
+        assertTrue(enforcer.areDynamicFieldsInitialized());
+
+        // Check the run-time schema was created.
+        assertThat(enforcer.getDesignSchema(), is(designSchema));
+        assertThat(enforcer.getRuntimeSchema(), not(nullValue()));
+
+        // Put values into the enforcer and get them as an IndexedRecord.
+        checkEnforcerWithComponentRecordData(enforcer, true);
+    }
+
     @Test
     public void testTypeConversion_toDate() {
         // The expected schema after enforcement.
@@ -699,6 +764,28 @@ public class IncomingSchemaEnforcerTest {
         assertThat(record.get(1), is((Object) new Date(1234567891011L)));
     }
 
+    /**
+     * Checks key field setting
+     */
+    @Test
+    public void testAddDynamicFieldKey() {
+        Schema expectedRuntimeSchema = SchemaBuilder.builder().record("Record").fields().name("id")
+                .prop(SchemaConstants.TALEND_COLUMN_IS_KEY, "true").type().intType().noDefault().endRecord();
+
+        Schema designSchema = SchemaBuilder.builder().record("Record").prop(SchemaConstants.INCLUDE_ALL_FIELDS, "true")
+                .prop(DiSchemaConstants.TALEND6_DYNAMIC_COLUMN_POSITION, "0").fields().endRecord();
+
+        IncomingSchemaEnforcer enforcer = new IncomingSchemaEnforcer(designSchema);
+
+        enforcer.addDynamicField("id", "id_Integer", null, null, null, false, true);
+
+        enforcer.createRuntimeSchema();
+        assertTrue(enforcer.areDynamicFieldsInitialized());
+
+        Schema actualRuntimeSchema = enforcer.getRuntimeSchema();
+        assertEquals(expectedRuntimeSchema, actualRuntimeSchema);
+    }
+
     /**
      * Checks {@link IncomingSchemaEnforcer#put()} converts string value to date according pattern specified in dynamic field
      * TODO (iv.gonchar): this is incorrect behavior, because avro record should not contain java.util.Date value. It should store

main/plugins/org.talend.designer.components.libs/libs_src/talend-httputil/pom.xml

@@ -4,7 +4,7 @@
 	<groupId>org.talend</groupId>
 	<artifactId>talend-httputil</artifactId>
 	<name>talend-httputil</name>
-	<version>1.0.5</version>
+	<version>1.0.6</version>
 
 	<properties>
 		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>
@@ -20,7 +20,7 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.10.1</version>
+            <version>2.11.4</version>
         </dependency>
         
         <dependency>

main/plugins/org.talend.designer.components.libs/libs_src/talend-mscrm/pom.xml

@@ -10,7 +10,6 @@
 	
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-		<cxf.version>3.1.1</cxf.version>
 		<odata.version>4.3.0</odata.version>
         <slf4j.version>1.7.12</slf4j.version>
 		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/pom.xml

@@ -0,0 +1,98 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>org.talend.components</groupId>
+	<artifactId>talend-parquet</artifactId>
+	<version>1.3</version>
+
+	<properties>
+		<parquet.version>1.10.1</parquet.version>
+		<hadoop.version>3.2.2</hadoop.version>
+		<jodd.version>6.0.1</jodd.version>
+		<hamcrest.version>1.3</hamcrest.version>
+		<junit.version>4.13.2</junit.version>
+		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>
+	</properties>
+	
+	<distributionManagement>
+		<snapshotRepository>
+			<id>talend_nexus_deployment</id>
+			<url>${talend.nexus.url}/nexus/content/repositories/TalendOpenSourceSnapshot/</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</snapshotRepository>
+		<repository>
+			<id>talend_nexus_deployment</id>
+			<url>${talend.nexus.url}/nexus/content/repositories/TalendOpenSourceRelease/</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+			<releases>
+				<enabled>true</enabled>
+			</releases>
+		</repository>
+	</distributionManagement>
+	
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.parquet</groupId>
+			<artifactId>parquet-hadoop</artifactId>
+			<version>${parquet.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.hadoop</groupId>
+			<artifactId>hadoop-client</artifactId>
+			<version>${hadoop.version}</version>
+			<scope>provided</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>slf4j-log4j12</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+		    <groupId>org.jodd</groupId>
+		    <artifactId>jodd-util</artifactId>
+		    <version>${jodd.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<version>${junit.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.hamcrest</groupId>
+			<artifactId>hamcrest-library</artifactId>
+			<version>${hamcrest.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.hamcrest</groupId>
+			<artifactId>hamcrest-library</artifactId>
+			<version>${hamcrest.version}</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<version>3.8.0</version>
+				<configuration>
+					<source>1.8</source>
+					<target>1.8</target>
+				</configuration>
+			</plugin>
+
+		</plugins>
+	</build>
+</project>

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/Group.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data;
+
+import org.talend.parquet.data.simple.NanoTime;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+abstract public class Group extends GroupValueSource {
+	private static final Logger LOG = LoggerFactory.getLogger(Group.class);
+
+	public void add(String field, int value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, long value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, float value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, double value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, String value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, NanoTime value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, boolean value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, Binary value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public void add(String field, Group value) {
+		add(getType().getFieldIndex(field), value);
+	}
+
+	public Group addGroup(String field) {
+		if (LOG.isDebugEnabled()) {
+			LOG.debug("add group {} to {}", field, getType().getName());
+		}
+		return addGroup(getType().getFieldIndex(field));
+	}
+
+	@Override
+	public Group getGroup(String field, int index) {
+		return getGroup(getType().getFieldIndex(field), index);
+	}
+
+	abstract public void add(int fieldIndex, int value);
+
+	abstract public void add(int fieldIndex, long value);
+
+	abstract public void add(int fieldIndex, String value);
+
+	abstract public void add(int fieldIndex, boolean value);
+
+	abstract public void add(int fieldIndex, NanoTime value);
+
+	abstract public void add(int fieldIndex, Binary value);
+
+	abstract public void add(int fieldIndex, float value);
+
+	abstract public void add(int fieldIndex, double value);
+
+	abstract public void add(int fieldIndex, Group value);
+
+	abstract public Group addGroup(int fieldIndex);
+
+	@Override
+	abstract public Group getGroup(int fieldIndex, int index);
+
+	public Group asGroup() {
+		return this;
+	}
+
+	public Group append(String fieldName, int value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	public Group append(String fieldName, float value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	public Group append(String fieldName, double value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	public Group append(String fieldName, long value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	public Group append(String fieldName, NanoTime value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	public Group append(String fieldName, String value) {
+		add(fieldName, Binary.fromString(value));
+		return this;
+	}
+
+	public Group append(String fieldName, boolean value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	public Group append(String fieldName, Binary value) {
+		add(fieldName, value);
+		return this;
+	}
+
+	abstract public void writeValue(int field, int index, RecordConsumer recordConsumer);
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/GroupFactory.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data;
+
+abstract public class GroupFactory {
+
+	abstract public Group newGroup();
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/GroupValueSource.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.GroupType;
+
+abstract public class GroupValueSource {
+
+	public int getFieldRepetitionCount(String field) {
+		return getFieldRepetitionCount(getType().getFieldIndex(field));
+	}
+
+	public GroupValueSource getGroup(String field, int index) {
+		return getGroup(getType().getFieldIndex(field), index);
+	}
+
+	public String getString(String field, int index) {
+		return getString(getType().getFieldIndex(field), index);
+	}
+
+	public int getInteger(String field, int index) {
+		return getInteger(getType().getFieldIndex(field), index);
+	}
+
+	public long getLong(String field, int index) {
+		return getLong(getType().getFieldIndex(field), index);
+	}
+
+	public double getDouble(String field, int index) {
+		return getDouble(getType().getFieldIndex(field), index);
+	}
+
+	public float getFloat(String field, int index) {
+		return getFloat(getType().getFieldIndex(field), index);
+	}
+
+	public boolean getBoolean(String field, int index) {
+		return getBoolean(getType().getFieldIndex(field), index);
+	}
+
+	public Binary getBinary(String field, int index) {
+		return getBinary(getType().getFieldIndex(field), index);
+	}
+
+	public Binary getInt96(String field, int index) {
+		return getInt96(getType().getFieldIndex(field), index);
+	}
+
+	abstract public int getFieldRepetitionCount(int fieldIndex);
+
+	abstract public GroupValueSource getGroup(int fieldIndex, int index);
+
+	abstract public String getString(int fieldIndex, int index);
+
+	abstract public Integer getInteger(int fieldIndex, int index);
+
+	abstract public Long getLong(int fieldIndex, int index);
+
+	abstract public Double getDouble(int fieldIndex, int index);
+
+	abstract public Float getFloat(int fieldIndex, int index);
+
+	abstract public Boolean getBoolean(int fieldIndex, int index);
+
+	abstract public Binary getBinary(int fieldIndex, int index);
+
+	abstract public Binary getInt96(int fieldIndex, int index);
+
+	abstract public String getValueToString(int fieldIndex, int index);
+
+	abstract public GroupType getType();
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/GroupWriter.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data;
+
+import org.apache.parquet.io.api.RecordConsumer;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.Type;
+
+public class GroupWriter {
+
+	private final RecordConsumer recordConsumer;
+	private final GroupType schema;
+
+	public GroupWriter(RecordConsumer recordConsumer, GroupType schema) {
+		this.recordConsumer = recordConsumer;
+		this.schema = schema;
+	}
+
+	public void write(Group group) {
+		recordConsumer.startMessage();
+		writeGroup(group, schema);
+		recordConsumer.endMessage();
+	}
+
+	private void writeGroup(Group group, GroupType type) {
+		int fieldCount = type.getFieldCount();
+		for (int field = 0; field < fieldCount; ++field) {
+			int valueCount = group.getFieldRepetitionCount(field);
+			if (valueCount > 0) {
+				Type fieldType = type.getType(field);
+				String fieldName = fieldType.getName();
+				recordConsumer.startField(fieldName, field);
+				for (int index = 0; index < valueCount; ++index) {
+					if (fieldType.isPrimitive()) {
+						group.writeValue(field, index, recordConsumer);
+					} else {
+						recordConsumer.startGroup();
+						writeGroup(group.getGroup(field, index), fieldType.asGroupType());
+						recordConsumer.endGroup();
+					}
+				}
+				recordConsumer.endField(fieldName, field);
+			}
+		}
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/BinaryValue.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class BinaryValue extends Primitive {
+
+	private final Binary binary;
+
+	public BinaryValue(Binary binary) {
+		this.binary = binary;
+	}
+
+	@Override
+	public Binary getBinary() {
+		return binary;
+	}
+
+	@Override
+	public String getString() {
+		return binary.toStringUsingUTF8();
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addBinary(binary);
+	}
+
+	@Override
+	public String toString() {
+		return getString();
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/BooleanValue.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class BooleanValue extends Primitive {
+
+	private final boolean bool;
+
+	public BooleanValue(boolean bool) {
+		this.bool = bool;
+	}
+
+	@Override
+	public String toString() {
+		return String.valueOf(bool);
+	}
+
+	@Override
+	public boolean getBoolean() {
+		return bool;
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addBoolean(bool);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/DoubleValue.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class DoubleValue extends Primitive {
+
+	private final double value;
+
+	public DoubleValue(double value) {
+		this.value = value;
+	}
+
+	@Override
+	public double getDouble() {
+		return value;
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addDouble(value);
+	}
+
+	@Override
+	public String toString() {
+		return String.valueOf(value);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/FloatValue.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class FloatValue extends Primitive {
+
+	private final float value;
+
+	public FloatValue(float value) {
+		this.value = value;
+	}
+
+	@Override
+	public float getFloat() {
+		return value;
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addFloat(value);
+	}
+
+	@Override
+	public String toString() {
+		return String.valueOf(value);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/Int96Value.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class Int96Value extends Primitive {
+
+	private final Binary value;
+
+	public Int96Value(Binary value) {
+		this.value = value;
+	}
+
+	@Override
+	public Binary getInt96() {
+		return value;
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addBinary(value);
+	}
+
+	@Override
+	public String toString() {
+		return "Int96Value{" + String.valueOf(value) + "}";
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/IntegerValue.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class IntegerValue extends Primitive {
+
+	private final int value;
+
+	public IntegerValue(int value) {
+		this.value = value;
+	}
+
+	@Override
+	public String toString() {
+		return String.valueOf(value);
+	}
+
+	@Override
+	public int getInteger() {
+		return value;
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addInteger(value);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/LongValue.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class LongValue extends Primitive {
+
+	private final long value;
+
+	public LongValue(long value) {
+		this.value = value;
+	}
+
+	@Override
+	public String toString() {
+		return String.valueOf(value);
+	}
+
+	@Override
+	public long getLong() {
+		return value;
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addLong(value);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/NanoTime.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+
+public class NanoTime extends Primitive {
+	private final int julianDay;
+	private final long timeOfDayNanos;
+
+	public static NanoTime fromBinary(Binary bytes) {
+		Preconditions.checkArgument(bytes.length() == 12, "Must be 12 bytes");
+		ByteBuffer buf = bytes.toByteBuffer();
+		buf.order(ByteOrder.LITTLE_ENDIAN);
+		long timeOfDayNanos = buf.getLong();
+		int julianDay = buf.getInt();
+		return new NanoTime(julianDay, timeOfDayNanos);
+	}
+
+	public static NanoTime fromInt96(Int96Value int96) {
+		ByteBuffer buf = int96.getInt96().toByteBuffer();
+		return new NanoTime(buf.getInt(), buf.getLong());
+	}
+
+	public NanoTime(int julianDay, long timeOfDayNanos) {
+		this.julianDay = julianDay;
+		this.timeOfDayNanos = timeOfDayNanos;
+	}
+
+	public int getJulianDay() {
+		return julianDay;
+	}
+
+	public long getTimeOfDayNanos() {
+		return timeOfDayNanos;
+	}
+
+	public Binary toBinary() {
+		ByteBuffer buf = ByteBuffer.allocate(12);
+		buf.order(ByteOrder.LITTLE_ENDIAN);
+		buf.putLong(timeOfDayNanos);
+		buf.putInt(julianDay);
+		buf.flip();
+		return Binary.fromConstantByteBuffer(buf);
+	}
+
+	public Int96Value toInt96() {
+		return new Int96Value(toBinary());
+	}
+
+	@Override
+	public void writeValue(RecordConsumer recordConsumer) {
+		recordConsumer.addBinary(toBinary());
+	}
+
+	@Override
+	public String toString() {
+		return "NanoTime{julianDay=" + julianDay + ", timeOfDayNanos=" + timeOfDayNanos + "}";
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/Primitive.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+
+public abstract class Primitive {
+
+	public String getString() {
+		throw new UnsupportedOperationException();
+	}
+
+	public int getInteger() {
+		throw new UnsupportedOperationException();
+	}
+
+	public long getLong() {
+		throw new UnsupportedOperationException();
+	}
+
+	public boolean getBoolean() {
+		throw new UnsupportedOperationException();
+	}
+
+	public Binary getBinary() {
+		throw new UnsupportedOperationException();
+	}
+
+	public Binary getInt96() {
+		throw new UnsupportedOperationException();
+	}
+
+	public float getFloat() {
+		throw new UnsupportedOperationException();
+	}
+
+	public double getDouble() {
+		throw new UnsupportedOperationException();
+	}
+
+	abstract public void writeValue(RecordConsumer recordConsumer);
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/SimpleGroup.java

@@ -0,0 +1,274 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.talend.parquet.data.Group;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.Type;
+
+public class SimpleGroup extends Group {
+
+	private final GroupType schema;
+	private final List<Object>[] data;
+
+	@SuppressWarnings("unchecked")
+	public SimpleGroup(GroupType schema) {
+		this.schema = schema;
+		this.data = new List[schema.getFields().size()];
+		for (int i = 0; i < schema.getFieldCount(); i++) {
+			this.data[i] = new ArrayList<>();
+		}
+	}
+
+	@Override
+	public String toString() {
+		return toString("");
+	}
+
+	private StringBuilder appendToString(StringBuilder builder, String indent) {
+		int i = 0;
+		for (Type field : schema.getFields()) {
+			String name = field.getName();
+			List<Object> values = data[i];
+			++i;
+			if (values != null && !values.isEmpty()) {
+				for (Object value : values) {
+					builder.append(indent).append(name);
+					if (value == null) {
+						builder.append(": NULL\n");
+					} else if (value instanceof Group) {
+						builder.append('\n');
+						((SimpleGroup) value).appendToString(builder, indent + "  ");
+					} else {
+						builder.append(": ").append(value.toString()).append('\n');
+					}
+				}
+			}
+		}
+		return builder;
+	}
+
+	public String toString(String indent) {
+		StringBuilder builder = new StringBuilder();
+		appendToString(builder, indent);
+		return builder.toString();
+	}
+
+	@Override
+	public Group addGroup(int fieldIndex) {
+		SimpleGroup g = new SimpleGroup(schema.getType(fieldIndex).asGroupType());
+		add(fieldIndex, g);
+		return g;
+	}
+
+	@Override
+	public Group getGroup(int fieldIndex, int index) {
+		return (Group) getValue(fieldIndex, index);
+	}
+
+	private Object getValue(int fieldIndex, int index) {
+		List<Object> list;
+		try {
+			list = data[fieldIndex];
+		} catch (IndexOutOfBoundsException e) {
+			throw new RuntimeException(
+					"not found " + fieldIndex + "(" + schema.getFieldName(fieldIndex) + ") in group:\n" + this);
+		}
+		try {
+			if(list == null || list.isEmpty()) {
+				return null;
+			}
+			return list.get(index);
+		} catch (IndexOutOfBoundsException e) {
+			throw new RuntimeException("not found " + fieldIndex + "(" + schema.getFieldName(fieldIndex)
+					+ ") element number " + index + " in group:\n" + this);
+		}
+	}
+
+	private void add(int fieldIndex, Primitive value) {
+		Type type = schema.getType(fieldIndex);
+		List<Object> list = data[fieldIndex];
+		if (!type.isRepetition(Type.Repetition.REPEATED) && !list.isEmpty()) {
+			throw new IllegalStateException(
+					"field " + fieldIndex + " (" + type.getName() + ") can not have more than one value: " + list);
+		}
+		list.add(value);
+	}
+
+	@Override
+	public int getFieldRepetitionCount(int fieldIndex) {
+		List<Object> list = data[fieldIndex];
+		return list == null ? 0 : list.size();
+	}
+
+	@Override
+	public String getValueToString(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return String.valueOf(value);
+	}
+
+	@Override
+	public String getString(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((BinaryValue) value).getString();
+	}
+
+	@Override
+	public Integer getInteger(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((IntegerValue)value).getInteger();
+	}
+
+	@Override
+	public Long getLong(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((LongValue)value).getLong();
+	}
+
+	@Override
+	public Double getDouble(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((DoubleValue)value).getDouble();
+	}
+
+	@Override
+	public Float getFloat(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((FloatValue)value).getFloat();
+	}
+
+	@Override
+	public Boolean getBoolean(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((BooleanValue) value).getBoolean();
+	}
+
+	@Override
+	public Binary getBinary(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((BinaryValue) value).getBinary();
+	}
+
+	public NanoTime getTimeNanos(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return NanoTime.fromInt96((Int96Value) value);
+	}
+
+	@Override
+	public Binary getInt96(int fieldIndex, int index) {
+		Object value = getValue(fieldIndex, index);
+		if(value == null) {
+			return null;
+		}
+		return ((Int96Value) value).getInt96();
+	}
+
+	@Override
+	public void add(int fieldIndex, int value) {
+		add(fieldIndex, new IntegerValue(value));
+	}
+
+	@Override
+	public void add(int fieldIndex, long value) {
+		add(fieldIndex, new LongValue(value));
+	}
+
+	@Override
+	public void add(int fieldIndex, String value) {
+		add(fieldIndex, new BinaryValue(Binary.fromString(value)));
+	}
+
+	@Override
+	public void add(int fieldIndex, NanoTime value) {
+		add(fieldIndex, value.toInt96());
+	}
+
+	@Override
+	public void add(int fieldIndex, boolean value) {
+		add(fieldIndex, new BooleanValue(value));
+	}
+
+	@Override
+	public void add(int fieldIndex, Binary value) {
+		switch (getType().getType(fieldIndex).asPrimitiveType().getPrimitiveTypeName()) {
+		case BINARY:
+		case FIXED_LEN_BYTE_ARRAY:
+			add(fieldIndex, new BinaryValue(value));
+			break;
+		case INT96:
+			add(fieldIndex, new Int96Value(value));
+			break;
+		default:
+			throw new UnsupportedOperationException(
+					getType().asPrimitiveType().getName() + " not supported for Binary");
+		}
+	}
+
+	@Override
+	public void add(int fieldIndex, float value) {
+		add(fieldIndex, new FloatValue(value));
+	}
+
+	@Override
+	public void add(int fieldIndex, double value) {
+		add(fieldIndex, new DoubleValue(value));
+	}
+
+	@Override
+	public void add(int fieldIndex, Group value) {
+		data[fieldIndex].add(value);
+	}
+
+	@Override
+	public GroupType getType() {
+		return schema;
+	}
+
+	@Override
+	public void writeValue(int field, int index, RecordConsumer recordConsumer) {
+		((Primitive) getValue(field, index)).writeValue(recordConsumer);
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/SimpleGroupFactory.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple;
+
+import org.talend.parquet.data.Group;
+import org.talend.parquet.data.GroupFactory;
+import org.apache.parquet.schema.MessageType;
+
+public class SimpleGroupFactory extends GroupFactory {
+
+	private final MessageType schema;
+
+	public SimpleGroupFactory(MessageType schema) {
+		this.schema = schema;
+	}
+
+	@Override
+	public Group newGroup() {
+		return new SimpleGroup(schema);
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/convert/GroupRecordConverter.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple.convert;
+
+import org.talend.parquet.data.Group;
+import org.talend.parquet.data.simple.SimpleGroupFactory;
+import org.apache.parquet.io.api.GroupConverter;
+import org.apache.parquet.io.api.RecordMaterializer;
+import org.apache.parquet.schema.MessageType;
+
+public class GroupRecordConverter extends RecordMaterializer<Group> {
+
+	private final SimpleGroupFactory simpleGroupFactory;
+
+	private SimpleGroupConverter root;
+
+	public GroupRecordConverter(MessageType schema) {
+		this.simpleGroupFactory = new SimpleGroupFactory(schema);
+		this.root = new SimpleGroupConverter(null, 0, schema) {
+			@Override
+			public void start() {
+				this.current = simpleGroupFactory.newGroup();
+			}
+
+			@Override
+			public void end() {
+			}
+		};
+	}
+
+	@Override
+	public Group getCurrentRecord() {
+		return root.getCurrentRecord();
+	}
+
+	@Override
+	public GroupConverter getRootConverter() {
+		return root;
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/convert/SimpleGroupConverter.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple.convert;
+
+import org.talend.parquet.data.Group;
+import org.apache.parquet.io.api.Converter;
+import org.apache.parquet.io.api.GroupConverter;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.Type;
+
+class SimpleGroupConverter extends GroupConverter {
+	private final SimpleGroupConverter parent;
+	private final int index;
+	protected Group current;
+	private Converter[] converters;
+
+	SimpleGroupConverter(SimpleGroupConverter parent, int index, GroupType schema) {
+		this.parent = parent;
+		this.index = index;
+
+		converters = new Converter[schema.getFieldCount()];
+
+		for (int i = 0; i < converters.length; i++) {
+			final Type type = schema.getType(i);
+			if (type.isPrimitive()) {
+				converters[i] = new SimplePrimitiveConverter(this, i);
+			} else {
+				converters[i] = new SimpleGroupConverter(this, i, type.asGroupType());
+			}
+
+		}
+	}
+
+	@Override
+	public void start() {
+		current = parent.getCurrentRecord().addGroup(index);
+	}
+
+	@Override
+	public Converter getConverter(int fieldIndex) {
+		return converters[fieldIndex];
+	}
+
+	@Override
+	public void end() {
+	}
+
+	public Group getCurrentRecord() {
+		return current;
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/data/simple/convert/SimplePrimitiveConverter.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.data.simple.convert;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.PrimitiveConverter;
+
+class SimplePrimitiveConverter extends PrimitiveConverter {
+
+	private final SimpleGroupConverter parent;
+	private final int index;
+
+	SimplePrimitiveConverter(SimpleGroupConverter parent, int index) {
+		this.parent = parent;
+		this.index = index;
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * 
+	 * @see org.apache.parquet.io.api.PrimitiveConverter#addBinary(Binary)
+	 */
+	@Override
+	public void addBinary(Binary value) {
+		parent.getCurrentRecord().add(index, value);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * 
+	 * @see org.apache.parquet.io.api.PrimitiveConverter#addBoolean(boolean)
+	 */
+	@Override
+	public void addBoolean(boolean value) {
+		parent.getCurrentRecord().add(index, value);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * 
+	 * @see org.apache.parquet.io.api.PrimitiveConverter#addDouble(double)
+	 */
+	@Override
+	public void addDouble(double value) {
+		parent.getCurrentRecord().add(index, value);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * 
+	 * @see org.apache.parquet.io.api.PrimitiveConverter#addFloat(float)
+	 */
+	@Override
+	public void addFloat(float value) {
+		parent.getCurrentRecord().add(index, value);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * 
+	 * @see org.apache.parquet.io.api.PrimitiveConverter#addInt(int)
+	 */
+	@Override
+	public void addInt(int value) {
+		parent.getCurrentRecord().add(index, value);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * 
+	 * @see org.apache.parquet.io.api.PrimitiveConverter#addLong(long)
+	 */
+	@Override
+	public void addLong(long value) {
+		parent.getCurrentRecord().add(index, value);
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/hadoop/TalendGroupReadSupport.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.hadoop;
+
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.hadoop.api.ReadSupport;
+import org.apache.parquet.io.api.RecordMaterializer;
+import org.apache.parquet.schema.MessageType;
+import org.talend.parquet.data.Group;
+import org.talend.parquet.data.simple.convert.GroupRecordConverter;
+
+public class TalendGroupReadSupport extends ReadSupport<Group> {
+
+	@Override
+	public org.apache.parquet.hadoop.api.ReadSupport.ReadContext init(Configuration configuration,
+			Map<String, String> keyValueMetaData, MessageType fileSchema) {
+		String partialSchemaString = configuration.get(ReadSupport.PARQUET_READ_SCHEMA);
+		MessageType requestedProjection = getSchemaForRead(fileSchema, partialSchemaString);
+		return new ReadContext(requestedProjection);
+	}
+
+	@Override
+	public RecordMaterializer<Group> prepareForRead(Configuration configuration, Map<String, String> keyValueMetaData,
+			MessageType fileSchema, org.apache.parquet.hadoop.api.ReadSupport.ReadContext readContext) {
+		return new GroupRecordConverter(readContext.getRequestedSchema());
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/hadoop/TalendGroupWriteSupport.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.hadoop;
+
+import static org.apache.parquet.schema.MessageTypeParser.parseMessageType;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.io.api.RecordConsumer;
+import org.apache.parquet.schema.MessageType;
+import org.talend.parquet.data.Group;
+import org.talend.parquet.data.GroupWriter;
+
+public class TalendGroupWriteSupport extends WriteSupport<Group> {
+
+	public static final String PARQUET_SCHEMA = "parquet.talend.schema";
+
+	public static void setSchema(MessageType schema, Configuration configuration) {
+		configuration.set(PARQUET_SCHEMA, schema.toString());
+	}
+
+	public static MessageType getSchema(Configuration configuration) {
+		return parseMessageType(Objects.requireNonNull(configuration.get(PARQUET_SCHEMA), PARQUET_SCHEMA));
+	}
+
+	private MessageType schema;
+	private GroupWriter groupWriter;
+	private Map<String, String> extraMetaData;
+
+	public TalendGroupWriteSupport() {
+		this(null, new HashMap<String, String>());
+	}
+
+	TalendGroupWriteSupport(MessageType schema) {
+		this(schema, new HashMap<String, String>());
+	}
+
+	TalendGroupWriteSupport(MessageType schema, Map<String, String> extraMetaData) {
+		this.schema = schema;
+		this.extraMetaData = extraMetaData;
+	}
+
+	@Override
+	public String getName() {
+		return "Talend";
+	}
+
+	@Override
+	public org.apache.parquet.hadoop.api.WriteSupport.WriteContext init(Configuration configuration) {
+		// if present, prefer the schema passed to the constructor
+		if (schema == null) {
+			schema = getSchema(configuration);
+		}
+		return new WriteContext(schema, this.extraMetaData);
+	}
+
+	@Override
+	public void prepareForWrite(RecordConsumer recordConsumer) {
+		groupWriter = new GroupWriter(recordConsumer, schema);
+	}
+
+	@Override
+	public void write(Group record) {
+		groupWriter.write(record);
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/hadoop/TalendInputFormat.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.hadoop;
+
+import org.apache.parquet.hadoop.ParquetInputFormat;
+import org.talend.parquet.data.Group;
+
+/**
+ * Example input format to read Parquet files
+ *
+ * This Input format uses a rather inefficient data model but works
+ * independently of higher level abstractions.
+ */
+public class TalendInputFormat extends ParquetInputFormat<Group> {
+
+	public TalendInputFormat() {
+		super(TalendGroupReadSupport.class);
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/hadoop/TalendOutputFormat.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.hadoop;
+
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.parquet.hadoop.ParquetOutputFormat;
+import org.apache.parquet.hadoop.util.ContextUtil;
+import org.apache.parquet.schema.MessageType;
+import org.talend.parquet.data.Group;
+
+/**
+ * An example output format
+ *
+ * must be provided the schema up front
+ * 
+ * @see TalendOutputFormat#setSchema(Job, MessageType)
+ * @see TalendGroupWriteSupport#PARQUET_SCHEMA
+ */
+public class TalendOutputFormat extends ParquetOutputFormat<Group> {
+
+	/**
+	 * set the schema being written to the job conf
+	 * 
+	 * @param job    a job
+	 * @param schema the schema of the data
+	 */
+	public static void setSchema(Job job, MessageType schema) {
+		TalendGroupWriteSupport.setSchema(schema, ContextUtil.getConfiguration(job));
+	}
+
+	/**
+	 * retrieve the schema from the conf
+	 * 
+	 * @param job a job
+	 * @return the schema
+	 */
+	public static MessageType getSchema(Job job) {
+		return TalendGroupWriteSupport.getSchema(ContextUtil.getConfiguration(job));
+	}
+
+	public TalendOutputFormat() {
+		super(new TalendGroupWriteSupport());
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/hadoop/TalendParquetWriter.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright (C) 2006-2021 Talend Inc. - www.talend.com
+ *
+ * Licensed 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.talend.parquet.hadoop;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.OutputFile;
+import org.apache.parquet.schema.MessageType;
+import org.talend.parquet.data.Group;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * An example file writer class.
+ */
+public class TalendParquetWriter extends ParquetWriter<Group> {
+
+	/**
+	 * Creates a Builder for configuring ParquetWriter with the example object
+	 *
+	 * @param file the output file to create
+	 * @return a {@link Builder} to create a {@link ParquetWriter}
+	 */
+	public static Builder builder(Path file) {
+		return new Builder(file);
+	}
+
+	/**
+	 * Creates a Builder for configuring ParquetWriter with the example object
+	 *
+	 * @param file the output file to create
+	 * @return a {@link Builder} to create a {@link ParquetWriter}
+	 */
+	public static Builder builder(OutputFile file) {
+		return new Builder(file);
+	}
+
+	/**
+	 * Create a new {@link TalendParquetWriter}.
+	 *
+	 * @param file                 The file name to write to.
+	 * @param writeSupport         The schema to write with.
+	 * @param compressionCodecName Compression code to use, or
+	 *                             CompressionCodecName.UNCOMPRESSED
+	 * @param blockSize            the block size threshold.
+	 * @param pageSize             See parquet write up. Blocks are subdivided into
+	 *                             pages for alignment and other purposes.
+	 * @param enableDictionary     Whether to use a dictionary to compress columns.
+	 * @param conf                 The Configuration to use.
+	 * @throws IOException
+	 */
+	TalendParquetWriter(Path file, WriteSupport<Group> writeSupport, CompressionCodecName compressionCodecName,
+			int blockSize, int pageSize, boolean enableDictionary, boolean enableValidation,
+			ParquetProperties.WriterVersion writerVersion, Configuration conf) throws IOException {
+		super(file, writeSupport, compressionCodecName, blockSize, pageSize, pageSize, enableDictionary,
+				enableValidation, writerVersion, conf);
+	}
+
+	public static class Builder extends ParquetWriter.Builder<Group, Builder> {
+		private MessageType type = null;
+		private Map<String, String> extraMetaData = new HashMap<String, String>();
+
+		private Builder(Path file) {
+			super(file);
+		}
+
+		private Builder(OutputFile file) {
+			super(file);
+		}
+
+		public Builder withType(MessageType type) {
+			this.type = type;
+			return this;
+		}
+
+		public Builder withExtraMetaData(Map<String, String> extraMetaData) {
+			this.extraMetaData = extraMetaData;
+			return this;
+		}
+
+		@Override
+		protected Builder self() {
+			return this;
+		}
+
+		@Override
+		protected WriteSupport<Group> getWriteSupport(Configuration conf) {
+			return new TalendGroupWriteSupport(type, extraMetaData);
+		}
+
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/utils/NanoTimeUtils.java

@@ -0,0 +1,131 @@
+package org.talend.parquet.utils;
+
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.Calendar;
+import java.util.GregorianCalendar;
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+import org.talend.parquet.data.simple.NanoTime;
+
+import jodd.time.JulianDate;
+
+public class NanoTimeUtils {
+
+	/**
+	 * Number of days between Julian day epoch (January 1, 4713 BC) and Unix day
+	 * epoch (January 1, 1970). The value of this constant is {@value}.
+	 */
+	public static final long JULIAN_EPOCH_OFFSET_DAYS = 2440588;
+	private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+	private static final long NANOS_PER_MILLISECOND = TimeUnit.MILLISECONDS.toNanos(1);
+
+	static final long NANOS_PER_HOUR = java.util.concurrent.TimeUnit.HOURS.toNanos(1);
+	static final long NANOS_PER_MINUTE = java.util.concurrent.TimeUnit.MINUTES.toNanos(1);
+	static final long NANOS_PER_SECOND = java.util.concurrent.TimeUnit.SECONDS.toNanos(1);
+	static final long NANOS_PER_DAY = java.util.concurrent.TimeUnit.DAYS.toNanos(1);
+
+	private static final ThreadLocal<java.util.Calendar> parquetGMTCalendar = new ThreadLocal<Calendar>();
+	private static final ThreadLocal<Calendar> parquetLocalCalendar = new ThreadLocal<Calendar>();
+
+	private static Calendar getGMTCalendar() {
+		// Calendar.getInstance calculates the current-time needlessly, so cache
+		// an instance.
+		if (parquetGMTCalendar.get() == null) {
+			parquetGMTCalendar.set(Calendar.getInstance(TimeZone.getTimeZone("GMT")));
+		}
+		return parquetGMTCalendar.get();
+	}
+
+	private static Calendar getLocalCalendar() {
+		if (parquetLocalCalendar.get() == null) {
+			parquetLocalCalendar.set(Calendar.getInstance());
+		}
+		return parquetLocalCalendar.get();
+	}
+
+	private static Calendar getCalendar(boolean skipConversion) {
+		Calendar calendar = skipConversion ? getLocalCalendar() : getGMTCalendar();
+		calendar.clear();
+		return calendar;
+	}
+
+	/**
+	 * Converts a timestamp to NanoTime.
+	 */
+	public static NanoTime getNanoTime(Timestamp ts, boolean skipConversion) {
+
+		Calendar calendar = getCalendar(skipConversion);
+		calendar.setTimeInMillis(ts.getTime());
+		int year = calendar.get(Calendar.YEAR);
+		if (calendar.get(Calendar.ERA) == GregorianCalendar.BC) {
+			year = 1 - year;
+		}
+		JulianDate jDateTime;
+		jDateTime = JulianDate.of(year, calendar.get(Calendar.MONTH) + 1, // java calendar index starting at 1.
+				calendar.get(Calendar.DAY_OF_MONTH), 0, 0, 0, 0);
+		int days = jDateTime.getJulianDayNumber();
+
+		long hour = calendar.get(Calendar.HOUR_OF_DAY);
+		long minute = calendar.get(Calendar.MINUTE);
+		long second = calendar.get(Calendar.SECOND);
+		long nanos = ts.getNanos();
+		long nanosOfDay = nanos + NANOS_PER_SECOND * second + NANOS_PER_MINUTE * minute + NANOS_PER_HOUR * hour;
+
+		return new NanoTime(days, nanosOfDay);
+	}
+
+	public static Timestamp getTimestamp(NanoTime nt, boolean skipConversion) {
+		int julianDay = nt.getJulianDay();
+		long nanosOfDay = nt.getTimeOfDayNanos();
+
+		long remainder = nanosOfDay;
+		julianDay += remainder / NANOS_PER_DAY;
+		remainder %= NANOS_PER_DAY;
+		if (remainder < 0) {
+			remainder += NANOS_PER_DAY;
+			julianDay--;
+		}
+
+		JulianDate jDateTime = new JulianDate((double) julianDay);
+		LocalDateTime datetime = jDateTime.toLocalDateTime();
+		Calendar calendar = getCalendar(skipConversion);
+		calendar.set(Calendar.YEAR, datetime.getYear());
+		calendar.set(Calendar.MONTH, datetime.getMonthValue() - 1);
+		calendar.set(Calendar.DAY_OF_MONTH, datetime.getYear());
+
+		int hour = (int) (remainder / (NANOS_PER_HOUR));
+		remainder = remainder % (NANOS_PER_HOUR);
+		int minutes = (int) (remainder / (NANOS_PER_MINUTE));
+		remainder = remainder % (NANOS_PER_MINUTE);
+		int seconds = (int) (remainder / (NANOS_PER_SECOND));
+		long nanos = remainder % NANOS_PER_SECOND;
+
+		calendar.set(Calendar.HOUR_OF_DAY, hour);
+		calendar.set(Calendar.MINUTE, minutes);
+		calendar.set(Calendar.SECOND, seconds);
+		Timestamp ts = new Timestamp(calendar.getTimeInMillis());
+		ts.setNanos((int) nanos);
+		return ts;
+	}
+
+	/**
+	 * Returns timestamp millis from NanoTime type value.
+	 *
+	 * @param nt NanoTime value
+	 * @return timestamp in millis
+	 */
+	public static long getTimestampMillis(NanoTime nt) {
+		long timeOfDayNanos = nt.getTimeOfDayNanos();
+		int julianDay = nt.getJulianDay();
+
+		return (julianDay - JULIAN_EPOCH_OFFSET_DAYS) * MILLIS_IN_DAY + (timeOfDayNanos / NANOS_PER_MILLISECOND);
+	}
+
+	public static Timestamp getTimestamp(NanoTime nt) {
+		Timestamp ts = new Timestamp(getTimestampMillis(nt));
+		ts.setNanos((int) (nt.getTimeOfDayNanos() % 1000000000));
+		return ts;
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/main/java/org/talend/parquet/utils/TalendParquetUtils.java

@@ -0,0 +1,231 @@
+package org.talend.parquet.utils;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.RoundingMode;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.DecimalMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Type.Repetition;
+import org.apache.parquet.schema.Types;
+import org.apache.parquet.schema.Types.GroupBuilder;
+import org.talend.parquet.data.Group;
+import org.talend.parquet.data.simple.NanoTime;
+
+public class TalendParquetUtils {
+
+	public static final String ARRAY_FIELD_NAME = "array";
+
+	public static PrimitiveType createPrimitiveType(String fieldName, boolean nullable, String primitiveType,
+			String originalTypeName) {
+		OriginalType originalType = null;
+		if (originalTypeName != null) {
+			originalType = OriginalType.valueOf(originalTypeName);
+		}
+		return new PrimitiveType((nullable ? Repetition.OPTIONAL : Repetition.REQUIRED),
+				PrimitiveTypeName.valueOf(primitiveType), fieldName, originalType);
+	}
+
+	public static PrimitiveType createDecimalType(String fieldName, boolean nullable, int precision, int scale) {
+		DecimalMetadata decimalMetadata = new DecimalMetadata(precision, scale);
+		return new PrimitiveType((nullable ? Repetition.OPTIONAL : Repetition.REQUIRED),
+				PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, 16, fieldName, OriginalType.DECIMAL, decimalMetadata, null);
+	}
+
+	public static Type createGroupElementType(String fieldName, Object element) {
+		if (element == null) {
+			return Types.repeated(PrimitiveTypeName.BINARY).as(OriginalType.UTF8).named(fieldName);
+		}
+		if (String.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.BINARY).as(OriginalType.UTF8).named(fieldName);
+		} else if (Double.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.DOUBLE).named(fieldName);
+		} else if (Float.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.FLOAT).named(fieldName);
+		} else if (Byte.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.INT32).as(OriginalType.INT_8).named(fieldName);
+		} else if (Short.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.INT32).as(OriginalType.INT_16).named(fieldName);
+		} else if (Integer.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.INT32).named(fieldName);
+		} else if (Long.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.INT64).named(fieldName);
+		} else if (Boolean.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.BOOLEAN).named(fieldName);
+		} else if (Date.class.isInstance(element)) {
+			return Types.repeated(PrimitiveTypeName.INT64).as(OriginalType.TIMESTAMP_MILLIS).named(fieldName);
+		} else if (Group.class.isInstance(element)) {
+			return ((Group) element).getType();
+		} else {
+			throw new IllegalArgumentException("Unsupported type: " + element.getClass().getCanonicalName()
+					+ " for group type field'" + fieldName + "'");
+		}
+	}
+
+	public static GroupType createGroupType(String fieldName, boolean nullable, Object element) {
+		GroupBuilder<GroupType> builder = null;
+		if (nullable) {
+			builder = Types.optionalGroup();
+		} else {
+			builder = Types.requiredGroup();
+		}
+		return builder.as(OriginalType.LIST).addField(createGroupElementType("array", element)).named(fieldName);
+	}
+
+	/*
+	 * Here group only support List value with one field
+	 */
+	public static List<Object> groupFieldValueToList(Group group) {
+		if (group == null) {
+			return null;
+		}
+		List<Object> values = new ArrayList<>();
+		int listSize = group.getFieldRepetitionCount(0);
+		for (int elementIndex = 0; elementIndex < listSize; elementIndex++) {
+			Type elelemntType = group.getType().getType(0);
+			if (elelemntType.isPrimitive()) {
+				PrimitiveType pType = elelemntType.asPrimitiveType();
+				switch (pType.getPrimitiveTypeName()) {
+				case INT64:
+					if (OriginalType.TIMESTAMP_MILLIS == elelemntType.getOriginalType()) {
+						values.add(new Date(group.getLong(0, elementIndex)));
+					} else {
+						values.add(group.getLong(0, elementIndex));
+					}
+					break;
+				case INT32:
+					values.add(group.getInteger(0, elementIndex));
+					break;
+				case BOOLEAN:
+					values.add(group.getBoolean(0, elementIndex));
+					break;
+				case INT96:
+					Binary value = group.getInt96(0, elementIndex);
+					if (value != null) {
+						NanoTime nanoTime = NanoTime.fromBinary(value);
+						values.add(new Date(NanoTimeUtils.getTimestamp(nanoTime, false).getTime()));
+					} else {
+						values.add(value);
+					}
+					break;
+				case FLOAT:
+					values.add(group.getFloat(0, elementIndex));
+					break;
+				case DOUBLE:
+					values.add(group.getDouble(0, elementIndex));
+					break;
+				default:
+					values.add(group.getValueToString(0, elementIndex));
+				}
+			} else {
+				values.add(groupFieldValueToList(group.getGroup(0, elementIndex)));
+			}
+		}
+		return values;
+	}
+
+	public static void writeGroupField(Group nestGroup, List<?> values) {
+		if (values == null || values.isEmpty()) {
+			return;
+		}
+		// only support one field currently
+		for (int i = 0; i < values.size(); i++) {
+			Object element = values.get(i);
+			if (String.class.isInstance(element)) {
+				nestGroup.add(0, (String) element);
+			} else if (Double.class.isInstance(element)) {
+				nestGroup.add(0, (Double) element);
+			} else if (Float.class.isInstance(element)) {
+				nestGroup.add(0, (Float) element);
+			} else if (Byte.class.isInstance(element)) {
+				nestGroup.add(0, (Byte) element);
+			} else if (Short.class.isInstance(element)) {
+				nestGroup.add(0, (Short) element);
+			} else if (Integer.class.isInstance(element)) {
+				nestGroup.add(0, (Integer) element);
+			} else if (Long.class.isInstance(element)) {
+				nestGroup.add(0, (Long) element);
+			} else if (Boolean.class.isInstance(element)) {
+				nestGroup.add(0, (Boolean) element);
+			} else if (Date.class.isInstance(element)) {
+				nestGroup.add(0, ((Date) element).getTime());
+			} else if (Group.class.isInstance(element)) {
+				nestGroup.add(0, (Group) element);
+			} else {
+				throw new IllegalArgumentException("Unsupported type: " + element.getClass().getCanonicalName()
+						+ " for group type field'" + nestGroup + "'");
+			}
+		}
+	}
+
+	public static BigDecimal binaryToDecimal(Binary value, int precision, int scale) {
+		/*
+		 * Precision <= 18 checks for the max number of digits for an unscaled long,
+		 * else treat with big integer conversion
+		 */
+		if (precision <= 18) {
+			ByteBuffer buffer = value.toByteBuffer();
+			byte[] bytes = buffer.array();
+			int start = buffer.arrayOffset() + buffer.position();
+			int end = buffer.arrayOffset() + buffer.limit();
+			long unscaled = 0L;
+			int i = start;
+			while (i < end) {
+				unscaled = (unscaled << 8 | bytes[i] & 0xff);
+				i++;
+			}
+			int bits = 8 * (end - start);
+			long unscaledNew = (unscaled << (64 - bits)) >> (64 - bits);
+			if (scale == 0 || unscaledNew <= -Math.pow(10, 18) || unscaledNew >= Math.pow(10, 18)) {
+				return new BigDecimal(unscaledNew);
+			} else {
+				return BigDecimal.valueOf(unscaledNew / Math.pow(10, scale));
+			}
+		} else {
+			return new BigDecimal(new BigInteger(value.getBytes()), scale);
+		}
+	}
+
+    public static Binary decimalToBinary(BigDecimal decimalValue, int scale) {
+        // First we need to make sure the BigDecimal matches our schema scale:
+        decimalValue = decimalValue.setScale(scale, RoundingMode.HALF_UP);
+
+        // Next we get the decimal value as one BigInteger (like there was no decimal
+        // point)
+        BigInteger unscaledDecimalValue = decimalValue.unscaledValue();
+
+        byte[] decimalBuffer = null;
+        // Finally we serialize the integer
+        byte[] decimalBytes = unscaledDecimalValue.toByteArray();
+
+        if (decimalValue.compareTo(BigDecimal.ZERO) < 0) {
+            decimalBuffer = new byte[] { -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1 };
+        } else {
+            decimalBuffer = new byte[16];
+        }
+        if (decimalBuffer.length >= decimalBytes.length) {
+            // Because we set our fixed byte array size as 16 bytes, we need to
+            // pad-left our original value's bytes with 0 or -1
+            int decimalBufferIndex = decimalBuffer.length - 1;
+            for (int i = decimalBytes.length - 1; i >= 0; i--) {
+                decimalBuffer[decimalBufferIndex] = decimalBytes[i];
+                decimalBufferIndex--;
+            }
+        } else {
+            throw new IllegalArgumentException(String
+                    .format("Decimal size: %d was greater than the allowed max: %d",
+                            decimalBytes.length, decimalBuffer.length));
+        }
+        return Binary.fromReusedByteArray(decimalBuffer);
+    }
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/test/java/org/talend/parquet/TestUtils.java

@@ -0,0 +1,86 @@
+/* 
+ * 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.talend.parquet;
+
+import java.io.IOException;
+import java.util.concurrent.Callable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.statistics.Statistics;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+
+public class TestUtils {
+
+	public static void enforceEmptyDir(Configuration conf, Path path) throws IOException {
+		FileSystem fs = path.getFileSystem(conf);
+		if (fs.exists(path)) {
+			if (!fs.delete(path, true)) {
+				throw new IOException("can not delete path " + path);
+			}
+		}
+		if (!fs.mkdirs(path)) {
+			throw new IOException("can not create path " + path);
+		}
+	}
+
+	/**
+	 * A convenience method to avoid a large number of @Test(expected=...) tests
+	 * 
+	 * @param message  A String message to describe this assertion
+	 * @param expected An Exception class that the Runnable should throw
+	 * @param callable A Callable that is expected to throw the exception
+	 */
+	public static void assertThrows(String message, Class<? extends Exception> expected, Callable callable) {
+		try {
+			callable.call();
+			Assert.fail("No exception was thrown (" + message + "), expected: " + expected.getName());
+		} catch (Exception actual) {
+			try {
+				Assert.assertEquals(message, expected, actual.getClass());
+			} catch (AssertionError e) {
+				e.addSuppressed(actual);
+				throw e;
+			}
+		}
+	}
+
+	public static void assertStatsValuesEqual(Statistics<?> stats1, Statistics<?> stats2) {
+		assertStatsValuesEqual(null, stats1, stats2);
+	}
+
+	// To be used to assert that the values (min, max, num-of-nulls) equals. It
+	// might be used in cases when creating
+	// Statistics object for the proper Type would require too much work/code
+	// duplications etc.
+	public static void assertStatsValuesEqual(String message, Statistics<?> expected, Statistics<?> actual) {
+		if (expected == actual) {
+			return;
+		}
+		if (expected == null || actual == null) {
+			Assert.assertEquals(expected, actual);
+		}
+		Assert.assertThat(actual, CoreMatchers.instanceOf(expected.getClass()));
+		Assert.assertArrayEquals(message, expected.getMaxBytes(), actual.getMaxBytes());
+		Assert.assertArrayEquals(message, expected.getMinBytes(), actual.getMinBytes());
+		Assert.assertEquals(message, expected.getNumNulls(), actual.getNumNulls());
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/test/java/org/talend/parquet/hadoop/TalendGroupReadSupportTest.java

@@ -0,0 +1,63 @@
+/* 
+ * 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.talend.parquet.hadoop;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.hadoop.api.ReadSupport;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.Test;
+import org.talend.parquet.hadoop.TalendGroupReadSupport;
+
+public class TalendGroupReadSupportTest {
+
+	private String fullSchemaStr = "message example {\n" + "required int32 line;\n" + "optional binary content;\n"
+			+ "}";
+
+	private String partialSchemaStr = "message example {\n" + "required int32 line;\n" + "}";
+
+	@Test
+	public void testInitWithoutSpecifyingRequestSchema() throws Exception {
+		TalendGroupReadSupport s = new TalendGroupReadSupport();
+		Configuration configuration = new Configuration();
+		Map<String, String> keyValueMetaData = new HashMap<String, String>();
+		MessageType fileSchema = MessageTypeParser.parseMessageType(fullSchemaStr);
+
+		ReadSupport.ReadContext context = s.init(configuration, keyValueMetaData, fileSchema);
+		assertEquals(context.getRequestedSchema(), fileSchema);
+	}
+
+	@Test
+	public void testInitWithPartialSchema() {
+		TalendGroupReadSupport s = new TalendGroupReadSupport();
+		Configuration configuration = new Configuration();
+		Map<String, String> keyValueMetaData = new HashMap<String, String>();
+		MessageType fileSchema = MessageTypeParser.parseMessageType(fullSchemaStr);
+		MessageType partialSchema = MessageTypeParser.parseMessageType(partialSchemaStr);
+		configuration.set(ReadSupport.PARQUET_READ_SCHEMA, partialSchemaStr);
+
+		ReadSupport.ReadContext context = s.init(configuration, keyValueMetaData, fileSchema);
+		assertEquals(context.getRequestedSchema(), partialSchema);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/test/java/org/talend/parquet/hadoop/TestParquetWriter.java

@@ -0,0 +1,169 @@
+/*
+ * 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.talend.parquet.hadoop;
+
+import static java.util.Arrays.asList;
+import static org.apache.parquet.column.Encoding.DELTA_BYTE_ARRAY;
+import static org.apache.parquet.column.Encoding.PLAIN;
+import static org.apache.parquet.column.Encoding.PLAIN_DICTIONARY;
+import static org.apache.parquet.column.Encoding.RLE_DICTIONARY;
+import static org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER;
+import static org.apache.parquet.hadoop.ParquetFileReader.readFooter;
+import static org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED;
+import static org.apache.parquet.schema.MessageTypeParser.parseMessageType;
+import static org.apache.parquet.schema.Type.Repetition.REQUIRED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.column.ParquetProperties.WriterVersion;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.example.GroupReadSupport;
+import org.apache.parquet.hadoop.example.GroupWriteSupport;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.InvalidSchemaException;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Types;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.talend.parquet.TestUtils;
+
+
+public class TestParquetWriter {
+
+  @Test
+  public void test() throws Exception {
+    Configuration conf = new Configuration();
+    Path root = new Path("target/tests/TestParquetWriter/");
+    TestUtils.enforceEmptyDir(conf, root);
+    MessageType schema = parseMessageType(
+        "message test { "
+        + "required binary binary_field; "
+        + "required int32 int32_field; "
+        + "required int64 int64_field; "
+        + "required boolean boolean_field; "
+        + "required float float_field; "
+        + "required double double_field; "
+        + "required fixed_len_byte_array(3) flba_field; "
+        + "required int96 int96_field; "
+        + "} ");
+    GroupWriteSupport.setSchema(schema, conf);
+    SimpleGroupFactory f = new SimpleGroupFactory(schema);
+    Map<String, Encoding> expected = new HashMap<String, Encoding>();
+    expected.put("10-" + ParquetProperties.WriterVersion.PARQUET_1_0, PLAIN_DICTIONARY);
+    expected.put("1000-" + ParquetProperties.WriterVersion.PARQUET_1_0, PLAIN);
+    expected.put("10-" + ParquetProperties.WriterVersion.PARQUET_2_0, RLE_DICTIONARY);
+    expected.put("1000-" + ParquetProperties.WriterVersion.PARQUET_2_0, DELTA_BYTE_ARRAY);
+    for (int modulo : asList(10, 1000)) {
+      for (WriterVersion version : WriterVersion.values()) {
+        Path file = new Path(root, version.name() + "_" + modulo);
+        ParquetWriter<Group> writer = new ParquetWriter<Group>(
+            file,
+            new GroupWriteSupport(),
+            UNCOMPRESSED, 1024, 1024, 512, true, false, version, conf);
+        for (int i = 0; i < 1000; i++) {
+          writer.write(
+              f.newGroup()
+              .append("binary_field", "test" + (i % modulo))
+              .append("int32_field", 32)
+              .append("int64_field", 64l)
+              .append("boolean_field", true)
+              .append("float_field", 1.0f)
+              .append("double_field", 2.0d)
+              .append("flba_field", "foo")
+              .append("int96_field", Binary.fromConstantByteArray(new byte[12])));
+        }
+        writer.close();
+        ParquetReader<Group> reader = ParquetReader.builder(new GroupReadSupport(), file).withConf(conf).build();
+        for (int i = 0; i < 1000; i++) {
+          Group group = reader.read();
+          assertEquals("test" + (i % modulo), group.getBinary("binary_field", 0).toStringUsingUTF8());
+          assertEquals(32, group.getInteger("int32_field", 0));
+          assertEquals(64l, group.getLong("int64_field", 0));
+          assertEquals(true, group.getBoolean("boolean_field", 0));
+          assertEquals(1.0f, group.getFloat("float_field", 0), 0.001);
+          assertEquals(2.0d, group.getDouble("double_field", 0), 0.001);
+          assertEquals("foo", group.getBinary("flba_field", 0).toStringUsingUTF8());
+          assertEquals(Binary.fromConstantByteArray(new byte[12]),
+              group.getInt96("int96_field",0));
+        }
+        reader.close();
+        ParquetMetadata footer = readFooter(conf, file, NO_FILTER);
+        for (BlockMetaData blockMetaData : footer.getBlocks()) {
+          for (ColumnChunkMetaData column : blockMetaData.getColumns()) {
+            if (column.getPath().toDotString().equals("binary_field")) {
+              String key = modulo + "-" + version;
+              Encoding expectedEncoding = expected.get(key);
+              assertTrue(
+                  key + ":" + column.getEncodings() + " should contain " + expectedEncoding,
+                  column.getEncodings().contains(expectedEncoding));
+            }
+          }
+        }
+        assertEquals("Object model property should be example",
+            "example", footer.getFileMetaData().getKeyValueMetaData()
+                .get(ParquetWriter.OBJECT_MODEL_NAME_PROP));
+      }
+    }
+  }
+
+  @Rule
+  public TemporaryFolder temp = new TemporaryFolder();
+
+  @Test
+  public void testBadWriteSchema() throws IOException {
+    final File file = temp.newFile("test.parquet");
+    file.delete();
+
+    TestUtils.assertThrows("Should reject a schema with an empty group",
+        InvalidSchemaException.class, new Callable<Void>() {
+          @Override
+          public Void call() throws IOException {
+            ExampleParquetWriter.builder(new Path(file.toString()))
+                .withType(Types.buildMessage()
+                    .addField(new GroupType(REQUIRED, "invalid_group"))
+                    .named("invalid_message"))
+                .build();
+            return null;
+          }
+        });
+
+    Assert.assertFalse("Should not create a file when schema is rejected",
+        file.exists());
+  }
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-parquet/src/test/java/org/talend/parquet/util/TalendParquetUtilsTest.java

@@ -0,0 +1,286 @@
+package org.talend.parquet.util;
+
+import static org.apache.parquet.schema.MessageTypeParser.parseMessageType;
+import static org.apache.parquet.schema.OriginalType.DECIMAL;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
+import static org.apache.parquet.schema.Type.Repetition.REQUIRED;
+
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.parquet.schema.DecimalMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+import org.hamcrest.MatcherAssert;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
+import org.junit.Test;
+import org.talend.parquet.data.Group;
+import org.talend.parquet.data.simple.SimpleGroup;
+import org.talend.parquet.utils.TalendParquetUtils;
+
+public class TalendParquetUtilsTest {
+
+	@Test
+	public void testGetStringList() {
+		MessageType schema = parseMessageType("message Schema { " //
+				+ "  optional int64 fieldo; " //
+				+ "  optional group field1 { " //
+				+ "    repeated binary field2 (UTF8); " //
+				+ "  } " //
+				+ "}"); //
+
+		Group group = new SimpleGroup(schema.getType(1).asGroupType());
+		group.add(0, "element 1");
+		group.add(0, "element 2");
+		group.add(0, "element 3");
+		group.add(0, "element 4");
+
+		List<Object> values = TalendParquetUtils.groupFieldValueToList(group);
+		MatcherAssert.assertThat("", values, Matchers.contains("element 1", "element 2", "element 3", "element 4"));
+
+	}
+
+	@Test
+	public void testGetIntList() {
+		MessageType schema = parseMessageType("message Schema { " //
+				+ "  optional int64 fieldo; " //
+				+ "  optional group field1 { " //
+				+ "    repeated int32 field2 ; " //
+				+ "  } " //
+				+ "}"); //
+
+		Group group = new SimpleGroup(schema.getType(1).asGroupType());
+		group.add(0, 123);
+		group.add(0, 345);
+		group.add(0, 431);
+
+		List<Object> values = TalendParquetUtils.groupFieldValueToList(group);
+		MatcherAssert.assertThat("", values, Matchers.contains(123, 345, 431));
+
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testNestGroupList() {
+		MessageType schema = parseMessageType("message Schema { " //
+				+ "  optional int64 fieldo; " //
+				+ "  optional group field1 { " //
+				+ "    repeated group field2 {" //
+				+ "      repeated double field3;" //
+				+ "    } " //
+				+ "  } " //
+				+ "}"); //
+
+		Group group = new SimpleGroup(schema.getType(1).asGroupType());
+
+		Group nest1 = new SimpleGroup(schema.getType(1).asGroupType().getType(0).asGroupType());
+		nest1.add(0, 123.0);
+		nest1.add(0, 345.0);
+		nest1.add(0, 431.0);
+
+		Group nest2 = new SimpleGroup(schema.getType(1).asGroupType().getType(0).asGroupType());
+		nest2.add(0, 2123.0);
+		nest2.add(0, 2345.0);
+		nest2.add(0, 2431.0);
+
+		group.add(0, nest1);
+		group.add(0, nest2);
+
+		List<Object> values = TalendParquetUtils.groupFieldValueToList(group);
+
+		MatcherAssert.assertThat("", (List<Object>) values.get(0), Matchers.contains(123.0, 345.0, 431.0));
+		MatcherAssert.assertThat("", (List<Object>) values.get(1), Matchers.contains(2123.0, 2345.0, 2431.0));
+
+	}
+
+	@Test
+	public void testNullGroupList() {
+		List<Object> values = TalendParquetUtils.groupFieldValueToList(null);
+		Assert.assertNull(values);
+	}
+
+	@Test
+	public void testCreateGroupElementType() {
+		Type emptyElement = TalendParquetUtils.createGroupElementType("field0", null);
+		Assert.assertEquals(PrimitiveTypeName.BINARY, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", "1");
+		Assert.assertEquals(PrimitiveTypeName.BINARY, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", 1.0);
+		Assert.assertEquals(PrimitiveTypeName.DOUBLE, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", 1.0f);
+		Assert.assertEquals(PrimitiveTypeName.FLOAT, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", 1);
+		Assert.assertEquals(PrimitiveTypeName.INT32, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", 1L);
+		Assert.assertEquals(PrimitiveTypeName.INT64, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", true);
+		Assert.assertEquals(PrimitiveTypeName.BOOLEAN, emptyElement.asPrimitiveType().getPrimitiveTypeName());
+
+		// Nest group
+		MessageType schema = parseMessageType("message Schema { " //
+				+ "  optional group field1 { " //
+				+ "    repeated group field2 {" //
+				+ "      repeated double field3;" //
+				+ "    } " //
+				+ "  } " //
+				+ "}"); //
+		Group group = new SimpleGroup(schema.getType(0).asGroupType());
+		Group nest1 = new SimpleGroup(schema.getType(0).asGroupType().getType(0).asGroupType());
+		nest1.add(0, 123.0);
+		nest1.add(0, 345.0);
+		nest1.add(0, 431.0);
+		Group nest2 = new SimpleGroup(schema.getType(0).asGroupType().getType(0).asGroupType());
+		nest2.add(0, 2123.0);
+		nest2.add(0, 2345.0);
+		nest2.add(0, 2431.0);
+
+		group.add(0, nest1);
+		group.add(0, nest2);
+		Assert.assertFalse("Should be group type", group.getType().isPrimitive());
+
+		Assert.assertEquals(2, group.getFieldRepetitionCount(0));
+
+		emptyElement = TalendParquetUtils.createGroupElementType("field0", group);
+		Assert.assertFalse("Should be group type", emptyElement.isPrimitive());
+		Assert.assertEquals(schema.getType(0).asGroupType(), emptyElement);
+	}
+
+	@Test
+	public void testCreateGroupType() {
+		GroupType emptyElement = TalendParquetUtils.createGroupType("field0", true, null);
+		Assert.assertEquals(OriginalType.LIST, emptyElement.asGroupType().getOriginalType());
+		Assert.assertEquals(OriginalType.UTF8, emptyElement.getType(0).asPrimitiveType().getOriginalType());
+
+		emptyElement = TalendParquetUtils.createGroupType("field0", true, 2);
+		Assert.assertEquals(OriginalType.LIST, emptyElement.asGroupType().getOriginalType());
+		Assert.assertEquals(PrimitiveTypeName.INT32, emptyElement.getType(0).asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupType("field0", true, Byte.valueOf("1"));
+		Assert.assertEquals(OriginalType.LIST, emptyElement.asGroupType().getOriginalType());
+		Assert.assertEquals(OriginalType.INT_8, emptyElement.getType(0).asPrimitiveType().getOriginalType());
+		Assert.assertEquals(PrimitiveTypeName.INT32, emptyElement.getType(0).asPrimitiveType().getPrimitiveTypeName());
+
+		emptyElement = TalendParquetUtils.createGroupType("field0", true, Short.valueOf("1"));
+		Assert.assertEquals(OriginalType.LIST, emptyElement.asGroupType().getOriginalType());
+		Assert.assertEquals(OriginalType.INT_16, emptyElement.getType(0).asPrimitiveType().getOriginalType());
+		Assert.assertEquals(PrimitiveTypeName.INT32, emptyElement.getType(0).asPrimitiveType().getPrimitiveTypeName());
+	}
+
+	@Test
+	public void testWriteGroupField() {
+		Group group = null;
+		MessageType schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated int32 array;" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		List<?> values = Arrays.asList(1, 2, 3);
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(3, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated int32 array(INT_8);" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList(Byte.valueOf("1"), Byte.valueOf("2"));
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(2, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated int32 array(INT_16);" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList(Short.valueOf("1"));
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(1, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated int64 array;" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList(1L, 2L, 3L);
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(3, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated double array;" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList(1.0, 2.0, 3.0);
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(3, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated float array;" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList(1.0f, 2.0f, 3.0f);
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(3, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated binary array (UTF8);" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList("element 1", "element 2");
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(2, group.getFieldRepetitionCount(0));
+
+		schema = parseMessageType("message Schema { " //
+				+ "    optional group field0 (LIST) {" + "    repeated boolean array ;" + "  } " //
+				+ "}"); //
+		group = new SimpleGroup(schema.getType(0).asGroupType());
+		values = Arrays.asList(true, false);
+		TalendParquetUtils.writeGroupField(group, values);
+		Assert.assertEquals(2, group.getFieldRepetitionCount(0));
+	}
+
+	@Test
+	public void testDecimalAnnotation() {
+		Group group = null;
+		MessageType schema = new MessageType("DecimalMessage", new PrimitiveType(REQUIRED, FIXED_LEN_BYTE_ARRAY, 16,
+				"aDecimal", DECIMAL, new DecimalMetadata(38, 2), null));
+		BigDecimal decimalValue = new BigDecimal("1234423199.9999");
+
+		group = new SimpleGroup(schema);
+		group.append("aDecimal", TalendParquetUtils.decimalToBinary(decimalValue, 5));
+		Assert.assertEquals(decimalValue.setScale(5), TalendParquetUtils.binaryToDecimal(group.getBinary(0, 0), 38, 5));
+
+		group = new SimpleGroup(schema);
+		group.append("aDecimal", TalendParquetUtils.decimalToBinary(decimalValue, 4));
+		Assert.assertEquals(decimalValue, TalendParquetUtils.binaryToDecimal(group.getBinary(0, 0), 38, 4));
+
+		decimalValue = new BigDecimal("1234");
+		group = new SimpleGroup(schema);
+		group.append("aDecimal", TalendParquetUtils.decimalToBinary(decimalValue, 0));
+		Assert.assertEquals(decimalValue, TalendParquetUtils.binaryToDecimal(group.getBinary(0, 0), 10, 0));
+
+		decimalValue = new BigDecimal("-93.5788130000");
+		group = new SimpleGroup(schema);
+		group.append("aDecimal", TalendParquetUtils.decimalToBinary(decimalValue, 10));
+		Assert.assertEquals(decimalValue, TalendParquetUtils.binaryToDecimal(group.getBinary(0, 0), 38, 10));
+		
+		decimalValue = new BigDecimal("-0.00");
+		group = new SimpleGroup(schema);
+		group.append("aDecimal", TalendParquetUtils.decimalToBinary(decimalValue, 2));
+		Assert.assertEquals(decimalValue, TalendParquetUtils.binaryToDecimal(group.getBinary(0, 0), 38, 2));
+		
+		decimalValue = new BigDecimal("0.000");
+		group = new SimpleGroup(schema);
+		group.append("aDecimal", TalendParquetUtils.decimalToBinary(decimalValue, 3));
+		Assert.assertEquals(decimalValue, TalendParquetUtils.binaryToDecimal(group.getBinary(0, 0), 38, 3));
+		
+	}
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-scp-helper/pom.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.talend.components</groupId>
+    <artifactId>talend-scp-helper</artifactId>
+    <version>1.0</version>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <ssd.version>2.5.1</ssd.version>
+		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>
+    </properties>
+	
+	
+	<distributionManagement>
+		<snapshotRepository>
+			<id>talend_nexus_deployment</id>
+			<url>${talend.nexus.url}/nexus/content/repositories/TalendOpenSourceSnapshot/</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</snapshotRepository>
+		<repository>
+			<id>talend_nexus_deployment</id>
+			<url>${talend.nexus.url}/nexus/content/repositories/TalendOpenSourceRelease/</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+			<releases>
+				<enabled>true</enabled>
+			</releases>
+		</repository>
+	</distributionManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.sshd</groupId>
+            <artifactId>sshd-scp</artifactId>
+            <version>${ssd.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.sshd</groupId>
+            <artifactId>sshd-common</artifactId>
+            <version>${ssd.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.sshd</groupId>
+            <artifactId>sshd-core</artifactId>
+            <version>${ssd.version}</version>
+        </dependency>
+    </dependencies>
+</project>

main/plugins/org.talend.designer.components.libs/libs_src/talend-scp-helper/src/main/java/org/talend/components/talendscp/TalendScpClient.java

@@ -0,0 +1,115 @@
+package org.talend.components.talendscp;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.FileSystem;
+import java.nio.file.Path;
+import java.nio.file.attribute.PosixFilePermission;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
+
+import org.apache.sshd.client.channel.ChannelExec;
+import org.apache.sshd.client.scp.AbstractScpClient;
+import org.apache.sshd.client.scp.DefaultScpClient;
+import org.apache.sshd.client.scp.DefaultScpStreamResolver;
+import org.apache.sshd.client.scp.ScpClient;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.FactoryManager;
+import org.apache.sshd.common.file.FileSystemFactory;
+import org.apache.sshd.common.file.util.MockFileSystem;
+import org.apache.sshd.common.file.util.MockPath;
+import org.apache.sshd.common.scp.ScpFileOpener;
+import org.apache.sshd.common.scp.ScpHelper;
+import org.apache.sshd.common.scp.ScpTimestamp;
+import org.apache.sshd.common.scp.ScpTransferEventListener;
+import org.apache.sshd.common.util.ValidateUtils;
+
+
+/*
+What we do here is just overwrite the ScpHelper to throw Exception when scp return status 1
+to keep the old behavior
+ */
+
+public class TalendScpClient extends DefaultScpClient {
+
+    public TalendScpClient(ClientSession clientSession, ScpFileOpener fileOpener,
+            ScpTransferEventListener eventListener) {
+        super(clientSession, fileOpener, eventListener);
+    }
+
+    @Override public void upload(InputStream local, String remote, long size, Collection<PosixFilePermission> perms,
+            ScpTimestamp time) throws IOException {
+        int namePos = ValidateUtils.checkNotNullAndNotEmpty(remote, "No remote location specified").lastIndexOf('/');
+        String name = (namePos < 0) ?
+                remote :
+                ValidateUtils.checkNotNullAndNotEmpty(remote.substring(namePos + 1), "No name value in remote=%s",
+                        remote);
+        Collection<Option> options = (time != null) ? EnumSet.of(Option.PreserveAttributes) : Collections.emptySet();
+        String cmd = ScpClient.createSendCommand(remote, options);
+        ClientSession session = getClientSession();
+        ChannelExec channel = openCommandChannel(session, cmd);
+        try (InputStream invOut = channel.getInvertedOut(); OutputStream invIn = channel.getInvertedIn()) {
+            // NOTE: we use a mock file system since we expect no invocations for it
+            ScpHelper helper = new TalendScpHelper(session, invOut, invIn, new MockFileSystem(remote), opener, listener);
+            Path mockPath = new MockPath(remote);
+            helper.sendStream(new DefaultScpStreamResolver(name, mockPath, perms, time, size, local, cmd),
+                    options.contains(Option.PreserveAttributes), ScpHelper.DEFAULT_SEND_BUFFER_SIZE);
+            handleCommandExitStatus(cmd, channel);
+        } finally {
+            channel.close(false);
+        }
+    }
+
+    @Override protected <T> void runUpload(String remote, Collection<Option> options, Collection<T> local,
+            AbstractScpClient.ScpOperationExecutor<T> executor) throws IOException {
+        local = ValidateUtils.checkNotNullAndNotEmpty(local, "Invalid argument local: %s", local);
+        remote = ValidateUtils.checkNotNullAndNotEmpty(remote, "Invalid argument remote: %s", remote);
+        if (local.size() > 1) {
+            options = addTargetIsDirectory(options);
+        }
+
+        String cmd = ScpClient.createSendCommand(remote, options);
+        ClientSession session = getClientSession();
+        ChannelExec channel = openCommandChannel(session, cmd);
+        try {
+            FactoryManager manager = session.getFactoryManager();
+            FileSystemFactory factory = manager.getFileSystemFactory();
+            FileSystem fs = factory.createFileSystem(session);
+
+            try (InputStream invOut = channel.getInvertedOut(); OutputStream invIn = channel.getInvertedIn()) {
+                ScpHelper helper = new TalendScpHelper(session, invOut, invIn, fs, opener, listener);
+                executor.execute(helper, local, options);
+            } finally {
+                try {
+                    fs.close();
+                } catch (UnsupportedOperationException e) {
+                    if (log.isDebugEnabled()) {
+                        log.debug("runUpload({}) {} => {} - failed ({}) to close file system={}: {}", session, remote,
+                                local, e.getClass().getSimpleName(), fs, e.getMessage());
+                    }
+                }
+            }
+            handleCommandExitStatus(cmd, channel);
+        } finally {
+            channel.close(false);
+        }
+    }
+
+    @Override public void download(String remote, OutputStream local) throws IOException {
+        String cmd = ScpClient.createReceiveCommand(remote, Collections.emptyList());
+        ClientSession session = getClientSession();
+        ChannelExec channel = openCommandChannel(session, cmd);
+        try (InputStream invOut = channel.getInvertedOut(); OutputStream invIn = channel.getInvertedIn()) {
+            // NOTE: we use a mock file system since we expect no invocations for it
+            ScpHelper helper =
+                    new TalendScpHelper(session, invOut, invIn, new MockFileSystem(remote), opener, listener);
+            helper.receiveFileStream(local, ScpHelper.DEFAULT_RECEIVE_BUFFER_SIZE);
+            handleCommandExitStatus(cmd, channel);
+        } finally {
+
+        }
+    }
+
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-scp-helper/src/main/java/org/talend/components/talendscp/TalendScpClientCreator.java

@@ -0,0 +1,22 @@
+package org.talend.components.talendscp;
+
+import org.apache.sshd.client.scp.AbstractScpClientCreator;
+import org.apache.sshd.client.scp.ScpClient;
+import org.apache.sshd.client.scp.ScpClientCreator;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.scp.ScpFileOpener;
+import org.apache.sshd.common.scp.ScpTransferEventListener;
+
+public class TalendScpClientCreator extends AbstractScpClientCreator implements ScpClientCreator {
+    public static final TalendScpClientCreator INSTANCE = new TalendScpClientCreator();
+
+    @Override
+    public ScpClient createScpClient(ClientSession session) {
+        return this.createScpClient(session, this.getScpFileOpener(), this.getScpTransferEventListener());
+    }
+
+    @Override public ScpClient createScpClient(ClientSession clientSession, ScpFileOpener scpFileOpener,
+            ScpTransferEventListener scpTransferEventListener) {
+        return new TalendScpClient(clientSession, scpFileOpener, scpTransferEventListener);
+    }
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-scp-helper/src/main/java/org/talend/components/talendscp/TalendScpHelper.java

@@ -0,0 +1,63 @@
+package org.talend.components.talendscp;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.FileSystem;
+
+import org.apache.sshd.common.scp.ScpException;
+import org.apache.sshd.common.scp.ScpFileOpener;
+import org.apache.sshd.common.scp.ScpHelper;
+import org.apache.sshd.common.scp.ScpTransferEventListener;
+import org.apache.sshd.common.session.Session;
+
+public class TalendScpHelper extends ScpHelper {
+
+    public TalendScpHelper(Session session, InputStream in, OutputStream out, FileSystem fileSystem,
+            ScpFileOpener opener, ScpTransferEventListener eventListener) {
+        super(session, in, out, fileSystem, opener, eventListener);
+    }
+
+    @Override
+    public int readAck(boolean canEof) throws IOException {
+        int c = in.read();
+        switch (c) {
+        case -1:
+            if (log.isDebugEnabled()) {
+                log.debug("readAck({})[EOF={}] received EOF", this, canEof);
+            }
+            if (!canEof) {
+                throw new EOFException("readAck - EOF before ACK");
+            }
+            break;
+        case OK:
+            if (log.isDebugEnabled()) {
+                log.debug("readAck({})[EOF={}] read OK", this, canEof);
+            }
+            break;
+        case WARNING: {
+            if (log.isDebugEnabled()) {
+                log.debug("readAck({})[EOF={}] read warning message", this, canEof);
+            }
+
+            String line = readLine();
+            log.warn("readAck({})[EOF={}] - Received warning: {}", this, canEof, line);
+            throw new ScpException("received error: " + line, c);
+        }
+        case ERROR: {
+            if (log.isDebugEnabled()) {
+                log.debug("readAck({})[EOF={}] read error message", this, canEof);
+            }
+            String line = readLine();
+            if (log.isDebugEnabled()) {
+                log.debug("readAck({})[EOF={}] received error: {}", this, canEof, line);
+            }
+            throw new ScpException("Received nack: " + line, c);
+        }
+        default:
+            break;
+        }
+        return c;
+    }
+}

main/plugins/org.talend.designer.components.localprovider/components/tAS400Output/tAS400Output_begin.javajet

@@ -86,39 +86,7 @@ if(columnList != null && columnList.size() > 0) {
 }
 %>
 
-<%
-if(("UPDATE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction) || ("UPDATE_OR_INSERT").equals(dataAction)) {
-    int updateKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isUpdateKey() && !column.isDynamic()) {
-                updateKeyCount++;
-            }
-        }
-        %>
-        int updateKeyCount_<%=cid%> = <%=updateKeyCount%>;
-        if(updateKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For update, Schema must have a key");
-        }
-        <%
-    }
-} else if(("DELETE").equals(dataAction)) {
-    int deleteKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isDeleteKey() && !column.isDynamic()) {
-                deleteKeyCount++;
-            }
-        }
-        %>
-        int deleteKeyCount_<%=cid%> = <%=deleteKeyCount%>;
-        if(deleteKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For delete, Schema must have a key");
-        }
-        <%
-    }
-}
-%>
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/DB/Output/CheckKeysForUpdateAndDelete.javajet"%>
 
 int nb_line_<%=cid%> = 0;
 int nb_line_update_<%=cid%> = 0;
@@ -132,7 +100,7 @@ int tmp_batchUpdateCount_<%=cid%> = 0;
 int deletedCount_<%=cid%>=0;
 int updatedCount_<%=cid%>=0;
 int insertedCount_<%=cid%>=0;
-
+int rowsToCommitCount_<%=cid%>=0;
 String tableName_<%=cid%> = <%=table%>;
 boolean whetherReject_<%=cid%> = false;
 

main/plugins/org.talend.designer.components.localprovider/components/tAS400Output/tAS400Output_main.javajet

@@ -313,6 +313,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                 insertedCount_<%=cid%> = insertedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().inserting(dbLog.var("nb_line"));%>
             } catch(java.sql.SQLException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -445,6 +446,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                 updatedCount_<%=cid%> = updatedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().updating(dbLog.var("nb_line"));%>
             } catch(java.sql.SQLException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -593,6 +595,7 @@ skeleton="../templates/db_output_bulk.skeleton"
         			nb_line_<%=cid%>++;
         			<%dbLog.data().updating(dbLog.var("nb_line"));%>
                 } catch(java.sql.SQLException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -669,6 +672,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                     nb_line_<%=cid%>++;
                     <%dbLog.data().inserting(dbLog.var("nb_line"));%>
                 } catch(java.sql.SQLException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -849,6 +853,7 @@ skeleton="../templates/db_output_bulk.skeleton"
     				<%dbLog.data().updating(dbLog.var("nb_line"));%>
     			}
                 } catch(java.sql.SQLException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -934,6 +939,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                 nb_line_<%=cid%>++;
 				<%dbLog.data().deleting(dbLog.var("nb_line"));%>
             } catch(java.sql.SQLException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {

main/plugins/org.talend.designer.components.localprovider/components/tAccessOutput/tAccessOutput_begin.javajet

@@ -58,39 +58,8 @@ if(columnList != null && columnList.size() > 0) {
 }
 %>
 
-<%
-if(("UPDATE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction) || ("UPDATE_OR_INSERT").equals(dataAction)) {
-    int updateKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isUpdateKey()) {
-                updateKeyCount++;
-            }
-        }
-        %>
-        int updateKeyCount_<%=cid%> = <%=updateKeyCount%>;
-        if(updateKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For update, Schema must have a key");
-        }
-        <%
-    }
-} else if(("DELETE").equals(dataAction)) {
-    int deleteKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isDeleteKey()) {
-                deleteKeyCount++;
-            }
-        }
-        %>
-        int deleteKeyCount_<%=cid%> = <%=deleteKeyCount%>;
-        if(deleteKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For delete, Schema must have a key");
-        }
-        <%
-    }
-}
-%>
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/DB/Output/CheckKeysForUpdateAndDelete.javajet"%>
+
 int nb_line_<%=cid%> = 0;
 int nb_line_update_<%=cid%> = 0;
 int nb_line_inserted_<%=cid%> = 0;
@@ -100,7 +69,7 @@ int nb_line_rejected_<%=cid%> = 0;
 int deletedCount_<%=cid%>=0;
 int updatedCount_<%=cid%>=0;
 int insertedCount_<%=cid%>=0;
-
+int rowsToCommitCount_<%=cid%>=0;
 String tableName_<%=cid%> = <%=tableName%>;
 boolean whetherReject_<%=cid%> = false;
 java.sql.Connection conn_<%=cid%> = null;

main/plugins/org.talend.designer.components.localprovider/components/tAccessOutput/tAccessOutput_main.javajet

@@ -144,6 +144,7 @@ skeleton="../templates/db_output_bulk.skeleton"
             	nb_line_<%=cid%>++;
             	<%dbLog.data().inserting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -234,6 +235,7 @@ skeleton="../templates/db_output_bulk.skeleton"
             	nb_line_<%=cid%>++;
             	<%dbLog.data().updating(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -352,6 +354,7 @@ skeleton="../templates/db_output_bulk.skeleton"
     				nb_line_<%=cid%>++;
     				<%dbLog.data().updating(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -414,6 +417,7 @@ skeleton="../templates/db_output_bulk.skeleton"
         			nb_line_<%=cid%>++;
                     <%dbLog.data().inserting(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -513,6 +517,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                  }
                  updatedCount_<%=cid%> = updatedCount_<%=cid%>+updateFlag_<%=cid%>;
              } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                  whetherReject_<%=cid%> = true;
                  <%
                  if (("true").equals(dieOnError)) {
@@ -577,6 +582,7 @@ skeleton="../templates/db_output_bulk.skeleton"
         			nb_line_<%=cid%>++;
                 	<%dbLog.data().inserting(dbLog.var("nb_line"));%>
                  } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                      whetherReject_<%=cid%> = true;
                      <%
                      if (("true").equals(dieOnError)) {
@@ -656,6 +662,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                 nb_line_<%=cid%>++;
                 <%dbLog.data().deleting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {

main/plugins/org.talend.designer.components.localprovider/components/tAdvancedFileOutputXML/tAdvancedFileOutputXML_begin.javajet

@@ -154,6 +154,7 @@ if ((metadatas != null) && (metadatas.size() > 0)) {
 						org.dom4j.Document doc4Str = org.dom4j.DocumentHelper.parseText("<root>"+ value + "</root>");
 						nestRoot.setContent(doc4Str.getRootElement().content());
 					} catch (java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 						nestRoot.setText(value);
 					}
 				}
@@ -252,6 +253,7 @@ if ((metadatas != null) && (metadatas.size() > 0)) {
 						doc_<%=cid %> = saxReader_<%=cid %>.read(createFile<%=cid %>);
 						nestXMLTool_<%=cid%>.removeWhiteSpaceTextNode(doc_<%=cid %>.getRootElement());
 					} catch (java.lang.Exception ex) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",ex.getMessage());
 						ex.printStackTrace();
 						throw new java.lang.Exception("can not find the file:" + fileName_<%=cid %>);
 					}

main/plugins/org.talend.designer.components.localprovider/components/tAmazonAuroraConnection/tAmazonAuroraConnection_java.xml

@@ -214,7 +214,7 @@
     <TEMPLATE_PARAM SOURCE="self.AUTO_COMMIT" TARGET="in.AUTO_COMMIT"/>
     </TEMPLATES>
      <IMPORTS>
-       <IMPORT NAME="Driver-MysqlConnector" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0" />
+       <IMPORT NAME="Driver-MysqlConnector" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49" />
        <IMPORT NAME="slf4j-api-1.7.25.jar" MODULE="slf4j-api-1.7.25.jar" MVN="mvn:org.slf4j/slf4j-api/1.7.25" REQUIRED_IF="(USE_SHARED_CONNECTION == 'true' AND SPECIFY_DATASOURCE_ALIAS=='false')" />
        <IMPORT NAME="slf4j-log4j12-1.7.25.jar" MODULE="slf4j-log4j12-1.7.25.jar" MVN="mvn:org.slf4j/slf4j-log4j12/1.7.25" REQUIRED_IF="(USE_SHARED_CONNECTION == 'true' AND SPECIFY_DATASOURCE_ALIAS=='false')" />
      </IMPORTS>

main/plugins/org.talend.designer.components.localprovider/components/tAmazonAuroraInput/tAmazonAuroraInput_java.xml

@@ -298,7 +298,7 @@
       <TEMPLATE_PARAM SOURCE="self.TRIM" TARGET="in.TRIM"/>
       </TEMPLATES>
        <IMPORTS>
-         <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0"  REQUIRED_IF="USE_EXISTING_CONNECTION == 'false'" />
+         <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="USE_EXISTING_CONNECTION == 'false'" />
        </IMPORTS>
       </CODEGENERATION>
 

main/plugins/org.talend.designer.components.localprovider/components/tAmazonAuroraOutput/tAmazonAuroraOutput_java.xml

@@ -524,7 +524,7 @@
       <TEMPLATE_PARAM SOURCE="self.PARALLELIZE_NUMBER" TARGET="out.PARALLELIZE_NUMBER"/>
     </TEMPLATES>
      <IMPORTS>
-       <IMPORT NAME="Driver-MysqlConnector" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0"  REQUIRED_IF="USE_EXISTING_CONNECTION == 'false'" />
+       <IMPORT NAME="Driver-MysqlConnector" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="USE_EXISTING_CONNECTION == 'false'" />
      </IMPORTS>
     </CODEGENERATION>
 

main/plugins/org.talend.designer.components.localprovider/components/tAmazonAuroraRow/tAmazonAuroraRow_java.xml

@@ -0,0 +1,325 @@
+<COMPONENT>
+    <HEADER
+            PLATEFORM="ALL"
+            SERIAL=""
+            VERSION="0.102"
+            STATUS="ALPHA"
+            COMPATIBILITY="ALL"
+            AUTHOR="Talend"
+            RELEASE_DATE="20070314A"
+            STARTABLE="true"
+            DATA_AUTO_PROPAGATE="false"
+            HAS_CONDITIONAL_OUTPUTS="true"
+            SCHEMA_AUTO_PROPAGATE="false"
+            PARTITIONING="AUTO"
+            LOG4J_ENABLED="true"
+    >
+        <SIGNATURE/>
+    </HEADER>
+
+    <FAMILIES>
+        <FAMILY>Databases/DB Specifics/Amazon/Aurora</FAMILY>
+        <FAMILY>Cloud/Amazon/Aurora</FAMILY>
+    </FAMILIES>
+
+    <DOCUMENTATION>
+        <URL/>
+    </DOCUMENTATION>
+
+    <CONNECTORS>
+        <CONNECTOR CTYPE="FLOW" MAX_INPUT="1" MAX_OUTPUT="1"/>
+        <CONNECTOR NAME="REJECT" NOT_SHOW_IF="(DIE_ON_ERROR == 'true')" CTYPE="FLOW" MAX_INPUT="0" MAX_OUTPUT="1" LINE_STYLE="2" COLOR="FF0000" BASE_SCHEMA="FLOW" />
+        <CONNECTOR CTYPE="ITERATE" MAX_OUTPUT="1" MAX_INPUT="1"/>
+        <CONNECTOR CTYPE="SUBJOB_OK" MAX_INPUT="1" />
+        <CONNECTOR CTYPE="SUBJOB_ERROR" MAX_INPUT="1" />
+        <CONNECTOR CTYPE="COMPONENT_OK"/>
+        <CONNECTOR CTYPE="COMPONENT_ERROR"/>
+        <CONNECTOR CTYPE="RUN_IF"/>
+    </CONNECTORS>
+
+    <PARAMETERS>
+        <PARAMETER
+                NAME="PROPERTY"
+                FIELD="PROPERTY_TYPE"
+                NUM_ROW="1"
+                REPOSITORY_VALUE="DATABASE:AMAZON_AURORA"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        />
+
+        <PARAMETER
+                NAME="DB_VERSION"
+                FIELD="CLOSED_LIST"
+                NUM_ROW="10"
+                REPOSITORY_VALUE="DB_VERSION"
+                READONLY="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+                SHOW="false">
+            <ITEMS DEFAULT="MYSQL_5">
+                <ITEM NAME="MYSQL_5" VALUE="MYSQL_5" />
+            </ITEMS>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="USE_EXISTING_CONNECTION"
+                FIELD="CHECK"
+                NUM_ROW="20"
+        >
+            <DEFAULT>false</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="CONNECTION"
+                FIELD="COMPONENT_LIST"
+                FILTER="tAmazonAuroraConnection"
+                REQUIRED="true"
+                NUM_ROW="20"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'true'"
+                DYNAMIC_SETTINGS="true"
+        />
+
+        <PARAMETER
+                NAME="HOST"
+                FIELD="TEXT"
+                NUM_ROW="30"
+                REPOSITORY_VALUE="SERVER_NAME"
+                REQUIRED="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="PORT"
+                FIELD="TEXT"
+                NUM_ROW="30"
+                REPOSITORY_VALUE="PORT"
+                REQUIRED="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        >
+            <DEFAULT>"3306"</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="DBNAME"
+                FIELD="TEXT"
+                NUM_ROW="40"
+                REPOSITORY_VALUE="SID"
+                REQUIRED="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="TYPE"
+                FIELD="TEXT"
+                NUM_ROW="40"
+                SHOW="false"
+                REPOSITORY_VALUE="TYPE"
+        >
+            <DEFAULT>AMAZON_AURORA</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="USER"
+                FIELD="TEXT"
+                NUM_ROW="50"
+                REPOSITORY_VALUE="USERNAME"
+                REQUIRED="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="PASS"
+                FIELD="PASSWORD"
+                NUM_ROW="50"
+                REPOSITORY_VALUE="PASSWORD"
+                REQUIRED="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="SCHEMA_FLOW"
+                FIELD="SCHEMA_TYPE"
+                NUM_ROW="60"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="SCHEMA_REJECT"
+                FIELD="SCHEMA_TYPE"
+                REQUIRED="true"
+                NUM_ROW="60"
+                CONTEXT="REJECT"
+                SHOW="true">
+            <TABLE READONLY="true">
+                <COLUMN NAME="errorCode" TYPE="id_String" LENGTH="255" READONLY="false" CUSTOM="true"/>
+                <COLUMN NAME="errorMessage" TYPE="id_String" LENGTH="255" READONLY="false" CUSTOM="true"/>
+            </TABLE>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="TABLE"
+                FIELD="DBTABLE"
+                NUM_ROW="60"
+                LOG4J_ENABLED="false"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="QUERYSTORE"
+                FIELD="QUERYSTORE_TYPE"
+                NUM_ROW="70"
+        >
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="QUERY"
+                FIELD="MEMO_SQL"
+                NUM_ROW="80"
+        >
+            <DEFAULT>"select id, name from employee"</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="DIE_ON_ERROR"
+                FIELD="CHECK"
+                NUM_ROW="90"
+        >
+            <DEFAULT>false</DEFAULT>
+        </PARAMETER>
+    </PARAMETERS>
+
+    <ADVANCED_PARAMETERS>
+        <PARAMETER NAME="PROPERTIES" FIELD="TEXT" NUM_ROW="4" REPOSITORY_VALUE="PROPERTIES_STRING" SHOW_IF="USE_EXISTING_CONNECTION == 'false'">
+            <DEFAULT>""</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="PROPAGATE_RECORD_SET"
+                FIELD="CHECK"
+                NUM_ROW="5"
+        >
+            <DEFAULT>false</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="RECORD_SET_COLUMN"
+                FIELD="COLUMN_LIST"
+                NUM_ROW="5"
+                SHOW_IF="PROPAGATE_RECORD_SET == 'true'"
+        />
+
+        <PARAMETER
+                NAME="USE_PREPAREDSTATEMENT"
+                FIELD="CHECK"
+                NUM_ROW="6"
+        >
+            <DEFAULT>false</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="SET_PREPAREDSTATEMENT_PARAMETERS"
+                FIELD="TABLE"
+                REQUIRED="false"
+                NUM_ROW="7"
+                NB_LINES="3"
+                SHOW_IF="(USE_PREPAREDSTATEMENT == 'true')"
+        >
+            <ITEMS>
+                <ITEM NAME="PARAMETER_INDEX" VALUE="" />
+                <ITEM NAME="PARAMETER_TYPE" VALUE="String" FIELD="CLOSED_LIST">
+                    <ITEMS DEFAULT="String">
+                        <ITEM NAME="BigDecimal"  	VALUE="BigDecimal"/>
+                        <ITEM NAME="Blob"  			VALUE="Blob"/>
+                        <ITEM NAME="Boolean"  		VALUE="Boolean"/>
+                        <ITEM NAME="Byte"  			VALUE="Byte"/>
+                        <ITEM NAME="Bytes"  		VALUE="Bytes"/>
+                        <ITEM NAME="Clob"  			VALUE="Clob"/>
+                        <ITEM NAME="Date" 			VALUE="Date"/>
+                        <ITEM NAME="Double"			VALUE="Double"/>
+                        <ITEM NAME="Float" 			VALUE="Float"/>
+                        <ITEM NAME="Int" 			VALUE="Int"/>
+                        <ITEM NAME="Long" 			VALUE="Long"/>
+                        <ITEM NAME="Object" 		VALUE="Object"/>
+                        <ITEM NAME="Short" 			VALUE="Short"/>
+                        <ITEM NAME="String" 		VALUE="String"/>
+                        <ITEM NAME="Time" 			VALUE="Time"/>
+                        <ITEM NAME="Null" 			VALUE="Null"/>
+                    </ITEMS>
+                </ITEM>
+                <ITEM NAME="PARAMETER_VALUE" VALUE="" />
+            </ITEMS>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="ENCODING"
+                FIELD="ENCODING_TYPE"
+                NUM_ROW="9"
+                REQUIRED="true"
+                SHOW="false"
+        >
+            <DEFAULT>"ISO-8859-15"</DEFAULT>
+        </PARAMETER>
+
+        <PARAMETER
+                NAME="COMMIT_EVERY"
+                FIELD="TEXT"
+                NUM_ROW="9"
+                REQUIRED="true"
+                SHOW_IF="USE_EXISTING_CONNECTION == 'false'"
+        >
+            <DEFAULT>10000</DEFAULT>
+        </PARAMETER>
+    </ADVANCED_PARAMETERS>
+
+    <CODEGENERATION>
+        <TEMPLATES INPUT="in" OUTPUT="out">
+            <TEMPLATE NAME="in" COMPONENT="tMysqlRow">
+                <LINK_TO NAME="out" CTYPE="FLOW"/>
+            </TEMPLATE>
+            <TEMPLATE NAME="out" COMPONENT="tDummyRow"/>
+            <TEMPLATE_PARAM SOURCE="self.PROPERTY" TARGET="in.PROPERTY"/>
+            <TEMPLATE_PARAM SOURCE="self.DB_VERSION" TARGET="in.DB_VERSION"/>
+            <TEMPLATE_PARAM SOURCE="self.USE_EXISTING_CONNECTION" TARGET="in.USE_EXISTING_CONNECTION"/>
+            <TEMPLATE_PARAM SOURCE="self.CONNECTION" TARGET="in.CONNECTION"/>
+            <TEMPLATE_PARAM SOURCE="self.HOST" TARGET="in.HOST"/>
+            <TEMPLATE_PARAM SOURCE="self.PORT" TARGET="in.PORT"/>
+            <TEMPLATE_PARAM SOURCE="self.DBNAME" TARGET="in.DBNAME"/>
+            <TEMPLATE_PARAM SOURCE="self.TYPE" TARGET="in.TYPE"/>
+            <TEMPLATE_PARAM SOURCE="self.USER" TARGET="in.USER"/>
+            <TEMPLATE_PARAM SOURCE="self.PASS" TARGET="in.PASS"/>
+            <TEMPLATE_PARAM SOURCE="self.SCHEMA_FLOW" TARGET="in.SCHEMA_FLOW"/>
+            <TEMPLATE_PARAM SOURCE="self.SCHEMA_REJECT" TARGET="in.SCHEMA_REJECT"/>
+            <TEMPLATE_PARAM SOURCE="self.TABLE" TARGET="in.TABLE"/>
+            <TEMPLATE_PARAM SOURCE="self.QUERYSTORE" TARGET="in.QUERYSTORE"/>
+            <TEMPLATE_PARAM SOURCE="self.QUERY" TARGET="in.QUERY"/>
+            <TEMPLATE_PARAM SOURCE="self.DIE_ON_ERROR" TARGET="in.DIE_ON_ERROR"/>
+            <TEMPLATE_PARAM SOURCE="self.PROPERTIES" TARGET="in.PROPERTIES"/>
+            <TEMPLATE_PARAM SOURCE="self.PROPAGATE_RECORD_SET" TARGET="in.PROPAGATE_RECORD_SET"/>
+            <TEMPLATE_PARAM SOURCE="self.RECORD_SET_COLUMN" TARGET="in.RECORD_SET_COLUMN"/>
+            <TEMPLATE_PARAM SOURCE="self.USE_PREPAREDSTATEMENT" TARGET="in.USE_PREPAREDSTATEMENT"/>
+            <TEMPLATE_PARAM SOURCE="self.SET_PREPAREDSTATEMENT_PARAMETERS" TARGET="in.SET_PREPAREDSTATEMENT_PARAMETERS"/>
+            <TEMPLATE_PARAM SOURCE="self.PARAMETER_INDEX" TARGET="in.PARAMETER_INDEX"/>
+            <TEMPLATE_PARAM SOURCE="self.PARAMETER_TYPE" TARGET="in.PARAMETER_TYPE"/>
+            <TEMPLATE_PARAM SOURCE="self.PARAMETER_VALUE" TARGET="in.PARAMETER_VALUE"/>
+            <TEMPLATE_PARAM SOURCE="self.ENCODING" TARGET="in.ENCODING"/>
+            <TEMPLATE_PARAM SOURCE="self.COMMIT_EVERY" TARGET="in.COMMIT_EVERY"/>
+
+        </TEMPLATES>
+        <IMPORTS>
+            <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
+        </IMPORTS>
+    </CODEGENERATION>
+
+    <RETURNS>
+        <RETURN NAME="QUERY" TYPE="id_String" AVAILABILITY="FLOW"/>
+    </RETURNS>
+</COMPONENT>

main/plugins/org.talend.designer.components.localprovider/components/tAmazonAuroraRow/tAmazonAuroraRow_messages.properties

@@ -0,0 +1,60 @@
+#Created by JInto - www.guh-software.de
+#Thu Aug 30 19:54:54 CST 2007
+COMMIT_EVERY.NAME=Commit every
+DBD-mysql.INFO=Required for MySQL
+DBNAME.NAME=Database
+DBTABLE.NAME=Table Name
+DIE_ON_ERROR.NAME=Die on error
+ENCODING.NAME=Encoding
+HELP=org.talend.help.tAmazonAuroraRow
+HOST.NAME=Host
+LONG_NAME=Executes an SQL query at each of Talend flow's iterations
+NB_LINE.NAME=Number of line
+PASS.NAME=Password
+PORT.NAME=Port
+PREPARED_STATEMENT.NAME=Prepared statement
+QUERY.NAME=Query
+REJECT.LINK=Rejects
+REJECT.MENU=Rejects
+SCHEMA_DB.NAME=Schema
+STATEMENT_PARAMS.ITEM.INPUT_COLUMN=Input column
+STATEMENT_PARAMS.NAME=Parameters
+TYPE.ITEM.MSSQL_ODBC=Microsoft SQL (Odbc driver)
+TYPE.ITEM.MYSQL=MySQL
+TYPE.NAME=Database Driver
+USER.NAME=Username
+USE_EXISTING_CONNECTION.NAME=Use an existing connection
+PROPERTIES.NAME=Additional JDBC Parameters
+PROPAGATE_RECORD_SET.NAME=Propagate QUERY's recordset
+RECORD_SET_COLUMN.NAME=use column
+TABLE.NAME=Table Name
+PROPERTY.NAME=Property Type
+CONNECTION.NAME=Component List
+SCHEMA_FLOW.NAME=Schema
+SCHEMA_REJECT.NAME=Reject Schema
+QUERYSTORE.NAME=Query Type
+
+DB_VERSION.NAME=DB Version
+DB_VERSION.ITEM.MYSQL_5=Mysql 5
+
+USE_PREPAREDSTATEMENT.NAME=Use PreparedStatement
+SET_PREPAREDSTATEMENT_PARAMETERS.NAME=Set PreparedStatement Parameters
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_INDEX=Parameter Index
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE=Parameter Type
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.BigDecimal=BigDecimal
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Blob=Blob
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Boolean=Boolean
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Byte=Byte
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Bytes=Bytes
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Clob=Clob
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Date=Date
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Double=Double
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Float=Float
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Int=Int
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Long=Long
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Object=Object
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Short=Short
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.String=String
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Time=Time
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_TYPE.ITEM.Null=Null
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_VALUE=Parameter Value

main/plugins/org.talend.designer.components.localprovider/components/tAmazonEMRListInstances/tAmazonEMRListInstances_java.xml

@@ -231,20 +231,20 @@
 			<IMPORT NAME="aws-java-sdk-1.11.848.jar" MODULE="aws-java-sdk-1.11.848.jar"
 				MVN="mvn:com.amazonaws/aws-java-sdk/1.11.848"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-core-2.10.1.jar" MODULE="jackson-core-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.10.1"
+			<IMPORT NAME="jackson-core-2.11.4.jar" MODULE="jackson-core-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-databind-2.10.1.jar" MODULE="jackson-databind-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.10.1"
+			<IMPORT NAME="jackson-databind-2.11.4.jar" MODULE="jackson-databind-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-annotations-2.10.1.jar" MODULE="jackson-annotations-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.10.1"
+			<IMPORT NAME="jackson-annotations-2.11.4.jar" MODULE="jackson-annotations-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.11.4"
 				REQUIRED="true" />
 			<IMPORT NAME="httpcore-4.4.13.jar" MODULE="httpcore-4.4.13.jar"
 				MVN="mvn:org.apache.httpcomponents/httpcore/4.4.13"
 				REQUIRED="true" />
-			<IMPORT NAME="httpclient-4.5.12.jar" MODULE="httpclient-4.5.12.jar"
-				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.12"
+			<IMPORT NAME="httpclient-4.5.13.jar" MODULE="httpclient-4.5.13.jar"
+				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.13"
 				REQUIRED="true" />
 			<IMPORT NAME="joda-time-2.8.1.jar" MODULE="joda-time-2.8.1.jar"
 				MVN="mvn:joda-time/joda-time/2.8.1"

main/plugins/org.talend.designer.components.localprovider/components/tAmazonEMRManage/tAmazonEMRManage_begin.javajet

@@ -320,6 +320,7 @@
 			try {
 				configs_<%=cid%> = java.util.Arrays.asList(mapper_<%=cid%>.readValue(<%=property_json%>, com.amazonaws.services.elasticmapreduce.model.Configuration[].class));
 			} catch (com.fasterxml.jackson.core.JsonProcessingException e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 				e.printStackTrace();
 			}
 		}

main/plugins/org.talend.designer.components.localprovider/components/tAmazonEMRManage/tAmazonEMRManage_java.xml

@@ -558,20 +558,20 @@
 			<IMPORT NAME="aws-java-sdk-1.11.848.jar" MODULE="aws-java-sdk-1.11.848.jar"
 				MVN="mvn:com.amazonaws/aws-java-sdk/1.11.848"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-core-2.10.1.jar" MODULE="jackson-core-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.10.1"
+			<IMPORT NAME="jackson-core-2.11.4.jar" MODULE="jackson-core-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-databind-2.10.1.jar" MODULE="jackson-databind-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.10.1"
+			<IMPORT NAME="jackson-databind-2.11.4.jar" MODULE="jackson-databind-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-annotations-2.10.1.jar" MODULE="jackson-annotations-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.10.1"
+			<IMPORT NAME="jackson-annotations-2.11.4.jar" MODULE="jackson-annotations-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.11.4"
 				REQUIRED="true" />
 			<IMPORT NAME="httpcore-4.4.13.jar" MODULE="httpcore-4.4.13.jar"
 				MVN="mvn:org.apache.httpcomponents/httpcore/4.4.13"
 				REQUIRED="true" />
-			<IMPORT NAME="httpclient-4.5.12.jar" MODULE="httpclient-4.5.12.jar"
-				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.12"
+			<IMPORT NAME="httpclient-4.5.13.jar" MODULE="httpclient-4.5.13.jar"
+				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.13"
 				REQUIRED="true" />
 			<IMPORT NAME="joda-time-2.8.1.jar" MODULE="joda-time-2.8.1.jar"
 				MVN="mvn:joda-time/joda-time/2.8.1"

main/plugins/org.talend.designer.components.localprovider/components/tAmazonEMRResize/tAmazonEMRResize_java.xml

@@ -288,20 +288,20 @@
 			<IMPORT NAME="aws-java-sdk-1.11.848.jar" MODULE="aws-java-sdk-1.11.848.jar"
 				MVN="mvn:com.amazonaws/aws-java-sdk/1.11.848"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-core-2.10.1.jar" MODULE="jackson-core-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.10.1"
+			<IMPORT NAME="jackson-core-2.11.4.jar" MODULE="jackson-core-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-databind-2.10.1.jar" MODULE="jackson-databind-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.10.1"
+			<IMPORT NAME="jackson-databind-2.11.4.jar" MODULE="jackson-databind-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-annotations-2.10.1.jar" MODULE="jackson-annotations-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.10.1"
+			<IMPORT NAME="jackson-annotations-2.11.4.jar" MODULE="jackson-annotations-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.11.4"
 				REQUIRED="true" />
 			<IMPORT NAME="httpcore-4.4.13.jar" MODULE="httpcore-4.4.13.jar"
 				MVN="mvn:org.apache.httpcomponents/httpcore/4.4.13"
 				REQUIRED="true" />
-			<IMPORT NAME="httpclient-4.5.12.jar" MODULE="httpclient-4.5.12.jar"
-				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.12"
+			<IMPORT NAME="httpclient-4.5.13.jar" MODULE="httpclient-4.5.13.jar"
+				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.13"
 				REQUIRED="true" />
 			<IMPORT NAME="joda-time-2.8.1.jar" MODULE="joda-time-2.8.1.jar"
 				MVN="mvn:joda-time/joda-time/2.8.1"

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlConnection/tAmazonMysqlConnection_begin.javajet

@@ -11,25 +11,10 @@ imports="
 	
 		public void createURL(INode node) {
 			super.createURL(node);
+			final boolean supportBulkComponent = false;
+			final String jdbcURL = "jdbc:mysql";
 %>
-<%
-			if(dbproperties == null || ("\"\"").equals(dbproperties) || ("").equals(dbproperties)) {
-%>
-		String url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%> + "?" + "rewriteBatchedStatements=true";
-<%
-			} else {
-%>
-		String proprtties_<%=cid%> = <%=dbproperties%>;
-		if (proprtties_<%=cid%> == null || proprtties_<%=cid%>.trim().length() == 0) {
-			proprtties_<%=cid%> += "rewriteBatchedStatements=true";
-		}else if (proprtties_<%=cid%> != null && !proprtties_<%=cid%>.contains("rewriteBatchedStatements")) {
-			proprtties_<%=cid%> += "&rewriteBatchedStatements=true";
-		}
-		
-		String url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%> + "?" + proprtties_<%=cid%>;
-<%
-			}
-%>
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/Mysql/jdbcurl4connection_output.javajet"%>
 <%	
 		}
 		

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlConnection/tAmazonMysqlConnection_java.xml

@@ -162,7 +162,7 @@
 
   <CODEGENERATION>
      <IMPORTS>
-       <IMPORT NAME="Driver-MysqlConnector" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0"  REQUIRED_IF="DB_VERSION == 'MYSQL_5'" />
+       <IMPORT NAME="Driver-MysqlConnector" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="DB_VERSION == 'MYSQL_5'" />
        <IMPORT NAME="Driver-MysqlConnector_Old" MODULE="mysql-connector-java-5.1.0-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.0-bin/6.0.0"  REQUIRED="false" />
        <IMPORT NAME="slf4j-api-1.7.25.jar" MODULE="slf4j-api-1.7.25.jar" MVN="mvn:org.slf4j/slf4j-api/1.7.25" REQUIRED_IF="USE_SHARED_CONNECTION =='true'" />
        <IMPORT NAME="slf4j-log4j12-1.7.25.jar" MODULE="slf4j-log4j12-1.7.25.jar" MVN="mvn:org.slf4j/slf4j-log4j12/1.7.25" REQUIRED_IF="USE_SHARED_CONNECTION =='true'" />

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlInput/tAmazonMysqlInput_begin.javajet

@@ -28,16 +28,10 @@ imports="
 	    }
 	    
 		public void setURL(INode node) {
-			
-			if(dbproperties == null || ("\"\"").equals(dbproperties) || ("").equals(dbproperties)) {
+			final String jdbcURL = "jdbc:mysql";
 %>
-				String url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%>;
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/Mysql/jdbcurl4basic.javajet"%>
 <%
-	        } else {
-%>
-				String url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%> + "?" + <%=dbproperties%>;
-<%
-	        }		
 		}
 		
 		public String getDirverClassName(INode node){

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlInput/tAmazonMysqlInput_java.xml

@@ -240,7 +240,7 @@
 
    <CODEGENERATION>
      <IMPORTS>
-       <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
+       <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
      </IMPORTS>
     </CODEGENERATION>
 

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlOutput/tAmazonMysqlOutput_begin.javajet

@@ -120,39 +120,10 @@ if(columnList != null && columnList.size() > 0) {
 }
 %>
 
-<%
-if(("UPDATE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction) || ("UPDATE_OR_INSERT").equals(dataAction)) {
-    int updateKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isUpdateKey() && !column.isDynamic()) {
-                updateKeyCount++;
-            }
-        }
-        %>
-        int updateKeyCount_<%=cid%> = <%=updateKeyCount%>;
-        if(updateKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For update, Schema must have a key");
-        }
-        <%
-    }
-} else if(("DELETE").equals(dataAction)) {
-    int deleteKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isDeleteKey() && !column.isDynamic()) {
-                deleteKeyCount++;
-            }
-        }
-        %>
-        int deleteKeyCount_<%=cid%> = <%=deleteKeyCount%>;
-        if(deleteKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For delete, Schema must have a key");
-        }
-        <%
-    }
-}
-%>
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/DB/Output/CheckKeysForUpdateAndDelete.javajet"%>
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/DB/Output/CheckKeysForUpdateOnDuplicateKey.javajet"%>
+
+
 
 int nb_line_<%=cid%> = 0;
 int nb_line_update_<%=cid%> = 0;
@@ -163,7 +134,7 @@ int nb_line_rejected_<%=cid%> = 0;
 int deletedCount_<%=cid%>=0;
 int updatedCount_<%=cid%>=0;
 int insertedCount_<%=cid%>=0;
-
+int rowsToCommitCount_<%=cid%>=0;
 int rejectedCount_<%=cid%>=0;
 
 String tableName_<%=cid%> = <%=table%>;
@@ -186,19 +157,13 @@ if(useExistingConnection) {
 	<%
 } else {
 	String javaDbDriver   = "org.gjt.mm.mysql.Driver";
+	
+	final boolean supportBulkComponent = false;
+	final String jdbcURL = "jdbc:mysql";
 	%>
-	String dbProperties_<%=cid%> = <%=dbproperties%>;
-	String url_<%=cid%> = null;
-	if(dbProperties_<%=cid%> == null || dbProperties_<%=cid%>.trim().length() == 0) {
-	    url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%> + "?" + "rewriteBatchedStatements=true";
-	} else {
-		String properties_<%=cid%> = <%=dbproperties%>;
-		if (!properties_<%=cid%>.contains("rewriteBatchedStatements")) {
-			properties_<%=cid%> += "&rewriteBatchedStatements=true";
-		}
-
-	    url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%> + "?" + properties_<%=cid%>;
-	}
+	
+	<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/Mysql/jdbcurl4connection_output.javajet"%>
+	
     String dbUser_<%=cid %> = <%=dbuser%>;
 
     <%

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlOutput/tAmazonMysqlOutput_end.javajet

@@ -131,6 +131,7 @@ skeleton="../templates/db_output_bulk.skeleton"
             	    	}
 
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);
@@ -211,6 +212,7 @@ skeleton="../templates/db_output_bulk.skeleton"
 	    	<%
 	    	}%>
     }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
     	<%if(("true").equals(dieOnError)) {
     	%>
     		throw(e);

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlOutput/tAmazonMysqlOutput_java.xml

@@ -423,7 +423,7 @@
 
   <CODEGENERATION>
      <IMPORTS>
-       <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
+       <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
      </IMPORTS>
     </CODEGENERATION>
 

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlOutput/tAmazonMysqlOutput_main.javajet

@@ -634,6 +634,7 @@ if(columnList != null && columnList.size() > 0) {
 	                    insertedCount_<%=cid%> = insertedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
 	                    <%dbLog.data().inserting(dbLog.var("nb_line"));%>
 	                } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 	                    whetherReject_<%=cid%> = true;
 	                    <%
 	                    if (("true").equals(dieOnError)) {
@@ -712,6 +713,7 @@ if(columnList != null && columnList.size() > 0) {
                     insertedCount_<%=cid%> = insertedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                     <%dbLog.data().inserting(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -847,6 +849,7 @@ if(columnList != null && columnList.size() > 0) {
                 updatedCount_<%=cid%> = updatedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().updating(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1000,6 +1003,7 @@ if(columnList != null && columnList.size() > 0) {
                     nb_line_<%=cid%>++;
         			<%dbLog.data().updating(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1076,6 +1080,7 @@ if(columnList != null && columnList.size() > 0) {
                     nb_line_<%=cid%>++;
                     <%dbLog.data().inserting(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1259,6 +1264,7 @@ if(columnList != null && columnList.size() > 0) {
     				<%dbLog.data().updating(dbLog.var("nb_line"));%>
     			}
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1346,6 +1352,7 @@ if(columnList != null && columnList.size() > 0) {
             	nb_line_<%=cid%>++;
             	<%dbLog.data().deleting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1413,6 +1420,7 @@ if(columnList != null && columnList.size() > 0) {
                 replaceCount_<%=cid%> = pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().replacing(null);%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1493,6 +1501,7 @@ if(columnList != null && columnList.size() > 0) {
                 count_on_duplicate_key_<%=cid%> = pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().insertingOnDuplicateKeyUpdating(null);%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1559,6 +1568,7 @@ if(columnList != null && columnList.size() > 0) {
                     insertedCount_<%=cid%> = insertedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                     <%dbLog.data().insertingOnDuplicateKeyUpdating(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1635,6 +1645,7 @@ if(columnList != null && columnList.size() > 0) {
             	    	}
             	    	%>
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);
@@ -1687,6 +1698,7 @@ if(columnList != null && columnList.size() > 0) {
             	    	}%>
             	    	batchSizeCounter_<%=cid%> = 0;
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);
@@ -1743,6 +1755,7 @@ if(columnList != null && columnList.size() > 0) {
             	    	}%>
                         batchSizeCounter_<%=cid%> = 0;
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
 						throw(e);

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlRow/tAmazonMysqlRow_begin.javajet

@@ -16,16 +16,10 @@ imports="
 	    }
 	    
 		public void setURL(INode node) {
-			
-			if(dbproperties == null || ("\"\"").equals(dbproperties) || ("").equals(dbproperties)) {
+			final String jdbcURL = "jdbc:mysql";
 %>
-				String url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%>;
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/Mysql/jdbcurl4basic.javajet"%>
 <%
-	        } else {
-%>
-				String url_<%=cid%> = "jdbc:mysql://" + <%=dbhost%> + ":" + <%=dbport%> + "/" + <%=dbname%> + "?" + <%=dbproperties%>;
-<%
-	        }		
 		}
 		
 		public String getDirverClassName(INode node){

main/plugins/org.talend.designer.components.localprovider/components/tAmazonMysqlRow/tAmazonMysqlRow_java.xml

@@ -281,7 +281,7 @@
 
   <CODEGENERATION>
      <IMPORTS>
-       <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.30-bin.jar" MVN="mvn:org.talend.libraries/mysql-connector-java-5.1.30-bin/6.0.0"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
+       <IMPORT NAME="Driver-Mysql5" MODULE="mysql-connector-java-5.1.49.jar" MVN="mvn:mysql/mysql-connector-java/5.1.49"  REQUIRED_IF="(DB_VERSION == 'MYSQL_5') AND (USE_EXISTING_CONNECTION == 'false')" />
      </IMPORTS>
     </CODEGENERATION>
 

main/plugins/org.talend.designer.components.localprovider/components/tAmazonOracleOutput/tAmazonOracleOutput_begin.javajet

@@ -144,39 +144,8 @@ for(Column column : stmtStructure) {
 }
 %>
 
-<%
-if(("UPDATE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction) || ("UPDATE_OR_INSERT").equals(dataAction)) {
-    int updateKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isUpdateKey() && !column.isDynamic()) {
-                updateKeyCount++;
-            }
-        }
-        %>
-        int updateKeyCount_<%=cid%> = <%=updateKeyCount%>;
-        if(updateKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For update, Schema must have a key");
-        }
-        <%
-    }
-} else if(("DELETE").equals(dataAction)) {
-    int deleteKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isDeleteKey() && !column.isDynamic()) {
-                deleteKeyCount++;
-            }
-        }
-        %>
-        int deleteKeyCount_<%=cid%> = <%=deleteKeyCount%>;
-        if(deleteKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For delete, Schema must have a key");
-        }
-        <%
-    }
-}
-%>
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/DB/Output/CheckKeysForUpdateAndDelete.javajet"%>
+
 int nb_line_<%=cid%> = 0;
 int nb_line_update_<%=cid%> = 0;
 int nb_line_inserted_<%=cid%> = 0;
@@ -188,6 +157,7 @@ int tmp_batchUpdateCount_<%=cid%> = 0;
 int deletedCount_<%=cid%>=0;
 int updatedCount_<%=cid%>=0;
 int insertedCount_<%=cid%>=0;
+int rowsToCommitCount_<%=cid%>=0;
 int rejectedCount_<%=cid%>=0;
 
 boolean whetherReject_<%=cid%> = false;
@@ -485,10 +455,12 @@ if (xmlTypeExist || clobTypeExist) {
 				// Close the temporary CLOB
 				tempClob.close();
 			} catch( java.sql.SQLException sqlexp){
+                globalMap.put("<%=cid%>_ERROR_MESSAGE",sqlexp.getMessage());
 	  			<%dbLog.logPrintedException("sqlexp.getMessage()");%>
 				tempClob.freeTemporary();
 				sqlexp.printStackTrace();
 			} catch(java.lang.Exception exp){
+                globalMap.put("<%=cid%>_ERROR_MESSAGE",exp.getMessage());
 	  			<%dbLog.logPrintedException("exp.getMessage()");%>
 				tempClob.freeTemporary();
 				exp.printStackTrace();

main/plugins/org.talend.designer.components.localprovider/components/tAmazonOracleOutput/tAmazonOracleOutput_end.javajet

@@ -128,6 +128,7 @@ imports="
 					stmtCreateIndex_<%=cid%>.execute(dropIndex_<%=cid %>);
 					<%dbLog.data().sqlExecuteDone(dbLog.var("dropIndex"));%>
 				} catch (java.sql.SQLException e) {
+                    globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 					<%dbLog.logIgnoredException("e.getMessage()");%>
 				}
 
@@ -182,6 +183,7 @@ imports="
 						<%dbLog.batch().executeDone(dbLog.str(dataAction));%>
             	    }
                 }catch (java.sql.BatchUpdateException e){
+                    globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);
@@ -242,4 +244,4 @@ imports="
 	}
     %>
 
-<%@ include file="../templates/DB/Output/DBOutputEndGlobalVars.javajet"%>
+<%@ include file="../templates/DB/Output/DBOutputEndGlobalVars.javajet"%>

main/plugins/org.talend.designer.components.localprovider/components/tAmazonOracleOutput/tAmazonOracleOutput_main.javajet

@@ -457,6 +457,7 @@ IMetadataColumn talendDynCol = null;
                 insertedCount_<%=cid%> = insertedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().inserting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+                globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -599,6 +600,7 @@ IMetadataColumn talendDynCol = null;
                 updatedCount_<%=cid%> = updatedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().updating(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+                globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -765,6 +767,7 @@ IMetadataColumn talendDynCol = null;
         			nb_line_<%=cid%>++;
         			<%dbLog.data().updating(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+                    globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -846,6 +849,7 @@ IMetadataColumn talendDynCol = null;
                     nb_line_<%=cid%>++;
                     <%dbLog.data().inserting(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+                    globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1037,6 +1041,7 @@ IMetadataColumn talendDynCol = null;
     				<%dbLog.data().updating(dbLog.var("nb_line"));%>
     			}
                 } catch(java.lang.Exception e) {
+                    globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1127,6 +1132,7 @@ IMetadataColumn talendDynCol = null;
                 nb_line_<%=cid%>++;
                 <%dbLog.data().deleting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+                globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1196,6 +1202,7 @@ IMetadataColumn talendDynCol = null;
 						pstmt_<%=cid %>.executeBatch();
 						<%dbLog.batch().executeDone(dbLog.str(dataAction));%>
                 	}catch (java.sql.BatchUpdateException e){
+                    globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);
@@ -1244,6 +1251,7 @@ IMetadataColumn talendDynCol = null;
 	                		pstmt_<%=cid %>.executeBatch();
 		                	<%dbLog.batch().executeDone(dbLog.str(dataAction));%>
 		                }catch (java.sql.BatchUpdateException e){
+                            globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 		                	<%if(("true").equals(dieOnError)) {
 		                	%>
 								throw(e);

main/plugins/org.talend.designer.components.localprovider/components/tAmazonOracleRow/tAmazonOracleRow_main.javajet

@@ -165,6 +165,7 @@ try {
 	}//end feature 0010425
 %>
 	} catch (java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 		whetherReject_<%=cid%> = true;
 		<%
 		if (("true").equals(dieOnError)) {

main/plugins/org.talend.designer.components.localprovider/components/tAmazonRedshiftManage/tAmazonRedshiftManage_java.xml

@@ -348,20 +348,20 @@
 			<IMPORT NAME="aws-java-sdk-1.11.848.jar" MODULE="aws-java-sdk-1.11.848.jar"
 				MVN="mvn:com.amazonaws/aws-java-sdk/1.11.848"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-core-2.10.1.jar" MODULE="jackson-core-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.10.1"
+			<IMPORT NAME="jackson-core-2.11.4.jar" MODULE="jackson-core-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-databind-2.10.1.jar" MODULE="jackson-databind-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.10.1"
+			<IMPORT NAME="jackson-databind-2.11.4.jar" MODULE="jackson-databind-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-databind/2.11.4"
 				REQUIRED="true" />
-			<IMPORT NAME="jackson-annotations-2.10.1.jar" MODULE="jackson-annotations-2.10.1.jar"
-				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.10.1"
+			<IMPORT NAME="jackson-annotations-2.11.4.jar" MODULE="jackson-annotations-2.11.4.jar"
+				MVN="mvn:com.fasterxml.jackson.core/jackson-annotations/2.11.4"
 				REQUIRED="true" />
 			<IMPORT NAME="httpcore-4.4.13.jar" MODULE="httpcore-4.4.13.jar"
 				MVN="mvn:org.apache.httpcomponents/httpcore/4.4.13"
 				REQUIRED="true" />
-			<IMPORT NAME="httpclient-4.5.12.jar" MODULE="httpclient-4.5.12.jar"
-				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.12"
+			<IMPORT NAME="httpclient-4.5.13.jar" MODULE="httpclient-4.5.13.jar"
+				MVN="mvn:org.apache.httpcomponents/httpclient/4.5.13"
 				REQUIRED="true" />
 			<IMPORT NAME="joda-time-2.8.1.jar" MODULE="joda-time-2.8.1.jar"
 				MVN="mvn:joda-time/joda-time/2.8.1"

main/plugins/org.talend.designer.components.localprovider/components/tApacheLogInput/tApacheLogInput_begin.javajet

@@ -160,6 +160,7 @@
 <%if(rejectConnName.equals(firstConnName)) {%> <%=firstConnName %> = null; <%}%>					
 					
     } catch (java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
         whetherReject_<%=cid%> = true;
         <%
         if (dieOnError) {

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseBulkExec/tAzureSynapseBulkExec_begin.javajet

@@ -523,6 +523,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                             isExistStmt_<%=cid%>.execute("SELECT TOP 1 1 FROM [" +  tableName_<%=cid%> + "]" );
                             whetherExist_<%=cid%> = true;
                         } catch (java.lang.Exception e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                             whetherExist_<%=cid%> = false;
                         }
                         isExistStmt_<%=cid%>.close();
@@ -679,6 +680,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                 }
 %>
             } catch (java.sql.SQLException e_<%=cid%>) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e_<%=cid%>.getMessage());
 <%
                 if(isLog4jEnabled) {
 %>
@@ -935,6 +937,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                                         isExistStmt_<%=cid%>.execute("SELECT TOP 1 1 FROM [" +  tableName_<%=cid%> + "]" );
                                         whetherExist_<%=cid%> = true;
                                     } catch (java.lang.Exception e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                                         whetherExist_<%=cid%> = false;
                                     }
                                     isExistStmt_<%=cid%>.close();
@@ -1165,6 +1168,7 @@ skeleton="../templates/db_output_bulk.skeleton"
                             }
             %>
                         } catch (java.sql.SQLException e_<%=cid%>) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e_<%=cid%>.getMessage());
             <%
                             if(isLog4jEnabled) {
             %>

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseBulkExec/tAzureSynapseBulkExec_java.xml

@@ -386,7 +386,7 @@
             <IMPORT NAME="commons-codec" MODULE="commons-codec-1.14.jar" MVN="mvn:commons-codec/commons-codec/1.14"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="gson" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="oauth2-oidc-sdk" MODULE="oauth2-oidc-sdk-6.5.jar" MVN="mvn:com.nimbusds/oauth2-oidc-sdk/6.5"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
-            <IMPORT NAME="json-smart" MODULE="json-smart-2.3.jar" MVN="mvn:net.minidev/json-smart/2.3"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
+            <IMPORT NAME="json-smart" MODULE="json-smart-2.4.7.jar" MVN="mvn:net.minidev/json-smart/2.4.7"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="nimbus-jose-jwt" MODULE="nimbus-jose-jwt-8.11.jar" MVN="mvn:com.nimbusds/nimbus-jose-jwt/8.11"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="javax.mail" MODULE="javax.mail-1.6.2.jar" MVN="mvn:com.sun.mail/javax.mail/1.6.2"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
 		</IMPORTS>

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseConnection/tAzureSynapseConnection_java.xml

@@ -209,7 +209,7 @@
         <IMPORT NAME="commons-codec" MODULE="commons-codec-1.14.jar" MVN="mvn:commons-codec/commons-codec/1.14"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
         <IMPORT NAME="gson" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
         <IMPORT NAME="oauth2-oidc-sdk" MODULE="oauth2-oidc-sdk-6.5.jar" MVN="mvn:com.nimbusds/oauth2-oidc-sdk/6.5"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
-        <IMPORT NAME="json-smart" MODULE="json-smart-2.3.jar" MVN="mvn:net.minidev/json-smart/2.3"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
+        <IMPORT NAME="json-smart" MODULE="json-smart-2.4.7.jar" MVN="mvn:net.minidev/json-smart/2.4.7"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
         <IMPORT NAME="nimbus-jose-jwt" MODULE="nimbus-jose-jwt-8.11.jar" MVN="mvn:com.nimbusds/nimbus-jose-jwt/8.11"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
         <IMPORT NAME="javax.mail" MODULE="javax.mail-1.6.2.jar" MVN="mvn:com.sun.mail/javax.mail/1.6.2"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true')" />
      </IMPORTS>

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseInput/tAzureSynapseInput_java.xml

@@ -269,7 +269,7 @@
         <IMPORT NAME="commons-codec" MODULE="commons-codec-1.14.jar" MVN="mvn:commons-codec/commons-codec/1.14"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="gson" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="oauth2-oidc-sdk" MODULE="oauth2-oidc-sdk-6.5.jar" MVN="mvn:com.nimbusds/oauth2-oidc-sdk/6.5"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
-        <IMPORT NAME="json-smart" MODULE="json-smart-2.3.jar" MVN="mvn:net.minidev/json-smart/2.3"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
+        <IMPORT NAME="json-smart" MODULE="json-smart-2.4.7.jar" MVN="mvn:net.minidev/json-smart/2.4.7"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="nimbus-jose-jwt" MODULE="nimbus-jose-jwt-8.11.jar" MVN="mvn:com.nimbusds/nimbus-jose-jwt/8.11"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="javax.mail" MODULE="javax.mail-1.6.2.jar" MVN="mvn:com.sun.mail/javax.mail/1.6.2"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
      </IMPORTS>

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseOutput/tAzureSynapseOutput_begin.javajet

@@ -94,6 +94,7 @@ int nb_line_rejected_<%=cid%> = 0;
 int deletedCount_<%=cid%>=0;
 int updatedCount_<%=cid%>=0;
 int insertedCount_<%=cid%>=0;
+int rowsToCommitCount_<%=cid%>=0;
 int rejectedCount_<%=cid%>=0;
 String dbschema_<%=cid%> = null;
 String tableName_<%=cid%> = null;
@@ -137,38 +138,10 @@ String identityKey = ElementParameterParser.getValue(node, "__IDENTITY_FIELD__")
 getManager(dbmsId, cid, node).setIsSpecifyIdentityKey(isSpecifyIdentityKey);
 %>
 
+<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/DB/Output/CheckKeysForUpdateAndDelete.javajet"%>
+
 <%
-if(("UPDATE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction) || ("UPDATE_OR_INSERT").equals(dataAction)) {
-    int updateKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isUpdateKey() && !column.isDynamic()) {
-                updateKeyCount++;
-            }
-        }
-        %>
-        int updateKeyCount_<%=cid%> = <%=updateKeyCount%>;
-        if(updateKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For update, Schema must have a key");
-        }
-        <%
-    }
-} else if(("DELETE").equals(dataAction)) {
-    int deleteKeyCount = 0;
-    if(stmtStructure != null) {
-        for(Column column : stmtStructure) {
-            if(column.isDeleteKey() && !column.isDynamic()) {
-                deleteKeyCount++;
-            }
-        }
-        %>
-        int deleteKeyCount_<%=cid%> = <%=deleteKeyCount%>;
-        if(deleteKeyCount_<%=cid%> < 1) {
-            throw new RuntimeException("For delete, Schema must have a key");
-        }
-        <%
-    }
-} else if (("INSERT_IF_NOT_EXIST").equals(dataAction)){
+ if (("INSERT_IF_NOT_EXIST").equals(dataAction)){
 	int insertKeyCount = 0;
     if(stmtStructure != null) {
         for(Column column : stmtStructure) {

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseOutput/tAzureSynapseOutput_end.javajet

@@ -409,6 +409,7 @@ skeleton="../templates/db_output_bulk.skeleton"
             	    	<%
             	    	}%>
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseOutput/tAzureSynapseOutput_java.xml

@@ -137,7 +137,7 @@
 		<PARAMETER NAME="DATA_ACTION" FIELD="CLOSED_LIST" NUM_ROW="70">
 			<ITEMS DEFAULT="INSERT">
 				<ITEM NAME="INSERT" VALUE="INSERT" />
-				<ITEM NAME="SINGLE_INSERT" VALUE="SINGLE_INSERT" />
+				<ITEM NAME="SINGLE_INSERT" VALUE="SINGLE_INSERT" SHOW_IF="false" />
 				<ITEM NAME="UPDATE" VALUE="UPDATE" />
 				<ITEM NAME="INSERT_OR_UPDATE" VALUE="INSERT_OR_UPDATE" />
 				<ITEM NAME="UPDATE_OR_INSERT" VALUE="UPDATE_OR_INSERT" />
@@ -340,7 +340,7 @@
             <IMPORT NAME="commons-codec" MODULE="commons-codec-1.14.jar" MVN="mvn:commons-codec/commons-codec/1.14"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="gson" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="oauth2-oidc-sdk" MODULE="oauth2-oidc-sdk-6.5.jar" MVN="mvn:com.nimbusds/oauth2-oidc-sdk/6.5"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
-            <IMPORT NAME="json-smart" MODULE="json-smart-2.3.jar" MVN="mvn:net.minidev/json-smart/2.3"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
+            <IMPORT NAME="json-smart" MODULE="json-smart-2.4.7.jar" MVN="mvn:net.minidev/json-smart/2.4.7"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="nimbus-jose-jwt" MODULE="nimbus-jose-jwt-8.11.jar" MVN="mvn:com.nimbusds/nimbus-jose-jwt/8.11"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
             <IMPORT NAME="javax.mail" MODULE="javax.mail-1.6.2.jar" MVN="mvn:com.sun.mail/javax.mail/1.6.2"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
 		</IMPORTS>

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseOutput/tAzureSynapseOutput_main.javajet

@@ -584,6 +584,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
                 insertedCount_<%=cid%> = insertedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
                 <%dbLog.data().inserting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -831,6 +832,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
                 updatedCount_<%=cid%> = updatedCount_<%=cid%> + pstmt_<%=cid %>.executeUpdate();
         	    <%dbLog.data().updating(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1002,6 +1004,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
                     nb_line_<%=cid%>++;
         			<%dbLog.data().updating(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1094,6 +1097,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
                     nb_line_<%=cid%>++;
                     <%dbLog.data().inserting(dbLog.var("nb_line"));%>
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1292,6 +1296,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
     				<%dbLog.data().updating(dbLog.var("nb_line"));%>
     			}
                 } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                     whetherReject_<%=cid%> = true;
                     <%
                     if (("true").equals(dieOnError)) {
@@ -1397,6 +1402,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
                 nb_line_<%=cid%>++;
                 <%dbLog.data().deleting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 whetherReject_<%=cid%> = true;
                 <%
                 if (("true").equals(dieOnError)) {
@@ -1516,6 +1522,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
                 nb_line_<%=cid%>++;
                 <%dbLog.data().inserting(dbLog.var("nb_line"));%>
             } catch(java.lang.Exception e) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
 				whetherReject_<%=cid%> = true;
                 <%
             if (("true").equals(dieOnError)) {
@@ -1590,6 +1597,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
 						}
 						<%dbLog.batch().executeDone(dbLog.str(dataAction));%>
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
                 		throw(e);
@@ -1624,6 +1632,7 @@ if((cid.equals("talendLogs_DB") || cid.equals("talendStats_DB") || cid.equals("t
 						}
 						<%dbLog.batch().executeDone(dbLog.str(dataAction));%>
                 }catch (java.sql.BatchUpdateException e){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e.getMessage());
                 	<%if(("true").equals(dieOnError)) {
                 	%>
 						throw(e);

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseOutput/tAzureSynapseOutput_messages.properties

@@ -22,7 +22,7 @@ CLEAR_TABLE.NAME=Clear data in table
 COMMIT_EVERY.NAME=Commit every
 DATA_ACTION.ITEM.DELETE=Delete
 DATA_ACTION.ITEM.INSERT=Insert
-DATA_ACTION.ITEM.SINGLE_INSERT=Single Insert Query
+DATA_ACTION.ITEM.SINGLE_INSERT=Single Insert Query (deprecated)
 DATA_ACTION.ITEM.INSERT_OR_UPDATE=Insert or update
 DATA_ACTION.ITEM.UPDATE=Update
 DATA_ACTION.ITEM.UPDATE_OR_INSERT=Update or insert

main/plugins/org.talend.designer.components.localprovider/components/tAzureSynapseRow/tAzureSynapseRow_java.xml

@@ -325,7 +325,7 @@
         <IMPORT NAME="commons-codec" MODULE="commons-codec-1.14.jar" MVN="mvn:commons-codec/commons-codec/1.14"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="gson" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="oauth2-oidc-sdk" MODULE="oauth2-oidc-sdk-6.5.jar" MVN="mvn:com.nimbusds/oauth2-oidc-sdk/6.5"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
-        <IMPORT NAME="json-smart" MODULE="json-smart-2.3.jar" MVN="mvn:net.minidev/json-smart/2.3"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
+        <IMPORT NAME="json-smart" MODULE="json-smart-2.4.7.jar" MVN="mvn:net.minidev/json-smart/2.4.7"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="nimbus-jose-jwt" MODULE="nimbus-jose-jwt-8.11.jar" MVN="mvn:com.nimbusds/nimbus-jose-jwt/8.11"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
         <IMPORT NAME="javax.mail" MODULE="javax.mail-1.6.2.jar" MVN="mvn:com.sun.mail/javax.mail/1.6.2"  REQUIRED_IF="(ACTIVE_DIR_AUTH == 'true') AND (USE_EXISTING_CONNECTION == 'false')" />
     </IMPORTS>

main/plugins/org.talend.designer.components.localprovider/components/tBigQueryBulkExec/tBigQueryBulkExec_begin.javajet

@@ -1,10 +1,10 @@
 <%@ jet
 	    imports="
-		org.talend.core.model.process.INode
-		org.talend.core.model.process.ElementParameterParser
+		org.talend.core.model.process.INode
+		org.talend.core.model.process.ElementParameterParser
 		org.talend.designer.codegen.config.CodeGeneratorArgument
-		org.talend.core.model.metadata.IMetadataTable
-		org.talend.core.model.metadata.IMetadataColumn
+		org.talend.core.model.metadata.IMetadataTable
+		org.talend.core.model.metadata.IMetadataColumn
 		org.talend.core.model.process.IConnection
 		java.util.List
 	    "
@@ -37,9 +37,10 @@
 	boolean useCustomNullMarker = ElementParameterParser.getBooleanValue(node, "__USE_CUSTOM_NULL_MARKER__");
 	String nullMarker = useCustomNullMarker ? ElementParameterParser.getValue(node, "__CUSTOM_NULL_MARKER__") : "\"\\\\N\"";
 	String passwordFieldName = "";
-
+
 	IConnection incomingConnection = null;
 	if(node.getUniqueName().startsWith("tBigQueryOutput_")) {
+        cid = cid.replace("_tBQBE", "");
 		List< ? extends IConnection> conns = node.getIncomingConnections();
 		IConnection virtConnection = null;
 		if(conns!=null && conns.size()>0){
@@ -49,7 +50,7 @@
 			incomingConnection = virtConnection.getSource().getIncomingConnections().get(0);
 		}
 	}
-
+
     if (authMode.equals("OAUTH")) {
 		%>
 		/* ----START-CREATING-CLIENT (OAuth 2.0)---- */
@@ -58,22 +59,22 @@
 	    <%
 	    passwordFieldName = "__CLIENT_SECRET__";
 	    %>
-
-	    <%@ include file="@{org.talend.designer.components.localprovider}/components/templates/password.javajet"%>
-
+
+	    <%@ include file="@{org.talend.designer.components.localprovider}/components/templates/password.javajet"%>
+
 		final String CLIENT_SECRET_<%=cid%> = "{\"web\": {\"client_id\": \""+<%=clientId%>+"\",\"client_secret\": \"" +decryptedPassword_<%=cid%>+ "\",\"auth_uri\": \"https://accounts.google.com/o/oauth2/auth\",\"token_uri\": \"https://accounts.google.com/o/oauth2/token\"}}";
 		final String PROJECT_ID_<%=cid %> = <%=projectId %>;
-
+
 		// Static variables for API scope, callback URI, and HTTP/JSON functions
 		final List<String> SCOPES_<%=cid%> = java.util.Arrays.asList("https://www.googleapis.com/auth/bigquery");
 		final String REDIRECT_URI_<%=cid%> = "urn:ietf:wg:oauth:2.0:oob";
 		final com.google.api.client.http.HttpTransport TRANSPORT_<%=cid %> = new com.google.api.client.http.javanet.NetHttpTransport();
 		final com.google.api.client.json.JsonFactory JSON_FACTORY_<%=cid %> = new com.google.api.client.json.jackson2.JacksonFactory();
-
+
 		com.google.api.client.googleapis.auth.oauth2.GoogleClientSecrets clientSecrets_<%=cid%> = com.google.api.client.googleapis.auth.oauth2.GoogleClientSecrets.load(
 						new com.google.api.client.json.jackson2.JacksonFactory(), new java.io.InputStreamReader(new java.io.ByteArrayInputStream(
 								CLIENT_SECRET_<%=cid%>.getBytes())));
-
+
 		com.google.api.client.googleapis.auth.oauth2.GoogleAuthorizationCodeFlow flow_<%=cid%> = null;
 		com.google.api.services.bigquery.Bigquery bigqueryclient_<%=cid%> = null;
 		long nb_line_<%=cid%> = 0;
@@ -94,6 +95,7 @@
 			properties_<%=cid%>.load(inputStream_<%=cid%>);
 			inputStream_<%=cid%>.close();
 		} catch (java.io.FileNotFoundException e_<%=cid%>) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e_<%=cid%>.getMessage());
 			<%
 			if(isLog4jEnabled){
 			%>
@@ -102,6 +104,7 @@
 			}
 			%>
 		} catch (java.io.IOException ee_<%=cid%>) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",ee_<%=cid%>.getMessage());
 			<%
 			if(isLog4jEnabled){
 			%>
@@ -111,7 +114,7 @@
 			%>
 		}
 		String storedRefreshToken_<%=cid%> = (String) properties_<%=cid%>.get("refreshtoken");
-
+
 		// Check to see if the an existing refresh token was loaded.
 		// If so, create a credential and call refreshToken() to get a new
 		// access token.
@@ -120,7 +123,7 @@
 			com.google.api.client.googleapis.auth.oauth2.GoogleCredential credential_<%=cid%> = new com.google.api.client.googleapis.auth.oauth2. GoogleCredential.Builder().setTransport(TRANSPORT_<%=cid%>)
 					.setJsonFactory(JSON_FACTORY_<%=cid%>).setClientSecrets(clientSecrets_<%=cid%>)
 					.build().setFromTokenResponse(new com.google.api.client.auth.oauth2.TokenResponse().setRefreshToken(storedRefreshToken_<%=cid%>));
-
+
 			credential_<%=cid%>.refreshToken();
 			<%
 			if(isLog4jEnabled){
@@ -213,12 +216,12 @@
 		/* ----END-CREATING-CLIENT (Cloud API)---- */
 
 		long nb_line_<%=cid%> = 0;
-  <%
+	<%
   } else if (authMode.equals("TOKEN")) {
-    if (ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) {%>
-       final String decryptedAccessToken_<%=cid%> = routines.system.PasswordEncryptUtil.decryptPassword(<%=ElementParameterParser.getEncryptedValue(node, "__ACCESS_TOKEN__")%>);
-  <%} else {%>
-       final String decryptedAccessToken_<%=cid%> = <%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>;
+    if (ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) {%>
+       final String decryptedAccessToken_<%=cid%> = routines.system.PasswordEncryptUtil.decryptPassword(<%=ElementParameterParser.getEncryptedValue(node, "__ACCESS_TOKEN__")%>);
+  <%} else {%>
+       final String decryptedAccessToken_<%=cid%> = <%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>;
   <%}%>
 	/* ----START-CREATING-CLIENT (OAuth based Token)---- */
 		final String PROJECT_ID_<%=cid %> = <%=projectId %>;
@@ -288,12 +291,12 @@
 				log.info("<%=cid%> - Upload Done.");
 			<%
 			}
-		} else {
-      boolean useExistingConn = false;
-      String gsProjectID = projectId;
-      String connection = cid;
-      %>
-      <%@ include file="@{org.talend.designer.components.localprovider}/components/templates/googleStorageConnection.javajet"%>
+		} else {
+      boolean useExistingConn = false;
+      String gsProjectID = projectId;
+      String connection = cid;
+			%>
+      <%@ include file="@{org.talend.designer.components.localprovider}/components/templates/googleStorageConnection.javajet"%>
 
 			java.io.File file_<%=cid%> = new java.io.File(<%=localFilename%>);
 
@@ -340,6 +343,7 @@
 				<% } %>
 				bigqueryclient_<%=cid%>.tables().delete(PROJECT_ID_<%=cid%>, <%=dataset%>, <%=table%>).execute();
 			} catch (com.google.api.client.googleapis.json.GoogleJsonResponseException e_<%=cid%>) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e_<%=cid%>.getMessage());
 				if (e_<%=cid%>.getDetails().getCode() != 404) {
 					<%
 					if(dieOnError) {
@@ -370,6 +374,7 @@
 				com.google.api.services.bigquery.model.Table getTable = bigqueryclient_<%=cid%>.tables().get(PROJECT_ID_<%=cid%>, <%=dataset%>, <%=table%>).execute();
 				queryLoad_<%=cid%>.setCreateDisposition("CREATE_NEVER");
 			} catch (com.google.api.client.googleapis.json.GoogleJsonResponseException e_<%=cid%>){
+globalMap.put("<%=cid%>_ERROR_MESSAGE",e_<%=cid%>.getMessage());
 				if (e_<%=cid%>.getDetails().getCode() == 404) {
 					com.google.api.services.bigquery.model.TableSchema schema_<%=cid%> = new com.google.api.services.bigquery.model.TableSchema();
 					<%
@@ -536,11 +541,12 @@
 		%>
 		com.google.api.services.bigquery.model.Job jobExec_<%=cid%> = null;
 		try {
-			<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
-      cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
+			<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
+      cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
       <% } %>
 			jobExec_<%=cid%> = insertReq_<%=cid%>.execute();
 		} catch (Exception ee_<%=cid%>) {
+globalMap.put("<%=cid%>_ERROR_MESSAGE",ee_<%=cid%>.getMessage());
 			<%
 			if(dieOnError) {
 			%>
@@ -560,14 +566,14 @@
 		}
 		if (jobExec_<%=cid%>.getStatus().getState().equals("RUNNING")
 				|| jobExec_<%=cid%>.getStatus().getState().equals("PENDING")) {
-					<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
-			cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
+					<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
+			cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
 					<% } %>
-			com.google.api.services.bigquery.model.Job pollJob_<%=cid%> = bigqueryclient_<%=cid%>.jobs().get(PROJECT_ID_<%=cid%>,jobExec_<%=cid%>.getJobReference().getJobId()).setLocation(jobExec_<%=cid%>.getJobReference().getLocation()).execute();
+            		com.google.api.services.bigquery.model.Job pollJob_<%=cid%> = bigqueryclient_<%=cid%>.jobs().get(PROJECT_ID_<%=cid%>,jobExec_<%=cid%>.getJobReference().getJobId()).setLocation(jobExec_<%=cid%>.getJobReference().getLocation()).execute();
 			while (pollJob_<%=cid%>.getStatus().getState().equals("RUNNING") || pollJob_<%=cid%>.getStatus().getState().equals("PENDING")) {
 				Thread.sleep(1000);
-				<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
-				cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
+				<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
+				cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
 				<% } %>
 				pollJob_<%=cid%> = bigqueryclient_<%=cid%>.jobs().get(PROJECT_ID_<%=cid%>,jobExec_<%=cid%>.getJobReference().getJobId()).setLocation(jobExec_<%=cid%>.getJobReference().getLocation()).execute();
 				System.out.println(String.format(
@@ -583,10 +589,12 @@
 				%>
 			}
 
-			<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
-      cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
+			<% if(authMode.equals("TOKEN") && !ElementParameterParser.canEncrypt(node, "__ACCESS_TOKEN__")) { %>
+      cred_<%=cid%>.setAccessToken(<%= ElementParameterParser.getValue(node, "__ACCESS_TOKEN__")%>);
       <% } %>
 			com.google.api.services.bigquery.model.Job doneJob_<%=cid%> = pollJob_<%=cid%>;
+            globalMap.put("<%=cid%>_JOBID",doneJob_<%=cid%>.getJobReference().getJobId());
+            globalMap.put("<%=cid%>_STATISTICS", doneJob_<%=cid%>.getStatistics().toString());
 
 			if ((doneJob_<%=cid%>.getStatus() != null) && (doneJob_<%=cid%>.getStatus().getErrors() != null)) {
 				status = "failure";
@@ -812,7 +820,10 @@
 			loadJobBuilder_<%=cid%>.setNullMarker(<%= nullMarker %>);
 			com.google.cloud.bigquery.Job job_<%=cid%> = bigquery_<%=cid%>.create(com.google.cloud.bigquery.JobInfo.of(loadJobBuilder_<%=cid%>.build()));
 			job_<%=cid%> = job_<%=cid%>.waitFor(com.google.cloud.RetryOption.initialRetryDelay(org.threeten.bp.Duration.ofSeconds(1)));
+                        com.google.gson.Gson gsonObject_<%=cid%> = new com.google.gson.Gson();
 			if (job_<%=cid%> != null && job_<%=cid%>.getStatus().getError() == null) {
+                globalMap.put("<%=cid%>_JOBID", job_<%=cid%>.getJobId().getJob());
+                globalMap.put("<%=cid%>_STATISTICS", gsonObject_<%=cid%>.toJson(job_<%=cid%>.getStatistics()));
 				<%
 				if(isLog4jEnabled){
 				%>
@@ -821,6 +832,11 @@
 				}
 				%>
 			} else {
+                if (job_<%=cid%> == null) {
+                    throw new RuntimeException("Job no longer exists");
+                }
+                globalMap.put("<%=cid%>_JOBID", job_<%=cid%>.getJobId().getJob());
+                globalMap.put("<%=cid%>_STATISTICS", gsonObject_<%=cid%>.toJson(job_<%=cid%>.getStatistics()));
 				List<com.google.cloud.bigquery.BigQueryError> errorList = job_<%=cid%>.getStatus().getExecutionErrors();
 				throw new RuntimeException("Job failed: " + errorList.get(errorList.size() - 1));
 			}

main/plugins/org.talend.designer.components.localprovider/components/tBigQueryBulkExec/tBigQueryBulkExec_java.xml

@@ -286,11 +286,11 @@
          <IMPORT NAME="google-oauth-client-1.31.0.jar" MODULE="google-oauth-client-1.31.0.jar" MVN="mvn:com.google.oauth-client/google-oauth-client/1.31.0" REQUIRED="true" />
          <IMPORT NAME="google-http-client-jackson2-1.38.0.jar" MODULE="google-http-client-jackson2-1.38.0.jar" MVN="mvn:com.google.http-client/google-http-client-jackson2/1.38.0"  REQUIRED="true" />
          <IMPORT NAME="guava-20.0.jar" MODULE="guava-20.0.jar" MVN="mvn:com.google.guava/guava/20.0" REQUIRED="true" />
-		<IMPORT NAME="jackson-core-2.10.1.jar" MODULE="jackson-core-2.10.1.jar" MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.10.1" REQUIRED="true" />
+		<IMPORT NAME="jackson-core-2.11.4.jar" MODULE="jackson-core-2.11.4.jar" MVN="mvn:com.fasterxml.jackson.core/jackson-core/2.11.4" REQUIRED="true" />
 		<!-- REQUIRED FOR GOOGLE STORAGE -->
 		<IMPORT NAME="jets3t-0.9.1" MODULE="jets3t-0.9.1.jar" MVN="mvn:org.talend.libraries/jets3t-0.9.1/6.0.0"  REQUIRED="true" />
 		<IMPORT NAME="commons-logging-1.2" MODULE="commons-logging-1.2.jar" MVN="mvn:commons-logging/commons-logging/1.2" REQUIRED="true" />
-		<IMPORT NAME="httpclient-4.5.12" MODULE="httpclient-4.5.12.jar" MVN="mvn:org.apache.httpcomponents/httpclient/4.5.12" REQUIRED="true" />
+		<IMPORT NAME="httpclient-4.5.13" MODULE="httpclient-4.5.13.jar" MVN="mvn:org.apache.httpcomponents/httpclient/4.5.13" REQUIRED="true" />
 		<IMPORT NAME="httpcore-4.4.13" MODULE="httpcore-4.4.13.jar" MVN="mvn:org.apache.httpcomponents/httpcore/4.4.13" REQUIRED="true" />
 		<IMPORT NAME="commons-codec-1.14" MODULE="commons-codec-1.14.jar" MVN="mvn:commons-codec/commons-codec/1.14" REQUIRED="true" />
 		<IMPORT NAME="google-cloud-bigquery-1.126.3.jar" MODULE="google-cloud-bigquery-1.126.3.jar" MVN="mvn:com.google.cloud/google-cloud-bigquery/1.126.3" REQUIRED_IF="(AUTH_MODE == 'SERVICEACCOUNT') OR (AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
@@ -312,7 +312,7 @@
     <IMPORT NAME="google-cloud-storage-1.104.0" MODULE="google-cloud-storage-1.104.0.jar" MVN="mvn:com.google.cloud/google-cloud-storage/1.104.0" REQUIRED_IF="(AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
       <!-- Transitive dependencies of google-cloud-storage -->
       <IMPORT NAME="google-api-services-storage-v1-rev20201112-1.31.0" MODULE="google-api-services-storage-v1-rev20201112-1.31.0.jar" MVN="mvn:com.google.apis/google-api-services-storage/v1-rev20201112-1.31.0" REQUIRED_IF="(AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
-      <IMPORT NAME="gson-2.8.6" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6" REQUIRED_IF="(AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
+      <IMPORT NAME="gson-2.8.6" MODULE="gson-2.8.6.jar" MVN="mvn:com.google.code.gson/gson/2.8.6" REQUIRED="true" />
       <IMPORT NAME="javax.annotation-api" MODULE="javax.annotation-api-1.3.jar" MVN="mvn:javax.annotation/javax.annotation-api/1.3" REQUIRED_IF="(AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
       <IMPORT NAME="protobuf-java-2.5.0" MODULE="protobuf-java-2.5.0.jar" MVN="mvn:com.google.protobuf/protobuf-java/2.5.0" REQUIRED_IF="(AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
       <IMPORT NAME="error_prone_annotation-2.1.3" MODULE="error_prone_annotation-2.1.3.jar" MVN="mvn:com.google.errorprone/error_prone_annotations/2.1.3" REQUIRED_IF="(AUTH_TYPE == 'GS_SERVICE_ACCOUNT') OR (AUTH_TYPE == 'TOKEN')" />
@@ -326,5 +326,7 @@
      </IMPORTS>
    </CODEGENERATION>
    <RETURNS>
+     <RETURN NAME="JOBID" TYPE="id_String" AVAILABILITY="AFTER"/>
+     <RETURN NAME="STATISTICS" TYPE="id_String" AVAILABILITY="AFTER"/>
    </RETURNS>
 </COMPONENT>