diff --git a/crates/core/src/operations/constraints.rs b/crates/core/src/operations/constraints.rs
index ee5ed16da9..1c5be2ccfc 100644
--- a/crates/core/src/operations/constraints.rs
+++ b/crates/core/src/operations/constraints.rs
@@ -41,6 +41,8 @@ pub struct ConstraintBuilder {
     commit_properties: CommitProperties,
 }
 
+impl super::Operation<()> for ConstraintBuilder {}
+
 impl ConstraintBuilder {
     /// Create a new builder
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/convert_to_delta.rs b/crates/core/src/operations/convert_to_delta.rs
index 351a596062..44475d0e92 100644
--- a/crates/core/src/operations/convert_to_delta.rs
+++ b/crates/core/src/operations/convert_to_delta.rs
@@ -116,6 +116,8 @@ impl Default for ConvertToDeltaBuilder {
     }
 }
 
+impl super::Operation<()> for ConvertToDeltaBuilder {}
+
 impl ConvertToDeltaBuilder {
     /// Create a new [`ConvertToDeltaBuilder`]
     pub fn new() -> Self {
diff --git a/crates/core/src/operations/create.rs b/crates/core/src/operations/create.rs
index b2092cafe8..04b71804d4 100644
--- a/crates/core/src/operations/create.rs
+++ b/crates/core/src/operations/create.rs
@@ -60,6 +60,8 @@ pub struct CreateBuilder {
     metadata: Option<HashMap<String, Value>>,
 }
 
+impl super::Operation<()> for CreateBuilder {}
+
 impl Default for CreateBuilder {
     fn default() -> Self {
         Self::new()
diff --git a/crates/core/src/operations/delete.rs b/crates/core/src/operations/delete.rs
index e28633ae17..6fb76891b1 100644
--- a/crates/core/src/operations/delete.rs
+++ b/crates/core/src/operations/delete.rs
@@ -82,6 +82,8 @@ pub struct DeleteMetrics {
     pub rewrite_time_ms: u128,
 }
 
+impl super::Operation<()> for DeleteBuilder {}
+
 impl DeleteBuilder {
     /// Create a new [`DeleteBuilder`]
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/drop_constraints.rs b/crates/core/src/operations/drop_constraints.rs
index 5a1e448812..081262753e 100644
--- a/crates/core/src/operations/drop_constraints.rs
+++ b/crates/core/src/operations/drop_constraints.rs
@@ -24,6 +24,8 @@ pub struct DropConstraintBuilder {
     commit_properties: CommitProperties,
 }
 
+impl super::Operation<()> for DropConstraintBuilder {}
+
 impl DropConstraintBuilder {
     /// Create a new builder
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/filesystem_check.rs b/crates/core/src/operations/filesystem_check.rs
index ada8f35251..0b5cfa5707 100644
--- a/crates/core/src/operations/filesystem_check.rs
+++ b/crates/core/src/operations/filesystem_check.rs
@@ -75,6 +75,8 @@ fn is_absolute_path(path: &str) -> DeltaResult<bool> {
     }
 }
 
+impl super::Operation<()> for FileSystemCheckBuilder {}
+
 impl FileSystemCheckBuilder {
     /// Create a new [`FileSystemCheckBuilder`]
     pub fn new(log_store: LogStoreRef, state: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/load.rs b/crates/core/src/operations/load.rs
index c89e63ebab..4bf439cd0d 100644
--- a/crates/core/src/operations/load.rs
+++ b/crates/core/src/operations/load.rs
@@ -23,6 +23,8 @@ pub struct LoadBuilder {
     columns: Option<Vec<String>>,
 }
 
+impl super::Operation<()> for LoadBuilder {}
+
 impl LoadBuilder {
     /// Create a new [`LoadBuilder`]
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/merge/mod.rs b/crates/core/src/operations/merge/mod.rs
index 206ee0e899..b8cf61c9a6 100644
--- a/crates/core/src/operations/merge/mod.rs
+++ b/crates/core/src/operations/merge/mod.rs
@@ -133,6 +133,8 @@ pub struct MergeBuilder {
     safe_cast: bool,
 }
 
+impl super::Operation<()> for MergeBuilder {}
+
 impl MergeBuilder {
     /// Create a new [`MergeBuilder`]
     pub fn new<E: Into<Expression>>(
diff --git a/crates/core/src/operations/mod.rs b/crates/core/src/operations/mod.rs
index 666b2dc66a..90653fea5f 100644
--- a/crates/core/src/operations/mod.rs
+++ b/crates/core/src/operations/mod.rs
@@ -14,6 +14,7 @@ use crate::errors::{DeltaResult, DeltaTableError};
 use crate::table::builder::DeltaTableBuilder;
 use crate::DeltaTable;
 use std::collections::HashMap;
+use std::sync::Arc;
 
 pub mod cast;
 pub mod convert_to_delta;
@@ -52,7 +53,13 @@ pub mod update;
 pub mod write;
 pub mod writer;
 
-// TODO make ops consume a snapshot ...
+/// The [Operation] trait defines common behaviors that all operations builders
+/// should have consistent
+pub(crate) trait Operation<State>: std::future::IntoFuture {
+    fn with(self, ware: Arc<dyn middleware::Transactional<State>>) -> Self where Self: Sized {
+        unimplemented!("The default Operation implementation cannot handle middleware!");
+    }
+}
 
 /// High level interface for executing commands against a DeltaTable
 pub struct DeltaOps(pub DeltaTable);
diff --git a/crates/core/src/operations/optimize.rs b/crates/core/src/operations/optimize.rs
index babe17a6a0..614fdf0d99 100644
--- a/crates/core/src/operations/optimize.rs
+++ b/crates/core/src/operations/optimize.rs
@@ -211,6 +211,8 @@ pub struct OptimizeBuilder<'a> {
     min_commit_interval: Option<Duration>,
 }
 
+impl super::Operation<()> for OptimizeBuilder<'_> {}
+
 impl<'a> OptimizeBuilder<'a> {
     /// Create a new [`OptimizeBuilder`]
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/restore.rs b/crates/core/src/operations/restore.rs
index a79fb389f1..4f4cd0568d 100644
--- a/crates/core/src/operations/restore.rs
+++ b/crates/core/src/operations/restore.rs
@@ -89,6 +89,8 @@ pub struct RestoreBuilder {
     commit_properties: CommitProperties,
 }
 
+impl super::Operation<()> for RestoreBuilder {}
+
 impl RestoreBuilder {
     /// Create a new [`RestoreBuilder`]
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/update.rs b/crates/core/src/operations/update.rs
index 6f8b595029..b0406fa9a5 100644
--- a/crates/core/src/operations/update.rs
+++ b/crates/core/src/operations/update.rs
@@ -98,6 +98,8 @@ pub struct UpdateMetrics {
     pub scan_time_ms: u64,
 }
 
+impl super::Operation<()> for UpdateBuilder {}
+
 impl UpdateBuilder {
     /// Create a new ['UpdateBuilder']
     pub fn new(log_store: LogStoreRef, snapshot: DeltaTableState) -> Self {
diff --git a/crates/core/src/operations/vacuum.rs b/crates/core/src/operations/vacuum.rs
index fd986b55d6..1541dbe486 100644
--- a/crates/core/src/operations/vacuum.rs
+++ b/crates/core/src/operations/vacuum.rs
@@ -95,6 +95,8 @@ pub struct VacuumBuilder {
     commit_properties: CommitProperties,
 }
 
+impl super::Operation<()> for VacuumBuilder {}
+
 /// Details for the Vacuum operation including which files were
 #[derive(Debug)]
 pub struct VacuumMetrics {
diff --git a/crates/core/src/operations/write.rs b/crates/core/src/operations/write.rs
index 10b48a768c..e87f3b6e50 100644
--- a/crates/core/src/operations/write.rs
+++ b/crates/core/src/operations/write.rs
@@ -152,6 +152,8 @@ pub struct WriteBuilder {
     configuration: HashMap<String, Option<String>>,
 }
 
+impl super::Operation<()> for WriteBuilder {}
+
 impl WriteBuilder {
     /// Create a new [`WriteBuilder`]
     pub fn new(log_store: LogStoreRef, snapshot: Option<DeltaTableState>) -> Self {