ARROW-10173: [Rust][DataFusion] Implement support for direct comparison to scalar values

rust/arrow/src/compute/kernels/comparison.rs

@@ -155,6 +155,34 @@ pub fn like_utf8(left: &StringArray, right: &StringArray) -> Result<BooleanArray
     Ok(PrimitiveArray::<BooleanType>::from(Arc::new(data)))
 }
 
+pub fn like_utf8_scalar(left: &StringArray, right: &str) -> Result<BooleanArray> {
+    let null_bit_buffer = left.data().null_buffer().cloned();
+    let re_pattern = right.replace("%", ".*").replace("_", ".");
+    let re = Regex::new(&re_pattern).map_err(|e| {
+        ArrowError::ComputeError(format!(
+            "Unable to build regex from LIKE pattern: {}",
+            e
+        ))
+    })?;
+
+    let mut result = BooleanBufferBuilder::new(left.len());
+    for i in 0..left.len() {
+        let haystack = left.value(i);
+        result.append(re.is_match(haystack))?;
+    }
+
+    let data = ArrayData::new(
+        DataType::Boolean,
+        left.len(),
+        None,
+        null_bit_buffer,
+        0,
+        vec![result.finish()],
+        vec![],
+    );
+    Ok(PrimitiveArray::<BooleanType>::from(Arc::new(data)))
+}
+
 pub fn nlike_utf8(left: &StringArray, right: &StringArray) -> Result<BooleanArray> {
     let mut map = HashMap::new();
     if left.len() != right.len() {
@@ -200,6 +228,34 @@ pub fn nlike_utf8(left: &StringArray, right: &StringArray) -> Result<BooleanArra
     Ok(PrimitiveArray::<BooleanType>::from(Arc::new(data)))
 }
 
+pub fn nlike_utf8_scalar(left: &StringArray, right: &str) -> Result<BooleanArray> {
+    let null_bit_buffer = left.data().null_buffer().cloned();
+    let re_pattern = right.replace("%", ".*").replace("_", ".");
+    let re = Regex::new(&re_pattern).map_err(|e| {
+        ArrowError::ComputeError(format!(
+            "Unable to build regex from LIKE pattern: {}",
+            e
+        ))
+    })?;
+
+    let mut result = BooleanBufferBuilder::new(left.len());
+    for i in 0..left.len() {
+        let haystack = left.value(i);
+        result.append(!re.is_match(haystack))?;
+    }
+
+    let data = ArrayData::new(
+        DataType::Boolean,
+        left.len(),
+        None,
+        null_bit_buffer,
+        0,
+        vec![result.finish()],
+        vec![],
+    );
+    Ok(PrimitiveArray::<BooleanType>::from(Arc::new(data)))
+}
+
 pub fn eq_utf8(left: &StringArray, right: &StringArray) -> Result<BooleanArray> {
     compare_op!(left, right, |a, b| a == b)
 }
@@ -1081,13 +1137,27 @@ mod tests {
         like_utf8,
         vec![true, true, true, false]
     );
+    test_utf8_scalar!(
+        test_utf8_array_like_scalar,
+        vec!["arrow", "parquet", "datafusion", "flight"],
+        "%ar%",
+        like_utf8_scalar,
+        vec![true, true, false, false]
+    );
     test_utf8!(
         test_utf8_array_nlike,
         vec!["arrow", "arrow", "arrow", "arrow"],
         vec!["arrow", "ar%", "%ro%", "foo"],
         nlike_utf8,
         vec![false, false, false, true]
     );
+    test_utf8_scalar!(
+        test_utf8_array_nlike_scalar,
+        vec!["arrow", "parquet", "datafusion", "flight"],
+        "%ar%",
+        nlike_utf8_scalar,
+        vec![false, false, true, true]
+    );
 
     test_utf8!(
         test_utf8_array_eq,

rust/datafusion/Cargo.toml

@@ -78,3 +78,7 @@ harness = false
 [[bench]]
 name = "math_query_sql"
 harness = false
+
+[[bench]]
+name = "filter_query_sql"
+harness = false

rust/datafusion/benches/filter_query_sql.rs

@@ -0,0 +1,91 @@
+// 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.
+
+use arrow::{
+    array::{Float32Array, Float64Array},
+    datatypes::{DataType, Field, Schema},
+    record_batch::RecordBatch,
+};
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::prelude::ExecutionContext;
+use datafusion::{datasource::MemTable, error::Result};
+use futures::executor::block_on;
+use std::sync::Arc;
+
+async fn query(ctx: &mut ExecutionContext, sql: &str) {
+    // execute the query
+    let df = ctx.sql(&sql).unwrap();
+    let results = df.collect().await.unwrap();
+
+    // display the relation
+    for _batch in results {
+        // println!("num_rows: {}", _batch.num_rows());
+    }
+}
+
+fn create_context(array_len: usize, batch_size: usize) -> Result<ExecutionContext> {
+    // define a schema.
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("f32", DataType::Float32, false),
+        Field::new("f64", DataType::Float64, false),
+    ]));
+
+    // define data.
+    let batches = (0..array_len / batch_size)
+        .map(|i| {
+            RecordBatch::try_new(
+                schema.clone(),
+                vec![
+                    Arc::new(Float32Array::from(vec![i as f32; batch_size])),
+                    Arc::new(Float64Array::from(vec![i as f64; batch_size])),
+                ],
+            )
+            .unwrap()
+        })
+        .collect::<Vec<_>>();
+
+    let mut ctx = ExecutionContext::new();
+
+    // declare a table in memory. In spark API, this corresponds to createDataFrame(...).
+    let provider = MemTable::new(schema, vec![batches])?;
+    ctx.register_table("t", Box::new(provider));
+
+    Ok(ctx)
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let array_len = 524_288; // 2^19
+    let batch_size = 4096; // 2^12
+
+    c.bench_function("filter_array", |b| {
+        let mut ctx = create_context(array_len, batch_size).unwrap();
+        b.iter(|| block_on(query(&mut ctx, "select f32, f64 from t where f32 >= f64")))
+    });
+
+    c.bench_function("filter_scalar", |b| {
+        let mut ctx = create_context(array_len, batch_size).unwrap();
+        b.iter(|| {
+            block_on(query(
+                &mut ctx,
+                "select f32, f64 from t where f32 >= 250 and f64 > 250",
+            ))
+        })
+    });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);