Revert "feat: drop unused rows as early as possible during row fetch"

This reverts commit d11cd2a.

Author: SkyFan2002

src/query/storages/fuse/src/operations/read/parquet_rows_fetcher.rs

@@ -25,7 +25,6 @@ use databend_common_catalog::plan::Projection;
 use databend_common_catalog::table::Table;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
-use databend_common_expression::BlockRowIndex;
 use databend_common_expression::DataBlock;
 use databend_common_expression::TableSchemaRef;
 use databend_common_storage::ColumnNodes;
@@ -79,23 +78,18 @@ impl<const BLOCKING_IO: bool> RowsFetcher for ParquetRowsFetcher<BLOCKING_IO> {
         let num_rows = row_ids.len();
         let mut part_set = HashSet::new();
         let mut row_set = Vec::with_capacity(num_rows);
-        let mut block_row_indices = HashMap::new();
         for row_id in row_ids {
             let (prefix, idx) = split_row_id(*row_id);
             part_set.insert(prefix);
             row_set.push((prefix, idx));
-            block_row_indices
-                .entry(prefix)
-                .or_insert(Vec::new())
-                .push((0u32, idx as u32, 1usize));
         }
 
         // Read blocks in `prefix` order.
         let part_set = part_set.into_iter().sorted().collect::<Vec<_>>();
-        let mut idx_map = part_set
+        let idx_map = part_set
             .iter()
             .enumerate()
-            .map(|(i, p)| (*p, (i, 0)))
+            .map(|(i, p)| (*p, i))
             .collect::<HashMap<_, _>>();
         // parts_per_thread = num_parts / max_threads
         // remain = num_parts % max_threads
@@ -123,15 +117,10 @@ impl<const BLOCKING_IO: bool> RowsFetcher for ParquetRowsFetcher<BLOCKING_IO> {
                 .iter()
                 .map(|idx| self.part_map[idx].clone())
                 .collect::<Vec<_>>();
-            let block_row_indices = part_set[begin..end]
-                .iter()
-                .map(|idx| block_row_indices.remove(idx).unwrap())
-                .collect::<Vec<_>>();
             tasks.push(Self::fetch_blocks(
                 self.reader.clone(),
                 parts,
                 self.settings,
-                block_row_indices,
             ));
             begin = end;
         }
@@ -152,11 +141,9 @@ impl<const BLOCKING_IO: bool> RowsFetcher for ParquetRowsFetcher<BLOCKING_IO> {
         // Take result rows from blocks.
         let indices = row_set
             .iter()
-            .map(|(prefix, _)| {
-                let (block_idx, row_idx_in_block) = idx_map.get_mut(prefix).unwrap();
-                let row_idx = *row_idx_in_block;
-                *row_idx_in_block += 1;
-                (*block_idx as u32, row_idx as u32, 1_usize)
+            .map(|(prefix, row_idx)| {
+                let block_idx = idx_map[prefix];
+                (block_idx as u32, *row_idx as u32, 1_usize)
             })
             .collect::<Vec<_>>();
 
@@ -256,35 +243,34 @@ impl<const BLOCKING_IO: bool> ParquetRowsFetcher<BLOCKING_IO> {
         reader: Arc<BlockReader>,
         parts: Vec<PartInfoPtr>,
         settings: ReadSettings,
-        block_row_indices: Vec<Vec<BlockRowIndex>>,
     ) -> Result<Vec<DataBlock>> {
-        let mut blocks = Vec::with_capacity(parts.len());
+        let mut chunks = Vec::with_capacity(parts.len());
         if BLOCKING_IO {
-            for (part, block_row_indices) in parts.iter().zip(block_row_indices.iter()) {
+            for part in parts.iter() {
                 let chunk = reader.sync_read_columns_data_by_merge_io(&settings, part, &None)?;
-                let block = Self::build_block(&reader, part, chunk)?;
-                let block =
-                    DataBlock::take_blocks(&[block], block_row_indices, block_row_indices.len());
-                blocks.push(block);
+                chunks.push(chunk);
             }
         } else {
-            for (part, block_row_indices) in parts.iter().zip(block_row_indices.iter()) {
-                let fuse_part = FuseBlockPartInfo::from_part(part)?;
+            for part in parts.iter() {
+                let part = FuseBlockPartInfo::from_part(part)?;
                 let chunk = reader
                     .read_columns_data_by_merge_io(
                         &settings,
-                        &fuse_part.location,
-                        &fuse_part.columns_meta,
+                        &part.location,
+                        &part.columns_meta,
                         &None,
                     )
                     .await?;
-                let block = Self::build_block(&reader, part, chunk)?;
-                let block =
-                    DataBlock::take_blocks(&[block], block_row_indices, block_row_indices.len());
-                blocks.push(block);
+                chunks.push(chunk);
             }
         }
-        Ok(blocks)
+        let fetched_blocks = chunks
+            .into_iter()
+            .zip(parts.iter())
+            .map(|(chunk, part)| Self::build_block(&reader, part, chunk))
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(fetched_blocks)
     }
 
     fn build_block(