updated relational sources to use ns

ruokun-niu · ruokun-niu · commit 54e57777b6c3 · 2025-03-10T13:55:53.000-07:00
diff --git a/query-container/query-host/src/api.rs b/query-container/query-host/src/api.rs
@@ -162,7 +162,7 @@ pub struct ChangeEvent {
 #[derive(Serialize, Deserialize, Debug)]
 struct ChangeEventTime {
     seq: u64,
-    ms: u64,
+    ns: u64,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -246,7 +246,7 @@ enum ElementType {
 
 impl ChangeEvent {
     pub fn get_timestamp(&self) -> u64 {
-        self.time.ms
+        self.time.ns
     }
 
     pub fn has_query(&self, query_id: &str) -> bool {
@@ -266,7 +266,7 @@ impl ChangeEvent {
             source_id: future_ref.element_ref.source_id.to_string(),
             time: ChangeEventTime {
                 seq: 0,
-                ms: future_ref.original_time,
+                ns: future_ref.original_time,
             },
             queries: vec![query_id.to_string()],
             op: ChangeType::Future,
@@ -293,7 +293,7 @@ impl TryInto<SourceChange> for ChangeEvent {
                         &self.source_id,
                         self.element_type
                             .ok_or(ApiError::BadRequest("missing element type".into()))?,
-                        self.time.ms,
+                        self.time.ns,
                     )?,
             },
             ChangeType::Update => SourceChange::Update {
@@ -304,12 +304,12 @@ impl TryInto<SourceChange> for ChangeEvent {
                         &self.source_id,
                         self.element_type
                             .ok_or(ApiError::BadRequest("missing element type".into()))?,
-                        self.time.ms,
+                        self.time.ns,
                     )?,
             },
             ChangeType::Delete => SourceChange::Delete {
                 metadata: ElementMetadata {
-                    effective_from: self.time.ms,
+                    effective_from: self.time.ns,
                     labels: Arc::from(match &self.before {
                         Some(before) => match &before.labels {
                             Some(labels) => labels.iter().map(|l| Arc::from(l.as_str())).collect(),
@@ -326,7 +326,7 @@ impl TryInto<SourceChange> for ChangeEvent {
             ChangeType::Future => SourceChange::Future {
                 future_ref: FutureElementRef {
                     element_ref: ElementReference::new(&self.source_id, &self.id),
-                    original_time: self.time.ms,
+                    original_time: self.time.ns,
                     due_time: self
                         .future_due_time
                         .ok_or(ApiError::BadRequest("Missing future due time".to_string()))?,
diff --git a/sources/cosmosdb/cosmosdb-ffcf-reactivator/Program.cs b/sources/cosmosdb/cosmosdb-ffcf-reactivator/Program.cs
@@ -1,17 +1,17 @@
-// Copyright 2024 The Drasi Authors.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
+// Copyright 2024 The Drasi Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 using System.Net;
 using System.Text;
 using Microsoft.Azure.Cosmos;
@@ -131,7 +131,7 @@ private static JObject FormatDebeziumEvent(JObject feedEvent, string changeEvent
                     throw new NotSupportedException();
             }            
             
-            result["ts_ms"] = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+            result["ts_ns"] = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000.0;
             result["schema"] = "";
 
             var isRelation = (bool)(source["_isEdge"] ?? false);
diff --git a/sources/relational/debezium-reactivator/Dockerfile b/sources/relational/debezium-reactivator/Dockerfile
@@ -16,14 +16,52 @@
 # Build stage
 #
 FROM --platform=$BUILDPLATFORM maven:3.9.9-amazoncorretto-21 AS build
+WORKDIR /home/app
+
+# Copy the source code, pom.xml, and lib folder
 COPY src /home/app/src
-COPY pom.xml /home/app
+COPY pom.xml /home/app/
+COPY lib /home/app/lib
+
+# Run the Maven build
 RUN mvn -f /home/app/pom.xml clean package
 
-RUN cd /home/app && mvn dependency:tree
+# Optional: Uncomment to debug dependency tree
+# RUN mvn dependency:tree
+
 #
 # Package stage
 #
 FROM --platform=$TARGETPLATFORM mcr.microsoft.com/openjdk/jdk:21-distroless
 COPY --from=build /home/app/target /home/app
-CMD ["-jar","/home/app/debezium-reactivator-1.0.jar"]
+CMD ["-jar", "/home/app/debezium-reactivator-1.0.jar"]
+
+# # Copyright 2024 The Drasi Authors.
+# #
+# # Licensed under the Apache License, Version 2.0 (the "License");
+# # you may not use this file except in compliance with the License.
+# # You may obtain a copy of the License at
+# #
+# #     http://www.apache.org/licenses/LICENSE-2.0
+# #
+# # Unless required by applicable law or agreed to in writing, software
+# # distributed under the License is distributed on an "AS IS" BASIS,
+# # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# # See the License for the specific language governing permissions and
+# # limitations under the License.
+
+# #
+# # Build stage
+# #
+# FROM --platform=$BUILDPLATFORM maven:3.9.9-amazoncorretto-21 AS build
+# COPY src /home/app/src
+# COPY pom.xml /home/app
+# RUN mvn -f /home/app/pom.xml clean package
+
+# RUN cd /home/app && mvn dependency:tree
+# #
+# # Package stage
+# #
+# FROM --platform=$TARGETPLATFORM mcr.microsoft.com/openjdk/jdk:21-distroless
+# COPY --from=build /home/app/target /home/app
+# CMD ["-jar","/home/app/debezium-reactivator-1.0.jar"]
diff --git a/sources/relational/debezium-reactivator/lib/source.sdk-0.1.4.jar b/sources/relational/debezium-reactivator/lib/source.sdk-0.1.4.jar
diff --git a/sources/relational/debezium-reactivator/pom.xml b/sources/relational/debezium-reactivator/pom.xml
@@ -128,9 +128,30 @@
           </configuration>
         </plugin>
         <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-install-plugin</artifactId>
+                <version>3.1.1</version>
+                <executions>
+                    <execution>
+                        <id>install-local-jar</id>
+                        <phase>initialize</phase>
+                        <goals>
+                            <goal>install-file</goal>
+                        </goals>
+                        <configuration>
+                            <file>lib/source.sdk-0.1.4.jar</file>
+                            <groupId>io.drasi</groupId>
+                            <artifactId>source.sdk</artifactId>
+                            <version>0.1.4</version>
+                            <packaging>jar</packaging>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        <!-- <plugin>
           <artifactId>maven-install-plugin</artifactId>
           <version>2.5.2</version>
-        </plugin>
+        </plugin> -->
         <plugin>
           <artifactId>maven-deploy-plugin</artifactId>
           <version>2.8.2</version>
diff --git a/sources/relational/debezium-reactivator/src/main/java/io/drasi/RelationalChangeConsumer.java b/sources/relational/debezium-reactivator/src/main/java/io/drasi/RelationalChangeConsumer.java
@@ -124,7 +124,11 @@ private SourceChange ExtractDrasiChange(JsonNode sourceChange) {
         }
 
         var nodeId = createNodeId(mapping.tableName, item.path(mapping.keyField).asText());
-        var timestamp = payload.path("ts_ms").asLong();
+        System.out.println("Payload: " + payload.toPrettyString()); // or payload.toString() for compact format
+
+        var timestamp = payload.path("ts_ns").asLong();
+        System.out.println("Timestamp: " + timestamp);
+
         var lsn = dbStrategy.extractLsn(source);
 
         return switch (changeType) {
diff --git a/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceChange.java b/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceChange.java
@@ -27,7 +27,7 @@ public abstract class SourceChange {
     protected abstract Op getOp();
 
     private String id;
-    private long tsMS;
+    private long tsNS;
     private JsonNode properties;
     private Map<String, Object> metadata;
     private List<String> labels;
@@ -38,9 +38,9 @@ public abstract class SourceChange {
     private String sourceTable;
 
 
-    protected SourceChange(String id, long tsMS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long lsn) {
+    protected SourceChange(String id, long tsNS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long lsn) {
         this.id = id;
-        this.tsMS = tsMS;
+        this.tsNS = tsNS;
         this.properties = properties;
         this.metadata = metadata;
         this.labels = labels;
@@ -49,9 +49,9 @@ protected SourceChange(String id, long tsMS, JsonNode properties, Map<String, Ob
         this.sourceTable = "node";
     }
 
-    protected SourceChange(String id, long tsMS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long lsn, String startId, String endId) {
+    protected SourceChange(String id, long tsNS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long lsn, String startId, String endId) {
         this.id = id;
-        this.tsMS = tsMS;
+        this.tsNS = tsNS;
         this.properties = properties;
         this.metadata = metadata;
         this.labels = labels;
@@ -67,7 +67,7 @@ public String toJson() {
         rgSource.put("db", Reactivator.SourceId());
         rgSource.put("table", sourceTable);
         rgSource.put("lsn", lsn);
-        rgSource.put("ts_ms", sourceTsMS);
+        rgSource.put("ts_ns", sourceTsMS);
 
         var payload = JsonNodeFactory.instance.objectNode();
         payload.set("source", rgSource);
@@ -93,7 +93,7 @@ public String toJson() {
                 result.put("op", "d");
                 break;
         }
-        result.put("ts_ms", tsMS);
+        result.put("ts_ns", tsNS);
         result.set("payload", payload);
 
         return result.toString();
diff --git a/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceDelete.java b/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceDelete.java
@@ -35,30 +35,30 @@ protected Op getOp() {
      * Create a delete operation of a node with the given id, timestamp, metadata and labels.
      *
      * @param id         The id of the element.
-     * @param tsMS       The timestamp of the event in milliseconds.
+     * @param tsNS       The timestamp of the event in nanoseconds.
      * @param metadata   The metadata of the element.
      * @param labels     The labels of the element.
      * @param sourceTsMS The timestamp of the event in the source database in milliseconds.
      * @param sequenceNumber The sequence number of the event in the source database.
      */
-    public SourceDelete(String id, long tsMS, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber) {
-        super(id, tsMS, null, metadata, labels, sourceTsMS, sequenceNumber);
+    public SourceDelete(String id, long tsNS, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber) {
+        super(id, tsNS, null, metadata, labels, sourceTsMS, sequenceNumber);
 
     }
 
     /**
      * Create a delete operation of a relation with the given id, timestamp, metadata, labels, startId and endId.
      *
      * @param id         The id of the element.
-     * @param tsMS       The timestamp of the event in milliseconds.
+     * @param tsNS       The timestamp of the event in nanoseconds.
      * @param metadata   The metadata of the element.
      * @param labels     The labels of the element.
      * @param sourceTsMS The timestamp of the event in the source database in milliseconds.
      * @param sequenceNumber The sequence number of the event in the source database.
      * @param startId    The id of the start node.
      * @param endId      The id of the end node.
      */
-    public SourceDelete(String id, long tsMS, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber, String startId, String endId) {
-        super(id, tsMS, null, metadata, labels, sourceTsMS, sequenceNumber, startId, endId);
+    public SourceDelete(String id, long tsNS, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber, String startId, String endId) {
+        super(id, tsNS, null, metadata, labels, sourceTsMS, sequenceNumber, startId, endId);
     }
 }
diff --git a/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceInsert.java b/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceInsert.java
@@ -35,23 +35,23 @@ protected Op getOp() {
      * Create a new insert operation of a node with the given id, timestamp, properties, metadata and labels.
      *
      * @param id         The id of the element.
-     * @param tsMS       The timestamp of the event in milliseconds.
+     * @param tsNS       The timestamp of the event in nanoseconds.
      * @param properties The properties of the element.
      * @param metadata   The metadata of the element.
      * @param labels     The labels of the element.
      * @param sourceTsMS The timestamp of the event in the source database in milliseconds.
      * @param sequenceNumber The sequence number of the event in the source database.
      */
-    public SourceInsert(String id, long tsMS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber) {
-        super(id, tsMS, properties, metadata, labels, sourceTsMS, sequenceNumber);
+    public SourceInsert(String id, long tsNS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber) {
+        super(id, tsNS, properties, metadata, labels, sourceTsMS, sequenceNumber);
 
     }
 
     /**
      * Create a new insert  operation of a relation with the given id, timestamp, properties, metadata, labels, startId and endId.
      *
      * @param id         The id of the element.
-     * @param tsMS       The timestamp of the event in milliseconds.
+     * @param tsNS       The timestamp of the event in nanoseconds.
      * @param properties The properties of the element.
      * @param metadata   The metadata of the element.
      * @param labels     The labels of the element.
@@ -60,7 +60,7 @@ public SourceInsert(String id, long tsMS, JsonNode properties, Map<String, Objec
      * @param startId    The id of the start node.
      * @param endId      The id of the end node.
      */
-    public SourceInsert(String id, long tsMS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber, String startId, String endId) {
-        super(id, tsMS, properties, metadata, labels, sourceTsMS, sequenceNumber, startId, endId);
+    public SourceInsert(String id, long tsNS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber, String startId, String endId) {
+        super(id, tsNS, properties, metadata, labels, sourceTsMS, sequenceNumber, startId, endId);
     }
 }
diff --git a/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceUpdate.java b/sources/sdk/java/src/main/java/io/drasi/source/sdk/models/SourceUpdate.java
@@ -35,23 +35,23 @@ protected Op getOp() {
      * Create an update operation of a node with the given id, timestamp, properties, metadata and labels.
      *
      * @param id         The id of the element.
-     * @param tsMS       The timestamp of the event in milliseconds.
+     * @param tsNS       The timestamp of the event in nanoseconds.
      * @param properties The properties of the element.
      * @param metadata   The metadata of the element.
      * @param labels     The labels of the element.
      * @param sourceTsMS The timestamp of the event in the source database in milliseconds.
      * @param sequenceNumber The sequence number of the event in the source database.
      */
-    public SourceUpdate(String id, long tsMS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber) {
-        super(id, tsMS, properties, metadata, labels, sourceTsMS, sequenceNumber);
+    public SourceUpdate(String id, long tsNS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber) {
+        super(id, tsNS, properties, metadata, labels, sourceTsMS, sequenceNumber);
 
     }
 
     /**
      * Create an update operation of a relation with the given id, timestamp, properties, metadata, labels, startId and endId.
      *
      * @param id         The id of the element.
-     * @param tsMS       The timestamp of the event in milliseconds.
+     * @param tsNS       The timestamp of the event in nanoseconds.
      * @param properties The properties of the element.
      * @param metadata   The metadata of the element.
      * @param labels     The labels of the element.
@@ -60,7 +60,7 @@ public SourceUpdate(String id, long tsMS, JsonNode properties, Map<String, Objec
      * @param startId    The id of the start node.
      * @param endId      The id of the end node.
      */
-    public SourceUpdate(String id, long tsMS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber, String startId, String endId) {
-        super(id, tsMS, properties, metadata, labels, sourceTsMS, sequenceNumber, startId, endId);
+    public SourceUpdate(String id, long tsNS, JsonNode properties, Map<String, Object> metadata, List<String> labels, long sourceTsMS, long sequenceNumber, String startId, String endId) {
+        super(id, tsNS, properties, metadata, labels, sourceTsMS, sequenceNumber, startId, endId);
     }
 }
diff --git a/sources/shared/change-dispatcher/src/main.rs b/sources/shared/change-dispatcher/src/main.rs
@@ -162,13 +162,6 @@ async fn process_changes(
                         ));
                     }
                 };
-            // dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherEnd_ms"] =
-            //     match serde_json::to_value(0) {
-            //         Ok(val) => val,
-            //         Err(_) => {
-            //             unreachable!();
-            //         }
-            //     };
 
             let subscriptions = match change_event["subscriptions"].as_array() {
                 Some(subs) => subs.clone(),
diff --git a/sources/shared/change-router/src/main.rs b/sources/shared/change-router/src/main.rs
diff --git a/sources/shared/query-api/src/api.rs b/sources/shared/query-api/src/api.rs
diff --git a/sources/shared/query-api/src/main.rs b/sources/shared/query-api/src/main.rs
