remove independent function

Author: ariesdevil

src/query/datablocks/src/data_block.rs

@@ -292,51 +292,6 @@ impl DataBlock {
         Ok(DataBlock::create(schema.clone(), columns))
     }
 
-    pub fn from_chunk_with_row_limit<A: AsRef<dyn Array>>(
-        schema: &DataSchemaRef,
-        chuck: &Chunk<A>,
-        num_rows: usize,
-    ) -> Result<Vec<DataBlock>> {
-        let columns: Vec<ColumnRef> = chuck
-            .columns()
-            .iter()
-            .zip(schema.fields().iter())
-            .map(|(col, f)| match f.is_nullable() {
-                true => col.into_nullable_column(),
-                false => col.into_column(),
-            })
-            .collect();
-
-        let block = DataBlock::create(schema.clone(), columns);
-
-        let total_rows_in_block: usize = block.num_rows();
-
-        if total_rows_in_block >= num_rows {
-            let num_blocks = if total_rows_in_block % num_rows == 0 {
-                total_rows_in_block / num_rows
-            } else {
-                total_rows_in_block / num_rows + 1
-            };
-            let mut blocks: Vec<DataBlock> = Vec::with_capacity(num_blocks);
-
-            let mut offset = 0;
-            let mut remain_rows = total_rows_in_block;
-            while remain_rows >= num_rows {
-                let cut = block.slice(offset, num_rows);
-                blocks.push(cut);
-                offset += num_rows;
-                remain_rows -= num_rows;
-            }
-            if remain_rows > 0 {
-                blocks.push(block.slice(offset, remain_rows));
-            }
-
-            Ok(blocks)
-        } else {
-            Ok(vec![block])
-        }
-    }
-
     pub fn get_serializers(&self) -> Result<Vec<TypeSerializerImpl>> {
         let columns_size = self.num_columns();
 

src/query/datablocks/tests/it/data_block.rs

@@ -71,39 +71,3 @@ fn test_data_block_convert() -> Result<()> {
 
     Ok(())
 }
-
-#[test]
-fn test_data_block_from_chunk_with_row_limit() -> Result<()> {
-    let schema = DataSchemaRefExt::create(vec![
-        DataField::new("a", i32::to_data_type()),
-        DataField::new("b", i32::to_data_type()),
-    ]);
-
-    let block = DataBlock::create(schema.clone(), vec![
-        Series::from_data(vec![1i32, 2, 3, 4, 5]),
-        Series::from_data(vec![1i32, 2, 3, 4, 5]),
-    ]);
-
-    assert_eq!(&schema, block.schema());
-    assert_eq!(5, block.num_rows());
-    assert_eq!(2, block.num_columns());
-
-    let chunk: Chunk<ArrayRef> = block.try_into().unwrap();
-
-    // first and last test.
-    assert_eq!(5, chunk.len());
-    assert_eq!(2, chunk.columns().len());
-
-    let new_blocks: Vec<DataBlock> =
-        DataBlock::from_chunk_with_row_limit(&schema, &chunk, 2).unwrap();
-
-    assert_eq!(3, new_blocks.len());
-    // first and last block test.
-    assert_eq!(2, new_blocks[0].num_rows());
-    assert_eq!(1, new_blocks[2].num_rows());
-
-    assert_eq!(2, new_blocks[0].num_columns());
-    assert_eq!(2, new_blocks[2].num_columns());
-
-    Ok(())
-}

src/query/pipeline/sources/src/processors/sources/input_formats/impls/input_format_parquet.rs

@@ -333,12 +333,20 @@ impl BlockBuilderTrait for ParquetBlockBuilder {
     fn deserialize(&mut self, mut batch: Option<RowGroupInMemory>) -> Result<Vec<DataBlock>> {
         if let Some(rg) = batch.as_mut() {
             let chunk = rg.get_arrow_chunk()?;
-            // Using `block_compact_thresholds` is enough here.
-            let blocks = DataBlock::from_chunk_with_row_limit(
-                &self.ctx.schema,
-                &chunk,
-                self.ctx.block_compact_thresholds.max_rows_per_block,
-            )?;
+            let block = DataBlock::from_chunk(&self.ctx.schema, &chunk)?;
+
+            let block_total_rows = block.num_rows();
+            let num_rows_per_block = self.ctx.block_compact_thresholds.max_rows_per_block;
+            let blocks: Vec<DataBlock> = (0..block_total_rows)
+                .step_by(num_rows_per_block)
+                .map(|idx| {
+                    if idx + num_rows_per_block < block_total_rows {
+                        block.slice(idx, num_rows_per_block)
+                    } else {
+                        block.slice(idx, block_total_rows - idx)
+                    }
+                })
+                .collect();
 
             Ok(blocks)
         } else {