Sources

-- Create a Kafka connection
CREATE SECRET kafka_password AS '...';

CREATE CONNECTION kafka_conn TO KAFKA (
    BROKER 'broker.kafka.example.com:9092',
    SASL MECHANISMS = 'PLAIN',
    SASL USERNAME = 'user',
    SASL PASSWORD = SECRET kafka_password
);

-- Create a source with Avro format
CREATE SOURCE events
FROM KAFKA CONNECTION kafka_conn (TOPIC 'events')
FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn;

-- Create a PostgreSQL connection
CREATE SECRET pg_password AS '...';

CREATE CONNECTION pg_conn TO POSTGRES (
    HOST 'postgres.example.com',
    PORT 5432,
    DATABASE 'production',
    USER 'materialize',
    PASSWORD SECRET pg_password
);

-- Create a source for all tables in a publication
CREATE SOURCE pg_source
FROM POSTGRES CONNECTION pg_conn
(PUBLICATION 'mz_source')
FOR ALL TABLES;

-- Create a MySQL connection
CREATE SECRET mysql_password AS '...';

CREATE CONNECTION mysql_conn TO MYSQL (
    HOST 'mysql.example.com',
    PORT 3306,
    USER 'materialize',
    PASSWORD SECRET mysql_password
);

-- Create a source for specific schemas and tables
CREATE SOURCE mysql_source
FROM MYSQL CONNECTION mysql_conn
FOR SCHEMAS (production, staging);

# MySQL server configuration
gtid_mode = ON
enforce_gtid_consistency = ON
binlog_format = ROW
binlog_row_image = FULL

CREATE CONNECTION sqlserver_conn TO SQL SERVER (
    HOST 'sqlserver.example.com',
    PORT 1433,
    DATABASE 'production',
    USER 'materialize',
    PASSWORD SECRET sqlserver_password
);

CREATE SOURCE sqlserver_source
FROM SQL SERVER CONNECTION sqlserver_conn
FOR TABLES (dbo.orders, dbo.customers);

-- Create a webhook source
CREATE SOURCE webhook_events FROM WEBHOOK
    BODY FORMAT JSON;

-- Get the webhook URL
SELECT url FROM mz_internal.mz_webhook_sources
WHERE name = 'webhook_events';

https://<HOST>/api/webhook/<database>/<schema>/<source_name>

CREATE SOURCE webhook_source FROM WEBHOOK
    BODY FORMAT JSON
    INCLUDE HEADER 'x-event-type' AS event_type
    INCLUDE HEADER 'timestamp' AS event_timestamp;

(data, time, diff)

{
  "before": {"id": 1, "name": "Alice", "count": 5},
  "after": {"id": 1, "name": "Alice", "count": 10}
}

({id: 1, name: "Alice", count: 5}, T, -1)  // Delete old value
({id: 1, name: "Alice", count: 10}, T, +1) // Insert new value

CREATE SOURCE user_profiles
FROM KAFKA CONNECTION kafka_conn (TOPIC 'users')
FORMAT JSON
ENVELOPE UPSERT;

CREATE SOURCE pg_source
FROM POSTGRES CONNECTION pg_conn
(PUBLICATION 'mz_source')
FOR ALL TABLES;

-- List all subsources
SHOW SOURCES;

         name         |   type
----------------------+-----------
 pg_source            | postgres
 pg_source_orders     | subsource
 pg_source_customers  | subsource
 pg_source_products   | subsource

SELECT * FROM pg_source_orders;

CREATE MATERIALIZED VIEW order_summary AS
SELECT 
    c.customer_name,
    COUNT(*) as order_count,
    SUM(o.total) as total_spent
FROM pg_source_orders o
JOIN pg_source_customers c ON o.customer_id = c.id
GROUP BY c.customer_name;

-- Create a dedicated cluster for sources
CREATE CLUSTER source_cluster SIZE = '100cc';

-- Create source in that cluster
CREATE SOURCE kafka_source
IN CLUSTER source_cluster
FROM KAFKA CONNECTION kafka_conn (TOPIC 'events')
FORMAT JSON;

SELECT id, replication_slot 
FROM mz_internal.mz_postgres_sources;

-- Check binlog retention (in seconds)
SHOW VARIABLES LIKE 'binlog_expire_logs_seconds';

-- Set to 7 days
SET GLOBAL binlog_expire_logs_seconds = 604800;

-- Check source status
SELECT id, name, type, size
FROM mz_sources;

-- Monitor replication lag
SELECT object_id, lag
FROM mz_internal.mz_wallclock_global_lag
WHERE object_id IN (SELECT id FROM mz_sources);

-- View source errors
SELECT * FROM mz_internal.mz_source_statuses
WHERE status = 'error';