使用 mssql-python 的稀疏欄位與欄位集合

稀疏欄位是 Microsoft SQL 針對擁有多個可空欄位資料表中 NULL 值的儲存優化。 用戶端應用程式將稀疏欄位視為規則欄位。 mssql-python 驅動程式的讀寫方式就像其他欄位一樣，無需特殊處理。

Feature	描述
稀疏欄位	NULL 值使用零儲存空間
欄位集合	所有稀疏欄位的 XML 表示法
寬桌	最多可支援30,000篇專欄

Note

稀疏欄位是伺服器端功能。 mssql-python 驅動程式不需要特殊設定或 API 來處理稀疏欄位。 客戶端唯一可見的差異是：當你使用欄位集合時，它們會回傳稀疏欄位值的 XML 表示。

最適合：

• 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 稀疏欄位