Introduce the Operation trait for all operations to implement

Currently this is nothing but a shim that ensures everything implements
IntoFuture, which they all already do, but in the future this will help
enforce consistency as well as provide common behaviors.

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 {

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 {

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()

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 {

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 {

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 {

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 {

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>>(

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);

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 {

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 {

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 {

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 {

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 {