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!(position_min [$($skip)*] {
455            fn position_min(
456                &self,
457                positions: &[$crate::Value],
458            ) -> ::core::option::Option<$crate::Value>
459            {
460                #[allow(unused_imports)]
461                use $crate::Source as _;
462                self.$inner.position_min(positions)
463            }
464        });
465        $crate::__faucet_forward_filter!(connector_name [$($skip)*] {
466            fn connector_name(&self) -> &'static str
467            {
468                #[allow(unused_imports)]
469                use $crate::Source as _;
470                self.$inner.connector_name()
471            }
472        });
473        $crate::__faucet_forward_filter!(set_roundtrip_recorder [$($skip)*] {
474            fn set_roundtrip_recorder(
475                &self,
476                recorder: ::std::sync::Arc<$crate::observability::RoundtripRecorder>,
477            ) -> ()
478            {
479                #[allow(unused_imports)]
480                use $crate::Source as _;
481                self.$inner.set_roundtrip_recorder(recorder)
482            }
483        });
484        $crate::__faucet_forward_filter!(set_call_context [$($skip)*] {
485            fn set_call_context(&self, ctx: $crate::io::CallContext) -> ()
486            {
487                #[allow(unused_imports)]
488                use $crate::Source as _;
489                self.$inner.set_call_context(ctx)
490            }
491        });
492        $crate::__faucet_forward_filter!(set_run_clock [$($skip)*] {
493            fn set_run_clock(
494                &self,
495                now: $crate::forward::__forward::chrono::DateTime<$crate::forward::__forward::chrono::Utc>,
496            ) -> ()
497            {
498                #[allow(unused_imports)]
499                use $crate::Source as _;
500                self.$inner.set_run_clock(now)
501            }
502        });
503        $crate::__faucet_forward_filter!(set_cancel [$($skip)*] {
504            fn set_cancel(&self, cancel: $crate::CancellationToken) -> bool
505            {
506                #[allow(unused_imports)]
507                use $crate::Source as _;
508                self.$inner.set_cancel(cancel)
509            }
510        });
511        $crate::__faucet_forward_filter!(dataset_uri [$($skip)*] {
512            fn dataset_uri(&self) -> ::std::string::String
513            {
514                #[allow(unused_imports)]
515                use $crate::Source as _;
516                self.$inner.dataset_uri()
517            }
518        });
519        $crate::__faucet_forward_filter!(check [$($skip)*] {
520            fn check<'life0, 'life1, 'async_trait>(
521                &'life0 self,
522                ctx: &'life1 $crate::check::CheckContext,
523            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::check::CheckReport>>
524            where
525                'life0: 'async_trait,
526                'life1: 'async_trait,
527                Self: 'async_trait,
528            {
529                #[allow(unused_imports)]
530                use $crate::Source as _;
531                self.$inner.check(ctx)
532            }
533        });
534    };
535}
536
537/// Generate the forwarding body of every [`Sink`](crate::Sink) hook inside a
538/// wrapper's `impl Sink for …` block.
539///
540/// `forward_sink!(field)` forwards every hook to `self.field`;
541/// `forward_sink!(field; except a, b)` leaves out the hooks the wrapper
542/// defines itself (or deliberately leaves at the trait default). See the
543/// [module docs](crate::forward) for the rules.
544///
545/// ```
546/// use faucet_core::{FaucetError, Sink, Value, async_trait, forward_sink};
547/// use std::sync::atomic::{AtomicUsize, Ordering};
548///
549/// /// Counts the rows written through it.
550/// struct Counted<S> {
551///     inner: S,
552///     rows: AtomicUsize,
553/// }
554///
555/// #[async_trait]
556/// impl<S: Sink> Sink for Counted<S> {
557///     async fn write_batch(&self, records: &[Value]) -> Result<usize, FaucetError> {
558///         let n = self.inner.write_batch(records).await?;
559///         self.rows.fetch_add(n, Ordering::Relaxed);
560///         Ok(n)
561///     }
562///     // The other write paths must count too, so they take the trait
563///     // default (which calls `write_batch` above) instead of forwarding.
564///     forward_sink!(inner; except write_batch, write_batch_partial,
565///         write_batch_idempotent, supports_idempotent_writes,
566///         supports_columnar, write_batch_columnar,
567///         native_load_capabilities, load_native);
568/// }
569/// ```
570///
571/// A name in `except` that is not a `Sink` hook does not compile:
572///
573/// ```compile_fail
574/// use faucet_core::{Sink, async_trait, forward_sink};
575///
576/// struct Wrapper {
577///     inner: Box<dyn Sink>,
578/// }
579///
580/// #[async_trait]
581/// impl Sink for Wrapper {
582///     forward_sink!(inner; except write_btach);
583/// }
584/// ```
585#[macro_export]
586macro_rules! forward_sink {
587    ($inner:tt) => { $crate::forward_sink!($inner; except); };
588    ($inner:tt; except $($skip:ident),* $(,)?) => {
589        $crate::__faucet_forward_known!(@sink [$($skip)*]);
590        $crate::__faucet_forward_filter!(write_batch [$($skip)*] {
591            fn write_batch<'life0, 'life1, 'async_trait>(
592                &'life0 self,
593                records: &'life1 [$crate::Value],
594            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
595            where
596                'life0: 'async_trait,
597                'life1: 'async_trait,
598                Self: 'async_trait,
599            {
600                #[allow(unused_imports)]
601                use $crate::Sink as _;
602                self.$inner.write_batch(records)
603            }
604        });
605        $crate::__faucet_forward_filter!(write_batch_reported [$($skip)*] {
606            fn write_batch_reported<'life0, 'life1, 'async_trait>(
607                &'life0 self,
608                records: &'life1 [$crate::Value],
609            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::accounting::WriteReport>>
610            where
611                'life0: 'async_trait,
612                'life1: 'async_trait,
613                Self: 'async_trait,
614            {
615                #[allow(unused_imports)]
616                use $crate::Sink as _;
617                self.$inner.write_batch_reported(records)
618            }
619        });
620        $crate::__faucet_forward_filter!(write_governed [$($skip)*] {
621            fn write_governed<'life0, 'life1, 'life2, 'async_trait>(
622                &'life0 self,
623                page: &'life1 $crate::governance::GovernedPage<'life2>,
624            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::accounting::WriteReport>>
625            where
626                'life0: 'async_trait,
627                'life1: 'async_trait,
628                'life2: 'async_trait,
629                Self: 'async_trait,
630            {
631                #[allow(unused_imports)]
632                use $crate::Sink as _;
633                self.$inner.write_governed(page)
634            }
635        });
636        $crate::__faucet_forward_filter!(flush [$($skip)*] {
637            fn flush<'life0, 'async_trait>(
638                &'life0 self,
639            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
640            where
641                'life0: 'async_trait,
642                Self: 'async_trait,
643            {
644                #[allow(unused_imports)]
645                use $crate::Sink as _;
646                self.$inner.flush()
647            }
648        });
649        $crate::__faucet_forward_filter!(write_batch_partial [$($skip)*] {
650            fn write_batch_partial<'life0, 'life1, 'async_trait>(
651                &'life0 self,
652                records: &'life1 [$crate::Value],
653            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::std::vec::Vec<$crate::traits::RowOutcome>>>
654            where
655                'life0: 'async_trait,
656                'life1: 'async_trait,
657                Self: 'async_trait,
658            {
659                #[allow(unused_imports)]
660                use $crate::Sink as _;
661                self.$inner.write_batch_partial(records)
662            }
663        });
664        $crate::__faucet_forward_filter!(batch_atomicity [$($skip)*] {
665            fn batch_atomicity(&self) -> $crate::dlq::BatchAtomicity
666            {
667                #[allow(unused_imports)]
668                use $crate::Sink as _;
669                self.$inner.batch_atomicity()
670            }
671        });
672        $crate::__faucet_forward_filter!(admit_page [$($skip)*] {
673            fn admit_page<'life0, 'life1, 'async_trait>(
674                &'life0 self,
675                records: &'life1 [$crate::Value],
676            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
677            where
678                'life0: 'async_trait,
679                'life1: 'async_trait,
680                Self: 'async_trait,
681            {
682                #[allow(unused_imports)]
683                use $crate::Sink as _;
684                self.$inner.admit_page(records)
685            }
686        });
687        $crate::__faucet_if_arrow! {
688            $crate::__faucet_forward_filter!(supports_columnar [$($skip)*] {
689                fn supports_columnar(&self) -> bool
690                {
691                    #[allow(unused_imports)]
692                    use $crate::Sink as _;
693                    self.$inner.supports_columnar()
694                }
695            });
696        }
697        $crate::__faucet_if_arrow! {
698            $crate::__faucet_forward_filter!(write_batch_columnar [$($skip)*] {
699                fn write_batch_columnar<'life0, 'life1, 'async_trait>(
700                    &'life0 self,
701                    batch: &'life1 $crate::forward::__forward::arrow::array::RecordBatch,
702                ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
703                where
704                    'life0: 'async_trait,
705                    'life1: 'async_trait,
706                    Self: 'async_trait,
707                {
708                    #[allow(unused_imports)]
709                    use $crate::Sink as _;
710                    self.$inner.write_batch_columnar(batch)
711                }
712            });
713        }
714        $crate::__faucet_forward_filter!(native_load_capabilities [$($skip)*] {
715            fn native_load_capabilities(
716                &self,
717            ) -> ::std::vec::Vec<$crate::native::NativeLoadCapability>
718            {
719                #[allow(unused_imports)]
720                use $crate::Sink as _;
721                self.$inner.native_load_capabilities()
722            }
723        });
724        $crate::__faucet_forward_filter!(load_native [$($skip)*] {
725            fn load_native<'life0, 'life1, 'async_trait>(
726                &'life0 self,
727                batch: $crate::native::NativeBatch,
728                scope: &'life1 str,
729                ctx: $crate::native::NativeLoadContext,
730            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
731            where
732                'life0: 'async_trait,
733                'life1: 'async_trait,
734                Self: 'async_trait,
735            {
736                #[allow(unused_imports)]
737                use $crate::Sink as _;
738                self.$inner.load_native(batch, scope, ctx)
739            }
740        });
741        $crate::__faucet_forward_filter!(supports_idempotent_writes [$($skip)*] {
742            fn supports_idempotent_writes(&self) -> bool
743            {
744                #[allow(unused_imports)]
745                use $crate::Sink as _;
746                self.$inner.supports_idempotent_writes()
747            }
748        });
749        $crate::__faucet_forward_filter!(sink_guarantee [$($skip)*] {
750            fn sink_guarantee(&self) -> $crate::idempotency::SinkGuarantee
751            {
752                #[allow(unused_imports)]
753                use $crate::Sink as _;
754                self.$inner.sink_guarantee()
755            }
756        });
757        $crate::__faucet_forward_filter!(capabilities [$($skip)*] {
758            fn capabilities(&self) -> $crate::capabilities::Capabilities
759            {
760                #[allow(unused_imports)]
761                use $crate::Sink as _;
762                $crate::capabilities::Capabilities::decorated_sink(
763                    self,
764                    &self.$inner.capabilities(),
765                )
766            }
767        });
768        $crate::__faucet_forward_filter!(write_batch_is_replay_safe [$($skip)*] {
769            fn write_batch_is_replay_safe(&self) -> bool
770            {
771                #[allow(unused_imports)]
772                use $crate::Sink as _;
773                self.$inner.write_batch_is_replay_safe()
774            }
775        });
776        $crate::__faucet_forward_filter!(dedups_by_key [$($skip)*] {
777            fn dedups_by_key(&self) -> bool
778            {
779                #[allow(unused_imports)]
780                use $crate::Sink as _;
781                self.$inner.dedups_by_key()
782            }
783        });
784        $crate::__faucet_forward_filter!(supported_write_modes [$($skip)*] {
785            fn supported_write_modes(&self) -> &'static [$crate::write_mode::WriteMode]
786            {
787                #[allow(unused_imports)]
788                use $crate::Sink as _;
789                self.$inner.supported_write_modes()
790            }
791        });
792        $crate::__faucet_forward_filter!(current_schema [$($skip)*] {
793            fn current_schema<'life0, 'async_trait>(
794                &'life0 self,
795            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<$crate::Value>>>
796            where
797                'life0: 'async_trait,
798                Self: 'async_trait,
799            {
800                #[allow(unused_imports)]
801                use $crate::Sink as _;
802                self.$inner.current_schema()
803            }
804        });
805        $crate::__faucet_forward_filter!(supports_schema_evolution [$($skip)*] {
806            fn supports_schema_evolution(&self) -> bool
807            {
808                #[allow(unused_imports)]
809                use $crate::Sink as _;
810                self.$inner.supports_schema_evolution()
811            }
812        });
813        $crate::__faucet_forward_filter!(evolve_schema [$($skip)*] {
814            fn evolve_schema<'life0, 'life1, 'async_trait>(
815                &'life0 self,
816                evolution: &'life1 $crate::drift::SchemaEvolution,
817            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
818            where
819                'life0: 'async_trait,
820                'life1: 'async_trait,
821                Self: 'async_trait,
822            {
823                #[allow(unused_imports)]
824                use $crate::Sink as _;
825                self.$inner.evolve_schema(evolution)
826            }
827        });
828        $crate::__faucet_forward_filter!(supports_cleanup [$($skip)*] {
829            fn supports_cleanup(&self) -> bool
830            {
831                #[allow(unused_imports)]
832                use $crate::Sink as _;
833                self.$inner.supports_cleanup()
834            }
835        });
836        $crate::__faucet_forward_filter!(supports_staged_load [$($skip)*] {
837            fn supports_staged_load(&self) -> bool
838            {
839                #[allow(unused_imports)]
840                use $crate::Sink as _;
841                self.$inner.supports_staged_load()
842            }
843        });
844        $crate::__faucet_forward_filter!(cleanup_scope [$($skip)*] {
845            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
846            fn cleanup_scope<'life0, 'life1, 'life2, 'async_trait>(
847                &'life0 self,
848                scope: &'life1 ::std::collections::BTreeMap<::std::string::String, $crate::Value>,
849                seen: &'life2 $crate::cleanup::SeenKeys,
850            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<u64>>
851            where
852                'life0: 'async_trait,
853                'life1: 'async_trait,
854                'life2: 'async_trait,
855                Self: 'async_trait,
856            {
857                #[allow(unused_imports)]
858                use $crate::Sink as _;
859                self.$inner.cleanup_scope(scope, seen)
860            }
861        });
862        $crate::__faucet_forward_filter!(write_batch_idempotent [$($skip)*] {
863            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
864            fn write_batch_idempotent<'life0, 'life1, 'life2, 'life3, 'async_trait>(
865                &'life0 self,
866                records: &'life1 [$crate::Value],
867                scope: &'life2 str,
868                token: &'life3 str,
869            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<usize>>
870            where
871                'life0: 'async_trait,
872                'life1: 'async_trait,
873                'life2: 'async_trait,
874                'life3: 'async_trait,
875                Self: 'async_trait,
876            {
877                #[allow(unused_imports)]
878                use $crate::Sink as _;
879                self.$inner.write_batch_idempotent(records, scope, token)
880            }
881        });
882        $crate::__faucet_forward_filter!(write_batch_idempotent_reported [$($skip)*] {
883            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
884            fn write_batch_idempotent_reported<'life0, 'life1, 'life2, 'life3, 'async_trait>(
885                &'life0 self,
886                records: &'life1 [$crate::Value],
887                scope: &'life2 str,
888                token: &'life3 str,
889            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::accounting::WriteReport>>
890            where
891                'life0: 'async_trait,
892                'life1: 'async_trait,
893                'life2: 'async_trait,
894                'life3: 'async_trait,
895                Self: 'async_trait,
896            {
897                #[allow(unused_imports)]
898                use $crate::Sink as _;
899                self.$inner.write_batch_idempotent_reported(records, scope, token)
900            }
901        });
902        $crate::__faucet_forward_filter!(last_committed_token [$($skip)*] {
903            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
904            fn last_committed_token<'life0, 'life1, 'async_trait>(
905                &'life0 self,
906                scope: &'life1 str,
907            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<::std::string::String>>>
908            where
909                'life0: 'async_trait,
910                'life1: 'async_trait,
911                Self: 'async_trait,
912            {
913                #[allow(unused_imports)]
914                use $crate::Sink as _;
915                self.$inner.last_committed_token(scope)
916            }
917        });
918        $crate::__faucet_forward_filter!(is_overwrite [$($skip)*] {
919            fn is_overwrite(&self) -> bool
920            {
921                #[allow(unused_imports)]
922                use $crate::Sink as _;
923                self.$inner.is_overwrite()
924            }
925        });
926        $crate::__faucet_forward_filter!(begin_overwrite [$($skip)*] {
927            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
928            fn begin_overwrite<'life0, 'async_trait>(
929                &'life0 self,
930            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
931            where
932                'life0: 'async_trait,
933                Self: 'async_trait,
934            {
935                #[allow(unused_imports)]
936                use $crate::Sink as _;
937                self.$inner.begin_overwrite()
938            }
939        });
940        $crate::__faucet_forward_filter!(commit_overwrite [$($skip)*] {
941            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
942            fn commit_overwrite<'life0, 'async_trait>(
943                &'life0 self,
944            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
945            where
946                'life0: 'async_trait,
947                Self: 'async_trait,
948            {
949                #[allow(unused_imports)]
950                use $crate::Sink as _;
951                self.$inner.commit_overwrite()
952            }
953        });
954        $crate::__faucet_forward_filter!(supports_rollback [$($skip)*] {
955            fn supports_rollback(&self) -> bool
956            {
957                #[allow(unused_imports)]
958                use $crate::Sink as _;
959                self.$inner.supports_rollback()
960            }
961        });
962        $crate::__faucet_forward_filter!(rollback_run [$($skip)*] {
963            fn rollback_run<'life0, 'life1, 'life2, 'async_trait>(
964                &'life0 self,
965                run_id: &'life1 str,
966                opts: &'life2 $crate::rollback::RollbackOptions,
967            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::rollback::RollbackOutcome>>
968            where
969                'life0: 'async_trait,
970                'life1: 'async_trait,
971                'life2: 'async_trait,
972                Self: 'async_trait,
973            {
974                #[allow(unused_imports)]
975                use $crate::Sink as _;
976                self.$inner.rollback_run(run_id, opts)
977            }
978        });
979        $crate::__faucet_forward_filter!(forget_run [$($skip)*] {
980            fn forget_run<'life0, 'life1, 'async_trait>(
981                &'life0 self,
982                run_id: &'life1 str,
983            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
984            where
985                'life0: 'async_trait,
986                'life1: 'async_trait,
987                Self: 'async_trait,
988            {
989                #[allow(unused_imports)]
990                use $crate::Sink as _;
991                self.$inner.forget_run(run_id)
992            }
993        });
994        $crate::__faucet_forward_filter!(rewind_commit_token [$($skip)*] {
995            fn rewind_commit_token<'life0, 'life1, 'life2, 'async_trait>(
996                &'life0 self,
997                scope: &'life1 str,
998                token: ::core::option::Option<&'life2 str>,
999            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
1000            where
1001                'life0: 'async_trait,
1002                'life1: 'async_trait,
1003                'life2: 'async_trait,
1004                Self: 'async_trait,
1005            {
1006                #[allow(unused_imports)]
1007                use $crate::Sink as _;
1008                self.$inner.rewind_commit_token(scope, token)
1009            }
1010        });
1011        $crate::__faucet_forward_filter!(readback_source [$($skip)*] {
1012            fn readback_source(
1013                &self,
1014            ) -> ::core::option::Option<(::std::string::String, $crate::Value)>
1015            {
1016                #[allow(unused_imports)]
1017                use $crate::Sink as _;
1018                self.$inner.readback_source()
1019            }
1020        });
1021        $crate::__faucet_forward_filter!(abort_overwrite [$($skip)*] {
1022            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
1023            fn abort_overwrite<'life0, 'async_trait>(
1024                &'life0 self,
1025            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
1026            where
1027                'life0: 'async_trait,
1028                Self: 'async_trait,
1029            {
1030                #[allow(unused_imports)]
1031                use $crate::Sink as _;
1032                self.$inner.abort_overwrite()
1033            }
1034        });
1035        $crate::__faucet_forward_filter!(complete_run [$($skip)*] {
1036            fn complete_run<'life0, 'async_trait>(
1037                &'life0 self,
1038            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<()>>
1039            where
1040                'life0: 'async_trait,
1041                Self: 'async_trait,
1042            {
1043                #[allow(unused_imports)]
1044                use $crate::Sink as _;
1045                self.$inner.complete_run()
1046            }
1047        });
1048        $crate::__faucet_forward_filter!(overwrite_staging_exists [$($skip)*] {
1049            #[allow(deprecated)] // forwards a hook the family drivers replace (RFC 0014)
1050            fn overwrite_staging_exists<'life0, 'async_trait>(
1051                &'life0 self,
1052            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<::core::option::Option<bool>>>
1053            where
1054                'life0: 'async_trait,
1055                Self: 'async_trait,
1056            {
1057                #[allow(unused_imports)]
1058                use $crate::Sink as _;
1059                self.$inner.overwrite_staging_exists()
1060            }
1061        });
1062        $crate::__faucet_forward_filter!(config_schema [$($skip)*] {
1063            fn config_schema(&self) -> $crate::Value
1064            {
1065                #[allow(unused_imports)]
1066                use $crate::Sink as _;
1067                self.$inner.config_schema()
1068            }
1069        });
1070        $crate::__faucet_forward_filter!(connector_name [$($skip)*] {
1071            fn connector_name(&self) -> &'static str
1072            {
1073                #[allow(unused_imports)]
1074                use $crate::Sink as _;
1075                self.$inner.connector_name()
1076            }
1077        });
1078        $crate::__faucet_forward_filter!(set_roundtrip_recorder [$($skip)*] {
1079            fn set_roundtrip_recorder(
1080                &self,
1081                recorder: ::std::sync::Arc<$crate::observability::RoundtripRecorder>,
1082            ) -> ()
1083            {
1084                #[allow(unused_imports)]
1085                use $crate::Sink as _;
1086                self.$inner.set_roundtrip_recorder(recorder)
1087            }
1088        });
1089        $crate::__faucet_forward_filter!(set_call_context [$($skip)*] {
1090            fn set_call_context(&self, ctx: $crate::io::CallContext) -> ()
1091            {
1092                #[allow(unused_imports)]
1093                use $crate::Sink as _;
1094                self.$inner.set_call_context(ctx)
1095            }
1096        });
1097        $crate::__faucet_forward_filter!(dataset_uri [$($skip)*] {
1098            fn dataset_uri(&self) -> ::std::string::String
1099            {
1100                #[allow(unused_imports)]
1101                use $crate::Sink as _;
1102                self.$inner.dataset_uri()
1103            }
1104        });
1105        $crate::__faucet_forward_filter!(local_outputs [$($skip)*] {
1106            fn local_outputs<'life0, 'async_trait>(
1107                &'life0 self,
1108            ) -> $crate::forward::__forward::Fut<'async_trait, ::std::vec::Vec<$crate::local_outputs::LocalOutput>>
1109            where
1110                'life0: 'async_trait,
1111                Self: 'async_trait,
1112            {
1113                #[allow(unused_imports)]
1114                use $crate::Sink as _;
1115                self.$inner.local_outputs()
1116            }
1117        });
1118        $crate::__faucet_forward_filter!(check [$($skip)*] {
1119            fn check<'life0, 'life1, 'async_trait>(
1120                &'life0 self,
1121                ctx: &'life1 $crate::check::CheckContext,
1122            ) -> $crate::forward::__forward::Fut<'async_trait, $crate::forward::__forward::Res<$crate::check::CheckReport>>
1123            where
1124                'life0: 'async_trait,
1125                'life1: 'async_trait,
1126                Self: 'async_trait,
1127            {
1128                #[allow(unused_imports)]
1129                use $crate::Sink as _;
1130                self.$inner.check(ctx)
1131            }
1132        });
1133    };
1134}
1135
1136#[doc(hidden)]
1137#[macro_export]
1138macro_rules! __faucet_forward_filter {
1139    (fetch_with_context [fetch_with_context $($rest:ident)*] $item:tt) => {};
1140    (fetch_all [fetch_all $($rest:ident)*] $item:tt) => {};
1141    (fetch_with_context_incremental [fetch_with_context_incremental $($rest:ident)*] $item:tt) => {};
1142    (fetch_all_incremental [fetch_all_incremental $($rest:ident)*] $item:tt) => {};
1143    (stream_pages [stream_pages $($rest:ident)*] $item:tt) => {};
1144    (supports_columnar [supports_columnar $($rest:ident)*] $item:tt) => {};
1145    (stream_batches [stream_batches $($rest:ident)*] $item:tt) => {};
1146    (native_output_formats [native_output_formats $($rest:ident)*] $item:tt) => {};
1147    (stream_native [stream_native $($rest:ident)*] $item:tt) => {};
1148    (config_schema [config_schema $($rest:ident)*] $item:tt) => {};
1149    (state_key [state_key $($rest:ident)*] $item:tt) => {};
1150    (apply_start_bookmark [apply_start_bookmark $($rest:ident)*] $item:tt) => {};
1151    (capture_resume_position [capture_resume_position $($rest:ident)*] $item:tt) => {};
1152    (lag [lag $($rest:ident)*] $item:tt) => {};
1153    (state_schema [state_schema $($rest:ident)*] $item:tt) => {};
1154    (migrate_state [migrate_state $($rest:ident)*] $item:tt) => {};
1155    (supports_exactly_once [supports_exactly_once $($rest:ident)*] $item:tt) => {};
1156    (consumes_destructively [consumes_destructively $($rest:ident)*] $item:tt) => {};
1157    (replay_guarantee [replay_guarantee $($rest:ident)*] $item:tt) => {};
1158    (capabilities [capabilities $($rest:ident)*] $item:tt) => {};
1159    (is_shardable [is_shardable $($rest:ident)*] $item:tt) => {};
1160    (enumerate_shards [enumerate_shards $($rest:ident)*] $item:tt) => {};
1161    (apply_shard [apply_shard $($rest:ident)*] $item:tt) => {};
1162    (range_digest [range_digest $($rest:ident)*] $item:tt) => {};
1163    (supports_discover [supports_discover $($rest:ident)*] $item:tt) => {};
1164    (discover [discover $($rest:ident)*] $item:tt) => {};
1165    (record_table [record_table $($rest:ident)*] $item:tt) => {};
1166    (position_le [position_le $($rest:ident)*] $item:tt) => {};
1167    (record_covered [record_covered $($rest:ident)*] $item:tt) => {};
1168    (position_min [position_min $($rest:ident)*] $item:tt) => {};
1169    (connector_name [connector_name $($rest:ident)*] $item:tt) => {};
1170    (set_roundtrip_recorder [set_roundtrip_recorder $($rest:ident)*] $item:tt) => {};
1171    (set_call_context [set_call_context $($rest:ident)*] $item:tt) => {};
1172    (set_run_clock [set_run_clock $($rest:ident)*] $item:tt) => {};
1173    (set_cancel [set_cancel $($rest:ident)*] $item:tt) => {};
1174    (dataset_uri [dataset_uri $($rest:ident)*] $item:tt) => {};
1175    (check [check $($rest:ident)*] $item:tt) => {};
1176    (write_batch [write_batch $($rest:ident)*] $item:tt) => {};
1177    (write_batch_reported [write_batch_reported $($rest:ident)*] $item:tt) => {};
1178    // A wrapper that changes `write_batch` takes the trait default for the
1179    // reported write, which goes through its own `write_batch` (#879).
1180    (write_batch_reported [write_batch $($rest:ident)*] $item:tt) => {};
1181    (write_governed [write_governed $($rest:ident)*] $item:tt) => {};
1182    // A wrapper changing the plain or reported write keeps the default governed write.
1183    (write_governed [write_batch $($rest:ident)*] $item:tt) => {};
1184    (write_governed [write_batch_reported $($rest:ident)*] $item:tt) => {};
1185    (flush [flush $($rest:ident)*] $item:tt) => {};
1186    (write_batch_partial [write_batch_partial $($rest:ident)*] $item:tt) => {};
1187    (batch_atomicity [batch_atomicity $($rest:ident)*] $item:tt) => {};
1188    (admit_page [admit_page $($rest:ident)*] $item:tt) => {};
1189    (write_batch_columnar [write_batch_columnar $($rest:ident)*] $item:tt) => {};
1190    (native_load_capabilities [native_load_capabilities $($rest:ident)*] $item:tt) => {};
1191    (load_native [load_native $($rest:ident)*] $item:tt) => {};
1192    (supports_idempotent_writes [supports_idempotent_writes $($rest:ident)*] $item:tt) => {};
1193    (sink_guarantee [sink_guarantee $($rest:ident)*] $item:tt) => {};
1194    (write_batch_is_replay_safe [write_batch_is_replay_safe $($rest:ident)*] $item:tt) => {};
1195    (dedups_by_key [dedups_by_key $($rest:ident)*] $item:tt) => {};
1196    (supported_write_modes [supported_write_modes $($rest:ident)*] $item:tt) => {};
1197    (current_schema [current_schema $($rest:ident)*] $item:tt) => {};
1198    (supports_schema_evolution [supports_schema_evolution $($rest:ident)*] $item:tt) => {};
1199    (evolve_schema [evolve_schema $($rest:ident)*] $item:tt) => {};
1200    (supports_cleanup [supports_cleanup $($rest:ident)*] $item:tt) => {};
1201    (supports_staged_load [supports_staged_load $($rest:ident)*] $item:tt) => {};
1202    (cleanup_scope [cleanup_scope $($rest:ident)*] $item:tt) => {};
1203    (write_batch_idempotent [write_batch_idempotent $($rest:ident)*] $item:tt) => {};
1204    (write_batch_idempotent_reported [write_batch_idempotent_reported $($rest:ident)*] $item:tt) => {};
1205    // A wrapper that changes `write_batch_idempotent` takes the trait default
1206    // for the reported write, which goes through its own change (#879).
1207    (write_batch_idempotent_reported [write_batch_idempotent $($rest:ident)*] $item:tt) => {};
1208    (last_committed_token [last_committed_token $($rest:ident)*] $item:tt) => {};
1209    (is_overwrite [is_overwrite $($rest:ident)*] $item:tt) => {};
1210    (begin_overwrite [begin_overwrite $($rest:ident)*] $item:tt) => {};
1211    (commit_overwrite [commit_overwrite $($rest:ident)*] $item:tt) => {};
1212    (supports_rollback [supports_rollback $($rest:ident)*] $item:tt) => {};
1213    (rollback_run [rollback_run $($rest:ident)*] $item:tt) => {};
1214    (forget_run [forget_run $($rest:ident)*] $item:tt) => {};
1215    (rewind_commit_token [rewind_commit_token $($rest:ident)*] $item:tt) => {};
1216    (readback_source [readback_source $($rest:ident)*] $item:tt) => {};
1217    (abort_overwrite [abort_overwrite $($rest:ident)*] $item:tt) => {};
1218    (complete_run [complete_run $($rest:ident)*] $item:tt) => {};
1219    (overwrite_staging_exists [overwrite_staging_exists $($rest:ident)*] $item:tt) => {};
1220    (local_outputs [local_outputs $($rest:ident)*] $item:tt) => {};
1221    ($hook:ident [] { $($item:tt)* }) => { $($item)* };
1222    ($hook:ident [$other:ident $($rest:ident)*] $item:tt) => {
1223        $crate::__faucet_forward_filter!($hook [$($rest)*] $item);
1224    };
1225}
1226
1227#[doc(hidden)]
1228#[macro_export]
1229macro_rules! __faucet_forward_known {
1230    (@source [fetch_with_context $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1231    (@source [fetch_all $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1232    (@source [fetch_with_context_incremental $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1233    (@source [fetch_all_incremental $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1234    (@source [stream_pages $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1235    (@source [supports_columnar $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1236    (@source [stream_batches $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1237    (@source [native_output_formats $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1238    (@source [stream_native $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1239    (@source [config_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1240    (@source [state_key $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1241    (@source [apply_start_bookmark $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1242    (@source [capture_resume_position $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1243    (@source [lag $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1244    (@source [state_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1245    (@source [migrate_state $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1246    (@source [supports_exactly_once $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1247    (@source [consumes_destructively $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1248    (@source [replay_guarantee $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1249    (@source [capabilities $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1250    (@source [is_shardable $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1251    (@source [enumerate_shards $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1252    (@source [apply_shard $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1253    (@source [range_digest $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1254    (@source [supports_discover $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1255    (@source [discover $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1256    (@source [record_table $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1257    (@source [position_le $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1258    (@source [record_covered $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1259    (@source [position_min $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1260    (@source [connector_name $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1261    (@source [set_roundtrip_recorder $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1262    (@source [set_call_context $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1263    (@source [set_run_clock $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1264    (@source [set_cancel $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1265    (@source [dataset_uri $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1266    (@source [check $($rest:ident)*]) => { $crate::__faucet_forward_known!(@source [$($rest)*]); };
1267    (@sink [write_batch $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1268    (@sink [write_batch_reported $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1269    (@sink [write_governed $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1270    (@sink [flush $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1271    (@sink [write_batch_partial $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1272    (@sink [batch_atomicity $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1273    (@sink [admit_page $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1274    (@sink [supports_columnar $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1275    (@sink [write_batch_columnar $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1276    (@sink [native_load_capabilities $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1277    (@sink [load_native $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1278    (@sink [supports_idempotent_writes $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1279    (@sink [sink_guarantee $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1280    (@sink [capabilities $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1281    (@sink [write_batch_is_replay_safe $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1282    (@sink [dedups_by_key $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1283    (@sink [supported_write_modes $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1284    (@sink [current_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1285    (@sink [supports_schema_evolution $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1286    (@sink [evolve_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1287    (@sink [supports_cleanup $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1288    (@sink [supports_staged_load $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1289    (@sink [cleanup_scope $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1290    (@sink [write_batch_idempotent $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1291    (@sink [write_batch_idempotent_reported $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1292    (@sink [last_committed_token $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1293    (@sink [is_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1294    (@sink [begin_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1295    (@sink [commit_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1296    (@sink [supports_rollback $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1297    (@sink [rollback_run $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1298    (@sink [forget_run $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1299    (@sink [rewind_commit_token $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1300    (@sink [readback_source $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1301    (@sink [abort_overwrite $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1302    (@sink [complete_run $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1303    (@sink [overwrite_staging_exists $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1304    (@sink [config_schema $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1305    (@sink [connector_name $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1306    (@sink [set_roundtrip_recorder $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1307    (@sink [set_call_context $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1308    (@sink [dataset_uri $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1309    (@sink [local_outputs $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1310    (@sink [check $($rest:ident)*]) => { $crate::__faucet_forward_known!(@sink [$($rest)*]); };
1311    (@$tr:ident []) => {};
1312    (@source [$other:ident $($rest:ident)*]) => {
1313        ::core::compile_error!(::core::concat!(
1314            "forward_source!: `", ::core::stringify!($other), "` is not a Source hook"
1315        ));
1316    };
1317    (@sink [$other:ident $($rest:ident)*]) => {
1318        ::core::compile_error!(::core::concat!(
1319            "forward_sink!: `", ::core::stringify!($other), "` is not a Sink hook"
1320        ));
1321    };
1322}
faucet_core/forward.rs

faucet_core/
forward.rs
