Use last_ingested_at as first_ingested_at on create case

Author: zprobst

nodestream_plugin_neo4j/ingest_query_builder.py

@@ -117,7 +117,7 @@ def _make_relationship(
         return f"{create_rel_query} {set_properties_query}"
 
     if set_first_ingested_at:
-        merge_rel_query = f"{merge_rel_query} ON CREATE SET {RELATIONSHIP_REF_NAME}.first_ingested_at = datetime()"
+        merge_rel_query = f"{merge_rel_query} ON CREATE SET {RELATIONSHIP_REF_NAME}.first_ingested_at = params.{generate_prefixed_param_name(PROPERTIES_PARAM_NAME, RELATIONSHIP_REF_NAME)}['last_ingested_at']"
 
     return f"{merge_rel_query} {set_properties_query}"
 
@@ -142,7 +142,7 @@ def generate_update_node_operation_query_statement(
             # property  and we are in the merge case. Because MATCH will not
             # create the node.
             if self.set_first_ingested_at:
-                query = f"{query} ON CREATE SET {GENERIC_NODE_REF_NAME}.first_ingested_at = datetime()"
+                query = f"{query} ON CREATE SET {GENERIC_NODE_REF_NAME}.first_ingested_at = params.{generate_prefixed_param_name(PROPERTIES_PARAM_NAME, GENERIC_NODE_REF_NAME)}['last_ingested_at']"
         else:
             query = str(_match_node(operation))
 

tests/e2e/test_neo4j_pipelines.py

@@ -20,6 +20,7 @@ def project():
 
 
 def validate_timestamps(session):
+    # Check that all nodes have a first_ingested_at property
     nodes_with_first_ingested_at = session.run(
         """
         MATCH (n) where n.first_ingested_at IS NOT NULL

tests/unit/test_ingest_query_builder.py

@@ -178,7 +178,7 @@ def test_generates_expected_queries(mocked_utcnow, query_builder, ttl, expected_
     ],
 )
 SIMPLE_NODE_EXPECTED_QUERY_WITH_CREATED_TIMESTAMP = QueryBatch(
-    "MERGE (node: TestType {id : params.__node_id}) ON CREATE SET node.first_ingested_at = datetime() SET node += params.__node_properties",
+    "MERGE (node: TestType {id : params.__node_id}) ON CREATE SET node.first_ingested_at = params.__node_properties['last_ingested_at'] SET node += params.__node_properties",
     [
         {
             "__node_id": "foo",
@@ -218,7 +218,7 @@ def test_generates_expected_queries(mocked_utcnow, query_builder, ttl, expected_
 )
 
 COMPLEX_NODE_EXPECTED_QUERY_WITH_CREATED_TIMESTAMP = QueryBatch(
-    "MERGE (node: ComplexType {id : params.__node_id}) ON CREATE SET node.first_ingested_at = datetime() WITH node, params CALL apoc.create.addLabels(node, params.__node_additional_labels) yield node as _ SET node += params.__node_properties",
+    "MERGE (node: ComplexType {id : params.__node_id}) ON CREATE SET node.first_ingested_at = params.__node_properties['last_ingested_at'] WITH node, params CALL apoc.create.addLabels(node, params.__node_additional_labels) yield node as _ SET node += params.__node_properties",
     [
         {
             "__node_id": "foo",
@@ -247,7 +247,7 @@ def test_generates_expected_queries(mocked_utcnow, query_builder, ttl, expected_
 )
 
 COMPLEX_NODE_TWO_EXPECTED_QUERY_WITH_CREATED_TIMESTAMP = QueryBatch(
-    "MERGE (node: ComplexType {id_part1 : params.__node_id_part1, id_part2 : params.__node_id_part2}) ON CREATE SET node.first_ingested_at = datetime() WITH node, params CALL apoc.create.addLabels(node, params.__node_additional_labels) yield node as _ SET node += params.__node_properties",
+    "MERGE (node: ComplexType {id_part1 : params.__node_id_part1, id_part2 : params.__node_id_part2}) ON CREATE SET node.first_ingested_at = params.__node_properties['last_ingested_at'] WITH node, params CALL apoc.create.addLabels(node, params.__node_additional_labels) yield node as _ SET node += params.__node_properties",
     [
         {
             "__node_id_part1": "foo",
@@ -325,6 +325,19 @@ def test_node_update_with_created_timestamp_generates_exepected_queries(
     ],
 )
 
+RELATIONSHIP_BETWEEN_TWO_NODES_EXPECTED_QUERY_WITH_CREATED_TIMESTAMP = QueryBatch(
+    """MATCH (from_node: TestType) WHERE from_node.id = params.__from_node_id MATCH (to_node: ComplexType) WHERE to_node.id = params.__to_node_id
+    MERGE (from_node)-[rel: RELATED_TO]->(to_node) ON CREATE SET rel.first_ingested_at = params.__rel_properties['last_ingested_at'] SET rel += params.__rel_properties
+    """,
+    [
+        {
+            "__from_node_id": "foo",
+            "__to_node_id": "foo",
+            "__rel_properties": RELATIONSHIP_BETWEEN_TWO_NODES.relationship.properties,
+        }
+    ],
+)
+
 RELATIONSHIP_BETWEEN_TWO_NODES_WITH_MULTI_KEY = RelationshipWithNodes(
     from_node=SIMPLE_NODE,
     to_node=COMPLEX_NODE_TWO,