faucet_core/
forward.rs

1//! Forwarding by construction for [`Source`](crate::Source) and
2//! [`Sink`](crate::Sink) wrappers (RFC 0013, #877).
3//!
4//! A wrapper (a decorator that holds an inner source or sink) must forward
5//! every defaulted hook, or the hook silently reverts to its trait default for
6//! everything behind the wrapper — exactly-once, upsert, the columnar fast
7//! path, multi-table routing. [`forward_source!`](crate::forward_source) and
8//! [`forward_sink!`](crate::forward_sink) generate the forwarding body for
9//! every hook, so a wrapper writes only the methods it changes and a hook
10//! added to the trait later reaches the inner connector without touching the
11//! wrapper.
12//!
13//! Both macros are invoked inside the wrapper's `impl` block (with or without
14//! `#[async_trait]`):
15//!
16//! ```text
17//! forward_sink!(inner);                                  // forward everything
18//! forward_sink!(inner; except write_batch, flush);       // the wrapper defines these
19//! ```
20//!
21//! `inner` names the field holding the wrapped connector: any type whose
22//! methods resolve to a `Source`/`Sink` (a `Box<dyn Sink>`, an `Arc<S>`, a
23//! `&'a S`, a concrete connector), or a tuple index such as `0`.
24//!
25//! Every hook listed after `except` is left out: the wrapper either defines it
26//! itself or deliberately takes the trait default. A name that is not a hook
27//! of the trait is a compile error, and defining a hook without listing it is
28//! a duplicate-definition error, so the list stays exact.
29//!
30//! Forwarding is verbatim, including the convenience hooks whose default
31//! delegates to another hook (`fetch_all`, `stream_pages`, `check`,
32//! `write_batch_partial`, `write_batch_idempotent`, …). A wrapper that changes
33//! a hook must also list the hooks that should go through its change rather
34//! than straight to the inner connector — a sink that rewrites records in
35//! `write_batch` lists `write_batch_partial`, `write_batch_idempotent`,
36//! `write_batch_columnar` and `load_native` too. `faucet-conformance`'s
37//! `assert_source_forwards_every_hook` / `assert_sink_forwards_every_hook`
38//! check the result.
39
40/// Paths the forwarding macros expand to. Not part of the public API.
41#[doc(hidden)]
42pub mod __forward {
43    #[cfg(feature = "arrow")]
44    pub use arrow;
45    pub use chrono;
46    /// Whether this build of `faucet-core` has the `arrow` hooks.
47    pub const ARROW: bool = cfg!(feature = "arrow");
48
49    /// The future an `#[async_trait]` hook returns.
50    pub type Fut<'a, T> = ::core::pin::Pin<Box<dyn ::core::future::Future<Output = T> + Send + 'a>>;
51    /// A hook's fallible result.
52    pub type Res<T> = Result<T, crate::FaucetError>;
53    /// The parent-record context a fetch receives.
54    pub type Ctx = std::collections::HashMap<String, crate::Value>;
55    /// A page stream a source returns.
56    pub type Pages<'a, T> = ::core::pin::Pin<Box<dyn crate::Stream<Item = Res<T>> + Send + 'a>>;
57}
58
59/// Emits its input only when `faucet-core` was built with `arrow`.
60#[cfg(feature = "arrow")]
61#[doc(hidden)]
62#[macro_export]
63macro_rules! __faucet_if_arrow {
64    ($($t:tt)*) => { $($t)* };
65}
66
67/// Emits its input only when `faucet-core` was built with `arrow`.
68#[cfg(not(feature = "arrow"))]
69#[doc(hidden)]
70#[macro_export]
71macro_rules! __faucet_if_arrow {
72    ($($t:tt)*) => {};
73}
74
75/// Generate the forwarding body of every [`Source`](crate::Source) hook
76/// inside a wrapper's `impl Source for …` block.
77///
78/// `forward_source!(field)` forwards every hook to `self.field`;
79/// `forward_source!(field; except a, b)` leaves out the hooks the wrapper
80/// defines itself (or deliberately leaves at the trait default). See the
81/// [module docs](crate::forward) for the rules.
82///
83/// ```
84/// use faucet_core::{FaucetError, Source, Value, async_trait, forward_source};
85/// use std::collections::HashMap;
86///
87/// /// Stamps a fixed state key on any source; everything else reaches `inner`.
88/// struct Keyed {
89///     inner: Box<dyn Source>,
90///     key: String,
91/// }
92///
93/// #[async_trait]
94/// impl Source for Keyed {
95///     fn state_key(&self) -> Option<String> {
96///         Some(self.key.clone())
97///     }
98///     forward_source!(inner; except state_key);
99/// }
100/// # struct Empty;
101/// # #[async_trait]
102/// # impl Source for Empty {
103/// #     async fn fetch_with_context(
104/// #         &self,
105/// #         _: &HashMap<String, Value>,
106/// #     ) -> Result<Vec<Value>, FaucetError> {
107/// #         Ok(vec![])
108/// #     }
109/// # }
110/// # let s = Keyed { inner: Box::new(Empty), key: "k".into() };
111/// # assert_eq!(s.state_key().as_deref(), Some("k"));
112/// ```
113#[macro_export]
114macro_rules! forward_source {
115    ($inner:tt) => { $crate::forward_source!($inner; except); };
116    ($inner:tt; except $($skip:ident),* $(,)?) => {
117        $crate::__faucet_forward_known!(@source [$($skip)*]);
118        $crate::__faucet_forward_filter!(fetch_with_context [$($skip)*] {
119            fn fetch_with_context<'life0, 'life1, 'async_trait>(
120                &'life0 self,
121                context: &'life1 $crate::forward::__forward::Ctx,
122            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::std::vec::Vec<$crate::Value>>>
123            where
124                'life0: 'async_trait,
125                'life1: 'async_trait,
126                Self: 'async_trait,
127            {
128                #[allow(unused_imports)]
129                use $crate::Source as _;
130                self.$inner.fetch_with_context(context)
131            }
132        });
133        $crate::__faucet_forward_filter!(fetch_all [$($skip)*] {
134            fn fetch_all<'life0, 'async_trait>(
135                &'life0 self,
136            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::std::vec::Vec<$crate::Value>>>
137            where
138                'life0: 'async_trait,
139                Self: 'async_trait,
140            {
141                #[allow(unused_imports)]
142                use $crate::Source as _;
143                self.$inner.fetch_all()
144            }
145        });
146        $crate::__faucet_forward_filter!(fetch_with_context_incremental [$($skip)*] {
147            fn fetch_with_context_incremental<'life0, 'life1, 'async_trait>(
148                &'life0 self,
149                context: &'life1 $crate::forward::__forward::Ctx,
150            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<(::std::vec::Vec<$crate::Value>, ::core::option::Option<$crate::Value>)>>
151            where
152                'life0: 'async_trait,
153                'life1: 'async_trait,
154                Self: 'async_trait,
155            {
156                #[allow(unused_imports)]
157                use $crate::Source as _;
158                self.$inner.fetch_with_context_incremental(context)
159            }
160        });
161        $crate::__faucet_forward_filter!(fetch_all_incremental [$($skip)*] {
162            fn fetch_all_incremental<'life0, 'async_trait>(
163                &'life0 self,
164            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<(::std::vec::Vec<$crate::Value>, ::core::option::Option<$crate::Value>)>>
165            where
166                'life0: 'async_trait,
167                Self: 'async_trait,
168            {
169                #[allow(unused_imports)]
170                use $crate::Source as _;
171                self.$inner.fetch_all_incremental()
172            }
173        });
174        $crate::__faucet_forward_filter!(stream_pages [$($skip)*] {
175            fn stream_pages<'a>(
176                &'a self,
177                context: &'a $crate::forward::__forward::Ctx,
178                batch_size: usize,
179            ) -> $crate::forward::__forward::Pages<'a, $crate::pipeline::StreamPage>
180            {
181                #[allow(unused_imports)]
182                use $crate::Source as _;
183                self.$inner.stream_pages(context, batch_size)
184            }
185        });
186        $crate::__faucet_if_arrow! {
187            $crate::__faucet_forward_filter!(supports_columnar [$($skip)*] {
188                fn supports_columnar(&self) -> bool
189                {
190                    #[allow(unused_imports)]
191                    use $crate::Source as _;
192                    self.$inner.supports_columnar()
193                }
194            });
195        }
196        $crate::__faucet_if_arrow! {
197            $crate::__faucet_forward_filter!(stream_batches [$($skip)*] {
198                fn stream_batches<'a>(
199                    &'a self,
200                    context: &'a $crate::forward::__forward::Ctx,
201                    batch_size: usize,
202                ) -> $crate::forward::__forward::Pages<'a, $crate::columnar::ColumnarPage>
203                {
204                    #[allow(unused_imports)]
205                    use $crate::Source as _;
206                    self.$inner.stream_batches(context, batch_size)
207                }
208            });
209        }
210        $crate::__faucet_forward_filter!(native_output_formats [$($skip)*] {
211            fn native_output_formats(&self) -> &'static [$crate::native::NativeFormat]
212            {
213                #[allow(unused_imports)]
214                use $crate::Source as _;
215                self.$inner.native_output_formats()
216            }
217        });
218        $crate::__faucet_forward_filter!(stream_native [$($skip)*] {
219            fn stream_native<'a>(
220                &'a self,
221                context: &'a $crate::forward::__forward::Ctx,
222                format: $crate::native::NativeFormat,
223                batch_size: usize,
224            ) -> $crate::forward::__forward::Pages<'a, $crate::native::NativeBatch>
225            {
226                #[allow(unused_imports)]
227                use $crate::Source as _;
228                self.$inner.stream_native(context, format, batch_size)
229            }
230        });
231        $crate::__faucet_forward_filter!(config_schema [$($skip)*] {
232            fn config_schema(&self) -> $crate::Value
233            {
234                #[allow(unused_imports)]
235                use $crate::Source as _;
236                self.$inner.config_schema()
237            }
238        });
239        $crate::__faucet_forward_filter!(state_key [$($skip)*] {
240            fn state_key(&self) -> ::core::option::Option<::std::string::String>
241            {
242                #[allow(unused_imports)]
243                use $crate::Source as _;
244                self.$inner.state_key()
245            }
246        });
247        $crate::__faucet_forward_filter!(apply_start_bookmark [$($skip)*] {
248            fn apply_start_bookmark<'life0, 'async_trait>(
249                &'life0 self,
250                bookmark: $crate::Value,
251            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
252            where
253                'life0: 'async_trait,
254                Self: 'async_trait,
255            {
256                #[allow(unused_imports)]
257                use $crate::Source as _;
258                self.$inner.apply_start_bookmark(bookmark)
259            }
260        });
261        $crate::__faucet_forward_filter!(capture_resume_position [$($skip)*] {
262            fn capture_resume_position<'life0, 'async_trait>(
263                &'life0 self,
264            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<$crate::Value>>>
265            where
266                'life0: 'async_trait,
267                Self: 'async_trait,
268            {
269                #[allow(unused_imports)]
270                use $crate::Source as _;
271                self.$inner.capture_resume_position()
272            }
273        });
274        $crate::__faucet_forward_filter!(lag [$($skip)*] {
275            fn lag<'life0, 'async_trait>(
276                &'life0 self,
277            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<$crate::lag::SourceLag>>>
278            where
279                'life0: 'async_trait,
280                Self: 'async_trait,
281            {
282                #[allow(unused_imports)]
283                use $crate::Source as _;
284                self.$inner.lag()
285            }
286        });
287        $crate::__faucet_forward_filter!(state_schema [$($skip)*] {
288            fn state_schema(&self) -> u32
289            {
290                #[allow(unused_imports)]
291                use $crate::Source as _;
292                self.$inner.state_schema()
293            }
294        });
295        $crate::__faucet_forward_filter!(migrate_state [$($skip)*] {
296            fn migrate_state(
297                &self,
298                from: u32,
299                data: $crate::Value,
300            ) -> $crate::forward::__forward::Res<$crate::Value>
301            {
302                #[allow(unused_imports)]
303                use $crate::Source as _;
304                self.$inner.migrate_state(from, data)
305            }
306        });
307        $crate::__faucet_forward_filter!(supports_exactly_once [$($skip)*] {
308            fn supports_exactly_once(&self) -> bool
309            {
310                #[allow(unused_imports)]
311                use $crate::Source as _;
312                self.$inner.supports_exactly_once()
313            }
314        });
315        $crate::__faucet_forward_filter!(consumes_destructively [$($skip)*] {
316            fn consumes_destructively(&self) -> bool
317            {
318                #[allow(unused_imports)]
319                use $crate::Source as _;
320                self.$inner.consumes_destructively()
321            }
322        });
323        $crate::__faucet_forward_filter!(replay_guarantee [$($skip)*] {
324            fn replay_guarantee(&self) -> $crate::idempotency::ReplayGuarantee
325            {
326                #[allow(unused_imports)]
327                use $crate::Source as _;
328                self.$inner.replay_guarantee()
329            }
330        });
331        $crate::__faucet_forward_filter!(capabilities [$($skip)*] {
332            fn capabilities(&self) -> $crate::capabilities::Capabilities
333            {
334                #[allow(unused_imports)]
335                use $crate::Source as _;
336                $crate::capabilities::Capabilities::decorated_source(
337                    self,
338                    &self.$inner.capabilities(),
339                )
340            }
341        });
342        $crate::__faucet_forward_filter!(is_shardable [$($skip)*] {
343            fn is_shardable(&self) -> bool
344            {
345                #[allow(unused_imports)]
346                use $crate::Source as _;
347                self.$inner.is_shardable()
348            }
349        });
350        $crate::__faucet_forward_filter!(enumerate_shards [$($skip)*] {
351            fn enumerate_shards<'life0, 'async_trait>(
352                &'life0 self,
353                target: usize,
354            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::std::vec::Vec<$crate::shard::ShardSpec>>>
355            where
356                'life0: 'async_trait,
357                Self: 'async_trait,
358            {
359                #[allow(unused_imports)]
360                use $crate::Source as _;
361                self.$inner.enumerate_shards(target)
362            }
363        });
364        $crate::__faucet_forward_filter!(apply_shard [$($skip)*] {
365            fn apply_shard<'life0, 'life1, 'async_trait>(
366                &'life0 self,
367                shard: &'life1 $crate::shard::ShardSpec,
368            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
369            where
370                'life0: 'async_trait,
371                'life1: 'async_trait,
372                Self: 'async_trait,
373            {
374                #[allow(unused_imports)]
375                use $crate::Source as _;
376                self.$inner.apply_shard(shard)
377            }
378        });
379        $crate::__faucet_forward_filter!(range_digest [$($skip)*] {
380            fn range_digest<'life0, 'life1, 'life2, 'life3, 'async_trait>(
381                &'life0 self,
382                range: &'life1 $crate::diff::KeyRange,
383                key: &'life2 str,
384                columns: &'life3 [::std::string::String],
385            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<$crate::diff::ServerDigest>>>
386            where
387                'life0: 'async_trait,
388                'life1: 'async_trait,
389                'life2: 'async_trait,
390                'life3: 'async_trait,
391                Self: 'async_trait,
392            {
393                #[allow(unused_imports)]
394                use $crate::Source as _;
395                self.$inner.range_digest(range, key, columns)
396            }
397        });
398        $crate::__faucet_forward_filter!(supports_discover [$($skip)*] {
399            fn supports_discover(&self) -> bool
400            {
401                #[allow(unused_imports)]
402                use $crate::Source as _;
403                self.$inner.supports_discover()
404            }
405        });
406        $crate::__faucet_forward_filter!(discover [$($skip)*] {
407            fn discover<'life0, 'async_trait>(
408                &'life0 self,
409            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::std::vec::Vec<$crate::discover::DatasetDescriptor>>>
410            where
411                'life0: 'async_trait,
412                Self: 'async_trait,
413            {
414                #[allow(unused_imports)]
415                use $crate::Source as _;
416                self.$inner.discover()
417            }
418        });
419        $crate::__faucet_forward_filter!(record_table [$($skip)*] {
420            fn record_table(
421                &self,
422                record: &$crate::Value,
423            ) -> ::core::option::Option<::std::string::String>
424            {
425                #[allow(unused_imports)]
426                use $crate::Source as _;
427                self.$inner.record_table(record)
428            }
429        });
430        $crate::__faucet_forward_filter!(position_le [$($skip)*] {
431            fn position_le(
432                &self,
433                a: &$crate::Value,
434                b: &$crate::Value,
435            ) -> ::core::option::Option<bool>
436            {
437                #[allow(unused_imports)]
438                use $crate::Source as _;
439                self.$inner.position_le(a, b)
440            }
441        });
442        $crate::__faucet_forward_filter!(record_covered [$($skip)*] {
443            fn record_covered(
444                &self,
445                record: &$crate::Value,
446                position: &$crate::Value,
447            ) -> ::core::option::Option<bool>
448            {
449                #[allow(unused_imports)]
450                use $crate::Source as _;
451                self.$inner.record_covered(record, position)
452            }
453        });
454        $crate::__faucet_forward_filter!(read_completeness [$($skip)*] {
455            fn read_completeness(&self) -> ::core::option::Option<$crate::cleanup::ReadCompleteness>
456            {
457                #[allow(unused_imports)]
458                use $crate::Source as _;
459                self.$inner.read_completeness()
460            }
461        });
462        $crate::__faucet_forward_filter!(position_min [$($skip)*] {
463            fn position_min(
464                &self,
465                positions: &[$crate::Value],
466            ) -> ::core::option::Option<$crate::Value>
467            {
468                #[allow(unused_imports)]
469                use $crate::Source as _;
470                self.$inner.position_min(positions)
471            }
472        });
473        $crate::__faucet_forward_filter!(connector_name [$($skip)*] {
474            fn connector_name(&self) -> &'static str
475            {
476                #[allow(unused_imports)]
477                use $crate::Source as _;
478                self.$inner.connector_name()
479            }
480        });
481        $crate::__faucet_forward_filter!(set_roundtrip_recorder [$($skip)*] {
482            fn set_roundtrip_recorder(
483                &self,
484                recorder: ::std::sync::Arc<$crate::observability::RoundtripRecorder>,
485            ) -> ()
486            {
487                #[allow(unused_imports)]
488                use $crate::Source as _;
489                self.$inner.set_roundtrip_recorder(recorder)
490            }
491        });
492        $crate::__faucet_forward_filter!(set_call_context [$($skip)*] {
493            fn set_call_context(&self, ctx: $crate::io::CallContext) -> ()
494            {
495                #[allow(unused_imports)]
496                use $crate::Source as _;
497                self.$inner.set_call_context(ctx)
498            }
499        });
500        $crate::__faucet_forward_filter!(set_run_clock [$($skip)*] {
501            fn set_run_clock(
502                &self,
503                now: $crate::forward::__forward::chrono::DateTime<$crate::forward::__forward::chrono::Utc>,
504            ) -> ()
505            {
506                #[allow(unused_imports)]
507                use $crate::Source as _;
508                self.$inner.set_run_clock(now)
509            }
510        });
511        $crate::__faucet_forward_filter!(set_cancel [$($skip)*] {
512            fn set_cancel(&self, cancel: $crate::CancellationToken) -> bool
513            {
514                #[allow(unused_imports)]
515                use $crate::Source as _;
516                self.$inner.set_cancel(cancel)
517            }
518        });
519        $crate::__faucet_forward_filter!(dataset_uri [$($skip)*] {
520            fn dataset_uri(&self) -> ::std::string::String
521            {
522                #[allow(unused_imports)]
523                use $crate::Source as _;
524                self.$inner.dataset_uri()
525            }
526        });
527        $crate::__faucet_forward_filter!(check [$($skip)*] {
528            fn check<'life0, 'life1, 'async_trait>(
529                &'life0 self,
530                ctx: &'life1 $crate::check::CheckContext,
531            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::check::CheckReport>>
532            where
533                'life0: 'async_trait,
534                'life1: 'async_trait,
535                Self: 'async_trait,
536            {
537                #[allow(unused_imports)]
538                use $crate::Source as _;
539                self.$inner.check(ctx)
540            }
541        });
542    };
543}
544
545/// Generate the forwarding body of every [`Sink`](crate::Sink) hook inside a
546/// wrapper's `impl Sink for …` block.
547///
548/// `forward_sink!(field)` forwards every hook to `self.field`;
549/// `forward_sink!(field; except a, b)` leaves out the hooks the wrapper
550/// defines itself (or deliberately leaves at the trait default). See the
551/// [module docs](crate::forward) for the rules.
552///
553/// ```
554/// use faucet_core::{FaucetError, Sink, Value, async_trait, forward_sink};
555/// use std::sync::atomic::{AtomicUsize, Ordering};
556///
557/// /// Counts the rows written through it.
558/// struct Counted<S> {
559///     inner: S,
560///     rows: AtomicUsize,
561/// }
562///
563/// #[async_trait]
564/// impl<S: Sink> Sink for Counted<S> {
565///     async fn write_batch(&self, records: &[Value]) -> Result<usize, FaucetError> {
566///         let n = self.inner.write_batch(records).await?;
567///         self.rows.fetch_add(n, Ordering::Relaxed);
568///         Ok(n)
569///     }
570///     // The other write paths must count too, so they take the trait
571///     // default (which calls `write_batch` above) instead of forwarding.
572///     forward_sink!(inner; except write_batch, write_batch_partial,
573///         write_batch_idempotent, supports_idempotent_writes,
574///         supports_columnar, write_batch_columnar,
575///         native_load_capabilities, load_native);
576/// }
577/// ```
578///
579/// A name in `except` that is not a `Sink` hook does not compile:
580///
581/// ```compile_fail
582/// use faucet_core::{Sink, async_trait, forward_sink};
583///
584/// struct Wrapper {
585///     inner: Box<dyn Sink>,
586/// }
587///
588/// #[async_trait]
589/// impl Sink for Wrapper {
590///     forward_sink!(inner; except write_btach);
591/// }
592/// ```
593#[macro_export]
594macro_rules! forward_sink {
595    ($inner:tt) => { $crate::forward_sink!($inner; except); };
596    ($inner:tt; except $($skip:ident),* $(,)?) => {
597        $crate::__faucet_forward_known!(@sink [$($skip)*]);
598        $crate::__faucet_forward_filter!(write_batch [$($skip)*] {
599            fn write_batch<'life0, 'life1, 'async_trait>(
600                &'life0 self,
601                records: &'life1 [$crate::Value],
602            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
603            where
604                'life0: 'async_trait,
605                'life1: 'async_trait,
606                Self: 'async_trait,
607            {
608                #[allow(unused_imports)]
609                use $crate::Sink as _;
610                self.$inner.write_batch(records)
611            }
612        });
613        $crate::__faucet_forward_filter!(write_batch_reported [$($skip)*] {
614            fn write_batch_reported<'life0, 'life1, 'async_trait>(
615                &'life0 self,
616                records: &'life1 [$crate::Value],
617            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::accounting::WriteReport>>
618            where
619                'life0: 'async_trait,
620                'life1: 'async_trait,
621                Self: 'async_trait,
622            {
623                #[allow(unused_imports)]
624                use $crate::Sink as _;
625                self.$inner.write_batch_reported(records)
626            }
627        });
628        $crate::__faucet_forward_filter!(write_governed [$($skip)*] {
629            fn write_governed<'life0, 'life1, 'life2, 'async_trait>(
630                &'life0 self,
631                page: &'life1 $crate::governance::GovernedPage<'life2>,
632            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::accounting::WriteReport>>
633            where
634                'life0: 'async_trait,
635                'life1: 'async_trait,
636                'life2: 'async_trait,
637                Self: 'async_trait,
638            {
639                #[allow(unused_imports)]
640                use $crate::Sink as _;
641                self.$inner.write_governed(page)
642            }
643        });
644        $crate::__faucet_forward_filter!(flush [$($skip)*] {
645            fn flush<'life0, 'async_trait>(
646                &'life0 self,
647            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
648            where
649                'life0: 'async_trait,
650                Self: 'async_trait,
651            {
652                #[allow(unused_imports)]
653                use $crate::Sink as _;
654                self.$inner.flush()
655            }
656        });
657        $crate::__faucet_forward_filter!(write_batch_partial [$($skip)*] {
658            fn write_batch_partial<'life0, 'life1, 'async_trait>(
659                &'life0 self,
660                records: &'life1 [$crate::Value],
661            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::std::vec::Vec<$crate::traits::RowOutcome>>>
662            where
663                'life0: 'async_trait,
664                'life1: 'async_trait,
665                Self: 'async_trait,
666            {
667                #[allow(unused_imports)]
668                use $crate::Sink as _;
669                self.$inner.write_batch_partial(records)
670            }
671        });
672        $crate::__faucet_forward_filter!(batch_atomicity [$($skip)*] {
673            fn batch_atomicity(&self) -> $crate::dlq::BatchAtomicity
674            {
675                #[allow(unused_imports)]
676                use $crate::Sink as _;
677                self.$inner.batch_atomicity()
678            }
679        });
680        $crate::__faucet_forward_filter!(admit_page [$($skip)*] {
681            fn admit_page<'life0, 'life1, 'async_trait>(
682                &'life0 self,
683                records: &'life1 [$crate::Value],
684            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
685            where
686                'life0: 'async_trait,
687                'life1: 'async_trait,
688                Self: 'async_trait,
689            {
690                #[allow(unused_imports)]
691                use $crate::Sink as _;
692                self.$inner.admit_page(records)
693            }
694        });
695        $crate::__faucet_if_arrow! {
696            $crate::__faucet_forward_filter!(supports_columnar [$($skip)*] {
697                fn supports_columnar(&self) -> bool
698                {
699                    #[allow(unused_imports)]
700                    use $crate::Sink as _;
701                    self.$inner.supports_columnar()
702                }
703            });
704        }
705        $crate::__faucet_if_arrow! {
706            $crate::__faucet_forward_filter!(write_batch_columnar [$($skip)*] {
707                fn write_batch_columnar<'life0, 'life1, 'async_trait>(
708                    &'life0 self,
709                    batch: &'life1 $crate::forward::__forward::arrow::array::RecordBatch,
710                ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
711                where
712                    'life0: 'async_trait,
713                    'life1: 'async_trait,
714                    Self: 'async_trait,
715                {
716                    #[allow(unused_imports)]
717                    use $crate::Sink as _;
718                    self.$inner.write_batch_columnar(batch)
719                }
720            });
721        }
722        $crate::__faucet_forward_filter!(native_load_capabilities [$($skip)*] {
723            fn native_load_capabilities(
724                &self,
725            ) -> ::std::vec::Vec<$crate::native::NativeLoadCapability>
726            {
727                #[allow(unused_imports)]
728                use $crate::Sink as _;
729                self.$inner.native_load_capabilities()
730            }
731        });
732        $crate::__faucet_forward_filter!(load_native [$($skip)*] {
733            fn load_native<'life0, 'life1, 'async_trait>(
734                &'life0 self,
735                batch: $crate::native::NativeBatch,
736                scope: &'life1 str,
737                ctx: $crate::native::NativeLoadContext,
738            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
739            where
740                'life0: 'async_trait,
741                'life1: 'async_trait,
742                Self: 'async_trait,
743            {
744                #[allow(unused_imports)]
745                use $crate::Sink as _;
746                self.$inner.load_native(batch, scope, ctx)
747            }
748        });
749        $crate::__faucet_forward_filter!(supports_idempotent_writes [$($skip)*] {
750            fn supports_idempotent_writes(&self) -> bool
751            {
752                #[allow(unused_imports)]
753                use $crate::Sink as _;
754                self.$inner.supports_idempotent_writes()
755            }
756        });
757        $crate::__faucet_forward_filter!(sink_guarantee [$($skip)*] {
758            fn sink_guarantee(&self) -> $crate::idempotency::SinkGuarantee
759            {
760                #[allow(unused_imports)]
761                use $crate::Sink as _;
762                self.$inner.sink_guarantee()
763            }
764        });
765        $crate::__faucet_forward_filter!(capabilities [$($skip)*] {
766            fn capabilities(&self) -> $crate::capabilities::Capabilities
767            {
768                #[allow(unused_imports)]
769                use $crate::Sink as _;
770                $crate::capabilities::Capabilities::decorated_sink(
771                    self,
772                    &self.$inner.capabilities(),
773                )
774            }
775        });
776        $crate::__faucet_forward_filter!(write_batch_is_replay_safe [$($skip)*] {
777            fn write_batch_is_replay_safe(&self) -> bool
778            {
779                #[allow(unused_imports)]
780                use $crate::Sink as _;
781                self.$inner.write_batch_is_replay_safe()
782            }
783        });
784        $crate::__faucet_forward_filter!(dedups_by_key [$($skip)*] {
785            fn dedups_by_key(&self) -> bool
786            {
787                #[allow(unused_imports)]
788                use $crate::Sink as _;
789                self.$inner.dedups_by_key()
790            }
791        });
792        $crate::__faucet_forward_filter!(supported_write_modes [$($skip)*] {
793            fn supported_write_modes(&self) -> &'static [$crate::write_mode::WriteMode]
794            {
795                #[allow(unused_imports)]
796                use $crate::Sink as _;
797                self.$inner.supported_write_modes()
798            }
799        });
800        $crate::__faucet_forward_filter!(current_schema [$($skip)*] {
801            fn current_schema<'life0, 'async_trait>(
802                &'life0 self,
803            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<$crate::Value>>>
804            where
805                'life0: 'async_trait,
806                Self: 'async_trait,
807            {
808                #[allow(unused_imports)]
809                use $crate::Sink as _;
810                self.$inner.current_schema()
811            }
812        });
813        $crate::__faucet_forward_filter!(supports_schema_evolution [$($skip)*] {
814            fn supports_schema_evolution(&self) -> bool
815            {
816                #[allow(unused_imports)]
817                use $crate::Sink as _;
818                self.$inner.supports_schema_evolution()
819            }
820        });
821        $crate::__faucet_forward_filter!(evolve_schema [$($skip)*] {
822            fn evolve_schema<'life0, 'life1, 'async_trait>(
823                &'life0 self,
824                evolution: &'life1 $crate::drift::SchemaEvolution,
825            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
826            where
827                'life0: 'async_trait,
828                'life1: 'async_trait,
829                Self: 'async_trait,
830            {
831                #[allow(unused_imports)]
832                use $crate::Sink as _;
833                self.$inner.evolve_schema(evolution)
834            }
835        });
836        $crate::__faucet_forward_filter!(supports_cleanup [$($skip)*] {
837            fn supports_cleanup(&self) -> bool
838            {
839                #[allow(unused_imports)]
840                use $crate::Sink as _;
841                self.$inner.supports_cleanup()
842            }
843        });
844        $crate::__faucet_forward_filter!(supports_staged_load [$($skip)*] {
845            fn supports_staged_load(&self) -> bool
846            {
847                #[allow(unused_imports)]
848                use $crate::Sink as _;
849                self.$inner.supports_staged_load()
850            }
851        });
852        $crate::__faucet_forward_filter!(cleanup_scope [$($skip)*] {
853            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
854            fn cleanup_scope<'life0, 'life1, 'life2, 'async_trait>(
855                &'life0 self,
856                scope: &'life1 ::std::collections::BTreeMap<::std::string::String, $crate::Value>,
857                seen: &'life2 $crate::cleanup::SeenKeys,
858            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<u64>>
859            where
860                'life0: 'async_trait,
861                'life1: 'async_trait,
862                'life2: 'async_trait,
863                Self: 'async_trait,
864            {
865                #[allow(unused_imports)]
866                use $crate::Sink as _;
867                self.$inner.cleanup_scope(scope, seen)
868            }
869        });
870        $crate::__faucet_forward_filter!(write_batch_idempotent [$($skip)*] {
871            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
872            fn write_batch_idempotent<'life0, 'life1, 'life2, 'life3, 'async_trait>(
873                &'life0 self,
874                records: &'life1 [$crate::Value],
875                scope: &'life2 str,
876                token: &'life3 str,
877            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
878            where
879                'life0: 'async_trait,
880                'life1: 'async_trait,
881                'life2: 'async_trait,
882                'life3: 'async_trait,
883                Self: 'async_trait,
884            {
885                #[allow(unused_imports)]
886                use $crate::Sink as _;
887                self.$inner.write_batch_idempotent(records, scope, token)
888            }
889        });
890        $crate::__faucet_forward_filter!(write_batch_idempotent_reported [$($skip)*] {
891            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
892            fn write_batch_idempotent_reported<'life0, 'life1, 'life2, 'life3, 'async_trait>(
893                &'life0 self,
894                records: &'life1 [$crate::Value],
895                scope: &'life2 str,
896                token: &'life3 str,
897            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::accounting::WriteReport>>
898            where
899                'life0: 'async_trait,
900                'life1: 'async_trait,
901                'life2: 'async_trait,
902                'life3: 'async_trait,
903                Self: 'async_trait,
904            {
905                #[allow(unused_imports)]
906                use $crate::Sink as _;
907                self.$inner.write_batch_idempotent_reported(records, scope, token)
908            }
909        });
910        $crate::__faucet_forward_filter!(last_committed_token [$($skip)*] {
911            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
912            fn last_committed_token<'life0, 'life1, 'async_trait>(
913                &'life0 self,
914                scope: &'life1 str,
915            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<::std::string::String>>>
916            where
917                'life0: 'async_trait,
918                'life1: 'async_trait,
919                Self: 'async_trait,
920            {
921                #[allow(unused_imports)]
922                use $crate::Sink as _;
923                self.$inner.last_committed_token(scope)
924            }
925        });
926        $crate::__faucet_forward_filter!(is_overwrite [$($skip)*] {
927            fn is_overwrite(&self) -> bool
928            {
929                #[allow(unused_imports)]
930                use $crate::Sink as _;
931                self.$inner.is_overwrite()
932            }
933        });
934        $crate::__faucet_forward_filter!(begin_overwrite [$($skip)*] {
935            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
936            fn begin_overwrite<'life0, 'async_trait>(
937                &'life0 self,
938            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
939            where
940                'life0: 'async_trait,
941                Self: 'async_trait,
942            {
943                #[allow(unused_imports)]
944                use $crate::Sink as _;
945                self.$inner.begin_overwrite()
946            }
947        });
948        $crate::__faucet_forward_filter!(commit_overwrite [$($skip)*] {
949            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
950            fn commit_overwrite<'life0, 'async_trait>(
951                &'life0 self,
952            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
953            where
954                'life0: 'async_trait,
955                Self: 'async_trait,
956            {
957                #[allow(unused_imports)]
958                use $crate::Sink as _;
959                self.$inner.commit_overwrite()
960            }
961        });
962        $crate::__faucet_forward_filter!(supports_rollback [$($skip)*] {
963            fn supports_rollback(&self) -> bool
964            {
965                #[allow(unused_imports)]
966                use $crate::Sink as _;
967                self.$inner.supports_rollback()
968            }
969        });
970        $crate::__faucet_forward_filter!(rollback_run [$($skip)*] {
971            fn rollback_run<'life0, 'life1, 'life2, 'async_trait>(
972                &'life0 self,
973                run_id: &'life1 str,
974                opts: &'life2 $crate::rollback::RollbackOptions,
975            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::rollback::RollbackOutcome>>
976            where
977                'life0: 'async_trait,
978                'life1: 'async_trait,
979                'life2: 'async_trait,
980                Self: 'async_trait,
981            {
982                #[allow(unused_imports)]
983                use $crate::Sink as _;
984                self.$inner.rollback_run(run_id, opts)
985            }
986        });
987        $crate::__faucet_forward_filter!(forget_run [$($skip)*] {
988            fn forget_run<'life0, 'life1, 'async_trait>(
989                &'life0 self,
990                run_id: &'life1 str,
991            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
992            where
993                'life0: 'async_trait,
994                'life1: 'async_trait,
995                Self: 'async_trait,
996            {
997                #[allow(unused_imports)]
998                use $crate::Sink as _;
999                self.$inner.forget_run(run_id)
1000            }
1001        });
1002        $crate::__faucet_forward_filter!(rewind_commit_token [$($skip)*] {
1003            fn rewind_commit_token<'life0, 'life1, 'life2, 'async_trait>(
1004                &'life0 self,
1005                scope: &'life1 str,
1006                token: ::core::option::Option<&'life2 str>,
1007            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
1008            where
1009                'life0: 'async_trait,
1010                'life1: 'async_trait,
1011                'life2: 'async_trait,
1012                Self: 'async_trait,
1013            {
1014                #[allow(unused_imports)]
1015                use $crate::Sink as _;
1016                self.$inner.rewind_commit_token(scope, token)
1017            }
1018        });
1019        $crate::__faucet_forward_filter!(readback_source [$($skip)*] {
1020            fn readback_source(
1021                &self,
1022            ) -> ::core::option::Option<(::std::string::String, $crate::Value)>
1023            {
1024                #[allow(unused_imports)]
1025                use $crate::Sink as _;
1026                self.$inner.readback_source()
1027            }
1028        });
1029        $crate::__faucet_forward_filter!(abort_overwrite [$($skip)*] {
1030            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
1031            fn abort_overwrite<'life0, 'async_trait>(
1032                &'life0 self,
1033            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
1034            where
1035                'life0: 'async_trait,
1036                Self: 'async_trait,
1037            {
1038                #[allow(unused_imports)]
1039                use $crate::Sink as _;
1040                self.$inner.abort_overwrite()
1041            }
1042        });
1043        $crate::__faucet_forward_filter!(complete_run [$($skip)*] {
1044            fn complete_run<'life0, 'async_trait>(
1045                &'life0 self,
1046            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
1047            where
1048                'life0: 'async_trait,
1049                Self: 'async_trait,
1050            {
1051                #[allow(unused_imports)]
1052                use $crate::Sink as _;
1053                self.$inner.complete_run()
1054            }
1055        });
1056        $crate::__faucet_forward_filter!(overwrite_staging_exists [$($skip)*] {
1057            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
1058            fn overwrite_staging_exists<'life0, 'async_trait>(
1059                &'life0 self,
1060            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<bool>>>
1061            where
1062                'life0: 'async_trait,
1063                Self: 'async_trait,
1064            {
1065                #[allow(unused_imports)]
1066                use $crate::Sink as _;
1067                self.$inner.overwrite_staging_exists()
1068            }
1069        });
1070        $crate::__faucet_forward_filter!(config_schema [$($skip)*] {
1071            fn config_schema(&self) -> $crate::Value
1072            {
1073                #[allow(unused_imports)]
1074                use $crate::Sink as _;
1075                self.$inner.config_schema()
1076            }
1077        });
1078        $crate::__faucet_forward_filter!(connector_name [$($skip)*] {
1079            fn connector_name(&self) -> &'static str
1080            {
1081                #[allow(unused_imports)]
1082                use $crate::Sink as _;
1083                self.$inner.connector_name()
1084            }
1085        });
1086        $crate::__faucet_forward_filter!(set_roundtrip_recorder [$($skip)*] {
1087            fn set_roundtrip_recorder(
1088                &self,
1089                recorder: ::std::sync::Arc<$crate::observability::RoundtripRecorder>,
1090            ) -> ()
1091            {
1092                #[allow(unused_imports)]
1093                use $crate::Sink as _;
1094                self.$inner.set_roundtrip_recorder(recorder)
1095            }
1096        });
1097        $crate::__faucet_forward_filter!(set_call_context [$($skip)*] {
1098            fn set_call_context(&self, ctx: $crate::io::CallContext) -> ()
1099            {
1100                #[allow(unused_imports)]
1101                use $crate::Sink as _;
1102                self.$inner.set_call_context(ctx)
1103            }
1104        });
1105        $crate::__faucet_forward_filter!(dataset_uri [$($skip)*] {
1106            fn dataset_uri(&self) -> ::std::string::String
1107            {
1108                #[allow(unused_imports)]
1109                use $crate::Sink as _;
1110                self.$inner.dataset_uri()
1111            }
1112        });
1113        $crate::__faucet_forward_filter!(local_outputs [$($skip)*] {
1114            fn local_outputs<'life0, 'async_trait>(
1115                &'life0 self,
1116            ) -> $crate::forward::__forward::Fut<'async_trait, ::std::vec::Vec<$crate::local_outputs::LocalOutput>>
1117            where
1118                'life0: 'async_trait,
1119                Self: 'async_trait,
1120            {
1121                #[allow(unused_imports)]
1122                use $crate::Sink as _;
1123                self.$inner.local_outputs()
1124            }
1125        });
1126        $crate::__faucet_forward_filter!(check [$($skip)*] {
1127            fn check<'life0, 'life1, 'async_trait>(
1128                &'life0 self,
1129                ctx: &'life1 $crate::check::CheckContext,
1130            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::check::CheckReport>>
1131            where
1132                'life0: 'async_trait,
1133                'life1: 'async_trait,
1134                Self: 'async_trait,
1135            {
1136                #[allow(unused_imports)]
1137                use $crate::Sink as _;
1138                self.$inner.check(ctx)
1139            }
1140        });
1141    };
1142}
1143
1144#[doc(hidden)]
1145#[macro_export]
1146macro_rules! __faucet_forward_filter {
1147    (fetch_with_context [fetch_with_context $($rest:ident)*] $item:tt) => {};
1148    (fetch_all [fetch_all $($rest:ident)*] $item:tt) => {};
1149    (fetch_with_context_incremental [fetch_with_context_incremental $($rest:ident)*] $item:tt) => {};
1150    (fetch_all_incremental [fetch_all_incremental $($rest:ident)*] $item:tt) => {};
1151    (stream_pages [stream_pages $($rest:ident)*] $item:tt) => {};
1152    (supports_columnar [supports_columnar $($rest:ident)*] $item:tt) => {};
1153    (stream_batches [stream_batches $($rest:ident)*] $item:tt) => {};
1154    (native_output_formats [native_output_formats $($rest:ident)*] $item:tt) => {};
1155    (stream_native [stream_native $($rest:ident)*] $item:tt) => {};
1156    (config_schema [config_schema $($rest:ident)*] $item:tt) => {};
1157    (state_key [state_key $($rest:ident)*] $item:tt) => {};
1158    (apply_start_bookmark [apply_start_bookmark $($rest:ident)*] $item:tt) => {};
1159    (capture_resume_position [capture_resume_position $($rest:ident)*] $item:tt) => {};
1160    (lag [lag $($rest:ident)*] $item:tt) => {};
1161    (state_schema [state_schema $($rest:ident)*] $item:tt) => {};
1162    (migrate_state [migrate_state $($rest:ident)*] $item:tt) => {};
1163    (supports_exactly_once [supports_exactly_once $($rest:ident)*] $item:tt) => {};
1164    (consumes_destructively [consumes_destructively $($rest:ident)*] $item:tt) => {};
1165    (replay_guarantee [replay_guarantee $($rest:ident)*] $item:tt) => {};
1166    (capabilities [capabilities $($rest:ident)*] $item:tt) => {};
1167    (is_shardable [is_shardable $($rest:ident)*] $item:tt) => {};
1168    (enumerate_shards [enumerate_shards $($rest:ident)*] $item:tt) => {};
1169    (apply_shard [apply_shard $($rest:ident)*] $item:tt) => {};
1170    (range_digest [range_digest $($rest:ident)*] $item:tt) => {};
1171    (supports_discover [supports_discover $($rest:ident)*] $item:tt) => {};
1172    (discover [discover $($rest:ident)*] $item:tt) => {};
1173    (record_table [record_table $($rest:ident)*] $item:tt) => {};
1174    (position_le [position_le $($rest:ident)*] $item:tt) => {};
1175    (record_covered [record_covered $($rest:ident)*] $item:tt) => {};
1176    (read_completeness [read_completeness $($rest:ident)*] $item:tt) => {};
1177    (position_min [position_min $($rest:ident)*] $item:tt) => {};
1178    (connector_name [connector_name $($rest:ident)*] $item:tt) => {};
1179    (set_roundtrip_recorder [set_roundtrip_recorder $($rest:ident)*] $item:tt) => {};
1180    (set_call_context [set_call_context $($rest:ident)*] $item:tt) => {};
1181    (set_run_clock [set_run_clock $($rest:ident)*] $item:tt) => {};
1182    (set_cancel [set_cancel $($rest:ident)*] $item:tt) => {};
1183    (dataset_uri [dataset_uri $($rest:ident)*] $item:tt) => {};
1184    (check [check $($rest:ident)*] $item:tt) => {};
1185    (write_batch [write_batch $($rest:ident)*] $item:tt) => {};
1186    (write_batch_reported [write_batch_reported $($rest:ident)*] $item:tt) => {};
1187    // A wrapper that changes `write_batch` takes the trait default for the
1188    // reported write, which goes through its own `write_batch` (#879).
1189    (write_batch_reported [write_batch $($rest:ident)*] $item:tt) => {};
1190    (write_governed [write_governed $($rest:ident)*] $item:tt) => {};
1191    // A wrapper changing the plain or reported write keeps the default governed write.
1192    (write_governed [write_batch $($rest:ident)*] $item:tt) => {};
1193    (write_governed [write_batch_reported $($rest:ident)*] $item:tt) => {};
1194    (flush [flush $($rest:ident)*] $item:tt) => {};
1195    (write_batch_partial [write_batch_partial $($rest:ident)*] $item:tt) => {};
1196    (batch_atomicity [batch_atomicity $($rest:ident)*] $item:tt) => {};
1197    (admit_page [admit_page $($rest:ident)*] $item:tt) => {};
1198    (write_batch_columnar [write_batch_columnar $($rest:ident)*] $item:tt) => {};
1199    (native_load_capabilities [native_load_capabilities $($rest:ident)*] $item:tt) => {};
1200    (load_native [load_native $($rest:ident)*] $item:tt) => {};
1201    (supports_idempotent_writes [supports_idempotent_writes $($rest:ident)*] $item:tt) => {};
1202    (sink_guarantee [sink_guarantee $($rest:ident)*] $item:tt) => {};
1203    (write_batch_is_replay_safe [write_batch_is_replay_safe $($rest:ident)*] $item:tt) => {};
1204    (dedups_by_key [dedups_by_key $($rest:ident)*] $item:tt) => {};
1205    (supported_write_modes [supported_write_modes $($rest:ident)*] $item:tt) => {};
1206    (current_schema [current_schema $($rest:ident)*] $item:tt) => {};
1207    (supports_schema_evolution [supports_schema_evolution $($rest:ident)*] $item:tt) => {};
1208    (evolve_schema [evolve_schema $($rest:ident)*] $item:tt) => {};
1209    (supports_cleanup [supports_cleanup $($rest:ident)*] $item:tt) => {};
1210    (supports_staged_load [supports_staged_load $($rest:ident)*] $item:tt) => {};
1211    (cleanup_scope [cleanup_scope $($rest:ident)*] $item:tt) => {};
1212    (write_batch_idempotent [write_batch_idempotent $($rest:ident)*] $item:tt) => {};
1213    (write_batch_idempotent_reported [write_batch_idempotent_reported $($rest:ident)*] $item:tt) => {};
1214    // A wrapper that changes `write_batch_idempotent` takes the trait default
1215    // for the reported write, which goes through its own change (#879).
1216    (write_batch_idempotent_reported [write_batch_idempotent $($rest:ident)*] $item:tt) => {};
1217    (last_committed_token [last_committed_token $($rest:ident)*] $item:tt) => {};
1218    (is_overwrite [is_overwrite $($rest:ident)*] $item:tt) => {};
1219    (begin_overwrite [begin_overwrite $($rest:ident)*] $item:tt) => {};
1220    (commit_overwrite [commit_overwrite $($rest:ident)*] $item:tt) => {};
1221    (supports_rollback [supports_rollback $($rest:ident)*] $item:tt) => {};
1222    (rollback_run [rollback_run $($rest:ident)*] $item:tt) => {};
1223    (forget_run [forget_run $($rest:ident)*] $item:tt) => {};
1224    (rewind_commit_token [rewind_commit_token $($rest:ident)*] $item:tt) => {};
1225    (readback_source [readback_source $($rest:ident)*] $item:tt) => {};
1226    (abort_overwrite [abort_overwrite $($rest:ident)*] $item:tt) => {};
1227    (complete_run [complete_run $($rest:ident)*] $item:tt) => {};
1228    (overwrite_staging_exists [overwrite_staging_exists $($rest:ident)*] $item:tt) => {};
1229    (local_outputs [local_outputs $($rest:ident)*] $item:tt) => {};
1230    ($hook:ident [] { $($item:tt)* }) => { $($item)* };
1231    ($hook:ident [$other:ident $($rest:ident)*] $item:tt) => {
1232        $crate::__faucet_forward_filter!($hook [$($rest)*] $item);
1233    };
1234}
1235
1236#[doc(hidden)]
1237#[macro_export]
1238macro_rules! __faucet_forward_known {
1239    (@source [fetch_with_context $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1240    (@source [fetch_all $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1241    (@source [fetch_with_context_incremental $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1242    (@source [fetch_all_incremental $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1243    (@source [stream_pages $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1244    (@source [supports_columnar $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1245    (@source [stream_batches $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1246    (@source [native_output_formats $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1247    (@source [stream_native $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1248    (@source [config_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1249    (@source [state_key $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1250    (@source [apply_start_bookmark $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1251    (@source [capture_resume_position $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1252    (@source [lag $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1253    (@source [state_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1254    (@source [migrate_state $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1255    (@source [supports_exactly_once $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1256    (@source [consumes_destructively $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1257    (@source [replay_guarantee $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1258    (@source [capabilities $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1259    (@source [is_shardable $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1260    (@source [enumerate_shards $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1261    (@source [apply_shard $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1262    (@source [range_digest $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1263    (@source [supports_discover $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1264    (@source [discover $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1265    (@source [record_table $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1266    (@source [position_le $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1267    (@source [record_covered $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1268    (@source [read_completeness $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1269    (@source [position_min $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1270    (@source [connector_name $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1271    (@source [set_roundtrip_recorder $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1272    (@source [set_call_context $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1273    (@source [set_run_clock $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1274    (@source [set_cancel $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1275    (@source [dataset_uri $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1276    (@source [check $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1277    (@sink [write_batch $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1278    (@sink [write_batch_reported $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1279    (@sink [write_governed $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1280    (@sink [flush $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1281    (@sink [write_batch_partial $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1282    (@sink [batch_atomicity $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1283    (@sink [admit_page $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1284    (@sink [supports_columnar $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1285    (@sink [write_batch_columnar $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1286    (@sink [native_load_capabilities $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1287    (@sink [load_native $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1288    (@sink [supports_idempotent_writes $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1289    (@sink [sink_guarantee $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1290    (@sink [capabilities $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1291    (@sink [write_batch_is_replay_safe $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1292    (@sink [dedups_by_key $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1293    (@sink [supported_write_modes $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1294    (@sink [current_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1295    (@sink [supports_schema_evolution $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1296    (@sink [evolve_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1297    (@sink [supports_cleanup $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1298    (@sink [supports_staged_load $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1299    (@sink [cleanup_scope $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1300    (@sink [write_batch_idempotent $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1301    (@sink [write_batch_idempotent_reported $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1302    (@sink [last_committed_token $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1303    (@sink [is_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1304    (@sink [begin_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1305    (@sink [commit_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1306    (@sink [supports_rollback $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1307    (@sink [rollback_run $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1308    (@sink [forget_run $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1309    (@sink [rewind_commit_token $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1310    (@sink [readback_source $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1311    (@sink [abort_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1312    (@sink [complete_run $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1313    (@sink [overwrite_staging_exists $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1314    (@sink [config_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1315    (@sink [connector_name $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1316    (@sink [set_roundtrip_recorder $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1317    (@sink [set_call_context $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1318    (@sink [dataset_uri $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1319    (@sink [local_outputs $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1320    (@sink [check $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1321    (@$tr:ident []) => {};
1322    (@source [$other:ident $($rest:ident)*]) => {
1323        ::core::compile_error!(::core::concat!(
1324            "forward_source!: `", ::core::stringify!($other), "` is not a Source hook"
1325        ));
1326    };
1327    (@sink [$other:ident $($rest:ident)*]) => {
1328        ::core::compile_error!(::core::concat!(
1329            "forward_sink!: `", ::core::stringify!($other), "` is not a Sink hook"
1330        ));
1331    };
1332}
faucet_core/forward.rs

faucet_core/
forward.rs
