Fix select from EmptyExec always return 0 row. (apache#1938)

* fix select from EmptyExec always return 0 row.

* fix clippy

ballista/rust/client/src/context.rs

@@ -542,4 +542,24 @@ mod tests {
         let results = df.collect().await.unwrap();
         pretty::print_batches(&results);
     }
+
+    #[tokio::test]
+    #[cfg(feature = "standalone")]
+    async fn test_empty_exec_with_one_row() {
+        use crate::context::BallistaContext;
+        use ballista_core::config::{
+            BallistaConfigBuilder, BALLISTA_WITH_INFORMATION_SCHEMA,
+        };
+
+        let config = BallistaConfigBuilder::default()
+            .set(BALLISTA_WITH_INFORMATION_SCHEMA, "true")
+            .build()
+            .unwrap();
+        let context = BallistaContext::standalone(&config, 1).await.unwrap();
+
+        let sql = "select EXTRACT(year FROM to_timestamp('2020-09-08T12:13:14+00:00'));";
+
+        let df = context.sql(sql).await.unwrap();
+        assert!(!df.collect().await.unwrap().is_empty());
+    }
 }

datafusion/src/physical_plan/empty.rs

@@ -108,7 +108,10 @@ impl ExecutionPlan for EmptyExec {
         children: Vec<Arc<dyn ExecutionPlan>>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         match children.len() {
-            0 => Ok(Arc::new(EmptyExec::new(false, self.schema.clone()))),
+            0 => Ok(Arc::new(EmptyExec::new(
+                self.produce_one_row,
+                self.schema.clone(),
+            ))),
             _ => Err(DataFusionError::Internal(
                 "EmptyExec wrong number of children".to_string(),
             )),
@@ -179,11 +182,15 @@ mod tests {
     #[test]
     fn with_new_children() -> Result<()> {
         let schema = test_util::aggr_test_schema();
-        let empty = EmptyExec::new(false, schema);
+        let empty = EmptyExec::new(false, schema.clone());
+        let empty_with_row = EmptyExec::new(true, schema);
 
         let empty2 = empty.with_new_children(vec![])?;
         assert_eq!(empty.schema(), empty2.schema());
 
+        let empty_with_row_2 = empty_with_row.with_new_children(vec![])?;
+        assert_eq!(empty_with_row.schema(), empty_with_row_2.schema());
+
         let too_many_kids = vec![empty2];
         assert!(
             empty.with_new_children(too_many_kids).is_err(),