fix: address review comments for ManifestGroupTranslator backport (#47954) (#47965)

pr: #47954
issue: #47858
Address review feedback on the cherry-pick of #47904 to 2.6:

- Replace WaitAllFutures with channel drain in catch blocks of
ManifestGroupTranslator::get_cells() and
GroupChunkTranslator::get_cells()
  to prevent deadlock when bounded channel is full and producers are
  blocked on push()
- Add AssertInfo to validate all_tables size matches batch.rg_count
  in LoadCellBatchAsync before indexing
- Remove redundant explicit Close() call in LoadWithStrategy since
  folly::makeGuard already handles cleanup on scope exit

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

internal/core/src/segcore/memory_planner.cpp

@@ -232,13 +232,6 @@ LoadWithStrategy(const std::vector<std::string>& remote_files,
                                  static_cast<size_t>(block.offset + i),
                                  table});
                         }
-                        auto close_status = row_group_reader->Close();
-                        AssertInfo(
-                            close_status.ok(),
-                            "[StorageV2] Failed to close row group reader "
-                            "for file " +
-                                file + " with error " +
-                                close_status.ToString());
                         return ret;
                     }));
             }
@@ -359,6 +352,11 @@ LoadCellBatchAsync(milvus::OpContext* op_ctx,
                        "[StorageV2] Failed to read batch: " +
                            tables_result.status().ToString());
             auto all_tables = std::move(tables_result).ValueOrDie();
+            AssertInfo(all_tables.size() == static_cast<size_t>(batch.rg_count),
+                       "reader returns less tables than expected, batch rg "
+                       "count: {}, result size: {}",
+                       batch.rg_count,
+                       all_tables.size());
 
             int64_t table_offset = 0;
             for (const auto& cell : batch.cells) {

internal/core/src/segcore/storagev2translator/GroupChunkTranslator.cpp

@@ -374,12 +374,16 @@ GroupChunkTranslator::get_cells(milvus::OpContext* ctx,
             completed_cells[cell_data->cid] =
                 load_group_chunk(cell_data->tables, cell_data->cid);
         }
-    } catch (std::exception& ex) {
-        // make sure all futures are handled, but still throw pop & load ex
+    } catch (...) {
+        // Drain the channel to unblock producers that may be stuck on push()
+        // to a full bounded channel. Without draining, producers block forever
+        // and their task_guard (which calls channel->close()) never executes.
+        std::shared_ptr<milvus::segcore::CellLoadResult> discard;
         try {
-            storage::WaitAllFutures(load_futures);
+            while (channel->pop(discard)) {
+            }
         } catch (...) {
-            LOG_WARN("WaitAllFutures exception swallowed");
+            LOG_WARN("drain channel exception swallowed");
         }
         throw;
     }

internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp

@@ -260,12 +260,16 @@ ManifestGroupTranslator::get_cells(
             completed_cells[cell_data->cid] =
                 load_group_chunk(cell_data->tables, cell_data->cid);
         }
-    } catch (std::exception& ex) {
-        // make sure all futures are handled, but still throw pop & load ex
+    } catch (...) {
+        // Drain the channel to unblock producers that may be stuck on push()
+        // to a full bounded channel. Without draining, producers block forever
+        // and their task_guard (which calls channel->close()) never executes.
+        std::shared_ptr<milvus::segcore::CellLoadResult> discard;
         try {
-            storage::WaitAllFutures(load_futures);
+            while (channel->pop(discard)) {
+            }
         } catch (...) {
-            LOG_WARN("WaitAllFutures exception swallowed");
+            LOG_WARN("drain channel exception swallowed");
         }
         throw;
     }