[BugFix] Fix the Spark Writer bug when the column name contains a dot(.) (#101)

Author: lixueclaire

spark/src/main/scala/com/alibaba/graphar/utils/IndexGenerator.scala

@@ -118,7 +118,7 @@ object IndexGenerator {
     val srcCol = GeneralParams.srcIndexCol;
     val indexCol = GeneralParams.vertexIndexCol;
     val srcPrimaryKey = GeneralParams.primaryCol;
-    val trans_df = spark.sql(f"select src_vertex.$indexCol%s as $srcCol%s, edge.* from edge inner join src_vertex on src_vertex.$srcPrimaryKey%s=edge.$srcColumnName%s")
+    val trans_df = spark.sql(f"select src_vertex.`$indexCol` as `$srcCol`, edge.* from edge inner join src_vertex on src_vertex.`$srcPrimaryKey`=edge.`$srcColumnName`")
     // drop the old src id col
     trans_df.drop(srcColumnName)
 	}
@@ -131,7 +131,7 @@ object IndexGenerator {
     val dstCol = GeneralParams.dstIndexCol;
     val indexCol = GeneralParams.vertexIndexCol;
     val dstPrimaryKey = GeneralParams.primaryCol;
-    val trans_df = spark.sql(f"select dst_vertex.$indexCol%s as $dstCol%s, edges.* from edges inner join dst_vertex on dst_vertex.$dstPrimaryKey%s=edges.$dstColumnName%s")
+    val trans_df = spark.sql(f"select dst_vertex.`$indexCol` as `$dstCol`, edges.* from edges inner join dst_vertex on dst_vertex.`$dstPrimaryKey`=edges.`$dstColumnName`")
     // drop the old dst id col
     trans_df.drop(dstColumnName)
 	}

spark/src/main/scala/com/alibaba/graphar/writer/EdgeWriter.scala

@@ -211,7 +211,7 @@ class EdgeWriter(prefix: String,  edgeInfo: EdgeInfo, adjListType: AdjListType.V
     val p_it = propertyGroup.getProperties().iterator
     while (p_it.hasNext()) {
       val property = p_it.next()
-      property_list += property.getName()
+      property_list += "`" + property.getName() + "`"
     }
     var chunk_index: Long = 0
     for (chunk <- chunks) {

spark/src/main/scala/com/alibaba/graphar/writer/VertexWriter.scala

@@ -82,7 +82,7 @@ class VertexWriter(prefix: String, vertexInfo: VertexInfo, vertexDf: DataFrame)
     val it = propertyGroup.getProperties().iterator
     while (it.hasNext()) {
       val property = it.next()
-      property_list += property.getName()
+      property_list += "`" + property.getName() + "`"
     }
     val pg_df = chunks.select(property_list.map(col): _*)
     FileSystem.writeDataFrame(pg_df, propertyGroup.getFile_type(), output_prefix)