Task 5 Report — Governed Polars cross-source execution and artifact handoff

Date: 2026-07-23 Branch: codex/data-rule-execution-m3a-m5 Implementation commits: 69d83ed, f4d798b, 8bffdb6

8bffdb6 is the final hash of the second remediation commit. It replaces the intermediate a17bcd0 after restoring the historical M3A acceptance document byte-for-byte; no production or test behavior changed in that amend.

Lifecycle boundary: physical Polars plans are created as compiled; Task 5 does not add a production promotion API. The real integration inserts a published fixture only to exercise the existing Runner publication gate.

Final outcome

Task 5 now provides a fail-closed PostgreSQL/MySQL → Parquet/MinIO → isolated Polars → cataloged Parquet path through the production repository, resolver, executor, HTTP token, and durable task-ledger combination.

The final implementation includes:

closed JSON Polars plans and shared compile/runtime semantic validation;
typed cast targets carrying nullability, Decimal precision/scale, and timestamptz timezone;
real rule-timezone semantics for date(), timestamp(), timestamp cast, and timestamptz cast;
server-owned Parquet refs, digest validation, TTL enforcement, and bounded metadata;
streamed MinIO staging to temporary files and trusted PyArrow footer preflight before Polars scans;
all allocation-heavy Polars scans, expression evaluation, regex, join, group/aggregate, counts, and output write in an isolated spawn child;
parent RSS monitoring, child OS RLIMIT_AS, bounded execution time, and fail-closed worker errors;
immutable catalog handoff keyed by (correlation_id, binding_id, artifact_kind);
production cleanup that removes the expired object and its catalog row;
minimal public artifact results without full schema fields;
standards-valid CycloneDX 1.5 runtime SBOM with deterministic offline schema validation.

The bound plan contains canonical RuleVersion, SchemaSnapshot, and DatasetBinding hashes plus dataops-polars-1.42.1 provenance. It contains no Python source, pickle, callable, module, caller-supplied path, arbitrary URL, secret, or serialized Polars internal plan.

Second-remediation RED/GREEN evidence

The second review was implemented as fail-first slices. Representative RED evidence captured during the work:

Immutable handoff tests failed before the catalog uniqueness changed from digest-based uniqueness to the correlation/binding/kind invariant and before retry cleanup/reuse existed.
Public-metadata tests failed while MinIO metadata still carried the base64 full schema contract and public results still exposed schema_fields.
Cleanup tests failed before a bounded production catalog cleanup entrypoint deleted both the MinIO object and the exact PostgreSQL row.
Compression-bomb and tempfile-staging tests failed before PyArrow footer row-group preflight and disk streaming existed.
Worker isolation tests failed before app/runner/polars_worker.py existed; the first adapter reconstruction run then exposed one schema-order failure (1 failed) because canonical fields are name-sorted while Parquet retains physical column order. Validation was corrected to compare the closed field set and exact per-field types.
Typed-cast RED: test_polars_cast_operations_carry_complete_typed_targets failed (1 failed) with missing target_field.
Timezone RED: test_polars_expression_date_and_timestamp_use_plan_timezone failed (1 failed) because timestamp() produced timezone-aware output for a timezone-free timestamp contract.
The final full-suite pre-close run found one stale test monkeypatch (1 failed, 553 passed) after in-memory BytesIO staging was removed; the regression check was changed to assert the stage implementation contains no BytesIO.
The first Ruff closeout found two style-only failures (unused io, nested with); both were fixed before final verification.

Second-remediation GREEN:

Focused compiler/adapter/worker/artifact/bootstrap/schema/SBOM command: 57 passed in 4.17s.
Final artifact/worker/runtime/SBOM recheck after the last portability and timestamp-cast changes: 30 passed in 3.92s.
Final real cross-source + HTTP/ledger integration: 1 passed in 2.03s.
Final full suite: 554 passed, 26 skipped, 59 subtests passed in 8.39s.
Ruff across every changed Python file: All checks passed!.
git diff --check: passed.
docker compose -f deploy/docker/docker-compose.yml config -q: passed.

Immutable artifact handoff, TTL, and cleanup

Migration 20260723_140 now enforces:

UNIQUE (correlation_id, binding_id, artifact_kind)

Its downgrade raises RuntimeError because removing the catalog would violate immutable runtime handoff evidence.

PostgresArtifactResolver.register implements the retry invariant:

first writer inserts and returns the new catalog artifact;
same triple + same digest reuses the existing catalog ref and deletes the newly generated duplicate object;
same triple + different digest fails closed and deletes the new object;
existing catalog metadata is rechecked against the retained store object.

Resolution requires the exact artifact kind, current binding hash, matching correlation prefix, valid TTL, digest, row count, and schema hash. The integration asserts one input, one lookup, and one output row/object after repeated deterministic execution.

cleanup-rule-artifacts --limit uses a bounded FOR UPDATE SKIP LOCKED selection, removes each expired store object, and deletes its exact catalog row in the same database transaction scope. Unit tests cover the CLI and object+row cleanup behavior.

The local database had an earlier applied copy of migration 140. Before real integration, its artifact table was verified empty and its old unique constraint was aligned to the final triple invariant. Migration source tests verify the clean-install definition.

Resource enforcement and worker isolation

Pre-allocation artifact checks

ArtifactStore.stage:

validates the server-owned ref, object stat, content type, compressed size, metadata digest, row count, schema digest, and TTL;
streams MinIO response chunks to a NamedTemporaryFile while hashing;
validates the downloaded digest;
reads the Parquet footer with pyarrow.parquet.ParquetFile;
rejects excessive footer row count or summed row-group uncompressed size;
checks only the lazy physical schema in the main process;
yields the path and always removes the temporary file.

Full schema contracts are no longer base64-encoded into MinIO metadata. MinIO metadata contains only bounded digest/count/expiry/size fields, with a 2,048-character aggregate ceiling. PostgreSQL remains the canonical store for the complete schema contract.

Isolated execution

The adapter stages paths but does not collect input frames. A spawn child:

warms Polars before establishing its runtime baseline;
applies RLIMIT_AS relative to baseline virtual memory;
validates the plan again;
scans input and lookup Parquet;
executes every allowlisted operation;
validates output row/type/nullability contracts;
writes the output Parquet path.

The parent independently watches incremental RSS with psutil, kills the child on limit breach, enforces a hard timeout, and returns only bounded metrics. Worker PID is deliberately not exposed in the public result.

Tests prove:

worker PID differs from the Runner PID;
a 128 MiB allocation fails under an 8 MiB allowance;
a highly compressed Parquet payload is rejected from footer bounds before scan exposure;
regex expansion, high-cardinality group/aggregate, and wide lookup join peak allocations fail inside the isolated worker.

Portability boundary: spawn and psutil are portable, but the OS hard limit requires Unix resource.RLIMIT_AS. Linux/macOS are supported targets. On a platform without RLIMIT_AS, the worker fails closed instead of silently running without an OS memory boundary.

Typed casts and timezone semantics

Each compiled cast operation now contains an exact target_field. The shared validator rejects a missing field, an unknown/extra shape, a target name/type mismatch, incomplete Decimal precision/scale, or missing timestamptz timezone. The worker no longer falls back to inferring cast details from output fields.

Executable tests verify:

string → Decimal(12, 2);
string → Datetime(time_zone="Asia/Shanghai");
offset timestamp → rule-local timezone-free timestamp;
date(offset_value) converts to the rule timezone before selecting the calendar date;
timestamp(offset_value) converts to the rule timezone and removes timezone metadata to match the platform's timezone-free timestamp type.

Real production combination

The integration discovers runtime values from deploy/docker/docker-compose.yml; credentials are not copied into source.

source PostgreSQL: live customer input;
source MySQL: live segment lookup;
MinIO: input, lookup, and output Parquet objects;
platform PostgreSQL: snapshots, bindings, published plan, artifact catalog, and durable task ledger;
Polars spawn worker: normalize → lookup join → assert → deduplicate;
Flask /v1/tasks/execute: signed short-lived task token;
PostgresTaskLedger: single-use claim and committed success record.

Evidence:

4 input rows → 2 output rows;
1 assertion reject;
1 deduplicated row;
exact enriched values verified after reread;
direct repeat reuses the stable output ref;
conflicting output digest fails and its new object is absent;
HTTP result returns the same stable output ref;
replaying the same token returns HTTP 409;
durable ledger records success / committed;
public result contains no schema_fields;
exactly three catalog rows and three correlation-scoped MinIO objects remain before cleanup.

The test finally removes only its exact correlation prefix, source tables, ledger record, catalog entries, plan/binding/deployment/rule rows, and schema snapshots, and asserts the MinIO prefix is empty.

CycloneDX 1.5 SBOM

docs/security/data-rule-runtime-sbom.json now uses standard CycloneDX fields:

licenses;
externalReferences;
properties;
component bom-ref and dependency relationships.

Pinned runtime components:

polars==1.42.1 — MIT;
polars-runtime-32==1.42.1 — MIT;
pyarrow==21.0.0 — Apache-2.0;
psutil==5.9.8 — BSD-3-Clause;
minio==7.2.10 — Apache-2.0.

The test validates the document with jsonschema against vendored official CycloneDX 1.5, SPDX, and JSF schemas. The official files' SHA-256 hashes are asserted before validation, so the check is deterministic and offline.

Key files in the final remediation

Production:

app/core/data_rules/compilers/polars.py
app/runner/artifacts.py
app/runner/polars_worker.py
app/runner/rule_polars.py
app/runner/bootstrap.py
migrations/versions/20260723_140_rule_run_artifacts.py
requirements.txt
docs/security/data-rule-runtime-sbom.json
docs/security/cyclonedx-1.5-schema/

Tests:

tests/core/data_rules/test_polars_compiler.py
tests/runner/test_artifacts.py
tests/runner/test_polars_worker.py
tests/runner/test_rule_polars.py
tests/runner/test_bootstrap.py
tests/integration/test_data_rule_polars_execution.py
tests/test_data_rule_runtime_sbom.py
tests/test_data_rule_schema.py

Remaining boundaries

Task 7 still owns test-evidence recording and plan publication. RulePlanExecutor continues to refuse compiled plans in the published runtime path.
quality_check remains fail-closed through the existing adapter; Task 5 does not claim an artifact-backed quality-check implementation.
ArtifactStore.read remains a compatibility/helper API that can return a bounded in-process frame. The production Polars adapter does not call it; it uses staged files plus the isolated worker.
Windows requires a separate Job Object implementation before Polars batch execution can be enabled there; the current implementation fails closed when RLIMIT_AS is unavailable.

task-5-report.md 12 KB 文件历史 原始文件