[SYSTEMML-995] Determine Frame Format If Needed

If `MLContextConversionUtil.dataFrameToFrameObject` receives a
DataFrame *without* any frame metadata, a new `FrameMetadata` will be
created with an empty `FrameFormat`, and so the subsequent
`isDataFrameWithIDColumn` function will always return false
([line 360](https://github.com/apache/incubator-systemml/blob/master/src/main/java/org/apache/sysml/api/mlcontext/MLContextConversionUtil.java#L360).
We should just create a new function similar to
`determineMatrixFormatIfNeeded` for frames, and call it before the
`isDataFrameWithIDColumn` function, as is done for DataFrame-matrix
conversions ([line 412](https://github.com/apache/incubator-systemml/blob/master/src/main/java/org/apache/sysml/api/mlcontext/MLContextConversionUtil.java#L412).

Author: dusenberrymw

src/main/java/org/apache/sysml/api/mlcontext/MLContextConversionUtil.java

@@ -357,6 +357,7 @@ public static FrameObject dataFrameToFrameObject(String variableName, DataFrame
 			//setup meta data and java spark context
 			if (frameMetadata == null)
 				frameMetadata = new FrameMetadata();
+			determineFrameFormatIfNeeded(dataFrame, frameMetadata);
 			boolean containsID = isDataFrameWithIDColumn(frameMetadata);
 			JavaSparkContext javaSparkContext = MLContextUtil
 					.getJavaSparkContext((MLContext) MLContextProxy.getActiveMLContextForAPI());
@@ -490,6 +491,33 @@ public static void determineMatrixFormatIfNeeded(DataFrame dataFrame, MatrixMeta
 		matrixMetadata.setMatrixFormat(mf);
 	}
 
+	/**
+	 * If the FrameFormat of the DataFrame has not been explicitly specified,
+	 * attempt to determine the proper FrameFormat.
+	 *
+	 * @param dataFrame
+	 *            the Spark {@code DataFrame}
+	 * @param frameMetadata
+	 *            the frame metadata, if available
+	 */
+	public static void determineFrameFormatIfNeeded(DataFrame dataFrame, FrameMetadata frameMetadata) {
+		FrameFormat frameFormat = frameMetadata.getFrameFormat();
+		if (frameFormat != null) {
+			return;
+		}
+
+		StructType schema = dataFrame.schema();
+		boolean hasID = false;
+		try {
+			schema.fieldIndex(RDDConverterUtils.DF_ID_COLUMN);
+			hasID = true;
+		} catch (IllegalArgumentException iae) {
+		}
+
+		FrameFormat ff = hasID ? FrameFormat.DF_WITH_INDEX : FrameFormat.DF;
+		frameMetadata.setFrameFormat(ff);
+	}
+
 	/**
 	 * Return whether or not the DataFrame has an ID column.
 	 * 

src/test/java/org/apache/sysml/test/integration/functions/mlcontext/DataFrameVectorScriptTest.java

@@ -271,24 +271,33 @@ private void testDataFrameScriptInput(ValueType[] schema, boolean containsID, bo
 
 			//create input data frame
 			DataFrame df = createDataFrame(sqlctx, mbA, containsID, schema);
-			FrameMetadata meta = new FrameMetadata(containsID ? FrameFormat.DF_WITH_INDEX : 
+
+			// Create full frame metadata, and empty frame metadata
+			FrameMetadata meta = new FrameMetadata(containsID ? FrameFormat.DF_WITH_INDEX :
 				FrameFormat.DF, mc2.getRows(), mc2.getCols());
-			
+			FrameMetadata metaEmpty = new FrameMetadata();
+
 			//create mlcontext
 			ml = new MLContext(sc);
 			ml.setExplain(true);
 
-			//run script and obtain result
+			//run scripts and obtain result
 			Script script1 = dml(
-					//"Xf = read($Xffile); Xm = as.matrix(Xf); write(Xm, $Xmfile);")
 					"Xm = as.matrix(Xf);")
 				.in("Xf", df, meta).out("Xm");
-			Matrix Xm = ml.execute(script1).getMatrix("Xm");
-			MatrixBlock mbB = Xm.toBinaryBlockMatrix().getMatrixBlock();
-			
+			Script script2 = dml(
+					"Xm = as.matrix(Xf);")
+					.in("Xf", df, metaEmpty).out("Xm");  // empty metadata
+			Matrix Xm1 = ml.execute(script1).getMatrix("Xm");
+			Matrix Xm2 = ml.execute(script2).getMatrix("Xm");
+			MatrixBlock mbB1 = Xm1.toBinaryBlockMatrix().getMatrixBlock();
+			MatrixBlock mbB2 = Xm2.toBinaryBlockMatrix().getMatrixBlock();
+
 			//compare frame blocks
-			double[][] B = DataConverter.convertToDoubleMatrix(mbB);
-			TestUtils.compareMatrices(A, B, rows1, cols, eps);
+			double[][] B1 = DataConverter.convertToDoubleMatrix(mbB1);
+			double[][] B2 = DataConverter.convertToDoubleMatrix(mbB2);
+			TestUtils.compareMatrices(A, B1, rows1, cols, eps);
+			TestUtils.compareMatrices(A, B2, rows1, cols, eps);
 		}
 		catch( Exception ex ) {
 			ex.printStackTrace();
@@ -364,4 +373,4 @@ private DataFrame createDataFrame(SQLContext sqlctx, MatrixBlock mb, boolean con
 		JavaRDD<Row> rowRDD = sc.parallelize(list);
 		return sqlctx.createDataFrame(rowRDD, dfSchema);
 	}
-}
+}