Skip to content

Poisson Workload Creator

parser = argparse.ArgumentParser(description='Create a Poisson workload.') module-attribute

args = parser.parse_args() module-attribute

PoissonArrivalPhase dataclass

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@dataclass(frozen=True)
class PoissonArrivalPhase:
    num_queries: int
    poisson_lambda: float

    @property
    def name(self) -> str:
        return f"{self.num_queries}q{self.poisson_lambda}l"

    def generate_rel_arrival_times_s(
        self,
        rng: np.random.Generator,
        reference_rel_time_s: float,
        include_gap_pre_first: bool = True,
    ) -> list[float]:
        inter_arrival_times_s = list(
            rng.exponential(1 / self.poisson_lambda, size=self.num_queries)
        )
        if not include_gap_pre_first:
            inter_arrival_times_s = [0] + inter_arrival_times_s[:-1]
        arrival_times_s = (
            np.cumsum(inter_arrival_times_s) + reference_rel_time_s
        )
        return arrival_times_s.tolist()

num_queries instance-attribute

poisson_lambda instance-attribute

name property

__init__(num_queries, poisson_lambda)

generate_rel_arrival_times_s(rng, reference_rel_time_s, include_gap_pre_first=True)

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def generate_rel_arrival_times_s(
    self,
    rng: np.random.Generator,
    reference_rel_time_s: float,
    include_gap_pre_first: bool = True,
) -> list[float]:
    inter_arrival_times_s = list(
        rng.exponential(1 / self.poisson_lambda, size=self.num_queries)
    )
    if not include_gap_pre_first:
        inter_arrival_times_s = [0] + inter_arrival_times_s[:-1]
    arrival_times_s = (
        np.cumsum(inter_arrival_times_s) + reference_rel_time_s
    )
    return arrival_times_s.tolist()

PoissonWorkloadCreator

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
class PoissonWorkloadCreator:
    @staticmethod
    def name_from_params(
        num_templates: int,
        num_query_texts_per_template: int,
        num_queries_per_query_text: int | None,
        phases: list[PoissonArrivalPhase],
        seed: int,
        num_total_queries: int | None = None,
    ) -> str:
        global_poisson_lambda = (
            phases[0].poisson_lambda if len(phases) == 1 else None
        )
        name = "_".join(
            [
                "poisson",
                str(num_templates),
                str(num_query_texts_per_template),
                str(num_queries_per_query_text),
                str(global_poisson_lambda),
                str(seed),
            ]
        )

        if num_total_queries is not None:
            name += f"_{num_total_queries}"

        if len(phases) > 1:
            phase_str = "_".join(phase.name for phase in phases)
            name += f"_phased_{phase_str}"

        return name

    @staticmethod
    def make_phased_profile(
        num_queries_per_phase: list[int],
        poisson_lambda_per_phase: list[float],
    ) -> list[PoissonArrivalPhase]:
        """
        Create a multi-phase workload with the given number of queries and
        Poisson lambdas. The i-th phase will have num_queries_per_phase[i]
        queries and a Poisson lambda of poisson_lambda_per_phase[i]. The lambda
        governs interarrival times *before* each query in a phase.
        """
        return [
            PoissonArrivalPhase(num_queries=num_q, poisson_lambda=lam)
            for num_q, lam in zip(
                num_queries_per_phase, poisson_lambda_per_phase
            )
        ]

    @staticmethod
    def make_bursty_profile(
        total_num_queries: int,
        num_lull_burst_cycles: int,
        lull_poisson_lambda: float,
        burst_poisson_lambda: float,
        fraction_of_queries_in_bursts: float,
    ) -> list[PoissonArrivalPhase]:
        """
        Generate a phased workload alternating between "lull" phases with low
        arrival rate and "burst" phases with high arrival rate.

        Parameters
        ----------
        name: Name for the profile.
        total_num_queries: Total number of queries across all phases.
        num_lull_burst_cycles: Number of alternating "lull" and "burst" cycles.
        lull_poisson_lambda: Poisson lambda for "lull" phases.
        burst_poisson_lambda: Poisson lambda for "burst" phases.
        fraction_of_queries_in_bursts: Fraction of queries that arriva during
        "burst" phases.
        """

        num_queries_per_burst = int(
            total_num_queries
            * fraction_of_queries_in_bursts
            / num_lull_burst_cycles
        )
        pessimistic_num_queries_per_lull = (
            total_num_queries - num_queries_per_burst * num_lull_burst_cycles
        ) // num_lull_burst_cycles

        remaining_queries = total_num_queries
        phases = []
        while remaining_queries > 0:
            # Add lull phase
            num_queries_this_lull = min(
                pessimistic_num_queries_per_lull, remaining_queries
            )
            remaining_queries -= num_queries_this_lull
            phases.append(
                PoissonArrivalPhase(
                    num_queries=num_queries_this_lull,
                    poisson_lambda=lull_poisson_lambda,
                )
            )

            # Add burst phase
            num_queries_this_burst = min(
                num_queries_per_burst, remaining_queries
            )
            remaining_queries -= num_queries_this_burst
            phases.append(
                PoissonArrivalPhase(
                    num_queries=num_queries_this_burst,
                    poisson_lambda=burst_poisson_lambda,
                )
            )

        return phases

    @staticmethod
    def create_poisson_workload(
        num_templates: int,
        num_query_texts_per_template: int,
        num_queries_per_query_text: int,
        poisson_lambda: float,
        seed: int,
        print_summary: bool = True,
    ) -> Workload:
        return PoissonWorkloadCreator.create_poisson_workload_phased(
            num_templates=num_templates,
            num_query_texts_per_template=num_query_texts_per_template,
            num_queries_per_query_text=num_queries_per_query_text,
            phases=[
                PoissonArrivalPhase(
                    num_queries=(
                        num_templates
                        * num_query_texts_per_template
                        * num_queries_per_query_text
                    ),
                    poisson_lambda=poisson_lambda,
                )
            ],
            seed=seed,
            print_summary=print_summary,
            include_total_in_name=False,
        )



    @staticmethod
    def create_poisson_workload_phased(
        num_templates: int,
        num_query_texts_per_template: int,
        num_queries_per_query_text: Optional[int],
        phases: list[PoissonArrivalPhase],
        seed: int,
        print_summary: bool = True,
        include_total_in_name: bool = False,
    ) -> Workload:
        num_total_queries = sum(phase.num_queries for phase in phases)
        rng = np.random.default_rng(seed)
        workload_name = PoissonWorkloadCreator.name_from_params(
            num_templates=num_templates,
            num_query_texts_per_template=num_query_texts_per_template,
            num_queries_per_query_text=num_queries_per_query_text,
            phases=phases,
            seed=seed,
            num_total_queries=(
                None if not include_total_in_name else num_total_queries
            ),
        )

        # Determine which templates to use
        all_templates = list(range(1, 100))
        rng.shuffle(all_templates)
        selected_templates = all_templates[:num_templates]

        # Create the queries in sorted order and then shuffle.
        if num_queries_per_query_text is None:
            num_queries_per_query_text = num_total_queries
        query_text_ids = []
        for template_idx in sorted(selected_templates):
            for query_text_idx in range(1, num_query_texts_per_template + 1):
                for _ in range(1, num_queries_per_query_text + 1):
                    query_text_ids.append(
                        f"ext_tpcds1000#{template_idx:03d}#{query_text_idx:03d}"
                    )
        rng.shuffle(query_text_ids)
        query_text_ids = query_text_ids[:num_total_queries]

        return PoissonWorkloadCreator._package_queries_into_workload(
            query_text_ids=query_text_ids,
            phases=phases,
            workload_name=workload_name,
            print_summary=print_summary,
            rng=rng,
        )

    @staticmethod
    def _package_queries_into_workload(
        query_text_ids: list[str],
        phases: list[PoissonArrivalPhase],
        workload_name: str,
        print_summary: bool,
        rng: np.random.Generator,
    ) -> Workload:

        # Create submission times using a Poisson process for each phase.
        reference_rel_time_s = 0.0
        rel_arrival_times_s = phases[0].generate_rel_arrival_times_s(
            rng=rng,
            reference_rel_time_s=reference_rel_time_s,
            include_gap_pre_first=False,
        )
        for phase in phases[1:]:
            reference_rel_time_s = rel_arrival_times_s[-1]
            phase_rel_arrival_times_s = phase.generate_rel_arrival_times_s(
                rng=rng,
                reference_rel_time_s=reference_rel_time_s,
                include_gap_pre_first=True,
            )
            rel_arrival_times_s.extend(phase_rel_arrival_times_s)

        # Package into workload.
        gen_start_time = datetime.fromisoformat("2026-01-01T00:00:00")
        records = []
        for i, (query_text_id, rel_arrival_time_s) in enumerate(
            zip(query_text_ids, rel_arrival_times_s)
        ):
            record = {
                "query_id": f"query_{i}",
                "abs_start_time": gen_start_time
                + timedelta(seconds=rel_arrival_time_s),
                "query_text_id": query_text_id,
                "repetition_id": f"query_{i}",
            }
            records.append(record)
        df = pd.DataFrame(records)
        out_path = pu.get_workloads_dir() / f"{workload_name}.parquet"
        df.to_parquet(out_path)

        # Print a nice summary
        workload_config = WorkloadConfig(
            workload_name=workload_name,
        )
        workload = Workload(workload_config)
        if print_summary:
            workload.print_summary()

        return workload

    @staticmethod
    def create_poisson_workload_with_n_queries(
        num_templates: int,
        num_query_texts_per_template: int,
        num_total_queries: int,
        poisson_lambda: float,
        seed: int,
        print_summary: bool = True,
    ) -> Workload:
        return PoissonWorkloadCreator.create_poisson_workload_phased(
            num_templates=num_templates,
            num_query_texts_per_template=num_query_texts_per_template,
            num_queries_per_query_text=None,
            phases=[
                PoissonArrivalPhase(
                    num_queries=num_total_queries,
                    poisson_lambda=poisson_lambda,
                )
            ],
            seed=seed,
            print_summary=print_summary,
            include_total_in_name=True,
        )

name_from_params(num_templates, num_query_texts_per_template, num_queries_per_query_text, phases, seed, num_total_queries=None) staticmethod

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@staticmethod
def name_from_params(
    num_templates: int,
    num_query_texts_per_template: int,
    num_queries_per_query_text: int | None,
    phases: list[PoissonArrivalPhase],
    seed: int,
    num_total_queries: int | None = None,
) -> str:
    global_poisson_lambda = (
        phases[0].poisson_lambda if len(phases) == 1 else None
    )
    name = "_".join(
        [
            "poisson",
            str(num_templates),
            str(num_query_texts_per_template),
            str(num_queries_per_query_text),
            str(global_poisson_lambda),
            str(seed),
        ]
    )

    if num_total_queries is not None:
        name += f"_{num_total_queries}"

    if len(phases) > 1:
        phase_str = "_".join(phase.name for phase in phases)
        name += f"_phased_{phase_str}"

    return name

make_phased_profile(num_queries_per_phase, poisson_lambda_per_phase) staticmethod

Create a multi-phase workload with the given number of queries and Poisson lambdas. The i-th phase will have num_queries_per_phase[i] queries and a Poisson lambda of poisson_lambda_per_phase[i]. The lambda governs interarrival times before each query in a phase.

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@staticmethod
def make_phased_profile(
    num_queries_per_phase: list[int],
    poisson_lambda_per_phase: list[float],
) -> list[PoissonArrivalPhase]:
    """
    Create a multi-phase workload with the given number of queries and
    Poisson lambdas. The i-th phase will have num_queries_per_phase[i]
    queries and a Poisson lambda of poisson_lambda_per_phase[i]. The lambda
    governs interarrival times *before* each query in a phase.
    """
    return [
        PoissonArrivalPhase(num_queries=num_q, poisson_lambda=lam)
        for num_q, lam in zip(
            num_queries_per_phase, poisson_lambda_per_phase
        )
    ]

make_bursty_profile(total_num_queries, num_lull_burst_cycles, lull_poisson_lambda, burst_poisson_lambda, fraction_of_queries_in_bursts) staticmethod

Generate a phased workload alternating between "lull" phases with low arrival rate and "burst" phases with high arrival rate.

Parameters

name: Name for the profile. total_num_queries: Total number of queries across all phases. num_lull_burst_cycles: Number of alternating "lull" and "burst" cycles. lull_poisson_lambda: Poisson lambda for "lull" phases. burst_poisson_lambda: Poisson lambda for "burst" phases. fraction_of_queries_in_bursts: Fraction of queries that arriva during "burst" phases.

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@staticmethod
def make_bursty_profile(
    total_num_queries: int,
    num_lull_burst_cycles: int,
    lull_poisson_lambda: float,
    burst_poisson_lambda: float,
    fraction_of_queries_in_bursts: float,
) -> list[PoissonArrivalPhase]:
    """
    Generate a phased workload alternating between "lull" phases with low
    arrival rate and "burst" phases with high arrival rate.

    Parameters
    ----------
    name: Name for the profile.
    total_num_queries: Total number of queries across all phases.
    num_lull_burst_cycles: Number of alternating "lull" and "burst" cycles.
    lull_poisson_lambda: Poisson lambda for "lull" phases.
    burst_poisson_lambda: Poisson lambda for "burst" phases.
    fraction_of_queries_in_bursts: Fraction of queries that arriva during
    "burst" phases.
    """

    num_queries_per_burst = int(
        total_num_queries
        * fraction_of_queries_in_bursts
        / num_lull_burst_cycles
    )
    pessimistic_num_queries_per_lull = (
        total_num_queries - num_queries_per_burst * num_lull_burst_cycles
    ) // num_lull_burst_cycles

    remaining_queries = total_num_queries
    phases = []
    while remaining_queries > 0:
        # Add lull phase
        num_queries_this_lull = min(
            pessimistic_num_queries_per_lull, remaining_queries
        )
        remaining_queries -= num_queries_this_lull
        phases.append(
            PoissonArrivalPhase(
                num_queries=num_queries_this_lull,
                poisson_lambda=lull_poisson_lambda,
            )
        )

        # Add burst phase
        num_queries_this_burst = min(
            num_queries_per_burst, remaining_queries
        )
        remaining_queries -= num_queries_this_burst
        phases.append(
            PoissonArrivalPhase(
                num_queries=num_queries_this_burst,
                poisson_lambda=burst_poisson_lambda,
            )
        )

    return phases

create_poisson_workload(num_templates, num_query_texts_per_template, num_queries_per_query_text, poisson_lambda, seed, print_summary=True) staticmethod

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@staticmethod
def create_poisson_workload(
    num_templates: int,
    num_query_texts_per_template: int,
    num_queries_per_query_text: int,
    poisson_lambda: float,
    seed: int,
    print_summary: bool = True,
) -> Workload:
    return PoissonWorkloadCreator.create_poisson_workload_phased(
        num_templates=num_templates,
        num_query_texts_per_template=num_query_texts_per_template,
        num_queries_per_query_text=num_queries_per_query_text,
        phases=[
            PoissonArrivalPhase(
                num_queries=(
                    num_templates
                    * num_query_texts_per_template
                    * num_queries_per_query_text
                ),
                poisson_lambda=poisson_lambda,
            )
        ],
        seed=seed,
        print_summary=print_summary,
        include_total_in_name=False,
    )

create_poisson_workload_phased(num_templates, num_query_texts_per_template, num_queries_per_query_text, phases, seed, print_summary=True, include_total_in_name=False) staticmethod

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@staticmethod
def create_poisson_workload_phased(
    num_templates: int,
    num_query_texts_per_template: int,
    num_queries_per_query_text: Optional[int],
    phases: list[PoissonArrivalPhase],
    seed: int,
    print_summary: bool = True,
    include_total_in_name: bool = False,
) -> Workload:
    num_total_queries = sum(phase.num_queries for phase in phases)
    rng = np.random.default_rng(seed)
    workload_name = PoissonWorkloadCreator.name_from_params(
        num_templates=num_templates,
        num_query_texts_per_template=num_query_texts_per_template,
        num_queries_per_query_text=num_queries_per_query_text,
        phases=phases,
        seed=seed,
        num_total_queries=(
            None if not include_total_in_name else num_total_queries
        ),
    )

    # Determine which templates to use
    all_templates = list(range(1, 100))
    rng.shuffle(all_templates)
    selected_templates = all_templates[:num_templates]

    # Create the queries in sorted order and then shuffle.
    if num_queries_per_query_text is None:
        num_queries_per_query_text = num_total_queries
    query_text_ids = []
    for template_idx in sorted(selected_templates):
        for query_text_idx in range(1, num_query_texts_per_template + 1):
            for _ in range(1, num_queries_per_query_text + 1):
                query_text_ids.append(
                    f"ext_tpcds1000#{template_idx:03d}#{query_text_idx:03d}"
                )
    rng.shuffle(query_text_ids)
    query_text_ids = query_text_ids[:num_total_queries]

    return PoissonWorkloadCreator._package_queries_into_workload(
        query_text_ids=query_text_ids,
        phases=phases,
        workload_name=workload_name,
        print_summary=print_summary,
        rng=rng,
    )

_package_queries_into_workload(query_text_ids, phases, workload_name, print_summary, rng) staticmethod

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
@staticmethod
def _package_queries_into_workload(
    query_text_ids: list[str],
    phases: list[PoissonArrivalPhase],
    workload_name: str,
    print_summary: bool,
    rng: np.random.Generator,
) -> Workload:

    # Create submission times using a Poisson process for each phase.
    reference_rel_time_s = 0.0
    rel_arrival_times_s = phases[0].generate_rel_arrival_times_s(
        rng=rng,
        reference_rel_time_s=reference_rel_time_s,
        include_gap_pre_first=False,
    )
    for phase in phases[1:]:
        reference_rel_time_s = rel_arrival_times_s[-1]
        phase_rel_arrival_times_s = phase.generate_rel_arrival_times_s(
            rng=rng,
            reference_rel_time_s=reference_rel_time_s,
            include_gap_pre_first=True,
        )
        rel_arrival_times_s.extend(phase_rel_arrival_times_s)

    # Package into workload.
    gen_start_time = datetime.fromisoformat("2026-01-01T00:00:00")
    records = []
    for i, (query_text_id, rel_arrival_time_s) in enumerate(
        zip(query_text_ids, rel_arrival_times_s)
    ):
        record = {
            "query_id": f"query_{i}",
            "abs_start_time": gen_start_time
            + timedelta(seconds=rel_arrival_time_s),
            "query_text_id": query_text_id,
            "repetition_id": f"query_{i}",
        }
        records.append(record)
    df = pd.DataFrame(records)
    out_path = pu.get_workloads_dir() / f"{workload_name}.parquet"
    df.to_parquet(out_path)

    # Print a nice summary
    workload_config = WorkloadConfig(
        workload_name=workload_name,
    )
    workload = Workload(workload_config)
    if print_summary:
        workload.print_summary()

    return workload

create_poisson_workload_with_n_queries(num_templates, num_query_texts_per_template, num_total_queries, poisson_lambda, seed, print_summary=True) staticmethod

Source code in src/autoslo/workload_definition/poisson_workload_creator.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@staticmethod
def create_poisson_workload_with_n_queries(
    num_templates: int,
    num_query_texts_per_template: int,
    num_total_queries: int,
    poisson_lambda: float,
    seed: int,
    print_summary: bool = True,
) -> Workload:
    return PoissonWorkloadCreator.create_poisson_workload_phased(
        num_templates=num_templates,
        num_query_texts_per_template=num_query_texts_per_template,
        num_queries_per_query_text=None,
        phases=[
            PoissonArrivalPhase(
                num_queries=num_total_queries,
                poisson_lambda=poisson_lambda,
            )
        ],
        seed=seed,
        print_summary=print_summary,
        include_total_in_name=True,
    )