mz_catalog Schema

-- List all databases and their owners
SELECT 
  d.name AS database_name,
  r.name AS owner_name
FROM mz_catalog.mz_databases d
JOIN mz_catalog.mz_roles r ON d.owner_id = r.id;

-- List all schemas in the current database
SELECT 
  s.name AS schema_name,
  d.name AS database_name
FROM mz_catalog.mz_schemas s
JOIN mz_catalog.mz_databases d ON s.database_id = d.id;

-- List all objects in the public schema
SELECT name, type 
FROM mz_catalog.mz_objects
WHERE schema_id = (SELECT id FROM mz_catalog.mz_schemas WHERE name = 'public')
ORDER BY type, name;

-- List all Kafka sources with their topics
SELECT 
  s.name AS source_name,
  s.type,
  ks.topic,
  c.name AS cluster_name
FROM mz_catalog.mz_sources s
JOIN mz_catalog.mz_kafka_sources ks ON s.id = ks.id
JOIN mz_catalog.mz_clusters c ON s.cluster_id = c.id
WHERE s.type = 'kafka';

-- Monitor Kafka source lag
SELECT
  s.name,
  ks.topic,
  st.offset_known - st.offset_committed AS lag
FROM mz_catalog.mz_sources s
JOIN mz_catalog.mz_kafka_sources ks ON s.id = ks.id
JOIN mz_internal.mz_source_statistics st ON s.id = st.id
WHERE st.snapshot_committed;

-- Find views that query a specific source
SELECT v.name, v.definition
FROM mz_catalog.mz_views v
WHERE v.definition LIKE '%my_source%';

-- List materialized views with their storage usage
SELECT 
  mv.name,
  c.name AS cluster_name,
  u.size_bytes / (1024 * 1024) AS size_mb
FROM mz_catalog.mz_materialized_views mv
JOIN mz_catalog.mz_clusters c ON mv.cluster_id = c.id
JOIN mz_catalog.mz_recent_storage_usage u ON mv.id = u.object_id
ORDER BY u.size_bytes DESC;

-- List all indexes and what they index
SELECT 
  i.name AS index_name,
  o.name AS indexed_object,
  o.type AS object_type,
  c.name AS cluster_name
FROM mz_catalog.mz_indexes i
JOIN mz_catalog.mz_objects o ON i.on_id = o.id
JOIN mz_catalog.mz_clusters c ON i.cluster_id = c.id;

-- Show index structure
SELECT 
  i.name AS index_name,
  ic.index_position,
  COALESCE(
    c.name,
    ic.on_expression
  ) AS column_def
FROM mz_catalog.mz_indexes i
JOIN mz_catalog.mz_index_columns ic ON i.id = ic.index_id
LEFT JOIN mz_catalog.mz_columns c 
  ON i.on_id = c.id 
  AND ic.on_position = c.position
WHERE i.name = 'my_index'
ORDER BY ic.index_position;

-- List all clusters with their configuration
SELECT 
  name,
  managed,
  size,
  replication_factor
FROM mz_catalog.mz_clusters
ORDER BY name;

-- Show cluster topology
SELECT 
  c.name AS cluster,
  r.name AS replica,
  r.size,
  r.availability_zone
FROM mz_catalog.mz_clusters c
JOIN mz_catalog.mz_cluster_replicas r ON c.id = r.cluster_id
ORDER BY c.name, r.name;

-- Calculate cluster costs
SELECT 
  c.name AS cluster,
  c.replication_factor,
  s.size,
  s.credits_per_hour,
  s.credits_per_hour * c.replication_factor AS total_credits_per_hour
FROM mz_catalog.mz_clusters c
JOIN mz_catalog.mz_cluster_replica_sizes s ON c.size = s.size
WHERE c.managed;

-- List all sinks and their destinations
SELECT 
  s.name AS sink_name,
  ks.topic AS kafka_topic,
  c.name AS cluster_name
FROM mz_catalog.mz_sinks s
JOIN mz_catalog.mz_kafka_sinks ks ON s.id = ks.id
JOIN mz_catalog.mz_clusters c ON s.cluster_id = c.id;

-- Show table schema
SELECT 
  c.name AS column_name,
  c.type AS data_type,
  c.nullable,
  c.default
FROM mz_catalog.mz_columns c
JOIN mz_catalog.mz_tables t ON c.id = t.id
WHERE t.name = 'my_table'
ORDER BY c.position;

-- List all users who can log in
SELECT name, rolsuper
FROM mz_catalog.mz_roles
WHERE rolcanlogin = true
ORDER BY name;

-- Show role hierarchy
SELECT 
  parent.name AS role,
  member.name AS member,
  grantor.name AS granted_by
FROM mz_catalog.mz_role_members rm
JOIN mz_catalog.mz_roles parent ON rm.role_id = parent.id
JOIN mz_catalog.mz_roles member ON rm.member = member.id
JOIN mz_catalog.mz_roles grantor ON rm.grantor = grantor.id;

-- Show recent DDL changes
SELECT 
  occurred_at,
  event_type,
  object_type,
  user,
  details->>'name' AS object_name
FROM mz_catalog.mz_audit_events
WHERE object_type IN ('source', 'view', 'materialized-view')
ORDER BY occurred_at DESC
LIMIT 50;

-- Top 10 objects by storage usage
SELECT 
  o.name,
  o.type,
  u.size_bytes / (1024 * 1024 * 1024.0) AS size_gb
FROM mz_catalog.mz_recent_storage_usage u
JOIN mz_catalog.mz_objects o ON u.object_id = o.id
ORDER BY u.size_bytes DESC
LIMIT 10;

-- Track storage growth over time for a specific object
SELECT 
  collection_timestamp,
  size_bytes / (1024 * 1024 * 1024.0) AS size_gb
FROM mz_catalog.mz_storage_usage
WHERE object_id = (SELECT id FROM mz_catalog.mz_materialized_views WHERE name = 'my_mv')
ORDER BY collection_timestamp DESC
LIMIT 24; -- Last 24 hours

-- Check frontier alignment across replicas
SELECT 
  o.name AS object,
  r.name AS replica,
  f.write_frontier
FROM mz_catalog.mz_cluster_replica_frontiers f
JOIN mz_catalog.mz_objects o ON f.object_id = o.id
JOIN mz_catalog.mz_cluster_replicas r ON f.replica_id = r.id
WHERE o.name = 'my_materialized_view';

SELECT
  o.name,
  s.snapshot_records_staged,
  s.snapshot_records_known,
  ROUND(
    100.0 * s.snapshot_records_staged / 
    NULLIF(s.snapshot_records_known, 0), 
    2
  ) AS snapshot_completed_pct
FROM mz_internal.mz_source_statistics s
JOIN mz_catalog.mz_objects o ON s.id = o.id
WHERE NOT s.snapshot_committed;

-- Indexes that might not be needed
SELECT 
  i.name AS index_name,
  o.name AS indexed_object
FROM mz_catalog.mz_indexes i
JOIN mz_catalog.mz_objects o ON i.on_id = o.id
WHERE NOT EXISTS (
  SELECT 1 
  FROM mz_internal.mz_compute_dependencies d
  WHERE d.dependency_id = i.id
);

SELECT 
  c.name AS cluster,
  COUNT(DISTINCT r.id) AS replica_count,
  s.size,
  s.cpu_nano_cores / 1000000000.0 AS cpu_cores,
  s.memory_bytes / (1024.0 * 1024 * 1024) AS memory_gb,
  s.credits_per_hour * COUNT(DISTINCT r.id) AS total_credits_per_hour
FROM mz_catalog.mz_clusters c
JOIN mz_catalog.mz_cluster_replicas r ON c.id = r.cluster_id
JOIN mz_catalog.mz_cluster_replica_sizes s ON r.size = s.size
GROUP BY c.name, s.size, s.cpu_nano_cores, s.memory_bytes, s.credits_per_hour
ORDER BY total_credits_per_hour DESC;