You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "ableegoldman (via GitHub)" <gi...@apache.org> on 2025/07/21 06:08:18 UTC
[I] Missing data when inserting into MemTable [datafusion]
ableegoldman opened a new issue, #16836:
URL: https://github.com/apache/datafusion/issues/16836
### Describe the bug
Hi, I've recently started using DataFusion and have run into an issue trying to copy some results into a local cache implemented using the Memtable. Here is the code:
```
// execute the query and get the DataFrame:
let df = self
.ctx
.execute_logical_plan(plan.clone())
.await
.map_err(anyhow::Error::msg)?;
// print total number of rows
info!("Number of DB results {}", df.clone().count().await?);
// insert into the local table
let plan = cached_table.provider
.insert_into(
&self.ctx.state(),
df.clone().create_physical_plan().await?,
InsertOp::Append,
)
.await?;
let task_ctx = self.ctx.task_ctx();
let mut stream = plan.execute(0, task_ctx)?;
// print number of rows inserted
while let Some(batch) = stream.try_next().await? {
let rows = batch.column_by_name("count");
info!(
"Inserted {:?} rows into cache table {}",
rows,
id
);
}
```
There are 731 result rows, but every time I run this only 87 rows are inserted into the MemTable/cache. I've confirmed this is the accurate count of rows inserted because some later code scans this cache and indeed finds only 87 rows. The number 87 is consistent for me, with the 731 original rows, but varies slightly depending on how many total rows there are -- for example, my teammate had 751 rows in his backing db, and saw it repeatedly insert only 90 rows instead of 87.
Any idea why only a small subset of these rows are being inserted? There is only 1 partition btw (according to `plan.output_partitioning().partition_count())`)
### To Reproduce
_No response_
### Expected behavior
_No response_
### 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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Missing data when inserting into MemTable [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on issue #16836:
URL: https://github.com/apache/datafusion/issues/16836#issuecomment-3113933948
> We do not have a reproducer yet. However, I would like to share some more data points with you. We suspect the issue is related to partitioning for two reasons:
>
> 1. If we set the target partitions to 1 with `let config = SessionConfig::new().with_target_partitions(1);`, the issue does not happen.
> 2. In the following call to `execute_input_stream()` the partition of the input stream is hard coded to 0. To my beginner's eyes it looks like that only partition 0 of the input stream is read and inserted into the sink table. Could that be?
>
>
>
> [datafusion/datafusion/datasource/src/sink.rs](https://github.com/apache/datafusion/blob/dbc03fa4f6d47c8f3b97f3a3d979945b2b7ccce7/datafusion/datasource/src/sink.rs#L227)
>
>
> Line 227
> in
> [dbc03fa](/apache/datafusion/commit/dbc03fa4f6d47c8f3b97f3a3d979945b2b7ccce7)
>
>
>
>
>
>
>
> let data = execute_input_stream(
>
> A useful info might also be the workaround we found. Instead of using `provider.insert_into()` we use `df.clone().write_table(sink_table_name, ...)`.
>
> I hope this is helpful!
Thank you, i think the partition 0 hard coded is right, DataSinkExec is designed as the final, global “write” step for DML operations (INSERT, COPY, EXPORT, etc.), and it intentionally runs on only one partition.
So it's possible before you sink, you don't merge into one partition data, and it requires: required_input_distribution = SinglePartition. So when you setting to partition 1, it will not happen, it's safe to into one partition.
Or call the higher‑level df.write_table(...), which includes that coalesce for you, so it will automatically into one partition before sink i think.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Missing data when inserting into MemTable [datafusion]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on issue #16836:
URL: https://github.com/apache/datafusion/issues/16836#issuecomment-3113873969
We do not have a reproducer yet. However, I would like to share some more data points with you. We suspect the issue is related to partitioning for two reasons:
1. If we set the target partitions to 1, the issue does not happen.
2. In the following call to `execute_input_stream()` the partition of the input stream is hard coded to 0. To my beginner's eyes it looks like that only partition 0 of the input stream is read and inserted into the sink table. Could that be?
https://github.com/apache/datafusion/blob/dbc03fa4f6d47c8f3b97f3a3d979945b2b7ccce7/datafusion/datasource/src/sink.rs#L227
A useful info might also be the workaround we found. Instead of using `provider.insert_into()` we use `df.clone().write_table(sink_table_name, ...)`.
I hope this is helpful!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Missing data when inserting into MemTable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #16836:
URL: https://github.com/apache/datafusion/issues/16836#issuecomment-3113408301
I agree this sounds like a bug. I am not sure what is going on.
I suggest a self contained reproducer is probably the most useful step to get more specific help
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Missing data when inserting into MemTable [datafusion]
Posted by "sap1ens (via GitHub)" <gi...@apache.org>.
sap1ens commented on issue #16836:
URL: https://github.com/apache/datafusion/issues/16836#issuecomment-3483194450
I faced a similar issue, and I had to add a RepartitionExec manually.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Missing data when inserting into MemTable [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on issue #16836:
URL: https://github.com/apache/datafusion/issues/16836#issuecomment-3113815429
I try to reproduce this, but it not reproduced, if we can reproduce it, i can help debugging, thanks!
```rust
// src/main.rs
use std::sync::Arc;
use arrow::array::UInt64Array;
use futures::{StreamExt, TryStreamExt}; // for try_next()
use datafusion::arrow::{
array::{Int64Array, StringArray},
datatypes::{DataType, Field, Schema},
record_batch::RecordBatch,
};
use datafusion::catalog::TableProvider;
use datafusion::datasource::MemTable;
use datafusion::execution::context::{SessionConfig, SessionContext};
use datafusion::logical_expr::dml::InsertOp;
#[tokio::main]
async fn main() -> datafusion::common::Result<()> {
// 1) Create a SessionContext, optionally customizing the number of target partitions
let config = SessionConfig::new();
let ctx = SessionContext::new_with_config(config);
// 2) Build a RecordBatch with two columns (i BIGINT, s VARCHAR), containing 100 rows
let schema = Arc::new(Schema::new(vec![
Field::new("i", DataType::Int64, false),
Field::new("s", DataType::Utf8, false),
]));
let i_array = Int64Array::from_iter_values(0_i64..100);
let s_array = StringArray::from(vec!["hello"; 100]);
let batch = RecordBatch::try_new(schema.clone(), vec![
Arc::new(i_array),
Arc::new(s_array),
])?;
// 3) Register the source MemTable, containing a single batch
let input_table = MemTable::try_new(schema.clone(), vec![vec![batch.clone()]])?;
ctx.register_table("input", Arc::new(input_table))?;
// 4) Register the cache MemTable (initially empty)
let empty_table = MemTable::try_new(schema.clone(), vec![vec![]])?;
let cache_provider = Arc::new(empty_table);
ctx.register_table("cache", cache_provider.clone())?;
// 5) Read the DataFrame from input and print its row count
let df_input = ctx.table("input").await?;
let orig_count = df_input.clone().count().await?;
println!("Row count in source table: {}", orig_count); // Should print 100
// 6) Build a physical INSERT plan using MemTable::insert_into
let plan = cache_provider
.insert_into(
&ctx.state(),
df_input.create_physical_plan().await?,
InsertOp::Append,
)
.await?;
// 7) Execute the insert and accumulate the number of inserted rows
let mut stream = plan.execute(0, ctx.task_ctx())?;
// 8) Read the `count` column from the resulting status batch
let inserted_rows = if let Some(batch) = stream.try_next().await? {
let col = batch
.column_by_name("count")
.expect("Missing `count` column");
// First try unsigned integer array
if let Some(arr) = col.as_any().downcast_ref::<UInt64Array>() {
arr.value(0)
}
// Then try signed integer array for compatibility
else if let Some(arr) = col.as_any().downcast_ref::<Int64Array>() {
arr.value(0) as u64
}
else {
panic!(
"`count` column is neither UInt64Array nor Int64Array, actual type: {:?}",
col.data_type()
);
}
} else {
0
};
println!("Number of rows reported inserted: {}", inserted_rows);
// 9) Scan the cache table again and verify its row count
let df_cache = ctx.table("cache").await?;
let cache_count = df_cache.count().await?;
println!("Row count from scanning cache table: {}", cache_count);
Ok(())
}
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org