How-To Read/Write Without Catalog v1.3

-- Setup PGD configuration for efficient testing cycles
SET bdr.taskmgr_nap_time = 10000;  -- AutoPartition worker runs every 10s vs default 290s
SET pgaa.max_replication_lag_s = 5;  -- Analytics replication lag threshold
SET pgaa.flush_task_interval_s = 5;  -- Frequent storage flush for testing

-- Initialize PGAA extension
CREATE EXTENSION IF NOT EXISTS pgaa CASCADE;

-- Create read-only storage location for existing Delta Lake data
SELECT bdr.replicate_ddl_command($$
  SELECT pgfs.create_storage_location(
    'biganimal-sample-data',
    's3://beacon-analytics-demo-data-us-east-1-prod',
    '{"aws_skip_signature": "true"}'
  );
$$);

-- Create development storage location for Iceberg tables
SELECT bdr.replicate_ddl_command($$
  SELECT pgfs.create_storage_location(
    'biganimal-sample-data-dev',
    's3://beacon-analytics-demo-data-us-east-1-dev',
    '{"aws_skip_signature": "true"}'
  );
$$);

-- Create appliance storage location with dynamic path construction
SELECT bdr.replicate_ddl_command($$
  SELECT pgfs.create_storage_location(
    'appliance-bucket',
    '$BUCKET_URL' || '/' ||
    (SELECT node_group_name FROM bdr.node_group WHERE node_group_name != 'world' LIMIT 1) ||
    '/analytics-offload'
  )
$$);

-- Validate storage location accessibility
CREATE TABLE sentinel_sl_read_test () USING PGAA
WITH (pgaa.format = 'iceberg', pgaa.storage_location = 'appliance-bucket', pgaa.path = 'nonexistent');
-- Should return ERROR: No valid Iceberg metadata file found (confirms bucket access)

-- Set as primary analytics write location
SELECT bdr.alter_node_group_option(
  (SELECT node_group_name FROM bdr.node_group WHERE node_group_name != 'world' LIMIT 1),
  'analytics_storage_location',
  'appliance-bucket'
);

-- Verify Seafowl query engine is running
SELECT application_name FROM pg_stat_activity WHERE application_name LIKE '%seafowl%';
-- Must show 'pgd seafowl_manager' for proper operation

-- Analyze storage statistics for Delta Lake tables
SELECT * FROM pgaa.lakehouse_table_stats('lineitem'::regclass);

-- Analyze storage statistics for Iceberg tables
SELECT * FROM pgaa.lakehouse_table_stats('iceberg_table'::regclass);

-- Create transactional table for offload testing
CREATE TABLE transactional_table (
  key BIGINT,
  inserted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  revenue DOUBLE PRECISION,
  comment TEXT,
  PRIMARY KEY (key, inserted_at)
);

-- Insert test data with temporal distribution
INSERT INTO transactional_table (
  SELECT
    g,
    '2010-01-01 00:00:00'::timestamp + INTERVAL '1 second' *
      (EXTRACT (EPOCH FROM (CURRENT_TIMESTAMP - '2010-01-01 00:00:00'::timestamp)) / 1000000 * g),
    g * 0.001,
    encode(sha256(g::text::bytea), 'hex')
  FROM generate_series(1, 1000000) g
);

-- Enable analytics replication for transactional table
ALTER TABLE transactional_table SET (pgd.replicate_to_analytics = TRUE);

-- Monitor replication progress
SELECT * FROM bdr.analytics_table;
-- Wait until table appears in results (~120s for initial sync)

-- Create reader interface for analytics data
CREATE TABLE transactional_table_analytics_reader ()
USING PGAA
WITH (
  pgaa.format = 'iceberg',
  pgaa.storage_location = 'appliance-bucket',
  pgaa.path = 'public.transactional_table'
);

-- Verify data integrity across storage tiers
SELECT COUNT(*) AS count, ROUND(SUM(revenue)) AS total_revenue
FROM transactional_table_analytics_reader;
-- Should return: count=1000000, total_revenue=500000500

-- Convert to analytics-only access
ALTER TABLE transactional_table
SET ACCESS METHOD PGAA,
SET (
  pgaa.auto_truncate = 'true',
  pgaa.format = 'iceberg',
  pgaa.storage_location = 'appliance-bucket',
  pgaa.path = 'public.transactional_table'
);

-- Verify conversion maintained data access
SELECT COUNT(*) AS count, ROUND(SUM(revenue)) AS total_revenue FROM transactional_table;

-- Restore table to transactional storage
ALTER TABLE transactional_table SET ACCESS METHOD heap, SET (pgd.replicate_to_analytics = FALSE);

-- Verify restoration with query plan analysis
EXPLAIN SELECT COUNT(*) AS count, ROUND(SUM(revenue)) AS total_revenue FROM transactional_table;
-- Should show Parallel Seq Scan indicating local heap access

-- Update data in transactional table
UPDATE transactional_table SET revenue = revenue * 2;

-- Re-enable analytics replication to propagate changes
ALTER TABLE transactional_table SET (pgd.replicate_to_analytics = TRUE);

-- Wait for synchronization (~120s) then verify changes propagated
SELECT COUNT(*) AS count, ROUND(SUM(revenue)) AS total_revenue
FROM transactional_table_analytics_reader;
-- Should show doubled revenue: total_revenue=1000001000

-- Perform partial offload with consistent transaction boundary
BEGIN;
CREATE TABLE transactional_table_before_2020 AS (
  SELECT * FROM transactional_table
  WHERE DATE_PART('year', inserted_at) < 2020
);
DELETE FROM transactional_table WHERE DATE_PART('year', inserted_at) < 2020;
COMMIT;

-- Configure partial table for analytics offload
ALTER TABLE transactional_table_before_2020 ADD PRIMARY KEY (key, inserted_at);
ALTER TABLE transactional_table_before_2020 SET (pgd.replicate_to_analytics = TRUE);

-- Convert to analytics-only access
ALTER TABLE transactional_table_before_2020
SET ACCESS METHOD PGAA,
SET (
  pgaa.auto_truncate = 'true',
  pgaa.format = 'iceberg',
  pgaa.storage_location = 'appliance-bucket',
  pgaa.path = 'public.transactional_table_before_2020'
);

-- Verify temporal data distribution
SELECT
  COUNT(*) FILTER (WHERE inserted_at < '2020-01-01') AS before_2020,
  COUNT(*) FILTER (WHERE inserted_at >= '2020-01-01') AS after_2020
FROM transactional_table_before_2020;

-- Create partitioned table structure
CREATE TABLE partitioned_table (
  key BIGINT,
  inserted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  revenue DOUBLE PRECISION,
  comment TEXT,
  PRIMARY KEY (key, inserted_at)
) PARTITION BY RANGE(inserted_at);

-- Configure AutoPartition without immediate offload
SELECT bdr.autopartition (
  relation := 'partitioned_table',
  partition_increment := '1 year',
  partition_initial_lowerbound := '2010-01-01 00:00:00',
  managed_locally := TRUE
);

-- Wait for partition creation across cluster
SELECT bdr.autopartition_wait_for_partitions_on_all_nodes('partitioned_table'::regclass, CURRENT_TIMESTAMP::text);

-- Load test data across temporal partitions
INSERT INTO partitioned_table (
  SELECT
    g,
    '2010-01-01 00:00:00'::timestamp + INTERVAL '1 second' *
      (EXTRACT (EPOCH FROM (CURRENT_TIMESTAMP - '2010-01-01 00:00:00'::timestamp)) / 1000000 * g),
    g * 0.001,
    encode(sha256(g::text::bytea), 'hex')
  FROM generate_series(1, 1000000) g
);

-- Create optimized row counting function
CREATE OR REPLACE FUNCTION temp_get_table_rowcount(schema_name text, table_name text)
RETURNS bigint AS $$
DECLARE count bigint;
BEGIN
    -- Uses DirectScan for PGAA tables (1.8s vs 20s without optimization)
    EXECUTE FORMAT('SELECT COUNT(*) FROM %I.%I', schema_name, table_name) INTO count;
    RETURN count;
END $$ LANGUAGE plpgsql;

-- Comprehensive partition storage analysis
WITH partitioned_table_info AS (
  SELECT
    nspc.nspname AS schema_name,
    child.relname AS table_name,
    pg_catalog.pg_get_expr(child.relpartbound, child.oid) AS partition_constraint,
    pg_get_partition_constraintdef(child.oid) AS partition_constraint_full
  FROM pg_inherits inh
    JOIN pg_class parent ON inh.inhparent = parent.oid
    JOIN pg_class child ON inh.inhrelid = child.oid
    JOIN pg_namespace nspp ON parent.relnamespace = nspp.oid
    JOIN pg_namespace nspc ON child.relnamespace = nspc.oid
  WHERE parent.relname = 'partitioned_table' AND nspp.nspname = 'public'
),

pgaa_tables AS (
  SELECT
    m.schema_name, m.table_name, m.format,
    CASE
      WHEN m.storage_location_name IS NOT NULL THEN FORMAT('pgfs://%s/%s', m.storage_location_name, m.path)
      WHEN (m.managed_by IS NOT NULL AND m.catalog_namespace IS NOT NULL AND m.catalog_table IS NOT NULL)
        THEN FORMAT('%s.%s.%s', m.managed_by, m.catalog_namespace, m.catalog_table)
    END AS path,
    temp_get_table_rowcount(m.schema_name, m.table_name) AS num_rows,
    s.total_size AS total_size,
    pti.partition_constraint
  FROM pgaa.get_all_analytics_table_settings() m
  LEFT JOIN LATERAL pgaa.lakehouse_table_stats(FORMAT('%I.%I', m.schema_name, m.table_name)::regclass) s ON TRUE
  JOIN partitioned_table_info pti ON m.schema_name = pti.schema_name AND m.table_name = pti.table_name
  WHERE m.storage_location_name IS NOT NULL OR (m.managed_by IS NOT NULL AND m.catalog_namespace IS NOT NULL AND m.catalog_table IS NOT NULL)
),

local_tables AS (
  SELECT
    pn.nspname AS schema_name, pc.relname AS table_name,
    temp_get_table_rowcount(pn.nspname, pc.relname) AS num_rows,
    pg_total_relation_size(FORMAT('%I.%I', pn.nspname, pc.relname)::regclass) AS total_size,
    pti.partition_constraint
  FROM pg_class pc
    JOIN pg_namespace pn ON pc.relnamespace = pn.oid
    JOIN pg_am am ON am.oid = pc.relam
    JOIN partitioned_table_info pti ON pn.nspname = pti.schema_name AND pc.relname = pti.table_name
  WHERE pn.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast', 'pg_internal', 'pgaa', 'pgfs', 'bdr')
    AND pc.relkind = 'r' AND am.amname != 'pgaa'
),

all_leaf_tables AS (
  SELECT schema_name, table_name, format AS pgaa_format, path AS pgaa_path, num_rows, total_size, partition_constraint FROM pgaa_tables
  UNION
  SELECT schema_name, table_name, NULL AS pgaa_format, NULL AS pgaa_path, num_rows, total_size, partition_constraint FROM local_tables
),

final_table_info AS (
  SELECT
    schema_name, table_name,
    CASE
      WHEN pgaa_format = 'iceberg' THEN 'iceberg'
      WHEN pgaa_format = 'delta' THEN 'delta'
      ELSE 'heap'
    END AS type,
    COALESCE(pgaa_path, 'local') AS location,
    num_rows, total_size AS total_size_bytes,
    SUBSTRING(partition_constraint, 'FROM \(''([^'']+)') AS start_bound,
    SUBSTRING(partition_constraint, 'TO \(''([^'']+)') AS end_bound
  FROM all_leaf_tables
  ORDER BY partition_constraint ASC
)

SELECT * FROM final_table_info;

-- Configure analytics offload for partitions older than one year
SELECT bdr.autopartition (
  relation := 'partitioned_table',
  partition_increment := '1 year',
  partition_initial_lowerbound := '2010-01-01 00:00:00',
  managed_locally := TRUE,
  analytics_offload_period := '1 year'
);

-- Monitor offload work queue until completion
SELECT
  wqs.workid,
  wis.ap_wi_started_at,
  wis.ap_wi_finished_at,
  wis.ap_wi_status,
  "partition",
  sql
FROM bdr.taskmgr_local_workitem_status wis
  RIGHT JOIN bdr.taskmgr_local_work_queue_status wqs ON wis.ap_wi_workid = wqs.workid
WHERE relname = 'public.partitioned_table'
AND sql NOT LIKE '%bdr\.autopartition_create_partition%'
ORDER BY ap_wi_finished_at DESC;

-- Examine storage tier assignments
\dt+
-- Historical partitions should show "pgaa" with "0 bytes"
-- Recent partitions should show "heap" with actual storage allocation

-- Run same storage analysis query as Step 18 to capture post-migration metrics

-- Verify data integrity across storage tiers
SELECT COUNT(*) AS count, ROUND(SUM(revenue)) AS total_revenue FROM partitioned_table;
-- Should maintain: count=1000000, total_revenue=500000500