fix(core): disable pushdown filter to keep the casting for timestamp …

…column (#907)

wren-core/core/src/mdl/context.rs

@@ -36,7 +36,6 @@ use datafusion::optimizer::eliminate_outer_join::EliminateOuterJoin;
 use datafusion::optimizer::extract_equijoin_predicate::ExtractEquijoinPredicate;
 use datafusion::optimizer::filter_null_join_keys::FilterNullJoinKeys;
 use datafusion::optimizer::propagate_empty_relation::PropagateEmptyRelation;
-use datafusion::optimizer::push_down_filter::PushDownFilter;
 use datafusion::optimizer::replace_distinct_aggregate::ReplaceDistinctWithAggregate;
 use datafusion::optimizer::scalar_subquery_to_join::ScalarSubqueryToJoin;
 use datafusion::optimizer::single_distinct_to_groupby::SingleDistinctToGroupBy;
@@ -176,7 +175,8 @@ fn optimize_rule_for_unparsing() -> Vec<Arc<dyn OptimizerRule + Send + Sync>> {
         // Filters can't be pushed down past Limits, we should do PushDownFilter after PushDownLimit
         // TODO: Sort with pushdown-limit doesn't support to be unparse
         // Arc::new(PushDownLimit::new()),
-        Arc::new(PushDownFilter::new()),
+        // Disable PushDownFilter to avoid the casting for bigquery (datetime/timestamp) column be removed
+        // Arc::new(PushDownFilter::new()),
         Arc::new(SingleDistinctToGroupBy::new()),
         // Disable SimplifyExpressions to avoid apply some function locally
         // Arc::new(SimplifyExpressions::new()),

wren-core/core/src/mdl/mod.rs

@@ -948,6 +948,45 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_disable_pushdown_filter() -> Result<()> {
+        let ctx = SessionContext::new();
+        ctx.register_batch("artist", artist())?;
+        let manifest = ManifestBuilder::new()
+            .catalog("wren")
+            .schema("test")
+            .model(
+                ModelBuilder::new("artist")
+                    .table_reference("artist")
+                    .column(
+                        ColumnBuilder::new("出道時間", "timestamp")
+                            .hidden(true)
+                            .build(),
+                    )
+                    .column(
+                        ColumnBuilder::new("cast_timestamp", "timestamp")
+                            .expression(r#"cast("出道時間" as timestamp with time zone)"#)
+                            .build(),
+                    )
+                    .build(),
+            )
+            .build();
+
+        let analyzed_mdl = Arc::new(AnalyzedWrenMDL::analyze(manifest)?);
+        let sql = r#"select count(*) from wren.test.artist where cast(cast_timestamp as timestamp) > timestamp '2011-01-01 21:00:00'"#;
+        let actual = transform_sql_with_ctx(
+            &SessionContext::new(),
+            Arc::clone(&analyzed_mdl),
+            &[],
+            sql,
+        )
+        .await?;
+        assert_eq!(actual,
+                   "SELECT count(*) FROM (SELECT artist.cast_timestamp FROM (SELECT CAST(artist.\"出道時間\" AS TIMESTAMP WITH TIME ZONE) AS cast_timestamp \
+                   FROM artist) AS artist) AS artist WHERE artist.cast_timestamp > CAST('2011-01-01 21:00:00' AS TIMESTAMP)");
+        Ok(())
+    }
+
     /// Return a RecordBatch with made up data about customer
     fn customer() -> RecordBatch {
         let custkey: ArrayRef = Arc::new(Int64Array::from(vec![1, 2, 3]));