dml: fixing cherry-pick errors

Author: Rudolf Lorenz

datafusion/core/src/dataframe/mod.rs

@@ -30,9 +30,7 @@ use crate::arrow::util::pretty;
 use crate::datasource::file_format::csv::CsvFormatFactory;
 use crate::datasource::file_format::format_as_file_type;
 use crate::datasource::file_format::json::JsonFormatFactory;
-use crate::datasource::{
-    provider_as_source, DefaultTableSource, MemTable, TableProvider,
-};
+use crate::datasource::{provider_as_source, MemTable, TableProvider};
 use crate::error::Result;
 use crate::execution::context::{SessionState, TaskContext};
 use crate::execution::FunctionRegistry;
@@ -64,7 +62,6 @@ use datafusion_functions_aggregate::expr_fn::{
 
 use async_trait::async_trait;
 use datafusion_catalog::Session;
-use datafusion_sql::TableReference;
 
 /// Contains options that control how data is
 /// written out from a DataFrame

datafusion/core/src/physical_planner.rs

@@ -24,7 +24,7 @@ use std::sync::Arc;
 use crate::datasource::file_format::file_type_to_format;
 use crate::datasource::listing::ListingTableUrl;
 use crate::datasource::physical_plan::FileSinkConfig;
-use crate::datasource::{source_as_provider, DefaultTableSource};
+use crate::datasource::source_as_provider;
 use crate::error::{DataFusionError, Result};
 use crate::execution::context::{ExecutionProps, SessionState};
 use crate::logical_expr::utils::generate_sort_key;

datafusion/expr/src/logical_plan/dml.rs

@@ -86,39 +86,6 @@ pub struct DmlStatement {
     /// The schema of the output relation
     pub output_schema: DFSchemaRef,
 }
-impl Eq for DmlStatement {}
-impl Hash for DmlStatement {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        self.table_name.hash(state);
-        self.target.schema().hash(state);
-        self.op.hash(state);
-        self.input.hash(state);
-        self.output_schema.hash(state);
-    }
-}
-
-impl PartialEq for DmlStatement {
-    fn eq(&self, other: &Self) -> bool {
-        self.table_name == other.table_name
-            && self.target.schema() == other.target.schema()
-            && self.op == other.op
-            && self.input == other.input
-            && self.output_schema == other.output_schema
-    }
-}
-
-impl Debug for DmlStatement {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        f.debug_struct("DmlStatement")
-            .field("table_name", &self.table_name)
-            .field("target", &"...")
-            .field("target_schema", &self.target.schema())
-            .field("op", &self.op)
-            .field("input", &self.input)
-            .field("output_schema", &self.output_schema)
-            .finish()
-    }
-}
 
 impl Debug for DmlStatement {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {

datafusion/proto/src/logical_plan/from_proto.rs

@@ -22,7 +22,7 @@ use datafusion_common::{
     exec_datafusion_err, internal_err, plan_datafusion_err, Result, ScalarValue,
     TableReference, UnnestOptions,
 };
-use datafusion_expr::dml::InsertOp;
+
 use datafusion_expr::expr::{Alias, Placeholder, Sort};
 use datafusion_expr::expr::{Unnest, WildcardOptions};
 use datafusion_expr::{
@@ -230,11 +230,11 @@ impl From<protobuf::dml_node::Type> for WriteOp {
         match t {
             protobuf::dml_node::Type::Update => WriteOp::Update,
             protobuf::dml_node::Type::Delete => WriteOp::Delete,
-            protobuf::dml_node::Type::InsertAppend => WriteOp::Insert(InsertOp::Append),
-            protobuf::dml_node::Type::InsertOverwrite => {
-                WriteOp::Insert(InsertOp::Overwrite)
-            }
-            protobuf::dml_node::Type::InsertReplace => WriteOp::Insert(InsertOp::Replace),
+            // TODO: support for insert append
+            protobuf::dml_node::Type::InsertAppend => WriteOp::InsertInto,
+            protobuf::dml_node::Type::InsertOverwrite => WriteOp::InsertOverwrite,
+            // TODO: support for insert replace
+            protobuf::dml_node::Type::InsertReplace => WriteOp::InsertOverwrite,
             protobuf::dml_node::Type::Ctas => WriteOp::Ctas,
         }
     }

datafusion/proto/src/logical_plan/mod.rs

@@ -21,8 +21,7 @@ use std::sync::Arc;
 
 use crate::protobuf::logical_plan_node::LogicalPlanType::CustomScan;
 use crate::protobuf::{
-    dml_node, ColumnUnnestListItem, ColumnUnnestListRecursion, CteWorkTableScanNode,
-    CustomTableScanNode, DmlNode, SortExprNodeCollection,
+    dml_node, CteWorkTableScanNode, CustomTableScanNode, DmlNode, SortExprNodeCollection,
 };
 use crate::{
     convert_required, into_required,
@@ -66,9 +65,9 @@ use datafusion_expr::{
         EmptyRelation, Extension, Join, JoinConstraint, Limit, Prepare, Projection,
         Repartition, Sort, SubqueryAlias, TableScan, Values, Window,
     },
-    AggregateUDF, ColumnUnnestList, DistinctOn, DmlStatement, DropView, Expr, FetchType,
-    LogicalPlan, LogicalPlanBuilder, RecursiveQuery, ScalarUDF, SkipType, SortExpr,
-    Statement, TableSource, Unnest, WindowUDF,
+    AggregateUDF, DistinctOn, DmlStatement, DropView, Expr, LogicalPlan,
+    LogicalPlanBuilder, RecursiveQuery, ScalarUDF, SortExpr, TableSource, Unnest,
+    WindowUDF,
 };
 
 use self::to_proto::{serialize_expr, serialize_exprs};
@@ -1695,7 +1694,7 @@ impl AsLogicalPlan for LogicalPlanNode {
             )),
             LogicalPlan::Dml(DmlStatement {
                 table_name,
-                target,
+                dst,
                 op,
                 input,
                 ..
@@ -1708,7 +1707,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                         input: Some(Box::new(input)),
                         target: Some(Box::new(from_table_source(
                             table_name.clone(),
-                            Arc::clone(target),
+                            Arc::clone(dst),
                             extension_codec,
                         )?)),
                         table_name: Some(table_name.clone().into()),

datafusion/proto/src/logical_plan/to_proto.rs

@@ -20,7 +20,6 @@
 //! processes.
 
 use datafusion_common::{TableReference, UnnestOptions};
-use datafusion_expr::dml::InsertOp;
 use datafusion_expr::expr::{
     self, Alias, Between, BinaryExpr, Cast, GroupingSet, InList, Like, Placeholder,
     ScalarFunction, Unnest,
@@ -706,11 +705,8 @@ impl From<JoinConstraint> for protobuf::JoinConstraint {
 impl From<&WriteOp> for protobuf::dml_node::Type {
     fn from(t: &WriteOp) -> Self {
         match t {
-            WriteOp::Insert(InsertOp::Append) => protobuf::dml_node::Type::InsertAppend,
-            WriteOp::Insert(InsertOp::Overwrite) => {
-                protobuf::dml_node::Type::InsertOverwrite
-            }
-            WriteOp::Insert(InsertOp::Replace) => protobuf::dml_node::Type::InsertReplace,
+            WriteOp::InsertInto => protobuf::dml_node::Type::InsertAppend,
+            WriteOp::InsertOverwrite => protobuf::dml_node::Type::InsertOverwrite,
             WriteOp::Delete => protobuf::dml_node::Type::Delete,
             WriteOp::Update => protobuf::dml_node::Type::Update,
             WriteOp::Ctas => protobuf::dml_node::Type::Ctas,

datafusion/proto/tests/cases/roundtrip_logical_plan.rs

@@ -341,70 +341,6 @@ async fn roundtrip_logical_plan_aggregation() -> Result<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn roundtrip_logical_plan_sort() -> Result<()> {
-    let ctx = SessionContext::new();
-
-    let schema = Schema::new(vec![
-        Field::new("a", DataType::Int64, true),
-        Field::new("b", DataType::Decimal128(15, 2), true),
-    ]);
-
-    ctx.register_csv(
-        "t1",
-        "tests/testdata/test.csv",
-        CsvReadOptions::default().schema(&schema),
-    )
-    .await?;
-
-    let query = "SELECT a, b FROM t1 ORDER BY b LIMIT 5";
-    let plan = ctx.sql(query).await?.into_optimized_plan()?;
-
-    let bytes = logical_plan_to_bytes(&plan)?;
-    let logical_round_trip = logical_plan_from_bytes(&bytes, &ctx)?;
-    assert_eq!(format!("{plan}"), format!("{logical_round_trip}"));
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn roundtrip_logical_plan_dml() -> Result<()> {
-    let ctx = SessionContext::new();
-    let schema = Schema::new(vec![
-        Field::new("a", DataType::Int64, true),
-        Field::new("b", DataType::Decimal128(15, 2), true),
-    ]);
-
-    ctx.register_csv(
-        "t1",
-        "tests/testdata/test.csv",
-        CsvReadOptions::default().schema(&schema),
-    )
-    .await?;
-    let queries = [
-        "INSERT INTO T1 VALUES (1, null)",
-        "INSERT OVERWRITE T1 VALUES (1, null)",
-        "REPLACE INTO T1 VALUES (1, null)",
-        "INSERT OR REPLACE INTO T1 VALUES (1, null)",
-        "DELETE FROM T1",
-        "UPDATE T1 SET a = 1",
-        "CREATE TABLE T2 AS SELECT * FROM T1",
-    ];
-    for query in queries {
-        let plan = ctx.sql(query).await?.into_optimized_plan()?;
-        let bytes = logical_plan_to_bytes(&plan)?;
-        let logical_round_trip = logical_plan_from_bytes(&bytes, &ctx)?;
-        assert_eq!(
-            format!("{plan}"),
-            format!("{logical_round_trip}"),
-            "failed query roundtrip: {}",
-            query
-        );
-    }
-
-    Ok(())
-}
-
 #[tokio::test]
 async fn roundtrip_logical_plan_copy_to_sql_options() -> Result<()> {
     let ctx = SessionContext::new();