Support consuming Substrait with compound signature function names

datafusion/substrait/Cargo.toml

@@ -37,11 +37,12 @@ chrono = { workspace = true }
 datafusion = { workspace = true, default-features = true }
 itertools = { workspace = true }
 object_store = { workspace = true }
+pbjson-types = "0.6"
 prost = "0.12"
-prost-types = "0.12"
-substrait = "0.34.0"
+substrait = { version = "0.34.0", features = ["serde"] }
 
 [dev-dependencies]
+serde_json = "1.0"
 tokio = { workspace = true }
 
 [features]

datafusion/substrait/src/logical_plan/consumer.rs

@@ -122,6 +122,15 @@ fn scalar_function_type_from_str(
     name: &str,
 ) -> Result<ScalarFunctionType> {
     let s = ctx.state();
+    let name = match name.rsplit_once(':') {
+        // Since 0.32.0, Substrait requires the function names to be in a compound format
+        // https://substrait.io/extensions/#function-signature-compound-names
+        // for example, `add:i8_i8`.
+        // On the consumer side, we don't really care about the signature though, just the name.
+        Some((name, _)) => name,
+        None => name,
+    };
+
     if let Some(func) = s.scalar_functions().get(name) {
         return Ok(ScalarFunctionType::Udf(func.to_owned()));
     }
@@ -1414,7 +1423,7 @@ pub(crate) fn from_substrait_literal(lit: &Literal) -> Result<ScalarValue> {
                         return substrait_err!("Interval year month value is empty");
                     };
                     let value_slice: [u8; 4] =
-                        raw_val.value.clone().try_into().map_err(|_| {
+                        (*raw_val.value).try_into().map_err(|_| {
                             substrait_datafusion_err!(
                                 "Failed to parse interval year month value"
                             )
@@ -1426,7 +1435,7 @@ pub(crate) fn from_substrait_literal(lit: &Literal) -> Result<ScalarValue> {
                         return substrait_err!("Interval day time value is empty");
                     };
                     let value_slice: [u8; 8] =
-                        raw_val.value.clone().try_into().map_err(|_| {
+                        (*raw_val.value).try_into().map_err(|_| {
                             substrait_datafusion_err!(
                                 "Failed to parse interval day time value"
                             )
@@ -1438,7 +1447,7 @@ pub(crate) fn from_substrait_literal(lit: &Literal) -> Result<ScalarValue> {
                         return substrait_err!("Interval month day nano value is empty");
                     };
                     let value_slice: [u8; 16] =
-                        raw_val.value.clone().try_into().map_err(|_| {
+                        (*raw_val.value).try_into().map_err(|_| {
                             substrait_datafusion_err!(
                                 "Failed to parse interval month day nano value"
                             )

datafusion/substrait/src/logical_plan/producer.rs

@@ -42,7 +42,7 @@ use datafusion::logical_expr::expr::{
 };
 use datafusion::logical_expr::{expr, Between, JoinConstraint, LogicalPlan, Operator};
 use datafusion::prelude::Expr;
-use prost_types::Any as ProtoAny;
+use pbjson_types::Any as ProtoAny;
 use substrait::proto::exchange_rel::{ExchangeKind, RoundRobin, ScatterFields};
 use substrait::proto::expression::literal::user_defined::Val;
 use substrait::proto::expression::literal::UserDefined;
@@ -487,7 +487,7 @@ pub fn to_substrait_rel(
                 .serialize_logical_plan(extension_plan.node.as_ref())?;
             let detail = ProtoAny {
                 type_url: extension_plan.node.name().to_string(),
-                value: extension_bytes,
+                value: extension_bytes.into(),
             };
             let mut inputs_rel = extension_plan
                 .node
@@ -1802,7 +1802,7 @@ fn to_substrait_literal(value: &ScalarValue) -> Result<Literal> {
                     }],
                     val: Some(Val::Value(ProtoAny {
                         type_url: INTERVAL_YEAR_MONTH_TYPE_URL.to_string(),
-                        value: bytes.to_vec(),
+                        value: bytes.to_vec().into(),
                     })),
                 }),
                 INTERVAL_YEAR_MONTH_TYPE_REF,
@@ -1825,7 +1825,7 @@ fn to_substrait_literal(value: &ScalarValue) -> Result<Literal> {
                     type_parameters: vec![i64_param.clone(), i64_param],
                     val: Some(Val::Value(ProtoAny {
                         type_url: INTERVAL_MONTH_DAY_NANO_TYPE_URL.to_string(),
-                        value: bytes.to_vec(),
+                        value: bytes.to_vec().into(),
                     })),
                 }),
                 INTERVAL_MONTH_DAY_NANO_TYPE_REF,
@@ -1848,7 +1848,7 @@ fn to_substrait_literal(value: &ScalarValue) -> Result<Literal> {
                     }],
                     val: Some(Val::Value(ProtoAny {
                         type_url: INTERVAL_DAY_TIME_TYPE_URL.to_string(),
-                        value: bytes.to_vec(),
+                        value: bytes.to_vec().into(),
                     })),
                 }),
                 INTERVAL_DAY_TIME_TYPE_REF,

datafusion/substrait/tests/cases/logical_plans.rs

@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Tests for reading substrait plans produced by other systems
+
+#[cfg(test)]
+mod tests {
+    use datafusion::common::Result;
+    use datafusion::prelude::{CsvReadOptions, SessionContext};
+    use datafusion_substrait::logical_plan::consumer::from_substrait_plan;
+    use std::fs::File;
+    use std::io::BufReader;
+    use substrait::proto::Plan;
+
+    #[tokio::test]
+    async fn function_compound_signature() -> Result<()> {
+        // DataFusion currently produces Substrait that refers to functions only by their name.
+        // However, the Substrait spec requires that functions be identified by their compound signature.
+        // This test confirms that DataFusion is able to consume plans following the spec, even though
+        // we don't yet produce such plans.
+        // Once we start producing plans with compound signatures, this test can be replaced by the roundtrip tests.
+
+        let ctx = create_context().await?;
+
+        // File generated with substrait-java's Isthmus:
+        // ./isthmus-cli/build/graal/isthmus "select not d from data" -c "create table data (d boolean)"
+        let path = "tests/testdata/select_not_bool.substrait.json";
+        let proto = serde_json::from_reader::<_, Plan>(BufReader::new(
+            File::open(path).expect("file not found"),
+        ))
+        .expect("failed to parse json");
+
+        let plan = from_substrait_plan(&ctx, &proto).await?;
+
+        assert_eq!(
+            format!("{:?}", plan),
+            "Projection: NOT DATA.a\
+            \n  TableScan: DATA projection=[a, b, c, d, e, f]"
+        );
+        Ok(())
+    }
+
+    async fn create_context() -> datafusion::common::Result<SessionContext> {
+        let ctx = SessionContext::new();
+        ctx.register_csv("DATA", "tests/testdata/data.csv", CsvReadOptions::new())
+            .await?;
+        Ok(ctx)
+    }
+}

datafusion/substrait/tests/cases/mod.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+mod logical_plans;
 mod roundtrip_logical_plan;
 mod roundtrip_physical_plan;
 mod serialize;

datafusion/substrait/tests/testdata/select_not_bool.substrait.json

@@ -0,0 +1,98 @@
+{
+  "extensionUris": [
+    {
+      "extensionUriAnchor": 1,
+      "uri": "/functions_boolean.yaml"
+    }
+  ],
+  "extensions": [
+    {
+      "extensionFunction": {
+        "extensionUriReference": 1,
+        "functionAnchor": 0,
+        "name": "not:bool"
+      }
+    }
+  ],
+  "relations": [
+    {
+      "root": {
+        "input": {
+          "project": {
+            "common": {
+              "emit": {
+                "outputMapping": [
+                  1
+                ]
+              }
+            },
+            "input": {
+              "read": {
+                "common": {
+                  "direct": {
+                  }
+                },
+                "baseSchema": {
+                  "names": [
+                    "D"
+                  ],
+                  "struct": {
+                    "types": [
+                      {
+                        "bool": {
+                          "typeVariationReference": 0,
+                          "nullability": "NULLABILITY_NULLABLE"
+                        }
+                      }
+                    ],
+                    "typeVariationReference": 0,
+                    "nullability": "NULLABILITY_REQUIRED"
+                  }
+                },
+                "namedTable": {
+                  "names": [
+                    "DATA"
+                  ]
+                }
+              }
+            },
+            "expressions": [
+              {
+                "scalarFunction": {
+                  "functionReference": 0,
+                  "args": [],
+                  "outputType": {
+                    "bool": {
+                      "typeVariationReference": 0,
+                      "nullability": "NULLABILITY_NULLABLE"
+                    }
+                  },
+                  "arguments": [
+                    {
+                      "value": {
+                        "selection": {
+                          "directReference": {
+                            "structField": {
+                              "field": 0
+                            }
+                          },
+                          "rootReference": {
+                          }
+                        }
+                      }
+                    }
+                  ],
+                  "options": []
+                }
+              }
+            ]
+          }
+        },
+        "names": [
+          "EXPR$0"
+        ]
+      }
+    }
+  ],
+  "expectedTypeUrls": []
+}