Default physical planner generates empty relation for DROP TABLE, CREATE MEMORY TABLE, etc

[alamb]

Describe the bug
When we ran some unsupported DDL in our system

drop table foo

The output would be an empty table (rather than an unsupported error):

++
++

However, running CREATE EXTERNAL TABLE results in an error as expected.

To Reproduce
Try and run drop table (not through the datafusion CLI which handles these LogicalPlan variants specially)

Expected behavior
The physical planner should generate an error for DDL statements that have a LogicalPlan variant as it does for CREATE EXTERNAL TABLE

https://github.com/apache/arrow-datafusion/blob/b654fdea697b9aec1cb487e292dd288e5c9d09e3/datafusion/core/src/physical_plan/planner.rs#L1024-L1057

Additional context
We saw this in IOx https://github.com/influxdata/influxdb_iox/issues/5802 and worked around it in https://github.com/influxdata/influxdb_iox/pull/5876 by explicitly generating the errors

[alamb]

I think this is a good first issue as the needed pattern already exists and the required changes are small and localized

[retikulum]

Hi. This seems interesting however when I try to produce the difference between drop table foo and CREATE EXTERNAL TABLE, they both output the same output. What did I do wrong? Output is the following:

++
++

My code for CREATE EXTERNAL TABLE:

use datafusion::error::Result;
use datafusion::prelude::*;


#[tokio::main]
async fn main() -> Result<()> {
    // create local execution context
    let ctx = SessionContext::new();

    // execute the query
    let df = ctx
        .sql("CREATE EXTERNAL TABLE foo stored as parquet location \"test.parquet\"")
        .await?;

    // print the results
    df.show().await?;

    Ok(())
}

My code for drop table test:

use datafusion::error::Result;
use datafusion::prelude::*;


#[tokio::main]
async fn main() -> Result<()> {
    // create local execution context
    let ctx = SessionContext::new();

    // register parquet file with the execution context
    ctx.register_parquet(
        "test",
        "test.parquet",
        ParquetReadOptions::default(),
    )
    .await?;

    // execute the query
    let df = ctx
        .sql("drop table test")
        .await?;
    // print the results

    df.show().await?;

    Ok(())
}

[alamb]

I think one difference may be that the SessionContext::sql has special handling for those statements (rather than running them through a logical plan): https://github.com/apache/arrow-datafusion/blob/b004ec7/datafusion/core/src/execution/context.rs#L286-L343

[Kikkon]

🤣 I'm interested in this, you can assign it to me

[Kikkon]

@alamb 🤣 If I understand correctly, just modify the return result of create_initial_plan, like this #3933.