apache / apache/datafusion

CREATE TABLE DDL does not save correct schema, resulting in mismatched plan vs execution (record batch) schema

Open
#7,636 8 comments 0 reactions 0 assignees View on GitHub
bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.