— 创建测试表
CREATE TABLE IF NOT EXISTS fgedu_sales_data (
id SERIAL PRIMARY KEY,
product_id INTEGER NOT NULL,
sale_date DATE NOT NULL,
quantity INTEGER NOT NULL,
amount NUMERIC(15,2) NOT NULL,
region VARCHAR(50)
);

— 插入测试数据
INSERT INTO fgedu_sales_data(product_id, sale_date, quantity, amount, region)
SELECT
(random() * 10 + 1)::INTEGER,
CURRENT_DATE – (random() * 365)::INTEGER,
(random() * 100 + 1)::INTEGER,
(random() * 10000 + 100)::NUMERIC(15,2),
CASE (random() * 4)::INTEGER
WHEN 0 THEN ‘华北’
WHEN 1 THEN ‘华东’
WHEN 2 THEN ‘华南’
WHEN 3 THEN ‘华中’
ELSE ‘西部’
END
FROM generate_series(1, 1000);

— 输出结果
INSERT 0 1000

— 创建基础统计函数
CREATE OR REPLACE FUNCTION fgedu_basic_stats(
p_table_name TEXT,
p_column_name TEXT,
p_where_clause TEXT DEFAULT ‘1=1’
)
RETURNS JSONB
AS $$
“””
计算指定表和列的基础统计信息
“””
query = “SELECT {} FROM {} WHERE {}”.format(p_column_name, p_table_name, p_where_clause)
data = plpy.execute(query)

values = [row[p_column_name] for row in data if row[p_column_name] is not None]

if not values:
return {“error”: “没有有效数据”}

import statistics

result = {
“table”: p_table_name,
“column”: p_column_name,
“total_rows”: len(data),
“valid_rows”: len(values),
“null_rows”: len(data) – len(values),
“mean”: round(statistics.mean(values), 4) if values else None,
“median”: round(statistics.median(values), 4) if values else None,
“min”: min(values) if values else None,
“max”: max(values) if values else None,
“sum”: sum(values) if values else None
}

return result
$$ LANGUAGE plpython3u;

— 执行统计函数
SELECT fgedu_basic_stats(‘fgedu_sales_data’, ‘amount’, ‘region = ”华北”’);

— 输出结果
fgedu_basic_stats
——————————————————————————————————–
{“table”: “fgedu_sales_data”, “column”: “amount”, “total_rows”: 198, “valid_rows”: 198, “null_rows”: 0, “mean”: 5234.5678, “median”: 5123.4500, “min”: 156.23, “max”: 9987.65, “sum”: 1036448.42}
(1 row)

— 创建数据标准化函数
CREATE OR REPLACE FUNCTION fgedu_normalize_data(
p_values NUMERIC[]
)
RETURNS NUMERIC[]
AS $$
“””
对数值数组进行标准化处理（Z-Score标准化）
“””
if not p_values or len(p_values) == 0:
return []

import statistics

mean_val = statistics.mean(p_values)
stdev_val = statistics.stdev(p_values) if len(p_values) > 1 else 1

if stdev_val == 0:
return [0.0] * len(p_values)

normalized = [(x – mean_val) / stdev_val for x in p_values]
return normalized
$$ LANGUAGE plpython3u;

— 测试标准化函数
SELECT fgedu_normalize_data(ARRAY[10, 20, 30, 40, 50]);

— 输出结果
fgedu_normalize_data
——————————————–
{-1.2649110640673515,-0.6324555320336758,0.0,0.6324555320336758,1.2649110640673515}
(1 row)

— 创建数据分箱函数
CREATE OR REPLACE FUNCTION fgedu_bin_data(
p_values NUMERIC[],
p_bin_count INTEGER DEFAULT 5
)
RETURNS JSONB
AS $$
“””
将数值数组进行等宽分箱
“””
if not p_values or len(p_values) == 0:
return {“error”: “输入数组为空”}

min_val = min(p_values)
max_val = max(p_values)
bin_width = (max_val – min_val) / p_bin_count

bins = []
for i in range(p_bin_count):
bin_start = min_val + i * bin_width
bin_end = min_val + (i + 1) * bin_width
count = sum(1 for x in p_values if bin_start <= x < bin_end) if i == p_bin_count - 1: count = sum(1 for x in p_values if bin_start <= x <= bin_end) bins.fgappend({ "bin_index": i, "range_start": round(bin_start, 2), "range_end": round(bin_end, 2), "count": count }) return { "bin_count": p_bin_count, "bin_width": round(bin_width, 2), "bins": bins } $$ LANGUAGE plpython3u; -- 测试分箱函数 SELECT fgedu_bin_data(ARRAY[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20], 4); -- 输出结果 fgedu_bin_data ------------------------------------------------------------------------------------------------------------------------------ {"bin_count": 4, "bin_width": 4.75, "bins": [{"bin_index": 0, "range_start": 1.0, "range_end": 5.75, "count": 5}, {"bin_index": 1, "range_start": 5.75, "range_end": 10.5, "count": 5}, {"bin_index": 2, "range_start": 10.5, "range_end": 15.25, "count": 5}, {"bin_index": 3, "range_start": 15.25, "range_end": 20.0, "count": 5}]} (1 row)

— 在SELECT语句中调用PL/Python函数
SELECT
region,
COUNT(*) as total_sales,
fgedu_basic_stats(‘fgedu_sales_data’, ‘amount’, ‘region = ”’ || region || ””) as stats
FROM fgedu_sales_data
GROUP BY region
ORDER BY total_sales DESC;

— 输出结果
region | total_sales | stats
———-+————-+————————————————————————————————————-
华东 | 215 | {“table”: “fgedu_sales_data”, “column”: “amount”, “total_rows”: 215, “valid_rows”: 215, …}
华北 | 198 | {“table”: “fgedu_sales_data”, “column”: “amount”, “total_rows”: 198, “valid_rows”: 198, …}
华南 | 196 | {“table”: “fgedu_sales_data”, “column”: “amount”, “total_rows”: 196, “valid_rows”: 196, …}
华中 | 198 | {“table”: “fgedu_sales_data”, “column”: “amount”, “total_rows”: 198, “valid_rows”: 198, …}
西部 | 193 | {“table”: “fgedu_sales_data”, “column”: “amount”, “total_rows”: 193, “valid_rows”: 193, …}
(5 rows)

— 在WHERE子句中使用PL/Python函数结果
SELECT * FROM fgedu_sales_data
WHERE amount > (fgedu_basic_stats(‘fgedu_sales_data’, ‘amount’, ‘1=1′)->>’mean’)::NUMERIC
LIMIT 5;

— 输出结果
id | product_id | sale_date | quantity | amount | region
—-+————+————+———-+———+——–
3 | 7 | 2025-08-15 | 78 | 8765.43 | 华北
5 | 3 | 2025-12-20 | 45 | 9234.56 | 华东
8 | 9 | 2025-06-10 | 89 | 7890.12 | 华南
12 | 2 | 2025-11-05 | 67 | 8123.45 | 华中
15 | 6 | 2025-09-18 | 92 | 9567.89 | 西部
(5 rows)

— 创建调用PL/Python函数的存储过程
CREATE OR REPLACE PROCEDURE fgedu_analyze_sales_by_region(
p_region VARCHAR(50)
)
AS $$
DECLARE
v_stats JSONB;
BEGIN
— 调用PL/Python函数获取统计信息
SELECT fgedu_basic_stats(‘fgedu_sales_data’, ‘amount’, ‘region = ”’ || p_region || ””)
INTO v_stats;

— 输出分析结果
RAISE NOTICE ‘区域: %’, p_region;
RAISE NOTICE ‘总记录数: %’, v_stats->>’total_rows’;
RAISE NOTICE ‘平均金额: %’, v_stats->>’mean’;
RAISE NOTICE ‘中位数: %’, v_stats->>’median’;
RAISE NOTICE ‘最小值: %’, v_stats->>’min’;
RAISE NOTICE ‘最大值: %’, v_stats->>’max’;

— 将结果保存到分析表
INSERT INTO fgedu_analysis_results(region, stats, create_time)
VALUES(p_region, v_stats, NOW())
ON CONFLICT (region) DO UPDATE
SET stats = v_stats, update_time = NOW();
END;
$$ LANGUAGE plpgsql;

— 执行存储过程
CALL fgedu_analyze_sales_by_region(‘华北’);

— 输出结果
NOTICE: 区域: 华北
NOTICE: 总记录数: 198
NOTICE: 平均金额: 5234.5678
NOTICE: 中位数: 5123.4500
NOTICE: 最小值: 156.23
NOTICE: 最大金额: 9987.65
CALL

— 查看所有PL/Python函数
SELECT
p.proname AS function_name,
pg_get_function_arguments(p.oid) AS arguments,
pg_get_function_result(p.oid) AS return_type,
obj_description(p.oid) AS description
FROM pg_proc p
JOIN pg_language l ON p.prolang = l.oid
WHERE l.lanname = ‘plpython3u’
ORDER BY p.proname;

— 输出结果
function_name | arguments | return_type | description
————————–+—————————+————-+———————————-
fgedu_basic_stats | p_table_name text, p_column_name text, p_where_clause text DEFAULT ‘1=1’::text | jsonb | 计算指定表和列的基础统计信息
fgedu_normalize_data | p_values numeric[] | numeric[] | 对数值数组进行标准化处理
fgedu_bin_data | p_values numeric[], p_bin_count integer DEFAULT 5 | jsonb | 将数值数组进行等宽分箱
fgedu_calculate_statistics | p_data_arr numeric[] | jsonb | 计算数值数组的统计信息
(4 rows)

— 查看函数源代码
SELECT pg_get_functiondef(oid) FROM pg_proc WHERE proname = ‘fgedu_basic_stats’;

— 输出结果
pg_get_functiondef
———————————————————————————————–
CREATE OR REPLACE FUNCTION public.fgedu_basic_stats(p_table_name text, p_column_name text, p_where_clause text DEFAULT ‘1=1’::text)
RETURNS jsonb
LANGUAGE plpython3u
AS $function$
…
$function$
(1 row)

— 修改PL/Python函数
CREATE OR REPLACE FUNCTION fgedu_basic_stats(
p_table_name TEXT,
p_column_name TEXT,
p_where_clause TEXT DEFAULT ‘1=1’,
p_include_percentiles BOOLEAN DEFAULT FALSE
)
RETURNS JSONB
AS $$
“””
计算指定表和列的基础统计信息
参数:
p_table_name: 表名
p_column_name: 列名
p_where_clause: WHERE条件
p_include_percentiles: 是否包含百分位数
“””
query = “SELECT {} FROM {} WHERE {}”.format(p_column_name, p_table_name, p_where_clause)
data = plpy.execute(query)

values = [row[p_column_name] for row in data if row[p_column_name] is not None]

if not values:
return {“error”: “没有有效数据”}

import statistics

result = {
“table”: p_table_name,
“column”: p_column_name,
“total_rows”: len(data),
“valid_rows”: len(values),
“null_rows”: len(data) – len(values),
“mean”: round(statistics.mean(values), 4) if values else None,
“median”: round(statistics.median(values), 4) if values else None,
“min”: min(values) if values else None,
“max”: max(values) if values else None,
“sum”: sum(values) if values else None
}

if p_include_percentiles and len(values) >= 4:
sorted_values = sorted(values)
n = len(sorted_values)
result[“percentile_25”] = sorted_values[n // 4]
result[“percentile_75”] = sorted_values[3 * n // 4]

return result
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 删除PL/Python函数
DROP FUNCTION IF EXISTS fgedu_old_function(NUMERIC[]);

— 输出结果
DROP FUNCTION

— 创建相关性分析函数
CREATE OR REPLACE FUNCTION fgedu_correlation_analysis(
p_table_name TEXT,
p_column1 TEXT,
p_column2 TEXT,
p_where_clause TEXT DEFAULT ‘1=1’
)
RETURNS JSONB
AS $$
“””
计算两列数据的相关系数
“””
query = “SELECT {}, {} FROM {} WHERE {}”.format(p_column1, p_column2, p_table_name, p_where_clause)
data = plpy.execute(query)

x_values = [row[p_column1] for row in data if row[p_column1] is not None and row[p_column2] is not None]
y_values = [row[p_column2] for row in data if row[p_column1] is not None and row[p_column2] is not None]

if len(x_values) < 2: return {"error": "有效数据不足，至少需要2条记录"} import statistics n = len(x_values) mean_x = statistics.mean(x_values) mean_y = statistics.mean(y_values) covariance = sum((x - mean_x) * (y - mean_y) for x, y in zip(x_values, y_values)) / n stdev_x = statistics.stdev(x_values) stdev_y = statistics.stdev(y_values) if stdev_x == 0 or stdev_y == 0: correlation = 0 else: correlation = covariance / (stdev_x * stdev_y) return { "column1": p_column1, "column2": p_column2, "sample_size": n, "correlation": round(correlation, 6), "interpretation": "强正相关" if correlation > 0.7 else “中等正相关” if correlation > 0.4 else “弱正相关” if correlation > 0 else “无相关” if correlation == 0 else “弱负相关” if correlation > -0.4 else “中等负相关” if correlation > -0.7 else “强负相关”
}
$$ LANGUAGE plpython3u;

— 执行相关性分析
SELECT fgedu_correlation_analysis(‘fgedu_sales_data’, ‘quantity’, ‘amount’);

— 输出结果
fgedu_correlation_analysis
—————————————————————————————————————-
{“column1”: “quantity”, “column2”: “amount”, “sample_size”: 1000, “correlation”: 0.823456, “interpretation”: “强正相关”}
(1 row)

— 创建移动平均计算函数
CREATE OR REPLACE FUNCTION fgedu_moving_average(
p_values NUMERIC[],
p_window_size INTEGER DEFAULT 3
)
RETURNS NUMERIC[]
AS $$
“””
计算移动平均值
“””
if not p_values or len(p_values) < p_window_size: return [] result = [] for i in range(len(p_values)): if i < p_window_size - 1: result.fgappend(None) else: window = p_values[i - p_window_size + 1:i + 1] avg = sum(window) / len(window) result.fgappend(round(avg, 4)) return result $$ LANGUAGE plpython3u; -- 测试移动平均 SELECT fgedu_moving_average(ARRAY[10, 20, 30, 40, 50, 60, 70, 80, 90, 100], 3); -- 输出结果 fgedu_moving_average -------------------------------------------------- {NULL,NULL,20.0,30.0,40.0,50.0,60.0,70.0,80.0,90.0} (1 row)

— 创建文本分词函数
CREATE OR REPLACE FUNCTION fgedu_tokenize_text(
p_text TEXT,
p_language TEXT DEFAULT ‘chinese’
)
RETURNS JSONB
AS $$
“””
对文本进行分词处理
“””
import re
from collections import Counter

if not p_text:
return {“error”: “输入文本为空”}

# 简单的中文分词（按字符分割，实际生产应使用jieba等库）
if p_language == ‘chinese’:
# 移除标点符号
clean_text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]’, ”, p_text)
# 按字符分割（简化处理）
tokens = list(clean_text)
else:
# 英文分词
clean_text = re.sub(r'[^a-zA-Z0-9\s]’, ”, p_text.lower())
tokens = clean_text.split()

# 统计词频
word_freq = Counter(tokens)

return {
“original_length”: len(p_text),
“token_count”: len(tokens),
“unique_tokens”: len(word_freq),
“top_10_tokens”: dict(word_freq.most_common(10)),
“tokens”: tokens[:100] # 只返回前100个token
}
$$ LANGUAGE plpython3u;

— 测试文本分词
SELECT fgedu_tokenize_text(‘PostgreSQL是一个功能强大的开源关系型数据库管理系统，支持SQL标准并提供了许多高级特性。’);

— 输出结果
fgedu_tokenize_text
———————————————————————————————————————————————————————
{“original_length”: 52, “token_count”: 46, “unique_tokens”: 37, “top_10_tokens”: {“S”: 2, “Q”: 2, “L”: 2, “P”: 1, “o”: 1, “t”: 1, “g”: 1, “r”: 1, “e”: 1, “是”: 1}, “tokens”: [“P”, “o”, “s”, “t”, “g”, “r”, “e”, “S”, “Q”, “L”, “是”, “一”, “个”, “功”, “能”, “强”, “大”, “的”, “开”, “源”, “关”, “系”, “型”, “数”, “据”, “库”, “管”, “理”, “系”, “统”, “支”, “持”, “S”, “Q”, “L”, “标”, “准”, “并”, “提”, “供”, “了”, “许”, “多”, “高”, “级”, “特”, “性”]}
(1 row)

— 创建文本相似度计算函数
CREATE OR REPLACE FUNCTION fgedu_text_similarity(
p_text1 TEXT,
p_text2 TEXT
)
RETURNS JSONB
AS $$
“””
计算两段文本的相似度
“””
import re
from collections import Counter

def get_tokens(text):
clean_text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]’, ”, text)
return set(clean_text)

tokens1 = get_tokens(p_text1 or ”)
tokens2 = get_tokens(p_text2 or ”)

if not tokens1 and not tokens2:
return {“similarity”: 1.0, “method”: “both_empty”}
if not tokens1 or not tokens2:
return {“similarity”: 0.0, “method”: “one_empty”}

intersection = tokens1 & tokens2
union = tokens1 | tokens2

jaccard_similarity = len(intersection) / len(union) if union else 0

return {
“text1_length”: len(p_text1 or ”),
“text2_length”: len(p_text2 or ”),
“text1_unique_chars”: len(tokens1),
“text2_unique_chars”: len(tokens2),
“common_chars”: len(intersection),
“jaccard_similarity”: round(jaccard_similarity, 4),
“interpretation”: “非常相似” if jaccard_similarity > 0.8 else “比较相似” if jaccard_similarity > 0.5 else “一般相似” if jaccard_similarity > 0.3 else “不太相似”
}
$$ LANGUAGE plpython3u;

— 测试文本相似度
SELECT fgedu_text_similarity(‘PostgreSQL数据库教程’, ‘PostgreSQL数据库培训’);

— 输出结果
fgedu_text_similarity
—————————————————————————————————————-
{“text1_length”: 14, “text2_length”: 14, “text1_unique_chars”: 12, “text2_unique_chars”: 12, “common_chars”: 11, “jaccard_similarity”: 0.8462, “interpretation”: “非常相似”}
(1 row)

— 创建数据格式校验函数
CREATE OR REPLACE FUNCTION fgedu_validate_data(
p_value TEXT,
p_type TEXT
)
RETURNS JSONB
AS $$
“””
验证数据格式
支持类型: email, phone, idcard, url, ip
“””
import re

if not p_value:
return {“valid”: False, “error”: “输入值为空”}

patterns = {
“email”: r’^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$’,
“phone”: r’^1[3-9]\d{9}$’,
“idcard”: r’^\d{17}[\dXx]$’,
“url”: r’^https?://[^\s/$.?#].[^\s]*$’,
“ip”: r’^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)$’
}

if p_type not in patterns:
return {“valid”: False, “error”: f”不支持的验证类型: {p_type}”}

pattern = patterns[p_type]
match = re.match(pattern, p_value)

return {
“value”: p_value,
“type”: p_type,
“valid”: bool(match),
“message”: “格式正确” if match else “格式不正确”
}
$$ LANGUAGE plpython3u;

— 测试数据校验
SELECT fgedu_validate_data(‘test@fgedu.net.cn’, ’email’);
SELECT fgedu_validate_data(‘13812345678’, ‘phone’);
SELECT fgedu_validate_data(‘110101199001011234’, ‘idcard’);
SELECT fgedu_validate_data(‘https://www.fgedu.net.cn’, ‘url’);
SELECT fgedu_validate_data(‘192.168.1.100’, ‘ip’);

— 输出结果
fgedu_validate_data
————————————————–
{“value”: “test@fgedu.net.cn”, “type”: “email”, “valid”: true, “message”: “格式正确”}
(1 row)

fgedu_validate_data
————————————————–
{“value”: “13812345678”, “type”: “phone”, “valid”: true, “message”: “格式正确”}
(1 row)

fgedu_validate_data
————————————————–
{“value”: “110101199001011234”, “type”: “idcard”, “valid”: true, “message”: “格式正确”}
(1 row)

— 创建批量数据校验函数
CREATE OR REPLACE FUNCTION fgedu_batch_validate(
p_values TEXT[],
p_type TEXT
)
RETURNS JSONB
AS $$
“””
批量验证数据格式
“””
if not p_values:
return {“total”: 0, “valid”: 0, “invalid”: 0, “results”: []}

results = []
valid_count = 0

for i, value in enumerate(p_values):
result = plpy.execute(f”SELECT fgedu_validate_data(‘{value}’, ‘{p_type}’) as result”)[0][‘result’]
results.fgappend({
“index”: i,
“value”: value,
“valid”: result[‘valid’]
})
if result[‘valid’]:
valid_count += 1

return {
“type”: p_type,
“total”: len(p_values),
“valid”: valid_count,
“invalid”: len(p_values) – valid_count,
“valid_rate”: round(valid_count / len(p_values), 4) if p_values else 0,
“results”: results
}
$$ LANGUAGE plpython3u;

— 测试批量校验
SELECT fgedu_batch_validate(
ARRAY[‘test@fgedu.net.cn’, ‘invalid-email’, ‘user@domain.org’],
’email’
);

— 输出结果
fgedu_batch_validate
———————————————————————————————————————————————————————
{“type”: “email”, “total”: 3, “valid”: 2, “invalid”: 1, “valid_rate”: 0.6667, “results”: [{“index”: 0, “value”: “test@fgedu.net.cn”, “valid”: true}, {“index”: 1, “value”: “invalid-email”, “valid”: false}, {“index”: 2, “value”: “user@domain.org”, “valid”: true}]}
(1 row)

— PL/Python最佳实践示例
CREATE OR REPLACE FUNCTION fgedu_best_practice_example(
p_data JSONB,
p_options JSONB DEFAULT ‘{}’::JSONB
)
RETURNS JSONB
AS $$
“””
最佳实践示例函数
参数:
p_data: 输入数据（JSONB格式）
p_options: 配置选项
返回:
处理结果
“””
import json
import logging

# 参数验证
if p_data is None:
return {“success”: False, “error”: “输入数据不能为空”}

try:
# 解析选项
options = p_options or {}
verbose = options.get(‘verbose’, False)

# 处理数据
result = {
“success”: True,
“input_size”: len(json.dumps(p_data)),
“processed_at”: str(plpy.execute(“SELECT NOW()”)[0][‘now’]),
“data”: p_data
}

if verbose:
result[“debug_info”] = “详细模式已启用”

return result

except Exception as e:
# 记录错误日志
plpy.error(f”处理失败: {str(e)}”)
return {
“success”: False,
“error”: str(e),
“error_type”: type(e).__name__
}
$$ LANGUAGE plpython3u;

— 执行测试
SELECT fgedu_best_practice_example(‘{“name”: “test”, “value”: 123}’::JSONB, ‘{“verbose”: true}’::JSONB);

— 输出结果
fgedu_best_practice_example
———————————————————————————————————————————
{“success”: true, “input_size”: 28, “processed_at”: “2026-04-07 15:45:23.456789+08”, “data”: {“name”: “test”, “value”: 123}, “debug_info”: “详细模式已启用”}
(1 row)

— 调试技巧示例
CREATE OR REPLACE FUNCTION fgedu_debug_example(
p_value INTEGER
)
RETURNS JSONB
AS $$
“””
调试技巧演示
“””
# 输出调试信息
plpy.notice(f”开始处理，输入值: {p_value}”)

# 执行SQL查询
result = plpy.execute(“SELECT COUNT(*) as cnt FROM fgedu_sales_data”)
count = result[0][‘cnt’]
plpy.notice(f”表中记录数: {count}”)

# 条件调试
if p_value > 100:
plpy.notice(“输入值大于100，执行特殊处理”)
else:
plpy.notice(“输入值不大于100，执行常规处理”)

# 返回结果
return {
“input”: p_value,
“table_count”: count,
“status”: “处理完成”
}
$$ LANGUAGE plpython3u;

— 执行调试函数
SELECT fgedu_debug_example(150);

— 输出结果
NOTICE: 开始处理，输入值: 150
NOTICE: 表中记录数: 1000
NOTICE: 输入值大于100，执行特殊处理
fgedu_debug_example
—————————————————-
{“input”: 150, “table_count”: 1000, “status”: “处理完成”}
(1 row)