class_docstr: Datasource for reading from and writing to MongoDB.
class_docstr_tok: ["Datasource","for","reading","from","and","writing","to","MongoDB","."]
code: def get_read_tasks(self, parallelism: int) -> List[ReadTask]:
      from bson.objectid import ObjectId
  
      self._get_or_create_client()
      coll = self._client[self._database][self._collection]
      match_query = self._get_match_query(self._pipeline)
      partitions_ids = list(
          coll.aggregate(
              [
                  {"$match": match_query},
                  {
                      "$bucketAuto": {
                          "groupBy": "$_id",
                          "buckets": parallelism,
                      }
                  },
              ],
              allowDiskUse=True,
          )
      )
  
      def make_block(
          uri: str,
          database: str,
          collection: str,
          pipeline: List[Dict],
          min_id: ObjectId,
          max_id: ObjectId,
          right_closed: bool,
          schema: "pymongoarrow.api.Schema",
          kwargs: dict,
      ) -> Block:
          import pymongo
          from pymongoarrow.api import aggregate_arrow_all
  
          # A range query over the partition.
          match = [
              {
                  "$match": {
                      "_id": {
                          "$gte": min_id,
                          "$lte" if right_closed else "$lt": max_id,
                      }
                  }
              }
          ]
          client = pymongo.MongoClient(uri)
          return aggregate_arrow_all(
              client[database][collection],
              match + pipeline,
              schema=schema,
              **kwargs,
          )
  
      read_tasks: List[ReadTask] = []
  
      for i, partition in enumerate(partitions_ids):
          metadata = BlockMetadata(
              num_rows=partition["count"],
              size_bytes=partition["count"] * self._avg_obj_size,
              schema=None,
              input_files=None,
              exec_stats=None,
          )
          make_block_args = (
              self._uri,
              self._database,
              self._collection,
              self._pipeline,
              partition["_id"]["min"],
              partition["_id"]["max"],
              i == len(partitions_ids) - 1,
              self._schema,
              self._mongo_args,
          )
          read_task = ReadTask(
              lambda args=make_block_args: [make_block(*args)],
              metadata,
          )
          read_tasks.append(read_task)
  
      return read_tasks
  
file_path: ray/python/ray/data/datasource/mongo_datasource.py
imports: import logging
  from typing import TYPE_CHECKING, Dict, List, Optional
  from ray.data.block import Block, BlockMetadata
  from ray.data.datasource.datasource import Datasource, ReadTask
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 5 for calling the MongoDatasource obj's underlying member method code with example usage: obj.get_read_tasks(parallelism) and returns: read_tasks, aggregate_arrow_all