Check create table input logical plan effectively empty (#489)

Author: osipovartem

crates/runtime/src/execution/query.rs

@@ -60,7 +60,7 @@ use super::catalogs::{catalog::DFCatalog, metastore::DFMetastore};
 use super::datafusion::context_provider::ExtendedSqlToRel;
 use super::error::{self as ex_error, ExecutionError, ExecutionResult};
 use super::session::UserSession;
-use super::utils::NormalizedIdent;
+use super::utils::{is_logical_plan_effectively_empty, NormalizedIdent};
 use crate::execution::datafusion::visitors::{functions_rewriter, json_element};
 use tracing_attributes::instrument;
 
@@ -463,7 +463,7 @@ impl UserQuery {
             ..
         })) = plan
         {
-            if matches!(*input, LogicalPlan::EmptyRelation(_)) {
+            if is_logical_plan_effectively_empty(&input) {
                 return created_entity_response();
             }
             let insert_plan = LogicalPlan::Dml(DmlStatement::new(
@@ -814,7 +814,7 @@ impl UserQuery {
             }
         }
         let select_query =
-            format!("SELECT {values} FROM {source_query} JOIN {target_table} {target_alias} ON {on}{where_clause_str}");
+            format!("SELECT {values} FROM {source_query} LEFT JOIN {target_table} {target_alias} ON {on}{where_clause_str}");
 
         // Construct the INSERT statement
         let insert_query = format!("INSERT INTO {target_table} ({columns}) {select_query}");

crates/runtime/src/execution/utils.rs

@@ -28,6 +28,8 @@ use chrono::DateTime;
 use datafusion::arrow::array::ArrayRef;
 use datafusion::arrow::datatypes::DataType;
 use datafusion::common::Result as DataFusionResult;
+use datafusion_common::ScalarValue;
+use datafusion_expr::{Expr, LogicalPlan};
 use embucket_metastore::SchemaIdent as MetastoreSchemaIdent;
 use embucket_metastore::TableIdent as MetastoreTableIdent;
 use sqlparser::ast::{Ident, ObjectName};
@@ -55,6 +57,23 @@ pub enum DataSerializationFormat {
     Json,
 }
 
+#[must_use]
+pub fn is_logical_plan_effectively_empty(plan: &LogicalPlan) -> bool {
+    match plan {
+        LogicalPlan::EmptyRelation(e) => !e.produce_one_row,
+        LogicalPlan::Projection(proj) => is_logical_plan_effectively_empty(&proj.input),
+        LogicalPlan::SubqueryAlias(alias) => is_logical_plan_effectively_empty(&alias.input),
+        LogicalPlan::Filter(filter) => {
+            let is_false_predicate = matches!(
+                filter.predicate,
+                Expr::Literal(ScalarValue::Boolean(Some(false)))
+            );
+            is_false_predicate || is_logical_plan_effectively_empty(&filter.input)
+        }
+        _ => false,
+    }
+}
+
 /*#[async_trait::async_trait]
 pub trait S3ClientValidation: Send + Sync {
     async fn get_aws_bucket_acl(