在 mssql-python 中使用稀疏列和列集

稀疏列是 Microsoft SQL Server 中一种用于优化包含许多可为 NULL 的列的表中 NULL 值存储的功能。 客户端应用程序将稀疏列视为常规列。 mssql-python驱动像读写其他列一样，无需特殊处理。

最适合：

• NULL 值占比为 20%–50% 以上的表。
• 带有变量属性的文档存储。
• EAV（实体-属性-值）模式。
• 传感器数据，带有许多可选读数。

创建稀疏列

在表格模式中定义稀疏列，方法是在经常保留NULL值的列上添加 SPARSE NULL 修饰符。

基本稀疏列表

创建一个带有稀疏列的表，用于可选属性。

CREATE TABLE ProductAttributes (
    ProductID INT PRIMARY KEY,
    ProductName NVARCHAR(100) NOT NULL,
    -- Sparse columns for optional attributes
    Color NVARCHAR(50) SPARSE NULL,
    Size NVARCHAR(20) SPARSE NULL,
    Weight DECIMAL(10,2) SPARSE NULL,
    Material NVARCHAR(100) SPARSE NULL,
    Warranty INT SPARSE NULL,
    Manufacturer NVARCHAR(100) SPARSE NULL
);

已设置列

添加一个列集，同时提供对所有稀疏列的XML访问。

CREATE TABLE ProductAttributesWithSet (
    ProductID INT PRIMARY KEY,
    ProductName NVARCHAR(100) NOT NULL,
    -- Column set provides XML access to all sparse columns
    SparseAttributes XML COLUMN_SET FOR ALL_SPARSE_COLUMNS,
    -- Sparse columns
    Color NVARCHAR(50) SPARSE NULL,
    Size NVARCHAR(20) SPARSE NULL,
    Weight DECIMAL(10,2) SPARSE NULL,
    Material NVARCHAR(100) SPARSE NULL,
    Warranty INT SPARSE NULL,
    Manufacturer NVARCHAR(100) SPARSE NULL
);

插入稀疏列数据

像操作常规列一样，按名称向稀疏列中插入数据。

插入单个列

插入具有特定稀疏列值的产品。

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

# Create the table with sparse columns
cursor.execute("DROP TABLE IF EXISTS ProductAttributes")
cursor.execute("""
    CREATE TABLE ProductAttributes (
        ProductID INT PRIMARY KEY,
        ProductName NVARCHAR(100) NOT NULL,
        Color NVARCHAR(50) SPARSE NULL,
        Size NVARCHAR(20) SPARSE NULL,
        Weight DECIMAL(10,2) SPARSE NULL,
        Material NVARCHAR(100) SPARSE NULL,
        Warranty INT SPARSE NULL,
        Manufacturer NVARCHAR(100) SPARSE NULL
    )
""")

# Insert with some sparse columns populated
cursor.execute("""
    INSERT INTO ProductAttributes (ProductID, ProductName, Color, Size)
    VALUES (%(id)s, %(name)s, %(color)s, %(size)s)
""", {"id": 1, "name": "T-Shirt", "color": "Blue", "size": "Large"})

# Insert with different sparse columns
cursor.execute("""
    INSERT INTO ProductAttributes (ProductID, ProductName, Weight, Material)
    VALUES (%(id)s, %(name)s, %(weight)s, %(material)s)
""", {"id": 2, "name": "Coffee Mug", "weight": 0.35, "material": "Ceramic"})

conn.commit()

通过列集（XML）插入

通过将XML传递到列集，一次性插入多个稀疏列值。

# Create the table with a column set
cursor.execute("DROP TABLE IF EXISTS ProductAttributesWithSet")
cursor.execute("""
    CREATE TABLE ProductAttributesWithSet (
        ProductID INT PRIMARY KEY,
        ProductName NVARCHAR(100) NOT NULL,
        SparseAttributes XML COLUMN_SET FOR ALL_SPARSE_COLUMNS,
        Color NVARCHAR(50) SPARSE NULL,
        Size NVARCHAR(20) SPARSE NULL,
        Weight DECIMAL(10,2) SPARSE NULL,
        Material NVARCHAR(100) SPARSE NULL,
        Warranty INT SPARSE NULL,
        Manufacturer NVARCHAR(100) SPARSE NULL
    )
""")

# Insert using column set XML
cursor.execute("""
    INSERT INTO ProductAttributesWithSet (ProductID, ProductName, SparseAttributes)
    VALUES (%(id)s, %(name)s, %(xml)s)
""", {
    "id": 3,
    "name": "Laptop Bag",
    "xml": "<Color>Black</Color><Size>Medium</Size><Material>Nylon</Material><Warranty>24</Warranty>"
})
conn.commit()

动态属性插入

创建一个接受动态属性作为字典并自动构建 XML 的函数。

def insert_with_attributes(cursor, product_id: int, name: str, attributes: dict):
    """Insert product with dynamic sparse column attributes."""
    # Build XML for column set
    xml_parts = [f"<{key}>{value}</{key}>" for key, value in attributes.items()]
    attributes_xml = "".join(xml_parts)
    
    cursor.execute("""
        INSERT INTO ProductAttributesWithSet (ProductID, ProductName, SparseAttributes)
        VALUES (%(id)s, %(name)s, %(xml)s)
    """, {"id": product_id, "name": name, "xml": attributes_xml or None})

# Usage
insert_with_attributes(cursor, 4, "Headphones", {
    "Color": "Silver",
    "Warranty": 12,
    "Manufacturer": "AudioTech"
})
conn.commit()

查询稀疏列

通过名称查询稀疏列，或通过列集一次性检索所有稀疏值。

查询单个列

使用标准SELECT语法查询特定的稀疏列。

# Query specific sparse columns
cursor.execute("""
    SELECT ProductID, ProductName, Color, Size
    FROM ProductAttributes
    WHERE Color IS NOT NULL
""")

for row in cursor:
    print(f"{row.ProductName}: {row.Color}, {row.Size}")

通过列集进行查询

从列集中以XML形式检索所有稀疏列值。

# Get column set XML
cursor.execute("""
    SELECT ProductID, ProductName, SparseAttributes
    FROM ProductAttributesWithSet
    WHERE ProductID = %(id)s
""", {"id": 3})

row = cursor.fetchone()
print(f"Product: {row.ProductName}")
print(f"Attributes XML: {row.SparseAttributes}")

在 Python 中解析列集 XML

从列集中解析 XML，并将其转换为 Python 字典，以便更方便地处理。

from xml.etree import ElementTree as ET

def get_product_attributes(cursor, product_id: int) -> dict:
    """Get product with parsed sparse attributes."""
    cursor.execute("""
        SELECT ProductName, SparseAttributes
        FROM ProductAttributesWithSet
        WHERE ProductID = %(id)s
    """, {"id": product_id})
    
    row = cursor.fetchone()
    if row is None:
        return None
    
    result = {"ProductName": row.ProductName}
    
    # Parse XML column set
    if row.SparseAttributes:
        # Wrap in root element for parsing
        xml_str = f"<root>{row.SparseAttributes}</root>"
        root = ET.fromstring(xml_str)
        
        for elem in root:
            result[elem.tag] = elem.text
    
    return result

# Usage
product = get_product_attributes(cursor, 3)
print(product)
# {'ProductName': 'Laptop Bag', 'Color': 'Black', 'Size': 'Medium', 'Material': 'Nylon', 'Warranty': '24'}

用SELECT * 查询

当你使用 SELECT * 时，列集会以单一 XML 列的形式返回，而不是单个稀疏列。

# SELECT * returns column set instead of individual sparse columns
cursor.execute("""
    SELECT * FROM ProductAttributesWithSet WHERE ProductID = %(id)s
""", {"id": 3})

row = cursor.fetchone()
# Returns: ProductID, ProductName, SparseAttributes (not individual columns)
print(f"Columns: {[col[0] for col in cursor.description]}")

显式查询单个列

要从具有列集的表中检索单个稀疏列，可以在SELECT子句中明确列出它们。

# To get individual sparse columns with column set table, list them explicitly
cursor.execute("""
    SELECT ProductID, ProductName, Color, Size, Weight, Material, Warranty, Manufacturer
    FROM ProductAttributesWithSet
    WHERE ProductID = %(id)s
""", {"id": 3})

# Now each sparse column is available as separate property
row = cursor.fetchone()
print(f"Color: {row.Color}, Material: {row.Material}")

更新稀疏列

按名称更新单个稀疏列，或通过列集 XML 一次性替换所有稀疏值。

更新各列

使用标准SQL UPDATE 语法更新特定的稀疏列值。

cursor.execute("""
    UPDATE ProductAttributes
    SET Color = %(color)s, Weight = %(weight)s
    WHERE ProductID = %(id)s
""", {"id": 1, "color": "Red", "weight": 0.2})
conn.commit()

通过列集进行更新

通过直接更新列集 XML 来替换所有稀疏列值。

# Replace all sparse column values via column set
cursor.execute("""
    UPDATE ProductAttributesWithSet
    SET SparseAttributes = %(xml)s
    WHERE ProductID = %(id)s
""", {
    "id": 3,
    "xml": "<Color>Navy</Color><Size>Large</Size><Material>Leather</Material>"
})
conn.commit()
# Note: This clears any sparse columns not included in the XML

通过列集进行部分更新

只更新特定的稀疏属性，同时保留未包含在更新中的其他属性值。

# To update only specific attributes, merge with existing
def update_attributes(cursor, product_id: int, updates: dict):
    """Update specific sparse attributes while preserving others."""
    # Get current attributes
    cursor.execute("""
        SELECT Color, Size, Weight, Material, Warranty, Manufacturer
        FROM ProductAttributesWithSet
        WHERE ProductID = %(id)s
    """, {"id": product_id})
    
    row = cursor.fetchone()
    if row is None:
        raise ValueError(f"Product {product_id} not found")
    
    # Merge updates
    current = {
        "Color": row.Color,
        "Size": row.Size,
        "Weight": row.Weight,
        "Material": row.Material,
        "Warranty": row.Warranty,
        "Manufacturer": row.Manufacturer
    }
    
    for key, value in updates.items():
        current[key] = value
    
    # Build XML with non-null values
    xml_parts = []
    for key, value in current.items():
        if value is not None:
            xml_parts.append(f"<{key}>{value}</{key}>")
    
    cursor.execute("""
        UPDATE ProductAttributesWithSet
        SET SparseAttributes = %(xml)s
        WHERE ProductID = %(id)s
    """, {"id": product_id, "xml": "".join(xml_parts) or None})

# Usage
update_attributes(cursor, 3, {"Color": "Brown", "Warranty": 36})
conn.commit()

动态列模式

构建灵活的查询，动态跨稀疏列搜索，利用允许列表验证列名并防止SQL注入攻击。

查询 EAV 样式数据

使用实体-属性-值（EAV）模式匹配，按特定属性名称和值搜索产品。

def find_products_by_attribute(cursor, attribute_name: str, attribute_value: str) -> list:
    """Find products with specific attribute value."""
    # Validate column name against allowed sparse columns to prevent SQL injection
    allowed_columns = {"Color", "Size", "Weight", "Material", "Warranty", "Manufacturer"}
    if attribute_name not in allowed_columns:
        raise ValueError(f"Invalid attribute: {attribute_name}")
    
    cursor.execute(f"""
        SELECT ProductID, ProductName, {attribute_name}
        FROM ProductAttributesWithSet
        WHERE {attribute_name} = %(value)s
    """, {"value": attribute_value})
    
    return cursor.fetchall()

# Find all blue products
blue_products = find_products_by_attribute(cursor, "Color", "Blue")

灵活属性搜索

搜索同时符合多个可选属性的产品。

def search_by_attributes(cursor, **attributes) -> list:
    """Search products by multiple optional attributes."""
    # Validate column names against allowed sparse columns to prevent SQL injection
    allowed_columns = {"Color", "Size", "Weight", "Material", "Warranty", "Manufacturer"}
    invalid = set(attributes.keys()) - allowed_columns
    if invalid:
        raise ValueError(f"Invalid attributes: {invalid}")
    
    conditions = ["1=1"]  # Always true base condition
    params = {}
    
    for i, (key, value) in enumerate(attributes.items()):
        if value is not None:
            conditions.append(f"{key} = %(attr_{i})s")
            params[f"attr_{i}"] = value
    
    query = f"""
        SELECT ProductID, ProductName, SparseAttributes
        FROM ProductAttributesWithSet
        WHERE {' AND '.join(conditions)}
    """
    
    cursor.execute(query, params)
    return cursor.fetchall()

# Search by multiple attributes
results = search_by_attributes(cursor, Color="Black", Material="Nylon")

性能注意事项

评估稀疏柱是否能带来存储效益，评估运营成本，并利用性能监控优化稀疏柱的设计。

何时使用稀疏列

适合使用稀疏列的对象包括：

• NULL 值占比超过 60%-70% 的列。
• 宽表格，带有许多可选列。
• 以存储优化为优先的工作负载。

以下情况下避免使用稀疏列：

• 大多数行都包含值（每个非 NULL 值会增加 4 字节的开销）。
• 该列频繁用于 WHERE 子句。
• 该列是聚类索引的一部分。

查看储蓄

比较同一表的稀疏版本和非稀疏版本的存储容量。

-- Compare storage with and without sparse
EXEC sp_spaceused 'ProductAttributes';
EXEC sp_spaceused 'ProductAttributesWithoutSparse';

索引注意事项

你可以为稀疏列创建索引。 过滤索引对稀疏数据效果良好，因为它们跳过了NULL行：

CREATE INDEX IX_Products_Color
ON ProductAttributes(Color)
WHERE Color IS NOT NULL;

批量操作

通过在将行传递给 bulkcopy() 之前先在 Python 中构建列集 XML，优化对多个具有稀疏列的产品的插入操作。

带有稀疏柱的批量插入

使用批量文案高效地插入多个带有稀疏属性的产品。

def bulk_insert_with_attributes(conn, products: list[dict]):
    """Bulk insert products with sparse attributes."""
    rows = []
    for product in products:
        attrs = product.get("attributes", {})
        xml = "".join(f"<{k}>{v}</{k}>" for k, v in attrs.items()) or None
        rows.append((product["id"], product["name"], xml))
    
    cursor = conn.cursor()
    result = cursor.bulkcopy("ProductAttributesWithSet", rows)
    return result["rows_copied"]

# Usage
products = [
    {"id": 100, "name": "Widget A", "attributes": {"Color": "Red", "Size": "Small"}},
    {"id": 101, "name": "Widget B", "attributes": {"Weight": 1.5, "Material": "Steel"}},
    {"id": 102, "name": "Widget C", "attributes": {}},  # No sparse attributes
]
bulk_insert_with_attributes(conn, products)
conn.commit()

最佳做法

遵循验证模式，使用列集以增加灵活性，并监控列稀疏度，确保稀疏列设计达到性能和可维护性目标。

验证稀疏列值

在插入数据前，验证所有稀疏属性都符合允许的集合。

# Sparse columns have the same constraints as regular columns
# The SPARSE keyword only affects storage

def validate_and_insert(cursor, product_id: int, name: str, attributes: dict):
    """Insert with validation."""
    allowed_attributes = {"Color", "Size", "Weight", "Material", "Warranty", "Manufacturer"}
    
    invalid = set(attributes.keys()) - allowed_attributes
    if invalid:
        raise ValueError(f"Unknown attributes: {invalid}")
    
    xml_parts = [f"<{k}>{v}</{k}>" for k, v in attributes.items()]
    
    cursor.execute("""
        INSERT INTO ProductAttributesWithSet (ProductID, ProductName, SparseAttributes)
        VALUES (%(id)s, %(name)s, %(xml)s)
    """, {"id": product_id, "name": name, "xml": "".join(xml_parts) or None})

使用列集以提高灵活性

列集简化了对稀疏列的操作：

• 添加新的稀疏列，无需修改代码。
• 存储动态属性。
• 自动序列化和反序列化 XML。

没有列集，你需要显式的列列表。 有了列集，XML 列会自动处理动态属性。

监控 NULL 百分比

分析某列的NULL百分比，判断它是否适合进行稀疏列优化。

def analyze_sparseness(cursor, table: str, column: str) -> float:
    """Check if column is a good sparse candidate."""
    # Validate identifiers to prevent SQL injection
    import re
    if not re.match(r'^[A-Za-z_][A-Za-z0-9_.]*$', table):
        raise ValueError(f"Invalid table name: {table}")
    if not re.match(r'^[A-Za-z_][A-Za-z0-9_]*$', column):
        raise ValueError(f"Invalid column name: {column}")

    cursor.execute(f"""
        SELECT 
            COUNT(*) AS TotalRows,
            SUM(CASE WHEN {column} IS NULL THEN 1 ELSE 0 END) AS NullRows
        FROM {table}
    """)
    
    row = cursor.fetchone()
    null_percentage = (row.NullRows / row.TotalRows * 100) if row.TotalRows > 0 else 0
    
    print(f"Column {column}: {null_percentage:.1f}% NULL")
    print(f"Recommendation: {'Good sparse candidate' if null_percentage > 60 else 'Keep as regular column'}")
    
    return null_percentage

相关内容

• 数据类型映射
• XML数据
• 批量复制操作
• Microsoft SQL 稀疏列