Streaming client improvements and snowflake loader features (#15)

* loaders: Add label management system for CSV-based enrichment

- Load labels from CSV files with automatic type detection
- Support hex string to binary conversion for Ethereum addresses
- Thread-safe label storage and retrieval
- Add LabelJoinConfig type for configuring joins

* streaming: Add unified stream state management for resume and dedup

- StreamStateStore interface with in-memory, null, and DB-backed
implementations
- Block range tracking with gap detection
- Reorg invalidation support

Key features:
- Resume from last processed position after crashes
- Exactly-once semantics via batch deduplication
- Gap detection and intelligent backfill
- Support for multiple networks and tables

* streaming: Add resilience features

- Exponential backoff with jitter for transient failures
- Adaptive rate limiting with automatic adjustment
- Back pressure detection and mitigation
- Error classification (transient vs permanent)
- Configurable retry policies

Features:
- Auto-detects rate limits and slows down requests
- Detects timeouts and adjusts batch sizes
- Production-tested configurations included

* *: Major base loader improvements for streaming and resilience

- Integrate state management for resume and deduplication
- Add label joining support with automatic type conversion
- Implement resilience features (retry, backpressure, rate limiting)
- Add metadata columns (_amp_batch_id) for reorg handling
- Support streaming with block ranges and reorg detection
- Separate _try_load_batch() for better error handling

* streaming: Enhance parallel execution; resumability & gap detection

- Add resume optimization that adjusts min_block based on persistent
state
- Implement gap-aware partitioning for intelligent backfill
- Add pre-flight table creation to avoid locking issues
- Improve error handling and logging for state operations
- Support label joining in parallel workers

Key features:
- Auto-detects processed ranges and skips already-loaded partitions
- Prioritizes gap filling before processing new data
- Efficient partition creation avoiding redundant work
- Visible logging for resume operations and adjustments

Resume workflow:
1. Query state store for max processed block
2. Adjust min_block to skip processed ranges
3. Detect gaps in processed data
4. Create partitions prioritizing gaps first
5. Process remaining historical data

* client: Integrate label manager into Client for enriched streaming

Add label management to Client class:
- Initialize LabelManager with configurable label directory
- Support loading labels from CSV files
- Pass label_manager to all loader instances
- Enable label joining in streaming queries via load() method

Updates:
- Client now supports label enrichment out of the box
- Loaders inherit label_manager from client
- Add pyarrow.csv dependency for label loading

* loaders: Update all impls for new base class interface

- PostgreSQL: Add reorg support with DELETE/UPDATE, metadata columns
- Redis: Add streaming metadata and batch ID support
- DeltaLake: Support new metadata columns
- Iceberg: Update for base class changes
- LMDB: Add metadata column support

All loaders now support:
- State-backed resume and deduplication
- Label joining via base class
- Resilience features (retry, backpressure)
- Reorg-aware streaming with metadata tracking

* test: Add comprehensive unit tests for streaming features

Add unit tests for all new streaming features:
- test_label_joining.py - Label enrichment with type conversion
- test_label_manager.py - CSV loading and label storage
- test_resilience.py - Retry, backoff, rate limiting
- test_resume_optimization.py - Resume position calculation
- test_stream_state.py - State store implementations
- test_streaming_helpers.py - Utility functions and batch ID generation
- test_streaming_types.py - BlockRange, ResumeWatermark types

* snowflake_loader: Major improvements with state management

- Add Snowflake-backed persistent state store (amp_stream_state table)
- Implement SnowflakeStreamStateStore with overlap detection
- Support multiple loading methods: stage, insert, pandas,
snowpipe_streaming
- Add connection pooling for parallel workers
- Implement reorg history tracking with simplified schema
- Support Parquet stage loading for better performance

State management features:
- Block-level overlap detection for different partition sizes
- MERGE-based upsert to prevent duplicate state entries
- Resume position calculation with gap detection
- Deduplication across runs

Performance improvements:
- Parallel stage loading with connection pool
- Optimized Parquet format for stage loads
- Efficient batch processing with metadata columns

* apps: Add Snowflake parallel loading applications

Add comprehensive demo applications for Snowflake loading:

1. snowflake_parallel_loader.py - Full-featured parallel loader
   - Configurable block ranges, workers, and partition sizes
   - Label joining with CSV files
   - State management with resume capability
   - Support for all Snowflake loading methods
   - Reorg history tracking
   - Clean formatted output with progress indicators

2. test_erc20_parallel_load.py - Simple ERC20 transfer loader
   - Basic parallel loading example
   - Good starting point for new users

3. test_erc20_labeled_parallel.py - Label-enriched example
   - Demonstrates label joining with token metadata
   - Shows how to enrich blockchain data

4. Query templates in apps/queries/
   - erc20_transfers.sql - Decode ERC20 Transfer events
   - README.md - Query documentation

* test: Add integration tests for loaders and streaming features

New tests:
- test_resilient_streaming.py - Resilience with real databases
- Enhanced Snowflake loader tests with state management
- Enhanced PostgreSQL tests with reorg handling
- Updated Redis, DeltaLake, Iceberg, LMDB loader tests

Integration test features:
- Real database containers (PostgreSQL, Redis, Snowflake)
- State persistence and resume testing
- Label joining with actual data
- Reorg detection and invalidation
- Parallel loading with multiple workers
- Error injection and recovery

Tests require Docker for database containers.

* infra: Add Docker and Kubernetes deployment configurations

Add containerization and orchestration support:
- General-purpose Dockerfile for amp-python
- Snowflake-specific Dockerfile with parallel loader
- GitHub Actions workflow for automated Docker publishing to ghcr.io
- Kubernetes deployment manifest for GKE with resource limits
- Comprehensive .dockerignore and .gitignore

Docker images:
- amp-python: Base image with all loaders
- amp-snowflake: Optimized for Snowflake parallel loading
  - Includes snowflake_parallel_loader.py as entrypoint
  - Pre-configured with Snowflake connector and dependencies

* docs: Add comprehensive documentation for new features

- All loading methods comparison (stage, insert, pandas, streaming)
- State management and resume capability
- Label joining for data enrichment
- Performance tuning and optimization
- Parallel loading configuration
- Reorg handling strategies
- Troubleshooting common issues

* data: Save performance benchmarks

* Formatting

* Linting fixes

* label manager: Remove data directory and document how to add label files

Users should now mount label CSV files at runtime using volume mounts
(Docker) or init containers with cloud storage (Kubernetes).

Changes
- Removed COPY data/ line from both Dockerfiles
- The /data directory is still created (mkdir -p /app /data) but empty
- Updated .gitignore to ignore entire data/ directory
- Removed data/** trigger from docker-publish workflow
- Added comprehensive docs/label_manager.md with:
  * Docker volume mount examples
  * Kubernetes init container pattern (recommended for large files)
  * ConfigMap examples (for small files <1MB)
  * PersistentVolume examples (for shared access)
  * Performance considerations and troubleshooting

* redis loader: Fix reorg handling when using string data structure

When data_structure='string', batch IDs are stored inside JSON values
rather than as hash fields. The reorg handler now checks the data
structure and uses GET+JSON parse for strings, HGET for hashes.

Author: fordN

.dockerignore

@@ -0,0 +1,68 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+*.egg
+*.egg-info/
+dist/
+build/
+.eggs/
+
+# Virtual environments
+.venv/
+venv/
+ENV/
+env/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+.DS_Store
+
+# Testing
+.pytest_cache/
+.coverage
+.coverage.*
+htmlcov/
+.tox/
+*.cover
+
+# Notebooks
+notebooks/
+*.ipynb
+.ipynb_checkpoints
+
+# Documentation
+docs/
+*.md
+!README.md
+!DOCKER_DEPLOY.md
+
+# Git
+.git/
+.gitignore
+.gitattributes
+
+# CI/CD
+.github/
+.gitlab-ci.yml
+
+# Local test data and logs
+tests/
+*.log
+/tmp/
+.test.env
+
+# UV/pip cache
+.uv/
+uv.lock
+
+# Docker
+Dockerfile*
+docker-compose*.yml
+.dockerignore

.github/workflows/docker-publish.yml

@@ -0,0 +1,89 @@
+name: Build and Push Docker Images
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - 'src/**'
+      - 'apps/**'
+      - 'Dockerfile*'
+      - 'pyproject.toml'
+      - '.github/workflows/docker-publish.yml'
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:  # Allow manual trigger
+    inputs:
+      tag:
+        description: 'Docker image tag suffix (default: latest)'
+        required: false
+        default: 'latest'
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  build-and-push:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    strategy:
+      matrix:
+        include:
+          - dockerfile: Dockerfile
+            suffix: ""
+            description: "Full image with all loader dependencies"
+          - dockerfile: Dockerfile.snowflake
+            suffix: "-snowflake"
+            description: "Snowflake-only image (minimal dependencies)"
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata for Docker
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          flavor: |
+            suffix=${{ matrix.suffix }},onlatest=true
+          tags: |
+            type=ref,event=branch
+            type=ref,event=pr
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=sha,prefix=sha-
+            type=raw,value=latest,enable={{is_default_branch}}
+
+      - name: Build and push Docker image (${{ matrix.description }})
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: ./${{ matrix.dockerfile }}
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          cache-from: type=gha,scope=${{ matrix.dockerfile }}
+          cache-to: type=gha,mode=max,scope=${{ matrix.dockerfile }}
+          platforms: linux/amd64,linux/arm64
+
+      - name: Image digest
+        run: |
+          echo "### ${{ matrix.description }}" >> $GITHUB_STEP_SUMMARY
+          echo "Digest: ${{ steps.meta.outputs.digest }}" >> $GITHUB_STEP_SUMMARY
+          echo "Tags: ${{ steps.meta.outputs.tags }}" >> $GITHUB_STEP_SUMMARY

.gitignore

@@ -0,0 +1,62 @@
+# Environment files
+.env
+.test.env
+*.env
+
+# Kubernetes secrets (NEVER commit these!)
+k8s/secret.yaml
+k8s/secrets.yaml
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+*.egg
+*.egg-info/
+dist/
+build/
+.eggs/
+
+# Virtual environments
+.venv/
+venv/
+ENV/
+env/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+.DS_Store
+
+# Testing
+.pytest_cache/
+.coverage
+.coverage.*
+htmlcov/
+.tox/
+*.cover
+.hypothesis/
+
+# Notebooks
+.ipynb_checkpoints/
+
+# Logs
+*.log
+/tmp/
+
+# UV/pip cache
+.uv/
+uv.lock
+
+# Data directories (local development)
+# Large datasets should be downloaded on-demand or mounted via ConfigMaps
+data/
+
+# Build artifacts
+*.tar.gz
+*.zip

Dockerfile

@@ -0,0 +1,93 @@
+# Multi-stage build for optimized image size
+# Stage 1: Build dependencies
+FROM python:3.12-slim AS builder
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install UV for fast dependency management
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+# Set working directory
+WORKDIR /app
+
+# Copy dependency files
+COPY pyproject.toml README.md ./
+
+# Install dependencies using UV (much faster than pip)
+# Install ALL dependencies including all loader dependencies
+# This ensures optional dependencies don't cause import errors
+RUN uv pip install --system --no-cache \
+    pandas>=2.3.1 \
+    pyarrow>=20.0.0 \
+    typer>=0.15.2 \
+    adbc-driver-manager>=1.5.0 \
+    adbc-driver-postgresql>=1.5.0 \
+    protobuf>=4.21.0 \
+    base58>=2.1.1 \
+    'eth-hash[pysha3]>=0.7.1' \
+    eth-utils>=5.2.0 \
+    google-cloud-bigquery>=3.30.0 \
+    google-cloud-storage>=3.1.0 \
+    arro3-core>=0.5.1 \
+    arro3-compute>=0.5.1 \
+    psycopg2-binary>=2.9.0 \
+    redis>=4.5.0 \
+    deltalake>=1.0.2 \
+    'pyiceberg[sql-sqlite]>=0.10.0' \
+    'pydantic>=2.0,<2.12' \
+    snowflake-connector-python>=4.0.0 \
+    snowpipe-streaming>=1.0.0 \
+    lmdb>=1.4.0
+
+# Stage 2: Runtime image
+FROM python:3.12-slim
+
+# Install runtime dependencies only
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libpq5 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Create non-root user for security
+RUN useradd -m -u 1000 amp && \
+    mkdir -p /app /data && \
+    chown -R amp:amp /app /data
+
+# Set working directory
+WORKDIR /app
+
+# Copy Python packages from builder
+COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
+
+# Copy UV from builder for package installation
+COPY --from=builder /usr/local/bin/uv /usr/local/bin/uv
+
+# Copy application code
+COPY --chown=amp:amp src/ ./src/
+COPY --chown=amp:amp apps/ ./apps/
+COPY --chown=amp:amp pyproject.toml README.md ./
+
+# Note: /data directory is created but empty by default
+# Mount data files at runtime using Kubernetes ConfigMaps or volumes
+
+# Install the amp package in the system Python (NOT editable for Docker)
+RUN uv pip install --system --no-cache .
+
+# Switch to non-root user
+USER amp
+
+# Set Python path
+ENV PYTHONPATH=/app
+ENV PYTHONUNBUFFERED=1
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+    CMD python -c "import sys; sys.exit(0)"
+
+# Default command - run ERC20 loader
+# Can be overridden with docker run arguments
+ENTRYPOINT ["python", "apps/test_erc20_labeled_parallel.py"]
+CMD ["--blocks", "100000", "--workers", "8", "--flush-interval", "0.5"]