Real-scale validation post-G0: configurable ingest cap + workers_500k metrics
Validated G0 substrate against the production workers_500k.parquet dataset (18 cols × 500,000 rows). Findings + one applied fix: Finding #1 (FIXED): ingestd's hardcoded 256 MiB cap rejected the 500K CSV (344 MiB) with 413. Cap fired correctly, no OOM. Extracted to [ingestd].max_ingest_bytes config field; default 256 MiB, override per deployment for known-large workloads. With cap bumped to 512 MiB, 500K ingest succeeds in 3.12s with ingestd peak RSS 209 MiB. Finding #2 (deferred): ingestd doesn't release memory between ingests. Go runtime conservative; long-running daemon, fine. Finding #3: DuckDB-via-httpfs is healthy at 500K. GROUP BY 45ms, count(*) 24ms, AVG 47ms, schema introspection 25ms. Sub-linear scaling vs 100K — the s3:// read path is not a bottleneck. Finding #4: ADR-010 type inference correctly handled real staffing data. worker_id → BIGINT, numeric scores → DOUBLE, multi-line resume_text → VARCHAR. 1000-row sample sufficient. Finding #5: Go's encoding/csv handles RFC 4180 quoted-comma fields and multi-line quoted text without LazyQuotes — confirming the D4 scrum's dismissal of Qwen's BLOCK on this point. Net: substrate handles production-scale data with one config knob. No correctness issues, no OOMs, no silent type errors. All 6 G0 smokes still PASS after the cap-config change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
+9
-2
@@ -29,7 +29,7 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
maxIngestBytes = 256 << 20 // matches storaged PUT cap
|
||||
defaultMaxIngestBytes = 256 << 20 // floor; configurable via [ingestd].max_ingest_bytes
|
||||
// Per scrum C-DRIFT (Opus WARN): parquet keys are content-addressed
|
||||
// by schema fingerprint — datasets/<name>/<fp_hex>.parquet — so a
|
||||
// schema-drift ingest writes to a different key from the live one.
|
||||
@@ -53,10 +53,16 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
maxBytes := cfg.Ingestd.MaxIngestBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = defaultMaxIngestBytes
|
||||
}
|
||||
|
||||
h := &handlers{
|
||||
storagedURL: strings.TrimRight(cfg.Ingestd.StoragedURL, "/"),
|
||||
catalogd: catalogclient.New(cfg.Ingestd.CatalogdURL),
|
||||
hc: &http.Client{Timeout: 5 * time.Minute},
|
||||
maxBytes: maxBytes,
|
||||
}
|
||||
|
||||
if err := shared.Run("ingestd", cfg.Ingestd.Bind, h.register); err != nil {
|
||||
@@ -69,6 +75,7 @@ type handlers struct {
|
||||
storagedURL string
|
||||
catalogd *catalogclient.Client
|
||||
hc *http.Client
|
||||
maxBytes int64
|
||||
}
|
||||
|
||||
func (h *handlers) register(r chi.Router) {
|
||||
@@ -100,7 +107,7 @@ func (h *handlers) handleIngest(w http.ResponseWriter, r *http.Request) {
|
||||
// be re-readable — the schema infer pre-reads N rows, and the
|
||||
// pqarrow writer needs all rows including those samples. We
|
||||
// take the simple path: the whole upload in memory, capped.
|
||||
r.Body = http.MaxBytesReader(w, r.Body, maxIngestBytes)
|
||||
r.Body = http.MaxBytesReader(w, r.Body, h.maxBytes)
|
||||
if err := r.ParseMultipartForm(64 << 20); err != nil {
|
||||
var maxErr *http.MaxBytesError
|
||||
if errors.As(err, &maxErr) || strings.Contains(err.Error(), "http: request body too large") {
|
||||
|
||||
Reference in New Issue
Block a user