Skip to content

CLAUDE.md

Latest commit

 

History

History
376 lines (294 loc) · 18.2 KB

CLAUDE.md

File metadata and controls

376 lines (294 loc) · 18.2 KB
Raw

Claude Code instructions for orcapod-python

Running commands

Always run Python commands via uv run, e.g.:

uv run pytest tests/
uv run python -c "..."

Never use python, pytest, or python3 directly.

Updating agent instructions

When adding or changing any instruction, update BOTH:

  • CLAUDE.md (for Claude Code)
  • .zed/rules (for Zed AI)

Design issues log

DESIGN_ISSUES.md at the project root is the canonical log of known design problems, bugs, and code quality issues.

When fixing a bug or addressing a design problem:

  1. Check DESIGN_ISSUES.md first — if a matching issue exists, update its status to in progress while working and resolved once done, adding a brief Fix: note.
  2. If no matching issue exists, ask the user whether it should be added before proceeding. If yes, add it (status open or in progress as appropriate).

When discovering a new issue that won't be fixed immediately, ask the user whether it should be logged in DESIGN_ISSUES.md before adding it.

Superpowers artifacts

Place all superpowers-related artifacts (design specs, plans, etc.) in the superpowers/ directory at the project root — not under docs/. The docs/ directory is reserved for actual library documentation.

  • Specs go in superpowers/specs/

Backward compatibility

This is a greenfield project pre-v0.1.0. Do not add backward-compatibility shims, re-exports, aliases, or deprecation wrappers when making design or implementation changes. Just change the code and update all references directly.

No sys.modules hacks

Never manipulate sys.modules directly (e.g. sys.modules.setdefault). If a subpackage import path doesn't work, create a proper re-export package with an __init__.py instead.

Docstrings

Use Google style Python docstrings everywhere.

Never mix in ReST markup. Do not use :meth:, :class:, :exc:, :attr:, :func:, or any other ReST cross-reference roles. Use plain backtick notation to refer to names (e.g. PollingSource, iter_data, CursorInvalidatedError).

Linear issue tracking

All work must be linked to a Linear issue. Before starting any feature, bug fix, or refactor:

  1. Check for an existing issue — search Linear for a corresponding issue.
  2. If none exists — ask the developer whether to create one. Do not proceed without either a linked issue or explicit approval to skip.
  3. When starting work on an issue — update its Linear status to In Progress.
  4. When a new issue is discovered during development (bug, design problem, deferred work), create a corresponding Linear issue using the template below.

When creating Linear issues, always use this template for the description:

## Overview
What is this project about? Describe the problem space and the high-level approach.

## Goals & Success Criteria
* Specific, measurable outcomes.

## Scope & Boundaries
(Optional — remove if not needed.)
In scope:
* ...
Out of scope:
* ...

## Dependencies & Risks
(Optional — remove if none.)
* ...

## Resources & References
(Optional — remove if none.)
* ...

## Milestones
(Optional — only for projects longer than ~4 weeks. Remove for shorter projects.)
* ...

Remove any optional sections that don't apply rather than leaving them empty.

Branches and PRs

When working on a feature, create and checkout a git branch using the gitBranchName returned by the primary Linear issue (e.g. eywalker/plt-911-add-documentation-for-orcapod-python).

Feature branch PRs always target main. Create a feature branch from main and open PRs against main.

If a feature branch / PR corresponds to multiple Linear issues, list all of them in the PR description body so that Linear's GitHub integration auto-tracks the PR against each issue. Use the format Fixes PLT-123 or Closes PLT-123 (GitHub magic words) for issues that the PR fully resolves, and simply mention PLT-456 for issues that are related but not fully resolved by the PR.

Responding to PR reviews

When asked to respond to PR reviewer comments:

  1. Read all comments carefully — fetch every review comment on the PR before forming any opinion.
  2. Evaluate each comment — decide whether to accept, partially accept, or decline, and why.
  3. Present a revision plan — show the full plan (table: comment summary, verdict, proposed action) and wait for user approval before touching any code or posting any replies.
  4. Fix, then reply — once approved, make all fixes in a single commit, then post replies to each reviewer comment explaining what was done (or why it was declined).

Never implement changes or reply to reviewers before the plan has been approved.

Git commits

Always use Conventional Commits style:

<type>(<optional scope>): <short description>

Common types: feat, fix, refactor, test, docs, chore, perf, ci.

Examples:

  • feat(schema): add optional_fields to Schema
  • fix(data_function): reject variadic parameters at construction
  • test(function_pod): add schema validation tests
  • refactor(schema_utils): use Schema.optional_fields directly

Project layout

src/orcapod/
├── types.py                    # Schema, ColumnConfig, ContentHash
├── system_constants.py         # Column prefixes and separators
├── errors.py                   # InputValidationError, DuplicateTagError, FieldNotResolvableError
├── config.py                   # Config dataclass
├── contexts/                   # DataContext (semantic_hasher, arrow_hasher, type_converter)
├── protocols/
│   ├── hashing_protocols.py    # PipelineElementProtocol, ContentIdentifiableProtocol
│   └── core_protocols/         # StreamProtocol, PodProtocol, SourceProtocol,
│                               # DataFunctionProtocol, DatagramProtocol, TagProtocol,
│                               # DataProtocol, TrackerProtocol
├── core/
│   ├── base.py                 # ContentIdentifiableBase, PipelineElementBase, TraceableBase
│   ├── static_output_pod.py    # StaticOutputPod (operator base), DynamicPodStream
│   ├── function_pod.py         # FunctionPod, FunctionPodStream, FunctionNode
│   ├── data_function.py      # DataFunctionBase, PythonDataFunction, CachedDataFunction
│   ├── operator_node.py        # OperatorNode (DB-backed operator execution)
│   ├── tracker.py              # Invocation tracking
│   ├── datagrams/
│   │   ├── datagram.py         # Datagram (unified dict/Arrow backing, lazy conversion)
│   │   └── tag_data.py       # Tag (+ system tags), Data (+ source info)
│   ├── sources/
│   │   ├── base.py             # RootSource (abstract, no upstream)
│   │   ├── arrow_table_source.py  # Core source — all other sources delegate to it
│   │   ├── derived_source.py   # DerivedSource (backed by FunctionNode/OperatorNode DB)
│   │   ├── csv_source.py, dict_source.py, list_source.py,
│   │   │   data_frame_source.py, delta_table_source.py  # Delegating wrappers
│   │   └── source_registry.py  # SourceRegistry for provenance resolution
│   ├── streams/
│   │   ├── base.py             # StreamBase (abstract)
│   │   └── arrow_table_stream.py  # ArrowTableStream (concrete, immutable)
│   └── operators/
│       ├── base.py             # UnaryOperator, BinaryOperator, NonZeroInputOperator
│       ├── join.py             # Join (N-ary inner join, commutative)
│       ├── merge_join.py       # MergeJoin (binary, colliding cols → sorted list[T])
│       ├── semijoin.py         # SemiJoin (binary, non-commutative)
│       ├── batch.py            # Batch (group rows by count, types become list[T])
│       ├── group_by.py         # GroupBy (many→one reduction keyed on tag values)
│       ├── column_selection.py # Select/Drop Tag/Data columns
│       ├── mappers.py          # MapTags, MapData (rename columns)
│       └── filters.py          # PolarsFilter
├── hashing/
│   └── semantic_hashing/       # BaseSemanticHasher, type handlers
├── semantic_types/             # Type conversion (Python ↔ Arrow)
├── databases/                  # ArrowDatabaseProtocol implementations (Delta Lake, in-memory)
└── utils/
    ├── arrow_data_utils.py     # System tag manipulation, source info, column helpers
    ├── arrow_utils.py          # Arrow table utilities
    ├── schema_utils.py         # Schema extraction, union, intersection, compatibility
    └── lazy_module.py          # LazyModule for deferred heavy imports

tests/
├── test_core/
│   ├── datagrams/              # Lazy conversion, dict/Arrow round-trip
│   ├── sources/                # Source construction, protocol conformance, DerivedSource
│   ├── streams/                # ArrowTableStream behavior
│   ├── function_pod/           # FunctionPod, FunctionNode, pipeline hash integration
│   ├── operators/              # All operators, OperatorNode, MergeJoin
│   └── data_function/        # DataFunction, CachedDataFunction
├── test_hashing/               # Semantic hasher, hash stability
├── test_databases/             # Delta Lake, in-memory, no-op databases
└── test_semantic_types/        # Type converter tests

Architecture overview

See orcapod-design.md at the project root for the full design specification.

Core data flow

RootSource → ArrowTableStream → [Operator / FunctionPod] → ArrowTableStream → ...

Every stream is an immutable sequence of (Tag, Data) pairs backed by a PyArrow Table. Tag columns are join keys and metadata; data columns are the data payload.

Core abstractions

Datagram (core/datagrams/datagram.py) — immutable data container with lazy dict ↔ Arrow conversion. Two specializations:

  • Tag — metadata columns + hidden system tag columns for provenance tracking
  • Data — data columns + per-column source info provenance tokens

Stream (core/streams/arrow_table_stream.py) — immutable (Tag, Data) sequence. Key methods: output_schema(), keys(), iter_data(), as_table().

Source (core/sources/) — produces a stream from external data. ArrowTableSource is the core implementation; CSV/Delta/DataFrame/Dict/List sources all delegate to it internally. Each source adds source-info columns and a system tag column. DerivedSource wraps a FunctionNode/OperatorNode's DB records as a new source.

Function Pod (core/function_pod.py) — wraps a DataFunction that transforms individual data. Never inspects tags. Two execution models:

  • FunctionPod → FunctionPodStream: lazy, in-memory
  • FunctionNode: DB-backed, two-phase (yield cached results first, then compute missing)

Operator (core/operators/) — structural pod transforming streams without synthesizing new data values. All subclass StaticOutputPod:

  • UnaryOperator — 1 input (Batch, Select/Drop columns, Map, Filter)
  • BinaryOperator — 2 inputs (MergeJoin, SemiJoin)
  • NonZeroInputOperator — 1+ inputs (Join)

OperatorNode (core/operator_node.py) — DB-backed operator execution, analogous to FunctionNode.

Strict operator / function pod boundary

Operator Function Pod
Inspects data content Never Yes
Inspects / uses tags Yes No
Can rename columns Yes No
Synthesizes new values No Yes
Stream arity Configurable Single in, single out

GroupBy is the only operator that reduces row count many→one: it collapses N rows sharing a tag tuple into one row with list-valued members. It still synthesizes no new data values — every emitted element came from an input row — and it keys only on tags, never on data.

Two identity chains

Every pipeline element has two parallel hashes:

  1. content_hash() — data-inclusive. Changes when data changes. Used for deduplication and memoization.
  2. pipeline_hash() — schema + topology only. Ignores data content. Used for DB path scoping so that different sources with identical schemas share database tables.

Base case: RootSource.pipeline_identity_structure() returns (tag_schema, data_schema). Each downstream node's pipeline hash commits to its own identity plus the pipeline hashes of its upstreams, forming a Merkle chain.

The pipeline hash uses a resolver pattern — PipelineElementProtocol objects route through pipeline_hash(), other ContentIdentifiable objects route through content_hash().

Column naming conventions

Prefix Meaning Example Controlled by
__ System metadata __data_id, __pod_version ColumnConfig(meta=True)
_source_ Source info provenance _source_age ColumnConfig(source=True)
_tag:: System tag _tag::source:abc123 ColumnConfig(system_tags=True)
_context_key Data context _context_key ColumnConfig(context=True)

Prefixes are computed from SystemConstant in system_constants.py. The constants singleton (with no global prefix) is used throughout.

System tag evolution rules

  1. Name-preserving — single-stream ops (filter, select, map). Column name and value pass through unchanged.

  2. Name-extending — multi-input ops (join, merge join). Each input's system tag column name gets ::{pipeline_hash}:{canonical_position} appended. Commutative operators canonically order inputs by pipeline_hash and sort system tag values per row.

  3. Reducing — many→one ops (Batch, GroupBy). User tag and data columns become list[T], and source-info columns become list[str] with one element per member. System tag columns must stay scalar, because _build_record_id_preimage (core/nodes/function_node.py) hashes them directly to derive record identity. They fold to a deterministic digest via arrow_utils.fold_system_tag_values and their column name gains ::{pipeline_hash} — a blend of rules 2 and 3.

    The fold is SHA-based (uuid5 for record_id, combine_hashes for source_id) and so is stable across processes. Never use hash() or a set-based construction there: the digest becomes a cache key, so a per-process value would look correct in a single-process test and miss the cache on every new driver run.

Schema types and ColumnConfig

Schema (types.py) — immutable Mapping[str, DataType] with optional_fields support. output_schema() always returns (tag_schema, data_schema) as a tuple of Schemas.

ColumnConfig (types.py) — frozen dataclass controlling which column groups are included. Fields: meta, context, source, system_tags, content_hash, sort_by_tags. Normalize via ColumnConfig.handle_config(columns, all_info) at the top of output_schema() and as_table() methods. all_info=True sets everything to True.

Key patterns

  • LazyModule("pyarrow") — deferred import for heavy deps (pyarrow, polars). Used in if TYPE_CHECKING: / else: blocks at module level.
  • Argument symmetry — each operator declares argument_symmetry(streams) returning frozenset (commutative) or tuple (ordered). Determines how upstream hashes combine.
  • StaticOutputPod.process() → DynamicPodStream — wraps static_process() output with timestamp-based staleness detection and automatic recomputation.
  • Source delegation — CSVSource, DictSource, etc. all create an internal ArrowTableSource and delegate every method to it.

Important implementation details

  • ArrowTableSource.__init__ raises ValueError if any tag_columns are not in the table.
  • ArrowTableStream requires at least one data column; raises ValueError otherwise.
  • FunctionNode.iter_data() Phase 1 returns ALL records in the shared pipeline_path DB table (not filtered to current inputs). Phase 2 skips inputs whose hash is already in the DB.
  • Empty data → ArrowTableSource raises ValueError("Table is empty").
  • DerivedSource before run() → raises ValueError (no computed records).
  • Join requires non-overlapping data columns; raises InputValidationError on collision.
  • MergeJoin requires colliding data columns to have identical types; merges into sorted list[T] with source columns reordered to match.
  • Operators predict their output schema (including system tag column names) without performing the actual computation. Verify a prediction against unary_static_process(stream).output_schema(...), not against process(...) (which delegates straight back to the pod, making the check circular) and not against as_table() (which legitimately differs: ArrowTableStream.output_schema ignores columns.source, so _source_*, _content_hash, and _context_key appear in the table but never in the schema).
  • GroupBy requires every column in by to be a scalar tag column; it raises InputValidationError for unknown, data, or list-valued columns (the last usually means the stream came from Batch — group before batching, not after). Members are sorted by non-key tag values with the system record_id as tiebreaker, so the hashed lists are stable across runs. Groups themselves are emitted in key order, so a reordered input yields a byte-identical table.
  • Data source-info values may be str, None, or list[...]; the Arrow and Python types are derived from the value, with None mapping to large_string. A many→one operator must emit a list for every row of a list-valued column — mixing a bare None into some rows makes per-row as_table() schemas diverge and pa.concat_tables fail on the barrier path.
  • Aggregating operators (Batch, GroupBy) must build list-valued columns via arrow_utils.build_aggregated_table, never pa.list_(field.type). Arrow cannot embed an extension type inside a list value field, so the naive call raises ArrowNotImplementedError on any logical-typed column (e.g. a pod annotated -> Path). The helper builds the list over the element's storage type and wraps it in the outer list[<element>] extension type from ListLogicalType.