Is this a bug in DataSink code, or is it a downstream bug that becomes somewhat hard to notice because there is no debug_assert?

I imagine it's downstream, and of course, the debug assert only catches bad RecordBatchStream impls that used the Adapter.

This specific bug may not even be caught by the debug assert if it doesn't use the Adapter somewhere.

Thank you @gatesn for the report.

Can you provide the full code or sql to reproduce the it? So we can solve it more quickly.

Yes, this code fails the schema assertion:

https://github.com/apache/datafusion/compare/main...gatesn:datafusion:ngates/record-batch-stream-schema?expand=1#diff-9b1672adeba35025e24d21f8d7da2f0e87487231775c5e470edf56886f90c837R811-R833

I wonder if it's worth adding that as a debug assertion?

I agree this would be good to add -- most similar check in DataFusion do a check and raise an DataFusionError;:Internal if hit

The only thing I would be worried about is the potential overhead. I get the feeling that this is something that should be done "once" but I'm not sure how we can do it. Maybe utilize the invariant and sanity checkers to check the contract somehow. Let's think about this.

The second (incorrect) invocation comes from session.sql("INSERT INTO my_tbl VALUES ('hello', 42::INT);")

          LogicalPlan::Dml(DmlStatement {
                table_name,
                op: WriteOp::Insert(insert_op),
                ..
            }) => {
                let name = table_name.table();
                let schema = session_state.schema_for_ref(table_name.clone())?;
                if let Some(provider) = schema.table(name).await? {
                    let input_exec = children.one()?;
                    provider
                        .insert_into(session_state, input_exec, *insert_op)
                        .await?
                } else {
                    return exec_err!("Table '{table_name}' does not exist");
                }
            }

I think we may can add check during the physical plan generation, we have the table schema, also we have the input_exec to insert into, so we can check here?

Is this overhead a problem if it's only a debug_assertion? It will be compiled out in release builds. Or are you looking for a release assertion to put somewhere?

Poked around and I think the specific issue here is due to the fact that the schema assigned to LogicalPlan::Values during planning defaults all its fields to nullable.

I've opened a PR to capture nullability from the table schema for the specified columns.

I am concerned whether union will also lead to similar behavior when the nullability of the two inputs is different.

@jonahgao Thank you for calling this out. I think you're right!

In fact, I think we could say more generally, this issue arises when the schema of the source of an INSERT statement contain fields that differ from the table schema in terms of nullability.

I think a better approach may be to map the schema of the source plan to a new schema that demonstrates parity in field nullability with the table schema. We could do this directly when planning the insert, after constructing the source LogicalPlan. In insert_to_plan.

I opened a new issue for this here - #14550

Would love a sanity check from you to verify the above makes sense!