Snowflake单VARIANT列加载CSV实现单文件单行存储修改方案

现有加载逻辑逐行生成表记录的核心原因有两点：一是文件格式默认按换行符切分内容逐行解析，二是解析后未按文件名做聚合，直接将单行结果写入表。以下是可直接落地的修改方案：

方案一：两步加载（稳定性最高，兼容非结构化CSV） #

该方案不会出现分隔符冲突问题，适合文件内容无统一规范的场景。

1. 第一步：创建临时中转表，逐行解析CSV内容
   先修正原有代码的文件格式错误（原逻辑错误指定TYPE=JSON读取CSV文件），将所有行的解析结果暂存：

   -- 创建临时中转表
   CREATE OR REPLACE TEMP TABLE rtf_lines_stage
   (
   LOADED_AT timestamp,
   FILENAME string,
   FILE_ROW_NUMBER int,
   ROW_DATA VARIANT
   );
   
   -- 逐行加载CSV到中转表，单行列数超过20可直接扩展$x对应字段
   COPY INTO rtf_lines_stage
   from 
   (
     SELECT
       CURRENT_TIMESTAMP as LOADED_AT,
       METADATA$FILENAME as FILENAME,
       METADATA$FILE_ROW_NUMBER as FILE_ROW_NUMBER,
       object_construct(
         'col_001', T.$1, 'col_002', T.$2, 'col_003', T.$3, 'col_004', T.$4,
         'col_005', T.$5, 'col_006', T.$6, 'col_007', T.$7, 'col_008', T.$8,
         'col_009', T.$9, 'col_010', T.$10, 'col_011', T.$11, 'col_012', T.$12,
         'col_013', T.$13, 'col_014', T.$14, 'col_015', T.$15, 'col_016', T.$16,
         'col_017', T.$17, 'col_018', T.$18, 'col_019', T.$19, 'col_020', T.$20
       ) as ROW_DATA
     FROM @%rtf_lines T
   )
   FILE_FORMAT = 
   (
     TYPE = CSV
     RECORD_DELIMITER = '\n'
     ESCAPE_UNENCLOSED_FIELD = NONE
     FIELD_OPTIONALLY_ENCLOSED_BY='0x22'
     EMPTY_FIELD_AS_NULL=FALSE  
   );
   

2. 第二步：按文件名聚合，写入目标表
   将同一文件的所有行字段按顺序重排col序号，拼接为单个JSON对象存入DATA列，每个文件仅生成1条记录：

   TRUNCATE TABLE rtf_lines;
   
   INSERT INTO rtf_lines (LOADED_AT, FILENAME, FILE_ROW_NUMBER, DATA)
   SELECT 
     MIN(LOADED_AT) as LOADED_AT,
     FILENAME,
     1 as FILE_ROW_NUMBER,
     object_construct_keep_null(
       LISTAGG(
         CASE 
           WHEN f.value::string = 'col_001' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 1, 3, '0'), ''',', ROW_DATA:col_001)
           WHEN f.value::string = 'col_002' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 2, 3, '0'), ''',', ROW_DATA:col_002)
           WHEN f.value::string = 'col_003' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 3, 3, '0'), ''',', ROW_DATA:col_003)
           WHEN f.value::string = 'col_004' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 4, 3, '0'), ''',', ROW_DATA:col_004)
           WHEN f.value::string = 'col_005' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 5, 3, '0'), ''',', ROW_DATA:col_005)
           WHEN f.value::string = 'col_006' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 6, 3, '0'), ''',', ROW_DATA:col_006)
           WHEN f.value::string = 'col_007' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 7, 3, '0'), ''',', ROW_DATA:col_007)
           WHEN f.value::string = 'col_008' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 8, 3, '0'), ''',', ROW_DATA:col_008)
           WHEN f.value::string = 'col_009' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 9, 3, '0'), ''',', ROW_DATA:col_009)
           WHEN f.value::string = 'col_010' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 10, 3, '0'), ''',', ROW_DATA:col_010)
           WHEN f.value::string = 'col_011' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 11, 3, '0'), ''',', ROW_DATA:col_011)
           WHEN f.value::string = 'col_012' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 12, 3, '0'), ''',', ROW_DATA:col_012)
           WHEN f.value::string = 'col_013' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 13, 3, '0'), ''',', ROW_DATA:col_013)
           WHEN f.value::string = 'col_014' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 14, 3, '0'), ''',', ROW_DATA:col_014)
           WHEN f.value::string = 'col_015' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 15, 3, '0'), ''',', ROW_DATA:col_015)
           WHEN f.value::string = 'col_016' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 16, 3, '0'), ''',', ROW_DATA:col_016)
           WHEN f.value::string = 'col_017' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 17, 3, '0'), ''',', ROW_DATA:col_017)
           WHEN f.value::string = 'col_018' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 18, 3, '0'), ''',', ROW_DATA:col_018)
           WHEN f.value::string = 'col_019' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 19, 3, '0'), ''',', ROW_DATA:col_019)
           WHEN f.value::string = 'col_020' THEN CONCAT('''col_', LPAD( (r.FILE_ROW_NUMBER-1)*20 + 20, 3, '0'), ''',', ROW_DATA:col_020)
         END, ','
       ) WITHIN GROUP (ORDER BY FILE_ROW_NUMBER, f.index)
     )::VARIANT as DATA
   FROM rtf_lines_stage r,
   LATERAL FLATTEN(input => OBJECT_KEYS(ROW_DATA)) f
   GROUP BY FILENAME;
   

若单文件列数超过20列，同步扩展中转表object_construct的字段定义和上述聚合逻辑的CASE分支即可，col序号会自动按行顺延，和预期格式完全匹配。

方案二：一步加载（适合文件内容无特殊字符场景） #

如果确认所有CSV文件内不会出现空字符\x00，可直接修改记录分隔符将整个文件读为单个字段，省略中转表步骤：

COPY INTO rtf_lines
from 
(
  SELECT
    CURRENT_TIMESTAMP as LOADED_AT,
    METADATA$FILENAME as FILENAME,
    1 as FILE_ROW_NUMBER,
    PARSE_JSON(
      CONCAT('{',
        LISTAGG(
          CONCAT('"col_', LPAD(r.INDEX*20 + c.INDEX + 1, 3, '0'), '":"', REPLACE(c.value,'"','\"'), '"'),
          ','
        ) WITHIN GROUP (ORDER BY r.INDEX, c.INDEX),
      '}')
    ) as DATA
  FROM @%rtf_lines T,
  LATERAL FLATTEN(input => SPLIT(T.$1, '\n')) r,
  LATERAL FLATTEN(input => SPLIT(r.value, ',')) c
  GROUP BY METADATA$FILENAME
)
FILE_FORMAT = 
(
  TYPE = CSV
  RECORD_DELIMITER = '\x00'
  ESCAPE_UNENCLOSED_FIELD = NONE
  FIELD_OPTIONALLY_ENCLOSED_BY=NONE
  EMPTY_FIELD_AS_NULL=FALSE  
);

效果验证 #

加载完成后查询目标表，每个CSV文件仅对应1条记录，DATA列按col_001、col_002的顺序存储文件内所有行的字段值，符合业务要求。

内容的提问来源于stack exchange，提问作者CodeM