fix overwrite parameter

vesoft-inc · Apr 18, 2023 · 8ecc822 · 8ecc822
1 parent e59ecca
commit 8ecc822
Show file tree

Hide file tree

Showing 9 changed files with 30 additions and 18 deletions.
diff --git a/nebula-spark-common/src/main/scala/com/vesoft/nebula/connector/writer/NebulaExecutor.scala b/nebula-spark-common/src/main/scala/com/vesoft/nebula/connector/writer/NebulaExecutor.scala
@@ -216,9 +216,7 @@ object NebulaExecutor {
   /**
     * construct insert statement for vertex
     */
-  def toExecuteSentence(tagName: String,
-                        vertices: NebulaVertices,
-                        overwrite: Boolean = true): String = {
+  def toExecuteSentence(tagName: String, vertices: NebulaVertices, overwrite: Boolean): String = {
     (if (overwrite) BATCH_INSERT_TEMPLATE else BATCH_INSERT_NO_OVERWRITE_TEMPLATE).format(
       DataTypeEnum.VERTEX.toString,
       tagName,
@@ -247,7 +245,7 @@ object NebulaExecutor {
   /**
     * construct insert statement for edge
     */
-  def toExecuteSentence(edgeName: String, edges: NebulaEdges, overwrite: Boolean = true): String = {
+  def toExecuteSentence(edgeName: String, edges: NebulaEdges, overwrite: Boolean): String = {
     val values = edges.values
       .map { edge =>
         val source = edges.getSourcePolicy match {

diff --git a/...-spark-connector/src/main/scala/com/vesoft/nebula/connector/writer/NebulaEdgeWriter.scala b/...-spark-connector/src/main/scala/com/vesoft/nebula/connector/writer/NebulaEdgeWriter.scala
@@ -83,7 +83,8 @@ class NebulaEdgeWriter(nebulaOptions: NebulaOptions,
   def execute(): Unit = {
     val nebulaEdges = NebulaEdges(propNames, edges.toList, srcPolicy, dstPolicy)
     val exec = nebulaOptions.writeMode match {
-      case WriteMode.INSERT => NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaEdges)
+      case WriteMode.INSERT =>
+        NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaEdges, nebulaOptions.overwrite)
       case WriteMode.UPDATE =>
         NebulaExecutor.toUpdateExecuteStatement(nebulaOptions.label, nebulaEdges)
       case WriteMode.DELETE =>

diff --git a/...park-connector/src/main/scala/com/vesoft/nebula/connector/writer/NebulaVertexWriter.scala b/...park-connector/src/main/scala/com/vesoft/nebula/connector/writer/NebulaVertexWriter.scala
@@ -67,7 +67,10 @@ class NebulaVertexWriter(nebulaOptions: NebulaOptions, vertexIndex: Int, schema:
   def execute(): Unit = {
     val nebulaVertices = NebulaVertices(propNames, vertices.toList, policy)
     val exec = nebulaOptions.writeMode match {
-      case WriteMode.INSERT => NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaVertices)
+      case WriteMode.INSERT =>
+        NebulaExecutor.toExecuteSentence(nebulaOptions.label,
+                                         nebulaVertices,
+                                         nebulaOptions.overwrite)
       case WriteMode.UPDATE =>
         NebulaExecutor.toUpdateExecuteStatement(nebulaOptions.label, nebulaVertices)
       case WriteMode.DELETE =>

diff --git a/...ark-connector/src/test/scala/com/vesoft/nebula/connector/writer/NebulaExecutorSuite.scala b/...ark-connector/src/test/scala/com/vesoft/nebula/connector/writer/NebulaExecutorSuite.scala
@@ -140,7 +140,7 @@ class NebulaExecutorSuite extends AnyFunSuite with BeforeAndAfterAll {
     vertices.append(NebulaVertex("\"vid2\"", props2))
 
     val nebulaVertices  = NebulaVertices(propNames, vertices.toList, None)
-    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices)
+    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices, true)
 
     val expectStatement = "INSERT vertex `person`(`col_string`,`col_fixed_string`,`col_bool`," +
       "`col_int`,`col_int64`,`col_double`,`col_date`,`col_geo`) VALUES \"vid1\": (" + props1
@@ -174,7 +174,7 @@ class NebulaExecutorSuite extends AnyFunSuite with BeforeAndAfterAll {
     vertices.append(NebulaVertex("vid2", props2))
 
     val nebulaVertices  = NebulaVertices(propNames, vertices.toList, Some(KeyPolicy.HASH))
-    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices)
+    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices, true)
 
     val expectStatement = "INSERT vertex `person`(`col_string`,`col_fixed_string`,`col_bool`," +
       "`col_int`,`col_int64`,`col_double`,`col_date`,`col_geo`) VALUES hash(\"vid1\"): (" + props1
@@ -208,14 +208,15 @@ class NebulaExecutorSuite extends AnyFunSuite with BeforeAndAfterAll {
     edges.append(NebulaEdge("\"vid2\"", "\"vid1\"", Some(2L), props2))
 
     val nebulaEdges   = NebulaEdges(propNames, edges.toList, None, None)
-    val edgeStatement = NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges)
+    val edgeStatement = NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges, true)
 
     val expectStatement = "INSERT edge `friend`(`col_string`,`col_fixed_string`,`col_bool`,`col_int`" +
       ",`col_int64`,`col_double`,`col_date`,`col_geo`) VALUES \"vid1\"->\"vid2\"@1: (" +
       props1.mkString(", ") + "), \"vid2\"->\"vid1\"@2: (" + props2.mkString(", ") + ")"
     assert(expectStatement.equals(edgeStatement))
 
-    val edgeWithoutOverwriteStatement = NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges)
+    val edgeWithoutOverwriteStatement =
+      NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges, true)
     val expectWithoutOverwriteStatement = "INSERT edge IF NOT EXISTS `friend`(`col_string`," +
       "`col_fixed_string`,`col_bool`,`col_int`,`col_int64`,`col_double`,`col_date`,`col_geo`) " +
       "VALUES \"vid1\"->\"vid2\"@1: (" + props1.mkString(", ") + "), \"vid2\"->\"vid1\"@2: (" +

diff --git a/...rk-connector_2.2/src/main/scala/com/vesoft/nebula/connector/writer/NebulaEdgeWriter.scala b/...rk-connector_2.2/src/main/scala/com/vesoft/nebula/connector/writer/NebulaEdgeWriter.scala
@@ -95,7 +95,8 @@ class NebulaEdgeWriter(nebulaOptions: NebulaOptions,
   def execute(): Unit = {
     val nebulaEdges = NebulaEdges(propNames, edges.toList, srcPolicy, dstPolicy)
     val exec = nebulaOptions.writeMode match {
-      case WriteMode.INSERT => NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaEdges)
+      case WriteMode.INSERT =>
+        NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaEdges, nebulaOptions.overwrite)
       case WriteMode.UPDATE =>
         NebulaExecutor.toUpdateExecuteStatement(nebulaOptions.label, nebulaEdges)
       case WriteMode.DELETE =>

diff --git a/...-connector_2.2/src/main/scala/com/vesoft/nebula/connector/writer/NebulaVertexWriter.scala b/...-connector_2.2/src/main/scala/com/vesoft/nebula/connector/writer/NebulaVertexWriter.scala
@@ -80,7 +80,10 @@ class NebulaVertexWriter(nebulaOptions: NebulaOptions, vertexIndex: Int, schema:
   private def execute(): Unit = {
     val nebulaVertices = NebulaVertices(propNames, vertices.toList, policy)
     val exec = nebulaOptions.writeMode match {
-      case WriteMode.INSERT => NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaVertices)
+      case WriteMode.INSERT =>
+        NebulaExecutor.toExecuteSentence(nebulaOptions.label,
+                                         nebulaVertices,
+                                         nebulaOptions.overwrite)
       case WriteMode.UPDATE =>
         NebulaExecutor.toUpdateExecuteStatement(nebulaOptions.label, nebulaVertices)
       case WriteMode.DELETE =>

diff --git a/...rk-connector_3.0/src/main/scala/com/vesoft/nebula/connector/writer/NebulaEdgeWriter.scala b/...rk-connector_3.0/src/main/scala/com/vesoft/nebula/connector/writer/NebulaEdgeWriter.scala
@@ -83,7 +83,8 @@ class NebulaEdgeWriter(nebulaOptions: NebulaOptions,
   def execute(): Unit = {
     val nebulaEdges = NebulaEdges(propNames, edges.toList, srcPolicy, dstPolicy)
     val exec = nebulaOptions.writeMode match {
-      case WriteMode.INSERT => NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaEdges)
+      case WriteMode.INSERT =>
+        NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaEdges, nebulaOptions.overwrite)
       case WriteMode.UPDATE =>
         NebulaExecutor.toUpdateExecuteStatement(nebulaOptions.label, nebulaEdges)
       case WriteMode.DELETE =>

diff --git a/...-connector_3.0/src/main/scala/com/vesoft/nebula/connector/writer/NebulaVertexWriter.scala b/...-connector_3.0/src/main/scala/com/vesoft/nebula/connector/writer/NebulaVertexWriter.scala
@@ -67,7 +67,10 @@ class NebulaVertexWriter(nebulaOptions: NebulaOptions, vertexIndex: Int, schema:
   def execute(): Unit = {
     val nebulaVertices = NebulaVertices(propNames, vertices.toList, policy)
     val exec = nebulaOptions.writeMode match {
-      case WriteMode.INSERT => NebulaExecutor.toExecuteSentence(nebulaOptions.label, nebulaVertices)
+      case WriteMode.INSERT =>
+        NebulaExecutor.toExecuteSentence(nebulaOptions.label,
+                                         nebulaVertices,
+                                         nebulaOptions.overwrite)
       case WriteMode.UPDATE =>
         NebulaExecutor.toUpdateExecuteStatement(nebulaOptions.label, nebulaVertices)
       case WriteMode.DELETE =>

diff --git a/...connector_3.0/src/test/scala/com/vesoft/nebula/connector/writer/NebulaExecutorSuite.scala b/...connector_3.0/src/test/scala/com/vesoft/nebula/connector/writer/NebulaExecutorSuite.scala
@@ -140,7 +140,7 @@ class NebulaExecutorSuite extends AnyFunSuite with BeforeAndAfterAll {
     vertices.append(NebulaVertex("\"vid2\"", props2))
 
     val nebulaVertices  = NebulaVertices(propNames, vertices.toList, None)
-    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices)
+    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices, true)
 
     val expectStatement = "INSERT vertex `person`(`col_string`,`col_fixed_string`,`col_bool`," +
       "`col_int`,`col_int64`,`col_double`,`col_date`,`col_geo`) VALUES \"vid1\": (" + props1
@@ -174,7 +174,7 @@ class NebulaExecutorSuite extends AnyFunSuite with BeforeAndAfterAll {
     vertices.append(NebulaVertex("vid2", props2))
 
     val nebulaVertices  = NebulaVertices(propNames, vertices.toList, Some(KeyPolicy.HASH))
-    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices)
+    val vertexStatement = NebulaExecutor.toExecuteSentence(tagName, nebulaVertices, true)
 
     val expectStatement = "INSERT vertex `person`(`col_string`,`col_fixed_string`,`col_bool`," +
       "`col_int`,`col_int64`,`col_double`,`col_date`,`col_geo`) VALUES hash(\"vid1\"): (" + props1
@@ -208,14 +208,15 @@ class NebulaExecutorSuite extends AnyFunSuite with BeforeAndAfterAll {
     edges.append(NebulaEdge("\"vid2\"", "\"vid1\"", Some(2L), props2))
 
     val nebulaEdges   = NebulaEdges(propNames, edges.toList, None, None)
-    val edgeStatement = NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges)
+    val edgeStatement = NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges, true)
 
     val expectStatement = "INSERT edge `friend`(`col_string`,`col_fixed_string`,`col_bool`,`col_int`" +
       ",`col_int64`,`col_double`,`col_date`,`col_geo`) VALUES \"vid1\"->\"vid2\"@1: (" +
       props1.mkString(", ") + "), \"vid2\"->\"vid1\"@2: (" + props2.mkString(", ") + ")"
     assert(expectStatement.equals(edgeStatement))
 
-    val edgeWithoutOverwriteStatement = NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges)
+    val edgeWithoutOverwriteStatement =
+      NebulaExecutor.toExecuteSentence(edgeName, nebulaEdges, true)
     val expectWithoutOverwriteStatement = "INSERT edge IF NOT EXISTS `friend`(`col_string`," +
       "`col_fixed_string`,`col_bool`,`col_int`,`col_int64`,`col_double`,`col_date`,`col_geo`) " +
       "VALUES \"vid1\"->\"vid2\"@1: (" + props1.mkString(", ") + "), \"vid2\"->\"vid1\"@2: (" +