code: def validate_statistics_internal(
      statistics: statistics_pb2.DatasetFeatureStatisticsList,
      schema: schema_pb2.Schema,
      environment: Optional[Text] = None,
      previous_span_statistics: Optional[
          statistics_pb2.DatasetFeatureStatisticsList
      ] = None,
      serving_statistics: Optional[
          statistics_pb2.DatasetFeatureStatisticsList
      ] = None,
      previous_version_statistics: Optional[
          statistics_pb2.DatasetFeatureStatisticsList
      ] = None,
      validation_options: Optional[vo.ValidationOptions] = None,
      enable_diff_regions: bool = False,
      custom_validation_config: Optional[
          custom_validation_config_pb2.CustomValidationConfig
      ] = None,
  ) -> anomalies_pb2.Anomalies:
      """Validates the input statistics against the provided input schema.
  
      This method validates the `statistics` against the `schema`. If an optional
      `environment` is specified, the `schema` is filtered using the
      `environment` and the `statistics` is validated against the filtered schema.
      The optional `previous_span_statistics`, `serving_statistics`, and
      `previous_version_statistics` are the statistics computed over the control
      data for drift detection, skew detection, and dataset-level anomaly detection
      across versions, respectively.
  
      Args:
        statistics: A DatasetFeatureStatisticsList protocol buffer denoting the
           statistics computed over the current data. Validation is currently
           supported only for lists with a single DatasetFeatureStatistics proto or
           lists with multiple DatasetFeatureStatistics protos corresponding to data
           slices that include the default slice (i.e., the slice with all
           examples). If a list with multiple DatasetFeatureStatistics protos is
           used, this function will validate the statistics corresponding to the
           default slice.
        schema: A Schema protocol buffer.
        environment: An optional string denoting the validation environment.
            Must be one of the default environments specified in the schema.
            By default, validation assumes that all Examples in a pipeline adhere
            to a single schema. In some cases introducing slight schema variations
            is necessary, for instance features used as labels are required during
            training (and should be validated), but are missing during serving.
            Environments can be used to express such requirements. For example,
            assume a feature named 'LABEL' is required for training, but is expected
            to be missing from serving. This can be expressed by defining two
            distinct environments in schema: ["SERVING", "TRAINING"] and
            associating 'LABEL' only with environment "TRAINING".
        previous_span_statistics: An optional DatasetFeatureStatisticsList protocol
            buffer denoting the statistics computed over an earlier data (for
            example, previous day's data). If provided, the
            `validate_statistics_internal` method will detect if there exists drift
            between current data and previous data. Configuration for drift
            detection can be done by specifying a `drift_comparator` in the schema.
        serving_statistics: An optional DatasetFeatureStatisticsList protocol
            buffer denoting the statistics computed over the serving data. If
            provided, the `validate_statistics_internal` method will identify if
            there exists distribution skew between current data and serving data.
            Configuration for skew detection can be done by specifying a
            `skew_comparator` in the schema.
        previous_version_statistics: An optional DatasetFeatureStatisticsList
            protocol buffer denoting the statistics computed over an earlier data
            (typically, previous run's data within the same day). If provided,
            the `validate_statistics_internal` method will detect if there exists a
            change in the number of examples between current data and previous
            version data. Configuration for such dataset-wide anomaly detection can
            be done by specifying a `num_examples_version_comparator` in the schema.
        validation_options: Optional input used to specify the options of this
            validation.
        enable_diff_regions: Specifies whether to include a comparison between the
            existing schema and the fixed schema in the Anomalies protocol buffer
            output.
        custom_validation_config: An optional config that can be used to specify
            custom validations to perform. If doing single-feature validations,
            the test feature will come from `statistics` and will be mapped to
            `feature` in the SQL query. If doing feature pair validations, the test
            feature will come from `statistics` and will be mapped to `feature_test`
            in the SQL query, and the base feature will come from
            `previous_statistics` and will be mapped to `feature_base` in the SQL
            query. Custom validations are not supported on Windows.
  
      Returns:
        An Anomalies protocol buffer.
  
      Raises:
        TypeError: If any of the input arguments is not of the expected type.
        ValueError: If the input statistics proto contains multiple datasets, none
            of which corresponds to the default slice.
      """
      if not isinstance(
          statistics, statistics_pb2.DatasetFeatureStatisticsList
      ):
          raise TypeError(
              "statistics is of type %s, should be "
              "a DatasetFeatureStatisticsList proto."
              % type(statistics).__name__
          )
  
      # This will raise an exception if there are multiple datasets, none of which
      # corresponds to the default slice.
      dataset_statistics = _get_default_dataset_statistics(statistics)
  
      if not isinstance(schema, schema_pb2.Schema):
          raise TypeError(
              "schema is of type %s, should be a Schema proto."
              % type(schema).__name__
          )
  
      if environment is not None:
          if environment not in schema.default_environment:
              raise ValueError(
                  "Environment %s not found in the schema."
                  % environment
              )
      else:
          environment = ""
  
      if previous_span_statistics is not None:
          if not isinstance(
              previous_span_statistics,
              statistics_pb2.DatasetFeatureStatisticsList,
          ):
              raise TypeError(
                  "previous_span_statistics is of type %s, should be "
                  "a DatasetFeatureStatisticsList proto."
                  % type(previous_span_statistics).__name__
              )
  
          previous_dataset_statistics = _get_default_dataset_statistics(
              previous_span_statistics
          )
  
      if serving_statistics is not None:
          if not isinstance(
              serving_statistics,
              statistics_pb2.DatasetFeatureStatisticsList,
          ):
              raise TypeError(
                  "serving_statistics is of type %s, should be "
                  "a DatasetFeatureStatisticsList proto."
                  % type(serving_statistics).__name__
              )
  
          serving_dataset_statistics = _get_default_dataset_statistics(
              serving_statistics
          )
  
      if previous_version_statistics is not None:
          if not isinstance(
              previous_version_statistics,
              statistics_pb2.DatasetFeatureStatisticsList,
          ):
              raise TypeError(
                  "previous_version_statistics is of type %s, should be "
                  "a DatasetFeatureStatisticsList proto."
                  % type(previous_version_statistics).__name__
              )
  
          previous_version_dataset_statistics = (
              _get_default_dataset_statistics(
                  previous_version_statistics
              )
          )
  
      # Serialize the input protos.
      serialized_schema = schema.SerializeToString()
      serialized_stats = dataset_statistics.SerializeToString()
      serialized_previous_span_stats = (
          previous_dataset_statistics.SerializeToString()
          if previous_span_statistics is not None
          else ""
      )
      serialized_serving_stats = (
          serving_dataset_statistics.SerializeToString()
          if serving_statistics is not None
          else ""
      )
      serialized_previous_version_stats = (
          previous_version_dataset_statistics.SerializeToString()
          if previous_version_statistics is not None
          else ""
      )
  
      features_needed_pb = validation_metadata_pb2.FeaturesNeededProto()
      if (
          validation_options is not None
          and validation_options.features_needed
      ):
          for (
              path,
              reason_list,
          ) in validation_options.features_needed.items():
              path_and_reason_feature_need = (
                  features_needed_pb.path_and_reason_feature_need.add()
              )
              path_and_reason_feature_need.path.CopyFrom(
                  path.to_proto()
              )
              for reason in reason_list:
                  r = (
                      path_and_reason_feature_need.reason_feature_needed.add()
                  )
                  r.comment = reason.comment
  
      serialized_features_needed = (
          features_needed_pb.SerializeToString()
      )
  
      validation_config = validation_config_pb2.ValidationConfig()
      if validation_options is not None:
          validation_config.new_features_are_warnings = (
              validation_options.new_features_are_warnings
          )
          for override in validation_options.severity_overrides:
              validation_config.severity_overrides.append(override)
      serialized_validation_config = (
          validation_config.SerializeToString()
      )
  
      anomalies_proto_string = (
          pywrap_tensorflow_data_validation.ValidateFeatureStatistics(
              tf.compat.as_bytes(serialized_stats),
              tf.compat.as_bytes(serialized_schema),
              tf.compat.as_bytes(environment),
              tf.compat.as_bytes(serialized_previous_span_stats),
              tf.compat.as_bytes(serialized_serving_stats),
              tf.compat.as_bytes(serialized_previous_version_stats),
              tf.compat.as_bytes(serialized_features_needed),
              tf.compat.as_bytes(serialized_validation_config),
              enable_diff_regions,
          )
      )
  
      # Parse the serialized Anomalies proto.
      result = anomalies_pb2.Anomalies()
      result.ParseFromString(anomalies_proto_string)
  
      if custom_validation_config is not None:
          serialized_previous_statistics = (
              previous_span_statistics.SerializeToString()
              if previous_span_statistics is not None
              else ""
          )
          custom_anomalies_string = pywrap_tensorflow_data_validation.CustomValidateStatistics(
              tf.compat.as_bytes(statistics.SerializeToString()),
              tf.compat.as_bytes(serialized_previous_statistics),
              tf.compat.as_bytes(
                  custom_validation_config.SerializeToString()
              ),
              tf.compat.as_bytes(environment),
          )
          custom_anomalies = anomalies_pb2.Anomalies()
          custom_anomalies.ParseFromString(custom_anomalies_string)
          result = _merge_custom_anomalies(result, custom_anomalies)
  
      return result
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import itertools
  import logging
  from typing import Callable, Iterable, List, Optional, Text, Tuple, Set
  import apache_beam
  import pyarrow
  import tensorflow
  from tensorflow_data_validation import constants
  from tensorflow_data_validation import types
  from tensorflow_data_validation.anomalies.proto import custom_validation_config_pb2
  from tensorflow_data_validation.anomalies.proto import validation_config_pb2
  from tensorflow_data_validation.anomalies.proto import validation_metadata_pb2
  from tensorflow_data_validation.api import validation_options
  from tensorflow_data_validation.pywrap.tensorflow_data_validation_extension import validation
  from tensorflow_data_validation.skew import feature_skew_detector
  from tensorflow_data_validation.skew.protos import feature_skew_results_pb2
  from tensorflow_data_validation.statistics import stats_impl
  from tensorflow_data_validation.statistics import stats_options
  from tensorflow_data_validation.utils import anomalies_util
  from tensorflow_data_validation.utils import slicing_util
  from tensorflow_data_validation.utils import stats_util
  from tensorflow_metadata.proto.v0 import anomalies_pb2
  from tensorflow_metadata.proto.v0 import schema_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 7 for calling a global function with example usage: validate_statistics_internal(statistics, schema, environment, previous_span_statistics, serving_statistics, previous_version_statistics, validation_options, enable_diff_regions, custom_validation_config) and returns: result