TUP-34282 CVE: log4j-core(1.2-api)-[2-2.16.0) (Fix junit) (#7096)

https://jira.talendforge.org/browse/TUP-34282

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

@@ -50,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.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/checkArchive/pom.xml

@@ -1,8 +1,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.libraries</groupId>
-  <artifactId>checkArchive-1.1-20190917</artifactId>
-  <version>6.0.0</version>
+  <groupId>org.talend.components</groupId>
+  <artifactId>checkArchive</artifactId>
+  <version>1.2-20210901</version>
   <name>checkArchive</name>
   <description>Dependence for tFileArchive and tFileUnAchive</description>
 <url>http://maven.apache.org</url>
@@ -10,7 +10,7 @@
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<talend.nexus.url>https://artifacts-oss.talend.com</talend.nexus.url>
-		<java.source.version>1.7</java.source.version>
+		<java.source.version>1.8</java.source.version>
 	</properties>
 
 	<distributionManagement>
@@ -41,7 +41,7 @@
 <dependency>
     <groupId>org.apache.commons</groupId>
     <artifactId>commons-compress</artifactId>
-    <version>1.19</version>
+    <version>1.21</version>
 </dependency>
 	</dependencies>
 	<build>

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/mdm_services_tomcat/src/main/java/org/talend/mdm/webservice/WSTransformerV2.java

@@ -22,6 +22,7 @@ import javax.xml.bind.annotation.XmlType;
  *         <element name="description" type="{http://www.w3.org/2001/XMLSchema}string" minOccurs="0"/>
  *         <element name="name" type="{http://www.w3.org/2001/XMLSchema}string" minOccurs="0"/>
  *         <element name="processSteps" type="{http://www.talend.com/mdm}WSTransformerProcessStep" maxOccurs="unbounded" minOccurs="0"/>
+ *         <element name="withAdminPermissions" type="{http://www.w3.org/2001/XMLSchema}boolean" minOccurs="0"/>
  *       </sequence>
  *     </restriction>
  *   </complexContent>
@@ -34,7 +35,8 @@ import javax.xml.bind.annotation.XmlType;
 @XmlType(name = "WSTransformerV2", propOrder = {
     "description",
     "name",
-    "processSteps"
+    "processSteps",
+    "withAdminPermissions"
 })
 public class WSTransformerV2 {
 
@@ -42,6 +44,7 @@ public class WSTransformerV2 {
     protected String name;
     @XmlElement(nillable = true)
     protected List<WSTransformerProcessStep> processSteps;
+    protected Boolean withAdminPermissions;
 
     /**
      * Default no-arg constructor
@@ -55,10 +58,11 @@ public class WSTransformerV2 {
      * Fully-initialising value constructor
      * 
      */
-    public WSTransformerV2(final String description, final String name, final List<WSTransformerProcessStep> processSteps) {
+    public WSTransformerV2(final String description, final String name, final List<WSTransformerProcessStep> processSteps, final Boolean withAdminPermissions) {
         this.description = description;
         this.name = name;
         this.processSteps = processSteps;
+        this.withAdminPermissions = withAdminPermissions;
     }
 
     /**
@@ -138,4 +142,28 @@ public class WSTransformerV2 {
         return this.processSteps;
     }
 
+    /**
+     * Gets the value of the withAdminPermissions property.
+     * 
+     * @return
+     *     possible object is
+     *     {@link Boolean }
+     *     
+     */
+    public Boolean isWithAdminPermissions() {
+        return withAdminPermissions;
+    }
+
+    /**
+     * Sets the value of the withAdminPermissions property.
+     * 
+     * @param value
+     *     allowed object is
+     *     {@link Boolean }
+     *     
+     */
+    public void setWithAdminPermissions(Boolean value) {
+        this.withAdminPermissions = value;
+    }
+    
 }

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

@@ -7,7 +7,7 @@
 	<groupId>org.talend.libraries</groupId>
 	<artifactId>talend-codegen-utils</artifactId>
 	<!-- release for revert version of library -->
-	<version>0.29.0</version>
+	<version>0.31.0</version>
 	<packaging>jar</packaging>
 	
 	<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,12 @@ 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
      *
@@ -165,7 +176,25 @@ public class IncomingSchemaEnforcer {
             fieldSchema = SchemaBuilder.nullable().type(fieldSchema);
         }
 
-        Schema.Field field = new Schema.Field(name, fieldSchema, description, (Object) null);
+        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);
@@ -271,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);
         }
@@ -390,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
      *
@@ -397,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.

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

@@ -69,13 +69,13 @@
 		<dependency>
 			<groupId>org.apache.logging.log4j</groupId>
 			<artifactId>log4j-api</artifactId>
-			<version>2.12.1</version>
+			<version>2.17.1</version>
 			<optional>true</optional>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.logging.log4j</groupId>
 			<artifactId>log4j-core</artifactId>
-			<version>2.12.1</version>
+			<version>2.17.1</version>
 			<optional>true</optional>
 		</dependency>
 	</dependencies>
@@ -97,4 +97,4 @@
 		</pluginManagement>
 	</build>
 
-</project>
+</project>

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-orc/pom.xml

@@ -0,0 +1,191 @@
+<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-orc</artifactId>
+	<version>1.0-20211008</version>
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<hadoop.version>3.2.2</hadoop.version>
+		<apache.orc.version>1.7.0</apache.orc.version>
+		<junit.jupiter.version>5.7.2</junit.jupiter.version>
+		<hamcrest.version>1.3</hamcrest.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.orc</groupId>
+			<artifactId>orc-core</artifactId>
+			<version>${apache.orc.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.hadoop</groupId>
+			<artifactId>hadoop-common</artifactId>
+			<version>${hadoop.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.eclipse.jetty</groupId>
+					<artifactId>jetty-server</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>org.eclipse.jetty</groupId>
+					<artifactId>jetty-servlet</artifactId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jersey-server</artifactId>
+					<groupId>com.sun.jersey</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jersey-servlet</artifactId>
+					<groupId>com.sun.jersey</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>slf4j-log4j12</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>log4j</artifactId>
+					<groupId>log4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>zookeeper</artifactId>
+					<groupId>org.apache.zookeeper</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jetty-webapp</artifactId>
+					<groupId>org.eclipse.jetty</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>javax.servlet-api</artifactId>
+					<groupId>javax.servlet</groupId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.hadoop</groupId>
+			<artifactId>hadoop-hdfs-client</artifactId>
+			<version>${hadoop.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.eclipse.jetty</groupId>
+					<artifactId>jetty-server</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>org.eclipse.jetty</groupId>
+					<artifactId>jetty-servlet</artifactId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jersey-server</artifactId>
+					<groupId>com.sun.jersey</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jersey-servlet</artifactId>
+					<groupId>com.sun.jersey</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>slf4j-log4j12</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>log4j</artifactId>
+					<groupId>log4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>zookeeper</artifactId>
+					<groupId>org.apache.zookeeper</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jetty-webapp</artifactId>
+					<groupId>org.eclipse.jetty</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>javax.servlet-api</artifactId>
+					<groupId>javax.servlet</groupId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<version>${junit.jupiter.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<version>${junit.jupiter.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-engine</artifactId>
+			<version>${junit.jupiter.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-params</artifactId>
+			<version>${junit.jupiter.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.platform</groupId>
+			<artifactId>junit-platform-launcher</artifactId>
+			<version>1.7.2</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-orc/src/main/java/org/talend/orc/ORCCommonUtils.java

@@ -0,0 +1,24 @@
+package org.talend.orc;
+
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+
+abstract class ORCCommonUtils {
+	/**
+	 * Check that the map type for the key is BYTES, LONG or DOUBLE and that the key
+	 * type is LONG, DOUBLE, BYTES, DECIMAL or TIMESTAMP.
+	 *
+	 * @param mapVector a MapColumnVector
+	 * @return true if the key and value types conform to the limits described
+	 *         above.
+	 */
+	public static boolean checkMapColumnVectorTypes(MapColumnVector mapVector) {
+		ColumnVector.Type keyType = mapVector.keys.type;
+		ColumnVector.Type valueType = mapVector.values.type;
+		return (keyType == ColumnVector.Type.BYTES || keyType == ColumnVector.Type.LONG
+				|| keyType == ColumnVector.Type.DOUBLE)
+				&& (valueType == ColumnVector.Type.LONG || valueType == ColumnVector.Type.DOUBLE
+						|| valueType == ColumnVector.Type.BYTES || valueType == ColumnVector.Type.DECIMAL
+						|| valueType == ColumnVector.Type.TIMESTAMP);
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-orc/src/main/java/org/talend/orc/ORCReadUtils.java

@@ -0,0 +1,442 @@
+package org.talend.orc;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.UnionColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.TypeDescription;
+
+public class ORCReadUtils {
+	public static Object readColumnByName(VectorizedRowBatch batch, String columnName, TypeDescription schema,
+			int rowNum) {
+		List<String> allColumnNames = schema.getFieldNames();
+		int colIndex = allColumnNames.indexOf(columnName);
+		if (colIndex < 0 || colIndex > batch.cols.length - 1) {
+			return null;
+		} else {
+			org.apache.hadoop.hive.ql.exec.vector.ColumnVector colVector = batch.cols[colIndex];
+			TypeDescription fieldType = schema.getChildren().get(colIndex);
+			int colRow = colVector.isRepeating ? 0 : rowNum;
+			Object value = readColumn(colVector, fieldType, colRow);
+			return value;
+		}
+
+	}
+
+	public static Object readColumn(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object columnObj = null;
+		if (!colVec.isNull[rowNum]) {
+			switch (colVec.type) {
+			case BYTES:
+				columnObj = readBytesVal(colVec, colType, rowNum);
+				break;
+			case DECIMAL:
+				columnObj = readDecimalVal(colVec, rowNum);
+				break;
+			case DOUBLE:
+				columnObj = readDoubleVal(colVec, colType, rowNum);
+				break;
+			case LIST:
+				columnObj = readListVal(colVec, colType, rowNum);
+				break;
+			case LONG:
+				columnObj = readLongVal(colVec, colType, rowNum);
+				break;
+			case MAP:
+				columnObj = readMapVal(colVec, colType, rowNum);
+				break;
+			case STRUCT:
+				columnObj = readStructVal(colVec, colType, rowNum);
+				break;
+			case TIMESTAMP:
+				columnObj = readTimestampVal(colVec, colType, rowNum);
+				break;
+			case UNION:
+				columnObj = readUnionVal(colVec, colType, rowNum);
+				break;
+			default:
+				throw new RuntimeException("readColumn: unsupported ORC file column type: " + colVec.type.name());
+			}
+		}
+		return columnObj;
+	}
+
+	private static Object readListVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object listValues = null;
+		if (!colVec.isNull[rowNum]) {
+			ListColumnVector listVector = (ListColumnVector) colVec;
+			ColumnVector listChildVector = listVector.child;
+			TypeDescription childType = colType.getChildren().get(0);
+			switch (listChildVector.type) {
+			case BYTES:
+				listValues = readBytesListValues(listVector, childType, rowNum);
+				break;
+			case DECIMAL:
+				listValues = readDecimalListValues(listVector, rowNum);
+				break;
+			case DOUBLE:
+				listValues = readDoubleListValues(listVector, rowNum);
+				break;
+			case LONG:
+				listValues = readLongListValues(listVector, childType, rowNum);
+				break;
+			case TIMESTAMP:
+				listValues = readTimestampListValues(listVector, childType, rowNum);
+				break;
+			default:
+				throw new RuntimeException(listVector.type.name() + " is not supported for ListColumnVectors");
+			}
+		}
+		return listValues;
+	}
+
+	private static List<Object> readLongListVector(LongColumnVector longVector, TypeDescription childType, int offset,
+			int numValues) {
+		List<Object> longList = new ArrayList<>();
+		for (int i = 0; i < numValues; i++) {
+			if (!longVector.isNull[offset + i]) {
+				long longVal = longVector.vector[offset + i];
+				if (childType.getCategory() == TypeDescription.Category.BOOLEAN) {
+					Boolean boolVal = longVal == 0 ? Boolean.valueOf(false) : Boolean.valueOf(true);
+					longList.add(boolVal);
+				} else if (childType.getCategory() == TypeDescription.Category.INT) {
+					Integer intObj = (int) longVal;
+					longList.add(intObj);
+				} else {
+					longList.add(longVal);
+				}
+			} else {
+				longList.add(null);
+			}
+		}
+		return longList;
+	}
+
+	private static Object readLongListValues(ListColumnVector listVector, TypeDescription childType, int rowNum) {
+		int offset = (int) listVector.offsets[rowNum];
+		int numValues = (int) listVector.lengths[rowNum];
+		LongColumnVector longVector = (LongColumnVector) listVector.child;
+		return readLongListVector(longVector, childType, offset, numValues);
+	}
+
+	private static Object readTimestampListVector(TimestampColumnVector timestampVector, TypeDescription childType,
+			int offset, int numValues) {
+		List<Object> timestampList = new ArrayList<>();
+		for (int i = 0; i < numValues; i++) {
+			if (!timestampVector.isNull[offset + i]) {
+				int nanos = timestampVector.nanos[offset + i];
+				long millisec = timestampVector.time[offset + i];
+				Timestamp timestamp = new Timestamp(millisec);
+				timestamp.setNanos(nanos);
+				if (childType.getCategory() == TypeDescription.Category.DATE) {
+					Date date = new Date(timestamp.getTime());
+					timestampList.add(date);
+				} else {
+					timestampList.add(timestamp);
+				}
+			} else {
+				timestampList.add(null);
+			}
+		}
+		return timestampList;
+	}
+
+	/**
+	 * Read either Timestamp or Date values, depending on the definition in the
+	 * schema.
+	 */
+	private static Object readTimestampListValues(ListColumnVector listVector, TypeDescription childType, int rowNum) {
+		int offset = (int) listVector.offsets[rowNum];
+		int numValues = (int) listVector.lengths[rowNum];
+		TimestampColumnVector timestampVec = (TimestampColumnVector) listVector.child;
+		return readTimestampListVector(timestampVec, childType, offset, numValues);
+	}
+
+	private static Object readDecimalListVector(DecimalColumnVector decimalVector, int offset, int numValues,
+			int batchRowNum) {
+		List<Object> decimalList = new ArrayList<>();
+		for (int i = 0; i < numValues; i++) {
+			if (!decimalVector.isNull[offset + i]) {
+				BigDecimal bigDecimal = decimalVector.vector[batchRowNum].getHiveDecimal().bigDecimalValue();
+				decimalList.add(bigDecimal);
+			} else {
+				decimalList.add(null);
+			}
+		}
+		return decimalList;
+	}
+
+	private static Object readDecimalListValues(ListColumnVector listVector, int rowNum) {
+		int offset = (int) listVector.offsets[rowNum];
+		int numValues = (int) listVector.lengths[rowNum];
+		DecimalColumnVector decimalVec = (DecimalColumnVector) listVector.child;
+		return readDecimalListVector(decimalVec, offset, numValues, rowNum);
+	}
+
+	private static Object readBytesListVector(BytesColumnVector bytesVec, TypeDescription childType, int offset,
+			int numValues) {
+		List<Object> bytesValList = new ArrayList<>();
+		for (int i = 0; i < numValues; i++) {
+			if (!bytesVec.isNull[offset + i]) {
+				byte[] byteArray = bytesVec.vector[offset + i];
+				int vecLen = bytesVec.length[offset + i];
+				int vecStart = bytesVec.start[offset + i];
+				byte[] vecCopy = Arrays.copyOfRange(byteArray, vecStart, vecStart + vecLen);
+				if (childType.getCategory() == TypeDescription.Category.STRING) {
+					String str = new String(vecCopy);
+					bytesValList.add(str);
+				} else {
+					bytesValList.add(vecCopy);
+				}
+			} else {
+				bytesValList.add(null);
+			}
+		}
+		return bytesValList;
+	}
+
+	private static Object readBytesListValues(ListColumnVector listVector, TypeDescription childType, int rowNum) {
+		int offset = (int) listVector.offsets[rowNum];
+		int numValues = (int) listVector.lengths[rowNum];
+		BytesColumnVector bytesVec = (BytesColumnVector) listVector.child;
+		return readBytesListVector(bytesVec, childType, offset, numValues);
+	}
+
+	private static Object readDoubleListVector(DoubleColumnVector doubleVec, int offset, int numValues) {
+		List<Object> doubleList = new ArrayList<>();
+		for (int i = 0; i < numValues; i++) {
+			if (!doubleVec.isNull[offset + i]) {
+				Double doubleVal = doubleVec.vector[offset + i];
+				doubleList.add(doubleVal);
+			} else {
+				doubleList.add(null);
+			}
+		}
+		return doubleList;
+	}
+
+	private static Object readDoubleListValues(ListColumnVector listVector, int rowNum) {
+
+		int offset = (int) listVector.offsets[rowNum];
+		int numValues = (int) listVector.lengths[rowNum];
+		DoubleColumnVector doubleVec = (DoubleColumnVector) listVector.child;
+		return readDoubleListVector(doubleVec, offset, numValues);
+	}
+
+	@SuppressWarnings("unchecked")
+	private static List<Object> readMapVector(ColumnVector mapVector, TypeDescription childType, int offset,
+			int numValues, int rowNum) {
+		List<Object> mapList = null;
+		switch (mapVector.type) {
+		case BYTES:
+			mapList = (List<Object>) readBytesListVector((BytesColumnVector) mapVector, childType, offset, numValues);
+			break;
+		case DECIMAL:
+			mapList = (List<Object>) readDecimalListVector((DecimalColumnVector) mapVector, offset, numValues, rowNum);
+			;
+			break;
+		case DOUBLE:
+			mapList = (List<Object>) readDoubleListVector((DoubleColumnVector) mapVector, offset, numValues);
+			break;
+		case LONG:
+			mapList = readLongListVector((LongColumnVector) mapVector, childType, offset, numValues);
+			break;
+		case TIMESTAMP:
+			mapList = (List<Object>) readTimestampListVector((TimestampColumnVector) mapVector, childType, offset,
+					numValues);
+			break;
+		default:
+			throw new RuntimeException(mapVector.type.name() + " is not supported for MapColumnVectors");
+		}
+		return mapList;
+	}
+
+	/**
+	 * <p>
+	 * Read a Map column value (e.g., a set of keys and their associated values).
+	 * </p>
+	 * <p>
+	 * The Map key and value types are the first and second children in the children
+	 * TypeDescription List. From the TypeDescription source:
+	 * </p>
+	 * 
+	 * <pre>
+	 * result.children.add(keyType);
+	 * result.children.add(valueType);
+	 * </pre>
+	 */
+	private static Object readMapVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Map<Object, Object> objMap = new HashMap<>();
+		MapColumnVector mapVector = (MapColumnVector) colVec;
+		if (ORCCommonUtils.checkMapColumnVectorTypes(mapVector)) {
+			int mapSize = (int) mapVector.lengths[rowNum];
+			int offset = (int) mapVector.offsets[rowNum];
+			List<TypeDescription> mapTypes = colType.getChildren();
+			TypeDescription keyType = mapTypes.get(0);
+			TypeDescription valueType = mapTypes.get(1);
+			ColumnVector keyChild = mapVector.keys;
+			ColumnVector valueChild = mapVector.values;
+			List<Object> keyList = readMapVector(keyChild, keyType, offset, mapSize, rowNum);
+			List<Object> valueList = readMapVector(valueChild, valueType, offset, mapSize, rowNum);
+			assert (keyList.size() == valueList.size());
+			for (int i = 0; i < keyList.size(); i++) {
+				objMap.put(keyList.get(i), valueList.get(i));
+			}
+		} else {
+			throw new RuntimeException("readMapVal: unsupported key or value types");
+		}
+		return objMap;
+	}
+
+	private static Object readUnionVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Pair<TypeDescription, Object> columnValuePair;
+		UnionColumnVector unionVector = (UnionColumnVector) colVec;
+		int tagVal = unionVector.tags[rowNum];
+		List<TypeDescription> unionFieldTypes = colType.getChildren();
+		if (tagVal < unionFieldTypes.size()) {
+			TypeDescription fieldType = unionFieldTypes.get(tagVal);
+			if (tagVal < unionVector.fields.length) {
+				ColumnVector fieldVector = unionVector.fields[tagVal];
+				int colRow = fieldVector.isRepeating ? 0 : rowNum;
+				Object unionValue = readColumn(fieldVector, fieldType, colRow);
+				columnValuePair = new ImmutablePair<>(fieldType, unionValue);
+			} else {
+				throw new RuntimeException("readUnionVal: union tag value out of range for union column vectors");
+			}
+		} else {
+			throw new RuntimeException("readUnionVal: union tag value out of range for union types");
+		}
+		return columnValuePair;
+	}
+
+	private static Object readStructVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object structObj = null;
+		if (!colVec.isNull[rowNum]) {
+			List<Object> fieldValList = new ArrayList<>();
+			StructColumnVector structVector = (StructColumnVector) colVec;
+			ColumnVector[] fieldVec = structVector.fields;
+			List<TypeDescription> fieldTypes = colType.getChildren();
+			assert (fieldVec.length == fieldTypes.size());
+			for (int i = 0; i < fieldVec.length; i++) {
+				int colRow = fieldVec[i].isRepeating ? 0 : rowNum;
+				Object fieldObj = readColumn(fieldVec[i], fieldTypes.get(i), colRow);
+				fieldValList.add(fieldObj);
+			}
+			structObj = fieldValList;
+		}
+		return structObj;
+	}
+
+	private static Object readTimestampVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object timestampVal = null;
+		if (!colVec.isNull[rowNum]) {
+			TimestampColumnVector timestampVec = (TimestampColumnVector) colVec;
+			int nanos = timestampVec.nanos[rowNum];
+			long millisec = timestampVec.time[rowNum];
+			Timestamp timestamp = new Timestamp(millisec);
+			timestamp.setNanos(nanos);
+			timestampVal = timestamp;
+			if (colType.getCategory() == TypeDescription.Category.DATE) {
+				timestampVal = new Date(timestamp.getTime());
+			}
+		}
+		return timestampVal;
+	}
+
+	private static Object readDecimalVal(ColumnVector colVec, int rowNum) {
+		Object decimalObj = null;
+		if (!colVec.isNull[rowNum]) {
+			DecimalColumnVector decimalVec = (DecimalColumnVector) colVec;
+			decimalObj = decimalVec.vector[rowNum].getHiveDecimal().bigDecimalValue().setScale(decimalVec.scale);
+		}
+		return decimalObj;
+	}
+
+	/**
+	 * Read a Long or Boolean value
+	 * 
+	 * @param colVec  the column vector
+	 * @param colType the type of the column
+	 * @param rowNum  the ORC file row number.
+	 * @return a Boolean or Long object
+	 */
+	private static Object readLongVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object colObj = null;
+		if (!colVec.isNull[rowNum]) {
+			LongColumnVector longVec = (LongColumnVector) colVec;
+			Long longVal = longVec.vector[rowNum];
+			colObj = longVal;
+			if (colType.getCategory() == TypeDescription.Category.INT) {
+				colObj = longVal.intValue();
+			} else if (colType.getCategory() == TypeDescription.Category.BOOLEAN) {
+				colObj = longVal == 1 ? Boolean.TRUE : Boolean.FALSE;
+			} else if (colType.getCategory() == TypeDescription.Category.DATE) {
+				colObj = new Date(longVal * 86400000);
+			} else if (colType.getCategory() == TypeDescription.Category.BYTE) {
+				colObj = longVal.byteValue();
+			} else if (colType.getCategory() == TypeDescription.Category.SHORT) {
+				colObj = longVal.shortValue();
+			}
+		}
+		return colObj;
+	}
+
+	/**
+	 * Read a Double or Float value
+	 * 
+	 * @param colVec  the column vector
+	 * @param colType the type of the column
+	 * @param rowNum  the ORC file row number.
+	 * @return a Double or Float object
+	 */
+	private static Object readDoubleVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object colObj = null;
+		if (!colVec.isNull[rowNum]) {
+			DoubleColumnVector longVec = (DoubleColumnVector) colVec;
+			Double doubleVal = longVec.vector[rowNum];
+			colObj = doubleVal;
+			if (colType.getCategory() == TypeDescription.Category.FLOAT) {
+				colObj = doubleVal.floatValue();
+			}
+		}
+		return colObj;
+	}
+
+	private static Object readBytesVal(ColumnVector colVec, TypeDescription colType, int rowNum) {
+		Object bytesObj = null;
+		if (!colVec.isNull[rowNum]) {
+			BytesColumnVector bytesVector = (BytesColumnVector) colVec;
+			byte[] columnBytes = bytesVector.vector[rowNum];
+			int vecLen = bytesVector.length[rowNum];
+			int vecStart = bytesVector.start[rowNum];
+			byte[] vecCopy = Arrays.copyOfRange(columnBytes, vecStart, vecStart + vecLen);
+			if (colType.getCategory() == TypeDescription.Category.STRING ||colType.getCategory() == TypeDescription.Category.VARCHAR) {
+				bytesObj = new String(vecCopy);
+			} else if (colType.getCategory() == TypeDescription.Category.CHAR) {
+				String charStr = new String(vecCopy);
+				bytesObj = charStr;
+			} else {
+				bytesObj = vecCopy;
+			}
+		}
+		return bytesObj;
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-orc/src/main/java/org/talend/orc/ORCWriteUtils.java

@@ -0,0 +1,884 @@
+package org.talend.orc;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DateColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MultiValuedColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.UnionColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
+import org.apache.orc.TypeDescription;
+
+public class ORCWriteUtils {
+	private static final Predicate<Object> isInteger = Integer.class::isInstance;
+	private static final Predicate<Object> isLong = Long.class::isInstance;
+	private static final Predicate<Object> isDouble = Double.class::isInstance;
+	private static final Predicate<Object> isString = String.class::isInstance;
+	private static final Predicate<Object> isBigDecimal = BigDecimal.class::isInstance;
+	private static final Predicate<Object> isDate = Date.class::isInstance;
+
+	public static void setColumnByName(VectorizedRowBatch batch, String columnName, TypeDescription schema,
+			Object colVal, int rowNum) {
+		List<String> allColumnNames = schema.getFieldNames();
+		int colIndex = allColumnNames.indexOf(columnName);
+		if (colIndex < 0 || colIndex > batch.cols.length - 1) {
+			return;
+		} else {
+			org.apache.hadoop.hive.ql.exec.vector.ColumnVector colVector = batch.cols[colIndex];
+			TypeDescription fieldType = schema.getChildren().get(colIndex);
+			setColumn(colVal, fieldType, columnName, colVector, rowNum);
+		}
+
+	}
+
+	/**
+	 * Add a column value that is a String or a byte[] array.
+	 *
+	 * @param colVal         the column value object
+	 * @param fieldName      the name of the field (for error reporting)
+	 * @param bytesColVector the BytesColumnVector that the byte array will be added
+	 *                       to.
+	 * @param rowNum         the ORC file row number
+	 */
+	private static void setByteColumnVector(Object colVal, String fieldName, BytesColumnVector bytesColVector,
+			int rowNum) {
+		if (colVal instanceof byte[] || colVal instanceof String || colVal instanceof Character) {
+			byte[] byteVec;
+			if (colVal instanceof String) {
+				String strVal = (String) colVal;
+				byteVec = strVal.getBytes(StandardCharsets.UTF_8);
+			} else if (colVal instanceof Character) {
+				String strVal = String.valueOf((char) colVal);
+				byteVec = strVal.getBytes(StandardCharsets.UTF_8);
+			} else {
+				byteVec = (byte[]) colVal;
+			}
+			bytesColVector.setRef(rowNum, byteVec, 0, byteVec.length);
+		} else {
+			throw new RuntimeException(orcExceptionMsg("byte[] or String type expected for field ", fieldName, rowNum));
+		}
+	}
+
+	private static void setDecimalVector(Object colVal, String fieldName, DecimalColumnVector decimalColVector,
+			int rowNum) {
+		if (colVal instanceof BigDecimal) {
+			BigDecimal bigDecimal = (BigDecimal) colVal;
+			decimalColVector.precision = (short) bigDecimal.precision();
+			decimalColVector.scale = (short) bigDecimal.scale();
+			HiveDecimal hiveDecimal = HiveDecimal.create(bigDecimal);
+			HiveDecimalWritable writeableDecimal = new HiveDecimalWritable(hiveDecimal);
+			decimalColVector.vector[rowNum] = writeableDecimal;
+		} else {
+			throw new RuntimeException(orcExceptionMsg("BigDecimal type expected for field ", fieldName, rowNum));
+		}
+	}
+
+	private static void setDoubleVector(Object colVal, String fieldName, DoubleColumnVector doubleVector, int rowNum) {
+		if (colVal instanceof Double) {
+			doubleVector.vector[rowNum] = (Double) colVal;
+		} else if (colVal instanceof Float) {
+			Float fltVal = (Float) colVal;
+			doubleVector.vector[rowNum] = fltVal.doubleValue();
+		} else if (colVal instanceof BigDecimal) {
+			doubleVector.vector[rowNum] = ((BigDecimal) colVal).doubleValue();
+		} else {
+			throw new RuntimeException(orcExceptionMsg("Double/Float/BigDecimal type expected for field ", fieldName, rowNum));
+		}
+	}
+
+	/**
+	 * Initialize a LongColumnVector value.
+	 * 
+	 * @param colVal     an object of type Boolean, Integer, Long or BigInteger.
+	 * @param fieldName  the field name in the schema
+	 * @param longVector the LongColumnVector
+	 * @param rowNum     the row number
+	 */
+	private static void setLongColumnVector(Object colVal, String fieldName, LongColumnVector longVector, int rowNum) {
+		if (colVal instanceof Boolean) {
+			Boolean bool = (Boolean) colVal;
+			longVector.vector[rowNum] = (bool.equals(Boolean.TRUE)) ? Long.valueOf(1) : Long.valueOf(0);
+		} else if (colVal instanceof Byte) {
+			longVector.vector[rowNum] = (Byte) colVal;
+		} else if (colVal instanceof Short) {
+			longVector.vector[rowNum] = (Short) colVal;
+		} else if (colVal instanceof Integer) {
+			longVector.vector[rowNum] = (Integer) colVal;
+		} else if (colVal instanceof Long) {
+			longVector.vector[rowNum] = (Long) colVal;
+		} else if (colVal instanceof BigInteger) {
+			BigInteger bigInt = (BigInteger) colVal;
+			longVector.vector[rowNum] = bigInt.longValue();
+		} else {
+			throw new RuntimeException(orcExceptionMsg("Long or Integer type expected for field ", fieldName, rowNum));
+		}
+	}
+
+	private static void setDateColumnVector(Object colVal, String fieldName, DateColumnVector dateVector, int rowNum) {
+		if (colVal instanceof Date) {
+			Date dateVal = (Date) colVal;
+			Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
+			cal.setTime(dateVal);
+			long epochDay = LocalDate.of(cal.get(Calendar.YEAR), cal.get(Calendar.MONTH)+1, cal.get(Calendar.DAY_OF_MONTH)).toEpochDay();
+			dateVector.vector[rowNum] = epochDay;
+		} else {
+			throw new RuntimeException(orcExceptionMsg("Date type expected for field ", fieldName, rowNum));
+		}
+	}
+
+	private static void setTimestampVector(Object colVal, String fieldName, TimestampColumnVector timestampVector,
+			int rowNum) {
+		if (colVal instanceof Timestamp) {
+			timestampVector.set(rowNum, (Timestamp) colVal);
+		} else if (colVal instanceof Date) {
+			Date date = (Date) colVal;
+			Timestamp ts = new Timestamp(date.getTime());
+
+			timestampVector.set(rowNum, ts);
+		} else {
+			throw new RuntimeException(
+					orcExceptionMsg("Date or Timestamp type expected for field ", fieldName, rowNum));
+		}
+	}
+
+	/**
+	 * <p>
+	 * A union column can contain column vectors of more than one type. In the
+	 * TypeDescription createUnion() is called to create a TypeDescription for a
+	 * union column. The union values are added by calling the addUnionChild()
+	 * method on this TypeDescription object.
+	 * </p>
+	 * <p>
+	 * The class fields in the UnionColumnVector are shown below:
+	 * </p>
+	 * 
+	 * <pre>
+	 *     public class UnionColumnVector extends ColumnVector {
+	 *        public int[] tags;
+	 *        public ColumnVector[] fields;
+	 * </pre>
+	 * <p>
+	 * A tag value (
+	 * 
+	 * <pre>
+	 * tags[rowNum]
+	 * </pre>
+	 * 
+	 * ) is associated with each field value (
+	 * 
+	 * <pre>
+	 * fields[rowNum])
+	 * </pre>
+	 * 
+	 * . The tag value serves as an index for the field type. For example, if there
+	 * are three field types defined:
+	 * <ol>
+	 * <li>Long</li>
+	 * <li>Double</li>
+	 * <li>String</li>
+	 * </ol>
+	 * The tag will have a value in the range of [0..2]
+	 * </p>
+	 * <p>
+	 * The tag value is needed to initialize the ColumnVector since without the tag
+	 * there is no way to know which union child should be initialized.
+	 * </p>
+	 *
+	 * @param colVal      a Pair&lt;ColumnVector.Type, Object&gt; object with the
+	 *                    union type and the object that will be used to initialize
+	 *                    the union child ColumnVector.
+	 * @param fieldName   The name of the union field
+	 * @param unionVector The UnionColumnVector to be initialized
+	 * @param rowNum      the ORC file row number.
+	 */
+	private static void setUnionColumnVector(Object colVal, TypeDescription unionTypeDesc, String fieldName,
+			UnionColumnVector unionVector, int rowNum) {
+		@SuppressWarnings("unchecked")
+		Pair<TypeDescription, Object> unionValuePair = (Pair<TypeDescription, Object>) colVal;
+		TypeDescription unionValType = unionValuePair.getLeft();
+		List<TypeDescription> unionChildrenTypes = unionTypeDesc.getChildren();
+		Object unionColVal = unionValuePair.getRight();
+		boolean found = false;
+		for (int i = 0; i < unionChildrenTypes.size(); i++) {
+			if (unionChildrenTypes.get(i).getCategory() == unionValType.getCategory()) {
+				unionVector.tags[rowNum] = i;
+				ColumnVector unionFieldVec = unionVector.fields[i];
+				setColumn(unionColVal, unionChildrenTypes.get(i), fieldName, unionFieldVec, rowNum);
+				found = true;
+				break;
+			}
+		}
+		if (!found) {
+			throw new RuntimeException("writeUnionColumnVector: Bad type enumeration "
+					+ unionValType.getCategory().getName() + " passed for field " + fieldName);
+		}
+	}
+
+	private static void setLongListVector(List<Object> longValList, LongColumnVector longVector, int offset,
+			String fieldName) {
+		for (int i = 0; i < longValList.size(); i++) {
+			Object objVal = longValList.get(i);
+			if (objVal != null) {
+				if (objVal instanceof Integer) {
+					longVector.vector[offset + i] = (Integer) objVal;
+				} else if (objVal instanceof Long) {
+					longVector.vector[offset + i] = (Long) objVal;
+				} else {
+					throw new RuntimeException("List<Integer> expected for field " + fieldName);
+				}
+			} else {
+				longVector.isNull[offset + i] = true;
+				longVector.noNulls = false;
+			}
+		}
+	}
+
+	private static void setLongList(List<Object> colValList, ListColumnVector listVector, String fieldName,
+			int rowNum) {
+		LongColumnVector longVector = (LongColumnVector) listVector.child;
+		int offset = (int) listVector.offsets[rowNum];
+		setLongListVector(colValList, longVector, offset, fieldName);
+	}
+
+	private static void setDoubleListVector(List<Object> doubleValList, DoubleColumnVector doubleVector, int offset,
+			String fieldName) {
+		for (int i = 0; i < doubleValList.size(); i++) {
+			Object objVal = doubleValList.get(i);
+			if (objVal != null) {
+				if (objVal instanceof Double) {
+					doubleVector.vector[offset + i] = (Double) objVal;
+				} else if (objVal instanceof Float) {
+					Float fltVal = (Float) objVal;
+					doubleVector.vector[offset + i] = fltVal.doubleValue();
+				} else {
+					throw new RuntimeException("List<Double> expected for field " + fieldName);
+				}
+			} else {
+				doubleVector.isNull[offset + i] = true;
+				doubleVector.noNulls = false;
+			}
+		}
+	}
+
+	private static void setDoubleList(List<Object> doubleValList, ListColumnVector listVector, String fieldName,
+			int rowNum) {
+		DoubleColumnVector vecChild = (DoubleColumnVector) listVector.child;
+		int offset = (int) listVector.offsets[rowNum];
+		setDoubleListVector(doubleValList, vecChild, offset, fieldName);
+	}
+
+	private static void setTimestampListVector(List<Object> valueList, TimestampColumnVector timestampVector,
+			int offset, String fieldName) {
+		for (int i = 0; i < valueList.size(); i++) {
+			Object objVal = valueList.get(i);
+			if (objVal != null) {
+				if (objVal instanceof Date) {
+					Timestamp ts = (objVal instanceof Timestamp) ? (Timestamp) objVal
+							: new Timestamp(((Date) objVal).getTime());
+					timestampVector.time[offset + i] = ts.getTime();
+					timestampVector.nanos[offset + i] = ts.getNanos();
+				} else {
+					throw new RuntimeException("List<Date> or List<Timestamp> expected for field " + fieldName);
+				}
+			} else {
+				timestampVector.isNull[offset + i] = true;
+				timestampVector.noNulls = false;
+			}
+		}
+	}
+
+	/**
+	 * Initialize the vector values for a ListColumnVector of Date or Timestamp
+	 * values.
+	 *
+	 * @param colValList a list of Timestamp or java.util.Date objects
+	 * @param listVector A ListColumnVector with a child that will contain the
+	 *                   vector values.
+	 * @param fieldName  The field name in the schema for this ORC element
+	 * @param rowNum     The ORC file row number
+	 */
+	private static void setTimestampList(List<Object> colValList, ListColumnVector listVector, String fieldName,
+			int rowNum) {
+		TimestampColumnVector timestampVector = (TimestampColumnVector) listVector.child;
+		int offset = (int) listVector.offsets[rowNum];
+		setTimestampListVector(colValList, timestampVector, offset, fieldName);
+	}
+
+	private static void setDecimalListVector(List<Object> decimalValList, DecimalColumnVector decimalVector, int offset,
+			String fieldName) {
+		for (int i = 0; i < decimalValList.size(); i++) {
+			Object objVal = decimalValList.get(i);
+			if (objVal != null) {
+				if (objVal instanceof BigDecimal) {
+					BigDecimal bigDecimal = (BigDecimal) objVal;
+					decimalVector.precision = (short) bigDecimal.precision();
+					decimalVector.scale = (short) bigDecimal.scale();
+					HiveDecimal hiveDecimal = HiveDecimal.create(bigDecimal);
+					HiveDecimalWritable writeableDecimal = new HiveDecimalWritable(hiveDecimal);
+					decimalVector.vector[offset + i] = writeableDecimal;
+				} else {
+					throw new RuntimeException("BigDecimal value expected for field " + fieldName);
+				}
+			} else {
+				decimalVector.isNull[offset + i] = true;
+				decimalVector.noNulls = false;
+			}
+		}
+	}
+
+	/**
+	 *
+	 * @param colValList a list of BigDecimal values to initialize the
+	 *                   ListColumnVector child
+	 * @param listVector the ListColumnVector with the DecimalColumnVector child
+	 * @param fieldName  the field name for the ListColumnVector/DecimalColumnVector
+	 *                   column
+	 * @param rowNum     the ORC file row number
+	 */
+	private static void setDecimalList(List<Object> colValList, ListColumnVector listVector, String fieldName,
+			int rowNum) {
+		DecimalColumnVector decimalVector = (DecimalColumnVector) listVector.child;
+		int offset = (int) listVector.offsets[rowNum];
+		setDecimalListVector(colValList, decimalVector, offset, fieldName);
+	}
+
+	private static void setBytesListVector(List<Object> valueList, BytesColumnVector bytesVector, int offset,
+			String fieldName) {
+		for (int i = 0; i < valueList.size(); i++) {
+			Object objVal = valueList.get(i);
+			if (objVal != null) {
+				if (objVal instanceof byte[] || objVal instanceof String) {
+					byte[] byteVec = (objVal instanceof byte[]) ? (byte[]) objVal
+							: ((String) objVal).getBytes(StandardCharsets.UTF_8);
+					bytesVector.vector[offset + i] = byteVec;
+					bytesVector.length[offset + i] = byteVec.length;
+				} else {
+					throw new RuntimeException("String or byte[] value expected for field " + fieldName);
+				}
+			} else {
+				bytesVector.isNull[offset + i] = true;
+				bytesVector.length[offset + i] = 0;
+				bytesVector.noNulls = false;
+			}
+		}
+	}
+
+	/**
+	 * Initialize a ListColumnVector with a BytesColumnVector child with byte[]
+	 * values.
+	 *
+	 * @param colValList a list of byte[] or String values
+	 * @param listVector the parent ListColumnVector
+	 * @param fieldName  the field name for the ORC column that contains the
+	 *                   ListColumnVector
+	 * @param rowNum     the ORC file row number
+	 */
+	private static void setBytesList(List<Object> colValList, ListColumnVector listVector, String fieldName,
+			int rowNum) {
+		BytesColumnVector bytesVector = (BytesColumnVector) listVector.child;
+		int offset = (int) listVector.offsets[rowNum];
+		setBytesListVector(colValList, bytesVector, offset, fieldName);
+	}
+
+	private static void setMultiValuedVectorParameters(MultiValuedColumnVector multiVector, int vecLength, int rowNum) {
+		multiVector.lengths[rowNum] = vecLength;
+		if (rowNum > 0) {
+			multiVector.offsets[rowNum] = multiVector.lengths[rowNum - 1] + multiVector.offsets[rowNum - 1];
+		}
+	}
+
+	private static void setListVectorParameters(ListColumnVector listVec, int maxBatchSize, int vecLength, int rowNum) {
+		setMultiValuedVectorParameters(listVec, vecLength, rowNum);
+		listVec.child.ensureSize(maxBatchSize * vecLength, true);
+	}
+
+	/**
+	 * Initialize a ListColumnVector. The child of the vector is limited to the
+	 * scalar types long, double, String (or byte[])), BigDecimal or Date (or
+	 * Timestamp).
+	 *
+	 * @param colVal     a List&lt;Object&gt;
+	 * @param typeDesc   the schema definition for this column
+	 * @param fieldName  the column field name
+	 * @param listVector the ListColumnVector parent of the vector type child
+	 * @param rowNum     the ORC file row number.
+	 */
+	private static void setListColumnVector(Object colVal, TypeDescription typeDesc, String fieldName,
+			ListColumnVector listVector, int rowNum) {
+		if (colVal instanceof List) {
+			@SuppressWarnings("unchecked")
+			List<Object> objValList = (List<Object>) colVal;
+			final int maxBatchSize = typeDesc.createRowBatch().getMaxSize();
+			setListVectorParameters(listVector, maxBatchSize, objValList.size(), rowNum);
+			ColumnVector.Type childType = listVector.child.type;
+			switch (childType) {
+			case LONG:
+				setLongList(objValList, listVector, fieldName, rowNum);
+				break;
+			case DOUBLE:
+				setDoubleList(objValList, listVector, fieldName, rowNum);
+				break;
+			case BYTES:
+				setBytesList(objValList, listVector, fieldName, rowNum);
+				break;
+			case DECIMAL:
+				setDecimalList(objValList, listVector, fieldName, rowNum);
+				break;
+			case TIMESTAMP:
+				setTimestampList(objValList, listVector, fieldName, rowNum);
+				break;
+			default:
+				throw new RuntimeException(childType.name() + " is not supported for ListColumnVector columns");
+			}
+		} else {
+			throw new RuntimeException("List value expected for field " + fieldName);
+		}
+	}
+
+	/**
+	 * Test that all elements in an Object list are of a particular type
+	 * 
+	 * @param objList  the Object list that is tested
+	 * @param typeTest a function that compares against a particular Object type
+	 * @return true if all elements are of the test type, false if one or more
+	 *         elements are not of that type.
+	 */
+	private static boolean isListType(List<Object> objList, Predicate<Object> typeTest) {
+		return !objList.stream().map(typeTest::test).collect(Collectors.toList()).contains(false);
+	}
+
+	/**
+	 * Initialize a ColumnVector with Long values.
+	 *
+	 * @param valueList a list of Long values
+	 * @param colVector the LongColumnVector that will be initialized with the Long
+	 *                  values
+	 * @param offset    the offset[rownum] value for the array
+	 * @param fieldName the field name for the Map column
+	 */
+	private static void setLongMapValues(List<Object> valueList, ColumnVector colVector, int offset, String fieldName) {
+		if (isListType(valueList, isLong) || isListType(valueList, isInteger)) {
+			LongColumnVector longVector = (LongColumnVector) colVector;
+			setLongListVector(valueList, longVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " Long values expected");
+		}
+	}
+
+	/**
+	 * Initialize a ColumnVector with Double values.
+	 *
+	 * @param valueList a list of Double values
+	 * @param colVector the DoubleColumnVector that will be initialized with the
+	 *                  Double values
+	 * @param offset    the offset[rownum] value for the array
+	 * @param fieldName the field name for the Map column
+	 */
+	private static void setDoubleMapValues(List<Object> valueList, ColumnVector colVector, int offset,
+			String fieldName) {
+		if (isListType(valueList, isDouble)) {
+			DoubleColumnVector doubleVector = (DoubleColumnVector) colVector;
+			setDoubleListVector(valueList, doubleVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " Double values expected");
+		}
+	}
+
+	/**
+	 * Initialize a ColumnVector with String values.
+	 *
+	 * @param valueList a list of String values
+	 * @param colVector the BytesColumnVector that will be initialized with the
+	 *                  String values
+	 * @param offset    the offset[rownum] value for the array
+	 * @param fieldName the field name for the Map column
+	 */
+	private static void setStringMapValues(List<Object> valueList, ColumnVector colVector, int offset,
+			String fieldName) {
+		if (isListType(valueList, isString)) {
+			BytesColumnVector doubleVector = (BytesColumnVector) colVector;
+			setBytesListVector(valueList, doubleVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " String values expected");
+		}
+	}
+
+	/**
+	 * Initialize a ColumnVector with BigDeciml values.
+	 *
+	 * @param valueList a list of BigDecimal
+	 * @param colVector the DecimalColumnVector that will be initialized with the
+	 *                  BigDecimal values
+	 * @param offset    the offset[rownum] value for the array
+	 * @param fieldName the field name for the Map column
+	 */
+	private static void setDecimalMapValues(List<Object> valueList, ColumnVector colVector, int offset,
+			String fieldName) {
+		if (isListType(valueList, isBigDecimal)) {
+			DecimalColumnVector decimalVector = (DecimalColumnVector) colVector;
+			setDecimalListVector(valueList, decimalVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " BigDecimal values expected");
+		}
+	}
+
+	/**
+	 * Initialize a ColumnVector with timestamp values.
+	 *
+	 * @param valueList a list of Date (or Timestamp) objects
+	 * @param colVector the TimestampColumnVector that will be initialized with the
+	 *                  Timestamp values
+	 * @param offset    the offset[rownum] value for the array
+	 * @param fieldName the field name for the Map column
+	 */
+	private static void setTimestampMapValues(List<Object> valueList, ColumnVector colVector, int offset,
+			String fieldName) {
+		if (isListType(valueList, isDate)) {
+			TimestampColumnVector timestampVector = (TimestampColumnVector) colVector;
+			setTimestampListVector(valueList, timestampVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " Date or Timestamp values expected");
+		}
+	}
+
+	/**
+	 * Set the MapColumn value array vector. The type for this vector is limited to
+	 * long, double, bytes (String), Decimal and Timestamp.
+	 *
+	 * @param valueList a list of Objects to initialize the Map column value array.
+	 * @param colVector the column array vector to be initialized with the map
+	 *                  values.
+	 * @param offset    the offset[rowNum] from the parent MapColumnVector
+	 * @param fieldName the name of the field for the MapColumnVector.
+	 */
+	private static void setMapValueVector(List<Object> valueList, ColumnVector colVector, int offset,
+			String fieldName) {
+		switch (colVector.type) {
+		case LONG:
+			setLongMapValues(valueList, colVector, offset, fieldName);
+			break;
+		case DOUBLE:
+			setDoubleMapValues(valueList, colVector, offset, fieldName);
+			break;
+		case BYTES:
+			setStringMapValues(valueList, colVector, offset, fieldName);
+			break;
+		case DECIMAL:
+			setDecimalMapValues(valueList, colVector, offset, fieldName);
+			break;
+		case TIMESTAMP:
+			setTimestampMapValues(valueList, colVector, offset, fieldName);
+			break;
+		default:
+			throw new RuntimeException(
+					"For field " + fieldName + " values must be long, double, String, BigDecimal or Timestamp");
+		}
+	}
+
+	/**
+	 * <p>
+	 * Initialize a MapColumnVector with Long key values.
+	 * </p>
+	 * 
+	 * @param mapSet    a set of {key, value} pairs, where the key values are Long
+	 *                  objects. The elements of this set will be used to initialize
+	 *                  the key and value array column vectors that are children of
+	 *                  the MapColumnVector.
+	 * @param mapVector the MapColumnVector. This ColumnVector has children for the
+	 *                  key and value arrays.
+	 * @param fieldName the field name for the map column vector column.
+	 * @param rowNum    the ORC file row number.
+	 */
+	private static void setLongKeyMap(Set<Map.Entry<Object, Object>> mapSet, MapColumnVector mapVector,
+			String fieldName, int rowNum) {
+		List<Object> keyValueList = mapSet.stream().map(Map.Entry::getKey).collect(Collectors.toList());
+		if (isListType(keyValueList, isLong)) {
+			LongColumnVector longVector = (LongColumnVector) mapVector.keys;
+			int offset = (int) mapVector.offsets[rowNum];
+			// set the key vector
+			setLongListVector(keyValueList, longVector, offset, fieldName);
+			// set the value vector
+			ColumnVector valueVector = mapVector.values;
+			List<Object> valueList = mapSet.stream().map(Map.Entry::getValue).collect(Collectors.toList());
+			setMapValueVector(valueList, valueVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " Long key type expected to match schema");
+		}
+	}
+
+	/**
+	 * <p>
+	 * Initialize a MapColumnVector with Double key values.
+	 * </p>
+	 * 
+	 * @param mapSet    a set of {key, value} pairs, where the key values are Double
+	 *                  objects. The elements of this set will be used to initialize
+	 *                  the key and value array column vectors that are children of
+	 *                  the MapColumnVector.
+	 * @param mapVector the MapColumnVector. This ColumnVector has children for the
+	 *                  key and value arrays.
+	 * @param fieldName the field name for the map column vector column.
+	 * @param rowNum    the ORC file row number.
+	 */
+	private static void setDoubleKeyMap(Set<Map.Entry<Object, Object>> mapSet, MapColumnVector mapVector,
+			String fieldName, int rowNum) {
+		List<Object> keyValueList = mapSet.stream().map(Map.Entry::getKey).collect(Collectors.toList());
+		if (isListType(keyValueList, isDouble)) {
+			DoubleColumnVector doubleVector = (DoubleColumnVector) mapVector.keys;
+			int offset = (int) mapVector.offsets[rowNum];
+			// set the key vector
+			setDoubleListVector(keyValueList, doubleVector, offset, fieldName);
+			// set the value vector
+			ColumnVector valueVector = mapVector.values;
+			List<Object> valueList = mapSet.stream().map(Map.Entry::getValue).collect(Collectors.toList());
+			setMapValueVector(valueList, valueVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " Long key type expected to match schema");
+		}
+	}
+
+	/**
+	 * <p>
+	 * Initialize a MapColumnVector with String key values.
+	 * </p>
+	 * 
+	 * @param mapSet    a set of {key, value} pairs, where the key values are String
+	 *                  objects. The elements of this set will be used to initialize
+	 *                  the key and value array column vectors that are children of
+	 *                  the MapColumnVector.
+	 * @param mapVector the MapColumnVector. This ColumnVector has children for the
+	 *                  key and value arrays.
+	 * @param fieldName the field name for the map column vector column.
+	 * @param rowNum    the ORC file row number.
+	 */
+	private static void setStringKeyMap(Set<Map.Entry<Object, Object>> mapSet, MapColumnVector mapVector,
+			String fieldName, int rowNum) {
+		List<Object> keyValueList = mapSet.stream().map(Map.Entry::getKey).collect(Collectors.toList());
+		if (isListType(keyValueList, isString)) {
+			BytesColumnVector byteVector = (BytesColumnVector) mapVector.keys;
+			int offset = (int) mapVector.offsets[rowNum];
+			// set the key array vector
+			setBytesListVector(keyValueList, byteVector, offset, fieldName);
+			// set the value array vector
+			ColumnVector valueVector = mapVector.values;
+			List<Object> valueList = mapSet.stream().map(Map.Entry::getValue).collect(Collectors.toList());
+			setMapValueVector(valueList, valueVector, offset, fieldName);
+		} else {
+			throw new RuntimeException("For field " + fieldName + " Long key type expected to match schema");
+		}
+	}
+
+	private static void setMapVectorParameters(MapColumnVector mapVec, int maxBatchSize, int vecLength, int rowNum) {
+		setMultiValuedVectorParameters(mapVec, vecLength, rowNum);
+		mapVec.keys.ensureSize(maxBatchSize + vecLength, true);
+		mapVec.values.ensureSize(maxBatchSize + vecLength, true);
+	}
+
+	/**
+	 * <p>
+	 * Set the Map key and value elements for a MapColumnVector
+	 * </p>
+	 * <p>
+	 * A MapColumnVector has a single ColumnVector type for each of the map key and
+	 * map values. For example, the ColumnVector for the key values could be a
+	 * BytesColumnVector (a set of String keys). The values could be a
+	 * LongColumnVector.
+	 * </p>
+	 * <p>
+	 * In the documentation there is no restriction given for the map key type. This
+	 * code limits the key types to scalar values: string, long, double.
+	 * </p>
+	 * </p>
+	 * <p>
+	 * The documentation does not limit the map value types. This code limites the
+	 * map values to the same types that are supported for ListColumnVectors: long,
+	 * double, bytes (String), Decimal and Timestamp.
+	 * </p>
+	 *
+	 * @param colVal    a HashMap object
+	 * @param typeDesc  the schema description for the MapColumnVector column
+	 * @param fieldName the field name of the MapColumnVector column
+	 * @param mapVector The parent MapColumnVector
+	 * @param rowNum    the ORC file column number.
+	 */
+	private static void setMapColumnVector(Object colVal, TypeDescription typeDesc, String fieldName,
+			MapColumnVector mapVector, int rowNum) {
+		if (colVal == null) {
+			mapVector.isNull[rowNum] = true;
+			mapVector.noNulls = false;
+		} else {
+			if (colVal instanceof HashMap) {
+				@SuppressWarnings("unchecked")
+				Map<Object, Object> rawMap = (HashMap<Object, Object>) colVal;
+				int mapLen = rawMap.size();
+				final int maxBatchSize = typeDesc.createRowBatch().getMaxSize();
+				setMapVectorParameters(mapVector, maxBatchSize, mapLen, rowNum);
+				if (ORCCommonUtils.checkMapColumnVectorTypes(mapVector)) {
+					Set<Map.Entry<Object, Object>> mapSet = rawMap.entrySet();
+					switch (mapVector.keys.type) {
+					case LONG:
+						setLongKeyMap(mapSet, mapVector, fieldName, rowNum);
+						break;
+					case DOUBLE:
+						setDoubleKeyMap(mapSet, mapVector, fieldName, rowNum);
+						break;
+					case BYTES:
+						setStringKeyMap(mapSet, mapVector, fieldName, rowNum);
+						break;
+					default: {
+						break;
+						/* This block left intentionally empty */
+					}
+					}
+				} else {
+					throw new RuntimeException(
+							"For field " + fieldName + " key types are limited to string, long and double. "
+									+ "value types are limited to long, double, String, decimal and timestamp");
+				}
+			}
+		}
+	}
+
+	/**
+	 * Set a column value in an ORC a row that will be written to the ORC file.
+	 *
+	 * @param colVal    an Object containing the values to be written to the column
+	 * @param typeDesc  the TypeDescription from the schema that defines the column
+	 * @param fieldName the column field name
+	 * @param vector    the ColumnVector that will be initialized with the values in
+	 *                  the colVal argument.
+	 * @param rowNum    the ORC file row number.
+	 */
+	public static void setColumn(Object colVal, TypeDescription typeDesc, String fieldName, ColumnVector vector,
+			int rowNum) {
+		if (colVal == null) {
+			vector.isNull[rowNum] = true;
+			vector.noNulls = false;
+		} else {
+			switch (vector.type) {
+			case LONG: {
+				if (vector instanceof DateColumnVector) {
+					DateColumnVector dateVector = (DateColumnVector) vector;
+					setDateColumnVector(colVal, fieldName, dateVector, rowNum);
+				} else {
+					LongColumnVector longVector = (LongColumnVector) vector;
+					setLongColumnVector(colVal, fieldName, longVector, rowNum);
+				}
+				break;
+			}
+			case DOUBLE: {
+				DoubleColumnVector doubleVector = (DoubleColumnVector) vector;
+				setDoubleVector(colVal, fieldName, doubleVector, rowNum);
+				break;
+			}
+			case BYTES: {
+				BytesColumnVector bytesColVector = (BytesColumnVector) vector;
+				setByteColumnVector(colVal, fieldName, bytesColVector, rowNum);
+				break;
+			}
+			case DECIMAL: {
+				DecimalColumnVector decimalVector = (DecimalColumnVector) vector;
+				setDecimalVector(colVal, fieldName, decimalVector, rowNum);
+				break;
+			}
+			case DECIMAL_64:
+				throw new RuntimeException("Field: " + fieldName + ", Decimal64ColumnVector is not supported");
+			case TIMESTAMP: {
+				TimestampColumnVector timestampVector = (TimestampColumnVector) vector;
+				setTimestampVector(colVal, fieldName, timestampVector, rowNum);
+				break;
+			}
+			case INTERVAL_DAY_TIME:
+				throw new RuntimeException("Field: " + fieldName + ", HiveIntervalDayTime is not supported");
+			case STRUCT: {
+				StructColumnVector structVector = (StructColumnVector) vector;
+//				setStructColumnVector(colVal, typeDesc, fieldName, structVector, rowNum);
+				break;
+			}
+			case LIST: {
+				ListColumnVector listVector = (ListColumnVector) vector;
+				setListColumnVector(colVal, typeDesc, fieldName, listVector, rowNum);
+				break;
+			}
+			case MAP: {
+				MapColumnVector mapVector = (MapColumnVector) vector;
+				setMapColumnVector(colVal, typeDesc, fieldName, mapVector, rowNum);
+				break;
+			}
+			case UNION: {
+				UnionColumnVector unionVector = (UnionColumnVector) vector;
+				setUnionColumnVector(colVal, typeDesc, fieldName, unionVector, rowNum);
+				break;
+			}
+			default:
+				throw new RuntimeException("setColumn: Internal error: unexpected ColumnVector subtype");
+			} // switch
+		} // else
+	} // setColumn
+
+	private static String orcExceptionMsg(String prefixMsg, String fieldName, int rowNum) {
+		return prefixMsg + fieldName + " in row " + rowNum;
+	}
+
+	public static TypeDescription detectType(Object value) {
+		TypeDescription type = null;
+		if (value != null) {
+			if (value instanceof Boolean) {
+				type = TypeDescription.createBoolean();
+			} else if (value instanceof Short) {
+				type = TypeDescription.createShort();
+			} else if (value instanceof Integer) {
+				type = TypeDescription.createInt();
+			} else if (value instanceof Long) {
+				type = TypeDescription.createLong();
+			} else if (value instanceof Timestamp) {
+				type = TypeDescription.createTimestamp();
+			} else if (value instanceof BigDecimal) {
+				type = TypeDescription.createDecimal();
+			} else if (value instanceof Byte) {
+				type = TypeDescription.createByte();
+			} else if (value instanceof Float) {
+				type = TypeDescription.createFloat();
+			} else if (value instanceof Double) {
+				type = TypeDescription.createDouble();
+			} else if (value instanceof String) {
+				type = TypeDescription.createString();
+			} else if (value instanceof Date) {
+				type = TypeDescription.createDate();
+			} else if (value instanceof byte[]) {
+				type = TypeDescription.createBinary();
+			} else {
+				throw new RuntimeException(
+						value.getClass().getName() + " is not supported for ListColumnVector columns");
+			}
+		} else {
+			type = TypeDescription.createString();
+		}
+
+		return type;
+	}
+}

main/plugins/org.talend.designer.components.libs/libs_src/talend-orc/src/test/java/org/talend/orc/ORCUtilsTest.java

@@ -0,0 +1,253 @@
+package org.talend.orc;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.CompressionKind;
+import org.apache.orc.OrcFile;
+import org.apache.orc.OrcFile.WriterOptions;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class ORCUtilsTest {
+
+	private static File localFolder;
+
+	@BeforeAll
+	static void setup() throws IOException {
+		localFolder = createTempDirectory();
+	}
+
+	@AfterAll
+	static void dispose() {
+		localFolder.delete();
+	}
+
+	/**
+	 * Test that an exception is thrown if a Date type is written.
+	 *
+	 * At the time this test was written, the ORC writer did not correctly write the
+	 * date epoch value to the ORC file. The value was written as a 32-bit int,
+	 * instead of a 64 bit long. As a result, the date is incorrect. A timestamp
+	 * value should be used instead.
+	 * 
+	 * @param tempDirPath
+	 * @throws ORCFileException
+	 * @throws InterruptedException
+	 */
+	@Test
+	void testAllDataTypes() throws Throwable {
+		String filePath = localFolder.getAbsolutePath() + "/testAllTypes.orc";
+		writeData(filePath);
+
+		readAndCheckData(filePath);
+
+	}
+
+	@Test
+	void testDetectType() {
+		Assertions.assertEquals(TypeDescription.Category.BOOLEAN, ORCWriteUtils.detectType(true).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.SHORT,
+				ORCWriteUtils.detectType(Short.valueOf("1")).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.INT, ORCWriteUtils.detectType(1).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.LONG, ORCWriteUtils.detectType(1L).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.TIMESTAMP,
+				ORCWriteUtils.detectType(new Timestamp(System.currentTimeMillis())).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.DECIMAL,
+				ORCWriteUtils.detectType(new BigDecimal("1")).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.BYTE,
+				ORCWriteUtils.detectType(Byte.valueOf("1")).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.FLOAT, ORCWriteUtils.detectType(1.0f).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.DOUBLE, ORCWriteUtils.detectType(1.0).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.STRING, ORCWriteUtils.detectType("test").getCategory());
+		Assertions.assertEquals(TypeDescription.Category.DATE, ORCWriteUtils.detectType(new Date()).getCategory());
+		Assertions.assertEquals(TypeDescription.Category.BINARY,
+				ORCWriteUtils.detectType("test".getBytes()).getCategory());
+	}
+
+	private void writeData(String filePath) throws Throwable {
+		TypeDescription schema = TypeDescription.createStruct();
+		schema.addField("t_boolean", TypeDescription.createBoolean());
+		schema.addField("t_byte", TypeDescription.createByte());
+		schema.addField("t_bytes", TypeDescription.createBinary());
+		schema.addField("t_char", TypeDescription.createChar());
+		schema.addField("t_date", TypeDescription.createDate());
+		schema.addField("t_ts", TypeDescription.createTimestamp());
+		schema.addField("t_double", TypeDescription.createDouble());
+		schema.addField("t_float", TypeDescription.createFloat());
+		schema.addField("t_decimal", TypeDescription.createDecimal().withPrecision(18).withScale(5));
+		schema.addField("t_int", TypeDescription.createInt());
+		schema.addField("t_long", TypeDescription.createLong());
+		schema.addField("t_short", TypeDescription.createShort());
+		schema.addField("t_string", TypeDescription.createString());
+		schema.addField("t_list", TypeDescription.createList(TypeDescription.createString()));
+
+		WriterOptions writerOption = OrcFile.writerOptions(new Configuration()) //
+				.overwrite(true) //
+				.compress(CompressionKind.valueOf("ZLIB")).setSchema(schema); //
+
+		Writer writer = OrcFile.createWriter(new Path(filePath), writerOption);
+		VectorizedRowBatch batch = schema.createRowBatch(100);
+		for (int r = 0; r < 1000; ++r) {
+			int row = batch.size++;
+			for (int i = 0; i < batch.cols.length; i++) {
+				ColumnVector vector = batch.cols[i];
+				TypeDescription type = schema.getChildren().get(i);
+				switch (vector.type) {
+				case BYTES:
+					if (type.getCategory() == TypeDescription.Category.BINARY) {
+						ORCWriteUtils.setColumn(("this is byte[] " + r).getBytes(), null, "t_bytes", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.STRING) {
+						if(r==666) {
+							ORCWriteUtils.setColumn(null, null, "t_string", vector, row);
+						}else {
+							ORCWriteUtils.setColumn(("this is String " + r), null, "t_string", vector, row);
+						}
+					} else if (type.getCategory() == TypeDescription.Category.CHAR) {
+						ORCWriteUtils.setColumn("talend".charAt(r % 6), null, "t_char", vector, row);
+					} else {
+						throw new RuntimeException(type.getCategory() + " is not supported as BYTES vector");
+					}
+					break;
+				case DECIMAL:
+					ORCWriteUtils.setColumn(new BigDecimal(r + ".12345"), null, "t_decimal", vector, row);
+					break;
+				case DOUBLE:
+					if (type.getCategory() == TypeDescription.Category.DOUBLE) {
+						ORCWriteUtils.setColumn(r + 0.123, null, "t_double", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.FLOAT) {
+						ORCWriteUtils.setColumn(r + 0.456f, null, "t_float", vector, row);
+					} else {
+						throw new RuntimeException(type.getCategory() + " is not supported as DOUBLE vector");
+					}
+					break;
+				case LONG:
+					if (type.getCategory() == TypeDescription.Category.BOOLEAN) {
+						ORCWriteUtils.setColumn(true, null, "t_boolean", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.BYTE) {
+						ORCWriteUtils.setColumn((byte)(r % 128), null, "t_byte", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.INT) {
+						ORCWriteUtils.setColumn(r, null, "t_int", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.SHORT) {
+						ORCWriteUtils.setColumn((short)(r % 256), null, "t_short", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.LONG) {
+						ORCWriteUtils.setColumn(r * 1000L, null, "t_long", vector, row);
+					} else if (type.getCategory() == TypeDescription.Category.DATE) {
+						Date d = new Date(1633687854031L);
+						ORCWriteUtils.setColumn(d, null, "t_date", vector, row);
+					} else {
+						throw new RuntimeException(type.getCategory() + " is not supported as LONG vector");
+					}
+					break;
+				case TIMESTAMP:
+					Timestamp ts = new java.sql.Timestamp(1633687854031L);
+					ts.setNanos(123456789);
+					ORCWriteUtils.setColumn(ts, null, "t_ts", vector, row);
+					break;
+				case LIST:
+					List<String> values = new ArrayList<>();
+					values.add("v1_" + r);
+					values.add("v2_" + r);
+					values.add("v3_" + r);
+					ORCWriteUtils.setColumn(values, ORCWriteUtils.detectType("v1_" + r), "t_ list", vector, row);
+					break;
+				default:
+					throw new RuntimeException(vector.type + " is not supported");
+
+				}
+			}
+			if (batch.size == batch.getMaxSize()) {
+				writer.addRowBatch(batch);
+				batch.reset();
+			}
+		}
+		if (batch.size != 0) {
+			writer.addRowBatch(batch);
+		}
+
+		writer.close();
+	}
+
+	private void readAndCheckData(String filePath) throws Throwable {
+
+		Reader reader = OrcFile.createReader(new Path(filePath), OrcFile.readerOptions(new Configuration()));
+		TypeDescription schema = reader.getSchema();
+		VectorizedRowBatch batch = schema.createRowBatch();
+		RecordReader rowIterator = reader.rows(reader.options().schema(schema));
+		int nuberLine = 0;
+		List<Object> nb_500 = new ArrayList<>();
+		List<Object> nb_666 = new ArrayList<>();
+		while (rowIterator.nextBatch(batch)) {
+			ColumnVector[] colVectors = batch.cols;
+			for (int row = 0; row < batch.size; ++row) {
+				nuberLine++;
+				for (String columnName : schema.getFieldNames()) {
+					ColumnVector colVector = colVectors[schema.getFieldNames().indexOf(columnName)];
+					int colRow = colVector.isRepeating ? 0 : row;
+					Object value = ORCReadUtils.readColumnByName(batch, columnName, schema, colRow);
+					if (nuberLine == 500) {
+						nb_500.add(value);
+					}else if (nuberLine == 667) {
+						nb_666.add(value);
+					}
+				}
+			}
+		}
+		Assertions.assertEquals(true, nb_500.get(0));
+		Assertions.assertEquals(Byte.valueOf("115"), nb_500.get(1));
+		Assertions.assertEquals("this is byte[] 499", new String((byte[]) nb_500.get(2)));
+		Assertions.assertEquals("a", nb_500.get(3));
+		Date t_date = (Date) nb_500.get(4);
+		Assertions.assertEquals((1633687854000L/86400000), t_date.getTime()/86400000);
+		Timestamp t_ts = (Timestamp) nb_500.get(5);
+		Assertions.assertEquals(1633687854123L, t_ts.getTime());
+		Assertions.assertEquals(123456789, t_ts.getNanos());
+		Assertions.assertEquals(499.123, nb_500.get(6));
+		Assertions.assertEquals(499.456f, (((float) nb_500.get(7)) * 1000) / 1000f);
+		Assertions.assertEquals(new BigDecimal("499.12345"), nb_500.get(8));
+		Assertions.assertEquals(499, nb_500.get(9));
+		Assertions.assertEquals(499000L, nb_500.get(10));
+		Assertions.assertEquals(Short.valueOf("243"), nb_500.get(11));
+		Assertions.assertEquals("this is String 499", nb_500.get(12));
+		Assertions.assertArrayEquals(Arrays.asList("v1_499", "v2_499", "v3_499").toArray(),
+				((List<Object>) nb_500.get(13)).toArray());
+		
+		//NB_LINE 666
+		Assertions.assertNull( nb_666.get(12));
+
+		rowIterator.close();
+
+	}
+
+	public static File createTempDirectory() throws IOException {
+		final File temp;
+
+		temp = File.createTempFile("temp", Long.toString(System.nanoTime()));
+		if (!temp.delete()) {
+			throw new IOException("Could not delete temp file: " + temp.getAbsolutePath());
+		}
+
+		if (!temp.mkdir()) {
+			throw new IOException("Could not create temp directory: " + temp.getAbsolutePath());
+		}
+
+		return temp;
+	}
+
+}

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.libs/libs_src/talendzip/pom.xml

@@ -4,7 +4,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>org.talend.components</groupId>
 	<artifactId>talendzip</artifactId>
-	<version>1.1-20201120</version>
+	<version>1.2-20210901</version>
 	<build>
 		<plugins>
 			<plugin>
@@ -54,17 +54,17 @@
 			<artifactId>zip4j</artifactId>
 			<version>1.3.3</version>
 		</dependency>
-		
+
 		<dependency>
 			<groupId>org.apache.commons</groupId>
 			<artifactId>commons-compress</artifactId>
-			<version>1.19</version>
+			<version>1.21</version>
 		</dependency>
-		
+
 		<dependency>
-			<groupId>org.talend.libraries</groupId>
-			<artifactId>checkArchive-1.1-20190917</artifactId>
-			<version>6.0.0</version>
+			<groupId>org.talend.components</groupId>
+			<artifactId>checkArchive</artifactId>
+			<version>1.2-20210901</version>
 		</dependency>
 	</dependencies>
 </project>

main/plugins/org.talend.designer.components.localprovider/components/tAS400Input/tAS400Input_java.xml

@@ -199,6 +199,23 @@
 			<ITEM NAME="TRIM" FIELD="CHECK"></ITEM>
 		</ITEMS>
 	</PARAMETER>
+	
+	<PARAMETER
+      NAME="SET_QUERY_TIMEOUT"
+      FIELD="CHECK"
+      NUM_ROW="50"
+      >
+      <DEFAULT>false</DEFAULT>
+    </PARAMETER>
+
+    <PARAMETER
+      NAME="QUERY_TIMEOUT_IN_SECONDS"
+      FIELD="TEXT"
+      NUM_ROW="50"
+      SHOW_IF="SET_QUERY_TIMEOUT == 'true'"
+      >
+      <DEFAULT>30</DEFAULT>
+    </PARAMETER>
    </ADVANCED_PARAMETERS>
 
    <CODEGENERATION>

main/plugins/org.talend.designer.components.localprovider/components/tAS400Input/tAS400Input_messages.properties

@@ -22,6 +22,9 @@ DB_VERSION.ITEM.V6R1_V7R2=V6R1 to V7R2
 DB_VERSION.ITEM.V7R1_V7R3=V7R1 to V7R3
 MAPPING.NAME=Mapping
 
+SET_QUERY_TIMEOUT.NAME=Set Query Timeout
+QUERY_TIMEOUT_IN_SECONDS.NAME=Timeout (seconds)
+
 NB_LINE.NAME=Number of line
 GUESS_SCHEMA.NAME=Guess Schema
 CONNECTION.NAME=Component List

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

@@ -49,10 +49,14 @@ skeleton="../templates/db_output_bulk.skeleton"
 	String commitEvery = ElementParameterParser.getValue(node, "__COMMIT_EVERY__");
 
 	boolean isEnableDebug = ("true").equals(ElementParameterParser.getValue(node,"__ENABLE_DEBUG_MODE__"));
-
+	
 	boolean useBatchSize = "true".equals(ElementParameterParser.getValue(node,"__USE_BATCH_SIZE__"));
 	String batchSize=ElementParameterParser.getValue(node,"__BATCH_SIZE__");
 
+	boolean setQueryTimeOut = "true".equals(ElementParameterParser.getValue(node,"__SET_QUERY_TIMEOUT__"));
+	
+	String queryTimeOutInSeconds = ElementParameterParser.getValue(node,"__QUERY_TIMEOUT_IN_SECONDS__");
+	
 	String rejectConnName = null;
 	List<? extends IConnection> rejectConns = node.getOutgoingConnections("REJECT");
 	if(rejectConns != null && rejectConns.size() > 0) {
@@ -174,43 +178,47 @@ if(columnList != null && columnList.size() > 0) {
     StringBuilder updateSetStmt = actionSQLMap.get(UPDATE_SET_STMT);
     StringBuilder updateWhereStmt = actionSQLMap.get(UPDATE_WHERE_STMT);
     StringBuilder deleteWhereStmt = actionSQLMap.get(DELETE_WHERE_STMT);
-
+    %>
+    java.sql.PreparedStatement pstmt_<%=cid %> = null;
+    java.sql.PreparedStatement pstmtInsert_<%=cid %> = null;
+    java.sql.PreparedStatement pstmtUpdate_<%=cid %> = null;
+    <%
 	if(("INSERT").equals(dataAction)) {
 	    %>
 	    String insert_<%=cid%> = "INSERT INTO " + <%=table%> + " (<%=insertColName.toString()%>) VALUES (<%=insertValueStmt.toString()%>)";
-	    java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>, java.sql.Statement.RETURN_GENERATED_KEYS);
+	    pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>, java.sql.Statement.RETURN_GENERATED_KEYS);
 		resourceMap.put("pstmt_<%=cid %>", pstmt_<%=cid %>);
 	    <%
 	} else if (("UPDATE").equals(dataAction)) {
 	    %>
 	    String update_<%=cid%> = "UPDATE " + <%=table%> + " SET <%=updateSetStmt.toString()%> WHERE <%=updateWhereStmt.toString()%>";
-	    java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
+	    pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
 		resourceMap.put("pstmt_<%=cid %>", pstmt_<%=cid %>);
 	    <%
 	} else if (("INSERT_OR_UPDATE").equals(dataAction)) {
 	    %>
-	    java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement("SELECT COUNT(1) FROM " + <%=table%> + " WHERE <%=updateWhereStmt.toString()%>");
+	    pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement("SELECT COUNT(1) FROM " + <%=table%> + " WHERE <%=updateWhereStmt.toString()%>");
 		resourceMap.put("pstmt_<%=cid %>", pstmt_<%=cid %>);
 	    String insert_<%=cid%> = "INSERT INTO " + <%=table%> + " (<%=insertColName.toString()%>) VALUES (<%=insertValueStmt.toString()%>)";
-	    java.sql.PreparedStatement pstmtInsert_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
+	    pstmtInsert_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
 		resourceMap.put("pstmtInsert_<%=cid %>", pstmtInsert_<%=cid %>);
 	    String update_<%=cid%> = "UPDATE " + <%=table%> + " SET <%=updateSetStmt.toString()%> WHERE <%=updateWhereStmt.toString()%>";
-	    java.sql.PreparedStatement pstmtUpdate_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
+	    pstmtUpdate_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
 		resourceMap.put("pstmtUpdate_<%=cid %>", pstmtUpdate_<%=cid %>);
 	    <%
 	} else if (("UPDATE_OR_INSERT").equals(dataAction)) {
 	    %>
 	    String update_<%=cid%> = "UPDATE " + <%=table%> + " SET <%=updateSetStmt.toString()%> WHERE <%=updateWhereStmt.toString()%>";
-	    java.sql.PreparedStatement pstmtUpdate_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
+	    pstmtUpdate_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
 		resourceMap.put("pstmtUpdate_<%=cid %>", pstmtUpdate_<%=cid %>);
 	    String insert_<%=cid%> = "INSERT INTO " + <%=table%> + " (<%=insertColName.toString()%>) VALUES (<%=insertValueStmt.toString()%>)";
-	    java.sql.PreparedStatement pstmtInsert_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
+	    pstmtInsert_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
 		resourceMap.put("pstmtInsert_<%=cid %>", pstmtInsert_<%=cid %>);
 	    <%
 	} else if (("DELETE").equals(dataAction)) {
 	    %>
 	    String delete_<%=cid%> = "DELETE FROM " + <%=table%> + " WHERE <%=deleteWhereStmt.toString()%>";
-	    java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(delete_<%=cid%>);
+	    pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(delete_<%=cid%>);
 		resourceMap.put("pstmt_<%=cid %>", pstmt_<%=cid %>);
 	    <%
 	}
@@ -220,6 +228,20 @@ if(columnList != null && columnList.size() > 0) {
        	<%@ include file="../templates/DB/Output/splitSQLForAllDBInBegin.javajet" %>
         <%
     }
+    if(setQueryTimeOut) {
+    	if(("INSERT").equals(dataAction) || ("UPDATE").equals(dataAction) 
+			|| ("DELETE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction)) {
+%>
+			pstmt_<%=cid %>.setQueryTimeout(<%= queryTimeOutInSeconds %>);
+<%
+    	}
+    	if(("UPDATE_OR_INSERT").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction)) {
+%>
+			pstmtUpdate_<%=cid %>.setQueryTimeout(<%= queryTimeOutInSeconds %>);
+			pstmtInsert_<%=cid %>.setQueryTimeout(<%= queryTimeOutInSeconds %>);
+<%
+    	}
+	}
 }
 } // end isDynamic
 if(isDynamic) {

main/plugins/org.talend.designer.components.localprovider/components/tAS400Output/tAS400Output_java.xml

@@ -322,6 +322,23 @@
 		NUM_ROW="20">
 		  <DEFAULT>false</DEFAULT>
 	</PARAMETER>
+	
+	<PARAMETER
+      NAME="SET_QUERY_TIMEOUT"
+      FIELD="CHECK"
+      NUM_ROW="30"
+      >
+      <DEFAULT>false</DEFAULT>
+    </PARAMETER>
+
+    <PARAMETER
+      NAME="QUERY_TIMEOUT_IN_SECONDS"
+      FIELD="TEXT"
+      NUM_ROW="30"
+      SHOW_IF="SET_QUERY_TIMEOUT == 'true'"
+      >
+      <DEFAULT>30</DEFAULT>
+    </PARAMETER>
 
   </ADVANCED_PARAMETERS>
 

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

@@ -41,10 +41,13 @@ skeleton="../templates/db_output_bulk.skeleton"
     String batchSize =ElementParameterParser.getValue(node, "__BATCH_SIZE__");
     boolean useExistingConnection = "true".equals(ElementParameterParser.getValue(node,"__USE_EXISTING_CONNECTION__"));
     boolean isEnableDebug = ("true").equals(ElementParameterParser.getValue(node,"__ENABLE_DEBUG_MODE__"));
-
 	//feature:2880
     getManager(dbmsId, cid, node);
     boolean whereSupportNull = ElementParameterParser.getValue(node, "__SUPPORT_NULL_WHERE__").equals("true");
+    
+    boolean setQueryTimeOut = "true".equals(ElementParameterParser.getValue(node,"__SET_QUERY_TIMEOUT__"));
+    
+    String queryTimeOutInSeconds = ElementParameterParser.getValue(node,"__QUERY_TIMEOUT_IN_SECONDS__");
 
 	String incomingConnName = null;
 	List<IMetadataColumn> columnList = getColumnList(node);
@@ -227,6 +230,20 @@ skeleton="../templates/db_output_bulk.skeleton"
        	<%@ include file="../templates/DB/Output/splitSQLForAllDBInDynamicMain.javajet" %>
         <%
     }
+    if(setQueryTimeOut) {
+    	if(("INSERT").equals(dataAction) || ("UPDATE").equals(dataAction) 
+			|| ("DELETE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction)) {
+%>
+			pstmt_<%=cid %>.setQueryTimeout(<%= queryTimeOutInSeconds %>);
+<%
+    	}
+    	if(("UPDATE_OR_INSERT").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction)) {
+%>
+			pstmtUpdate_<%=cid %>.setQueryTimeout(<%= queryTimeOutInSeconds %>);
+			pstmtInsert_<%=cid %>.setQueryTimeout(<%= queryTimeOutInSeconds %>);
+<%
+    	}
+	}
 }
 %>
 	}
@@ -313,6 +330,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 +463,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 +612,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 +689,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 +870,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 +956,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/tAS400Output/tAS400Output_messages.properties

@@ -75,4 +75,7 @@ SUPPORT_NULL_WHERE.NAME=Support null in "SQL WHERE" statement
 CONNECTION.NAME=Component List
 USE_BATCH_SIZE.NAME=Use Batch Size
 BATCH_SIZE.NAME=Batch Size
-NB_LINE_REJECTED.NAME=NB Line Rejected
+NB_LINE_REJECTED.NAME=NB Line Rejected
+
+SET_QUERY_TIMEOUT.NAME=Set Query Timeout
+QUERY_TIMEOUT_IN_SECONDS.NAME=Timeout (seconds)

main/plugins/org.talend.designer.components.localprovider/components/tAS400Row/tAS400Row_java.xml

@@ -251,6 +251,23 @@
 		REQUIRED="true" SHOW_IF="USE_EXISTING_CONNECTION == 'false'">
 		<DEFAULT>10000</DEFAULT>
 	</PARAMETER>
+	
+	<PARAMETER
+      NAME="SET_QUERY_TIMEOUT"
+      FIELD="CHECK"
+      NUM_ROW="10"
+      >
+      <DEFAULT>false</DEFAULT>
+    </PARAMETER>
+
+    <PARAMETER
+      NAME="QUERY_TIMEOUT_IN_SECONDS"
+      FIELD="TEXT"
+      NUM_ROW="10"
+      SHOW_IF="SET_QUERY_TIMEOUT == 'true'"
+      >
+      <DEFAULT>30</DEFAULT>
+    </PARAMETER>
   </ADVANCED_PARAMETERS>
 
    <CODEGENERATION>

main/plugins/org.talend.designer.components.localprovider/components/tAS400Row/tAS400Row_messages.properties

@@ -52,4 +52,7 @@ 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
+SET_PREPAREDSTATEMENT_PARAMETERS.ITEM.PARAMETER_VALUE=Parameter Value
+
+SET_QUERY_TIMEOUT.NAME=Set Query Timeout
+QUERY_TIMEOUT_IN_SECONDS.NAME=Timeout (seconds)

main/plugins/org.talend.designer.components.localprovider/components/tAccessConnection/tAccessConnection_java.xml

@@ -118,7 +118,7 @@
 		<IMPORT NAME="Driver-JACKCESS" MODULE="jackcess-2.1.12.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess/2.1.12" REQUIRED="true"/>
 		
 		<IMPORT NAME="jackcess-encrypt-2.1.4" MODULE="jackcess-encrypt-2.1.4.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess-encrypt/2.1.4" REQUIRED="true"/>
-		<IMPORT NAME="bcprov-jdk15on-1.68" MODULE="bcprov-jdk15on-1.68.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.68" REQUIRED="true"/>
+		<IMPORT NAME="bcprov-jdk15on-1.69" MODULE="bcprov-jdk15on-1.69.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.69" REQUIRED="true"/>
 		<IMPORT NAME="talend-ucanaccess-utils-1.0.0" MODULE="talend-ucanaccess-utils-1.0.0.jar" MVN="mvn:org.talend.libraries/talend-ucanaccess-utils-1.0.0/6.4.0"  UrlPath="platform:/plugin/org.talend.libraries.jdbc.access/lib/talend-ucanaccess-utils-1.0.0.jar" REQUIRED="true"/>
 		
 		<IMPORT NAME="common2.6" MODULE="commons-lang-2.6.jar" MVN="mvn:commons-lang/commons-lang/2.6"   UrlPath="platform:/plugin/org.talend.libraries.apache.common/lib/commons-lang-2.6.jar" REQUIRED="true"/>

main/plugins/org.talend.designer.components.localprovider/components/tAccessInput/tAccessInput_java.xml

@@ -121,7 +121,7 @@
 		<IMPORT NAME="Driver-JACKCESS" MODULE="jackcess-2.1.12.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess/2.1.12" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		
 		<IMPORT NAME="jackcess-encrypt-2.1.4" MODULE="jackcess-encrypt-2.1.4.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess-encrypt/2.1.4" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
-		<IMPORT NAME="bcprov-jdk15on-1.68" MODULE="bcprov-jdk15on-1.68.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.68" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
+		<IMPORT NAME="bcprov-jdk15on-1.69" MODULE="bcprov-jdk15on-1.69.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.69" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		<IMPORT NAME="talend-ucanaccess-utils-1.0.0" MODULE="talend-ucanaccess-utils-1.0.0.jar" MVN="mvn:org.talend.libraries/talend-ucanaccess-utils-1.0.0/6.4.0"  UrlPath="platform:/plugin/org.talend.libraries.jdbc.access/lib/talend-ucanaccess-utils-1.0.0.jar" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		
 		<IMPORT NAME="common2.6" MODULE="commons-lang-2.6.jar" MVN="mvn:commons-lang/commons-lang/2.6"   UrlPath="platform:/plugin/org.talend.libraries.apache.common/lib/commons-lang-2.6.jar" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>

main/plugins/org.talend.designer.components.localprovider/components/tAccessOutput/tAccessOutput_java.xml

@@ -207,7 +207,7 @@
 		<IMPORT NAME="Driver-JACKCESS" MODULE="jackcess-2.1.12.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess/2.1.12" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		
 		<IMPORT NAME="jackcess-encrypt-2.1.4" MODULE="jackcess-encrypt-2.1.4.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess-encrypt/2.1.4" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
-		<IMPORT NAME="bcprov-jdk15on-1.68" MODULE="bcprov-jdk15on-1.68.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.68" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
+		<IMPORT NAME="bcprov-jdk15on-1.69" MODULE="bcprov-jdk15on-1.69.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.69" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		<IMPORT NAME="talend-ucanaccess-utils-1.0.0" MODULE="talend-ucanaccess-utils-1.0.0.jar" MVN="mvn:org.talend.libraries/talend-ucanaccess-utils-1.0.0/6.4.0"  UrlPath="platform:/plugin/org.talend.libraries.jdbc.access/lib/talend-ucanaccess-utils-1.0.0.jar" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		
 		<IMPORT NAME="common2.6" MODULE="commons-lang-2.6.jar" MVN="mvn:commons-lang/commons-lang/2.6"   UrlPath="platform:/plugin/org.talend.libraries.apache.common/lib/commons-lang-2.6.jar" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>

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/tAccessRow/tAccessRow_java.xml

@@ -181,7 +181,7 @@
 		<IMPORT NAME="Driver-JACKCESS" MODULE="jackcess-2.1.12.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess/2.1.12" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		
 		<IMPORT NAME="jackcess-encrypt-2.1.4" MODULE="jackcess-encrypt-2.1.4.jar" MVN="mvn:com.healthmarketscience.jackcess/jackcess-encrypt/2.1.4" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
-		<IMPORT NAME="bcprov-jdk15on-1.68" MODULE="bcprov-jdk15on-1.68.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.68" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
+		<IMPORT NAME="bcprov-jdk15on-1.69" MODULE="bcprov-jdk15on-1.69.jar" MVN="mvn:org.bouncycastle/bcprov-jdk15on/1.69" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		<IMPORT NAME="talend-ucanaccess-utils-1.0.0" MODULE="talend-ucanaccess-utils-1.0.0.jar" MVN="mvn:org.talend.libraries/talend-ucanaccess-utils-1.0.0/6.4.0"  UrlPath="platform:/plugin/org.talend.libraries.jdbc.access/lib/talend-ucanaccess-utils-1.0.0.jar" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>
 		
 		<IMPORT NAME="common2.6" MODULE="commons-lang-2.6.jar" MVN="mvn:commons-lang/commons-lang/2.6"   UrlPath="platform:/plugin/org.talend.libraries.apache.common/lib/commons-lang-2.6.jar" REQUIRED_IF="(USE_EXISTING_CONNECTION == 'false')"/>

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

@@ -388,9 +388,12 @@ class GenerateToolByDom4j{
 			for(XMLNode attri:node.attributes){
 				addAttribute(currEleName,attri);
 			}
-			if(node.name.indexOf(":")>0){
+			int idx = node.name.indexOf(":");
+			if(idx > 0){
+				String prefix = node.name.substring(0, idx);
+				String localName = node.name.substring(idx + 1);
 %>
-			<%touchXMLNode.getXMLNode(currEleName);%>.setName("<%=node.name%>");
+			<%touchXMLNode.getXMLElement(currEleName);%>.setQName(org.dom4j.DocumentHelper.createQName("<%=localName%>",<%touchXMLNode.getXMLElement(currEleName);%>.getNamespaceForPrefix("<%=prefix%>")));
 <%
 			}
 			int index = 0;

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);
 					}
 				}
@@ -173,7 +174,7 @@ if ((metadatas != null) && (metadatas.size() > 0)) {
 					if (nestRoot != null) {
 						boolean isDefaultNameSpaceAtRoot = "".equals(nestRoot.getQName().getNamespace().getPrefix());
 						for (org.dom4j.Element tmp: (java.util.List<org.dom4j.Element>) nestRoot.elements()) {
-							if (("").equals(tmp.getQName().getNamespace().getURI()) && ("").equals(tmp.getQName().getNamespace().getPrefix()) && isDefaultNameSpaceAtRoot) {
+							if (isDefaultNameSpaceAtRoot && ("").equals(tmp.getQName().getNamespace().getURI()) && ("").equals(tmp.getQName().getNamespace().getPrefix())) {
 								tmp.setQName(org.dom4j.DocumentHelper.createQName(tmp.getName(), nestRoot.getQName().getNamespace()));
 							}
 							replaceDefaultNameSpace(tmp);
@@ -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

@@ -243,8 +243,8 @@
 			<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

@@ -570,8 +570,8 @@
 			<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

@@ -300,8 +300,8 @@
 			<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

@@ -157,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

@@ -455,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

@@ -360,8 +360,8 @@
 			<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"