code: def new_dataset(expr, deltas, missing_values):
      """
      Creates or returns a dataset from a pair of blaze expressions.
  
      Parameters
      ----------
      expr : Expr
          The blaze expression representing the first known values.
      deltas : Expr
          The blaze expression representing the deltas to the data.
      missing_values : frozenset((name, value) pairs
          Association pairs column name and missing_value for that column.
  
          This needs to be a frozenset rather than a dict or tuple of tuples
          because we want a collection that's unordered but still hashable.
  
      Returns
      -------
      ds : type
          A new dataset type.
  
      Notes
      -----
      This function is memoized. repeated calls with the same inputs will return
      the same type.
      """
      missing_values = dict(missing_values)
      class_dict = {"ndim": 2 if SID_FIELD_NAME in expr.fields else 1}
      for name, type_ in expr.dshape.measure.fields:
          # Don't generate a column for sid or timestamp, since they're
          # implicitly the labels if the arrays that will be passed to pipeline
          # Terms.
          if name in (SID_FIELD_NAME, TS_FIELD_NAME):
              continue
          type_ = datashape_type_to_numpy(type_)
          if can_represent_dtype(type_):
              col = Column(
                  type_,
                  missing_values.get(name, NotSpecified),
              )
          else:
              col = NonPipelineField(name, type_)
          class_dict[name] = col
  
      name = expr._name
      if name is None:
          name = next(_new_names)
  
      # unicode is a name error in py3 but the branch is only hit
      # when we are in python 2.
      if PY2 and isinstance(name, unicode):  # pragma: no cover # noqa
          name = name.encode("utf-8")
  
      return type(name, (DataSet,), class_dict)
  
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 2 for calling a global function with example usage: new_dataset(expr, deltas, missing_values) and returns: type