Dataherald多数据源支持详解

Dataherald作为企业级自然语言转SQL引擎，提供了对多种主流数据库和数据仓库的深度集成支持。本文详细解析了Dataherald对PostgreSQL、MySQL、Snowflake、BigQuery、Databricks和Redshift等数据源的连接配置、多schema支持机制、安全加密处理以及性能优化特性。通过统一的连接管理架构和智能的SQL生成引擎，Dataherald能够实现跨schema查询和复杂的数据分析，为企业用户提供无缝的自然语言数据查询体验。

PostgreSQL与MySQL集成

Dataherald作为企业级自然语言转SQL引擎，对PostgreSQL和MySQL这两种主流关系型数据库提供了深度集成支持。通过专门的扫描器实现和统一的连接管理机制，Dataherald能够智能地处理这两种数据库的元数据扫描、查询日志分析以及低基数值识别等关键功能。

连接配置与URI格式

Dataherald使用标准化的URI格式来配置数据库连接，PostgreSQL和MySQL的连接URI遵循统一的模式：

PostgreSQL连接URI格式：

postgresql://username:password@host:port/database

MySQL连接URI格式：

mysql://username:password@host:port/database

在实际配置中，可以通过API创建数据库连接：

curl -X 'POST' \
  'http://localhost/api/v1/database-connections' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "alias": "production_postgres",
  "use_ssh": false,
  "connection_uri": "postgresql://user:pass@localhost:5432/mydb"
}'

数据库方言支持枚举

Dataherald通过SupportedDialects枚举类明确定义了支持的数据库类型：

class SupportedDialects(Enum):
    POSTGRES = "postgresql"
    MYSQL = "mysql"
    MSSQL = "mssql"
    DATABRICKS = "databricks"
    SNOWFLAKE = "snowflake"
    # ... 其他数据库支持

这种枚举设计确保了类型安全，并在连接URI解析时自动识别数据库类型。

PostgreSQL专用扫描器实现

Dataherald为PostgreSQL实现了专门的扫描器PostgreSqlScanner，它继承自抽象扫描器基类，并重写了关键方法：

class PostgreSqlScanner(AbstractScanner):
    def cardinality_values(self, column: Column, db_engine: SQLDatabase) -> list | None:
        rs = db_engine.engine.execute(
            f"SELECT n_distinct, most_common_vals FROM pg_stats 
             WHERE tablename = '{column.table.name}' AND attname = '{column.name}'"
        ).fetchall()
        
        if len(rs) > 0 and 1 < rs[0]["n_distinct"] <= 100:
            return rs[0]["most_common_vals"]
        return None

PostgreSQL扫描器利用pg_stats系统表来获取列的统计信息，特别是n_distinct（不同值数量）和most_common_vals（最常见值），这对于识别低基数列和优化自然语言查询至关重要。

MySQL扫描处理机制

虽然代码中没有显示专门的MySQL扫描器，但Dataherald通过统一的SQLAlchemy扫描器来处理MySQL数据库。MySQL的元数据扫描依赖于SQLAlchemy的反射机制：

# 在SQLAlchemy扫描器中统一处理所有数据库
def scan_single_table(self, meta: MetaData, table: str, db_engine: SQLDatabase, 
                     db_connection_id: str, repository: TableDescriptionRepository,
                     scanner_service: AbstractScanner, schema: str | None = None):
    # 获取表结构信息
    table_obj = meta.tables[table]
    columns_info = []
    
    for column in table_obj.columns:
        column_detail = self.get_processed_column(meta, table, 
                                                column.__dict__, db_engine, 
                                                scanner_service)
        columns_info.append(column_detail)
    
    # 存储表描述信息
    table_description = TableDescription(
        db_connection_id=db_connection_id,
        table_name=table,
        columns=columns_info,
        schema=schema
    )
    repository.save_table_info(table_description)

多schema支持特性

PostgreSQL支持多schema配置，这在企业环境中特别有用。Dataherald允许在单个数据库连接中配置多个schema：

curl -X 'POST' \
  'http://localhost/api/v1/database-connections' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "alias": "multi_schema_postgres",
  "use_ssh": false,
  "connection_uri": "postgresql://user:pass@localhost:5432/mydb",
  "schemas": ["public", "sales", "hr", "finance"]
}'

这种多schema支持使得Dataherald能够在跨schema查询时保持上下文一致性，特别是在处理关联查询时。

安全与加密处理

Dataherald对数据库连接信息进行加密存储，确保敏感数据的安全性：

@validator("connection_uri", pre=True, always=True)
def connection_uri_format(cls, value: str, values):
    fernet_encrypt = FernetEncrypt()
    try:
        fernet_encrypt.decrypt(value)
    except Exception:
        dialect_prefix = cls.get_dialect(value)
        values["dialect"] = cls.set_dialect(dialect_prefix)
        value = fernet_encrypt.encrypt(value)
    return value

性能优化特性

PostgreSQL和MySQL在Dataherald中都受益于以下性能优化特性：

1. 连接池管理：通过SQLAlchemy管理数据库连接池
2. 元数据缓存：扫描后的表结构信息被缓存以提高查询性能
3. 异步处理：数据库扫描操作支持异步执行
4. 批量处理：支持批量表扫描和元数据更新

错误处理与兼容性

Dataherald针对不同数据库的方言差异进行了兼容性处理：

def get_processed_column(self, meta: MetaData, table: str, column: dict, 
                        db_engine: SQLDatabase, scanner_service: AbstractScanner):
    # 处理不同数据库的类型映射
    column_type = str(column['type'])
    if 'postgresql' in str(db_engine.engine.url):
        # PostgreSQL特定类型处理
        if 'TIMESTAMP' in column_type:
            column_type = 'timestamp'
    elif 'mysql' in str(db_engine.engine.url):
        # MySQL特定类型处理
        if 'DATETIME' in column_type:
            column_type = 'datetime'
    
    return ColumnDetail(
        name=column['name'],
        type=column_type,
        low_cardinality=scanner_service.cardinality_values(column, db_engine)
    )

查询生成优化

基于扫描到的元数据信息，Dataherald能够为PostgreSQL和MySQL生成更加准确和优化的SQL查询：

flowchart TD
    A[自然语言查询] --> B[数据库连接识别]
    B --> C{数据库类型判断}
    C -->|PostgreSQL| D[使用pg_stats优化查询]
    C -->|MySQL| E[使用信息模式优化查询]
    D --> F[生成PostgreSQL方言SQL]
    E --> G[生成MySQL方言SQL]
    F --> H[执行查询并返回结果]
    G --> H

这种深度集成使得Dataherald不仅能够理解数据库结构，还能根据具体的数据库特性生成最优化的查询语句，大大提高了自然语言转SQL的准确性和性能。

通过这种系统化的集成方法，Dataherald为PostgreSQL和MySQL用户提供了无缝的自然语言查询体验，无论是简单的数据检索还是复杂的多表关联查询，都能获得准确和高效的SQL生成结果。

Snowflake与BigQuery配置详解

Dataherald作为企业级自然语言转SQL引擎，对主流数据仓库提供了深度支持，其中Snowflake和BigQuery作为云数据仓库的代表，在Dataherald中具有特殊的配置要求和优化特性。本文将深入解析这两种数据源的配置细节、连接机制和最佳实践。

连接字符串格式解析

Dataherald采用标准化的连接字符串格式来支持多种数据源，对于Snowflake和BigQuery，连接字符串的构造遵循特定的模式。

Snowflake连接字符串

Snowflake的连接字符串采用以下格式：

snowflake://<username>:<password>@<organization>-<account>/<database>/<schema>

示例配置：

# Snowflake标准连接配置
connection_uri = "snowflake://user123:password456@myorg-account123/mydatabase/myschema"

# 多schema支持配置
connection_uri = "snowflake://user123:password456@myorg-account123/mydatabase"
schemas = ["schema1", "schema2", "schema3"]

BigQuery连接字符串

BigQuery的连接字符串相对简洁，但需要额外的凭据文件支持：

bigquery://project-id/dataset

示例配置：

# BigQuery标准连接配置
connection_uri = "bigquery://my-gcp-project/my-dataset"
path_to_credentials_file = "/path/to/service-account-key.json"

多Schema支持机制

Dataherald对Snowflake和BigQuery提供了原生的多Schema支持，这是通过特殊的URI处理机制实现的。

Schema URI处理流程

Dataherald通过正则表达式模式匹配来处理Schema信息：

flowchart TD
    A[原始连接URI] --> B{是否包含Schema?}
    B -->|是| C[移除现有Schema]
    B -->|否| D[保持原URI]
    C --> E[添加目标Schema]
    D --> E
    E --> F[生成最终连接URI]

具体的URI处理逻辑在DatabaseConnectionService类中实现：

def add_schema_in_uri(self, connection_uri: str, schema: str, dialect: str) -> str:
    connection_uri = self.remove_schema_in_uri(connection_uri, dialect)
    if dialect in ["snowflake", "bigquery"]:
        return f"{connection_uri}/{schema}"
    # 其他数据源处理逻辑...

正则表达式模式匹配

Dataherald使用以下正则表达式模式来处理不同数据源的URI：

数据源	正则表达式模式	用途
Snowflake	([^:/]+)://([^:]+):([^@]+)@([^:/]+)(?::(\d+))?/([^/]+)	提取连接参数
BigQuery	([^:/]+)://([^/]+)	基础URI匹配

凭据文件管理

BigQuery连接需要服务账户密钥文件，Dataherald提供了灵活的凭据文件管理机制。

凭据文件路径处理

Dataherald支持本地文件路径和S3存储的凭据文件：

# 本地文件路径配置
path_to_credentials_file = "/path/to/bigquery-key.json"

# S3存储配置
path_to_credentials_file = "s3://my-bucket/credentials/bigquery-key.json"

自动下载机制

当检测到S3路径时，Dataherald会自动下载凭据文件：

if file_path and file_path.lower().startswith("s3"):
    s3 = S3()
    file_path = s3.download(file_path)
if db_uri.lower().startswith("bigquery"):
    db_uri = db_uri + f"?credentials_path={file_path}"

加密安全机制

所有敏感信息在存储时都会进行加密处理：

from dataherald.utils.encrypt import FernetEncrypt

fernet_encrypt = FernetEncrypt()
encrypted_uri = fernet_encrypt.encrypt(connection_uri)
decrypted_uri = fernet_encrypt.decrypt(encrypted_uri)

连接池管理

Dataherald实现了智能的连接池管理，避免重复创建数据库连接：

class DBConnections:
    db_connections = {}

    @staticmethod
    def add(uri, engine):
        DBConnections.db_connections[uri] = engine

    @staticmethod
    def get(uri):
        return DBConnections.db_connections.get(uri)

最佳实践配置

Snowflake最佳实践

1. 多Schema配置：

{
  "alias": "snowflake_prod",
  "connection_uri": "snowflake://user:pass@org-account/database",
  "schemas": ["sales", "marketing", "finance"]
}

2. 连接参数优化：

# 建议设置连接超时和查询行数限制
UPPER_LIMIT_QUERY_RETURN_ROWS = 100
DH_ENGINE_TIMEOUT = 120

BigQuery最佳实践

1. 服务账户权限： 确保服务账户具有以下权限：

• bigquery.jobs.create
• bigquery.tables.getData
• bigquery.tables.list

2. 凭据文件安全：

# 使用环境变量或密钥管理系统
import os
path_to_credentials_file = os.getenv('BIGQUERY_CREDENTIALS_PATH')

错误处理与调试

Dataherald提供了详细的错误处理机制：

try:
    sql_database = SQLDatabase.get_sql_engine(database_connection, True)
    tables = sql_database.get_tables_and_views()
except InvalidDBConnectionError as e:
    logger.error(f"数据库连接失败: {e.description}")
except EmptyDBError as e:
    logger.warning("数据库为空，可能是权限问题")

性能优化建议

1. 连接复用：利用Dataherald的连接池机制，避免频繁创建新连接
2. Schema预加载：在创建连接时预加载所有需要的Schema信息
3. 查询优化：设置合适的UPPER_LIMIT_QUERY_RETURN_ROWS限制返回行数
4. 超时配置：根据网络状况调整DH_ENGINE_TIMEOUT参数

通过以上配置和最佳实践，可以确保Snowflake和BigQuery在Dataherald中达到最佳性能和稳定性。Dataherald的多数据源支持架构使得企业能够无缝集成不同的数据仓库，为自然语言查询提供统一接口。

Databricks与Redshift连接

Dataherald作为企业级自然语言转SQL引擎，对现代数据仓库提供了全面的支持，其中Databricks和Amazon Redshift作为两大主流数据平台，在Dataherald中得到了深度集成和优化支持。本节将详细解析Dataherald如何实现与这两种数据仓库的无缝连接。

连接架构设计

Dataherald采用统一的SQLAlchemy引擎架构来处理所有支持的数据库连接，通过抽象化的接口设计，为不同数据源提供一致的访问体验。对于Databricks和Redshift，系统实现了专门的方言支持和连接参数处理机制。

flowchart TD
    A[客户端请求] --> B[DatabaseConnectionService]
    B --> C{数据库类型判断}
    C -->|Databricks| D[处理Databricks连接参数]
    C -->|Redshift| E[处理Redshift连接参数]
    
    D --> F[构建Databricks连接URI]
    E --> G[构建Redshift连接URI]
    
    F --> H[SQLAlchemy引擎创建]
    G --> H
    
    H --> I[连接池管理]
    I --> J[返回SQLDatabase实例]

Databricks连接实现

Dataherald通过Databricks官方提供的SQLAlchemy方言支持来实现连接，需要配置以下关键参数：

连接URI格式：

databricks://token:<personal-access-token>@<workspace-url>?http_path=<http-path>&catalog=<catalog-name>&schema=<schema-name>

参数说明表：

参数	类型	必需	描述
token	string	是	Databricks个人访问令牌
workspace-url	string	是	Databricks工作区URL地址
http-path	string	是	SQL仓库的HTTP路径
catalog	string	否	Unity Catalog目录名称
schema	string	否	数据库schema名称

代码示例 - Databricks连接配置：

from dataherald.sql_database.models.types import DatabaseConnection, SupportedDialects
from dataherald.sql_database.services.database_connection import DatabaseConnectionService

# 创建Databricks数据库连接
databricks_conn = DatabaseConnection(
    alias="production_databricks",
    dialect=SupportedDialects.DATABRICKS,
    connection_uri="databricks://token:[e-mail]?http_path=sql/protocolv1/o/123456789/1234-567890-abcd&catalog=main&schema=default",
    use_ssh=False
)

# 通过服务获取SQL数据库实例
service = DatabaseConnectionService(scanner, storage)
sql_db = service.get_sql_database(databricks_conn)

Redshift连接实现

对于Amazon Redshift，Dataherald使用PostgreSQL兼容的连接方式，通过redshift+psycopg2方言驱动实现连接：

连接URI格式：

redshift+psycopg2://<username>:<password>@<hostname>:<port>/<database-name>

参数说明表：

参数	类型	必需	描述
username	string	是	Redshift数据库用户名
password	string	是	Redshift数据库密码
hostname	string	是	Redshift集群端点地址
port	integer	否	连接端口，默认为5439
database-name	string	是	要连接的数据库名称

代码示例 - Redshift连接配置：

from dataherald.sql_database.models.types import DatabaseConnection, SupportedDialects

# 创建Redshift数据库连接
redshift_conn = DatabaseConnection(
    alias="analytics_redshift",
    dialect=SupportedDialects.REDSHIFT,
    connection_uri="redshift+psycopg2://admin:[e-mail]:5439/production",
    use_ssh=False
)

# Redshift特有的扫描器实现
from dataherald.db_scanner.services.redshift_scanner import RedshiftScanner

redshift_scanner = RedshiftScanner()
# 使用HLL算法进行基数估算
cardinality_values = redshift_scanner.cardinality_values(column, db_engine)

多Schema支持机制

Dataherald为Databricks和Redshift提供了不同的多Schema处理策略：

Databricks多Schema支持：

# Databricks使用URL参数处理多schema
def add_schema_in_uri(self, connection_uri: str, schema: str, dialect: str) -> str:
    if dialect == "databricks":
        # 移除现有schema参数
        connection_uri = re.sub(r"&schema=[^&]*", "", connection_uri)
        return f"{connection_uri}&schema={schema}"

Redshift Schema处理：

# Redshift默认不支持多schema配置
if database_connection.dialect in ["redshift", "awsathena", "mssql", "mysql"]:
    raise SchemaNotSupportedError(
        "Schema not supported for this db",
        description=f"The {database_connection.dialect} dialect doesn't support schemas"
    )

安全与加密机制

Dataherald对所有敏感连接信息采用Fernet加密算法进行保护：

from dataherald.utils.encrypt import FernetEncrypt

# 连接字符串加密
fernet_encrypt = FernetEncrypt()
encrypted_uri = fernet_encrypt.encrypt(connection_uri)

# 使用时解密
decrypted_uri = fernet_encrypt.decrypt(encrypted_uri)

连接池与性能优化

Dataherald实现了连接池管理机制，避免频繁创建和销毁数据库连接：

class DBConnections:
    db_connections = {}

    @staticmethod
    def add(uri, engine):
        DBConnections.db_connections[uri] = engine

    @staticmethod
    def get(uri):
        return DBConnections.db_connections.get(uri)

错误处理与重试机制

系统为不同的数据库类型提供了专门的错误处理：

class SSHInvalidDatabaseConnectionError(CustomError):
    pass

class InvalidDBConnectionError(CustomError):
    pass

class SQLInjectionError(CustomError):
    pass

实际应用场景

企业数据湖查询： 通过Databricks连接，企业用户可以自然语言查询Delta Lake中的数据：

-- 生成的SQL示例
SELECT * FROM main.sales.orders WHERE order_date >= '2024-01-01'

数据仓库分析： Redshift连接支持复杂的分析查询：

-- 生成的Redshift SQL
SELECT product_category, SUM(sales_amount) 
FROM sales_fact 
WHERE date_trunc('month', sale_date) = '2024-01-01'
GROUP BY product_category

Dataherald的Databricks和Redshift连接实现体现了现代数据平台集成的最佳实践，通过统一的接口设计、安全的数据处理和优化的性能特性，为企业用户提供了强大而灵活的自然语言数据查询能力。

多schema支持与跨库查询

Dataherald作为企业级自然语言转SQL引擎，在多数据源支持方面提供了强大的多schema管理能力和跨schema查询功能。这一特性使得用户能够在复杂的数据库环境中无缝地进行数据查询和分析。

多schema架构设计

Dataherald的多schema支持采用了一种智能的架构设计，通过统一的连接管理机制来处理不同数据库系统的schema差异。系统支持以下数据库的多schema操作：

数据库类型	Schema支持状态	连接URI格式示例
Snowflake	✅ 完全支持	snowflake://user:pass@org-account/database/schema
BigQuery	✅ 完全支持	bigquery://project/dataset
Databricks	✅ 完全支持	databricks://token@host?schema=schema_name
PostgreSQL	✅ 完全支持	postgresql://user:pass@host/db?options=-csearch_path=schema
MySQL	❌ 不支持	仅支持默认schema
SQL Server	❌ 不支持	仅支持默认schema

多schema连接配置

在创建数据库连接时，可以通过schemas参数指定多个schema：

# 多schema连接配置示例
{
    "alias": "enterprise_data",
    "use_ssh": false,
    "connection_uri": "snowflake://user:password@organization-account/database",
    "schemas": ["sales", "marketing", "finance", "operations"]
}

这种配置方式允许Dataherald同时访问多个schema中的表，为跨schema查询奠定了基础。

跨schema查询机制

Dataherald的SQL生成引擎具备智能的跨schema查询能力。当用户提出涉及多个schema的自然语言查询时，系统会自动识别相关表并生成正确的跨schema SQL语句。

flowchart TD
    A[用户自然语言查询] --> B[查询解析与意图识别]
    B --> C[识别涉及的表和schema]
    C --> D{是否跨schema?}
    D -->|是| E[构建跨schema SQL]
    D -->|否| F[构建单schema SQL]
    E --> G[SQL语法验证与优化]
    F --> G
    G --> H[执行查询并返回结果]

表关联与上下文管理

在多schema环境中，Dataherald通过以下机制管理表关联和上下文：

1. Schema感知的表扫描：系统会扫描所有配置的schema，收集表结构和元数据信息
2. 智能表识别：基于表名、列名和描述信息，识别跨schema的表关联关系
3. 上下文过滤：根据查询涉及的schema，自动过滤无关的上下文信息

# 表过滤逻辑示例
def filter_tables_by_schema(db_scan: List[TableDescription], schemas: List[str]):
    """根据schema过滤表信息"""
    if not schemas:
        return db_scan
    return [table for table in db_scan if table.schema_name in schemas]

跨schema查询示例

假设我们有一个包含多个schema的Snowflake数据库，用户可以进行如下跨schema查询：

自然语言查询: "显示销售部门的员工在2023年的业绩数据"

生成的SQL:

SELECT 
    e.employee_name,
    s.sales_amount,
    d.department_name
FROM sales.employee_sales s
JOIN hr.employees e ON s.employee_id = e.employee_id
JOIN hr.departments d ON e.department_id = d.department_id
WHERE d.department_name = 'Sales'
AND EXTRACT(YEAR FROM s.sale_date) = 2023
ORDER BY s.sales_amount DESC;

权限与安全考虑

在多schema环境中，Dataherald充分考虑了权限管理和安全性：

1. 连接级别权限：数据库连接使用具有适当权限的凭据
2. Schema访问控制：通过配置的schemas参数限制可访问的schema范围
3. 数据加密：所有连接信息在存储时进行加密处理
4. 查询验证：生成的SQL语句会进行语法和权限验证

性能优化策略

为了确保跨schema查询的性能，Dataherald实现了以下优化策略：

1. 上下文限制：只加载相关schema的表信息到上下文窗口
2. 智能缓存：对频繁访问的跨schema查询结果进行缓存
3. 查询优化：生成优化的JOIN语句和WHERE条件
4. 并行处理：对多个schema的元数据采集进行并行处理

错误处理与回退机制

当跨schema查询遇到问题时，Dataherald提供了完善的错误处理机制：

1. Schema不存在错误：检测并提示不存在的schema
2. 表不存在错误：验证表在各schema中的存在性
3. 权限不足错误：捕获并报告权限相关的错误
4. 连接超时处理：实现连接超时的重试机制

最佳实践建议

在使用Dataherald的多schema功能时，建议遵循以下最佳实践：

1. 合理的schema划分：按业务领域合理划分schema，避免过度碎片化
2. 一致的命名规范：在不同schema中使用一致的表和列命名规范
3. 适当的权限管理：为Dataherald连接配置最小必要的权限
4. 定期元数据更新：定期更新schema和表的元数据信息
5. 监控与优化：监控跨schema查询性能，适时进行优化

通过以上机制和策略，Dataherald为企业用户提供了强大而可靠的多schema支持和跨schema查询能力，使得复杂的多数据源环境下的自然语言查询变得简单高效。

Dataherald通过系统化的多数据源集成架构，为企业用户提供了强大而灵活的自然语言转SQL能力。系统支持主流关系型数据库（PostgreSQL、MySQL）和现代数据仓库（Snowflake、BigQuery、Databricks、Redshift），具备统一的连接管理、多schema支持、安全加密和性能优化等特性。智能的跨schema查询机制和错误处理能力确保了在复杂数据环境下的查询准确性和可靠性。通过遵循最佳实践，企业可以充分利用Dataherald的多数据源支持能力，实现高效的自然语言数据查询和分析。