alamb commented on code in PR #25041: URL: https://github.com/apache/datafusion/pull/25041#discussion_r4050693688
########## datafusion/datasource-parquet/src/sink/tests.rs: ########## @@ -0,0 +1,467 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::*; +use arrow::array::{ + ArrayRef, BooleanArray, Int64Array, Int64Builder, ListBuilder, StringArray, +}; +use arrow::compute::concat_batches; +use arrow::datatypes::{DataType, Field}; +use bytes::Bytes; +use datafusion_execution::memory_pool::UnboundedMemoryPool; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::basic::Compression; +use std::io::Write; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::task::{Context, Poll}; +use std::time::Duration; + +struct TestOutput(SharedBuffer); Review Comment: These tests seem very verbose to me and they aren't testing the public API surface (aka these are all basically unit tests) Like @kosiew suggests, I think we should have an "end to end" type test -- namely create a SQL (or DataFrame test) that sets the row group limit and then writes into multiple parquet files and verifies the written row group limit is honored. I don't think this kind of low level / orchestration test is super useful ########## datafusion/common/src/config.rs: ########## @@ -1451,14 +1451,15 @@ config_namespace! { /// either limit is reached, whichever comes first. pub max_row_group_size: usize, default = 1024 * 1024 - /// (writing) Target maximum size of each row group in bytes. When set, - /// the writer flushes whenever either this limit or `max_row_group_size` - /// is reached, whichever comes first. Useful for bounding writer memory - /// on wide schemas where a row-count limit can map to very different - /// byte sizes. Matches the behavior of `parquet.block.size` in - /// parquet-mr. If `None` (the default), only the row-count limit - /// applies. Currently only honored when `allow_single_file_parallelism` - /// is `false`; by default the parallel file writer ignores this limit. + /// (writing) Target maximum estimated encoded size of each row group in bytes. + /// When set, either this target or `max_row_group_size` triggers a flush. The + /// first batch, subject to the row limit, is written before its size can be + /// estimated. Subsequent batches are split using the observed average row size, + /// so this is not a hard byte or memory limit. The parallel writer synchronizes Review Comment: I found this pretty wordy and hard to read. I think the old text is clearer as it focuses on the user visible behavior (the memory usage for wide schemas) while this description describes a lot of internal implementation detail I don't think is particularly helpful to users For example " Subsequent batches are split using the observed average row size, so this is not a hard byte or memory limit." is a lot of words, and I think we could simplify it by just adding "best effort" to the summary if you wanted to emphasize this point > "/// (writing) Target maximum estimated encoded size of each row group in bytes (best effort)" ########## datafusion/datasource-parquet/src/sink.rs: ########## @@ -514,57 +516,48 @@ impl ParquetSink { } } -/// Consumes a stream of [ArrowLeafColumn] via a channel and serializes them using an [ArrowColumnWriter] -/// Once the channel is exhausted, returns the ArrowColumnWriter. +/// A leaf array and its number of root records. Nested leaf arrays can have a +/// different length, so the dispatcher supplies the record count explicitly. +struct ColumnInput { Review Comment: Why do we need this? Isn't the row count the same across all leaves? ########## datafusion/datasource-parquet/src/sink/tests.rs: ########## @@ -0,0 +1,467 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::*; +use arrow::array::{ + ArrayRef, BooleanArray, Int64Array, Int64Builder, ListBuilder, StringArray, +}; +use arrow::compute::concat_batches; +use arrow::datatypes::{DataType, Field}; +use bytes::Bytes; +use datafusion_execution::memory_pool::UnboundedMemoryPool; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::basic::Compression; +use std::io::Write; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::task::{Context, Poll}; +use std::time::Duration; + +struct TestOutput(SharedBuffer); Review Comment: Or put another way, what additional coverage is this adding that is not already covered by the slt test? ########## datafusion/datasource-parquet/src/sink.rs: ########## @@ -66,6 +66,7 @@ use parquet::file::properties::{ use parquet::file::writer::SerializedFileWriter; use tokio::io::{AsyncWrite, AsyncWriteExt}; use tokio::sync::mpsc::{self, Receiver, Sender}; +use tokio::sync::watch; Review Comment: I think this file has a bunch more work than just fixing row group counts -- it contains some sort of structural refactor as well as a fix for an error case (maybe?) Could you please break this into parts: 1. The fix for the closed-channel error 2. The row count propagation to the parallel writer 3. the refactoring here -- 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]
