fix: release v0.7.1 with clearer initialization reports

Author: Gurov Oleg

INITIALIZATION.md

@@ -22,6 +22,15 @@ flight, regardless of dataset size. A batch commits independently; the loader
 does not create one transaction for the entire dataset. Progress logs report
 committed rows and batches per task.
 
+The report shows a compact stage summary, with table conversions and vacuum
+operations grouped together. Expand **data loading by task** for PostgreSQL
+INSERT/COPY row counts, completed batches and cumulative worker time. Inserted
+rows are measured before post-load updates or deduplication; they are not final
+table cardinalities. Worker times overlap during parallel loading and must not
+be added to elapsed stage times. DDL and maintenance details are collapsed and
+do not display artificial zero row counters. A task targeting a partitioned
+parent can populate several partitions without separate partition-level jobs.
+
 Before each iteration, `--reset-mode database` (default) recreates the disposable
 database. `--reset-mode schema` resets the schemas in `LoadPlan.schemas` inside
 a pre-created dedicated database; it requires fast initialization and `--init-fsync keep`.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pg-perf-bench"
-version = "0.7.0"
+version = "0.7.1"
 description = "Reproducible PostgreSQL TPS benchmarks with environment evidence and report comparison"
 readme = "README.md"
 requires-python = ">=3.10"

src/pg_perf_bench/__init__.py

@@ -1,3 +1,3 @@
 """PostgreSQL performance benchmark and evidence collector."""
 
-__version__ = '0.7.0'
+__version__ = '0.7.1'

src/pg_perf_bench/initialization.py

@@ -418,37 +418,155 @@ async def statement(phase, sql):
 
 
 def build_initialization_section(benchmark_runs, options):
+    """Separate measured loading counters from DDL and maintenance timings."""
+    labels = {
+        'schema': 'Create schema',
+        'unlogged': 'Set tables UNLOGGED',
+        'prepare': 'Prepare generator',
+        'data': 'Load data',
+        'after_data': 'Post-load processing',
+        'logged': 'Set tables LOGGED',
+        'indexes': 'Build indexes',
+        'constraints': 'Apply constraints',
+        'finalize': 'Finalize dataset',
+        'vacuum': 'Vacuum and freeze',
+        'analyze': 'Analyze relations',
+        'restore_settings_and_sync': 'Restore durability settings',
+    }
     reports = {}
     for run in benchmark_runs:
         evidence = run.get('initialization')
         if evidence is None:
             continue
         index = run['iteration']['index']
-        rows = []
+        groups = {}
+        loads = []
+        operations = []
         for phase in evidence['phases']:
-            tasks = phase.get('tasks', [])
-            rows.append(
-                [
-                    phase['name'],
-                    round(phase['elapsed_seconds'], 3),
-                    sum(task['rows'] for task in tasks),
-                    sum(task['batches'] for task in tasks),
-                ]
-            )
+            kind = phase['name'].split(':', 1)[0]
+            group = groups.setdefault(kind, {'seconds': 0.0, 'phases': [], 'tasks': []})
+            group['seconds'] += phase['elapsed_seconds']
+            group['phases'].append(phase)
+            group['tasks'].extend(phase.get('tasks', []))
+            if kind == 'data':
+                for task in phase.get('tasks', []):
+                    loads.append(
+                        [
+                            task['name'],
+                            round(task['worker_seconds'], 3),
+                            f'Rows inserted: {task["rows"]:,}; batches: {task["batches"]:,}',
+                        ]
+                    )
+            elif kind in ('indexes', 'constraints'):
+                # Aggregate timings already appear in the summary. Listing them
+                # again beside individual tasks invites double-counting.
+                action = 'Build index' if kind == 'indexes' else 'Apply constraint'
+                for task in phase.get('tasks', []):
+                    operations.append(
+                        [
+                            action,
+                            round(task['worker_seconds'], 3),
+                            f'{task["name"]}; completed executions: {task["batches"]:,}. '
+                            'Time includes execution and commit.',
+                        ]
+                    )
+            else:
+                relation = phase['name'].partition(':')[2]
+                description = {
+                    'schema': 'Create profile schemas and supporting objects: '
+                    + ', '.join(evidence.get('schemas', [])),
+                    'unlogged': f'Set {relation} to UNLOGGED before bulk loading.',
+                    'logged': (
+                        f'Set {relation} to LOGGED before bulk loading.'
+                        if options['table_mode'] == 'logged'
+                        else f'Restore WAL logging for {relation} after loading.'
+                    ),
+                    'vacuum': f'VACUUM (FREEZE, ANALYZE) {relation}: freeze rows and '
+                    'refresh visibility and planner statistics.',
+                    'prepare': 'Execute profile setup SQL for data generation.',
+                    'after_data': 'Execute profile post-load SQL, including declared '
+                    'data corrections; affected rows are not recorded.',
+                    'finalize': 'Execute profile finalization SQL, such as materialized '
+                    'view refresh and workload bounds setup.',
+                    'analyze': 'ANALYZE profile relations, including partitioned parents.',
+                    'restore_settings_and_sync': 'Restore pre-load durability settings; '
+                    'checkpoint and sync only if fsync changed.',
+                }.get(kind, f'Execute custom initialization step {phase["name"]}.')
+                operations.append(
+                    [
+                        labels.get(kind, kind),
+                        round(phase['elapsed_seconds'], 3),
+                        description,
+                    ]
+                )
+
+        summary = []
+        for kind, group in groups.items():
+            tasks = group['tasks']
+            if kind == 'data':
+                inserted = sum(task['rows'] for task in tasks)
+                batches = sum(task['batches'] for task in tasks)
+                work = f'{inserted:,} rows inserted; {len(tasks):,} load tasks; {batches:,} batches'
+            elif kind in ('indexes', 'constraints'):
+                work = f'Tasks: {len(tasks):,}; executions: {sum(t["batches"] for t in tasks):,}'
+            elif kind in ('unlogged', 'logged', 'vacuum'):
+                work = f'{len(group["phases"]):,} relations'
+            else:
+                work = {
+                    'schema': 'Create tables and supporting objects',
+                    'prepare': 'Install generator helpers',
+                    'after_data': 'Apply profile post-processing; row counts not collected',
+                    'finalize': 'Run profile finalization SQL',
+                    'analyze': 'Refresh planner statistics',
+                    'restore_settings_and_sync': 'Restore settings; sync storage if fsync changed',
+                }.get(kind, 'Execute profile SQL; row counts not collected')
+            summary.append([labels.get(kind, kind), round(group['seconds'], 3), work])
         barrier = evidence['replication_barrier']
-        rows.append(['replica_replay_barrier', round(barrier['elapsed_seconds'], 3), None, None])
+        summary.append(
+            [
+                'Wait for replica replay',
+                round(barrier['elapsed_seconds'], 3),
+                f'{barrier["replicas"]} replicas caught up to {barrier["target_lsn"]}',
+            ]
+        )
         reports[f'iteration_{index}'] = {
-            'header': f'Iteration {index}: initialization phases',
-            'description': f'Workers: {options["workers"]}; batch rows: {options["batch_rows"]}; '
-            f'tables during load: {options["table_mode"]}; fsync during load: {options["fsync"]}; '
-            f'loader synchronous_commit: {options["synchronous_commit"]}. '
-            f'Final fsync: {evidence["fsync_after"]}; replicas caught up: {barrier["replicas"]}; '
-            f'replay target: {barrier["target_lsn"]}. '
-            'Rows and batches describe data jobs; elapsed times are wall-clock seconds.',
+            'header': f'Iteration {index}: initialization summary',
+            'description': f'Up to {options["workers"]} load workers; '
+            f'batch limit: {options["batch_rows"]:,} rows. '
+            f'Tables during load: {options["table_mode"]}; fsync policy: {options["fsync"]}; '
+            f'synchronous_commit policy: {options["synchronous_commit"]}; '
+            f'fsync after initialization: {evidence["fsync_after"]}. '
+            'Times are measured elapsed seconds per stage; per-relation operations are grouped. '
+            'Parallel worker times are not added to stage elapsed time.',
             'item_type': 'table',
             'state': 'expanded',
             'collection_status': 'ok',
-            'theader': ['phase', 'elapsed_seconds', 'rows', 'jobs'],
-            'data': rows,
+            'theader': ['Stage', 'Elapsed (s)', 'Work performed'],
+            'data': summary,
+        }
+        reports[f'iteration_{index}_loading'] = {
+            'header': f'Iteration {index}: data loading by task',
+            'description': 'Inserted rows come from PostgreSQL INSERT/COPY completion counts. '
+            'They are not final table sizes: post-load processing may update or delete rows. '
+            'Batches count successful loader executions, including single-statement tasks. '
+            'Worker time is cumulative execution time including commit, not wall-clock time. '
+            'Task names identify loader jobs, not necessarily physical tables; inserting into '
+            'a partitioned parent routes rows to its partitions without separate loader jobs.',
+            'item_type': 'table',
+            'state': 'collapsed',
+            'collection_status': 'ok',
+            'theader': ['Load task', 'Worker time (s)', 'Details'],
+            'data': loads,
+        }
+        reports[f'iteration_{index}_operations'] = {
+            'header': f'Iteration {index}: SQL and maintenance details',
+            'description': 'Individual operations, without the stage totals already shown above. '
+            'Each row describes the work actually timed. Index tasks can run in parallel, '
+            'so adding these times does not give total initialization elapsed time.',
+            'item_type': 'table',
+            'state': 'collapsed',
+            'collection_status': 'ok',
+            'theader': ['Operation', 'Time (s)', 'Description'],
+            'data': operations,
         }
     return {'header': 'Database initialization', 'state': 'expanded', 'reports': reports}

tests/test_initialization.py

@@ -301,3 +301,95 @@ def test_recovery_rejects_sql_failover_when_host_transport_still_targets_old_pri
     )
     with pytest.raises(ConfigurationError, match='target different servers'):
         asyncio.run(guard._identity())
+
+
+def test_initialization_report_separates_row_counts_from_maintenance_times():
+    from pg_perf_bench.initialization import build_initialization_section
+
+    phases = [
+        {'name': 'schema', 'elapsed_seconds': 0.1},
+        {'name': 'unlogged:first', 'elapsed_seconds': 1},
+        {'name': 'unlogged:second', 'elapsed_seconds': 2},
+        {
+            'name': 'data',
+            'elapsed_seconds': 5,
+            'tasks': [
+                {'name': 'first', 'rows': 7, 'batches': 2, 'worker_seconds': 4},
+                {'name': 'payment', 'rows': 3, 'batches': 1, 'worker_seconds': 9},
+            ],
+        },
+        {'name': 'after_data', 'elapsed_seconds': 0.5},
+        {
+            'name': 'indexes',
+            'elapsed_seconds': 2,
+            'tasks': [{'name': 'first_pk', 'rows': 0, 'batches': 1, 'worker_seconds': 2}],
+        },
+    ]
+    runs = [
+        {
+            'iteration': {'index': 1},
+            'initialization': {
+                'phases': phases,
+                'fsync_after': 'on',
+                'replication_barrier': {'elapsed_seconds': 0.2, 'replicas': 2, 'target_lsn': '0/A'},
+            },
+        }
+    ]
+    options = dict(
+        workers=4, batch_rows=100000, table_mode='unlogged', fsync='keep', synchronous_commit='keep'
+    )
+    reports = build_initialization_section(runs, options)['reports']
+    summary = reports['iteration_1']
+    assert summary['theader'] == ['Stage', 'Elapsed (s)', 'Work performed']
+    assert ['Set tables UNLOGGED', 3, '2 relations'] in summary['data']
+    assert ['Load data', 5, '10 rows inserted; 2 load tasks; 3 batches'] in summary['data']
+    assert ['Build indexes', 2, 'Tasks: 1; executions: 1'] in summary['data']
+    loading = reports['iteration_1_loading']
+    assert loading['data'] == [
+        ['first', 4, 'Rows inserted: 7; batches: 2'],
+        ['payment', 9, 'Rows inserted: 3; batches: 1'],
+    ]
+    assert loading['theader'] == ['Load task', 'Worker time (s)', 'Details']
+    assert 'not final table sizes' in loading['description']
+    assert 'partitioned parent' in loading['description']
+    operations = reports['iteration_1_operations']
+    assert operations['theader'] == ['Operation', 'Time (s)', 'Description']
+    assert len(operations['data']) == 5  # No duplicate aggregate index timing.
+    assert any(row[0] == 'Build index' and 'first_pk' in row[2] for row in operations['data'])
+    assert not any(row[0] == 'Build indexes' for row in operations['data'])
+    assert any('first to UNLOGGED' in row[2] for row in operations['data'])
+    assert any('second to UNLOGGED' in row[2] for row in operations['data'])
+    assert not any('Timed operation;' in row[2] for row in operations['data'])
+    assert reports['iteration_1_operations']['state'] == 'collapsed'
+    assert all(len(row) == 3 for row in reports['iteration_1_operations']['data'])
+    assert build_initialization_section([{'iteration': {'index': 2}}], options)['reports'] == {}
+
+
+@pytest.mark.parametrize(
+    ('table_mode', 'description'),
+    [
+        ('logged', 'Set payment to LOGGED before bulk loading.'),
+        ('unlogged', 'Restore WAL logging for payment after loading.'),
+    ],
+)
+def test_initialization_report_logged_description_matches_load_mode(table_mode, description):
+    from pg_perf_bench.initialization import build_initialization_section
+
+    conversion = {'name': 'logged:payment', 'elapsed_seconds': 0.5}
+    data = {'name': 'data', 'elapsed_seconds': 1, 'tasks': []}
+    phases = [conversion, data] if table_mode == 'logged' else [data, conversion]
+    run = {
+        'iteration': {'index': 1},
+        'initialization': {
+            'phases': phases,
+            'fsync_after': 'on',
+            'replication_barrier': {'elapsed_seconds': 0, 'replicas': 0, 'target_lsn': '0/A'},
+        },
+    }
+    options = dict(
+        workers=4, batch_rows=100000, table_mode=table_mode, fsync='keep', synchronous_commit='keep'
+    )
+
+    reports = build_initialization_section([run], options)['reports']
+
+    assert reports['iteration_1_operations']['data'] == [['Set tables LOGGED', 0.5, description]]