Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

support the vid with specific string prefix #149

Merged
merged 4 commits into from
Jul 6, 2023
Merged

support the vid with specific string prefix #149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
35cab68
check port and add test
Nicole00 Jul 4, 2023
a0b49a4
release the before test
Nicole00 Jul 4, 2023
3ff02f5
support the vid with specific string prefix
Nicole00 Jul 5, 2023
0fdb82f
fix test
Nicole00 Jul 6, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 14 additions & 8 deletions conf-template/client_import/csv_datasource.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@
nebula.fields: [nebula-field-0, nebula-field-1, nebula-field-2]
vertex: {
field: csv-field-0
# add the prefix for vertex id value, eg: original id is 12345, and the real id will be: tag1_12345
prefix:"tag1"
udf: {
separator: "_"
oldColNames: [parquet-field-0, parquet-field-1]
Expand All @@ -90,13 +92,17 @@
path: "hdfs://ip:port/path/test.csv"
fields: [csv-field-0, csv-field-1, csv-field-2]
nebula.fields: [nebula-field-0, nebula-field-1, nebula-field-2]
source: csv-field-0
target: csv-field-1
ranking: csv-field-2
separator: ","
header: true
batch: 2000
partition: 60
}
source: {
field: csv-field-0
# add the prefix for source id value, eg: original id is 12345, and the real id will be: edge1_12345
prefix:"edge1"
}
target: csv-field-1
ranking: csv-field-2
separator: ","
header: true
batch: 2000
partition: 60
}
]
}
10 changes: 9 additions & 1 deletion exchange-common/src/main/scala/com/vesoft/exchange/common/config/Configs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,7 +404,9 @@ object Configs {

// You can specified the vertex field name via the com.vesoft.exchange.common.config item `vertex`
// If you want to qualified the key policy, you can wrap them into a block.
var prefix: String = null
val vertexField = if (tagConfig.hasPath("vertex.field")) {
prefix = getOrElse(tagConfig, "vertex.prefix", null)
tagConfig.getString("vertex.field")
} else {
tagConfig.getString("vertex")
Expand Down Expand Up @@ -455,6 +457,7 @@ object Configs {
nebulaFields,
vertexField,
policyOpt,
prefix,
batch,
partition,
checkPointPath,
Expand Down Expand Up @@ -504,8 +507,10 @@ object Configs {
val sinkConfig = dataSinkConfig(sinkCategory, nebulaConfig)
LOG.info(s"Sink Config ${sourceConfig}")

var sourcePrefix: String = null
val sourceField = if (!isGeo) {
if (edgeConfig.hasPath("source.field")) {
sourcePrefix = getOrElse(edgeConfig, "source.prefix", null)
edgeConfig.getString("source.field")
} else {
edgeConfig.getString("source")
Expand All @@ -524,8 +529,9 @@ object Configs {
} else {
None
}

var targetPrefix: String = null
val targetField: String = if (edgeConfig.hasPath("target.field")) {
targetPrefix = getOrElse(edgeConfig, "target.prefix", null)
edgeConfig.getString("target.field")
} else {
edgeConfig.getString("target")
Expand Down Expand Up @@ -591,9 +597,11 @@ object Configs {
nebulaFields,
sourceField,
sourcePolicy,
sourcePrefix,
ranking,
targetField,
targetPolicy,
targetPrefix,
isGeo,
latitude,
longitude,
Expand Down
3 changes: 3 additions & 0 deletions exchange-common/src/main/scala/com/vesoft/exchange/common/config/SchemaConfigs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ case class TagConfigEntry(override val name: String,
override val nebulaFields: List[String],
vertexField: String,
vertexPolicy: Option[KeyPolicy.Value],
vertexPrefix: String,
override val batch: Int,
override val partition: Int,
override val checkPointPath: Option[String],
Expand Down Expand Up @@ -110,9 +111,11 @@ case class EdgeConfigEntry(override val name: String,
override val nebulaFields: List[String],
sourceField: String,
sourcePolicy: Option[KeyPolicy.Value],
sourcePrefix: String,
rankingField: Option[String],
targetField: String,
targetPolicy: Option[KeyPolicy.Value],
targetPrefix: String,
isGeo: Boolean,
latitude: Option[String],
longitude: Option[String],
Expand Down
12 changes: 7 additions & 5 deletions exchange-common/src/test/scala/com/vesoft/exchange/common/utils/NebulaUtilsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ class NebulaUtilsSuite {
nebulaFields,
"id",
Some(KeyPolicy.UUID),
null,
1,
1,
Some(""))
Expand Down Expand Up @@ -136,6 +137,7 @@ class NebulaUtilsSuite {
wrongNebulaFields,
"id",
Some(KeyPolicy.UUID),
null,
1,
1,
Some(""))
Expand Down Expand Up @@ -222,18 +224,18 @@ class NebulaUtilsSuite {

@Test
def getAddressFromString(): Unit = {
var addr = "127.0.0.1:9669"
var addr = "127.0.0.1:9669"
var hostAddress = NebulaUtils.getAddressFromString(addr)
assert("127.0.0.1".equals(hostAddress.getHost))
assert(hostAddress.getPort==9669)
assert(hostAddress.getPort == 9669)

addr="localhost:9669"
addr = "localhost:9669"
hostAddress = NebulaUtils.getAddressFromString(addr)
assert("localhost".equals(hostAddress.getHost))

addr = "www.baidu.com:22"
hostAddress = NebulaUtils.getAddressFromString(addr)
assert(hostAddress.getPort==22)
assert(hostAddress.getPort == 22)

addr = "[2023::2]:65535"
hostAddress = NebulaUtils.getAddressFromString(addr)
Expand All @@ -242,7 +244,7 @@ class NebulaUtilsSuite {
addr = "2023::3"
hostAddress = NebulaUtils.getAddressFromString(addr)
assert(hostAddress.getHost.equals("2023::3"))
assert(hostAddress.getPort== -1)
assert(hostAddress.getPort == -1)

// bad address
addr = "localhost:65536"
Expand Down
16 changes: 12 additions & 4 deletions ...xchange_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,9 @@ class EdgeProcessor(spark: SparkSession,
s"only int vidType can use policy, but your vidType is FIXED_STRING.your row data is $row")
false
} else true
idFlag && policyFlag

val udfFlag = isVidStringType || policy.isEmpty || (edgeConfig.sourcePrefix == null && edgeConfig.targetPrefix == null)
idFlag && policyFlag && udfFlag
}

/**
Expand All @@ -254,13 +256,15 @@ class EdgeProcessor(spark: SparkSession,
"source_field",
row,
edgeConfig.sourcePolicy,
isVidStringType)
isVidStringType,
edgeConfig.sourcePrefix)

val targetField = processField(edgeConfig.targetField,
"target_field",
row,
edgeConfig.targetPolicy,
isVidStringType)
isVidStringType,
edgeConfig.targetPrefix)

val values = for {
property <- fieldKeys if property.trim.length != 0
Expand All @@ -282,7 +286,8 @@ class EdgeProcessor(spark: SparkSession,
fieldType: String,
row: Row,
policy: Option[KeyPolicy.Value],
isVidStringType: Boolean): String = {
isVidStringType: Boolean,
prefix: String): String = {
var fieldValue = if (edgeConfig.isGeo && "source_field".equals(fieldType)) {
val lat = row.getDouble(row.schema.fieldIndex(edgeConfig.latitude.get))
val lng = row.getDouble(row.schema.fieldIndex(edgeConfig.longitude.get))
Expand All @@ -292,6 +297,9 @@ class EdgeProcessor(spark: SparkSession,
val value = row.get(index).toString.trim
if (value.equals(DEFAULT_EMPTY_VALUE)) "" else value
}
if (prefix != null) {
fieldValue = prefix + "_" + fieldValue
}
// process string type vid
if (policy.isEmpty && isVidStringType) {
fieldValue = NebulaUtils.escapeUtil(fieldValue).mkString("\"", "", "\"")
Expand Down
6 changes: 6 additions & 0 deletions ...nge_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,9 @@ class VerticesProcessor(spark: SparkSession,
printChoice(streamFlag, s"vertexId must exist and cannot be null, your row data is $row")
return false
}
if (!isVidStringType && (tagConfig.vertexPolicy.isEmpty && tagConfig.vertexPrefix != null)) {
printChoice(streamFlag, s"space vidType is int, does not support prefix for vid")
}

val vertexId = row.get(index).toString.trim
// process int type vid
Expand Down Expand Up @@ -223,6 +226,9 @@ class VerticesProcessor(spark: SparkSession,
if (vertexId.equals(DEFAULT_EMPTY_VALUE)) {
vertexId = ""
}
if (tagConfig.vertexPrefix != null) {
vertexId = tagConfig.vertexPrefix + "_" + vertexId
}

if (tagConfig.vertexPolicy.isEmpty && isVidStringType) {
vertexId = NebulaUtils.escapeUtil(vertexId).mkString("\"", "", "\"")
Expand Down
26 changes: 26 additions & 0 deletions ...ge_spark_2.2/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,9 +84,11 @@ class EdgeProcessorSuite {
nebulaKeys,
"src",
None,
null,
None,
"dst",
None,
null,
false,
None,
None,
Expand Down Expand Up @@ -115,9 +117,11 @@ class EdgeProcessorSuite {
nebulaKeys,
"src",
Some(KeyPolicy.HASH),
null,
None,
"dst",
Some(KeyPolicy.HASH),
null,
false,
None,
None,
Expand Down Expand Up @@ -145,6 +149,28 @@ class EdgeProcessorSuite {
assert(edge.destination.equals("\"2\""))
assert(edge.toString.equals(
"Edge: \"1\"->\"2\"@0 values: \"\", \"fixedBob\", 12, 200, 1000, 100000, date(\"2021-01-01\"), datetime(\"2021-01-01T12:00:00.100\"), time(\"12:00:00.100\"), 345436232, true, 12.01, 22.12, ST_GeogFromText(\"POINT(3 8)\")"))

val edgeConfigEntryWithPrefix = EdgeConfigEntry("friend",
null,
null,
fieldKeys,
nebulaKeys,
"src",
None,
"src",
None,
"dst",
None,
"dst",
false,
None,
None,
10,
10,
None)
val edgeWithPrefix = processClazz.convertToEdge(row, edgeConfigEntryWithPrefix, true, fieldKeys, map)
assert(edgeWithPrefix.source.equals("\"src_1\""))
assert(edgeWithPrefix.destination.equals("\"dst_2\""))
}

@Test
Expand Down
37 changes: 35 additions & 2 deletions ...park_2.2/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ class VerticesProcessorSuite {
val stringIdRow = new GenericRowWithSchema(stringIdValue.toArray, schema)
val intIdRow = new GenericRowWithSchema(intIdValue.toArray, schema)
val tagConfigEntry =
TagConfigEntry("person", null, null, List(), List(), "id", None, 10, 10, None)
TagConfigEntry("person", null, null, List(), List(), "id", None, null, 10, 10, None)

// test for string id value without policy
assert(processClazz.isVertexValid(stringIdRow, tagConfigEntry, false, true))
Expand All @@ -96,7 +96,17 @@ class VerticesProcessorSuite {

// test for string id value with policy
val tagConfigEntryWithPolicy =
TagConfigEntry("person", null, null, List(), List(), "id", Some(KeyPolicy.HASH), 10, 10, None)
TagConfigEntry("person",
null,
null,
List(),
List(),
"id",
Some(KeyPolicy.HASH),
null,
10,
10,
None)
assert(!processClazz.isVertexValid(stringIdRow, tagConfigEntryWithPolicy, true, true))
assertThrows[AssertionError](
processClazz.isVertexValid(stringIdRow, tagConfigEntryWithPolicy, false, true))
Expand All @@ -107,6 +117,23 @@ class VerticesProcessorSuite {
assert(processClazz.isVertexValid(stringIdRow, tagConfigEntryWithPolicy, false, false))
assertThrows[AssertionError](
processClazz.isVertexValid(stringIdRow, tagConfigEntryWithPolicy, false, true))

// test for prefix for id value
val tagConfigEntryWithPrefix =
TagConfigEntry("person",
null,
null,
List(),
List(),
"id",
Some(KeyPolicy.HASH),
"prefix",
10,
10,
None)
assert(processClazz.isVertexValid(stringIdRow, tagConfigEntryWithPrefix, false, false))
assertThrows[AssertionError](
processClazz.isVertexValid(stringIdRow, tagConfigEntryWithPrefix, false, true))
}

@Test
Expand All @@ -117,6 +144,12 @@ class VerticesProcessorSuite {
assert(vertex.vertexID.equals("\"1\""))
assert(vertex.toString.equals(
"Vertex ID: \"1\", Values: \"\", \"fixedBob\", 12, 200, 1000, 100000, date(\"2021-01-01\"), datetime(\"2021-01-01T12:00:00.100\"), time(\"12:00:00.100\"), 345436232, true, 12.01, 22.12, ST_GeogFromText(\"POINT(3 8)\")"))

val tagConfigEntryWithPrefix =
TagConfigEntry("person", null, null, List(), List(), "id", None, "prefix", 10, 10, None)
val vertexWithPrefix =
processClazz.convertToVertex(row, tagConfigEntryWithPrefix, true, fieldKeys, map)
assert(vertexWithPrefix.vertexID.equals("\"prefix_1\""))
}

@Test
Expand Down
16 changes: 12 additions & 4 deletions ...xchange_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,9 @@ class EdgeProcessor(spark: SparkSession,
s"only int vidType can use policy, but your vidType is FIXED_STRING.your row data is $row")
false
} else true
idFlag && policyFlag

val udfFlag = isVidStringType || policy.isEmpty || (edgeConfig.sourcePrefix == null && edgeConfig.targetPrefix == null)
idFlag && policyFlag && udfFlag
}

/**
Expand All @@ -273,13 +275,15 @@ class EdgeProcessor(spark: SparkSession,
"source_field",
row,
edgeConfig.sourcePolicy,
isVidStringType)
isVidStringType,
edgeConfig.sourcePrefix)

val targetField = processField(edgeConfig.targetField,
"target_field",
row,
edgeConfig.targetPolicy,
isVidStringType)
isVidStringType,
edgeConfig.targetPrefix)

val values = for {
property <- fieldKeys if property.trim.length != 0
Expand All @@ -301,7 +305,8 @@ class EdgeProcessor(spark: SparkSession,
fieldType: String,
row: Row,
policy: Option[KeyPolicy.Value],
isVidStringType: Boolean): String = {
isVidStringType: Boolean,
prefix: String): String = {
var fieldValue = if (edgeConfig.isGeo && "source_field".equals(fieldType)) {
val lat = row.getDouble(row.schema.fieldIndex(edgeConfig.latitude.get))
val lng = row.getDouble(row.schema.fieldIndex(edgeConfig.longitude.get))
Expand All @@ -311,6 +316,9 @@ class EdgeProcessor(spark: SparkSession,
val value = row.get(index).toString.trim
if (value.equals(DEFAULT_EMPTY_VALUE)) "" else value
}
if (prefix != null) {
fieldValue = prefix + "_" + fieldValue
}
// process string type vid
if (policy.isEmpty && isVidStringType) {
fieldValue = NebulaUtils.escapeUtil(fieldValue).mkString("\"", "", "\"")
Expand Down
Loading