Skip to main content

impulse_query_engine.analyze.query.query_builder

QueryBuilder​

class QueryBuilder()

__init__​

def __init__(db: "impulse_query_engine.analyze.MeasurementDB")

Initialize the QueryBuilder.

Arguments:

  • db (impulse_query_engine.analyze.MeasurementDB): Measurement database object.

where​

def where(*args)

Add filter expressions to the query.

Arguments:

  • *args (list): Filter expressions to be added.

Returns:

QueryBuilder: The updated QueryBuilder instance.

filter​

def filter(*args)

Alias for where().

Arguments:

  • *args (list): Filter expressions to be added.

Returns:

QueryBuilder: The updated QueryBuilder instance.

havingTag​

def havingTag(**kwargs)

Add tag-based filters to the query.

Arguments:

  • **kwargs (dict): Tag-value pairs to filter by.

Returns:

QueryBuilder: The updated QueryBuilder instance.

tag​

def tag(key: str, cast_type: str | None = None) -> TagSelector

Create a tag selector for the given key.

Arguments:

  • key (str): Name of the tag (element_id in the EAV table).
  • cast_type (str or None): Spark type to cast the tag value to before comparison (e.g. "int", "double", "string").

Returns:

TagSelector: Tag selector object.

metric​

def metric(name) -> MetricSelector

Create a metric selector for the given name.

Arguments:

  • name (str): Name of the metric.

Returns:

MetricSelector: Metric selector object.

channel​

def channel(**kwargs) -> TimeSeriesSelector

Create a time series selector for the given channel tags.

Arguments:

  • **kwargs (dict): Channel tag-value pairs.

Returns:

TimeSeriesSelector: Time series selector object.

select​

def select(*args) -> Self

Set the selection expressions for the query.

Arguments:

  • *args (list): Selection expressions.

Returns:

QueryBuilder: The updated QueryBuilder instance.

solve​

def solve(spark,
          solver: QuerySolver = BlobSolver(),
          pre_filtered_containers_df: DataFrame = None) -> DataFrame

Execute the query using the specified solver and return a Spark DataFrame.

Arguments:

  • spark (SparkSession): Spark session used for query execution.
  • solver (QuerySolver): Query solver to use (default is BlobSolver).
  • pre_filtered_containers_df (DataFrame): Pre-filtered container metrics DataFrame for incremental processing. When provided, only these containers will be processed. When None, all containers matching query filters are processed (full mode).

Returns:

pyspark.sql.DataFrame: DataFrame containing query results.

solve_calculated_channels​

def solve_calculated_channels(
        spark,
        solver: QuerySolver = BlobSolver(),
        pre_filtered_containers_df: DataFrame = None) -> DataFrame

Compute calculated channels and return a narrow silver-shaped DataFrame.

Every selection must be a :class:CalculatedChannel. This runs the same metadata filter pipeline as :meth:solve (resolving the input channels each calculated channel depends on), then evaluates each calculated channel per container and emits rows in the silver channel_data shape — container_id, channel_id, tstart, tend, value — plus a single identity MapType(string, string) column holding each channel's identity dict.

Arguments:

  • spark (SparkSession): Spark session used for query execution.
  • solver (QuerySolver): Query solver to use. Must implement solve_calculated_channels (DefaultSolver does); the default BlobSolver does not.
  • pre_filtered_containers_df (DataFrame): Pre-filtered container metrics for incremental processing. When provided, only these containers are processed; when None, all containers matching the query filters are processed.

Raises:

  • ValueError: If any selection is not a CalculatedChannel, or if a wrapped expression does not evaluate to a SampleSeries.

Returns:

pyspark.sql.DataFrame: Narrow DataFrame [container_id, channel_id, tstart, tend, value, identity].

toPandas​

def toPandas(spark, solver: QuerySolver = BlobSolver()) -> pd.DataFrame

Execute the query and collect results into a Pandas DataFrame.

Arguments:

  • spark (SparkSession): Spark session used for query execution.
  • solver (QuerySolver): Query solver to use (default is BlobSolver).

Returns:

pd.DataFrame: Pandas DataFrame containing query results.