Exchange UT

.github/workflows/maven.yml

@@ -0,0 +1,46 @@
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+
+name: Java CI with Maven
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches:
+      - master
+      - 'v[0-9]+.*'
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up JDK 1.8
+      uses: actions/setup-java@v1
+      with:
+        java-version: 1.8
+
+    - name: Cache the Maven packages to speed up build
+      uses: actions/cache@v2
+      with:
+        path: ~/.m2/repository
+        key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
+        restore-keys: ${{ runner.os }}-maven-
+
+    - name: Install nebula-graph
+      run: |
+        mkdir tmp
+        pushd tmp
+        git clone https://github.com/vesoft-inc/nebula-docker-compose.git
+        pushd nebula-docker-compose/
+        cp ../../nebula-exchange/src/test/resources/docker-compose.yaml .
+        docker-compose up -d
+        sleep 10
+        popd
+        popd
+
+    - name: Build with Maven
+      run: mvn -B package

nebula-exchange/src/main/scala/com/vesoft/nebula/exchange/config/Configs.scala

@@ -593,8 +593,8 @@ object Configs {
         val fields: ListBuffer[String] = new ListBuffer[String]
         fields.append(config.getStringList("fields").asScala: _*)
 
-        if (config.hasPath("vertex")) {
-          fields.append(config.getString("vertex"))
+        if (config.hasPath("vertex.field")) {
+          fields.append(config.getString("vertex.field"))
         }
         if (config.hasPath("source.field")) {
           fields.append(config.getString("source.field"))

nebula-exchange/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala

@@ -162,7 +162,6 @@ class EdgeProcessor(data: DataFrame,
             val codec       = new NebulaCodecImpl()
 
             import java.nio.ByteBuffer
-            val order = ByteOrder.nativeOrder
             val srcBytes = if (vidType == VidType.INT) {
               ByteBuffer
                 .allocate(8)

nebula-exchange/src/main/scala/com/vesoft/nebula/exchange/processor/Processor.scala

@@ -39,10 +39,7 @@ trait Processor extends Serializable {
     * DataTime type: add datetime() function for attribute value.
     * eg: convert attribute value 2020-01-01T22:30:40 to datetime("2020-01-01T22:30:40")
     */
-  def extraValueForClient(row: Row,
-                          field: String,
-                          fieldTypeMap: Map[String, Int],
-                          toBytes: Boolean = false): Any = {
+  def extraValueForClient(row: Row, field: String, fieldTypeMap: Map[String, Int]): Any = {
     val index = row.schema.fieldIndex(field)
 
     if (row.isNullAt(index)) return null
@@ -54,10 +51,10 @@ trait Processor extends Serializable {
           value = ""
         }
         val result = NebulaUtils.escapeUtil(value).mkString("\"", "", "\"")
-        if (toBytes) result.getBytes else result
+        result
       }
       case PropertyType.DATE     => "date(\"" + row.get(index) + "\")"
-      case PropertyType.DATETIME => "datatime(\"" + row.get(index) + "\")"
+      case PropertyType.DATETIME => "datetime(\"" + row.get(index) + "\")"
       case PropertyType.TIME     => "time(\"" + row.get(index) + "\")"
       case PropertyType.TIMESTAMP => {
         val value = row.get(index).toString

nebula-exchange/src/main/scala/com/vesoft/nebula/exchange/utils/NebulaUtils.scala

@@ -47,7 +47,11 @@ object NebulaUtils {
   }
 
   def isNumic(str: String): Boolean = {
-    for (char <- str.toCharArray) {
+    val newStr: String = if (str.startsWith("-")) {
+      str.substring(1)
+    } else { str }
+
+    for (char <- newStr.toCharArray) {
       if (!Character.isDigit(char)) return false
     }
     true
@@ -80,16 +84,14 @@ object NebulaUtils {
   }
 
   def getPartitionId(id: String, partitionSize: Int, vidType: VidType.Value): Int = {
-    val hashValue: Int = if (vidType == VidType.INT) {
-      val value = java.lang.Long.parseUnsignedLong(id)
-      (Math.floorMod(value, partitionSize) + 1).toInt
+    val hashValue: Long = if (vidType == VidType.STRING) {
+      MurmurHash2.hash64(id.getBytes, id.length, 0xc70f6907)
     } else {
-      val hash     = MurmurHash2.hash64(id.getBytes, id.length, 0xc70f6907)
-      val value    = UnsignedLong.fromLongBits(hash)
-      val partSize = UnsignedLong.fromLongBits(partitionSize)
-      value.mod(partSize).intValue + 1
+      id.toLong
     }
-    hashValue
+    val unsignedValue = UnsignedLong.fromLongBits(hashValue)
+    val partSize      = UnsignedLong.fromLongBits(partitionSize)
+    unsignedValue.mod(partSize).intValue + 1
   }
 
   def escapePropName(nebulaFields: List[String]): List[String] = {