能否将动态查询结果读取至‘通用’对象？技术实现问询

一、执行存储SELECT语句的替代方式 #

除了EXECUTE IMMEDIATE，还有以下几种可行方案，具体取决于你使用的数据库：

• 数据库内置工具包/存储过程：比如Oracle的DBMS_SQL包（比EXECUTE IMMEDIATE更灵活，支持动态获取结果集元数据），SQL Server的sp_executesql，PostgreSQL的动态游标结合EXECUTE ... INTO。这类方式本质还是动态SQL，但能提供更细粒度的执行控制。
• 脚本/ETL工具：如果不想在数据库层面处理，可以用Python、Shell脚本读取表中的SQL语句，逐行执行。比如用Python的psycopg2（PostgreSQL）、pyodbc（通用）库，循环读取表内记录，执行SQL后直接处理结果。
• 数据库批量执行特性：部分数据库支持结合系统表或定时任务批量执行存储的SQL，比如PostgreSQL搭配pg_cron+动态SQL，但核心逻辑仍需依赖动态执行。

二、处理不确定字段数量与类型的通用方案 #

把所有类型转成varchar是最直接的方案，但并非最优，以下是更灵活的替代思路：

1. 使用JSON/键值对结构 #

将每行查询结果序列化为JSON对象，字段名作为键，值保留原类型（数据库需支持JSON类型）。比如PostgreSQL用row_to_json()，Oracle用JSON_OBJECT()，SQL Server用FOR JSON AUTO。这种方式既保留数据类型信息，又能适配任意字段数量，写入文件时JSON格式也易于解析。
示例（PostgreSQL）：

EXECUTE 'SELECT row_to_json(t) FROM (' || your_sql_statement || ') t';

2. 动态获取元数据+通用容器 #

执行SQL前，先通过数据库系统表获取结果集元数据（字段名、数据类型），再动态读取结果：

• PostgreSQL：查询pg_catalog.pg_attribute获取字段信息
• Oracle：通过DBMS_SQL.DESCRIBE_COLUMNS获取列信息
• SQL Server：使用sp_describe_first_result_set或查询sys.columns

拿到元数据后，用通用数据结构（比如Python的字典列表、Java的List<Map<String, Object>>）存储结果，既保留原类型，又能适配任意字段数量。

3. 转varchar的优化方案 #

如果坚持用字符串类型，需针对性处理特殊格式：日期类型指定统一格式（TO_CHAR(date_col, 'YYYY-MM-DD HH24:MI:SS')），数字类型保留足够精度（TO_CHAR(num_col, '9999999999.999')），避免数据丢失。

三、写入文件的实现 #

根据前面的方案，写入文件分两种方式：

• 数据库直接导出：如果用JSON或字符串格式，部分数据库支持直接导出到文件。比如PostgreSQL的COPY命令：

COPY (SELECT row_to_json(t) FROM (SELECT num1, num2, string1 FROM abc) t) TO '/path/to/output.json';

SQL Server可以用BCP工具导出：

bcp "SELECT * FROM xyz" queryout "output.csv" -S server -d db -U user -P password -c

• 应用层导出：用脚本读取结果后写入文件，比如Python示例：

import psycopg2
import json

conn = psycopg2.connect("dbname=your_db user=your_user")
cur = conn.cursor()

# 读取存储的SQL语句
cur.execute("SELECT sql_text FROM your_sql_table")
sql_statements = [row[0] for row in cur.fetchall()]

with open("output.json", "w") as f:
    for sql in sql_statements:
        cur.execute(sql)
        columns = [desc[0] for desc in cur.description]
        rows = cur.fetchall()
        # 转成字典列表再序列化JSON
        result = [dict(zip(columns, row)) for row in rows]
        json.dump(result, f)
        f.write("\n")

cur.close()
conn.close()

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