Decoding large int

docker_clickhouse/.env.example

@@ -1,3 +1,3 @@
-CLICKHOUSE_USER=default
-CLICKHOUSE_PASSWORD=default
+CLICKHOUSE_USER=tiders_user
+CLICKHOUSE_PASSWORD=tiders_password
 CLICKHOUSE_DB=tiders

docker_postgres/.env.example

@@ -1,3 +1,3 @@
-POSTGRES_USER=postgres
-POSTGRES_PASSWORD=secret
+POSTGRES_USER=tiders_user
+POSTGRES_PASSWORD=tiders_password
 POSTGRES_DB=tiders

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "tiders"
-version = "0.1.5"
+version = "0.1.6"
 description = "Library for building blockchain pipelines"
 repository = "https://github.com/yulesa/tiders"
 homepage = "https://github.com/yulesa/tiders"
@@ -12,7 +12,7 @@ authors = [
 ]
 requires-python = ">=3.11,<3.14"
 dependencies = [
-    "tiders-core>=0.2.0",
+    "tiders-core>=0.3.0",
     "pyarrow>=22.0.0",
     "pyyaml>=6.0",
     "click>=8.0",
@@ -21,7 +21,7 @@ dependencies = [
 
 [project.optional-dependencies]
 duckdb = ["duckdb>=1.4.1"]
-clickhouse = ["clickhouse-connect>=0.9.2"]
+clickhouse = ["clickhouse-connect>=1.0"]
 delta_lake = ["deltalake>=1.2.1"]
 iceberg = ["pyiceberg[sql-sqlite]>=0.10.0"]
 polars = ["polars>=1.34.0"]

src/tiders/cli/tiders_yaml_parser.py

@@ -134,6 +134,7 @@
     AddColumnsConfig,
     CopyColumnsConfig,
     U256ToBinaryConfig,
+    LargeIntColumnsToBinaryConfig,
     Writer,
     WriterKind,
 )
@@ -989,6 +990,7 @@ def _parse_step_config(kind: StepKind, raw: dict[str, Any], path: str) -> Any:
             input_table=raw.get("input_table", "logs"),
             output_table=raw.get("output_table", "decoded_logs"),
             hstack=raw.get("hstack", True),
+            large_int_as_binary=raw.get("large_int_as_binary", False),
         )
 
     if kind == StepKind.CAST_BY_TYPE:
@@ -1078,6 +1080,31 @@ def _parse_step_config(kind: StepKind, raw: dict[str, Any], path: str) -> Any:
             )
         return U256ToBinaryConfig(tables=raw.get("tables"))
 
+    if kind == StepKind.LARGE_INT_COLUMNS_TO_BINARY:
+        valid_keys = {f.name for f in dataclasses.fields(LargeIntColumnsToBinaryConfig)}
+        unknown = set(raw.keys()) - valid_keys
+        if unknown:
+            raise YamlConfigError(
+                f"Unknown large_int_columns_to_binary config keys: {sorted(unknown)}. "
+                f"Valid keys: {sorted(valid_keys)}.",
+                cfg_path,
+            )
+        if "table_name" not in raw:
+            raise YamlConfigError(
+                "large_int_columns_to_binary requires 'config.table_name' (string).",
+                cfg_path,
+            )
+        if "columns" not in raw:
+            raise YamlConfigError(
+                "large_int_columns_to_binary requires 'config.columns' "
+                "(list of column names).",
+                cfg_path,
+            )
+        return LargeIntColumnsToBinaryConfig(
+            table_name=raw["table_name"],
+            columns=list(raw["columns"]),
+        )
+
     if kind == StepKind.SET_CHAIN_ID:
         valid_keys = {f.name for f in dataclasses.fields(SetChainIdConfig)}
         unknown = set(raw.keys()) - valid_keys

src/tiders/config.py

@@ -63,6 +63,7 @@ class StepKind(str, Enum):
     CAST_BY_TYPE = "cast_by_type"
     BASE58_ENCODE = "base58_encode"
     U256_TO_BINARY = "u256_to_binary"
+    LARGE_INT_COLUMNS_TO_BINARY = "large_int_columns_to_binary"
     SVM_DECODE_INSTRUCTIONS = "svm_decode_instructions"
     SVM_DECODE_LOGS = "svm_decode_logs"
     JOIN_BLOCK_DATA = "join_block_data"
@@ -372,6 +373,11 @@ class EvmDecodeEventsConfig:
             (default ``"decoded_logs"``).
         hstack: When ``True`` (the default), decoded columns are horizontally
             stacked with the original input columns.
+        large_int_as_binary: When ``True``, signed and unsigned integers wider
+            than 64 bits (``int128``/``int256``/``uint128``/``uint256``) are
+            emitted as 32-byte big-endian ``Binary`` columns (two's-complement
+            for signed) instead of ``Decimal128``/``Decimal256``. Use this to
+            preserve the full unsigned range of ``uint256`` losslessly.
     """
 
     event_signature: str
@@ -380,6 +386,7 @@ class EvmDecodeEventsConfig:
     input_table: str = "logs"
     output_table: str = "decoded_logs"
     hstack: bool = True
+    large_int_as_binary: bool = False
 
 
 @dataclass
@@ -484,6 +491,25 @@ class U256ToBinaryConfig:
     tables: Optional[list[str]] = None
 
 
+@dataclass
+class LargeIntColumnsToBinaryConfig:
+    """Configuration for the column-scoped large-integer-to-binary step.
+
+    Converts named scale-0 ``Decimal128`` / ``Decimal256`` columns in a single
+    table to fixed-width big-endian two's-complement ``Binary`` (16 bytes and
+    32 bytes respectively). Matches the wire format produced by
+    ``evm_decode_events`` with ``large_int_as_binary=True``.
+
+    Attributes:
+        table_name: The name of the table whose columns should be converted.
+        columns: List of column names to convert. Each must be a scale-0
+            ``Decimal128`` or ``Decimal256`` column.
+    """
+
+    table_name: str
+    columns: list[str]
+
+
 @dataclass
 class Base58EncodeConfig:
     """Configuration for the Base58-encoding step.
@@ -935,6 +961,7 @@ class Step:
         | CastConfig
         | HexEncodeConfig
         | U256ToBinaryConfig
+        | LargeIntColumnsToBinaryConfig
         | CastByTypeConfig
         | Base58EncodeConfig
         | SvmDecodeInstructionsConfig
@@ -1020,6 +1047,7 @@ class Pipeline:
     "CastConfig",
     "HexEncodeConfig",
     "U256ToBinaryConfig",
+    "LargeIntColumnsToBinaryConfig",
     "CastByTypeConfig",
     "Base58EncodeConfig",
     "SvmDecodeInstructionsConfig",

src/tiders/pipeline.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import logging
+import time
 from dataclasses import asdict
 from enum import Enum
 
@@ -44,6 +45,7 @@
     Step,
     StepKind,
     U256ToBinaryConfig,
+    LargeIntColumnsToBinaryConfig,
     SvmDecodeInstructionsConfig,
     SvmDecodeLogsConfig,
 )
@@ -147,6 +149,9 @@ def process_steps(
         elif step.kind == StepKind.U256_TO_BINARY:
             assert isinstance(step.config, U256ToBinaryConfig)
             data = step_def.u256_to_binary.execute(data, step.config)
+        elif step.kind == StepKind.LARGE_INT_COLUMNS_TO_BINARY:
+            assert isinstance(step.config, LargeIntColumnsToBinaryConfig)
+            data = step_def.large_int_columns_to_binary.execute(data, step.config)
         elif step.kind == StepKind.BASE58_ENCODE:
             assert isinstance(step.config, Base58EncodeConfig)
             data = step_def.base58_encode.execute(data, step.config)
@@ -266,6 +271,84 @@ def merge_data(data: list[Dict[str, pa.Table]]) -> Dict[str, pa.Table]:
     return out
 
 
+def _fmt_dur(s: float) -> str:
+    s = int(s)
+    if s < 60:
+        return f"{s}s"
+    if s < 3600:
+        return f"{s // 60}m{s % 60:02d}s"
+    return f"{s // 3600}h{(s % 3600) // 60:02d}m"
+
+
+class _ProgressTracker:
+    """Per-batch progress logger driven by tiders-core stream accessors."""
+
+    def __init__(self, stream, name: str):
+        self.stream = stream
+        self.name = name
+        self.batch = 0
+        self.prev_last = stream.from_block - 1
+        self.t_start = time.monotonic()
+
+    def log_header(self) -> None:
+        to = self.stream.to_block
+        if to is None:
+            logger.info(
+                f"pipeline '{self.name}' starting | "
+                f"from_block {self.stream.from_block:,} | to_block <head> | tailing"
+            )
+        else:
+            logger.info(
+                f"pipeline '{self.name}' starting | "
+                f"from_block {self.stream.from_block:,} | to_block {to:,} | "
+                f"range {to - self.stream.from_block:,} blocks"
+            )
+
+    def log_batch(self, t_ingest: float, t_steps: float, t_write: float) -> None:
+        self.batch += 1
+        last = self.stream.last_block
+        if last is None:
+            return
+        lo, hi = self.prev_last + 1, last
+        self.prev_last = last
+
+        elapsed = time.monotonic() - self.t_start
+        done = last - self.stream.from_block + 1
+        rate = done / elapsed if elapsed > 0 else 0.0
+
+        to = self.stream.to_block
+        if to is None:
+            progress = f"{last:,} (live, no to_block)"
+            eta = ""
+        else:
+            total = max(1, to - self.stream.from_block + 1)
+            pct = 100.0 * done / total
+            progress = f"{last:,}/{to:,} {pct:5.1f}%"
+            remaining = max(0, to - last)
+            eta_s = remaining / rate if rate > 0 else 0.0
+            eta = f" | eta {_fmt_dur(eta_s)}"
+
+        logger.info(
+            f"batch {self.batch:<3} | blocks {lo:,}–{hi:,} ({hi - lo + 1}) "
+            f"| {progress} | {rate:7,.0f} blk/s{eta} "
+            f"| ingest {t_ingest:.2f}s · steps {t_steps:.2f}s · write {t_write:.2f}s"
+        )
+
+    def log_footer(self) -> None:
+        elapsed = time.monotonic() - self.t_start
+        last = (
+            self.stream.last_block
+            if self.stream.last_block is not None
+            else self.stream.from_block
+        )
+        done = last - self.stream.from_block + 1
+        rate = done / elapsed if elapsed > 0 else 0.0
+        logger.info(
+            f"pipeline '{self.name}' done | {done:,} blocks in {_fmt_dur(elapsed)} "
+            f"| avg {rate:,.0f} blk/s | {self.batch} batches"
+        )
+
+
 async def run_pipeline(pipeline: Pipeline, pipeline_name: Optional[str] = None):
     """Execute a full pipeline: ingest, transform, and write data.
 
@@ -316,9 +399,13 @@ async def run_pipeline(pipeline: Pipeline, pipeline_name: Optional[str] = None):
             )
 
     stream = start_stream(pipeline.provider, pipeline.query)
+    progress = _ProgressTracker(stream, pipeline_name or "<unnamed>")
+    progress.log_header()
 
     while True:
+        t0 = time.monotonic()
         data = await stream.next()
+        t_ingest = time.monotonic() - t0
         if data is None:
             break
 
@@ -335,11 +422,19 @@ async def run_pipeline(pipeline: Pipeline, pipeline_name: Optional[str] = None):
         for table_name, table_batch in data.items():
             tables[table_name] = pa.Table.from_batches([table_batch])
 
+        t1 = time.monotonic()
         processed = await asyncio.to_thread(process_steps, tables, pipeline.steps)
+        t_steps = time.monotonic() - t1
 
         logger.debug("Pushing data to writer")
 
+        t2 = time.monotonic()
         await asyncio.gather(*[w.push_data(processed) for w in writers])
+        t_write = time.monotonic() - t2
+
+        progress.log_batch(t_ingest, t_steps, t_write)
+
+    progress.log_footer()
 
 
 __all__ = ["run_pipeline"]

src/tiders/steps/__init__.py

@@ -13,6 +13,7 @@
     cast_by_type,
     base58_encode,
     u256_to_binary,
+    large_int_columns_to_binary,
     svm_decode_instructions,
     svm_decode_logs,
     set_chain_id,
@@ -45,6 +46,7 @@
     "cast_by_type",
     "base58_encode",
     "u256_to_binary",
+    "large_int_columns_to_binary",
     "svm_decode_instructions",
     "svm_decode_logs",
     "set_chain_id",

src/tiders/steps/evm_decode_events.py

@@ -41,10 +41,13 @@ def execute(
                 config.allow_decode_fail,
                 config.filter_by_topic0,
                 config.hstack,
+                config.large_int_as_binary,
             )
         )
 
-    decoded_schema = evm_event_signature_to_arrow_schema(config.event_signature)
+    decoded_schema = evm_event_signature_to_arrow_schema(
+        config.event_signature, config.large_int_as_binary
+    )
     if config.hstack:
         schema = pa.schema(list(decoded_schema) + list(input_table.schema))
     else: