fix: DuckDB and server-to-server architecture#145
Merged
Conversation
Adds the byte-addressing seam to Core's cross-cutting capability surface: ISupportsByteLocation resolves where an item's bytes live — a ByteLocation closed sum over LocalFile / RemoteUri-with-access- handoff — without loading rows, so a native-reading consumer (an embedded engine, a bulk copier) can be pointed straight at storage. - ComposedStorageAdapter implements the capability by delegating to its medium; a non-addressable medium fails as a typed value, never a silent fallback that materialises rows. - IItem<T>.LocateBytes() demands the capability at wire-up, unwrapping a .Constrain() wrapper exactly as AsStream() does, and throws a loud error naming the item when the adapter cannot honor it. - FileStorageMedium answers with the absolute file path; S3StorageMedium routes through the IS3Gateway seam — the credential owner — so the production gateway mints the s3:// URI plus endpoint/region/credential entries per call and the local stub hands back the backing file itself. - Ships the ISupportsByteLocationLaws kit (addressability honesty, address stability, write-target addressability) with subclasses for the file medium, the composed adapter, and S3 over the local stub, plus targeted tests for the incapable-adapter and constrained paths. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
Adds the Flowthru.Extensions.DuckDB extension: a wide-transform step wired between ordinary Parquet catalog items whose entire body is SQL executed inside the embedded DuckDB engine. Rows never enter the CLR — inputs bind as read_parquet views via the byte-location capability, the result is verified against the output item's declared schema (DESCRIBE, before any write), and written engine-side with COPY (FORMAT PARQUET). Extension surface: - AddDuckDbTransform intent verbs (1-input and multi-relation shapes) with DuckDbInputRelation bindings defaulting to item labels - IDuckDbEngine + InProcessDuckDbEngine, registered by UseDuckDb() with options bound from Flowthru:DuckDb - Engine registers as a cache-neutral (AffectsOutputs=false), capacity-constrained service dependency mirroring the Python worker; scheduler holds concurrent transforms to MaxConcurrentTransforms - Typed failures: RuntimeError.SchemaMismatch for result-schema disagreement, DuckDbRuntimeError (FTDDB40xx) for remote-URI endpoints and engine errors — all as FlowIO error values, nothing thrown Core (additive): - IStepNode.DeclaredUncacheableReason seam + StepUncacheableReason.DeclaredByStep case: engine transforms opt out of caching loudly (SQL text isn't in the cache identity yet), visible everywhere uncacheable reasons surface Parquet (additive fix): - ParquetAdapter's DTO fast path now defers to the schema-mirroring runtime DTO when file column types disagree with TRow (external writers mark all columns optional; the fast path was a hard definition-level mismatch on any externally-written file) Benchmark (category-gated, [Explicit]): 5M-row composite-key sort — LINQ OrderBy 8.87s / 3.38GiB managed allocations vs DuckDB 0.39s / 99.6KiB (23x, row-count-independent CLR allocations). Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
Schema-validated SQL per ADR-0024's v1 relational surface: every AddDuckDbTransform step is now checked before any step runs — empty in-memory tables are built from the declared input record schemas (named per the step's relation bindings), the transform SQL is DESCRIBEd against them (binding without executing), and the described result schema is verified against the declared output schema. Zero I/O against real data; failures name the step, the relation binding, and the offending column(s), and aggregate applicatively with all other pre-flight errors. Core (the ADR-0021 "symmetric follow-up"): - IFlowValidationHook gains MinimumDepth self-classification on the ValidationDepth ladder (default Shallow), mirroring IRegistrationValidationHook; PreFlightPipeline admits hermetic-classified flow hooks at Hermetic scope. - FlowthruService now merges DI-registered IFlowValidationHooks with builder-registered ones and filters by the run's depth — the plural DI surface extensions already use (Python) finally reaches the hosted pre-flight path. - ValidationDepth.Hermetic / PreFlightScope.Hermetic docs sharpened: the promise is "no external reach", with an explicit carve-out for process-local computation (embedded in-memory engine as a metadata-only type-checker) so the hermetic classification of this check stays honest rather than silently stretched. Extension (Flowthru.Extensions.DuckDB): - DuckDbSqlSchemaCheck: the shared hermetic check core, reusing the runtime's DuckDbTypeMap/DuckDbSchemaVerifier and SQL-text discipline (hoisted into DuckDbSql) so pre-flight binds byte-identical SQL. - DuckDbPreFlightError (FTDDB3001 preparation failure / FTDDB3002 result-schema mismatch / FTDDB3003 unmodellable input schema), wrapped through PreFlightError.External. - DuckDbTransformValidationHook (HookId "duckdb.sql-schema", MinimumDepth Hermetic), registered by UseDuckDb() via TryAddEnumerable. - Design-time surfaces: DuckDbTransformStep.Validate() now runs the real check (so FUnitContext.Validate(step) fails a unit test on a schema-breaking SQL edit), plus BuiltFlow.ValidateDuckDbTransforms() for whole-flow test assertions. The engine never calls Validate() on the run path — the happy path pays only the pre-flight check itself. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
Teach the engine transform to run S3-Parquet -> S3-Parquet entirely
in-engine, closing the FTDDB4001 gap for s3:// URIs:
- InProcessDuckDbEngine handles ByteLocation.RemoteUri for s3://
endpoints: loads httpfs (INSTALL on first use when downloads are
allowed), mints one temporary DuckDB secret per distinct object from
the gateway's access handoff, each SCOPEd to exactly that object's
URI so multi-credential inputs never bleed into each other, then
read_parquet('s3://...') / COPY ... TO 's3://...' with no CLR
buffering and no CLR rows.
- Credential material lives only in the transform's private in-memory
connection; engine error text is scrubbed of credential values
before surfacing (DuckDB errors echo the offending SQL).
- httpfs is not statically linked in DuckDB.NET.Data.Full: new
DuckDbEngineOptions.ExtensionDirectory (offline pre-provisioning)
and AllowExtensionDownload (default true; false makes a missing
httpfs the typed FTDDB4003 failure and disables DuckDB autoinstall
for the connection).
- Non-s3 remote schemes keep the typed FTDDB4001 error with an
honestly narrowed message naming the offending scheme.
- DuckDbBoundRelation/DuckDbTransformRequest now carry ByteLocation
endpoints instead of pre-collapsed local paths.
- README: S3 support section (gateway-minted, per-call,
connection-scoped credential flow), httpfs loading story, refreshed
limitations.
Tests: offline secret-SQL planning/redaction units, scheme rejection
(step + engine-defensive), typed FTDDB4003 failure mode, s3:read
conflict-key inheritance over S3-backed endpoints (ADR-0023), an
always-on offline tier over the file-backed S3 stub, and an
env-gated (TestCapabilities.AwsS3) integration tier covering S3->S3,
mixed local+S3 inputs, per-endpoint secret scoping, and the
zero-CLR-rows guarantee over S3 — verified green against a real MinIO.
Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
Engine-delegated transform steps are now first-class cacheable, with the query in the key, per ADR-0024's cache-identity decision. Core (general seam, no DuckDB coupling): - IStepNode.DeclaredCacheIdentity: a defaulted member (following the DeclaredUncacheableReason precedent) through which a step declares the output-affecting wire-up data missing from CodeVersion — an opaque, ordinal-stable token folded into the composite fingerprint wherever code version + input fingerprints combine (CachePlanBuilder phase 3 and CacheManifestStore's post-run twin). - ComposeStepFingerprint gains the declared segment; a null identity composes byte-identically to the pre-seam shape, so existing manifests stay valid for ordinary steps (no schema bump). - The post-run twin now also mirrors the DeclaredUncacheableReason opt-out, keeping its eligibility a true mirror of pre-flight. - DeclaredUncacheableReason/DeclaredByStep stay public surface; docs now route steps whose wire-up data CAN be fingerprinted to the new seam and reserve the opt-out for data that can't. DuckDB implements it (stopgap removed): - DeclaredCacheIdentity = SHA-256 of the exact SQL bytes (no normalization), the engine version, the relation-name -> item-label bindings (same SQL over reordered bindings reads different data), and the output-affecting write options (compression, row-group size both change the produced file's bytes). Engine tuning (memory, threads, concurrency) is deliberately excluded. - CodeVersion = the extension assembly version (the transform machinery's identity), satisfying cache-plan eligibility. - IDuckDbEngine.EngineVersion: the engine is the authority on its own version; InProcessDuckDbEngine probes DuckDBConnection.ServerVersion on a throwaway in-memory connection once per process, degrading to a per-process-unique sentinel (never a hit, never a throw) if the native library can't load. - The DeclaredByStep uncacheable stopgap from #135 is gone; engine steps no longer appear in uncacheable reporting. Tests: composite back-compat pin + declared-identity hit/stale/cascade at the core plan level; DuckDB identity axes (SQL bytes, engine version via fake, options, bindings, tuning exclusion) plus end-to-end hit, per-axis invalidation (engine bump via a version-spoofing wrapper over the real engine), and downstream-of-engine-step cascade through a real IFlowthruService with a file-backed manifest. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
Adds the medium-agnostic bulk-transfer surface from the transfer half of the engine-delegated-transforms design, scoped to the streaming rung: - ISupportsBulkExport / ISupportsBulkImport capability pair in Core's cross-cutting storage surface, so any two extensions can pair on provider + wire format without referencing each other, plus IItem.TryGetBulkExport/TryGetBulkImport discovery accessors. - ISupportsStreamingSink<TRow>, the write-side sibling of ISupportsStreamingView, so a target item can open the batch sink the streaming rung drives. - AddBulkTransfer(source, target, options): an intent verb emitting an on-DAG identity step between two endpoint items (the AddBulkLoad mechanics), inheriting scheduling, caching, pre-flight, and the conflict keys of both endpoints. - Pre-flight rung negotiation (zero-I/O, runs at every scope including Hermetic): probes both endpoints' capabilities, checks pair compatibility, selects a rung, and reports the selection - including the downgrade rationale - in the run's validation output. A pairing with no executable rung fails pre-flight with the new PreFlightError.BulkTransferRungUnavailable (FT3008). - BulkTransferOptions.RequireNative: an unavailable native path becomes a pre-flight error; always errors in this slice since the native rung's execution machinery lands in a later issue. - FUnitContext.NegotiateTransfer so step tests can assert the reported rung without building a flow. - StreamingItem now surfaces its origin's ServiceDependencies, so streaming views inherit conflict keys consistently. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
Flip the native rung from "known but unavailable" to executable. A homogeneous Postgres pair now negotiates BulkTransferRung.Native at pre-flight (visibly, naming the postgresql/pgcopy-binary pairing) and executes as a raw binary COPY byte passthrough — no CLR rows. Heterogeneous pairs still take the streaming fallback visibly, and RequireNative passes for matched pairs and errors otherwise. Core: - BulkImportChannel (Stream + CompleteAsync): commit only on explicit completion; dispose-without-complete is the abort signal, so a torn transfer can never commit. Mirrors the IFlowSink discipline. - BulkTransferBytePump: bounded-buffer (80 KB), cancellation-aware pump disposing both channels on every path, preserving typed errors. - Negotiation selects Native for matched capability pairs; endpoint items execute the rung by re-probing the same capability resolvers negotiation used. Flowthru.Extensions.EFCore.Npgsql (new package): - NpgsqlCopyStorageAdapter<T>: EFCore adapter decorator claiming postgresql/pgcopy-binary, exporting via COPY TO STDOUT and importing via COPY FROM STDIN (FORMAT BINARY) on Npgsql's raw binary COPY API (no EFCore.BulkExtensions — #129). Table, schema, and column list resolve from the EF model; non-Npgsql providers fail at construction. - Load semantics are explicit: NpgsqlBulkImportMode.Replace (default, transactional TRUNCATE + load) or Append; the import runs in one transaction, so a failed transfer rolls the target back untouched. - Streaming view + transactional streaming sink so mixed pairings execute the fallback rung against Postgres endpoints. - .NpgsqlTable<TRow, TContext>() catalog builder. Tests: negotiation/pump/adapter tiers run offline; a Docker-gated Postgres tier (FLOWTHRU_PG_TEST_CONNSTRING escape hatch for external servers) covers both rungs end-to-end through the same AddBulkTransfer wiring plus rollback on mid-COPY failure; a category-gated benchmark measured the native rung 19.6x faster than streaming on 200k rows. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
…nsfer Records the architecture decision consolidating #126/#127: wide transforms and homogeneous transfers execute outside the CLR as delegated arrows between ordinary typed items. Adds the wide-vs-narrow transform vocabulary to the Extension Developer glossary. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
The bulk-transfer rung pre-flight code from the AddBulkTransfer work was emitted from FlowthruDiagnosticCodes but never added to the diagnostic-ID registry; the tests:_test:diagnostic-id-registration check catches exactly this. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
…ement The restored step-shape hook compared C# type names positionally against decorator-declared names, which cannot succeed for byte[] artifact or directory payloads — no decorator spelling matches 'Byte[]'. Name agreement is now enforced only at positions whose underlying type carries [FlowthruSchema]; descriptive names for artifact payloads remain FTPY2007's design-time concern. Arity checks are unchanged. Surfaced by FlowthruCoverage's heatmap/icicle steps, the first bytes-output Python steps to run under the revived hook. Claude-Session: https://claude.ai/code/session_01DbyQtC1gFwbv9VAsyZ4AWQ
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.