[MINOR] off-by-one error in CSV header writing

src/main/java/org/apache/sysds/runtime/instructions/spark/utils/FrameRDDConverterUtils.java

@@ -663,9 +663,9 @@ public Iterator<String> call(Tuple2<Long, FrameBlock> arg0)
 			//handle header information and frame meta data
 			if( ix==1 ) {
 				if( _props.hasHeader() ) {
-					for(int j = 1; j <= blk.getNumColumns(); j++) {
-						sb.append(blk.getColumnNames()[j] 
-							+ ((j<blk.getNumColumns()-1)?_props.getDelim():""));
+					for(int j = 0; j < blk.getNumColumns(); j++) {
+						sb.append(blk.getColumnNames()[j])
+								.append(j < blk.getNumColumns() - 1 ? _props.getDelim() : "");
 					}
 					ret.add(sb.toString());
 					sb.setLength(0); //reset

src/test/java/org/apache/sysds/test/functions/io/csv/FrameCSVHeaderNamesTest.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.test.functions.io.csv;
+
+import java.util.Collections;
+
+import org.apache.sysds.api.DMLScript;
+import org.apache.sysds.common.Types;
+import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.common.Types.FileFormat;
+import org.apache.sysds.runtime.frame.data.FrameBlock;
+import org.apache.sysds.runtime.io.FileFormatPropertiesCSV;
+import org.apache.sysds.runtime.io.FrameReader;
+import org.apache.sysds.runtime.io.FrameReaderFactory;
+import org.apache.sysds.runtime.io.FrameWriter;
+import org.apache.sysds.runtime.io.FrameWriterFactory;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestConfiguration;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class FrameCSVHeaderNamesTest extends AutomatedTestBase {
+	private static final String TEST_NAME = "FrameCSVHeaderNames";
+	private static final String TEST_DIR = "functions/frame/";
+	private static final String TEST_CLASS_DIR =
+		TEST_DIR + FrameCSVHeaderNamesTest.class.getSimpleName() + "/";
+
+	private static final int ROWS = 42;
+	private static final String[] COLUMN_NAMES = new String[] {"customer_id", "signup_date", "score"};
+
+	@Override
+	public void setUp() {
+		addTestConfiguration(TEST_NAME, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME, new String[] {"B"}));
+	}
+
+	@Test
+	public void testFrameCSVHeaderNamesCP() {
+		runCSVHeaderNamesTest(ExecType.CP);
+	}
+
+	@Test
+	public void testFrameCSVHeaderNamesSpark() {
+		runCSVHeaderNamesTest(ExecType.SPARK);
+	}
+
+	private void runCSVHeaderNamesTest(ExecType et) {
+		Types.ExecMode platformOld = setExecMode(et);
+		boolean sparkConfigOld = DMLScript.USE_LOCAL_SPARK_CONFIG;
+		setOutputBuffering(true);
+		try {
+			getAndLoadTestConfiguration(TEST_NAME);
+			String HOME = SCRIPT_DIR + TEST_DIR;
+			fullDMLScriptName = HOME + TEST_NAME + ".dml";
+			programArgs = new String[] {
+				"-args", input("A"), String.valueOf(ROWS), String.valueOf(COLUMN_NAMES.length), output("B")
+			};
+
+			Types.ValueType[] schema = Collections.nCopies(
+				COLUMN_NAMES.length, Types.ValueType.FP64).toArray(new Types.ValueType[0]);
+			FrameBlock inputFrame = new FrameBlock(schema);
+			inputFrame.setColumnNames(COLUMN_NAMES);
+
+			double[][] data = getRandomMatrix(ROWS, schema.length, -10, 10, 0.8, 7);
+			TestUtils.initFrameData(inputFrame, data, schema, ROWS);
+
+			FrameWriter writer = FrameWriterFactory.createFrameWriter(
+				FileFormat.CSV, new FileFormatPropertiesCSV(true, ",", false));
+			writer.writeFrameToHDFS(inputFrame, input("A"), ROWS, schema.length);
+
+			runTest(true, false, null, -1);
+
+			FrameReader reader = FrameReaderFactory.createFrameReader(
+				FileFormat.CSV, new FileFormatPropertiesCSV(true, ",", false));
+			FrameBlock outputFrame = reader.readFrameFromHDFS(output("B"), ROWS, schema.length);
+			Assert.assertArrayEquals("Column names were not preserved across CSV read/write.",
+				COLUMN_NAMES, outputFrame.getColumnNames());
+		}
+		catch(Exception ex) {
+			throw new RuntimeException(ex);
+		}
+		finally {
+			rtplatform = platformOld;
+			DMLScript.USE_LOCAL_SPARK_CONFIG = sparkConfigOld;
+		}
+	}
+}

src/test/scripts/functions/frame/FrameCSVHeaderNames.dml

@@ -0,0 +1,23 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+X = read($1, rows=$2, cols=$3, data_type="frame", format="csv", header=TRUE);
+write(X, $4, format="csv", header=TRUE);