class_docstr: A PipelineLoader for datasets constructed with ``from_blaze``.
  
  Parameters
  ----------
  dsmap : mapping, optional
      An initial mapping of datasets to ``ExprData`` objects.
      NOTE: Further mutations to this map will not be reflected by this
      object.
  data_query_time : time, optional
      The time to use for the data query cutoff.
  data_query_tz : tzinfo or str, optional
      The timezeone to use for the data query cutoff.
  pool : Pool, optional
      The pool to use to run blaze queries concurrently. This object must
      support ``imap_unordered``, ``apply`` and ``apply_async`` methods.
  
  Attributes
  ----------
  pool : Pool
      The pool to use to run blaze queries concurrently. This object must
      support ``imap_unordered``, ``apply`` and ``apply_async`` methods.
      It is possible to change the pool after the loader has been
      constructed. This allows us to set a new pool for the ``global_loader``
      like: ``global_loader.pool = multiprocessing.Pool(4)``.
  
  See Also
  --------
  :class:`catalyst.utils.pool.SequentialPool`
  :class:`multiprocessing.Pool`
class_docstr_tok: ["A","PipelineLoader","for","datasets","constructed","with","``","from_blaze","``",".","Parameters","--","--","--","--","--","dsmap",":","mapping",",","optional","An","initial","mapping","of","datasets","to","``","ExprData","``","objects",".","NOTE",":","Further","mutations","to","this","map","will","not","be","reflected","by","this","object",".","data_query_time",":","time",",","optional","The","time","to","use","for","the","data","query","cutoff",".","data_query_tz",":","tzinfo","or","str",",","optional","The","timezeone","to","use","for","the","data","query","cutoff",".","pool",":","Pool",",","optional","The","pool","to","use","to","run","blaze","queries","concurrently",".","This","object","must","support","``","imap_unordered","``",",","``","apply","``","and","``","apply_async","``","methods",".","Attributes","--","--","--","--","--","pool",":","Pool","The","pool","to","use","to","run","blaze","queries","concurrently",".","This","object","must","support","``","imap_unordered","``",",","``","apply","``","and","``","apply_async","``","methods",".","It","is","possible","to","change","the","pool","after","the","loader","has","been","constructed",".","This","allows","us","to","set","a","new","pool","for","the","``","global_loader","``","like",":","``","global_loader.pool","=","multiprocessing.Pool","(","4",")","``",".","See","Also","--","--","--","--",":","class",":","`","catalyst.utils.pool.SequentialPool","`",":","class",":","`","multiprocessing.Pool","`"]
code: def _load_dataset(self, dates, assets, mask, columns):
      try:
          (dataset,) = set(map(getdataset, columns))
      except ValueError:
          raise AssertionError(
              "all columns must come from the same dataset"
          )
  
      (
          expr,
          deltas,
          checkpoints,
          odo_kwargs,
          apply_deltas_adjustments,
      ) = self[dataset]
      have_sids = dataset.ndim == 2
      asset_idx = pd.Series(index=assets, data=np.arange(len(assets)))
      assets = list(map(int, assets))  # coerce from numpy.int64
      added_query_fields = {AD_FIELD_NAME, TS_FIELD_NAME} | (
          {SID_FIELD_NAME} if have_sids else set()
      )
      requested_columns = set(map(getname, columns))
      colnames = sorted(added_query_fields | requested_columns)
  
      data_query_time = self._data_query_time
      data_query_tz = self._data_query_tz
      lower_dt, upper_dt = normalize_data_query_bounds(
          dates[0],
          dates[-1],
          data_query_time,
          data_query_tz,
      )
  
      def collect_expr(e, lower):
          """Materialize the expression as a dataframe.
  
          Parameters
          ----------
          e : Expr
              The baseline or deltas expression.
          lower : datetime
              The lower time bound to query.
  
          Returns
          -------
          result : pd.DataFrame
              The resulting dataframe.
  
          Notes
          -----
          This can return more data than needed. The in memory reindex will
          handle this.
          """
          predicate = e[TS_FIELD_NAME] <= upper_dt
          if lower is not None:
              predicate &= e[TS_FIELD_NAME] >= lower
  
          return odo(e[predicate][colnames], pd.DataFrame, **odo_kwargs)
  
      lower, materialized_checkpoints = get_materialized_checkpoints(
          checkpoints, colnames, lower_dt, odo_kwargs
      )
  
      materialized_expr = self.pool.apply_async(
          collect_expr, (expr, lower)
      )
      materialized_deltas = (
          self.pool.apply(collect_expr, (deltas, lower))
          if deltas is not None
          else pd.DataFrame(columns=colnames)
      )
  
      if materialized_checkpoints is not None:
          materialized_expr = pd.concat(
              (
                  materialized_checkpoints,
                  materialized_expr.get(),
              ),
              ignore_index=True,
              copy=False,
          )
  
      # It's not guaranteed that assets returned by the engine will contain
      # all sids from the deltas table; filter out such mismatches here.
      if not materialized_deltas.empty and have_sids:
          materialized_deltas = materialized_deltas[
              materialized_deltas[SID_FIELD_NAME].isin(assets)
          ]
  
      if data_query_time is not None:
          for m in (materialized_expr, materialized_deltas):
              m.loc[:, TS_FIELD_NAME] = m.loc[:, TS_FIELD_NAME].astype(
                  "datetime64[ns]"
              )
  
              normalize_timestamp_to_query_time(
                  m,
                  data_query_time,
                  data_query_tz,
                  inplace=True,
                  ts_field=TS_FIELD_NAME,
              )
  
      # Inline the deltas that changed our most recently known value.
      # Also, we reindex by the dates to create a dense representation of
      # the data.
      sparse_output, non_novel_deltas = overwrite_novel_deltas(
          materialized_expr,
          materialized_deltas,
          dates,
      )
      # If we ever have cases where we find out about multiple asof_dates'
      # data on the same TS, we want to make sure that last_in_date_group
      # selects the correct last asof_date's value.
      sparse_output.sort_values(AD_FIELD_NAME, inplace=True)
      non_novel_deltas.sort_values(AD_FIELD_NAME, inplace=True)
      if AD_FIELD_NAME not in requested_columns:
          sparse_output.drop(AD_FIELD_NAME, axis=1, inplace=True)
  
      sparse_deltas = last_in_date_group(
          non_novel_deltas,
          dates,
          assets,
          reindex=False,
          have_sids=have_sids,
      )
      dense_output = last_in_date_group(
          sparse_output,
          dates,
          assets,
          reindex=True,
          have_sids=have_sids,
      )
      ffill_across_cols(
          dense_output, columns, {c.name: c.name for c in columns}
      )
  
      # By default, no non-novel deltas are applied.
      def no_adjustments_from_deltas(*args):
          return {}
  
      adjustments_from_deltas = no_adjustments_from_deltas
      if have_sids:
          if apply_deltas_adjustments:
              adjustments_from_deltas = (
                  adjustments_from_deltas_with_sids
              )
          column_view = identity
      else:
          # If we do not have sids, use the column view to make a single
          # column vector which is unassociated with any assets.
          column_view = op.itemgetter(np.s_[:, np.newaxis])
          if apply_deltas_adjustments:
              adjustments_from_deltas = adjustments_from_deltas_no_sids
          mask = np.full(
              shape=(len(mask), 1),
              fill_value=True,
              dtype=bool_dtype,
          )
  
      return {
          column: AdjustedArray(
              column_view(
                  dense_output[column.name].values.astype(column.dtype),
              ),
              mask,
              adjustments_from_deltas(
                  dates,
                  sparse_output[TS_FIELD_NAME].values,
                  column_idx,
                  column.name,
                  asset_idx,
                  sparse_deltas,
              ),
              column.missing_value,
          )
          for column_idx, column in enumerate(columns)
      }
  
imports: from __future__ import division, absolute_import
  from abc import ABCMeta, abstractproperty
  from collections import namedtuple, defaultdict
  from functools import partial
  from itertools import count
  import warnings
  from weakref import WeakKeyDictionary
  import blaze
  from datashape import Date, DateTime, Option, isrecord, isscalar
  import numpy
  from odo import odo
  import pandas
  from six import with_metaclass, PY2, itervalues, iteritems
  from toolz import complement, compose, flip, groupby, identity, memoize, merge
  import toolz.curried.operator
  from catalyst.pipeline.common import AD_FIELD_NAME, SID_FIELD_NAME, TS_FIELD_NAME
  from catalyst.pipeline.data.dataset import DataSet, Column
  from catalyst.pipeline.loaders.utils import check_data_query_args, last_in_date_group, normalize_data_query_bounds, normalize_timestamp_to_query_time, ffill_across_cols
  from catalyst.pipeline.sentinels import NotSpecified
  from catalyst.lib.adjusted_array import AdjustedArray, can_represent_dtype
  from catalyst.lib.adjustment import make_adjustment_from_indices, OVERWRITE
  from catalyst.utils.input_validation import expect_element, ensure_timezone, optionally
  from catalyst.utils.numpy_utils import bool_dtype
  from catalyst.utils.pool import SequentialPool
  from catalyst.utils.preprocess import preprocess
label_desc: Use image node_id 7 for calling the BlazeLoader obj's underlying member method code with example usage: obj._load_dataset(dates, assets, mask, columns) and returns: unknown, odo, dict