DWH

SQL queries and Stream Load inserts into StarRocks

The DWH sub-client provides access to the IYREE Data Warehouse powered by StarRocks. Execute SQL queries and bulk-insert data via Stream Load.

from iyree import IyreeClient

with IyreeClient(api_key="my-key") as client:
    result = client.dwh.sql("SELECT * FROM orders LIMIT 10")

---

sql

Execute a SQL query against StarRocks. Uses the StarRocks SQL HTTP API under the hood. The response is streamed as NDJSON and parsed incrementally.

Only SELECT, SHOW, EXPLAIN, and KILL statements are supported. DDL and DML statements (e.g. CREATE, INSERT, UPDATE, DELETE) are not allowed.

result = client.dwh.sql(
    "SELECT status, COUNT(*) AS cnt FROM orders GROUP BY status",
    session_variables={"query_timeout": 60},
)

Parameters

query · str · required

SQL query string to execute.

session_variables · Dict[str, Any]

Optional StarRocks session variables applied for this query.

session_variables={"query_timeout": 60, "parallel_fragment_exec_instance_num": 4}

Returns DwhQueryResult

columns · List[ColumnMeta]

Column metadata from the query result.

Show ColumnMeta

name · str

Column name.

type · str

Column data type as reported by StarRocks.

rows · List[List[Any]]

Data rows. Each row is a list of values in column order.

statistics · Dict[str, Any]

Query execution statistics from StarRocks.

connection_id · int

Connection identifier from StarRocks.

Result methods

to_dicts() · () -> List[Dict[str, Any]]

Convert rows to a list of {column_name: value} dictionaries.

result = client.dwh.sql("SELECT id, name FROM users")
for row in result.to_dicts():
    print(row)  # {"id": 1, "name": "Alice"}

to_dataframe() · () -> pd.DataFrame

Convert to a pandas DataFrame. Requires iyree[pandas].

df = client.dwh.sql("SELECT * FROM orders").to_dataframe()
print(df.head())

Errors

Exception	Condition
IyreeError	Query failure or HTTP error
IyreeAuthError	Invalid API key
IyreeTimeoutError	Request timed out

Examples

Basic query

result = client.dwh.sql("SELECT 1 AS n")
print(result.rows)       # [[1]]
print(result.to_dicts()) # [{"n": 1}]

---

raw_sql

Execute any SQL statement against StarRocks via the streaming /rawSql endpoint. Unlike sql(), this method supports all SQL statement types — including INSERT INTO ... SELECT, CREATE, ALTER, DROP, DELETE, and more.

Use raw_sql() for DML/DDL operations that sql() does not support. For SELECT queries, prefer sql() — it returns richer metadata (column types, statistics, connection ID). For inserting data directly (from your application), prefer insert() which uses the optimized Stream Load path.

Common use cases:

• INSERT INTO ... SELECT — transform and load data between tables
• INSERT INTO ... SELECT * FROM FILES(...) — load data from external files
• DELETE FROM — delete rows by condition
• CREATE TABLE, ALTER TABLE, DROP TABLE — DDL operations

result = client.dwh.raw_sql(
    "INSERT INTO analytics.orders_agg SELECT region, COUNT(*) FROM orders GROUP BY region"
)
print(f"Affected rows: {result.affected_rows}")

Parameters

query · str · required

SQL statement to execute. Any valid StarRocks SQL is accepted.

Returns DwhRawSqlResult

The result adapts to the statement type. For SELECT-like statements, columns and rows are populated. For DML/DDL, affected_rows is populated instead.

columns · List[str]

Column names from the result. Empty for DML/DDL statements.

rows · List[Dict[str, Any]]

Data rows as dictionaries ({column_name: value}). Empty for DML/DDL statements.

row_count · int · default: "0"

Number of rows returned (for SELECT-like statements). 0 for DML/DDL.

affected_rows · int · default: "0"

Number of rows affected (for DML/DDL statements). 0 for SELECT.

Result methods

to_dicts() · () -> List[Dict[str, Any]]

Return rows as a list of {column_name: value} dicts. Since rows are already stored as dicts, this is an identity operation.

result = client.dwh.raw_sql("SHOW TABLES")
for row in result.to_dicts():
    print(row)

to_dataframe() · () -> pd.DataFrame

Convert to a pandas DataFrame. Requires iyree[pandas].

df = client.dwh.raw_sql("SHOW TABLES").to_dataframe()

Errors

Exception	Condition
IyreeError	Query failure or mid-stream error
IyreeAuthError	Invalid API key
IyreeTimeoutError	Request timed out

Examples

INSERT INTO SELECT

result = client.dwh.raw_sql("""
    INSERT INTO analytics.daily_revenue
    SELECT DATE(created_at) AS day, SUM(amount) AS revenue
    FROM orders
    WHERE created_at >= '2024-01-01'
    GROUP BY DATE(created_at)
""")
print(f"Affected rows: {result.affected_rows}")

---

insert

Insert data into a StarRocks table via Stream Load. Supports CSV strings, raw bytes, JSON lists, and pandas DataFrames.

result = client.dwh.insert("staging_table", "col1,col2\n1,hello\n2,world")
print(f"Loaded {result.number_loaded_rows} rows")

Parameters

table · str · required

Target table name in StarRocks.

data · str | bytes | List[dict] | pd.DataFrame · required

Payload to insert. The SDK auto-detects the type and adjusts the format accordingly:

• str — encoded as UTF-8, sent as-is
• bytes — sent as-is
• List[dict] — serialized to JSON, format switched to json, strip_outer_array enabled
• pd.DataFrame — exported to CSV (headerless), column names extracted automatically

format · str · default: "\"csv\""

Data format: "csv" or "json". Automatically overridden to "json" when data is a list.

label · str

Idempotency label. When provided, re-sending the same label is safe — StarRocks deduplicates.

columns · List[str]

Column mapping for CSV data. Automatically populated from DataFrame column names when inserting a DataFrame.

column_separator · str · default: "\",\""

Column separator for CSV format.

timeout · float

Request timeout override in seconds. Defaults to stream_load_timeout from configuration.

**stream_load_headers · str

Extra StarRocks Stream Load headers passed as keyword arguments (e.g. max_filter_ratio="0.1").

Returns StreamLoadResult

txn_id · int

StarRocks transaction ID.

label · str

Label used for the load operation.

status · str

Operation status: Success, Fail, Publish Timeout, Label Already Exists, etc.

message · str

Human-readable status message.

number_total_rows · int

Total rows processed.

number_loaded_rows · int

Rows successfully loaded.

number_filtered_rows · int

Rows filtered out.

number_unselected_rows · int

Rows not selected.

load_bytes · int

Bytes loaded.

load_time_ms · int

Load duration in milliseconds.

error_url · str

URL to fetch detailed error info when rows are filtered. None if no errors.

Errors

Exception	Condition
IyreeStreamLoadError	Stream Load status is Fail
IyreeDuplicateLabelError	Label already exists (extends IyreeStreamLoadError)
IyreeTimeoutError	Upload timed out

Examples

CSV string

result = client.dwh.insert(
    "users",
    "1,Alice\n2,Bob",
    columns=["id", "name"],
    label="load_users_001",
)
print(f"Loaded {result.number_loaded_rows} rows in {result.load_time_ms}ms")