Reading Data from Streams

-- Stream all events as they arrive
SELECT * FROM events;

-- Select specific columns
SELECT event_id, user_id, event_type 
FROM events;

-- Query all historical data
SELECT * FROM table(events);

-- Query with ordering
SELECT * FROM table(events)
ORDER BY event_time DESC
LIMIT 100;

-- Filter by column value
SELECT * FROM transactions
WHERE amount > 1000;

-- Multiple conditions
SELECT * FROM events
WHERE event_type = 'purchase' 
  AND user_id > 100;

-- String matching
SELECT * FROM logs
WHERE message LIKE '%error%';

-- Events from last hour
SELECT * FROM events
WHERE _tp_time > now() - INTERVAL 1 HOUR;

-- Specific date range (historical)
SELECT * FROM table(events)
WHERE _tp_time >= '2024-01-01'
  AND _tp_time < '2024-02-01';

-- Events from today
SELECT * FROM table(events)
WHERE to_date(_tp_time) = today();

-- Select specific columns
SELECT user_id, event_type, event_time
FROM events;

-- Column aliases
SELECT 
  user_id as uid,
  event_type as type,
  event_time as ts
FROM events;

-- Exclude columns with EXCEPT
SELECT * EXCEPT (_tp_time, _tp_sn)
FROM events;

-- Apply functions
SELECT 
  upper(event_type) as event_type,
  amount * 1.1 as amount_with_tax,
  to_string(event_time) as time_str
FROM events;

-- Conditional expressions
SELECT 
  order_id,
  CASE 
    WHEN amount < 100 THEN 'small'
    WHEN amount < 1000 THEN 'medium'
    ELSE 'large'
  END as order_size
FROM orders;

-- Start from earliest available data
SELECT * FROM events
SETTINGS seek_to = 'earliest';

-- Start from latest (default)
SELECT * FROM events
SETTINGS seek_to = 'latest';

-- Specify timestamp
SELECT * FROM events
SETTINGS seek_to = '2024-01-01 00:00:00';

-- Limit parallel processing
SELECT * FROM events
SETTINGS max_threads = 4;

-- Memory limits
SELECT * FROM events
SETTINGS max_memory_usage = 10000000000;  -- 10GB

-- This runs continuously
SELECT device_id, temperature 
FROM sensor_readings
WHERE temperature > 80;

-- This processes bounded data and terminates
SELECT device_id, avg(temperature) as avg_temp
FROM table(sensor_readings)
GROUP BY device_id
ORDER BY avg_temp DESC;

-- Subquery for filtering
SELECT * FROM events
WHERE user_id IN (
  SELECT user_id FROM table(premium_users)
);

-- Derived tables
SELECT 
  date,
  avg(daily_total) as avg_daily
FROM (
  SELECT 
    to_date(_tp_time) as date,
    sum(amount) as daily_total
  FROM table(transactions)
  GROUP BY date
)
GROUP BY date;

WITH high_value_users AS (
  SELECT user_id, sum(amount) as total
  FROM table(transactions)
  WHERE _tp_time > now() - INTERVAL 30 DAY
  GROUP BY user_id
  HAVING total > 10000
)
SELECT 
  e.event_type,
  count() as event_count
FROM events e
INNER JOIN high_value_users h ON e.user_id = h.user_id
GROUP BY e.event_type;

-- Limit results in historical query
SELECT * FROM table(events)
ORDER BY _tp_time DESC
LIMIT 100;

-- Process first 1000 events then stop
SELECT * FROM events
LIMIT 1000;

-- Limit per time window
SELECT count(*) as cnt
FROM events
LIMIT 5 EMIT PERIODIC INTERVAL 1 SECOND;

-- Distinct event types (historical)
SELECT DISTINCT event_type 
FROM table(events);

-- Count distinct users
SELECT count(DISTINCT user_id) as unique_users
FROM table(events)
WHERE to_date(_tp_time) = today();

-- Parse JSON string column
SELECT 
  event_id,
  json_extract_string(metadata, 'user_agent') as user_agent,
  json_extract_int(metadata, 'session_duration') as duration
FROM events;

-- JSON path expressions
SELECT 
  json_extract(payload, '$.user.id') as user_id,
  json_extract(payload, '$.items[0].price') as first_item_price
FROM orders;

-- Monitor high-value transactions in real-time
SELECT 
  transaction_id,
  user_id,
  amount,
  currency,
  created_at
FROM transactions
WHERE amount > 10000
  AND status = 'completed';

-- Analyze yesterday's user activity
SELECT 
  event_type,
  count() as event_count,
  count(DISTINCT user_id) as unique_users,
  avg(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) as purchase_rate
FROM table(events)
WHERE to_date(_tp_time) = today() - 1
GROUP BY event_type
ORDER BY event_count DESC;

-- Stream only anomalous sensor readings
SELECT 
  sensor_id,
  temperature,
  humidity,
  reading_time,
  CASE
    WHEN temperature > 100 THEN 'temp_high'
    WHEN temperature < -20 THEN 'temp_low'
    WHEN humidity > 95 THEN 'humidity_high'
    ELSE 'normal'
  END as alert_type
FROM iot_sensors
WHERE temperature > 100 
   OR temperature < -20 
   OR humidity > 95
SETTINGS seek_to = 'earliest';

-- Get latest value per device in last 5 minutes
SELECT 
  device_id,
  any_last(temperature) as latest_temp,
  max(_tp_time) as last_reading
FROM table(sensor_readings)
WHERE _tp_time > now() - INTERVAL 5 MINUTE
GROUP BY device_id;