diff --git a/docker_clickhouse/.env.example b/docker_clickhouse/.env.example index 98fb78f..02b33b7 100644 --- a/docker_clickhouse/.env.example +++ b/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 \ No newline at end of file diff --git a/docker_postgres/.env.example b/docker_postgres/.env.example index 97b199e..cd103ff 100644 --- a/docker_postgres/.env.example +++ b/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 \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 6f00197..ff75a8b 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/tiders/cli/tiders_yaml_parser.py b/src/tiders/cli/tiders_yaml_parser.py index 2b1c3ca..d138e84 100644 --- a/src/tiders/cli/tiders_yaml_parser.py +++ b/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 diff --git a/src/tiders/config.py b/src/tiders/config.py index 36a5c7b..21f4e6f 100644 --- a/src/tiders/config.py +++ b/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", diff --git a/src/tiders/pipeline.py b/src/tiders/pipeline.py index cc02bf3..c495702 100644 --- a/src/tiders/pipeline.py +++ b/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 | 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 "") + 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"] diff --git a/src/tiders/steps/__init__.py b/src/tiders/steps/__init__.py index 307d81d..62d67f4 100644 --- a/src/tiders/steps/__init__.py +++ b/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", diff --git a/src/tiders/steps/evm_decode_events.py b/src/tiders/steps/evm_decode_events.py index b23e579..de6070c 100644 --- a/src/tiders/steps/evm_decode_events.py +++ b/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: diff --git a/src/tiders/steps/large_int_columns_to_binary.py b/src/tiders/steps/large_int_columns_to_binary.py new file mode 100644 index 0000000..5755473 --- /dev/null +++ b/src/tiders/steps/large_int_columns_to_binary.py @@ -0,0 +1,79 @@ +"""Column-scoped large-integer-to-binary conversion 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) via the Rust-backed ``tiders_core.large_ints_to_binary`` +function. Matches the wire format produced by ``evm_decode_events`` with +``large_int_as_binary=True``. +""" + +from typing import Dict +from copy import deepcopy + +import pyarrow as pa +from tiders_core import large_ints_to_binary + +from tiders.config import LargeIntColumnsToBinaryConfig + + +def execute( + data: Dict[str, pa.Table], config: LargeIntColumnsToBinaryConfig +) -> Dict[str, pa.Table]: + """Convert specific large-integer columns to binary in a single table. + + Only the table matching ``config.table_name`` is affected. Each column + listed in ``config.columns`` must be a scale-0 ``Decimal128`` or + ``Decimal256``; it is converted to fixed-width big-endian two's-complement + ``Binary``. Other columns are left untouched. + + The conversion is delegated to ``tiders_core.large_ints_to_binary`` by + building a sub-batch of just the selected columns, casting it, and + splicing the results back into the original batch. + + Args: + data: A dictionary mapping table names to PyArrow Tables. + config: A :class:`LargeIntColumnsToBinaryConfig` specifying the target + table and the columns to convert. + + Returns: + A new data dictionary with the converted columns applied. + """ + data = deepcopy(data) + + table = data.get(config.table_name) + if table is None: + return data + + columns = list(config.columns) + if not columns: + return data + + out_batches = [] + + for batch in table.to_batches(): + sub_batch = batch.select(columns) + converted = large_ints_to_binary(sub_batch) + + converted_by_name = { + name: converted.column(i) for i, name in enumerate(converted.schema.names) + } + converted_fields = { + name: converted.schema.field(i) + for i, name in enumerate(converted.schema.names) + } + + arrays = [] + fields = [] + for i, name in enumerate(batch.schema.names): + if name in converted_by_name: + arrays.append(converted_by_name[name]) + fields.append(converted_fields[name]) + else: + arrays.append(batch.column(i)) + fields.append(batch.schema.field(i)) + + out_batches.append(pa.RecordBatch.from_arrays(arrays, schema=pa.schema(fields))) + + data[config.table_name] = pa.Table.from_batches(out_batches) + + return data diff --git a/src/tiders/writers/clickhouse.py b/src/tiders/writers/clickhouse.py index 7b473c1..eb671cb 100644 --- a/src/tiders/writers/clickhouse.py +++ b/src/tiders/writers/clickhouse.py @@ -117,8 +117,10 @@ class Writer(DataWriter): def __init__(self, config: ClickHouseWriterConfig): if config.client is not None: self.client = config.client + self.database = config.client.database else: self.client = None + self.database = config.database self._client_config = { "host": config.host, "port": config.port, @@ -155,7 +157,7 @@ async def _check_table_exists(self, table_name: str) -> bool: """Return ``True`` if a table with the given name exists in the current database.""" assert self.client is not None res = await self.client.query( - f"SELECT count() > 0 as table_exists FROM system.tables WHERE database = '{self.client.client.database}' AND name = '{table_name}'" + f"SELECT count() > 0 as table_exists FROM system.tables WHERE database = '{self.database}' AND name = '{table_name}'" ) return bool(res.result_rows[0][0])