Tpt commented on code in PR #24924:
URL: https://github.com/apache/datafusion/pull/24924#discussion_r4065487926
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -1763,6 +1784,154 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn test_copy_respects_soft_max_bytes_per_output_file() -> Result<()>
{
+ struct TestCase {
+ format: &'static str,
+ options: &'static str,
+ // Arrow, Avro, and the parallel Parquet writer buffer about 1 MB
+ // before forwarding bytes to the object writer (and therefore the
+ // demuxer). Their batches must be larger than that buffer for
+ // rotation to be observable.
+ payload_size: usize,
+ }
+
+ let test_cases = [
+ TestCase {
+ format: "csv",
+ options: "",
+ payload_size: 32 * 1024,
+ },
+ TestCase {
+ format: "csv",
+ options: "OPTIONS ('format.compression' 'gzip')",
+ payload_size: 32 * 1024,
+ },
+ TestCase {
+ format: "json",
+ options: "",
+ payload_size: 32 * 1024,
+ },
+ TestCase {
+ format: "json",
+ options: "OPTIONS ('format.compression' 'zstd')",
+ payload_size: 32 * 1024,
+ },
+ TestCase {
+ format: "arrow",
+ options: "",
+ payload_size: 1536 * 1024,
+ },
+ #[cfg(feature = "parquet")]
+ TestCase {
+ format: "parquet",
+ options: "OPTIONS ('format.compression' 'uncompressed',
'format.max_row_group_size' '1', 'format.allow_single_file_parallelism'
'false')",
+ payload_size: 32 * 1024,
+ },
+ #[cfg(feature = "parquet")]
+ TestCase {
+ format: "parquet",
+ options: "OPTIONS ('format.compression' 'zstd(3)',
'format.max_row_group_size' '1')",
+ payload_size: 1536 * 1024,
+ },
+ #[cfg(feature = "avro")]
+ TestCase {
+ format: "avro",
+ options: "",
+ payload_size: 1536 * 1024,
+ },
+ ];
+
+ for test_case in test_cases {
+ let mut config_map = HashMap::new();
+ config_map.insert(
+ "datafusion.execution.minimum_parallel_output_files".into(),
+ "1".into(),
+ );
+ config_map.insert(
+ "datafusion.execution.soft_max_rows_per_output_file".into(),
+ "1000000".into(),
+ );
+ config_map.insert(
+ "datafusion.execution.soft_max_bytes_per_output_file".into(),
+ "1024".into(),
+ );
+ // Use the smallest supported buffer so the writer cannot get far
+ // ahead of the demuxer. There is still an intentional delay, so
the
+ // test only requires that rotation eventually occurs.
+ config_map.insert(
+
"datafusion.execution.max_buffered_batches_per_output_file".into(),
+ "2".into(),
+ );
+ let config = SessionConfig::from_string_hash_map(&config_map)?;
+ let ctx = SessionContext::new_with_config(config);
+
+ // A deterministic high-entropy payload avoids compression reducing
Review Comment:
Good point. Thanks! Done 1554c472dfadd38599d6b4b3703696deb965ad8e
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
