code: def from_blaze(
      expr,
      deltas="auto",
      checkpoints="auto",
      loader=None,
      resources=None,
      odo_kwargs=None,
      missing_values=None,
      no_deltas_rule="warn",
      no_checkpoints_rule="warn",
      apply_deltas_adjustments=True,
  ):
      """Create a Pipeline API object from a blaze expression.
  
      Parameters
      ----------
      expr : Expr
          The blaze expression to use.
      deltas : Expr, 'auto' or None, optional
          The expression to use for the point in time adjustments.
          If the string 'auto' is passed, a deltas expr will be looked up
          by stepping up the expression tree and looking for another field
          with the name of ``expr._name`` + '_deltas'. If None is passed, no
          deltas will be used.
      checkpoints : Expr, 'auto' or None, optional
          The expression to use for the forward fill checkpoints.
          If the string 'auto' is passed, a checkpoints expr will be looked up
          by stepping up the expression tree and looking for another field
          with the name of ``expr._name`` + '_checkpoints'. If None is passed,
          no checkpoints will be used.
      loader : BlazeLoader, optional
          The blaze loader to attach this pipeline dataset to. If None is passed,
          the global blaze loader is used.
      resources : dict or any, optional
          The data to execute the blaze expressions against. This is used as the
          scope for ``bz.compute``.
      odo_kwargs : dict, optional
          The keyword arguments to pass to odo when evaluating the expressions.
      missing_values : dict[str -> any], optional
          A dict mapping column names to missing values for those columns.
          Missing values are required for integral columns.
      no_deltas_rule : {'warn', 'raise', 'ignore'}, optional
          What should happen if ``deltas='auto'`` but no deltas can be found.
          'warn' says to raise a warning but continue.
          'raise' says to raise an exception if no deltas can be found.
          'ignore' says take no action and proceed with no deltas.
      no_checkpoints_rule : {'warn', 'raise', 'ignore'}, optional
          What should happen if ``checkpoints='auto'`` but no checkpoints can be
          found. 'warn' says to raise a warning but continue.
          'raise' says to raise an exception if no deltas can be found.
          'ignore' says take no action and proceed with no deltas.
      apply_deltas_adjustments : bool, optional
          Whether or not deltas adjustments should be applied for this dataset.
          True by default because not applying deltas adjustments is an exception
          rather than the rule.
  
      Returns
      -------
      pipeline_api_obj : DataSet or BoundColumn
          Either a new dataset or bound column based on the shape of the expr
          passed in. If a table shaped expression is passed, this will return
          a ``DataSet`` that represents the whole table. If an array-like shape
          is passed, a ``BoundColumn`` on the dataset that would be constructed
          from passing the parent is returned.
      """
      if "auto" in {deltas, checkpoints}:
          invalid_nodes = tuple(
              filter(is_invalid_deltas_node, expr._subterms())
          )
          if invalid_nodes:
              raise TypeError(
                  "expression with auto %s may only contain (%s) nodes,"
                  " found: %s"
                  % (
                      " or ".join(
                          ["deltas"]
                          if deltas is not None
                          else [] + ["checkpoints"]
                          if checkpoints is not None
                          else [],
                      ),
                      ", ".join(
                          map(get__name__, valid_deltas_node_types)
                      ),
                      ", ".join(
                          set(
                              map(
                                  compose(get__name__, type),
                                  invalid_nodes,
                              )
                          ),
                      ),
                  ),
              )
      deltas = _get_metadata(
          "deltas",
          expr,
          deltas,
          no_deltas_rule,
      )
      checkpoints = _get_metadata(
          "checkpoints",
          expr,
          checkpoints,
          no_checkpoints_rule,
      )
  
      # Check if this is a single column out of a dataset.
      if bz.ndim(expr) != 1:
          raise TypeError(
              "expression was not tabular or array-like,"
              " %s dimensions: %d"
              % (
                  "too many" if bz.ndim(expr) > 1 else "not enough",
                  bz.ndim(expr),
              ),
          )
  
      single_column = None
      if isscalar(expr.dshape.measure):
          # This is a single column. Record which column we are to return
          # but create the entire dataset.
          single_column = rename = expr._name
          field_hit = False
          if not isinstance(expr, traversable_nodes):
              raise TypeError(
                  "expression '%s' was array-like but not a simple field of"
                  " some larger table" % str(expr),
              )
          while isinstance(expr, traversable_nodes):
              if isinstance(expr, bz.expr.Field):
                  if not field_hit:
                      field_hit = True
                  else:
                      break
              rename = expr._name
              expr = expr._child
          dataset_expr = expr.relabel({rename: single_column})
      else:
          dataset_expr = expr
  
      measure = dataset_expr.dshape.measure
      if not isrecord(measure) or AD_FIELD_NAME not in measure.names:
          raise TypeError(
              "The dataset must be a collection of records with at least an"
              " '{ad}' field. Fields provided: '{fields}'\nhint: maybe you need"
              " to use `relabel` to change your field names".format(
                  ad=AD_FIELD_NAME,
                  fields=measure,
              ),
          )
      _check_datetime_field(AD_FIELD_NAME, measure)
      dataset_expr, deltas, checkpoints = _ensure_timestamp_field(
          dataset_expr,
          deltas,
          checkpoints,
      )
  
      if deltas is not None and (
          sorted(deltas.dshape.measure.fields) != sorted(measure.fields)
      ):
          raise TypeError(
              "baseline measure != deltas measure:\n%s != %s"
              % (
                  measure,
                  deltas.dshape.measure,
              ),
          )
      if checkpoints is not None and (
          sorted(checkpoints.dshape.measure.fields)
          != sorted(measure.fields)
      ):
          raise TypeError(
              "baseline measure != checkpoints measure:\n%s != %s"
              % (
                  measure,
                  checkpoints.dshape.measure,
              ),
          )
  
      # Ensure that we have a data resource to execute the query against.
      _check_resources("expr", dataset_expr, resources)
      _check_resources("deltas", deltas, resources)
      _check_resources("checkpoints", checkpoints, resources)
  
      # Create or retrieve the Pipeline API dataset.
      if missing_values is None:
          missing_values = {}
      ds = new_dataset(
          dataset_expr, deltas, frozenset(missing_values.items())
      )
  
      # Register our new dataset with the loader.
      (loader if loader is not None else global_loader)[ds] = ExprData(
          bind_expression_to_resources(dataset_expr, resources),
          bind_expression_to_resources(deltas, resources)
          if deltas is not None
          else None,
          bind_expression_to_resources(checkpoints, resources)
          if checkpoints is not None
          else None,
          odo_kwargs=odo_kwargs,
          apply_deltas_adjustments=apply_deltas_adjustments,
      )
      if single_column is not None:
          # We were passed a single column, extract and return it.
          return getattr(ds, single_column)
      return ds
  
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 8 for calling a global function with example usage: from_blaze(expr, deltas, checkpoints, loader, resources, odo_kwargs, missing_values, no_deltas_rule, no_checkpoints_rule, apply_deltas_adjustments) and returns: ds, getattr