I tried running the Snowflake Openflow Connector for PostgreSQL on AWS

AWSTemplateFormatVersion: "2010-09-09"

Parameters:
  ProjectName:
    Type: String
    Default: openflow-pg
    Description: Prefix for resource names and tags

  VpcCidr:
    Type: String
    Default: 10.0.0.0/16

  PublicSubnet1Cidr:
    Type: String
    Default: 10.0.101.0/24
    Description: Public subnet for NLB / NAT (AZ #1)

  PublicSubnet2Cidr:
    Type: String
    Default: 10.0.102.0/24
    Description: Public subnet for NLB (AZ #2)

  ComputeSubnet1Cidr:
    Type: String
    Default: 10.0.1.0/24
    Description: Private subnet for EKS (AZ #1)

  ComputeSubnet2Cidr:
    Type: String
    Default: 10.0.2.0/24
    Description: Private subnet for EKS (AZ #2)

  DbSubnet1Cidr:
    Type: String
    Default: 10.0.3.0/24
    Description: DB subnet for RDS (AZ #1)

  DbSubnet2Cidr:
    Type: String
    Default: 10.0.4.0/24
    Description: DB subnet for RDS (AZ #2)

Resources:
  Vpc:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: !Ref VpcCidr
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-vpc"

  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-igw"

  VpcGatewayAttachment:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId: !Ref Vpc
      InternetGatewayId: !Ref InternetGateway

  PublicSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref PublicSubnet1Cidr
      AvailabilityZone: !Select [0, !GetAZs ""]
      MapPublicIpOnLaunch: true
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-public-1"
        # Tag for AWS Load Balancer Controller to discover subnets for creating external NLB
        - Key: kubernetes.io/role/elb
          Value: "1"

  PublicSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref PublicSubnet2Cidr
      AvailabilityZone: !Select [1, !GetAZs ""]
      MapPublicIpOnLaunch: true
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-public-2"
        - Key: kubernetes.io/role/elb
          Value: "1"

  ComputeSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref ComputeSubnet1Cidr
      AvailabilityZone: !Select [0, !GetAZs ""]
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-compute-1"
        # Discovery tag for internal LB subnets
        - Key: kubernetes.io/role/internal-elb
          Value: "1"

  ComputeSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref ComputeSubnet2Cidr
      AvailabilityZone: !Select [1, !GetAZs ""]
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-compute-2"
        - Key: kubernetes.io/role/internal-elb
          Value: "1"

  DbSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref DbSubnet1Cidr
      AvailabilityZone: !Select [0, !GetAZs ""]
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-db-1"

  DbSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref DbSubnet2Cidr
      AvailabilityZone: !Select [1, !GetAZs ""]
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-db-2"

  NatEip:
    Type: AWS::EC2::EIP
    DependsOn: VpcGatewayAttachment
    Properties:
      Domain: vpc
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-nat-eip"

  NatGateway:
    Type: AWS::EC2::NatGateway
    Properties:
      AllocationId: !GetAtt NatEip.AllocationId
      SubnetId: !Ref PublicSubnet1
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-nat"

  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref Vpc
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-public-rt"

  PublicDefaultRoute:
    Type: AWS::EC2::Route
    DependsOn: VpcGatewayAttachment
    Properties:
      RouteTableId: !Ref PublicRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway

  PublicSubnet1RouteAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PublicRouteTable
      SubnetId: !Ref PublicSubnet1

  PublicSubnet2RouteAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PublicRouteTable
      SubnetId: !Ref PublicSubnet2

  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref Vpc
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-private-rt"

  PrivateDefaultRoute:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId: !Ref NatGateway

  ComputeSubnet1RouteAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref ComputeSubnet1

  ComputeSubnet2RouteAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref ComputeSubnet2

  DbRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref Vpc
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-db-rt"

  DbSubnet1RouteAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref DbRouteTable
      SubnetId: !Ref DbSubnet1

  DbSubnet2RouteAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref DbRouteTable
      SubnetId: !Ref DbSubnet2

  OpenflowPrivateSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: !Sub "${ProjectName} Openflow BYOC private SG (self ingress, all egress)"
      VpcId: !Ref Vpc
      SecurityGroupEgress:
        - IpProtocol: "-1"
          CidrIp: 0.0.0.0/0
          Description: allow all outbound
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-openflow-private-sg"

  OpenflowPrivateSgSelfIngress:
    Type: AWS::EC2::SecurityGroupIngress
    Properties:
      GroupId: !Ref OpenflowPrivateSecurityGroup
      IpProtocol: "-1"
      SourceSecurityGroupId: !Ref OpenflowPrivateSecurityGroup
      Description: allow all inbound from itself

Outputs:
  VpcId:
    Description: VPC to select in the Snowflake BYOC template (BYO-VPC)
    Value: !Ref Vpc
    Export:
      Name: !Sub "${ProjectName}-VpcId"

  VpcCidr:
    Value: !Ref VpcCidr
    Export:
      Name: !Sub "${ProjectName}-VpcCidr"

  ComputeSubnetIds:
    Description: Private subnets for EKS to select in the Snowflake BYOC template (2 subnets)
    Value: !Sub "${ComputeSubnet1},${ComputeSubnet2}"
    Export:
      Name: !Sub "${ProjectName}-ComputeSubnetIds"

  ComputeSubnet1Id:
    Value: !Ref ComputeSubnet1
  ComputeSubnet2Id:
    Value: !Ref ComputeSubnet2

  PublicSubnetIds:
    Description: Public subnets for NLB ingress (2 subnets)
    Value: !Sub "${PublicSubnet1},${PublicSubnet2}"
    Export:
      Name: !Sub "${ProjectName}-PublicSubnetIds"

  DbSubnetIds:
    Description: DB subnets to pass to the RDS DB Subnet Group (2 subnets)
    Value: !Sub "${DbSubnet1},${DbSubnet2}"
    Export:
      Name: !Sub "${ProjectName}-DbSubnetIds"

  ComputeSubnetCidrs:
    Description: Compute subnet CIDRs to use as allowed source in the RDS security group
    Value: !Sub "${ComputeSubnet1Cidr},${ComputeSubnet2Cidr}"
    Export:
      Name: !Sub "${ProjectName}-ComputeSubnetCidrs"

  OpenflowPrivateSecurityGroupId:
    Description: SG ID to pass to PrivateSecurityGroup in the Snowflake BYOC template
    Value: !Ref OpenflowPrivateSecurityGroup
    Export:
      Name: !Sub "${ProjectName}-OpenflowPrivateSgId"

AWSTemplateFormatVersion: "2010-09-09"

Parameters:
  ProjectName:
    Type: String
    Default: openflow-pg
    Description: Same prefix as 01-network.yaml (used for Export references)

  DBName:
    Type: String
    Default: appdb
    Description: Initial database name (DB to connect the connector to)

  MasterUsername:
    Type: String
    Default: pgadmin
    Description: RDS master username

  ParameterGroupFamily:
    Type: String
    Default: postgres16
    Description: DB parameter group family (must match EngineVersion)

  EngineVersion:
    Type: String
    Default: "16.13"
    Description: >-
      PostgreSQL engine version (must match ParameterGroupFamily)

  DBInstanceClass:
    Type: String
    Default: db.t3.small
    Description: Small instance class for sample purposes

  AllocatedStorage:
    Type: Number
    Default: 20
    Description: Storage (GB)

Resources:
  # Secrets Manager for DB credentials
  DBSecret:
    Type: AWS::SecretsManager::Secret
    # Delete secret when stack is deleted for sample environment
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      Name: !Sub "${ProjectName}/rds/master"
      Description: RDS for PostgreSQL master credentials
      GenerateSecretString:
        SecretStringTemplate: !Sub '{"username":"${MasterUsername}"}'
        GenerateStringKey: password
        PasswordLength: 24
        # Exclude characters that tend to cause issues in JDBC URLs / shell
        ExcludeCharacters: '"@/\ ''`'
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-rds-master-secret"

  DBSubnetGroup:
    Type: AWS::RDS::DBSubnetGroup
    Properties:
      DBSubnetGroupDescription: !Sub "${ProjectName} DB subnet group"
      SubnetIds: !Split
        - ","
        - !ImportValue
          "Fn::Sub": "${ProjectName}-DbSubnetIds"
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-db-subnet-group"

  DBSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: !Sub "${ProjectName} RDS PostgreSQL access (VPC internal only)"
      VpcId: !ImportValue
        "Fn::Sub": "${ProjectName}-VpcId"
      SecurityGroupIngress:
        # Allow access from VPC CIDR in this verification
        - IpProtocol: tcp
          FromPort: 5432
          ToPort: 5432
          CidrIp: !ImportValue
            "Fn::Sub": "${ProjectName}-VpcCidr"
          Description: PostgreSQL from within VPC (EKS/Openflow runtime)
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-rds-sg"

  DBParameterGroup:
    Type: AWS::RDS::DBParameterGroup
    Properties:
      Description: !Sub "${ProjectName} PostgreSQL params with logical replication"
      Family: !Ref ParameterGroupFamily
      Parameters:
        # RDS-specific parameter to enable wal_level=logical
        rds.logical_replication: "1"
        # At least 1 slot / 2 WAL senders per connector
        max_replication_slots: "10"
        max_wal_senders: "10"
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-pg-params"

  DBInstance:
    Type: AWS::RDS::DBInstance
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      DBInstanceIdentifier: !Sub "${ProjectName}-postgres"
      Engine: postgres
      EngineVersion: !Ref EngineVersion
      DBName: !Ref DBName
      DBInstanceClass: !Ref DBInstanceClass
      AllocatedStorage: !Ref AllocatedStorage
      StorageType: gp3
      StorageEncrypted: true
      MultiAZ: false
      PubliclyAccessible: false
      DeletionProtection: false
      BackupRetentionPeriod: 1
      AutoMinorVersionUpgrade: true
      MasterUsername: !Ref MasterUsername
      MasterUserPassword: !Sub "{{resolve:secretsmanager:${DBSecret}:SecretString:password}}"
      DBParameterGroupName: !Ref DBParameterGroup
      DBSubnetGroupName: !Ref DBSubnetGroup
      VPCSecurityGroups:
        - !Ref DBSecurityGroup
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-postgres"

Outputs:
  DBEndpointAddress:
    Description: RDS endpoint (used in the connector's JDBC URL)
    Value: !GetAtt DBInstance.Endpoint.Address
    Export:
      Name: !Sub "${ProjectName}-RdsEndpoint"

  DBEndpointPort:
    Value: !GetAtt DBInstance.Endpoint.Port

  DBName:
    Value: !Ref DBName

  JdbcUrl:
    Description: JDBC URL that can be set directly in the connector
    Value: !Sub
      - "jdbc:postgresql://${Addr}:${Port}/${Db}"
      - Addr: !GetAtt DBInstance.Endpoint.Address
        Port: !GetAtt DBInstance.Endpoint.Port
        Db: !Ref DBName

  MasterSecretArn:
    Description: Secrets Manager ARN for master credentials
    Value: !Ref DBSecret

  MasterUsername:
    Value: !Ref MasterUsername

AWSTemplateFormatVersion: "2010-09-09"

Parameters:
  ProjectName:
    Type: String
    Default: openflow-pg
    Description: Same prefix as 01-network.yaml (used for Export references)

  InstanceType:
    Type: String
    Default: t3.micro

  LatestAmiId:
    Type: AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
    Default: /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64
    Description: Automatically resolve the latest Amazon Linux 2023 AMI from SSM public parameters

Resources:
  BastionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "${ProjectName}-bastion-role"
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-bastion-role"

  BastionInstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      InstanceProfileName: !Sub "${ProjectName}-bastion-profile"
      Roles:
        - !Ref BastionRole

  BastionSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: !Sub "${ProjectName} bastion (SSM only, no inbound)"
      VpcId: !ImportValue
        "Fn::Sub": "${ProjectName}-VpcId"
      SecurityGroupEgress:
        - IpProtocol: "-1"
          CidrIp: 0.0.0.0/0
          Description: allow all outbound
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-bastion-sg"

  BastionInstance:
    Type: AWS::EC2::Instance
    Properties:
      InstanceType: !Ref InstanceType
      ImageId: !Ref LatestAmiId
      IamInstanceProfile: !Ref BastionInstanceProfile
      SubnetId: !Select
        - 0
        - !Split
          - ","
          - !ImportValue
            "Fn::Sub": "${ProjectName}-ComputeSubnetIds"
      SecurityGroupIds:
        - !Ref BastionSecurityGroup
      UserData:
        Fn::Base64: |
          #!/bin/bash
          # Install psql client (Amazon Linux 2023)
          dnf install -y postgresql16 || dnf install -y postgresql15
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-bastion"

Outputs:
  BastionInstanceId:
    Description: Bastion instance ID
    Value: !Ref BastionInstance

  SsmStartSessionCommand:
    Description: SSM session start command
    Value: !Sub "aws ssm start-session --target ${BastionInstance} --region ${AWS::Region}"

-- Schema
CREATE SCHEMA IF NOT EXISTS app;

-- Tables
CREATE TABLE app.customers (
    customer_id  SERIAL PRIMARY KEY,
    name         VARCHAR(100) NOT NULL,
    email        VARCHAR(200) NOT NULL,
    created_at   TIMESTAMP    NOT NULL DEFAULT now()
);
CREATE TABLE app.products (
    product_id   SERIAL PRIMARY KEY,
    name         VARCHAR(100)   NOT NULL,
    price        NUMERIC(10,2)  NOT NULL,
    in_stock     BOOLEAN        NOT NULL DEFAULT true
);
CREATE TABLE app.orders (
    order_id     SERIAL PRIMARY KEY,
    customer_id  INTEGER      NOT NULL REFERENCES app.customers(customer_id),
    product_id   INTEGER      NOT NULL REFERENCES app.products(product_id),
    quantity     INTEGER      NOT NULL DEFAULT 1,
    status       VARCHAR(20)  NOT NULL DEFAULT 'pending',
    order_date   TIMESTAMP    NOT NULL DEFAULT now()
);

-- Initial data
INSERT INTO app.customers (name, email) VALUES
    ('Taro Yamada',   'taro@example.com'),
    ('Hanako Suzuki', 'hanako@example.com'),
    ('Jiro Tanaka',   'jiro@example.com');

INSERT INTO app.products (name, price, in_stock) VALUES
    ('Keyboard', 3980.00, true),
    ('Mouse',    1980.00, true),
    ('Monitor',  29800.00, false);

INSERT INTO app.orders (customer_id, product_id, quantity, status) VALUES
    (1, 1, 2, 'confirmed'),
    (2, 2, 1, 'pending'),
    (3, 3, 1, 'shipped');

-- check data
SELECT 'customers' AS t, count(*) FROM app.customers
UNION ALL SELECT 'products', count(*) FROM app.products
UNION ALL SELECT 'orders',   count(*) FROM app.orders;

-- Check logical replication
SHOW wal_level;          -- Expected value: logical
SHOW max_replication_slots;
SHOW max_wal_senders;

-- user for openflow connector
CREATE USER openflow_user WITH PASSWORD '<CONNECTOR_PASSWORD>';

-- In RDS, the REPLICATION attribute cannot be granted directly to non-superusers,
-- so instead grant the rds_replication role to allow logical replication.
GRANT rds_replication TO openflow_user;

-- Grant for Snapshot / CDC
-- Grant read permissions to the database, schema, etc. where test data was created as the data source
GRANT CONNECT ON DATABASE appdb TO openflow_user;
GRANT USAGE   ON SCHEMA  app    TO openflow_user;
GRANT SELECT  ON ALL TABLES IN SCHEMA app TO openflow_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA app GRANT SELECT ON TABLES TO openflow_user;

-- PUBLICATION
CREATE PUBLICATION openflow_pub
  FOR TABLE app.customers, app.products, app.orders
  WITH (publish_via_partition_root = true);

-- Check
SELECT pubname, puballtables FROM pg_publication WHERE pubname = 'openflow_pub';
SELECT schemaname, tablename FROM pg_publication_tables WHERE pubname = 'openflow_pub';

-- Create Openflow Admin Role
USE ROLE ACCOUNTADMIN;

CREATE ROLE IF NOT EXISTS OPENFLOW_ADMIN
  COMMENT = 'Role to manage Openflow deployment/runtime';

-- Grant to your working user
GRANT ROLE OPENFLOW_ADMIN TO USER <your_user>;

-- Grant for Openflow
GRANT CREATE OPENFLOW DATA PLANE INTEGRATION ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE OPENFLOW RUNTIME INTEGRATION   ON ACCOUNT TO ROLE OPENFLOW_ADMIN;

-- OAuth login to the runtime/connector uses the DEFAULT_ROLE of the working user.
-- Even if the working user's role is changed to `OPENFLOW_ADMIN`,
-- if DEFAULT_ROLE is ACCOUNTADMIN / ORGADMIN / GLOBALORGADMIN / SECURITYADMIN,
-- access to Openflow will be denied
-- (the error `The role requested has been explicitly blocked ...` is returned)
-- Therefore, change DEFAULT_ROLE to the non-privileged OPENFLOW_ADMIN and set secondary roles to ALL.
-- After making the change, sign out and sign back in before operating Openflow.
ALTER USER <your_user> SET DEFAULT_ROLE = OPENFLOW_ADMIN;
ALTER USER <your_user> SET DEFAULT_SECONDARY_ROLES = ('ALL');

-- Create Event Table
-- Table to record logs/metrics for the Openflow runtime
CREATE DATABASE IF NOT EXISTS OPENFLOW;
CREATE SCHEMA   IF NOT EXISTS OPENFLOW.TELEMETRY;
CREATE EVENT TABLE IF NOT EXISTS OPENFLOW.TELEMETRY.EVENTS;

GRANT USAGE ON DATABASE OPENFLOW              TO ROLE OPENFLOW_ADMIN;
GRANT USAGE ON SCHEMA   OPENFLOW.TELEMETRY    TO ROLE OPENFLOW_ADMIN;

SHOW OPENFLOW DATA PLANE INTEGRATIONS;
ALTER OPENFLOW DATA PLANE INTEGRATION <deployment_name>
  SET EVENT_TABLE = 'OPENFLOW.TELEMETRY.EVENTS';

USE ROLE ACCOUNTADMIN;

-- Create database and role for destination
CREATE DATABASE IF NOT EXISTS OPENFLOW_PG_DB
  COMMENT = 'Destination DB for Openflow PostgreSQL connector';
CREATE ROLE IF NOT EXISTS OPENFLOW_PG_CONNECTOR_ROLE
  COMMENT = 'Role for the PostgreSQL connector to write to the destination DB';

-- The connector auto-generates schemas/tables, so grant USAGE and CREATE SCHEMA on the DB
GRANT USAGE        ON DATABASE OPENFLOW_PG_DB TO ROLE OPENFLOW_PG_CONNECTOR_ROLE;
GRANT CREATE SCHEMA ON DATABASE OPENFLOW_PG_DB TO ROLE OPENFLOW_PG_CONNECTOR_ROLE;

-- Create warehouse
CREATE WAREHOUSE IF NOT EXISTS OPENFLOW_PG_WH
  WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND   = 60
  AUTO_RESUME    = TRUE
  INITIALLY_SUSPENDED = TRUE
  COMMENT = 'Warehouse for Openflow PostgreSQL connector';

GRANT USAGE, OPERATE ON WAREHOUSE OPENFLOW_PG_WH TO ROLE OPENFLOW_PG_CONNECTOR_ROLE;

-- Grant to the working user who creates the Openflow Runtime,
-- so it appears in the dropdown list on the runtime creation screen
GRANT ROLE OPENFLOW_PG_CONNECTOR_ROLE TO USER <your_user>;

SELECT 'customers' AS t, COUNT(*) FROM OPENFLOW_PG_DB."app"."customers"
UNION ALL SELECT 'products', COUNT(*) FROM OPENFLOW_PG_DB."app"."products"
UNION ALL SELECT 'orders',   COUNT(*) FROM OPENFLOW_PG_DB."app"."orders"
ORDER BY t;

-- insert
INSERT INTO app.customers (name, email) VALUES
    ('Saburo Sato', 'saburo@example.com');

INSERT INTO app.orders (customer_id, product_id, quantity, status) VALUES
    (4, 1, 5, 'pending');   -- References customer_id=4 added above

-- update
UPDATE app.products
   SET in_stock = true, price = 27800.00
 WHERE name = 'Monitor';

UPDATE app.orders
   SET status = 'shipped'
 WHERE order_id = 2;

-- delete
DELETE FROM app.orders
 WHERE order_id = 3;

-- check count
SELECT 'customers' AS t, count(*) FROM app.customers
UNION ALL SELECT 'products', count(*) FROM app.products
UNION ALL SELECT 'orders',   count(*) FROM app.orders;

-- Verify count changes (insert / delete)
SELECT 'customers' AS t, COUNT(*) FROM OPENFLOW_PG_DB."app"."customers"   -- 3 -> 4 (INSERT)
UNION ALL SELECT 'orders', COUNT(*) FROM OPENFLOW_PG_DB."app"."orders"    -- INSERT 1, DELETE 1
ORDER BY t;

-- Verify update changes (update)
SELECT "product_id", "name", "price", "in_stock"
FROM OPENFLOW_PG_DB."app"."products"
WHERE "name" = 'Monitor';   -- Confirm price=27800.00, in_stock=TRUE

-- Check metadata columns
SELECT "order_id", "status",
       "_SNOWFLAKE_INSERTED_AT",
       "_SNOWFLAKE_UPDATED_AT",
       "_SNOWFLAKE_DELETED"
FROM OPENFLOW_PG_DB."app"."orders"
ORDER BY "order_id";