Method read

dataflow/utils/storage.py:353–387

Read data from Myscale/ClickHouse table.

(self, output_type: Literal["dataframe", "dict"])

        self.validate_required_params()

    def read(self, output_type: Literal["dataframe", "dict"]) -> Any:
        """
        Read data from Myscale/ClickHouse table.
        """
        where_clauses = []
        params = {}
        if self.pipeline_id:
            where_clauses.append("pipeline_id = %(pipeline_id)s")
            params['pipeline_id'] = self.pipeline_id
        if self.input_task_id:
            where_clauses.append("task_id = %(task_id)s")
            params['task_id'] = self.input_task_id
        where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
        limit_offset = f"LIMIT {self.page_size} OFFSET {(self.page_num-1)*self.page_size}" if self.page_size else ""
        sql = f"SELECT * FROM {self.table} {where_sql} {limit_offset}"
        self.logger.info(f"Reading from DB: {sql} with params {params}")
        result = self.client.execute(sql, params, with_column_types=True)
        rows, col_types = result
        columns = [col[0] for col in col_types]
        df = pd.DataFrame(rows, columns=columns)
        # 解析 data 字段为 dict
        if 'data' not in df.columns:
            raise ValueError("Result does not contain required 'data' field.")

        # 只保留 data 字段
        data_series = df['data'].apply(safe_json_loads)

        if output_type == "dataframe":
            # 返回只有 data 一列的 DataFrame
            return pd.DataFrame({'data': data_series})
        elif output_type == "dict":
            # 返回 data 字段的 dict 列表
            return list(data_series)
        else:
            raise ValueError(f"Unsupported output type: {output_type}")

    def write(self, data: Any) -> Any:
        """