Skip to content

Commit 77bf476

Browse files
kurodo3[bot]claudeeywalker
authored
feat(schema): pdb/rdb v0→v1 schema versioning + migration (ITL-535) (#232)
* docs(migrations): add schema versioning + pdb/rdb v0→v1 migration design spec Design spec for ITL-535 covering the Orcapod schema versioning framework and the concrete v0→v1 migration for pipeline DB and result DB tables. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs(migrations): update schema versioning spec with detection, NodeConfig, table_exists - Add SchemaVersionError hard-stop on old schema detection (replaces warning) - Add NodeConfig.ignore_schema for opt-in tolerance of specific old versions - Redesign detection flow: check v1 path first, only inspect v0 if v1 absent - Add per-v1-path process-level cache (_checked_pdb_paths / _checked_rdb_paths) - Add ArrowDatabaseProtocol.table_exists() as required new method - Add golden fixture testing strategy with sample tables for each schema version Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs(migrations): add implementation plan for schema versioning + pdb/rdb v0→v1 migration (ITL-535) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat(schema): add ContentHash.from_prefixed_digest, SchemaVersionError, NodeConfig.ignore_schema Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat(databases): add table_exists() to ArrowDatabaseProtocol and all backends Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat(constants): add PIPELINE_DB_SCHEMA_VERSION and RESULT_DB_SCHEMA_VERSION Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat(schema): FunctionJobNode + ResultCache path versioning, binary ContentHash, schema detection - FunctionJobNode: pipeline DB now uses versioned path (pdb_v1 suffix) - FunctionJobNode: meta_table hash columns changed to large_binary - FunctionJobNode: _ensure_pdb_schema() detects legacy v0 tables - ResultCache: result DB now uses versioned path (rdb_v1 suffix) - ResultCache: INPUT_DATA_HASH_COL changed to large_binary - ResultCache: _ensure_rdb_schema() detects legacy v0 tables - CachedFunctionPod: set_ignore_schema() propagates to ResultCache - Remove ITL-508 guard (superseded by schema versioning) - Update tests for new binary hash format and versioned paths Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix(schema): _ResultDatabaseReader.record_path now returns versioned path (rdb_v1 suffix) * feat(migrations): add MigrationResult dataclass and migrate_result_v0_to_v1() and migrate_pipeline_v0_to_v1() and migrate_node() * feat(cli): add orcapod migrate pipeline-db and result-db sub-commands * fix(test): derive CLI test cwd from __file__ instead of hardcoded absolute path The hardcoded Kurodo agent path caused all 6 CLI smoke tests to fail on GitHub Actions runners with FileNotFoundError. Use Path(__file__).parent.parent.parent to resolve the repo root portably. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix(migrate): address PR review comments - cli/migrate.py: fix result_path derivation for pipeline-db command; rdb is scoped to pod.uri (without schema:/instance: suffix), not the full node identity path. Add _result_path_from_node_path() helper. - result_cache.py: add _ensure_rdb_schema() guard to store() and get_all_records() so schema detection fires on any DB access, not only on lookup(). - migrations/pipeline_db.py: clarify docstrings — only __input_data_hash is backfilled from rdb; __output_data_hash stays null when absent (rdb doesn't store it, not counted as unresolvable). Add inline note on the upfront rdb-index memory tradeoff. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix(migrate): address PR review round 2 — schema guard, rdb perf, public API - result_cache: remove premature _checked_rdb_paths.add before SchemaVersionError raise (path was cached even when error fired) - result_cache: add public base_record_path property so migration utilities no longer need _record_path private access - result_cache: add _hash_val_to_binary helper to tolerate both str and bytes/memoryview ContentHash values in store() - function_node: expand pdb SchemaVersionError message with actual node path and migration command hint - pipeline_db/result_db: add track_skipped=True parameter to skip the v1 pre-scan on fresh large-table migrations - pipeline_db: replace full upfront rdb index load with per-batch get_records_by_ids lookup to avoid memory spike on large tables - pipeline_db: migrate_node() uses cache.base_record_path instead of private cache._record_path Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --------- Co-authored-by: agent-kurodo[bot] <268466204+agent-kurodo[bot]@users.noreply.github.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: Edgar Y. Walker <eywalker@users.noreply.github.com>
1 parent 4f7f6fb commit 77bf476

39 files changed

Lines changed: 4887 additions & 93 deletions

‎src/orcapod/cli/__init__.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import typer
1010

1111
from orcapod.cli.warm_cache import warm_cache
12+
from orcapod.cli.migrate import migrate_app
1213

1314
app = typer.Typer(
1415
name="orcapod",
@@ -23,3 +24,4 @@ def _main() -> None:
2324

2425

2526
app.command("warm-cache")(warm_cache)
27+
app.add_typer(migrate_app, name="migrate")

‎src/orcapod/cli/migrate.py‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
"""``orcapod migrate`` sub-commands.
2+
3+
Provides ``orcapod migrate pipeline-db`` and ``orcapod migrate result-db``
4+
for upgrading v0 pipeline/result DB tables to the v1 schema.
5+
"""
6+
from __future__ import annotations
7+
8+
import json
9+
10+
import typer
11+
12+
migrate_app = typer.Typer(
13+
name="migrate",
14+
help="Migrate Orcapod pipeline and result DB tables to the current schema version.",
15+
no_args_is_help=True,
16+
)
17+
18+
19+
def _result_path_from_node_path(node_path_str: str) -> tuple[str, ...]:
20+
"""Derive the rdb record path (pod URI) from a node identity path string.
21+
22+
A node identity path has the form ``pod/uri/schema:<hash>`` (and optionally
23+
``instance:<hash>`` for fine-grained paths). The result DB is scoped to
24+
just the pod URI portion — i.e. all components before any ``schema:`` or
25+
``instance:`` segment.
26+
27+
Args:
28+
node_path_str: Slash-separated node identity path as passed on the CLI.
29+
30+
Returns:
31+
Tuple of path components with ``schema:`` and ``instance:`` suffix
32+
components stripped.
33+
"""
34+
return tuple(
35+
p for p in node_path_str.split("/")
36+
if not p.startswith("schema:") and not p.startswith("instance:")
37+
)
38+
39+
40+
@migrate_app.command("pipeline-db")
41+
def migrate_pipeline_db(
42+
pipeline_db_path: str = typer.Argument(..., help="Path to the pipeline DB (Delta Lake root)."),
43+
result_db_path: str = typer.Argument(..., help="Path to the result DB (Delta Lake root)."),
44+
node_paths: list[str] = typer.Argument(..., help="One or more bare v0 node paths (slash-separated, e.g. 'my_node/schema:abc123')."),
45+
dry_run: bool = typer.Option(False, "--dry-run", help="Count rows to migrate without writing."),
46+
batch_size: int = typer.Option(500, "--batch-size", help="Rows processed per batch."),
47+
progress: bool = typer.Option(True, "--progress/--no-progress", help="Log progress messages."),
48+
json_summary: bool = typer.Option(False, "--json-summary", help="Print JSON summary to stdout on completion."),
49+
) -> None:
50+
"""Migrate one or more pipeline DB node paths from v0 to v1 schema."""
51+
from orcapod.databases.delta_lake_databases import DeltaTableDatabase
52+
from orcapod.migrations.pipeline_db import migrate_pipeline_v0_to_v1
53+
54+
pipeline_db = DeltaTableDatabase(base_path=pipeline_db_path)
55+
result_db = DeltaTableDatabase(base_path=result_db_path)
56+
57+
for node_path_str in node_paths:
58+
pipeline_path = tuple(node_path_str.split("/"))
59+
# The rdb is scoped to the pod URI (function_pod.uri), which is the
60+
# node identity path with schema:/instance: suffix components removed.
61+
result_path = _result_path_from_node_path(node_path_str)
62+
63+
if progress:
64+
typer.echo(f"Migrating pipeline DB: {pipeline_db_path}")
65+
typer.echo(f" node path: {node_path_str}")
66+
67+
result = migrate_pipeline_v0_to_v1(
68+
pipeline_db=pipeline_db,
69+
pipeline_path=pipeline_path,
70+
result_db=result_db,
71+
result_path=result_path,
72+
dry_run=dry_run,
73+
batch_size=batch_size,
74+
progress=progress,
75+
)
76+
77+
if progress:
78+
typer.echo(
79+
f" migrated: {result.rows_migrated} "
80+
f"skipped (already v1): {result.rows_skipped} "
81+
f"unresolvable: {result.rows_unresolvable}"
82+
)
83+
typer.echo(f" elapsed: {result.elapsed_s:.1f}s")
84+
85+
if json_summary:
86+
summary = {
87+
"rows_total": result.rows_total,
88+
"rows_migrated": result.rows_migrated,
89+
"rows_skipped": result.rows_skipped,
90+
"rows_unresolvable": result.rows_unresolvable,
91+
"elapsed_s": result.elapsed_s,
92+
"dry_run": result.dry_run,
93+
}
94+
typer.echo(json.dumps(summary))
95+
96+
97+
@migrate_app.command("result-db")
98+
def migrate_result_db(
99+
result_db_path: str = typer.Argument(..., help="Path to the result DB (Delta Lake root)."),
100+
record_paths: list[str] = typer.Argument(..., help="One or more bare v0 record paths (slash-separated)."),
101+
dry_run: bool = typer.Option(False, "--dry-run", help="Count rows to migrate without writing."),
102+
batch_size: int = typer.Option(500, "--batch-size", help="Rows processed per batch."),
103+
progress: bool = typer.Option(True, "--progress/--no-progress", help="Log progress messages."),
104+
json_summary: bool = typer.Option(False, "--json-summary", help="Print JSON summary to stdout on completion."),
105+
) -> None:
106+
"""Migrate one or more result DB record paths from v0 to v1 schema."""
107+
from orcapod.databases.delta_lake_databases import DeltaTableDatabase
108+
from orcapod.migrations.result_db import migrate_result_v0_to_v1
109+
110+
result_db = DeltaTableDatabase(base_path=result_db_path)
111+
112+
for record_path_str in record_paths:
113+
result_path = tuple(record_path_str.split("/"))
114+
115+
if progress:
116+
typer.echo(f"Migrating result DB: {result_db_path}")
117+
typer.echo(f" record path: {record_path_str}")
118+
119+
result = migrate_result_v0_to_v1(
120+
result_db=result_db,
121+
result_path=result_path,
122+
dry_run=dry_run,
123+
batch_size=batch_size,
124+
progress=progress,
125+
)
126+
127+
if progress:
128+
typer.echo(
129+
f" migrated: {result.rows_migrated} "
130+
f"skipped (already v1): {result.rows_skipped}"
131+
)
132+
typer.echo(f" elapsed: {result.elapsed_s:.1f}s")
133+
134+
if json_summary:
135+
summary = {
136+
"rows_total": result.rows_total,
137+
"rows_migrated": result.rows_migrated,
138+
"rows_skipped": result.rows_skipped,
139+
"rows_unresolvable": result.rows_unresolvable,
140+
"elapsed_s": result.elapsed_s,
141+
"dry_run": result.dry_run,
142+
}
143+
typer.echo(json.dumps(summary))

‎src/orcapod/core/cached_function_pod.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,9 +63,19 @@ def result_database(self) -> ArrowDatabaseProtocol:
6363

6464
@property
6565
def record_path(self) -> tuple[str, ...]:
66-
"""Return the path to the cached records in the result store."""
66+
"""Return the path to the cached records in the result store (versioned)."""
6767
return self._cache.record_path
6868

69+
def set_ignore_schema(self, ignore_schema: tuple[str, ...] | None) -> None:
70+
"""Propagate ``ignore_schema`` setting to the underlying ``ResultCache``.
71+
72+
Args:
73+
ignore_schema: Tuple of schema version strings to tolerate (e.g.
74+
``("v0",)``), or ``None`` to use the default (raise on any
75+
old schema).
76+
"""
77+
self._cache.set_ignore_schema(ignore_schema)
78+
6979
def lookup_cached_data(self, data: DataProtocol) -> DataProtocol | None:
7080
"""Look up a cached result for ``data`` without triggering computation.
7181

‎src/orcapod/core/data_function.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -963,7 +963,7 @@ def set_auto_flush(self, on: bool = True) -> None:
963963

964964
@property
965965
def record_path(self) -> tuple[str, ...]:
966-
"""Return the path to the record in the result store."""
966+
"""Return the path to the cached records in the result store (versioned)."""
967967
return self._cache.record_path
968968

969969
def call(

‎src/orcapod/core/nodes/function_node.py‎

Lines changed: 85 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
from orcapod.core.streams.base import StreamBase
3232
from orcapod.core.tracker import DEFAULT_TRACKER_MANAGER
3333
from orcapod.core.datagrams.tag_data import EmptyData, Tag
34-
from orcapod.errors import EphemeralResultMissingError, PipelineJobRequiredError
34+
from orcapod.errors import EphemeralResultMissingError, PipelineJobRequiredError, SchemaVersionError
3535
from orcapod.protocols.core_protocols import (
3636
FunctionPodProtocol,
3737
DataFunctionExecutorProtocol,
@@ -46,7 +46,7 @@
4646
DataExecutionLoggerProtocol,
4747
ExecutionObserverProtocol,
4848
)
49-
from orcapod.system_constants import constants
49+
from orcapod.system_constants import constants, PIPELINE_DB_SCHEMA_VERSION, RESULT_DB_SCHEMA_VERSION
5050
from orcapod.types import (
5151
ColumnConfig,
5252
ContentHash,
@@ -86,6 +86,11 @@
8686
# record_and_forward() before downstream emission.
8787
_TAG_NODE_INPUT_REF = "_tag_node_input_ref"
8888

89+
# Module-level set of versioned pdb paths that have been checked for legacy schema.
90+
# Keyed by the full versioned path (node_identity_path + (PIPELINE_DB_SCHEMA_VERSION,)).
91+
# Populated on first access; prevents repeated table_exists checks across instances.
92+
_checked_pdb_paths: set[tuple[str, ...]] = set()
93+
8994

9095
def _executor_supports_concurrent(
9196
data_function: DataFunctionProtocol,
@@ -685,8 +690,13 @@ def __init__(
685690

686691
@property
687692
def record_path(self) -> tuple[str, ...]:
688-
"""Path to cached records in the result store."""
689-
return self._record_path
693+
"""Path to cached records in the result store (versioned).
694+
695+
Returns the v1 schema path: ``_record_path + (RESULT_DB_SCHEMA_VERSION,)``.
696+
Matches ``CachedFunctionPod.record_path`` semantics so that stub nodes
697+
(loaded from a saved job) resolve to the same storage location as live nodes.
698+
"""
699+
return self._record_path + (RESULT_DB_SCHEMA_VERSION,)
690700

691701

692702
# ---------------------------------------------------------------------------
@@ -786,6 +796,66 @@ def node_config(self) -> NodeConfig:
786796
@node_config.setter
787797
def node_config(self, value: NodeConfig) -> None:
788798
self._node_config = value
799+
if self._cached_function_pod is not None:
800+
self._cached_function_pod.set_ignore_schema(value.ignore_schema)
801+
if self._ephemeral_cached_pod is not None:
802+
self._ephemeral_cached_pod.set_ignore_schema(value.ignore_schema)
803+
804+
@property
805+
def _versioned_pipeline_path(self) -> tuple[str, ...]:
806+
"""Pipeline DB path with schema version suffix appended.
807+
808+
All pipeline DB reads and writes use this path (not ``node_identity_path``
809+
directly) so that v0 and v1 schema data live at physically separate locations.
810+
"""
811+
return self.node_identity_path + (PIPELINE_DB_SCHEMA_VERSION,)
812+
813+
def _ensure_pdb_schema(self) -> None:
814+
"""Check once per versioned path that no legacy v0 schema table is present.
815+
816+
Detection flow:
817+
1. If the v1 versioned path already exists → already migrated, skip check.
818+
2. If v1 absent → check whether the unversioned (v0) path exists.
819+
3. If v0 exists AND the version is not in ``ignore_schema`` → raise.
820+
4. If v0 exists AND version is ignored → log a warning and proceed.
821+
5. If v0 also absent → first use, proceed normally.
822+
823+
Results are cached per versioned path (module-level set) so the check
824+
happens at most once per process per versioned path.
825+
826+
Raises:
827+
SchemaVersionError: When a v0 table is found and ``ignore_schema``
828+
does not include ``"v0"``.
829+
"""
830+
global _checked_pdb_paths
831+
versioned_path = self._versioned_pipeline_path
832+
if versioned_path in _checked_pdb_paths:
833+
return
834+
if self._pipeline_database is None:
835+
return
836+
# v1 path exists → schema is current, mark as checked
837+
if self._pipeline_database.table_exists(versioned_path):
838+
_checked_pdb_paths.add(versioned_path)
839+
return
840+
# v1 absent — check for legacy v0 table at the unversioned path
841+
if self._pipeline_database.table_exists(self.node_identity_path):
842+
ignore = self._node_config.ignore_schema or ()
843+
if "v0" not in ignore:
844+
node_path_str = "/".join(self.node_identity_path)
845+
raise SchemaVersionError(
846+
f"Pipeline DB at {self.node_identity_path!r} contains a legacy v0 schema table.\n"
847+
"Run migration first:\n"
848+
f" orcapod migrate pipeline-db <PIPELINE_DB_PATH> <RESULT_DB_PATH> {node_path_str}\n"
849+
"To suppress this error and recompute all results instead, set:\n"
850+
' node.node_config = NodeConfig(ignore_schema=("v0",))'
851+
)
852+
logger.warning(
853+
"Pipeline DB at %r has a legacy v0 schema table; "
854+
"proceeding without migration because ignore_schema includes 'v0'. "
855+
"All results will be recomputed from scratch.",
856+
self.node_identity_path,
857+
)
858+
_checked_pdb_paths.add(versioned_path)
789859

790860
# ------------------------------------------------------------------
791861
# attach_databases
@@ -1088,7 +1158,7 @@ def _filter_by_content_hash(self, table: "pa.Table") -> "pa.Table":
10881158
f"required column {col_name!r} is missing from the stored table. "
10891159
"This may indicate records written by an older version of the code."
10901160
)
1091-
own_hash = self.content_hash().to_string()
1161+
own_hash = self.content_hash().to_prefixed_digest()
10921162
mask = pc.equal(table.column(col_name), own_hash)
10931163
return table.filter(mask)
10941164

@@ -1624,30 +1694,14 @@ def add_pipeline_record(
16241694
result keyed by the same hash (= the downstream's ``INPUT_DATA_HASH_COL``).
16251695
"""
16261696
self._require_pipeline_database()
1697+
self._ensure_pdb_schema()
16271698
base_entry_id = self.compute_base_entry_id(tag, input_data)
16281699

1629-
# Guard against pre-ITL-508 pipeline DB records that are missing the new
1630-
# versioning columns. If such records exist, fail fast with a clear message
1631-
# rather than letting the subsequent filter crash with a cryptic Arrow error.
1632-
_all_existing = self._pipeline_database.get_all_records(self.node_identity_path)
1633-
if _all_existing is not None and _all_existing.num_rows > 0:
1634-
_missing = [
1635-
col
1636-
for col in (_PIPELINE_BASE_ENTRY_ID_COL, _PIPELINE_RECOMPUTATION_INDEX_COL)
1637-
if col not in _all_existing.schema.names
1638-
]
1639-
if _missing:
1640-
raise ValueError(
1641-
f"Pipeline database at {self.node_identity_path!r} contains records "
1642-
f"that are missing required ITL-508 columns: {_missing!r}. "
1643-
"Please clear or migrate the pipeline database before using this node."
1644-
)
1645-
16461700
# Determine the next recomputation index by querying all existing rows
16471701
# for this base_entry_id. No await is used here, so within a single-threaded
16481702
# asyncio event loop this read-then-write sequence is uninterrupted.
16491703
existing = self._pipeline_database.get_records_with_column_value(
1650-
self.node_identity_path,
1704+
self._versioned_pipeline_path,
16511705
{_PIPELINE_BASE_ENTRY_ID_COL: base_entry_id},
16521706
)
16531707
if existing is None or existing.num_rows == 0:
@@ -1674,14 +1728,14 @@ def add_pipeline_record(
16741728
[data_record_id.bytes], type=pa.large_binary()
16751729
),
16761730
constants.NODE_CONTENT_HASH_COL: pa.array(
1677-
[self.content_hash().to_string()], type=pa.large_string()
1731+
[self.content_hash().to_prefixed_digest()], type=pa.large_binary()
16781732
),
16791733
constants.INPUT_DATA_HASH_COL: pa.array(
1680-
[input_data.content_hash().to_string()], type=pa.large_string()
1734+
[input_data.content_hash().to_prefixed_digest()], type=pa.large_binary()
16811735
),
16821736
constants.OUTPUT_DATA_HASH_COL: pa.array(
1683-
[output_data_hash.to_string() if output_data_hash is not None else None],
1684-
type=pa.large_string(),
1737+
[output_data_hash.to_prefixed_digest() if output_data_hash is not None else None],
1738+
type=pa.large_binary(),
16851739
),
16861740
f"{constants.META_PREFIX}input_data{constants.CONTEXT_KEY}": pa.array(
16871741
[input_data.data_context_key], type=pa.large_string()
@@ -1709,7 +1763,7 @@ def add_pipeline_record(
17091763
)
17101764

17111765
self._pipeline_database.add_record(
1712-
self.node_identity_path,
1766+
self._versioned_pipeline_path,
17131767
versioned_entry_id,
17141768
combined_record,
17151769
skip_duplicates=True,
@@ -1848,8 +1902,9 @@ def _fetch_joined_records(
18481902
if self._cached_function_pod is None or self._pipeline_database is None:
18491903
return None
18501904

1905+
self._ensure_pdb_schema()
18511906
taginfo = self._pipeline_database.get_all_records(
1852-
self.node_identity_path,
1907+
self._versioned_pipeline_path,
18531908
record_id_column=_PIPELINE_ENTRY_ID_COL,
18541909
)
18551910

@@ -1965,7 +2020,7 @@ def _fetch_joined_records(
19652020
)
19662021
cached_hash = None
19672022
else:
1968-
cached_hash = ContentHash.from_string(raw_hash)
2023+
cached_hash = ContentHash.from_prefixed_digest(raw_hash)
19692024
empty_data_tokens[base_eid] = EmptyData(
19702025
cached_content_hash=cached_hash,
19712026
data_context=self.data_context,

0 commit comments

Comments
 (0)