Snowflake Openflow Connector for PostgreSQL を AWS で動かしてみた

AWSTemplateFormatVersion: "2010-09-09"

Parameters:
  ProjectName:
    Type: String
    Default: openflow-pg
    Description: リソース名・タグの接頭辞

  VpcCidr:
    Type: String
    Default: 10.0.0.0/16

  PublicSubnet1Cidr:
    Type: String
    Default: 10.0.101.0/24
    Description: NLB / NAT 用パブリックサブネット (AZ #1)

  PublicSubnet2Cidr:
    Type: String
    Default: 10.0.102.0/24
    Description: NLB 用パブリックサブネット (AZ #2)

  ComputeSubnet1Cidr:
    Type: String
    Default: 10.0.1.0/24
    Description: EKS 用プライベートサブネット (AZ #1)

  ComputeSubnet2Cidr:
    Type: String
    Default: 10.0.2.0/24
    Description: EKS 用プライベートサブネット (AZ #2)

  DbSubnet1Cidr:
    Type: String
    Default: 10.0.3.0/24
    Description: RDS 用 DB サブネット (AZ #1)

  DbSubnet2Cidr:
    Type: String
    Default: 10.0.4.0/24
    Description: RDS 用 DB サブネット (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"
        # AWS Load Balancer Controller が外部 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"
        # 内部 LB 用サブネットの探索タグ
        - 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: Snowflake BYOC テンプレート (BYO-VPC) で選択する VPC
    Value: !Ref Vpc
    Export:
      Name: !Sub "${ProjectName}-VpcId"

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

  ComputeSubnetIds:
    Description: Snowflake BYOC テンプレートで選択する EKS 用プライベートサブネット (2つ)
    Value: !Sub "${ComputeSubnet1},${ComputeSubnet2}"
    Export:
      Name: !Sub "${ProjectName}-ComputeSubnetIds"

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

  PublicSubnetIds:
    Description: NLB ingress 用パブリックサブネット (2つ)
    Value: !Sub "${PublicSubnet1},${PublicSubnet2}"
    Export:
      Name: !Sub "${ProjectName}-PublicSubnetIds"

  DbSubnetIds:
    Description: RDS DB Subnet Group に渡す DB サブネット (2つ)
    Value: !Sub "${DbSubnet1},${DbSubnet2}"
    Export:
      Name: !Sub "${ProjectName}-DbSubnetIds"

  ComputeSubnetCidrs:
    Description: RDS セキュリティグループの許可元に使う compute サブネット CIDR
    Value: !Sub "${ComputeSubnet1Cidr},${ComputeSubnet2Cidr}"
    Export:
      Name: !Sub "${ProjectName}-ComputeSubnetCidrs"

  OpenflowPrivateSecurityGroupId:
    Description: Snowflake BYOC テンプレートの PrivateSecurityGroup に渡す SG ID
    Value: !Ref OpenflowPrivateSecurityGroup
    Export:
      Name: !Sub "${ProjectName}-OpenflowPrivateSgId"

AWSTemplateFormatVersion: "2010-09-09"

Parameters:
  ProjectName:
    Type: String
    Default: openflow-pg
    Description: 01-network.yaml と同一の接頭辞 (Export 参照に使用)

  DBName:
    Type: String
    Default: appdb
    Description: 初期データベース名 (コネクタの接続先 DB)

  MasterUsername:
    Type: String
    Default: pgadmin
    Description: RDS マスタユーザ名

  ParameterGroupFamily:
    Type: String
    Default: postgres16
    Description: DB パラメータグループのファミリ (EngineVersion と整合させること)

  EngineVersion:
    Type: String
    Default: "16.13"
    Description: >-
      PostgreSQL エンジンバージョン (ParameterGroupFamily と整合させること)

  DBInstanceClass:
    Type: String
    Default: db.t3.small
    Description: サンプル用なので、小さめのインスタンスクラスにします

  AllocatedStorage:
    Type: Number
    Default: 20
    Description: ストレージ(GB)

Resources:
  # DB認証情報格納用のSecrets Manager
  DBSecret:
    Type: AWS::SecretsManager::Secret
    # サンプル環境のためスタック削除時にシークレットも削除する
    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
        # JDBC URL / シェルで問題になりやすい文字を除外
        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:
        # 今回はVPC CIDRからのアクセスを許可
        - 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:
        # wal_level=logical を有効にする RDS 固有パラメータ
        rds.logical_replication: "1"
        # コネクタ 1 台あたり最低 1 スロット / 2 WAL sender
        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 エンドポイント (コネクタの 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
    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
    Value: !Ref DBSecret

  MasterUsername:
    Value: !Ref MasterUsername

AWSTemplateFormatVersion: "2010-09-09"

Parameters:
  ProjectName:
    Type: String
    Default: openflow-pg
    Description: 01-network.yaml と同一の接頭辞 (Export 参照に使用)

  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: 最新 Amazon Linux 2023 AMI を SSM パブリックパラメータから自動解決

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
          # psql クライアントを導入 (Amazon Linux 2023)
          dnf install -y postgresql16 || dnf install -y postgresql15
      Tags:
        - Key: Name
          Value: !Sub "${ProjectName}-bastion"

Outputs:
  BastionInstanceId:
    Description: 踏み台のインスタンス ID
    Value: !Ref BastionInstance

  SsmStartSessionCommand:
    Description: SSM セッション開始コマンド
    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;          -- 期待値: logical
SHOW max_replication_slots;
SHOW max_wal_senders;

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

-- RDS では非スーパーユーザに REPLICATION 属性を直接付けられないため、
-- 代わりに rds_replication ロールを付与して論理レプリケーションを許可する。
GRANT rds_replication TO openflow_user;

-- Grant for Snapshot / CDC
-- データソースとしてテストデータを作成したデータベースやスキーマ等へ読み取り権限を付与
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 = 'Openflow デプロイ/ランタイムを管理するロール';

-- 自分の作業ユーザに付与
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 ログインは、作業ユーザの DEFAULT_ROLE を使う。
-- 作業ユーザのロールを `OPENFLOW_ADMIN` に変更していたとしても、
-- DEFAULT_ROLEが ACCOUNTADMIN / ORGADMIN / GLOBALORGADMIN / SECURITYADMIN だと
-- Openflowにアクセス拒否される
-- (`The role requiested has been explicity blocked ...` エラーが返る)
-- そのため、DEFAULT_ROLEを非特権のOPENFLOW_ADMINに変更して、セカンダリロールをALLにする
-- 変更後は一度サインアウト→再サインインしてからOpenflowを操作すること
ALTER USER <your_user> SET DEFAULT_ROLE = OPENFLOW_ADMIN;
ALTER USER <your_user> SET DEFAULT_SECONDARY_ROLES = ('ALL');

-- Create Event Table
-- Openflowランタイムのログ/メトリクスを記録するテーブル
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 = 'Openflow PostgreSQL コネクタの宛先 DB';
CREATE ROLE IF NOT EXISTS OPENFLOW_PG_CONNECTOR_ROLE
  COMMENT = 'PostgreSQL コネクタが宛先 DB へ書き込むためのロール';

-- コネクタはスキーマ/テーブルを自動生成するため、DB への USAGE と CREATE SCHEMA を付与
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 = 'Openflow PostgreSQL コネクタ用ウェアハウス';

GRANT USAGE, OPERATE ON WAREHOUSE OPENFLOW_PG_WH TO ROLE OPENFLOW_PG_CONNECTOR_ROLE;

-- Openflow Runtimeを作成する作業ユーザに対してGrantすることで、
-- ランタイム作成時の画面上のドロップダウンリストに表示させる
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');   -- 上で追加した customer_id=4 を参照

-- 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;

-- 件数の変化確認 (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;

-- 更新の変化確認 (update)
SELECT "product_id", "name", "price", "in_stock"
FROM OPENFLOW_PG_DB."app"."products"
WHERE "name" = 'Monitor';   -- price=27800.00, in_stock=TRUE になっていること

-- メタデータ列確認
SELECT "order_id", "status",
       "_SNOWFLAKE_INSERTED_AT",
       "_SNOWFLAKE_UPDATED_AT",
       "_SNOWFLAKE_DELETED"
FROM OPENFLOW_PG_DB."app"."orders"
ORDER BY "order_id";