[Feature][Connector-v2] Neo4j source connector #2777

getChan · 2022-09-18T11:31:28Z

Purpose of this pull request

To add Neo4j Source Connector

Check list

Code changed are covered with tests, or it does not need tests for reason:
If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs

CalvinKirs · 2022-09-19T02:36:36Z

hi, thanks for your contribution, could you add the e2e test for this PR?

https://github.com/apache/incubator-seatunnel/tree/dev/seatunnel-e2e

hailin0 · 2022-09-19T05:48:39Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSourceConfig.java

+    public static final String PLUGIN_NAME = "Neo4j";
+
+    public static final String KEY_NEO4J_URI = "uri";
+    public static final String KEY_USERNAME = "username";
+    public static final String KEY_PASSWORD = "password";
+    public static final String KEY_BEARER_TOKEN = "bearer_token";
+    public static final String KEY_KERBEROS_TICKET = "kerberos_ticket"; // Base64 encoded
+
+    public static final String KEY_DATABASE = "database";
+    public static final String KEY_QUERY = "query";
+    public static final String KEY_MAX_TRANSACTION_RETRY_TIME = "max_transaction_retry_time";
+    public static final String KEY_MAX_CONNECTION_TIMEOUT = "max_connection_timeout";


Extract common configuration to Neo4jCommonConfig?

hailin0 · 2022-09-19T05:55:32Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceReader.java

+        if (dataType.equals(BasicType.STRING_TYPE)) {
+            return value.asString();
+        } else if (dataType.equals(BasicType.BOOLEAN_TYPE)) {
+            return value.asBoolean();
+        } else if (dataType.equals(BasicType.LONG_TYPE)) {
+            return value.asLong();
+        } else if (dataType.equals(BasicType.DOUBLE_TYPE)) {
+            return value.asDouble();
+        } else if (dataType.equals(BasicType.VOID_TYPE)) {
+            return null;
+        } else if (dataType.equals(PrimitiveByteArrayType.INSTANCE)) {
+            return value.asByteArray();
+        } else if (dataType.equals(LocalTimeType.LOCAL_DATE_TYPE)) {
+            return value.asLocalDate();
+        } else if (dataType.equals(LocalTimeType.LOCAL_TIME_TYPE)) {
+            return value.asLocalTime();
+        } else if (dataType.equals(LocalTimeType.LOCAL_DATE_TIME_TYPE)) {
+            return value.asLocalDateTime();
+        } else if (dataType instanceof MapType) {
+            if (!((MapType<?, ?>) dataType).getKeyType().equals(BasicType.STRING_TYPE)) {
+                throw new IllegalArgumentException("Key Type of MapType must String type");
+            }
+            return value.asMap();
+        } else if (dataType.equals(BasicType.INT_TYPE)) {
+            return value.asInt();
+        } else if (dataType.equals(BasicType.FLOAT_TYPE)) {
+            return value.asFloat();
+        } else {
+            throw new IllegalArgumentException("not supported data type: " + dataType);
+        }


Use switch(dataType.getSqlType()) ?

getChan · 2022-09-22T14:08:54Z

@hailin0
thanks for your review. I fixed it.
and add e2e test.

EricJoy2048 · 2022-09-26T09:20:06Z

...-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSource.java

+        }
+        neo4jSourceConfig.setQuery(pluginConfig.getString(KEY_QUERY));
+
+        final CheckResult schemaConfigCheck = CheckConfigUtil.checkAllExists(pluginConfig, SeaTunnelSchema.SCHEMA);


Why not use CheckConfigUtil.checkAllExists(pluginConfig, KEY_QUERY, SeaTunnelSchema.SCHEMA)

EricJoy2048 · 2022-09-26T09:43:48Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceReader.java

+        Objects.requireNonNull(value);
+
+        switch (dataType.getSqlType()) {
+            case STRING:


Can you case Neo4j.List to SqlType.ARRAY?

EricJoy2048 · 2022-09-26T09:47:09Z

...connector-neo4j-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/neo4j/Neo4jIT.java

+        // then
+        Assertions.assertEquals(0, execResult.getExitCode());
+
+        final Result result = neo4jSession.run("MATCH (a:Person) RETURN a.name, a.age");


Please test all of the datatype.

EricJoy2048 · 2022-09-26T09:47:18Z

...connector-neo4j-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/neo4j/Neo4jIT.java

+    @Test
+    public void testSource() throws IOException, InterruptedException {
+        // given
+        neo4jSession.run("CREATE (a:Person {name: 'foo', age: 10})");


Same as above.

EricJoy2048 · 2022-09-26T09:48:06Z

...connector-neo4j-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/neo4j/Neo4jIT.java

+        Container.ExecResult execResult = executeSeaTunnelSparkJob("/neo4j/fake_to_neo4j.conf");
+
+        // then
+        Assertions.assertEquals(0, execResult.getExitCode());


Same as the flink e2e.

hailin0 · 2022-09-24T02:35:14Z

...flink-connector-v2-e2e/connector-neo4j-flink-e2e/src/test/resources/neo4j/fake_to_neo4j.conf

+            name=STRING
+            age=INT


test all datatypes?

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/mongodb_source_and_sink.conf#L36

hailin0 · 2022-09-24T02:36:41Z

...ink-connector-v2-e2e/connector-neo4j-flink-e2e/src/test/resources/neo4j/neo4j_to_assert.conf

+        query = "MATCH (a:Person) RETURN a.name, a.age"
+
+        schema {
+            fields {


test all datatypes?

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/mongodb_source_and_sink.conf#L36

hailin0 · 2022-09-24T02:37:22Z

...spark-connector-v2-e2e/connector-neo4j-spark-e2e/src/test/resources/neo4j/fake_to_neo4j.conf

+    FakeSource {
+      result_table_name = "fake"
+      schema = {
+        fields = {


test all datatypes?

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/mongodb_source_and_sink.conf#L36

hailin0 · 2022-09-24T02:37:30Z

...ark-connector-v2-e2e/connector-neo4j-spark-e2e/src/test/resources/neo4j/neo4j_to_assert.conf

+        query = "MATCH (a:Person) RETURN a.name, a.age"
+
+        schema {
+            fields {


test all datatypes?

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/mongodb_source_and_sink.conf#L36

hailin0 · 2022-09-24T02:39:16Z

...connector-neo4j-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/neo4j/Neo4jIT.java

+    }
+
+    @Test
+    public void testSink() throws IOException, InterruptedException {


Can you test source and sink at the once job？

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/mongodb/MongodbIT.java#L106

getChan · 2022-10-09T17:03:42Z

@EricJoy2048 @hailin0
I applied your reviews.
except for TIME datatypes at spark. It does not supported at spark

EricJoy2048 · 2022-10-13T05:54:02Z

@EricJoy2048 @hailin0 I applied your reviews. except for TIME datatypes at spark. It does not supported at spark

Yes I know.

EricJoy2048

LGTM

hailin0

LGTM

...connector-neo4j-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/neo4j/Neo4jIT.java

...connector-neo4j-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/neo4j/Neo4jIT.java

hailin0 · 2022-10-13T06:30:20Z

docs/en/connector-v2/source/Neo4j.md

+                a.age=INT
+                a.name=STRING


Better not to include . ？

e.g.

xxx_age = INT
xxx_name = STRING

@hailin0
I think the dot(.) expression is familiar to neo4j users.
because dot means reference of node's property.

is the dot naming in the field forbidden in SeaTunnel?

@hailin0 PTAL

is the dot naming in the field forbidden in SeaTunnel?

You could use, seatunnel currently has no restrictions on field names

…-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/neo4j/Neo4jIT.java Co-authored-by: hailin0 <[email protected]>

…-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/neo4j/Neo4jIT.java Co-authored-by: hailin0 <[email protected]>

…onnectorv2-neo4j-source

EricJoy2048 · 2022-10-15T10:51:37Z

Hi, @getChan Thanks for your contribution, Please resolve conflicts.

# Conflicts: # seatunnel-e2e/seatunnel-flink-connector-v2-e2e/pom.xml # seatunnel-e2e/seatunnel-spark-connector-v2-e2e/pom.xml

hailin0

LGTM

EricJoy2048

+1

CalvinKirs · 2022-10-21T06:33:19Z

Thanks:)

warm-inf added 14 commits August 31, 2022 23:00

ConnectorV2 Neo4j Source - Add Classes

c421c97

Merge branch 'dev' into connectorv2-neo4j-source

488b659

Merge branch 'dev' into connectorv2-neo4j-source

79d8ec7

ConnectorV2 Neo4j Source - licence header

e173e38

ConnectorV2 Neo4j Source - init config

147b0f9

Merge branch 'dev' into connectorv2-neo4j-source

b5541ff

ConnectorV2 Neo4j Source - single split source

3f44ca3

ConnectorV2 Neo4j Source - convert type from neo4j to seatunnel

60abba3

ConnectorV2 Neo4j Source - not support byte, short type. overflow

78564f4

Merge branch 'dev' into connectorv2-neo4j-source

2f0192b

ConnectorV2 Neo4j Source - write doc

833bef0

ConnectorV2 Neo4j Source - codestyle

f29850f

ConnectorV2 Neo4j Source - test convertType

7ba61e3

Merge branch 'dev' into connectorv2-neo4j-source

e1bf217

CalvinKirs added the connectors-v2 label Sep 19, 2022

hailin0 reviewed Sep 19, 2022

View reviewed changes

warm-inf added 3 commits September 20, 2022 21:23

ConnectorV2 Neo4j Source - extract Neo4jCommonConfig

167ec1b

ConnectorV2 Neo4j Source - convertType use switch syntax

a8e4da5

Merge branch 'dev' into connectorv2-neo4j-source

1eb378c

EricJoy2048 mentioned this pull request Sep 21, 2022

[Umbrella][Connector] New SeaTunnel API Connectors #1946

Closed

warm-inf added 9 commits September 21, 2022 22:46

ConnectorV2 Neo4j Source - e2e test

47d23ff

Merge branch 'dev' into connectorv2-neo4j-source

fd911ea

ConnectorV2 Neo4j Source - fix codestyle

91eeee6

ConnectorV2 Neo4j Source - sink assert

8f28e0c

ConnectorV2 Neo4j Source - dbname

ecd5180

ConnectorV2 Neo4j Source - allow except neo4j scheme

4758e49

ConnectorV2 Neo4j Source - testcontainer hostname

0b2e919

ConnectorV2 Neo4j Source - testcontainer hostname

235baf1

ConnectorV2 Neo4j Source - fix IT

5712932

EricJoy2048 reviewed Sep 26, 2022

View reviewed changes

hailin0 reviewed Sep 26, 2022

View reviewed changes

warm-inf added 2 commits September 26, 2022 22:17

ConnectorV2 Neo4j Source - merge source config validation

f804be9

ConnectorV2 Neo4j Source - support ARRAY type of seaTunnelDataType

34fbafa

EricJoy2048 mentioned this pull request Oct 8, 2022

[Umbrella][Connector-V2] SeaTunnel Connector V2 Tracking Table #3018

Open

3 tasks

warm-inf added 3 commits October 8, 2022 23:14

Merge branch 'dev' into connectorv2-neo4j-source

d8842d8

ConnectorV2 Neo4j Source - fix convertType, complete flink e2e

fb3a716

ConnectorV2 Neo4j Source - complete spark e2e

84d35da

EricJoy2048 previously approved these changes Oct 13, 2022

View reviewed changes

hailin0 reviewed Oct 13, 2022

View reviewed changes

Update seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-neo4j…

a860b1b

…-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/neo4j/Neo4jIT.java Co-authored-by: hailin0 <[email protected]>

getChan dismissed EricJoy2048’s stale review via a860b1b October 13, 2022 12:53

getChan and others added 5 commits October 13, 2022 21:55

Update seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-neo4j…

198e719

…-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/neo4j/Neo4jIT.java Co-authored-by: hailin0 <[email protected]>

ConnectorV2 Neo4j Source E2E - import DockerLoggerFactory

96aff47

Merge branch 'apache:dev' into connectorv2-neo4j-source

703a127

Merge branch 'dev' into connectorv2-neo4j-source

d303d6d

Merge remote-tracking branch 'origin/connectorv2-neo4j-source' into c…

80bbe19

…onnectorv2-neo4j-source

getChan and others added 2 commits October 16, 2022 21:44

Merge branch 'dev' into connectorv2-neo4j-source

1a14b84

# Conflicts: # seatunnel-e2e/seatunnel-flink-connector-v2-e2e/pom.xml # seatunnel-e2e/seatunnel-spark-connector-v2-e2e/pom.xml

Merge branch 'dev' into connectorv2-neo4j-source

5b821e1

CalvinKirs approved these changes Oct 21, 2022

View reviewed changes

hailin0 approved these changes Oct 21, 2022

View reviewed changes

EricJoy2048 approved these changes Oct 21, 2022

View reviewed changes

EricJoy2048 merged commit 38b0daf into apache:dev Oct 21, 2022

Carl-Zhou-CN pushed a commit to Carl-Zhou-CN/incubator-seatunnel that referenced this pull request Oct 27, 2022

[Feature][Connector-v2] Neo4j source connector (apache#2777)

55cdbc8

Carl-Zhou-CN pushed a commit to Carl-Zhou-CN/incubator-seatunnel that referenced this pull request Oct 31, 2022

[Feature][Connector-v2] Neo4j source connector (apache#2777)

bf5328a

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][Connector-v2] Neo4j source connector #2777

[Feature][Connector-v2] Neo4j source connector #2777

getChan commented Sep 18, 2022

CalvinKirs commented Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

getChan commented Sep 22, 2022

EricJoy2048 Sep 26, 2022

EricJoy2048 Sep 26, 2022

EricJoy2048 Sep 26, 2022

EricJoy2048 Sep 26, 2022

EricJoy2048 Sep 26, 2022

hailin0 Sep 24, 2022

hailin0 Sep 24, 2022

hailin0 Sep 24, 2022

hailin0 Sep 24, 2022

hailin0 Sep 24, 2022

getChan commented Oct 9, 2022

EricJoy2048 commented Oct 13, 2022

EricJoy2048 left a comment

hailin0 left a comment

hailin0 Oct 13, 2022

getChan Oct 13, 2022

EricJoy2048 Oct 14, 2022

hailin0 Oct 21, 2022

EricJoy2048 commented Oct 15, 2022

hailin0 left a comment

EricJoy2048 left a comment

CalvinKirs commented Oct 21, 2022

[Feature][Connector-v2] Neo4j source connector #2777

[Feature][Connector-v2] Neo4j source connector #2777

Conversation

getChan commented Sep 18, 2022

Purpose of this pull request

Check list

CalvinKirs commented Sep 19, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

getChan commented Sep 22, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

getChan commented Oct 9, 2022

EricJoy2048 commented Oct 13, 2022

EricJoy2048 left a comment

Choose a reason for hiding this comment

hailin0 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

EricJoy2048 commented Oct 15, 2022

hailin0 left a comment

Choose a reason for hiding this comment

EricJoy2048 left a comment

Choose a reason for hiding this comment

CalvinKirs commented Oct 21, 2022