Reorg aware streaming support

[fordN]

Reorg Aware Streaming Support for amp-python

Summary

This PR adds comprehensive blockchain reorganization (reorg) handling and streaming support to the amp-python client library. It enables real-time data streaming with automatic handling of blockchain reorganizations across all supported data loaders.

Key Features

1. Streaming Infrastructure

• New streaming module with iterator, types, and reorg handling
• ResponseBatchWithReorg type for streaming data and reorg events
• BlockRange metadata tracking for multi-network support
• Streaming iterator with automatic reorg detection

2. Enhanced Client API

• Enhanced load() method in QueryBuilder with streaming support
• New stream parameter enables continuous data streaming
• Support for reorg handling via handle_reorgs parameter
• Backward compatible - existing batch operations unchanged

3. Universal Reorg Support

• All 6 data loaders now support blockchain reorganizations:
  • PostgreSQL: Efficient JSONB-based deletion
  • Redis: Sorted set index for O(log N) lookups
  • Snowflake: SQL with FLATTEN for JSON operations
  • Iceberg: Scan and overwrite approach
  • DeltaLake: PyArrow compute masks for filtering
  • LMDB: Two-pass key-based deletion

4. Metadata Architecture

• Standardized _meta_block_ranges column across all loaders
• JSON format: [{"network": "ethereum", "start": 100, "end": 110}]
• Supports multi-network scenarios (bridges, cross-chain DEX, etc.)

Changes by Component

Core Library (552 lines added)

• src/amp/client.py: Enhanced QueryBuilder.load() with streaming support
• src/amp/loaders/base.py: Added load_stream_continuous() and _handle_reorg()
• src/amp/streaming/: New module with iterator, types, and reorg handling

Data Loaders (740 lines added)

• PostgreSQL: Single SQL DELETE with JSONB operations
• Redis: Leverages sorted set indexes for efficient deletion
• Snowflake: Native JSON SQL functions
• Iceberg: Full scan with filtered overwrite
• DeltaLake: Arrow compute operations
• LMDB: Key iteration and batch deletion

Testing (2,063 lines added)

• 110 new integration tests covering all loaders
• Comprehensive unit tests for streaming types
• Test scenarios: empty tables, missing metadata, multi-network, overlapping ranges

Documentation (355 lines added)

• Detailed reorg handling guide for each loader
• Implementation details and code examples

Usage Example

# Batch mode (existing)
result = client.query("SELECT * FROM blocks").load(postgres_loader, "blocks")

# Streaming mode (new)
for result in client.query("SELECT * FROM blocks").load(
    postgres_loader, 
    "blocks", 
    stream=True,
    handle_reorgs=True
):
    print(f"Loaded {result.rows_loaded} rows")
    if result.is_reorg:
        print(f"Handled reorg: {result.invalidation_ranges}")

Commits

ef17ef7 streaming: Base streaming implementations
c58b9fc loaders/base: Add streaming concepts to DataLoader class
5a23c12 client: Add streaming support to QueryBuilder and Client
cc5730f tests/unit: Add unit tests for reorg handling
d157e9d postgresql loader: Add reorg aware streaming support
876ade4 tests/performance: Fix config fixture usage in Redis loader tests
f1fea01 redis loader: Add reorg aware streaming support
073ada5 iceberg loader: Add reorg aware streaming support
21fd0ed deltalake loader: Add reorg aware streaming support
ea045c8 lmdb loader: Add reorg aware streaming support
a76fa1b snowflake loader: Add reorg aware streaming support
9c5aec4 docs: Summarizing reorg handling approach for each loader

[incrypto32]

Wouldn't a full table scan be problematic for very large datasets?

[incrypto32]

This is very inn-efficient, crates huge arrays unneccesarily. A simple bool flag list would work here instead

[incrypto32]

LGTM apart from the full table scans for Iceberg and Deltalake loader and the deletion logic in Deltalake loader.

I'll add the commits to fix that

[fordN]

I'm going to go ahead and merge this and consider the specific loader implementations to be in beta (needing optimization and hardening).