Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions datafusion/core/src/execution/context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -534,6 +534,13 @@ impl SessionContext {
let physical = DataFrame::new(self.state(), input);

let batches: Vec<_> = physical.collect_partitioned().await?;
let schema = {
Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This solves the bug but feels like the input.schema() should already be correct. We shouldn't need to reach into the record batch to find out the true schema

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I haven't investigated enough to know what is the "correct" fix.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree input.schema() should be fixed (in the sense that in the RecordBatches produced don't match the schema declared) that is the root cause of the issue

Perhaps this is related to #7636 that you filed

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See also #7636 (comment)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was my initial and quick attempt at fixing #7636 before I dove deeper toward a more correct solution. I'm going to close this out since it should be replaced by #7638

if let Some(batch) = batches.first().and_then(|b| b.first()) {
batch.schema().clone()
} else {
schema
}
};
let table = Arc::new(
// pass constraints to the mem table.
MemTable::try_new(schema, batches)?.with_constraints(constraints),
Expand Down