COPY INTO 와 스테이지 (대량 적재, 파일 포맷, 언로드)

CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake-load'
  STORAGE_ALLOWED_LOCATIONS = ('s3://my-lake/raw/');

DESC INTEGRATION s3_int;   -- STORAGE_AWS_IAM_USER_ARN, STORAGE_AWS_EXTERNAL_ID 를 AWS 신뢰 정책에

CREATE STAGE raw.s3_stage
  URL = 's3://my-lake/raw/'
  STORAGE_INTEGRATION = s3_int
  FILE_FORMAT = raw.csv_fmt;

CREATE FILE FORMAT raw.csv_fmt
  TYPE = CSV
  FIELD_DELIMITER = ','
  PARSE_HEADER = TRUE                   -- 헤더로 열 이름 매칭 (MATCH_BY_COLUMN_NAME 과 함께)
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  NULL_IF = ('', 'NULL')
  EMPTY_FIELD_AS_NULL = TRUE
  ENCODING = 'UTF8';

CREATE FILE FORMAT raw.parquet_fmt TYPE = PARQUET;
CREATE FILE FORMAT raw.json_fmt TYPE = JSON STRIP_OUTER_ARRAY = TRUE;

COPY INTO raw.orders
FROM @raw.s3_stage/orders/2026/10/
FILE_FORMAT = (FORMAT_NAME = 'raw.csv_fmt')
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE
ON_ERROR = 'ABORT_STATEMENT'
PATTERN = '.*[.]csv[.]gz';

-- 실제로 넣지 않고 검사만
COPY INTO raw.orders FROM @raw.s3_stage/orders/ VALIDATION_MODE = 'RETURN_ERRORS';

SELECT file_name, status, row_count, row_parsed, first_error_message, last_load_time
FROM snowflake.account_usage.copy_history
WHERE table_name = 'ORDERS' AND last_load_time > DATEADD(day, -1, CURRENT_TIMESTAMP())
ORDER BY last_load_time DESC;

COPY INTO raw.orders (id, ordered_at, amount, src_file)
FROM (
  SELECT $1::NUMBER, TO_TIMESTAMP_NTZ($2), $3::NUMBER(12,2), METADATA$FILENAME
  FROM @raw.s3_stage/orders/
)
FILE_FORMAT = (FORMAT_NAME = 'raw.csv_fmt');

COPY INTO @raw.s3_stage/export/orders_
FROM (SELECT * FROM mart.orders WHERE d = '2026-10-20')
FILE_FORMAT = (TYPE = PARQUET)
HEADER = TRUE
MAX_FILE_SIZE = 268435456     -- 약 256 MB
OVERWRITE = TRUE;