Gpdb header support

extensions/spring-xd-extension-gpfdist/src/main/java/org/springframework/xd/greenplum/config/GPFDistSinkOptionsMetadata.java

@@ -68,6 +68,9 @@ public class GPFDistSinkOptionsMetadata {
 
 	private String sqlAfter;
 
+	private boolean header = false;
+
+
 	public int getPort() {
 		return port;
 	}
@@ -257,4 +260,12 @@ public void setSqlAfter(String sqlAfter) {
 		this.sqlAfter = sqlAfter;
 	}
 
+    @ModuleOption("header ")
+    public void setHeader(boolean header){
+        this.header = header;
+    }
+
+    public boolean isHeader(){
+        return this.header;
+    }
 }

extensions/spring-xd-extension-gpfdist/src/main/java/org/springframework/xd/greenplum/support/ReadableTableFactoryBean.java

@@ -16,12 +16,12 @@
 
 package org.springframework.xd.greenplum.support;
 
-import java.util.Arrays;
-import java.util.List;
-
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.InitializingBean;
 
+import java.util.Arrays;
+import java.util.List;
+
 /**
  * @since 1.2
  * @author Janne Valkealahti
@@ -57,6 +57,8 @@ public class ReadableTableFactoryBean implements FactoryBean<ReadableTable>, Ini
 
 	private SegmentRejectType segmentRejectType;
 
+	private boolean header;
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		if (controlFile != null) {
@@ -75,6 +77,7 @@ public ReadableTable getObject() throws Exception {
 		w.setLogErrorsInto(logErrorsInto);
 		w.setSegmentRejectLimit(segmentRejectLimit);
 		w.setSegmentRejectType(segmentRejectType);
+        w.setFormatHeader(header);
 
 		if (format == Format.TEXT) {
 			Character delim = delimiter != null ? delimiter : Character.valueOf('\t');
@@ -206,4 +209,12 @@ public void setFormat(Format format) {
 		this.format = format;
 	}
 
+    public boolean isHeader() {
+        return header;
+    }
+
+    public void setHeader(boolean header) {
+        this.header = header;
+    }
+
 }

extensions/spring-xd-extension-gpfdist/src/main/java/org/springframework/xd/greenplum/support/SqlUtils.java

@@ -16,10 +16,10 @@
 
 package org.springframework.xd.greenplum.support;
 
-import java.util.List;
-
 import org.springframework.util.StringUtils;
 
+import java.util.List;
+
 public abstract class SqlUtils {
 
 	public static String createExternalReadableTable(LoadConfiguration config, String prefix,
@@ -101,6 +101,10 @@ else if (StringUtils.hasText(externalTable.getColumns())) {
 			buf.append(StringUtils.arrayToCommaDelimitedString(externalTable.getForceQuote()));
 		}
 
+		if(externalTable.isFormatHeader()){
+			buf.append(" HEADER ");
+		}
+
 		buf.append(" )");
 
 		if (externalTable.getEncoding() != null) {

extensions/spring-xd-extension-gpfdist/src/test/java/org/springframework/xd/greenplum/support/LoadReadableTableFactoryBeanTest.java

@@ -0,0 +1,23 @@
+package org.springframework.xd.greenplum.support;
+
+import org.junit.Test;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Created by cq on 4/4/16.
+ */
+public class LoadReadableTableFactoryBeanTest {
+
+    @Test
+    public void testLoadHeaderConfiguration() throws Exception {
+        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+                "org/springframework/xd/greenplum/support/LoadReadableTableFactoryBeanTest.xml");
+        ReadableTableFactoryBean factoryBean = context.getBean("&greenplumReadableTable", ReadableTableFactoryBean.class);
+        assertThat(factoryBean.isHeader(),is(true));
+        context.close();
+
+    }
+}

extensions/spring-xd-extension-gpfdist/src/test/resources/org/springframework/xd/greenplum/support/LoadReadableTableFactoryBeanTest.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+    <bean id="greenplumReadableTable" class="org.springframework.xd.greenplum.support.ReadableTableFactoryBean">
+        <property name="controlFile" ref="greenplumControlFile" />
+        <property name="delimiter" value="|" />
+        <property name="header" value="true" />
+        <property name="locations" value="#{T(org.springframework.xd.greenplum.support.NetworkUtils).getGPFDistUri(5000)}" />
+    </bean>
+
+    <bean id="greenplumControlFile" class="org.springframework.xd.greenplum.support.ControlFileFactoryBean">
+        <property name="controlFileResource" value="" />
+    </bean>
+
+</beans>

modules/sink/gpfdist/config/gpfdist.xml

@@ -60,6 +60,7 @@
     <beans:bean id="greenplumReadableTable" class="org.springframework.xd.greenplum.support.ReadableTableFactoryBean">
         <beans:property name="controlFile" ref="greenplumControlFile" />
         <beans:property name="delimiter" value="${columnDelimiter:}" />
+        <beans:property name="header" value="${header:}" />
         <beans:property name="locations" value="#{T(org.springframework.xd.greenplum.support.NetworkUtils).getGPFDistUri(${port})}" />
     </beans:bean>