arrow_sql_server/write/
profile.rs

1//! Benchmark-only profiling support for the write path.
2
3use std::time::Duration;
4
5#[cfg(feature = "bench-profile")]
6mod enabled {
7    use std::{
8        cell::{Cell, RefCell},
9        time::Duration,
10    };
11
12    use tiberius::BulkLoadStats;
13
14    /// Accumulated timings for the direct raw writer path.
15    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
16    pub struct DirectWriteProfile {
17        /// Time spent measuring runtime batches before encoding.
18        pub measure_batch: Duration,
19        /// Time spent splitting measured batches into bounded row ranges.
20        pub row_range_split: Duration,
21        /// Time spent encoding/appending raw rows before Tiberius writes packets.
22        pub append_encode: Duration,
23        /// Total time spent in raw-row send calls, including append encoding.
24        pub send_total: Duration,
25        /// Number of rows accepted by the direct writer.
26        pub rows: u64,
27        /// Number of batches accepted by the direct writer.
28        pub batches: u64,
29        /// Number of raw row ranges sent.
30        pub row_ranges: u64,
31        /// Total encoded raw row bytes sent or appended.
32        pub encoded_bytes: u64,
33        /// Largest encoded row range in bytes.
34        pub max_row_range_bytes: u64,
35        /// Non-null SQL Server `nvarchar` payload bytes after UTF-16 encoding.
36        pub nvarchar_utf16_bytes: u64,
37        /// Non-null SQL Server `varchar` payload bytes after ASCII validation.
38        pub varchar_bytes: u64,
39        /// Non-null SQL Server `varbinary` payload bytes.
40        pub varbinary_bytes: u64,
41        /// Number of null cells observed by the profiled direct writer path.
42        pub null_cells: u64,
43        /// Number of bulk packet drain attempts inside Tiberius.
44        pub packet_write_calls: u64,
45        /// Number of full TDS bulk-load packets written before finalization.
46        pub packets_written: u64,
47        /// Total packet payload bytes written before finalization.
48        pub packet_payload_bytes: u64,
49        /// Largest full packet payload written before finalization.
50        pub max_packet_payload_bytes: u64,
51        /// Largest buffered bulk-load byte count before a packet drain attempt.
52        pub max_buffered_bytes_before_write: u64,
53        /// Buffered bulk-load tail left after the latest packet drain attempt.
54        pub buffered_bytes_after_last_write: u64,
55        /// Final `EndOfMessage` packet payload bytes written during finalization.
56        pub finalized_packet_payload_bytes: u64,
57        /// Time spent inside bulk-load packet drain attempts.
58        pub bulk_write_packets_elapsed: Duration,
59        /// Number of lower-level connection writes issued by bulk load.
60        pub bulk_write_to_wire_calls: u64,
61        /// Time spent awaiting lower-level connection writes from bulk load.
62        pub bulk_write_to_wire_elapsed: Duration,
63        /// Payload bytes passed to lower-level connection writes from bulk load.
64        pub bulk_write_to_wire_payload_bytes: u64,
65        /// Slowest lower-level connection write awaited by bulk load.
66        pub bulk_max_write_to_wire_elapsed: Duration,
67        /// Largest lower-level connection write payload from bulk load.
68        pub bulk_max_write_to_wire_payload_bytes: u64,
69        /// Number of explicit bulk-load flushes.
70        pub bulk_flush_calls: u64,
71        /// Time spent awaiting explicit bulk-load flushes.
72        pub bulk_flush_elapsed: Duration,
73        /// Slowest explicit bulk-load flush.
74        pub bulk_max_flush_elapsed: Duration,
75        /// Time spent finalizing the bulk-load request.
76        pub bulk_finalize_elapsed: Duration,
77        /// Time spent awaiting the final `EndOfMessage` packet write.
78        pub bulk_finalize_write_to_wire_elapsed: Duration,
79        /// Time spent awaiting the final explicit flush.
80        pub bulk_finalize_flush_elapsed: Duration,
81        /// Time spent waiting for the server result after final bulk flush.
82        pub bulk_finalize_result_elapsed: Duration,
83        /// Number of bulk-load packets passed through the framed connection sink.
84        pub bulk_connection_write_calls: u64,
85        /// Payload bytes passed through the framed connection sink.
86        pub bulk_connection_write_payload_bytes: u64,
87        /// Time spent waiting for the framed connection sink to accept packets.
88        pub bulk_connection_write_ready_elapsed: Duration,
89        /// Time spent encoding packets into the framed connection sink.
90        pub bulk_connection_write_encode_elapsed: Duration,
91        /// Time spent flushing packets through the framed connection sink.
92        pub bulk_connection_write_flush_elapsed: Duration,
93        /// Slowest framed connection sink readiness wait.
94        pub bulk_connection_write_max_ready_elapsed: Duration,
95        /// Slowest packet encode into the framed connection sink.
96        pub bulk_connection_write_max_encode_elapsed: Duration,
97        /// Slowest framed connection sink flush.
98        pub bulk_connection_write_max_flush_elapsed: Duration,
99        /// Largest payload passed through the framed connection sink.
100        pub bulk_connection_write_max_payload_bytes: u64,
101        /// Number of bulk-load packets written through the direct packet path.
102        pub bulk_direct_packet_write_calls: u64,
103        /// Payload bytes written through the direct packet path.
104        pub bulk_direct_packet_payload_bytes: u64,
105        /// Header bytes written through the direct packet path.
106        pub bulk_direct_packet_header_bytes: u64,
107        /// Largest packet payload written through the direct packet path.
108        pub bulk_direct_packet_max_payload_bytes: u64,
109        /// Number of final `EndOfMessage` direct packets.
110        pub bulk_direct_packet_final_calls: u64,
111        /// Payload bytes in final `EndOfMessage` direct packets.
112        pub bulk_direct_packet_final_payload_bytes: u64,
113        /// Header bytes in final `EndOfMessage` direct packets.
114        pub bulk_direct_packet_final_header_bytes: u64,
115        /// Direct packet writes observed on raw, non-TLS streams.
116        pub bulk_direct_packet_raw_stream_calls: u64,
117        /// Direct packet writes observed on TLS streams.
118        pub bulk_direct_packet_tls_stream_calls: u64,
119        /// Low-level write calls issued by the direct packet path.
120        pub bulk_direct_packet_low_level_write_calls: u64,
121        /// Low-level bytes written by the direct packet path.
122        pub bulk_direct_packet_low_level_write_bytes: u64,
123        /// Largest low-level write accepted by the direct packet path.
124        pub bulk_direct_packet_max_low_level_write_bytes: u64,
125        /// Time spent in low-level writes for the direct packet path.
126        pub bulk_direct_packet_write_elapsed: Duration,
127        /// Slowest low-level write in the direct packet path.
128        pub bulk_direct_packet_max_write_elapsed: Duration,
129        /// Low-level write calls used for direct packet headers.
130        pub bulk_direct_packet_header_write_calls: u64,
131        /// Header bytes accepted by low-level writes.
132        pub bulk_direct_packet_header_write_bytes: u64,
133        /// Largest header byte count accepted by one low-level write.
134        pub bulk_direct_packet_header_max_write_bytes: u64,
135        /// Time spent in low-level header writes.
136        pub bulk_direct_packet_header_write_elapsed: Duration,
137        /// Slowest low-level header write.
138        pub bulk_direct_packet_header_max_write_elapsed: Duration,
139        /// Low-level header writes that accepted fewer bytes than remained.
140        pub bulk_direct_packet_header_partial_writes: u64,
141        /// Low-level write calls used for direct packet payloads.
142        pub bulk_direct_packet_payload_write_calls: u64,
143        /// Payload bytes accepted by low-level writes.
144        pub bulk_direct_packet_payload_write_bytes: u64,
145        /// Largest payload byte count accepted by one low-level write.
146        pub bulk_direct_packet_payload_max_write_bytes: u64,
147        /// Time spent in low-level payload writes.
148        pub bulk_direct_packet_payload_write_elapsed: Duration,
149        /// Slowest low-level payload write.
150        pub bulk_direct_packet_payload_max_write_elapsed: Duration,
151        /// Low-level payload writes that accepted fewer bytes than remained.
152        pub bulk_direct_packet_payload_partial_writes: u64,
153        /// Number of low-level `poll_write` attempts.
154        pub bulk_direct_packet_poll_write_polls: u64,
155        /// Number of low-level `poll_write` attempts that returned `Pending`.
156        pub bulk_direct_packet_poll_write_pending_count: u64,
157        /// Time spent waiting after low-level `poll_write` returned `Pending`.
158        pub bulk_direct_packet_poll_write_pending_elapsed: Duration,
159        /// Slowest wait after low-level `poll_write` returned `Pending`.
160        pub bulk_direct_packet_poll_write_max_pending_elapsed: Duration,
161        /// Number of low-level `poll_write` attempts that returned ready.
162        pub bulk_direct_packet_poll_write_ready_count: u64,
163        /// Time spent in ready low-level `poll_write` attempts.
164        pub bulk_direct_packet_poll_write_ready_elapsed: Duration,
165        /// Slowest ready low-level `poll_write` attempt.
166        pub bulk_direct_packet_poll_write_max_ready_elapsed: Duration,
167        /// Explicit flush calls issued by the direct packet path.
168        pub bulk_direct_packet_flush_calls: u64,
169        /// Time spent flushing the direct packet path.
170        pub bulk_direct_packet_flush_elapsed: Duration,
171        /// Slowest flush in the direct packet path.
172        pub bulk_direct_packet_max_flush_elapsed: Duration,
173        /// Number of direct packet flush polls that returned `Pending`.
174        pub bulk_direct_packet_flush_pending_count: u64,
175        /// Time spent waiting after direct packet flush polls returned `Pending`.
176        pub bulk_direct_packet_flush_pending_elapsed: Duration,
177        /// Slowest wait after a direct packet flush poll returned `Pending`.
178        pub bulk_direct_packet_flush_max_pending_elapsed: Duration,
179    }
180
181    impl DirectWriteProfile {
182        /// Returns approximate send time outside the append/encode closure.
183        pub fn send_without_append_encode(&self) -> Duration {
184            self.send_total.saturating_sub(self.append_encode)
185        }
186    }
187
188    thread_local! {
189        static DIRECT_PROFILE: RefCell<Option<DirectWriteProfile>> = const { RefCell::new(None) };
190        static DIRECT_DATE_FAST_PATH_DISABLED: Cell<bool> = const { Cell::new(false) };
191        static DIRECT_FIXED_WIDTH_FAST_PATH_DISABLED: Cell<bool> = const { Cell::new(false) };
192    }
193
194    /// Scoped benchmark override that disables the direct date fast path.
195    #[derive(Debug)]
196    pub struct DirectDateFastPathOverride {
197        previous: bool,
198    }
199
200    /// Scoped benchmark override that disables the whole fixed-width fast path.
201    #[derive(Debug)]
202    pub struct DirectFixedWidthFastPathOverride {
203        previous: bool,
204    }
205
206    impl Drop for DirectDateFastPathOverride {
207        fn drop(&mut self) {
208            DIRECT_DATE_FAST_PATH_DISABLED.with(|disabled| disabled.set(self.previous));
209        }
210    }
211
212    impl Drop for DirectFixedWidthFastPathOverride {
213        fn drop(&mut self) {
214            DIRECT_FIXED_WIDTH_FAST_PATH_DISABLED.with(|disabled| disabled.set(self.previous));
215        }
216    }
217
218    /// Disables direct date fixed-width fast-path encoding for the current scope.
219    pub fn disable_direct_date_fast_path_for_scope() -> DirectDateFastPathOverride {
220        let previous = DIRECT_DATE_FAST_PATH_DISABLED.with(|disabled| {
221            let previous = disabled.get();
222            disabled.set(true);
223            previous
224        });
225
226        DirectDateFastPathOverride { previous }
227    }
228
229    /// Disables direct fixed-width fast-path encoding for the current scope.
230    pub fn disable_direct_fixed_width_fast_path_for_scope() -> DirectFixedWidthFastPathOverride {
231        let previous = DIRECT_FIXED_WIDTH_FAST_PATH_DISABLED.with(|disabled| {
232            let previous = disabled.get();
233            disabled.set(true);
234            previous
235        });
236
237        DirectFixedWidthFastPathOverride { previous }
238    }
239
240    pub(crate) fn direct_date_fast_path_disabled() -> bool {
241        DIRECT_DATE_FAST_PATH_DISABLED.with(Cell::get)
242    }
243
244    pub(crate) fn direct_fixed_width_fast_path_disabled() -> bool {
245        DIRECT_FIXED_WIDTH_FAST_PATH_DISABLED.with(Cell::get)
246    }
247
248    /// Starts direct writer profiling on the current thread.
249    ///
250    /// Existing data for the current thread is cleared.
251    pub fn start_direct_write_profile() {
252        DIRECT_PROFILE.with(|profile| {
253            *profile.borrow_mut() = Some(DirectWriteProfile::default());
254        });
255    }
256
257    /// Stops direct writer profiling on the current thread and returns the data.
258    pub fn finish_direct_write_profile() -> Option<DirectWriteProfile> {
259        DIRECT_PROFILE.with(|profile| profile.borrow_mut().take())
260    }
261
262    pub(crate) fn record_measure_batch(elapsed: Duration) {
263        with_profile(|profile| profile.measure_batch += elapsed);
264    }
265
266    pub(crate) fn record_row_range_split(elapsed: Duration) {
267        with_profile(|profile| profile.row_range_split += elapsed);
268    }
269
270    pub(crate) fn record_append_encode(elapsed: Duration) {
271        with_profile(|profile| profile.append_encode += elapsed);
272    }
273
274    pub(crate) fn record_send_total(elapsed: Duration) {
275        with_profile(|profile| profile.send_total += elapsed);
276    }
277
278    pub(crate) fn record_accepted_batch(rows: usize) {
279        with_profile(|profile| {
280            profile.rows = profile.rows.saturating_add(usize_to_u64_saturating(rows));
281            profile.batches = profile.batches.saturating_add(1);
282        });
283    }
284
285    pub(crate) fn record_row_range(encoded_bytes: usize) {
286        with_profile(|profile| {
287            let encoded_bytes = usize_to_u64_saturating(encoded_bytes);
288            profile.row_ranges = profile.row_ranges.saturating_add(1);
289            profile.encoded_bytes = profile.encoded_bytes.saturating_add(encoded_bytes);
290            profile.max_row_range_bytes = profile.max_row_range_bytes.max(encoded_bytes);
291        });
292    }
293
294    pub(crate) fn record_nvarchar_utf16_bytes(encoded_bytes: usize) {
295        with_profile(|profile| {
296            profile.nvarchar_utf16_bytes = profile
297                .nvarchar_utf16_bytes
298                .saturating_add(usize_to_u64_saturating(encoded_bytes));
299        });
300    }
301
302    pub(crate) fn record_varchar_bytes(encoded_bytes: usize) {
303        with_profile(|profile| {
304            profile.varchar_bytes = profile
305                .varchar_bytes
306                .saturating_add(usize_to_u64_saturating(encoded_bytes));
307        });
308    }
309
310    pub(crate) fn record_varbinary_bytes(encoded_bytes: usize) {
311        with_profile(|profile| {
312            profile.varbinary_bytes = profile
313                .varbinary_bytes
314                .saturating_add(usize_to_u64_saturating(encoded_bytes));
315        });
316    }
317
318    pub(crate) fn record_null_cell() {
319        with_profile(|profile| {
320            profile.null_cells = profile.null_cells.saturating_add(1);
321        });
322    }
323
324    pub(crate) fn record_bulk_load_stats(stats: BulkLoadStats) {
325        let packet = stats.packet;
326        let timing = stats.write_timing;
327        let connection = timing.connection_write;
328        let direct = timing.direct_packet_write;
329
330        with_profile(|profile| {
331            profile.packet_write_calls = profile
332                .packet_write_calls
333                .saturating_add(packet.write_packets_calls);
334            profile.packets_written = profile
335                .packets_written
336                .saturating_add(packet.packets_written);
337            profile.packet_payload_bytes = profile
338                .packet_payload_bytes
339                .saturating_add(packet.packet_payload_bytes);
340            profile.max_packet_payload_bytes = profile
341                .max_packet_payload_bytes
342                .max(usize_to_u64_saturating(packet.max_packet_payload_bytes));
343            profile.max_buffered_bytes_before_write =
344                profile
345                    .max_buffered_bytes_before_write
346                    .max(usize_to_u64_saturating(
347                        packet.max_buffered_bytes_before_write,
348                    ));
349            profile.buffered_bytes_after_last_write =
350                usize_to_u64_saturating(packet.buffered_bytes_after_last_write);
351            profile.finalized_packet_payload_bytes = profile
352                .finalized_packet_payload_bytes
353                .saturating_add(usize_to_u64_saturating(
354                    packet.finalized_packet_payload_bytes,
355                ));
356            profile.bulk_write_packets_elapsed += timing.write_packets_elapsed;
357            profile.bulk_write_to_wire_calls = profile
358                .bulk_write_to_wire_calls
359                .saturating_add(timing.write_to_wire_calls);
360            profile.bulk_write_to_wire_elapsed += timing.write_to_wire_elapsed;
361            profile.bulk_write_to_wire_payload_bytes = profile
362                .bulk_write_to_wire_payload_bytes
363                .saturating_add(timing.write_to_wire_payload_bytes);
364            profile.bulk_max_write_to_wire_elapsed = profile
365                .bulk_max_write_to_wire_elapsed
366                .max(timing.max_write_to_wire_elapsed);
367            profile.bulk_max_write_to_wire_payload_bytes = profile
368                .bulk_max_write_to_wire_payload_bytes
369                .max(usize_to_u64_saturating(
370                    timing.max_write_to_wire_payload_bytes,
371                ));
372            profile.bulk_flush_calls = profile.bulk_flush_calls.saturating_add(timing.flush_calls);
373            profile.bulk_flush_elapsed += timing.flush_elapsed;
374            profile.bulk_max_flush_elapsed =
375                profile.bulk_max_flush_elapsed.max(timing.max_flush_elapsed);
376            profile.bulk_finalize_elapsed += timing.finalize_elapsed;
377            profile.bulk_finalize_write_to_wire_elapsed += timing.finalize_write_to_wire_elapsed;
378            profile.bulk_finalize_flush_elapsed += timing.finalize_flush_elapsed;
379            profile.bulk_finalize_result_elapsed += timing.finalize_result_elapsed;
380            profile.bulk_connection_write_calls = profile
381                .bulk_connection_write_calls
382                .saturating_add(connection.calls);
383            profile.bulk_connection_write_payload_bytes = profile
384                .bulk_connection_write_payload_bytes
385                .saturating_add(connection.payload_bytes);
386            profile.bulk_connection_write_ready_elapsed += connection.ready_elapsed;
387            profile.bulk_connection_write_encode_elapsed += connection.encode_elapsed;
388            profile.bulk_connection_write_flush_elapsed += connection.flush_elapsed;
389            profile.bulk_connection_write_max_ready_elapsed = profile
390                .bulk_connection_write_max_ready_elapsed
391                .max(connection.max_ready_elapsed);
392            profile.bulk_connection_write_max_encode_elapsed = profile
393                .bulk_connection_write_max_encode_elapsed
394                .max(connection.max_encode_elapsed);
395            profile.bulk_connection_write_max_flush_elapsed = profile
396                .bulk_connection_write_max_flush_elapsed
397                .max(connection.max_flush_elapsed);
398            profile.bulk_connection_write_max_payload_bytes = profile
399                .bulk_connection_write_max_payload_bytes
400                .max(usize_to_u64_saturating(connection.max_payload_bytes));
401            profile.bulk_direct_packet_write_calls = profile
402                .bulk_direct_packet_write_calls
403                .saturating_add(direct.calls);
404            profile.bulk_direct_packet_payload_bytes = profile
405                .bulk_direct_packet_payload_bytes
406                .saturating_add(direct.payload_bytes);
407            profile.bulk_direct_packet_header_bytes = profile
408                .bulk_direct_packet_header_bytes
409                .saturating_add(direct.header_bytes);
410            profile.bulk_direct_packet_max_payload_bytes = profile
411                .bulk_direct_packet_max_payload_bytes
412                .max(usize_to_u64_saturating(direct.max_payload_bytes));
413            profile.bulk_direct_packet_final_calls = profile
414                .bulk_direct_packet_final_calls
415                .saturating_add(direct.final_calls);
416            profile.bulk_direct_packet_final_payload_bytes = profile
417                .bulk_direct_packet_final_payload_bytes
418                .saturating_add(direct.final_payload_bytes);
419            profile.bulk_direct_packet_final_header_bytes = profile
420                .bulk_direct_packet_final_header_bytes
421                .saturating_add(direct.final_header_bytes);
422            profile.bulk_direct_packet_raw_stream_calls = profile
423                .bulk_direct_packet_raw_stream_calls
424                .saturating_add(direct.raw_stream_calls);
425            profile.bulk_direct_packet_tls_stream_calls = profile
426                .bulk_direct_packet_tls_stream_calls
427                .saturating_add(direct.tls_stream_calls);
428            profile.bulk_direct_packet_low_level_write_calls = profile
429                .bulk_direct_packet_low_level_write_calls
430                .saturating_add(direct.write_calls);
431            profile.bulk_direct_packet_low_level_write_bytes = profile
432                .bulk_direct_packet_low_level_write_bytes
433                .saturating_add(direct.write_bytes);
434            profile.bulk_direct_packet_max_low_level_write_bytes = profile
435                .bulk_direct_packet_max_low_level_write_bytes
436                .max(usize_to_u64_saturating(direct.max_write_bytes));
437            profile.bulk_direct_packet_write_elapsed += direct.write_elapsed;
438            profile.bulk_direct_packet_max_write_elapsed = profile
439                .bulk_direct_packet_max_write_elapsed
440                .max(direct.max_write_elapsed);
441            profile.bulk_direct_packet_header_write_calls = profile
442                .bulk_direct_packet_header_write_calls
443                .saturating_add(direct.header_write_calls);
444            profile.bulk_direct_packet_header_write_bytes = profile
445                .bulk_direct_packet_header_write_bytes
446                .saturating_add(direct.header_write_bytes);
447            profile.bulk_direct_packet_header_max_write_bytes = profile
448                .bulk_direct_packet_header_max_write_bytes
449                .max(usize_to_u64_saturating(direct.header_max_write_bytes));
450            profile.bulk_direct_packet_header_write_elapsed += direct.header_write_elapsed;
451            profile.bulk_direct_packet_header_max_write_elapsed = profile
452                .bulk_direct_packet_header_max_write_elapsed
453                .max(direct.header_max_write_elapsed);
454            profile.bulk_direct_packet_header_partial_writes = profile
455                .bulk_direct_packet_header_partial_writes
456                .saturating_add(direct.header_partial_writes);
457            profile.bulk_direct_packet_payload_write_calls = profile
458                .bulk_direct_packet_payload_write_calls
459                .saturating_add(direct.payload_write_calls);
460            profile.bulk_direct_packet_payload_write_bytes = profile
461                .bulk_direct_packet_payload_write_bytes
462                .saturating_add(direct.payload_write_bytes);
463            profile.bulk_direct_packet_payload_max_write_bytes = profile
464                .bulk_direct_packet_payload_max_write_bytes
465                .max(usize_to_u64_saturating(direct.payload_max_write_bytes));
466            profile.bulk_direct_packet_payload_write_elapsed += direct.payload_write_elapsed;
467            profile.bulk_direct_packet_payload_max_write_elapsed = profile
468                .bulk_direct_packet_payload_max_write_elapsed
469                .max(direct.payload_max_write_elapsed);
470            profile.bulk_direct_packet_payload_partial_writes = profile
471                .bulk_direct_packet_payload_partial_writes
472                .saturating_add(direct.payload_partial_writes);
473            profile.bulk_direct_packet_poll_write_polls = profile
474                .bulk_direct_packet_poll_write_polls
475                .saturating_add(direct.poll_write_polls);
476            profile.bulk_direct_packet_poll_write_pending_count = profile
477                .bulk_direct_packet_poll_write_pending_count
478                .saturating_add(direct.poll_write_pending_count);
479            profile.bulk_direct_packet_poll_write_pending_elapsed +=
480                direct.poll_write_pending_elapsed;
481            profile.bulk_direct_packet_poll_write_max_pending_elapsed = profile
482                .bulk_direct_packet_poll_write_max_pending_elapsed
483                .max(direct.poll_write_max_pending_elapsed);
484            profile.bulk_direct_packet_poll_write_ready_count = profile
485                .bulk_direct_packet_poll_write_ready_count
486                .saturating_add(direct.poll_write_ready_count);
487            profile.bulk_direct_packet_poll_write_ready_elapsed += direct.poll_write_ready_elapsed;
488            profile.bulk_direct_packet_poll_write_max_ready_elapsed = profile
489                .bulk_direct_packet_poll_write_max_ready_elapsed
490                .max(direct.poll_write_max_ready_elapsed);
491            profile.bulk_direct_packet_flush_calls = profile
492                .bulk_direct_packet_flush_calls
493                .saturating_add(direct.flush_calls);
494            profile.bulk_direct_packet_flush_elapsed += direct.flush_elapsed;
495            profile.bulk_direct_packet_max_flush_elapsed = profile
496                .bulk_direct_packet_max_flush_elapsed
497                .max(direct.max_flush_elapsed);
498            profile.bulk_direct_packet_flush_pending_count = profile
499                .bulk_direct_packet_flush_pending_count
500                .saturating_add(direct.flush_pending_count);
501            profile.bulk_direct_packet_flush_pending_elapsed += direct.flush_pending_elapsed;
502            profile.bulk_direct_packet_flush_max_pending_elapsed = profile
503                .bulk_direct_packet_flush_max_pending_elapsed
504                .max(direct.flush_max_pending_elapsed);
505        });
506    }
507
508    fn with_profile(update: impl FnOnce(&mut DirectWriteProfile)) {
509        DIRECT_PROFILE.with(|profile| {
510            if let Some(profile) = profile.borrow_mut().as_mut() {
511                update(profile);
512            }
513        });
514    }
515
516    fn usize_to_u64_saturating(value: usize) -> u64 {
517        u64::try_from(value).unwrap_or(u64::MAX)
518    }
519}
520
521#[cfg(not(feature = "bench-profile"))]
522mod disabled {
523    use std::time::Duration;
524
525    pub(crate) fn record_measure_batch(_elapsed: Duration) {}
526
527    pub(crate) fn record_row_range_split(_elapsed: Duration) {}
528
529    pub(crate) fn record_append_encode(_elapsed: Duration) {}
530
531    pub(crate) fn record_send_total(_elapsed: Duration) {}
532
533    pub(crate) fn record_accepted_batch(_rows: usize) {}
534
535    pub(crate) fn record_row_range(_encoded_bytes: usize) {}
536
537    pub(crate) fn record_nvarchar_utf16_bytes(_encoded_bytes: usize) {}
538
539    pub(crate) fn record_varchar_bytes(_encoded_bytes: usize) {}
540
541    pub(crate) fn record_varbinary_bytes(_encoded_bytes: usize) {}
542
543    pub(crate) fn record_null_cell() {}
544
545    pub(crate) fn direct_date_fast_path_disabled() -> bool {
546        false
547    }
548
549    pub(crate) fn direct_fixed_width_fast_path_disabled() -> bool {
550        false
551    }
552}
553
554#[cfg(not(feature = "bench-profile"))]
555pub(crate) use disabled::*;
556#[cfg(feature = "bench-profile")]
557pub use enabled::{
558    DirectDateFastPathOverride, DirectFixedWidthFastPathOverride, DirectWriteProfile,
559    disable_direct_date_fast_path_for_scope, disable_direct_fixed_width_fast_path_for_scope,
560    finish_direct_write_profile, start_direct_write_profile,
561};
562#[cfg(feature = "bench-profile")]
563pub(crate) use enabled::{
564    direct_date_fast_path_disabled, direct_fixed_width_fast_path_disabled, record_accepted_batch,
565    record_append_encode, record_bulk_load_stats, record_measure_batch, record_null_cell,
566    record_nvarchar_utf16_bytes, record_row_range, record_row_range_split, record_send_total,
567    record_varbinary_bytes, record_varchar_bytes,
568};
569
570pub(crate) fn record_elapsed<T>(start: std::time::Instant, record: fn(Duration), value: T) -> T {
571    record(start.elapsed());
572    value
573}
574
575#[cfg(all(test, feature = "bench-profile"))]
576mod tests {
577    use std::{thread, time::Duration};
578
579    #[test]
580    fn direct_profile_accumulates_and_resets_current_thread_data() {
581        super::start_direct_write_profile();
582        super::record_measure_batch(Duration::from_millis(2));
583        super::record_row_range_split(Duration::from_millis(3));
584        super::record_append_encode(Duration::from_millis(5));
585        super::record_send_total(Duration::from_millis(13));
586        super::record_accepted_batch(7);
587        super::record_row_range(11);
588        super::record_nvarchar_utf16_bytes(17);
589        super::record_varchar_bytes(18);
590        super::record_varbinary_bytes(19);
591        super::record_null_cell();
592        super::record_bulk_load_stats(tiberius::BulkLoadStats {
593            packet: tiberius::BulkLoadPacketStats {
594                write_packets_calls: 23,
595                packets_written: 29,
596                packet_payload_bytes: 31,
597                max_packet_payload_bytes: 37,
598                max_buffered_bytes_before_write: 41,
599                buffered_bytes_after_last_write: 43,
600                finalized_packet_payload_bytes: 47,
601            },
602            write_timing: tiberius::BulkLoadWriteTimingStats {
603                write_packets_elapsed: Duration::from_millis(53),
604                write_to_wire_calls: 59,
605                write_to_wire_elapsed: Duration::from_millis(61),
606                write_to_wire_payload_bytes: 67,
607                max_write_to_wire_elapsed: Duration::from_millis(71),
608                max_write_to_wire_payload_bytes: 73,
609                flush_calls: 79,
610                flush_elapsed: Duration::from_millis(83),
611                max_flush_elapsed: Duration::from_millis(89),
612                finalize_elapsed: Duration::from_millis(97),
613                finalize_write_to_wire_elapsed: Duration::from_millis(101),
614                finalize_flush_elapsed: Duration::from_millis(103),
615                finalize_result_elapsed: Duration::from_millis(107),
616                connection_write: tiberius::BulkLoadConnectionWriteStats {
617                    calls: 109,
618                    payload_bytes: 113,
619                    ready_elapsed: Duration::from_millis(127),
620                    encode_elapsed: Duration::from_millis(131),
621                    flush_elapsed: Duration::from_millis(137),
622                    max_ready_elapsed: Duration::from_millis(139),
623                    max_encode_elapsed: Duration::from_millis(149),
624                    max_flush_elapsed: Duration::from_millis(151),
625                    max_payload_bytes: 157,
626                },
627                direct_packet_write: tiberius::BulkLoadDirectPacketWriteStats {
628                    calls: 163,
629                    payload_bytes: 167,
630                    header_bytes: 173,
631                    max_payload_bytes: 179,
632                    final_calls: 181,
633                    final_payload_bytes: 191,
634                    final_header_bytes: 193,
635                    raw_stream_calls: 197,
636                    tls_stream_calls: 199,
637                    write_calls: 179,
638                    write_bytes: 181,
639                    max_write_bytes: 191,
640                    write_elapsed: Duration::from_millis(193),
641                    max_write_elapsed: Duration::from_millis(197),
642                    header_write_calls: 211,
643                    header_write_bytes: 223,
644                    header_max_write_bytes: 227,
645                    header_write_elapsed: Duration::from_millis(229),
646                    header_max_write_elapsed: Duration::from_millis(233),
647                    header_partial_writes: 239,
648                    payload_write_calls: 241,
649                    payload_write_bytes: 251,
650                    payload_max_write_bytes: 257,
651                    payload_write_elapsed: Duration::from_millis(263),
652                    payload_max_write_elapsed: Duration::from_millis(269),
653                    payload_partial_writes: 271,
654                    poll_write_polls: 277,
655                    poll_write_pending_count: 281,
656                    poll_write_pending_elapsed: Duration::from_millis(283),
657                    poll_write_max_pending_elapsed: Duration::from_millis(293),
658                    poll_write_ready_count: 307,
659                    poll_write_ready_elapsed: Duration::from_millis(311),
660                    poll_write_max_ready_elapsed: Duration::from_millis(313),
661                    flush_calls: 199,
662                    flush_elapsed: Duration::from_millis(211),
663                    max_flush_elapsed: Duration::from_millis(223),
664                    flush_pending_count: 317,
665                    flush_pending_elapsed: Duration::from_millis(331),
666                    flush_max_pending_elapsed: Duration::from_millis(337),
667                },
668            },
669        });
670
671        let profile = super::finish_direct_write_profile().unwrap();
672
673        assert_eq!(profile.measure_batch, Duration::from_millis(2));
674        assert_eq!(profile.row_range_split, Duration::from_millis(3));
675        assert_eq!(profile.append_encode, Duration::from_millis(5));
676        assert_eq!(profile.send_total, Duration::from_millis(13));
677        assert_eq!(
678            profile.send_without_append_encode(),
679            Duration::from_millis(8)
680        );
681        assert_eq!(profile.rows, 7);
682        assert_eq!(profile.batches, 1);
683        assert_eq!(profile.row_ranges, 1);
684        assert_eq!(profile.encoded_bytes, 11);
685        assert_eq!(profile.max_row_range_bytes, 11);
686        assert_eq!(profile.nvarchar_utf16_bytes, 17);
687        assert_eq!(profile.varchar_bytes, 18);
688        assert_eq!(profile.varbinary_bytes, 19);
689        assert_eq!(profile.null_cells, 1);
690        assert_eq!(profile.packet_write_calls, 23);
691        assert_eq!(profile.packets_written, 29);
692        assert_eq!(profile.packet_payload_bytes, 31);
693        assert_eq!(profile.max_packet_payload_bytes, 37);
694        assert_eq!(profile.max_buffered_bytes_before_write, 41);
695        assert_eq!(profile.buffered_bytes_after_last_write, 43);
696        assert_eq!(profile.finalized_packet_payload_bytes, 47);
697        assert_eq!(
698            profile.bulk_write_packets_elapsed,
699            Duration::from_millis(53)
700        );
701        assert_eq!(profile.bulk_write_to_wire_calls, 59);
702        assert_eq!(
703            profile.bulk_write_to_wire_elapsed,
704            Duration::from_millis(61)
705        );
706        assert_eq!(profile.bulk_write_to_wire_payload_bytes, 67);
707        assert_eq!(
708            profile.bulk_max_write_to_wire_elapsed,
709            Duration::from_millis(71)
710        );
711        assert_eq!(profile.bulk_max_write_to_wire_payload_bytes, 73);
712        assert_eq!(profile.bulk_flush_calls, 79);
713        assert_eq!(profile.bulk_flush_elapsed, Duration::from_millis(83));
714        assert_eq!(profile.bulk_max_flush_elapsed, Duration::from_millis(89));
715        assert_eq!(profile.bulk_finalize_elapsed, Duration::from_millis(97));
716        assert_eq!(
717            profile.bulk_finalize_write_to_wire_elapsed,
718            Duration::from_millis(101)
719        );
720        assert_eq!(
721            profile.bulk_finalize_flush_elapsed,
722            Duration::from_millis(103)
723        );
724        assert_eq!(
725            profile.bulk_finalize_result_elapsed,
726            Duration::from_millis(107)
727        );
728        assert_eq!(profile.bulk_connection_write_calls, 109);
729        assert_eq!(profile.bulk_connection_write_payload_bytes, 113);
730        assert_eq!(
731            profile.bulk_connection_write_ready_elapsed,
732            Duration::from_millis(127)
733        );
734        assert_eq!(
735            profile.bulk_connection_write_encode_elapsed,
736            Duration::from_millis(131)
737        );
738        assert_eq!(
739            profile.bulk_connection_write_flush_elapsed,
740            Duration::from_millis(137)
741        );
742        assert_eq!(
743            profile.bulk_connection_write_max_ready_elapsed,
744            Duration::from_millis(139)
745        );
746        assert_eq!(
747            profile.bulk_connection_write_max_encode_elapsed,
748            Duration::from_millis(149)
749        );
750        assert_eq!(
751            profile.bulk_connection_write_max_flush_elapsed,
752            Duration::from_millis(151)
753        );
754        assert_eq!(profile.bulk_connection_write_max_payload_bytes, 157);
755        assert_eq!(profile.bulk_direct_packet_write_calls, 163);
756        assert_eq!(profile.bulk_direct_packet_payload_bytes, 167);
757        assert_eq!(profile.bulk_direct_packet_header_bytes, 173);
758        assert_eq!(profile.bulk_direct_packet_max_payload_bytes, 179);
759        assert_eq!(profile.bulk_direct_packet_final_calls, 181);
760        assert_eq!(profile.bulk_direct_packet_final_payload_bytes, 191);
761        assert_eq!(profile.bulk_direct_packet_final_header_bytes, 193);
762        assert_eq!(profile.bulk_direct_packet_raw_stream_calls, 197);
763        assert_eq!(profile.bulk_direct_packet_tls_stream_calls, 199);
764        assert_eq!(profile.bulk_direct_packet_low_level_write_calls, 179);
765        assert_eq!(profile.bulk_direct_packet_low_level_write_bytes, 181);
766        assert_eq!(profile.bulk_direct_packet_max_low_level_write_bytes, 191);
767        assert_eq!(
768            profile.bulk_direct_packet_write_elapsed,
769            Duration::from_millis(193)
770        );
771        assert_eq!(
772            profile.bulk_direct_packet_max_write_elapsed,
773            Duration::from_millis(197)
774        );
775        assert_eq!(profile.bulk_direct_packet_header_write_calls, 211);
776        assert_eq!(profile.bulk_direct_packet_header_write_bytes, 223);
777        assert_eq!(profile.bulk_direct_packet_header_max_write_bytes, 227);
778        assert_eq!(
779            profile.bulk_direct_packet_header_write_elapsed,
780            Duration::from_millis(229)
781        );
782        assert_eq!(
783            profile.bulk_direct_packet_header_max_write_elapsed,
784            Duration::from_millis(233)
785        );
786        assert_eq!(profile.bulk_direct_packet_header_partial_writes, 239);
787        assert_eq!(profile.bulk_direct_packet_payload_write_calls, 241);
788        assert_eq!(profile.bulk_direct_packet_payload_write_bytes, 251);
789        assert_eq!(profile.bulk_direct_packet_payload_max_write_bytes, 257);
790        assert_eq!(
791            profile.bulk_direct_packet_payload_write_elapsed,
792            Duration::from_millis(263)
793        );
794        assert_eq!(
795            profile.bulk_direct_packet_payload_max_write_elapsed,
796            Duration::from_millis(269)
797        );
798        assert_eq!(profile.bulk_direct_packet_payload_partial_writes, 271);
799        assert_eq!(profile.bulk_direct_packet_poll_write_polls, 277);
800        assert_eq!(profile.bulk_direct_packet_poll_write_pending_count, 281);
801        assert_eq!(
802            profile.bulk_direct_packet_poll_write_pending_elapsed,
803            Duration::from_millis(283)
804        );
805        assert_eq!(
806            profile.bulk_direct_packet_poll_write_max_pending_elapsed,
807            Duration::from_millis(293)
808        );
809        assert_eq!(profile.bulk_direct_packet_poll_write_ready_count, 307);
810        assert_eq!(
811            profile.bulk_direct_packet_poll_write_ready_elapsed,
812            Duration::from_millis(311)
813        );
814        assert_eq!(
815            profile.bulk_direct_packet_poll_write_max_ready_elapsed,
816            Duration::from_millis(313)
817        );
818        assert_eq!(profile.bulk_direct_packet_flush_calls, 199);
819        assert_eq!(
820            profile.bulk_direct_packet_flush_elapsed,
821            Duration::from_millis(211)
822        );
823        assert_eq!(
824            profile.bulk_direct_packet_max_flush_elapsed,
825            Duration::from_millis(223)
826        );
827        assert_eq!(profile.bulk_direct_packet_flush_pending_count, 317);
828        assert_eq!(
829            profile.bulk_direct_packet_flush_pending_elapsed,
830            Duration::from_millis(331)
831        );
832        assert_eq!(
833            profile.bulk_direct_packet_flush_max_pending_elapsed,
834            Duration::from_millis(337)
835        );
836        assert!(super::finish_direct_write_profile().is_none());
837    }
838
839    #[test]
840    fn scoped_fast_path_overrides_do_not_cross_threads() -> Result<(), String> {
841        assert!(!super::direct_date_fast_path_disabled());
842        assert!(!super::direct_fixed_width_fast_path_disabled());
843
844        let _date_guard = super::disable_direct_date_fast_path_for_scope();
845        let _fixed_width_guard = super::disable_direct_fixed_width_fast_path_for_scope();
846
847        assert!(super::direct_date_fast_path_disabled());
848        assert!(super::direct_fixed_width_fast_path_disabled());
849
850        let handle = thread::spawn(|| {
851            (
852                super::direct_date_fast_path_disabled(),
853                super::direct_fixed_width_fast_path_disabled(),
854            )
855        });
856        let thread_state = handle
857            .join()
858            .map_err(|_| "fast path override thread panicked".to_owned())?;
859
860        assert_eq!(thread_state, (false, false));
861        Ok(())
862    }
863
864    #[test]
865    fn scoped_fast_path_overrides_restore_previous_thread_state() {
866        assert!(!super::direct_date_fast_path_disabled());
867        assert!(!super::direct_fixed_width_fast_path_disabled());
868
869        {
870            let _date_guard = super::disable_direct_date_fast_path_for_scope();
871            let _fixed_width_guard = super::disable_direct_fixed_width_fast_path_for_scope();
872
873            assert!(super::direct_date_fast_path_disabled());
874            assert!(super::direct_fixed_width_fast_path_disabled());
875        }
876
877        assert!(!super::direct_date_fast_path_disabled());
878        assert!(!super::direct_fixed_width_fast_path_disabled());
879    }
880}
arrow_sql_server/write/profile.rs

arrow_sql_server/write/
profile.rs
