Add '#[deny(clippy::missing_trait_methods)]' to wrapper/delegation trait impls (spiceai#10795)

* Add '#[deny(clippy::missing_trait_methods)]' to wrapper/delegation trait impls

* implement missing

* fix: implement missing delegation methods caught by missing_trait_methods lint

* fix: rustfmt get_logical_plan signature in indexed.rs

* fix: implement missing TableProvider delegation methods in PartitionTableProvider

* fix: correct ScanResult return type and IndexedMemTable::scan_with_args delegation

- runtime-table-partition/src/provider.rs:490: use datafusion::error::Result
  instead of bare Result<T> (no type alias in scope; std::result::Result
  requires two generic args, causing E0107 compile error on all CI jobs)

- data_components/src/arrow/indexed.rs: unpack ScanArgs and call self.scan()
  instead of forwarding to self.inner.scan_with_args(), preserving the
  index fast-path logic that IndexedMemTable::scan implements

* fix: replace redundant closure with method reference (clippy pedantic)

* fix: replace redundant closure with method reference in scan_with_args

* fix: add missing trait method forwarding impls for #[deny(clippy::missing_trait_methods)]

Adds explicit implementations for all provided/default trait methods across
wrapper and stub types so the deny attribute compiles cleanly:

TableProvider wrappers (delegate to inner):
- AcceleratedTable: get_table_definition, get_logical_plan, get_column_default,
  scan_with_args, statistics, truncate
- UpsertDedupTableProvider: same set
- EmbeddingTable: scan_with_args, truncate
- LocationPruningListingTable: get_logical_plan, get_column_default,
  scan_with_args, statistics, insert_into, delete_from

TableProvider with no inner (None / NotImplemented stubs):
- SwappableTableProvider: get_table_definition/get_logical_plan/get_column_default
  return None (can't borrow from current() temporary Arc); scan_with_args and
  truncate delegate via current()
- DatasetTableProvider: all missing methods return Internal errors or None
- RerankUDTFProvider: metadata returns None; DML returns NotImplemented
- ReciprocalRankFusion: same pattern

CatalogProvider wrapper:
- RefreshingCatalogProvider: register_schema, deregister_schema forwarded to inner

DataConnector wrappers:
- DeferredConnector: all missing methods added; stream methods return false/None
  (connector is a pre-initialization placeholder); setup/metrics methods
  delegate to self.inner
- EmbeddingConnector: resolve_refresh_mode, initialization_for_dataset
- FullTextConnector: initialization_for_dataset

Embed wrapper:
- TaskEmbed: cache, get_cached_embed, put_cached_embed delegated to self.inner

* fix: remove #[deny(clippy::missing_trait_methods)] from AcceleratedTable, DeferredConnector, DatasetTableProvider

* linting

* fix: implement missing TableProvider trait methods for LocationPruningListingTable and EmbeddingTable

- Add  and  to LocationPruningListingTable, delegating to inner
- Add  and  to EmbeddingTable, delegating to base_table
- Add missing imports: std::borrow::Cow and datafusion::logical_expr::LogicalPlan

* fix AcceleratedTable

* comments

* clippy

* remove nasa

---------

Co-authored-by: Jeadie <[email protected]>

Author: Jeadie

Cargo.lock

@@ -200,6 +200,7 @@ impl DeletionFilteringVortexFormat {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl FileFormat for DeletionFilteringVortexFormat {
     fn as_any(&self) -> &dyn Any {

crates/cayenne/src/provider/vortex_format.rs

@@ -571,6 +571,7 @@ impl PrimaryKeyValue {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl TableProvider for IndexedMemTable {
     fn as_any(&self) -> &dyn Any {
@@ -806,6 +807,47 @@ impl TableProvider for IndexedMemTable {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         self.inner.delete_from(state, filters).await
     }
+
+    fn get_table_definition(&self) -> Option<&str> {
+        self.inner.get_table_definition()
+    }
+
+    fn get_logical_plan(
+        &self,
+    ) -> Option<std::borrow::Cow<'_, datafusion::logical_expr::LogicalPlan>> {
+        self.inner.get_logical_plan()
+    }
+
+    async fn scan_with_args<'a>(
+        &self,
+        state: &dyn Session,
+        args: datafusion::catalog::ScanArgs<'a>,
+    ) -> Result<datafusion::catalog::ScanResult> {
+        let filters = args.filters().unwrap_or(&[]);
+        let projection = args.projection().map(<[usize]>::to_vec);
+        let limit = args.limit();
+        let plan = self
+            .scan(state, projection.as_ref(), filters, limit)
+            .await?;
+        Ok(datafusion::catalog::ScanResult::new(plan))
+    }
+
+    fn statistics(&self) -> Option<datafusion::common::Statistics> {
+        self.inner.statistics()
+    }
+
+    async fn update(
+        &self,
+        state: &dyn Session,
+        assignments: Vec<(String, Expr)>,
+        filters: Vec<Expr>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        self.inner.update(state, assignments, filters).await
+    }
+
+    async fn truncate(&self, state: &dyn Session) -> Result<Arc<dyn ExecutionPlan>> {
+        self.inner.truncate(state).await
+    }
 }
 
 #[async_trait]

crates/data_components/src/arrow/indexed.rs

@@ -110,6 +110,7 @@ impl DuckDbFederatedTableWriter {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl TableProvider for DuckDbFederatedTableWriter {
     fn as_any(&self) -> &dyn Any {
@@ -196,6 +197,27 @@ impl TableProvider for DuckDbFederatedTableWriter {
 
         Ok(Arc::new(DataSinkExec::new(input, Arc::new(sink), None)) as _)
     }
+
+    async fn delete_from(
+        &self,
+        state: &dyn Session,
+        filters: Vec<Expr>,
+    ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        self.read_provider.delete_from(state, filters).await
+    }
+
+    async fn update(
+        &self,
+        state: &dyn Session,
+        assignments: Vec<(String, Expr)>,
+        filters: Vec<Expr>,
+    ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        self.read_provider.update(state, assignments, filters).await
+    }
+
+    async fn truncate(&self, state: &dyn Session) -> DFResult<Arc<dyn ExecutionPlan>> {
+        self.read_provider.truncate(state).await
+    }
 }
 
 struct DuckDbFederatedDataSink {

crates/data_components/src/ducklake/writer.rs

@@ -199,6 +199,7 @@ impl RefreshSkipTableProvider for S3SingleFileCached {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl TableProvider for S3SingleFileCached {
     fn as_any(&self) -> &dyn Any {
@@ -258,6 +259,35 @@ impl TableProvider for S3SingleFileCached {
         // The metadata check happens during refresh, not during scan
         self.inner.scan(state, projection, filters, limit).await
     }
+
+    async fn scan_with_args<'a>(
+        &self,
+        state: &dyn Session,
+        args: datafusion::catalog::ScanArgs<'a>,
+    ) -> DataFusionResult<datafusion::catalog::ScanResult> {
+        self.inner.scan_with_args(state, args).await
+    }
+
+    async fn delete_from(
+        &self,
+        state: &dyn Session,
+        filters: Vec<Expr>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        self.inner.delete_from(state, filters).await
+    }
+
+    async fn update(
+        &self,
+        state: &dyn Session,
+        assignments: Vec<(String, Expr)>,
+        filters: Vec<Expr>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        self.inner.update(state, assignments, filters).await
+    }
+
+    async fn truncate(&self, state: &dyn Session) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        self.inner.truncate(state).await
+    }
 }
 
 #[cfg(test)]

crates/data_components/src/s3_single_file_cached.rs

@@ -7,6 +7,7 @@ description = "Type-safe object storage with optimistic concurrency control"
 
 [dependencies]
 async-trait.workspace = true
+bytes.workspace = true
 futures.workspace = true
 object_store.workspace = true
 parking_lot.workspace = true

crates/object_store_occ/Cargo.toml

@@ -34,9 +34,11 @@ limitations under the License.
 use std::fmt::{self, Display, Formatter};
 use std::fs::{File, OpenOptions};
 use std::io;
+use std::ops::Range;
 use std::path::PathBuf;
 
 use async_trait::async_trait;
+use bytes::Bytes;
 use futures::stream::BoxStream;
 use object_store::local::LocalFileSystem;
 use object_store::path::Path;
@@ -141,6 +143,7 @@ impl Display for LocalConditionalPut {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl ObjectStore for LocalConditionalPut {
     async fn put_opts(
@@ -237,6 +240,53 @@ impl ObjectStore for LocalConditionalPut {
     async fn rename_if_not_exists(&self, from: &Path, to: &Path) -> Result<(), ObjectStoreError> {
         self.inner.rename_if_not_exists(from, to).await
     }
+
+    async fn put(
+        &self,
+        location: &Path,
+        payload: PutPayload,
+    ) -> Result<PutResult, ObjectStoreError> {
+        self.put_opts(location, payload, PutOptions::default())
+            .await
+    }
+
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<Box<dyn MultipartUpload>, ObjectStoreError> {
+        self.inner.put_multipart(location).await
+    }
+
+    async fn get(&self, location: &Path) -> Result<GetResult, ObjectStoreError> {
+        self.inner.get(location).await
+    }
+
+    async fn get_range(
+        &self,
+        location: &Path,
+        range: Range<u64>,
+    ) -> Result<Bytes, ObjectStoreError> {
+        self.inner.get_range(location, range).await
+    }
+
+    async fn get_ranges(
+        &self,
+        location: &Path,
+        ranges: &[Range<u64>],
+    ) -> Result<Vec<Bytes>, ObjectStoreError> {
+        self.inner.get_ranges(location, ranges).await
+    }
+
+    fn delete_stream<'a>(
+        &'a self,
+        locations: BoxStream<'a, Result<Path, ObjectStoreError>>,
+    ) -> BoxStream<'a, Result<Path, ObjectStoreError>> {
+        self.inner.delete_stream(locations)
+    }
+
+    async fn rename(&self, from: &Path, to: &Path) -> Result<(), ObjectStoreError> {
+        self.inner.rename(from, to).await
+    }
 }
 
 #[cfg(test)]

crates/object_store_occ/src/local_conditional_put.rs

@@ -19,7 +19,7 @@ use std::{any::Any, borrow::Cow, sync::Arc};
 use async_trait::async_trait;
 use datafusion::{
     arrow::datatypes::SchemaRef,
-    catalog::{Session, TableProvider},
+    catalog::{ScanArgs, ScanResult, Session, TableProvider},
     common::{Constraints, Statistics},
     datasource::TableType,
     error::Result as DataFusionResult,
@@ -95,6 +95,7 @@ impl IndexedTableProvider {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl TableProvider for IndexedTableProvider {
     fn as_any(&self) -> &dyn Any {
@@ -175,4 +176,16 @@ impl TableProvider for IndexedTableProvider {
     ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
         self.underlying.update(state, assignments, filters).await
     }
+
+    async fn scan_with_args<'a>(
+        &self,
+        state: &dyn Session,
+        args: ScanArgs<'a>,
+    ) -> DataFusionResult<ScanResult> {
+        self.underlying.scan_with_args(state, args).await
+    }
+
+    async fn truncate(&self, state: &dyn Session) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        self.underlying.truncate(state).await
+    }
 }

crates/runtime-datafusion-index/src/provider.rs

@@ -708,6 +708,7 @@ impl SpiceObjectStoreRegistry {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 impl ObjectStoreRegistry for SpiceObjectStoreRegistry {
     fn register_store(
         &self,
@@ -724,6 +725,10 @@ impl ObjectStoreRegistry for SpiceObjectStoreRegistry {
             Ok(store)
         })
     }
+
+    fn deregister_store(&self, url: &Url) -> datafusion::error::Result<Arc<dyn ObjectStore>> {
+        self.inner.deregister_store(url)
+    }
 }
 
 // This method uses unwrap_or_default, however it should never fail on the initialization. See

crates/runtime-object-store/src/registry/mod.rs

@@ -288,6 +288,7 @@ impl PartitionTableProvider {
     }
 }
 
+#[deny(clippy::missing_trait_methods)]
 #[async_trait]
 impl TableProvider for PartitionTableProvider {
     fn as_any(&self) -> &dyn Any {
@@ -468,6 +469,47 @@ impl TableProvider for PartitionTableProvider {
         Ok(plan)
     }
 
+    fn get_table_definition(&self) -> Option<&str> {
+        None
+    }
+
+    fn get_logical_plan(
+        &self,
+    ) -> Option<std::borrow::Cow<'_, datafusion::logical_expr::LogicalPlan>> {
+        None
+    }
+
+    fn get_column_default(&self, _column: &str) -> Option<&Expr> {
+        None
+    }
+
+    async fn scan_with_args<'a>(
+        &self,
+        state: &dyn Session,
+        args: datafusion::catalog::ScanArgs<'a>,
+    ) -> datafusion::error::Result<datafusion::catalog::ScanResult> {
+        let filters = args.filters().unwrap_or(&[]);
+        let projection = args.projection().map(<[usize]>::to_vec);
+        let limit = args.limit();
+        let plan = self
+            .scan(state, projection.as_ref(), filters, limit)
+            .await?;
+        Ok(datafusion::catalog::ScanResult::new(plan))
+    }
+
+    fn statistics(&self) -> Option<datafusion::common::Statistics> {
+        None
+    }
+
+    async fn truncate(
+        &self,
+        _state: &dyn Session,
+    ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+        Err(DataFusionError::NotImplemented(
+            "PartitionTableProvider does not support truncate".to_string(),
+        ))
+    }
+
     async fn insert_into(
         &self,
         _state: &dyn Session,