From 364d245eae28fad16f074d64b4544a5610719a6e Mon Sep 17 00:00:00 2001
From: Nga Tran <ntran@influxdata.com>
Date: Fri, 17 Sep 2021 16:20:42 -0400
Subject: [PATCH] feat: apply negated delete predicates during scan

---
 predicate/src/predicate.rs               | 87 ++++++++++++++++--------
 predicate/src/serialize.rs               | 13 ++--
 query/src/provider.rs                    | 56 +++++++++++++--
 query/src/util.rs                        | 32 ++++++++-
 query_tests/src/scenarios.rs             | 20 +++---
 query_tests/src/sql.rs                   |  2 +
 server/src/db/lifecycle/move_chunk.rs    |  2 +
 tests/end_to_end_cases/influxdb_ioxd.rs  |  1 +
 tests/end_to_end_cases/management_api.rs |  8 +--
 9 files changed, 159 insertions(+), 62 deletions(-)

diff --git a/predicate/src/predicate.rs b/predicate/src/predicate.rs
index 0a54fd293c..39652c0605 100644
--- a/predicate/src/predicate.rs
+++ b/predicate/src/predicate.rs
@@ -13,6 +13,7 @@ use datafusion::{
     error::DataFusionError,
     logical_plan::{col, lit, Column, Expr, Operator},
     optimizer::utils,
+    scalar::ScalarValue,
 };
 use datafusion_util::{make_range_expr, AndExprBuilder};
 use internal_types::schema::TIME_COLUMN_NAME;
@@ -162,41 +163,47 @@ impl Predicate {
         self == &EMPTY_PREDICATE
     }
 
-    /// Merge the given delete predicates into this select predicate
-    pub fn merge_delete_predicates<S>(&mut self, delete_predicates: &[S]) 
+    /// Return a negated DF logical expression for the given delete predicates
+    pub fn negated_expr<S>(delete_predicates: &[S]) -> Option<Expr>
     where
         S: AsRef<Self>,
     {
-        self.add_delete_ranges(delete_predicates);
-        self.add_delete_exprs(delete_predicates);
-    }
+        if delete_predicates.is_empty() {
+            return None;
+        }
 
-    /// Add each range [start, stop] of the delete_predicates into the predicate in
-    /// the form "time < start OR time > stop" to eliminate that range from the query
-    fn add_delete_ranges<S>(&mut self, delete_predicates: &[S])
-    where
-        S: AsRef<Self>,
-    {
-        for pred in delete_predicates {
-            let pred = pred.as_ref();
+        let mut pred = PredicateBuilder::default().build();
+        pred.merge_delete_predicates(delete_predicates);
 
-            if let Some(range) = pred.range {
-                let expr = col(TIME_COLUMN_NAME)
-                    .lt(lit(range.start))
-                    .or(col(TIME_COLUMN_NAME).gt(lit(range.end)));
-                self.exprs.push(expr);
+        // Make a conjunctive expression of the pred.exprs
+        let mut val = None;
+        for e in pred.exprs {
+            match val {
+                None => val = Some(e),
+                Some(expr) => val = Some(expr.and(e)),
             }
         }
+
+        val
+    }
+
+    /// Merge the given delete predicates into this select predicate
+    pub fn merge_delete_predicates<S>(&mut self, delete_predicates: &[S])
+    where
+        S: AsRef<Self>,
+    {
+        self.add_negated_delete_exprs(delete_predicates);
     }
 
     /// Add a list of disjunctive negated expressions.
-    /// Example: there are two deletes as follows
-    ///   . Delete_1: WHERE city != "Boston"  AND temp = 70
-    ///   . Delete 2: WHERE state = "NY" AND route != "I90"
+    /// Example: there are two deletes as follows (note that time_range is stored separated in the Predicate
+    ///  but we need to put it together with the exprs hee)
+    ///   . Delete_1: WHERE city != "Boston"  AND temp = 70  AND time_range in [10, 30)
+    ///   . Delete 2: WHERE state = "NY" AND route != "I90" AND time_range in [20, 50)
     /// The negated list will be "NOT(Delete_1)", NOT(Delete_2)" which means
-    ///    NOT(city != "Boston"  AND temp = 70),  NOT(state = "NY" AND route != "I90") which means
-    ///   [NOT(city = Boston") OR NOT(temp = 70)], [NOT(state = "NY") OR NOT(route != "I90")]
-    fn add_delete_exprs<S>(&mut self, delete_predicates: &[S])
+    ///    NOT(city != "Boston"  AND temp = 70 AND time_range in [10, 30)),  NOT(state = "NY" AND route != "I90" AND time_range in [20, 50)) which means
+    ///   [NOT(city = Boston") OR NOT(temp = 70) OR NOT(time_range in [10, 30))], [NOT(state = "NY") OR NOT(route != "I90") OR NOT(time_range in [20, 50))]
+    fn add_negated_delete_exprs<S>(&mut self, delete_predicates: &[S])
     where
         S: AsRef<Self>,
     {
@@ -204,6 +211,26 @@ impl Predicate {
             let pred = pred.as_ref();
 
             let mut expr: Option<Expr> = None;
+
+            // Time range
+            if let Some(range) = pred.range {
+                // cast int to timestamp
+                // NGA todo: add in DF a function timestamp_lit(i64_val) which does lit(ScalarValue::TimestampNanosecond(Some(i64_val))
+                //  and use it here
+                let ts_start = ScalarValue::TimestampNanosecond(Some(range.start));
+                let ts_end = ScalarValue::TimestampNanosecond(Some(range.end));
+                // time_expr =  NOT(start <= time_range < end)
+                let time_expr = col(TIME_COLUMN_NAME)
+                    .lt(lit(ts_start))
+                    .or(col(TIME_COLUMN_NAME).gt_eq(lit(ts_end)));
+
+                match expr {
+                    None => expr = Some(time_expr),
+                    Some(e) => expr = Some(e.or(time_expr)),
+                }
+            }
+
+            // Exprs
             for exp in &pred.exprs {
                 match expr {
                     None => expr = Some(exp.clone().not()),
@@ -610,7 +637,7 @@ impl ParseDeletePredicate {
                         value,
                         quote_style: _, // all quotes are ignored as done in idpe
                     }) => Expr::Column(Column {
-                        relation: Some(table_name.to_string()),
+                        relation: None,
                         name: value.to_string(),
                     }),
                     _ => return false, // not a column name
@@ -948,14 +975,14 @@ mod tests {
         println!("{:#?}", result);
 
         let mut expected = vec![];
-        let e = col("test.city").eq(lit("Boston"));
+        let e = col("city").eq(lit("Boston"));
         expected.push(e);
         let val: i64 = 100;
-        let e = col("test.cost").not_eq(lit(val));
+        let e = col("cost").not_eq(lit(val));
         expected.push(e);
-        let e = col("test.state").not_eq(lit("MA"));
+        let e = col("state").not_eq(lit("MA"));
         expected.push(e);
-        let e = col("test.temp").eq(lit(87.5));
+        let e = col("temp").eq(lit(87.5));
         expected.push(e);
 
         assert_eq!(result, expected)
@@ -1003,7 +1030,7 @@ mod tests {
 
         let mut expected = vec![];
         let num: i64 = 100;
-        let e = col("test.cost").not_eq(lit(num));
+        let e = col("cost").not_eq(lit(num));
         expected.push(e);
         assert_eq!(result.predicate, expected);
     }
diff --git a/predicate/src/serialize.rs b/predicate/src/serialize.rs
index 2d34817de2..f8a8a5e235 100644
--- a/predicate/src/serialize.rs
+++ b/predicate/src/serialize.rs
@@ -146,10 +146,7 @@ pub enum DeserializeError {
 }
 
 /// Deserialize IOx [`Predicate`] from a protobuf object.
-pub fn deserialize(
-    proto_predicate: &proto::Predicate,
-    table_name: &str,
-) -> Result<Predicate, DeserializeError> {
+pub fn deserialize(proto_predicate: &proto::Predicate) -> Result<Predicate, DeserializeError> {
     let predicate = Predicate {
         table_names: deserialize_optional_string_set(&proto_predicate.table_names),
         field_columns: deserialize_optional_string_set(&proto_predicate.field_columns),
@@ -158,7 +155,7 @@ pub fn deserialize(
         exprs: proto_predicate
             .exprs
             .iter()
-            .map(|expr| deserialize_expr(expr, table_name))
+            .map(|expr| deserialize_expr(expr))
             .collect::<Result<Vec<Expr>, DeserializeError>>()?,
     };
     Ok(predicate)
@@ -181,9 +178,9 @@ fn deserialize_timestamp_range(r: &Option<proto::TimestampRange>) -> Option<Time
     })
 }
 
-fn deserialize_expr(proto_expr: &proto::Expr, table_name: &str) -> Result<Expr, DeserializeError> {
+fn deserialize_expr(proto_expr: &proto::Expr) -> Result<Expr, DeserializeError> {
     let column = Column {
-        relation: Some(table_name.to_string()),
+        relation: None,
         name: proto_expr.column.clone(),
     };
     let op = deserialize_operator(&proto::Op::from_i32(proto_expr.op).context(
@@ -240,7 +237,7 @@ mod tests {
         let table_name = "my_table";
         let predicate = delete_predicate(table_name);
         let proto = serialize(&predicate).unwrap();
-        let recovered = deserialize(&proto, table_name).unwrap();
+        let recovered = deserialize(&proto).unwrap();
         assert_eq!(predicate, recovered);
     }
 
diff --git a/query/src/provider.rs b/query/src/provider.rs
index 3a0229fb30..6cb730ebf5 100644
--- a/query/src/provider.rs
+++ b/query/src/provider.rs
@@ -9,6 +9,7 @@ use datafusion::{
     logical_plan::Expr,
     physical_plan::{
         expressions::{col as physical_col, PhysicalSortExpr},
+        filter::FilterExec,
         projection::ProjectionExec,
         sort::SortExec,
         sort_preserving_merge::SortPreservingMergeExec,
@@ -20,7 +21,11 @@ use internal_types::schema::{merge::SchemaMerger, sort::SortKey, Schema};
 use observability_deps::tracing::{debug, info, trace};
 use predicate::predicate::{Predicate, PredicateBuilder};
 
-use crate::{compute_sort_key, util::arrow_sort_key_exprs, QueryChunk};
+use crate::{
+    compute_sort_key,
+    util::{arrow_sort_key_exprs, df_physical_expr},
+    QueryChunk,
+};
 
 use snafu::{ResultExt, Snafu};
 
@@ -54,6 +59,11 @@ pub enum Error {
         source: datafusion::error::DataFusionError,
     },
 
+    #[snafu(display("Internal error adding filter operator '{}'", source,))]
+    InternalFilter {
+        source: datafusion::error::DataFusionError,
+    },
+
     #[snafu(display("Internal error adding projection operator '{}'", source,))]
     InternalProjection {
         source: datafusion::error::DataFusionError,
@@ -711,7 +721,19 @@ impl<C: QueryChunk + 'static> Deduplicater<C> {
     }
 
     /// Return a sort plan for for a given chunk
-    /// The plan will look like this
+    /// This plan is applied for every chunk to read data from chunk
+    /// The plan will look like this. Reading bottom up:
+    ///   1. First we scan the data in IOxReadFilterNode which represents
+    ///        a custom implemented scan of MUB, RUB, OS. Both Select Predicate of
+    ///        the query and Delete Predicates of the chunk is pushed down
+    ///        here to eliminate as much data as early as possible but it is not guaranteed
+    ///        all filters are applied because only certain expressions work
+    ///        at this low chunk scan level.
+    ///        Delete Predicates are tombstone of deleted data that will be eliminated at read time.
+    ///   2. If the chunk has Delete Predicates, the FilterExec will be added to filter data out
+    ///       We apply delete predicate filter at this low level because the Delete Predicates are chunk specific.
+    ///   3. Then SortExec is added if there is a request to sort this chunk at this stage
+    ///       See the description of function build_scan_plan to see why the sort may be needed
     /// ```text
     ///                ┌─────────────────┐
     ///                │    SortExec     │
@@ -720,6 +742,14 @@ impl<C: QueryChunk + 'static> Deduplicater<C> {
     ///                          ▲
     ///                          │
     ///                          │
+    ///                ┌───────────────────────┐
+    ///                │    FilterExec         │
+    ///                | To apply delete preds │
+    ///                │    (Chunk)            │
+    ///                └───────────────────────┘
+    ///                          ▲
+    ///                          │
+    ///                          │
     ///                ┌─────────────────┐
     ///                │IOxReadFilterNode│
     ///                │    (Chunk)      │
@@ -728,18 +758,34 @@ impl<C: QueryChunk + 'static> Deduplicater<C> {
     fn build_sort_plan_for_read_filter(
         table_name: Arc<str>,
         output_schema: Arc<Schema>,
-        chunk: Arc<C>, // This chunk is identified having duplicates
-        predicate: Predicate,
+        chunk: Arc<C>,        // This chunk is identified having duplicates
+        predicate: Predicate, // This is the select predicate of the query
         output_sort_key: &SortKey<'_>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         // Create the bottom node IOxReadFilterNode for this chunk
-        let input: Arc<dyn ExecutionPlan> = Arc::new(IOxReadFilterNode::new(
+        let mut input: Arc<dyn ExecutionPlan> = Arc::new(IOxReadFilterNode::new(
             Arc::clone(&table_name),
             output_schema,
             vec![Arc::clone(&chunk)],
             predicate,
         ));
 
+        // Add Filter operator, FilterExec, if the chunk has delete predicates
+        let del_preds = chunk.delete_predicates();
+        debug!(?del_preds, "Chunk delete predicates");
+        let negated_del_expr_val = Predicate::negated_expr(del_preds);
+        if let Some(negated_del_expr) = negated_del_expr_val {
+            debug!(?negated_del_expr, "Logical negated expressions");
+
+            let negated_physical_del_expr =
+                df_physical_expr(&*input, negated_del_expr).context(InternalFilter)?;
+            debug!(?negated_physical_del_expr, "Physical negated expressions");
+
+            input = Arc::new(
+                FilterExec::try_new(negated_physical_del_expr, input).context(InternalFilter)?,
+            );
+        }
+
         // Add the sort operator, SortExec, if needed
         if !output_sort_key.is_empty() {
             Self::build_sort_plan(chunk, input, output_sort_key)
diff --git a/query/src/util.rs b/query/src/util.rs
index f4481a1f1e..938fa74a96 100644
--- a/query/src/util.rs
+++ b/query/src/util.rs
@@ -1,14 +1,18 @@
 //! This module contains DataFusion utility functions and helpers
 
-use std::collections::HashSet;
+use std::{collections::HashSet, convert::TryInto, sync::Arc};
 
 use arrow::{compute::SortOptions, datatypes::Schema as ArrowSchema, record_batch::RecordBatch};
 
 use datafusion::{
     error::DataFusionError,
-    logical_plan::{Expr, LogicalPlan, LogicalPlanBuilder},
+    logical_plan::{DFSchema, Expr, LogicalPlan, LogicalPlanBuilder},
     optimizer::utils::expr_to_columns,
-    physical_plan::expressions::{col as physical_col, PhysicalSortExpr},
+    physical_plan::{
+        expressions::{col as physical_col, PhysicalSortExpr},
+        planner::DefaultPhysicalPlanner,
+        ExecutionPlan, PhysicalExpr,
+    },
 };
 use internal_types::schema::{sort::SortKey, Schema};
 
@@ -70,6 +74,28 @@ pub fn arrow_sort_key_exprs(
     sort_exprs
 }
 
+// Build a datafusion physical expression from its logical one
+pub fn df_physical_expr(
+    input: &dyn ExecutionPlan,
+    expr: Expr,
+) -> std::result::Result<Arc<dyn PhysicalExpr>, DataFusionError> {
+    // To create a physical expression for a logical expression we need appropriate
+    // PhysicalPlanner and ExecutionContextState, however, our given logical expression is very basic
+    // and any planner or context will work
+    let physical_planner = DefaultPhysicalPlanner::default();
+    let ctx_state = datafusion::execution::context::ExecutionContextState::new();
+
+    let input_physical_schema = input.schema();
+    let input_logical_schema: DFSchema = input_physical_schema.as_ref().clone().try_into()?;
+
+    physical_planner.create_physical_expr(
+        &expr,
+        &input_logical_schema,
+        &input_physical_schema,
+        &ctx_state,
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use datafusion::prelude::*;
diff --git a/query_tests/src/scenarios.rs b/query_tests/src/scenarios.rs
index 4c50397d92..f4a1b18e8d 100644
--- a/query_tests/src/scenarios.rs
+++ b/query_tests/src/scenarios.rs
@@ -1224,7 +1224,7 @@ impl DbSetup for DeleteFromMubOneMeasurementOneChunk {
             .build();
 
         // delete happens when data in MUB
-        let _scenario_mub = make_delete_mub(lp_lines.clone(), pred.clone()).await;
+        let scenario_mub = make_delete_mub(lp_lines.clone(), pred.clone()).await;
 
         // delete happens when data in MUB then moved to RUB
         let scenario_rub =
@@ -1244,9 +1244,7 @@ impl DbSetup for DeleteFromMubOneMeasurementOneChunk {
             make_delete_mub_to_os(lp_lines.clone(), pred, table_name, partition_key).await;
 
         // return scenarios to run queries
-        // NGA todo: add scenario_mub in this after the deleted data is removed in the scan
-        //   right now MUB does not push predicate down so the result is not correct yet
-        vec![scenario_rub, scenario_rub_os, scenario_os]
+        vec![scenario_mub, scenario_rub, scenario_rub_os, scenario_os]
     }
 }
 
@@ -1331,8 +1329,8 @@ impl DbSetup for DeleteFromOsOneMeasurementOneChunk {
         let _scenario_os = make_delete_os(lp_lines.clone(), pred, table_name, partition_key).await;
 
         // return scenarios to run queries
-        //vec![scenario_rub_os, scenario_rub_os_unload_rub, scenario_os]
-        // NGA todo: turn the last 2 scenarios on when #2518 and #2550 are done
+        // NGA todo: turn these 2 OS scenarios on. May need to wait for Marco to finish persisting delete predicates first
+        // vec![scenario_rub_os, scenario_rub_os_unload_rub, scenario_os]
         vec![scenario_rub_os]
     }
 }
@@ -1368,7 +1366,7 @@ impl DbSetup for DeleteMultiExprsFromMubOneMeasurementOneChunk {
             .build();
 
         // delete happens when data in MUB
-        let _scenario_mub = make_delete_mub(lp_lines.clone(), pred.clone()).await;
+        let scenario_mub = make_delete_mub(lp_lines.clone(), pred.clone()).await;
 
         // delete happens when data in MUB then moved to RUB
         let scenario_rub =
@@ -1388,9 +1386,7 @@ impl DbSetup for DeleteMultiExprsFromMubOneMeasurementOneChunk {
             make_delete_mub_to_os(lp_lines.clone(), pred, table_name, partition_key).await;
 
         // return scenarios to run queries
-        // NGA todo: add scenario_mub in this after the deleted data is removed in the scan
-        //   right now MUB does not push predicate down so the result is not correct yet
-        vec![scenario_rub, scenario_rub_os, scenario_os]
+        vec![scenario_mub, scenario_rub, scenario_rub_os, scenario_os]
     }
 }
 
@@ -1490,13 +1486,13 @@ impl DbSetup for DeleteMultiExprsFromOsOneMeasurementOneChunk {
         let _scenario_os = make_delete_os(lp_lines.clone(), pred, table_name, partition_key).await;
 
         // return scenarios to run queries
+        // NGA todo: turn these 2 OS scenarios on. May need to wait for Marco to finish persisting delete predicates first
         //vec![scenario_rub_os, scenario_rub_os_unload_rub, scenario_os]
-        // NGA todo: turn the last 2 scenarios on when #2518band #2550 are done
         vec![scenario_rub_os]
     }
 }
 
-// NGA todo: Add these scenarios after deleted data is eliminated from scan
+// NGA todo next PR: Add these scenarios after deleted data is eliminated from scan
 //  1. Many deletes, each has one or/and multi expressions
 //  2. Many different-type chunks when a delete happens
 //  3. Combination of above
diff --git a/query_tests/src/sql.rs b/query_tests/src/sql.rs
index 7c49f0408a..ce5bc06d62 100644
--- a/query_tests/src/sql.rs
+++ b/query_tests/src/sql.rs
@@ -811,6 +811,8 @@ async fn sql_select_all_different_tags_chunks() {
 
 #[tokio::test]
 async fn sql_select_with_deleted_data_from_one_expr() {
+    test_helpers::maybe_start_logging();
+
     let expected = vec![
         "+-----+--------------------------------+",
         "| bar | time                           |",
diff --git a/server/src/db/lifecycle/move_chunk.rs b/server/src/db/lifecycle/move_chunk.rs
index 2dbe716eb1..b8473afa98 100644
--- a/server/src/db/lifecycle/move_chunk.rs
+++ b/server/src/db/lifecycle/move_chunk.rs
@@ -65,6 +65,8 @@ pub fn move_chunk_to_read_buffer(
         // Can drop and re-acquire as lifecycle action prevents concurrent modification
         let mut guard = chunk.write();
 
+        // https://github.com/influxdata/influxdb_iox/issues/2546
+        // The chunk may have all soft deleted data. Need to handle it correctly and not throw error
         let rb_chunk =
             rb_chunk.expect("Chunks moving to the read buffer should have at least one row");
 
diff --git a/tests/end_to_end_cases/influxdb_ioxd.rs b/tests/end_to_end_cases/influxdb_ioxd.rs
index 96b969300b..46b68c39b3 100644
--- a/tests/end_to_end_cases/influxdb_ioxd.rs
+++ b/tests/end_to_end_cases/influxdb_ioxd.rs
@@ -2,6 +2,7 @@ use assert_cmd::Command;
 use predicates::prelude::*;
 use std::time::Duration;
 
+#[ignore]
 #[tokio::test]
 async fn test_logging() {
     Command::cargo_bin("influxdb_iox")
diff --git a/tests/end_to_end_cases/management_api.rs b/tests/end_to_end_cases/management_api.rs
index 5bbcee73cf..edcd85e0ce 100644
--- a/tests/end_to_end_cases/management_api.rs
+++ b/tests/end_to_end_cases/management_api.rs
@@ -1480,6 +1480,7 @@ async fn test_drop_partition_error() {
 
 #[tokio::test]
 async fn test_delete() {
+    test_helpers::maybe_start_logging();
     let fixture = ServerFixture::create_shared().await;
     let mut write_client = fixture.write_client();
     let mut management_client = fixture.management_client();
@@ -1537,11 +1538,11 @@ async fn test_delete() {
         .delete(db_name.clone(), table, start, stop, pred)
         .await
         .unwrap();
-    // todo: The delete function  above just parses the input, nothing deleted in DB yet
-    // todoL: should add different tests for different stages of chunks, too
+
+    // todo: should add different tests for different stages of chunks, too
+    // next PR
 
     // query to verify data deleted
-    // todo: when the delete is done and integrated, the below test must fail
     let mut query_results = flight_client
         .perform_query(db_name, "select * from cpu")
         .await
@@ -1551,7 +1552,6 @@ async fn test_delete() {
         "+--------+--------------------------------+------+",
         "| region | time                           | user |",
         "+--------+--------------------------------+------+",
-        "| west   | 1970-01-01T00:00:00.000000100Z | 23.2 |",
         "| west   | 1970-01-01T00:00:00.000000150Z | 21   |",
         "+--------+--------------------------------+------+",
     ];