[Feat][Format][Spark] Add nullable key in meta-data

spark/graphar/src/main/scala/com/alibaba/graphar/EdgeInfo.scala

@@ -17,9 +17,10 @@
 package com.alibaba.graphar
 
 import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.{SparkSession}
-import org.yaml.snakeyaml.{Yaml, DumperOptions}
+import org.apache.spark.sql.SparkSession
+import org.yaml.snakeyaml.{DumperOptions, Yaml}
 import org.yaml.snakeyaml.constructor.Constructor
+
 import scala.beans.BeanProperty
 import org.yaml.snakeyaml.LoaderOptions
 
@@ -230,6 +231,31 @@ class EdgeInfo() {
     throw new IllegalArgumentException("Property not found: " + property_name)
   }
 
+  /**
+   * Check the property is nullable key of edge info.
+   *
+   * @param property_name
+   *   name of the property.
+   * @return
+   *   true if the property is the nullable key of edge info, false if not. If
+   *   edge info not contains the property, raise an IllegalArgumentException
+   *   error.
+   */
+  def isNullableKey(property_name: String): Boolean = {
+    val len: Int = property_groups.size
+    for (i <- 0 to len - 1) {
+      val pg: PropertyGroup = property_groups.get(i)
+      val properties = pg.getProperties
+      val num = properties.size
+      for (j <- 0 to num - 1) {
+        if (properties.get(j).getName == property_name) {
+          return properties.get(j).getIs_nullable
+        }
+      }
+    }
+    throw new IllegalArgumentException("Property not found: " + property_name)
+  }
+
   /** Get Primary key of edge info. */
   def getPrimaryKey(): String = {
     val len: Int = property_groups.size

spark/graphar/src/main/scala/com/alibaba/graphar/GraphInfo.scala

@@ -174,6 +174,7 @@ class Property() {
   @BeanProperty var name: String = ""
   @BeanProperty var data_type: String = ""
   @BeanProperty var is_primary: Boolean = false
+  private var is_nullable: Option[Boolean] = Option.empty
 
   /** Get data type in gar of the property */
   def getData_type_in_gar: GarType.Value = {
@@ -185,7 +186,8 @@ class Property() {
       case other: Property =>
         this.name == other.name &&
           this.data_type == other.data_type &&
-          this.is_primary == other.is_primary
+          this.is_primary == other.is_primary &&
+          this.getIs_nullable == other.getIs_nullable
       case _ => false
     }
   }
@@ -195,8 +197,21 @@ class Property() {
     data.put("name", name)
     data.put("data_type", data_type)
     data.put("is_primary", new java.lang.Boolean(is_primary))
+    data.put("is_nullable", new java.lang.Boolean(getIs_nullable))
     return data
   }
+
+  def getIs_nullable(): Boolean = {
+    if (is_nullable.isEmpty) {
+      !is_primary
+    } else {
+      is_nullable.get
+    }
+  }
+
+  def setIs_nullable(is_nullable: Boolean): Unit = {
+    this.is_nullable = Option(is_nullable)
+  }
 }
 
 /** PropertyGroup is a class to store the property group information. */

spark/graphar/src/main/scala/com/alibaba/graphar/VertexInfo.scala

@@ -145,6 +145,30 @@ class VertexInfo() {
     throw new IllegalArgumentException("Property not found: " + property_name)
   }
 
+  /**
+   * Check if the property is nullable key.
+   *
+   * @param property_name
+   *   name of the property to check.
+   * @return
+   *   true if the property if a nullable key of vertex info, otherwise return
+   *   false.
+   */
+  def isNullableKey(property_name: String): Boolean = {
+    val len: Int = property_groups.size
+    for (i <- 0 to len - 1) {
+      val pg: PropertyGroup = property_groups.get(i)
+      val properties = pg.getProperties
+      val num = properties.size
+      for (j <- 0 to num - 1) {
+        if (properties.get(j).getName == property_name) {
+          return properties.get(j).getIs_nullable
+        }
+      }
+    }
+    throw new IllegalArgumentException("Property not found: " + property_name)
+  }
+
   /**
    * Get primary key of vertex info.
    *

spark/graphar/src/test/scala/com/alibaba/graphar/TestGraphInfo.scala

@@ -79,6 +79,7 @@ class GraphInfoSuite extends AnyFunSuite {
     assert(vertex_info.containPropertyGroup(property_group))
     assert(vertex_info.getPropertyType("id") == GarType.INT64)
     assert(vertex_info.isPrimaryKey("id"))
+    assert(vertex_info.isNullableKey("id") == false)
     assert(
       vertex_info.getFilePath(property_group, 0) == "vertex/person/id/chunk0"
     )
@@ -95,6 +96,9 @@ class GraphInfoSuite extends AnyFunSuite {
     assert(vertex_info.containPropertyGroup(property_group_2))
     assert(vertex_info.getPropertyType("firstName") == GarType.STRING)
     assert(vertex_info.isPrimaryKey("firstName") == false)
+    assert(vertex_info.isNullableKey("id") == false)
+    assert(vertex_info.isNullableKey("firstName"))
+    assert(vertex_info.isNullableKey("gender"))
     assert(
       vertex_info.getFilePath(
         property_group_2,
@@ -118,10 +122,13 @@ class GraphInfoSuite extends AnyFunSuite {
       vertex_info.getPropertyGroup("not_exist")
     )
     assertThrows[IllegalArgumentException](
-      vertex_info.getPropertyType("now_exist")
+      vertex_info.getPropertyType("not_exist")
     )
     assertThrows[IllegalArgumentException](
-      vertex_info.isPrimaryKey("now_exist")
+      vertex_info.isPrimaryKey("not_exist")
+    )
+    assertThrows[IllegalArgumentException](
+      vertex_info.isNullableKey("not_exist")
     )
   }
 
@@ -239,6 +246,7 @@ class GraphInfoSuite extends AnyFunSuite {
       edge_info.getPropertyType(property_name) == property.getData_type_in_gar
     )
     assert(edge_info.isPrimaryKey(property_name) == property.getIs_primary)
+    assert(edge_info.isNullableKey(property_name) == property.getIs_nullable)
     assert(
       edge_info.getPropertyFilePath(
         property_group,
@@ -341,6 +349,9 @@ class GraphInfoSuite extends AnyFunSuite {
       ) == property_2.getData_type_in_gar
     )
     assert(edge_info.isPrimaryKey(property_name_2) == property_2.getIs_primary)
+    assert(
+      edge_info.isNullableKey(property_name_2) == property_2.getIs_nullable
+    )
     assert(
       edge_info.getPropertyFilePath(
         property_group_2,
@@ -402,6 +413,7 @@ class GraphInfoSuite extends AnyFunSuite {
       edge_info.getPropertyType("not_exist")
     )
     assertThrows[IllegalArgumentException](edge_info.isPrimaryKey("not_exist"))
+    assertThrows[IllegalArgumentException](edge_info.isNullableKey("not_exist"))
   }
 
   test("== of Property/PropertyGroup/AdjList") {