matthewgapp opened a new issue, #7636:
URL: https://github.com/apache/arrow-datafusion/issues/7636
### Describe the bug
Row_number function places a non-nullable constraint on its generated field
and thus the resulting schema should result in a schema that shows 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. (not
quite sure why it doesn't complain on queries that aren't joins).
### Error message produced with minimal repo below
```text
hread '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
```
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<Field> {
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.
### To Reproduce
### Minimal repro
Run this script which will result in the error
```text
hread '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
// 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 logical plan shouldn't conflict with the observed record batches during
execution. No panic should occur.
### Additional context
_No response_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
