CREATE TABLE DDL does not save correct schema, resulting in mismatched plan vs execution (record batch) schema
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
`ROW_NUMBER()` function places a non-nullable constraint on its generated field and thus the resulting schema should label that column as `nulllable: false`. But instead, the logical plan resulting from a table created using `CREATE TABLE ...` shows a schema with that field as `nullable: true`. This results in a runtime panic with queries that involve joins (although, I'm not quite sure why it doesn't complain on queries that aren't joins).
### Error message produced with minimal repo below
```text
Thread 'main' panicked at 'query failed to execute: External(ArrowError(InvalidArgumentError("batches[0] schema is different with argument schema.\n batches[0] schema: Schema { fields: [Field { name: \"id\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"row_num\", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} },\n argument schema: Schema { fields: [Field { name: \"id\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"row_num\", data_type: UInt64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }\n ")))', src/main.rs:42:10
```
### To Reproduce
### Minimal repro
Run this script which will result in the error
```text
Thread 'main' panicked at 'query failed to execute: External(ArrowError(InvalidArgumentError("batches[0] schema is different with argument schema.\n batches[0] schema: Schema { fields: [Field { name: \"id\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"row_num\", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} },\n argument schema: Schema { fields: [Field { name: \"id\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"row_num\", data_type: UInt64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }\n ")))', src/main.rs:42:10
```
```rust
use std::sync::Arc;
use datafusion::{
arrow::{
array::{Int32Array, StringArray},
datatypes::{DataType, Field, Schema},
record_batch::RecordBatch,
util::pretty::print_batches,
},
datasource::MemTable,
prelude::{SessionConfig, SessionContext},
};
#[tokio::main(flavor = "current_thread")]
async fn main() {
let config = SessionConfig::new()
.with_create_default_catalog_and_schema(true)
.with_information_schema(true);
let ctx = SessionContext::with_config(config);
ctx.register_table("source_table", Arc::new(create_mem_table()))
.unwrap();
let create_table_query = r#"create table customers as SELECT *, ROW_NUMBER() OVER (ORDER BY id) AS row_num FROM source_table"#;
let _ = ctx
.sql(create_table_query)
.await
.unwrap()
.collect()
.await
.unwrap();
let batches = ctx
// performing a (cross) join query because joins seem to complain about execution schema vs plan schema mismatch
.sql("select a.*, b.* from customers a, customers b")
.await
.unwrap()
.collect()
.await
.expect("query failed to execute");
print_batches(&batches).unwrap();
}
// just some random able that we'll append a row_num column to
fn create_mem_table() -> MemTable {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("name", DataType::Utf8, false),
]));
let ids = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
let names = Arc::new(StringArray::from(vec![
"Alice", "Bob", "Charlie", "David", "Eve",
]));
let batch = RecordBatch::try_new(schema.clone(), vec![ids as _, names as _]).unwrap();
MemTable::try_new(schema, vec![vec![batch]]).unwrap()
}
```
### Expected behavior
The schema that is saved when using create table should be correct (i.e., it should capture nullable: false requirements on fields). The logical plan shouldn't conflict with the observed record batches during execution. No panic should occur.
### Additional context
A bit more context:
This is where nullable false is set. It's not being picked up in the create table statement.
https://github.com/apache/arrow-datafusion/blob/78d9613e81557ca5e5db8b75e5c7dec47ccee0a1/datafusion/physical-expr/src/window/row_number.rs#L54
```rust
fn field(&self) -> Result {
let nullable = false;
let data_type = DataType::UInt64;
Ok(Field::new(self.name(), data_type, nullable))
}
```
I haven't investigated how this field property on the `WindowExpr` is actually used (or omitted) when constructing the logical plan.
Contributor guide
Research direction
Reproduce the failure with the minimal Rust program in src/main.rs, then inspect the referenced datafusion/physical-expr/src/window/row_number.rs implementation and trace how CREATE TABLE stores its schema. Done means the saved plan schema matches the record-batch schema for the ROW_NUMBER() field and the cross-join query completes without a panic.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, sql
- Domain
- databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100