@@ -111,6 +111,14 @@ export class ClickhouseEventRepository implements IEventRepository {
111
111
span . setAttribute ( "flush_id" , flushId ) ;
112
112
span . setAttribute ( "event_count" , events . length ) ;
113
113
114
+ const firstEvent = events [ 0 ] ;
115
+
116
+ if ( firstEvent ) {
117
+ logger . debug ( "ClickhouseEventRepository.flushBatch first event" , {
118
+ event : firstEvent ,
119
+ } ) ;
120
+ }
121
+
114
122
const [ insertError , insertResult ] = await this . _clickhouse . taskEvents . insert ( events ) ;
115
123
116
124
if ( insertError ) {
@@ -147,7 +155,7 @@ export class ClickhouseEventRepository implements IEventRepository {
147
155
project_id : event . projectId ,
148
156
task_identifier : event . taskSlug ,
149
157
run_id : event . runId ,
150
- start_time : event . startTime . toString ( ) ,
158
+ start_time : formatClickhouseDate64NanosecondsEpochString ( event . startTime . toString ( ) ) ,
151
159
duration : ( event . duration ?? 0 ) . toString ( ) ,
152
160
trace_id : event . traceId ,
153
161
span_id : event . spanId ,
@@ -177,7 +185,11 @@ export class ClickhouseEventRepository implements IEventRepository {
177
185
}
178
186
179
187
// Only return events where the event start_time is greater than the span start_time
180
- return records . filter ( ( r ) => BigInt ( r . start_time ) > BigInt ( event . startTime ) ) ;
188
+ return records . filter (
189
+ ( r ) =>
190
+ convertClickhouseDate64NanosecondsEpochStringToBigInt ( r . start_time ) >
191
+ BigInt ( event . startTime )
192
+ ) ;
181
193
}
182
194
183
195
private createTaskEventV1InputFromSpanEvent (
@@ -209,7 +221,9 @@ export class ClickhouseEventRepository implements IEventRepository {
209
221
project_id : event . projectId ,
210
222
task_identifier : event . taskSlug ,
211
223
run_id : event . runId ,
212
- start_time : convertDateToNanoseconds ( spanEvent . time ) . toString ( ) ,
224
+ start_time : formatClickhouseDate64NanosecondsEpochString (
225
+ convertDateToNanoseconds ( spanEvent . time ) . toString ( )
226
+ ) ,
213
227
duration : "0" , // Events have no duration
214
228
trace_id : event . traceId ,
215
229
span_id : event . spanId ,
@@ -243,7 +257,9 @@ export class ClickhouseEventRepository implements IEventRepository {
243
257
project_id : event . projectId ,
244
258
task_identifier : event . taskSlug ,
245
259
run_id : event . runId ,
246
- start_time : convertDateToNanoseconds ( spanEvent . time ) . toString ( ) ,
260
+ start_time : formatClickhouseDate64NanosecondsEpochString (
261
+ convertDateToNanoseconds ( spanEvent . time ) . toString ( )
262
+ ) ,
247
263
duration : "0" , // Events have no duration
248
264
trace_id : event . traceId ,
249
265
span_id : event . spanId ,
@@ -271,7 +287,9 @@ export class ClickhouseEventRepository implements IEventRepository {
271
287
project_id : event . projectId ,
272
288
task_identifier : event . taskSlug ,
273
289
run_id : event . runId ,
274
- start_time : convertDateToNanoseconds ( spanEvent . time ) . toString ( ) ,
290
+ start_time : formatClickhouseDate64NanosecondsEpochString (
291
+ convertDateToNanoseconds ( spanEvent . time ) . toString ( )
292
+ ) ,
275
293
duration : "0" , // Events have no duration
276
294
trace_id : event . traceId ,
277
295
span_id : event . spanId ,
@@ -303,7 +321,9 @@ export class ClickhouseEventRepository implements IEventRepository {
303
321
project_id : event . projectId ,
304
322
task_identifier : event . taskSlug ,
305
323
run_id : event . runId ,
306
- start_time : convertDateToNanoseconds ( spanEvent . time ) . toString ( ) ,
324
+ start_time : formatClickhouseDate64NanosecondsEpochString (
325
+ convertDateToNanoseconds ( spanEvent . time ) . toString ( )
326
+ ) ,
307
327
duration : "0" , // Events have no duration
308
328
trace_id : event . traceId ,
309
329
span_id : event . spanId ,
@@ -439,7 +459,7 @@ export class ClickhouseEventRepository implements IEventRepository {
439
459
project_id : options . environment . projectId ,
440
460
task_identifier : options . taskSlug ,
441
461
run_id : options . attributes . runId ,
442
- start_time : startTime . toString ( ) ,
462
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
443
463
duration : duration . toString ( ) ,
444
464
trace_id : traceId ,
445
465
span_id : spanId ,
@@ -540,7 +560,7 @@ export class ClickhouseEventRepository implements IEventRepository {
540
560
project_id : options . environment . projectId ,
541
561
task_identifier : options . taskSlug ,
542
562
run_id : options . attributes . runId ,
543
- start_time : startTime . toString ( ) ,
563
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
544
564
duration : String ( options . incomplete ? 0 : duration ) ,
545
565
trace_id : traceId ,
546
566
span_id : spanId ,
@@ -574,7 +594,7 @@ export class ClickhouseEventRepository implements IEventRepository {
574
594
project_id : options . environment . projectId ,
575
595
task_identifier : options . taskSlug ,
576
596
run_id : options . attributes . runId ,
577
- start_time : startTime . toString ( ) ,
597
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
578
598
duration : String ( options . incomplete ? 0 : duration ) ,
579
599
trace_id : traceId ,
580
600
span_id : spanId ,
@@ -623,7 +643,7 @@ export class ClickhouseEventRepository implements IEventRepository {
623
643
project_id : run . projectId ,
624
644
task_identifier : run . taskIdentifier ,
625
645
run_id : run . friendlyId ,
626
- start_time : startTime . toString ( ) ,
646
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
627
647
duration : calculateDurationFromStart ( startTime , endTime ?? new Date ( ) ) . toString ( ) ,
628
648
trace_id : run . traceId ,
629
649
span_id : run . spanId ,
@@ -671,7 +691,7 @@ export class ClickhouseEventRepository implements IEventRepository {
671
691
project_id : run . projectId ,
672
692
task_identifier : run . taskIdentifier ,
673
693
run_id : blockedRun . friendlyId ,
674
- start_time : startTime . toString ( ) ,
694
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
675
695
duration : calculateDurationFromStart ( startTime , endTime ?? new Date ( ) ) . toString ( ) ,
676
696
trace_id : blockedRun . traceId ,
677
697
span_id : spanId ,
@@ -711,7 +731,7 @@ export class ClickhouseEventRepository implements IEventRepository {
711
731
project_id : run . projectId ,
712
732
task_identifier : run . taskIdentifier ,
713
733
run_id : run . friendlyId ,
714
- start_time : startTime . toString ( ) ,
734
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
715
735
duration : calculateDurationFromStart ( startTime , endTime ?? new Date ( ) ) . toString ( ) ,
716
736
trace_id : run . traceId ,
717
737
span_id : run . spanId ,
@@ -757,7 +777,7 @@ export class ClickhouseEventRepository implements IEventRepository {
757
777
project_id : run . projectId ,
758
778
task_identifier : run . taskIdentifier ,
759
779
run_id : run . friendlyId ,
760
- start_time : startTime . toString ( ) ,
780
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
761
781
duration : calculateDurationFromStart ( startTime , endTime ?? new Date ( ) ) . toString ( ) ,
762
782
trace_id : run . traceId ,
763
783
span_id : run . spanId ,
@@ -803,7 +823,7 @@ export class ClickhouseEventRepository implements IEventRepository {
803
823
project_id : run . projectId ,
804
824
task_identifier : run . taskIdentifier ,
805
825
run_id : run . friendlyId ,
806
- start_time : startTime . toString ( ) ,
826
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
807
827
duration : "0" ,
808
828
trace_id : run . traceId ,
809
829
span_id : run . spanId ,
@@ -847,7 +867,7 @@ export class ClickhouseEventRepository implements IEventRepository {
847
867
project_id : run . projectId ,
848
868
task_identifier : run . taskIdentifier ,
849
869
run_id : run . friendlyId ,
850
- start_time : startTime . toString ( ) ,
870
+ start_time : formatClickhouseDate64NanosecondsEpochString ( startTime . toString ( ) ) ,
851
871
duration : calculateDurationFromStart ( startTime , cancelledAt ) . toString ( ) ,
852
872
trace_id : run . traceId ,
853
873
span_id : run . spanId ,
@@ -1807,3 +1827,17 @@ function isLogEvent(kind: string): boolean {
1807
1827
function calculateEndTimeFromStartTime ( startTime : Date , duration : number ) : Date {
1808
1828
return new Date ( startTime . getTime ( ) + duration / 1_000_000 ) ;
1809
1829
}
1830
+
1831
+ // This will take a string like "1759427319944999936" and return "1759427319.944999936"
1832
+ function formatClickhouseDate64NanosecondsEpochString ( date : string ) : string {
1833
+ if ( date . length !== 19 ) {
1834
+ return date ;
1835
+ }
1836
+
1837
+ return date . substring ( 0 , 10 ) + "." + date . substring ( 10 ) ;
1838
+ }
1839
+
1840
+ function convertClickhouseDate64NanosecondsEpochStringToBigInt ( date : string ) : bigint {
1841
+ const parts = date . split ( "." ) ;
1842
+ return BigInt ( parts . join ( "" ) ) ;
1843
+ }
0 commit comments