feat: parse json predicates

delta-sharing/client/src/gen/delta_sharing.sharing.v1.rs

@@ -302,6 +302,31 @@ pub struct MetadatDelta {
     #[prost(int64, optional, tag="4")]
     pub num_files: ::core::option::Option<i64>,
 }
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct JsonPredicate {
+    /// The operation to perform.
+    #[prost(string, tag="1")]
+    pub op: ::prost::alloc::string::String,
+    /// The inputs to the operation - if any.
+    #[prost(message, repeated, tag="2")]
+    pub children: ::prost::alloc::vec::Vec<JsonPredicate>,
+    /// Specifies the name of a column.
+    ///
+    /// This is only applicable to column ops.
+    #[prost(string, optional, tag="3")]
+    pub name: ::core::option::Option<::prost::alloc::string::String>,
+    /// Specifies the value of a literal.
+    ///
+    /// This is only applicable to literal ops.
+    #[prost(string, optional, tag="4")]
+    pub value: ::core::option::Option<::prost::alloc::string::String>,
+    /// Specifies the value type of a column or a literal op.
+    ///
+    /// This is only applicate to column and literal ops.
+    #[prost(string, optional, tag="5")]
+    pub value_type: ::core::option::Option<::prost::alloc::string::String>,
+}
 /// Request to list shares.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

delta-sharing/client/src/gen/delta_sharing.sharing.v1.serde.rs

@@ -1212,6 +1212,166 @@ impl<'de> serde::Deserialize<'de> for GetTableVersionResponse {
         deserializer.deserialize_struct("delta_sharing.sharing.v1.GetTableVersionResponse", FIELDS, GeneratedVisitor)
     }
 }
+impl serde::Serialize for JsonPredicate {
+    #[allow(deprecated)]
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        use serde::ser::SerializeStruct;
+        let mut len = 0;
+        if !self.op.is_empty() {
+            len += 1;
+        }
+        if !self.children.is_empty() {
+            len += 1;
+        }
+        if self.name.is_some() {
+            len += 1;
+        }
+        if self.value.is_some() {
+            len += 1;
+        }
+        if self.value_type.is_some() {
+            len += 1;
+        }
+        let mut struct_ser = serializer.serialize_struct("delta_sharing.sharing.v1.JsonPredicate", len)?;
+        if !self.op.is_empty() {
+            struct_ser.serialize_field("op", &self.op)?;
+        }
+        if !self.children.is_empty() {
+            struct_ser.serialize_field("children", &self.children)?;
+        }
+        if let Some(v) = self.name.as_ref() {
+            struct_ser.serialize_field("name", v)?;
+        }
+        if let Some(v) = self.value.as_ref() {
+            struct_ser.serialize_field("value", v)?;
+        }
+        if let Some(v) = self.value_type.as_ref() {
+            struct_ser.serialize_field("valueType", v)?;
+        }
+        struct_ser.end()
+    }
+}
+impl<'de> serde::Deserialize<'de> for JsonPredicate {
+    #[allow(deprecated)]
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        const FIELDS: &[&str] = &[
+            "op",
+            "children",
+            "name",
+            "value",
+            "value_type",
+            "valueType",
+        ];
+
+        #[allow(clippy::enum_variant_names)]
+        enum GeneratedField {
+            Op,
+            Children,
+            Name,
+            Value,
+            ValueType,
+        }
+        impl<'de> serde::Deserialize<'de> for GeneratedField {
+            fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                struct GeneratedVisitor;
+
+                impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+                    type Value = GeneratedField;
+
+                    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                        write!(formatter, "expected one of: {:?}", &FIELDS)
+                    }
+
+                    #[allow(unused_variables)]
+                    fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+                    where
+                        E: serde::de::Error,
+                    {
+                        match value {
+                            "op" => Ok(GeneratedField::Op),
+                            "children" => Ok(GeneratedField::Children),
+                            "name" => Ok(GeneratedField::Name),
+                            "value" => Ok(GeneratedField::Value),
+                            "valueType" | "value_type" => Ok(GeneratedField::ValueType),
+                            _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+                        }
+                    }
+                }
+                deserializer.deserialize_identifier(GeneratedVisitor)
+            }
+        }
+        struct GeneratedVisitor;
+        impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+            type Value = JsonPredicate;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("struct delta_sharing.sharing.v1.JsonPredicate")
+            }
+
+            fn visit_map<V>(self, mut map_: V) -> std::result::Result<JsonPredicate, V::Error>
+                where
+                    V: serde::de::MapAccess<'de>,
+            {
+                let mut op__ = None;
+                let mut children__ = None;
+                let mut name__ = None;
+                let mut value__ = None;
+                let mut value_type__ = None;
+                while let Some(k) = map_.next_key()? {
+                    match k {
+                        GeneratedField::Op => {
+                            if op__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("op"));
+                            }
+                            op__ = Some(map_.next_value()?);
+                        }
+                        GeneratedField::Children => {
+                            if children__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("children"));
+                            }
+                            children__ = Some(map_.next_value()?);
+                        }
+                        GeneratedField::Name => {
+                            if name__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("name"));
+                            }
+                            name__ = map_.next_value()?;
+                        }
+                        GeneratedField::Value => {
+                            if value__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("value"));
+                            }
+                            value__ = map_.next_value()?;
+                        }
+                        GeneratedField::ValueType => {
+                            if value_type__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("valueType"));
+                            }
+                            value_type__ = map_.next_value()?;
+                        }
+                    }
+                }
+                Ok(JsonPredicate {
+                    op: op__.unwrap_or_default(),
+                    children: children__.unwrap_or_default(),
+                    name: name__,
+                    value: value__,
+                    value_type: value_type__,
+                })
+            }
+        }
+        deserializer.deserialize_struct("delta_sharing.sharing.v1.JsonPredicate", FIELDS, GeneratedVisitor)
+    }
+}
 impl serde::Serialize for ListSchemaTablesRequest {
     #[allow(deprecated)]
     fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>

delta-sharing/common/Cargo.toml

@@ -46,7 +46,8 @@ sqlx = { version = "0.8", optional = true, features = [
 ] }
 
 # integration test dependencies (in alphabetical order)
-http-body-util = { version = "0.1.0", optional = true}
+http-body-util = { version = "0.1.0", optional = true }
+mime = { version = "0.3", optional = true }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full"] }
@@ -69,4 +70,4 @@ grpc = ["dep:tonic"]
 sqlx = ["dep:sqlx"]
 
 # exposes functions that help testing cutom handler / router implementations.
-integration = ["http-body-util"]
+integration = ["http-body-util", "mime"]

delta-sharing/common/src/error.rs

@@ -32,6 +32,9 @@ pub enum Error {
     #[error("Invalid identifier: {0}")]
     InvalidIdentifier(#[from] uuid::Error),
 
+    #[error("Invalid predicate: {0}")]
+    InvalidPredicate(String),
+
     #[error("Generic error: {0}")]
     Generic(String),
 
@@ -58,6 +61,10 @@ impl Error {
     pub fn invalid_argument(msg: impl Into<String>) -> Self {
         Self::InvalidArgument(msg.into())
     }
+
+    pub fn invalid_predicate(msg: impl Into<String>) -> Self {
+        Self::InvalidPredicate(msg.into())
+    }
 }
 
 #[cfg(feature = "grpc")]
@@ -79,6 +86,7 @@ impl From<Error> for Status {
             Error::MissingRecipient => {
                 Status::invalid_argument("Failed to extract recipient from request")
             }
+            Error::InvalidPredicate(msg) => Status::invalid_argument(msg),
             Error::AlreadyExists => Status::already_exists("The resource already exists."),
             Error::InvalidIdentifier(_) => Status::internal("Invalid uuid identifier"),
             Error::InvalidArgument(message) => Status::invalid_argument(message),
@@ -141,6 +149,13 @@ mod server {
                     error!("Invalid argument: {}", message);
                     INVALID_ARGUMENT
                 }
+                Error::InvalidPredicate(msg) => {
+                    error!("Invalid predicate: {}", msg);
+                    (
+                        StatusCode::BAD_REQUEST,
+                        "Invalid predicate provided in the request.",
+                    )
+                }
                 Error::InvalidIdentifier(_) => {
                     error!("Invalid uuid identifier");
                     INTERNAL_ERROR

delta-sharing/common/src/gen/delta_sharing.sharing.v1.rs

@@ -302,6 +302,31 @@ pub struct MetadatDelta {
     #[prost(int64, optional, tag="4")]
     pub num_files: ::core::option::Option<i64>,
 }
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct JsonPredicate {
+    /// The operation to perform.
+    #[prost(string, tag="1")]
+    pub op: ::prost::alloc::string::String,
+    /// The inputs to the operation - if any.
+    #[prost(message, repeated, tag="2")]
+    pub children: ::prost::alloc::vec::Vec<JsonPredicate>,
+    /// Specifies the name of a column.
+    ///
+    /// This is only applicable to column ops.
+    #[prost(string, optional, tag="3")]
+    pub name: ::core::option::Option<::prost::alloc::string::String>,
+    /// Specifies the value of a literal.
+    ///
+    /// This is only applicable to literal ops.
+    #[prost(string, optional, tag="4")]
+    pub value: ::core::option::Option<::prost::alloc::string::String>,
+    /// Specifies the value type of a column or a literal op.
+    ///
+    /// This is only applicate to column and literal ops.
+    #[prost(string, optional, tag="5")]
+    pub value_type: ::core::option::Option<::prost::alloc::string::String>,
+}
 /// Request to list shares.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]