## 1. PL/pgSQL函数

### 1.1 创建简单函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_add(a INTEGER, b INTEGER)
RETURNS INTEGER AS $$
BEGIN
RETURN a + b;
END;
$$ LANGUAGE plpgsql;
“`

### 1.2 创建带默认参数的函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_greet(name VARCHAR DEFAULT ‘World’)
RETURNS VARCHAR AS $$
BEGIN
RETURN ‘Hello, ‘ || name || ‘!’;
END;
$$ LANGUAGE plpgsql;
“`

### 1.3 创建带异常处理的函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_divide(a NUMERIC, b NUMERIC)
RETURNS NUMERIC AS $$
BEGIN
IF b = 0 THEN
RAISE EXCEPTION ‘Division by zero’;
END IF;
RETURN a / b;
EXCEPTION
WHEN OTHERS THEN
RAISE EXCEPTION ‘Error: %’, SQLERRM;
END;
$$ LANGUAGE plpgsql;
“`

### 1.4 创建返回表的函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_get_students()
RETURNS TABLE(id INTEGER, name VARCHAR, age INTEGER)
AS $$
BEGIN
RETURN QUERY
SELECT s.id, s.name, s.age
FROM fgedu_student s;
END;
$$ LANGUAGE plpgsql;
“`

## 2. SQL函数

### 2.1 创建简单SQL函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_square(x INTEGER)
RETURNS INTEGER AS $$
SELECT x * x;
$$ LANGUAGE sql;
“`

### 2.2 创建带参数的SQL函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_calculate_discount(price NUMERIC, discount_rate NUMERIC)
RETURNS NUMERIC AS $$
SELECT price * (1 – discount_rate / 100);
$$ LANGUAGE sql;
“`

## 3. 其他语言函数

### 3.1 PL/Python函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_python_upper(text VARCHAR)
RETURNS VARCHAR AS $$
return text.upper()
$$ LANGUAGE plpython3u;
“`

### 3.2 PL/Perl函数

“`sql
CREATE OR REPLACE FUNCTION fgedu_perl_upper(text VARCHAR)
RETURNS VARCHAR AS $$
return uc($_[0]);
$$ LANGUAGE plperl;
“`

## 1. 创建操作符

### 1.1 创建简单操作符

“`sql
— 首先创建实现函数
CREATE OR REPLACE FUNCTION fgedu_cube(n INTEGER)
RETURNS INTEGER AS $$
BEGIN
RETURN n * n * n;
END;
$$ LANGUAGE plpgsql;

— 创建操作符
CREATE OPERATOR ^# (
PROCEDURE = fgedu_cube,
LEFTARG = INTEGER,
RIGHTARG = NONE
);
“`

### 1.2 创建二元操作符

“`sql
— 首先创建实现函数
CREATE OR REPLACE FUNCTION fgedu_power(base INTEGER, exponent INTEGER)
RETURNS INTEGER AS $$
DECLARE
result INTEGER := 1;
i INTEGER;
BEGIN
FOR i IN 1..exponent LOOP
result := result * base;
END LOOP;
RETURN result;
END;
$$ LANGUAGE plpgsql;

— 创建操作符
CREATE OPERATOR ** (
PROCEDURE = fgedu_power,
LEFTARG = INTEGER,
RIGHTARG = INTEGER
);
“`

### 1.3 创建带优先级和结合性的操作符

“`sql
— 创建操作符，设置优先级为5，左结合
CREATE OPERATOR @@ (
PROCEDURE = fgedu_custom_operation,
LEFTARG = INTEGER,
RIGHTARG = INTEGER,
PRIORITY = 5,
ASSOCIATION = LEFT
);
“`

## 2. 操作符重载

### 2.1 为不同类型重载操作符

“`sql
— 为INTEGER类型创建操作符
CREATE OR REPLACE FUNCTION fgedu_add(integer, integer)
RETURNS integer AS $$
SELECT $1 + $2;
$$ LANGUAGE sql;

— 为NUMERIC类型创建操作符
CREATE OR REPLACE FUNCTION fgedu_add(numeric, numeric)
RETURNS numeric AS $$
SELECT $1 + $2;
$$ LANGUAGE sql;

— 创建操作符
CREATE OPERATOR +# (
PROCEDURE = fgedu_add,
LEFTARG = INTEGER,
RIGHTARG = INTEGER
);

CREATE OPERATOR +# (
PROCEDURE = fgedu_add,
LEFTARG = NUMERIC,
RIGHTARG = NUMERIC
);
“`

## 3. 操作符索引支持

### 3.1 创建操作符类

“`sql
— 创建操作符类
CREATE OPERATOR CLASS gist_point_ops
DEFAULT FOR TYPE point USING gist AS
OPERATOR 1 <-> (point, point),
OPERATOR 2 <@ (point, box), OPERATOR 3 @> (point, box),
OPERATOR 4 ~= (point, point),
FUNCTION 1 gist_point_compress(internal),
FUNCTION 2 gist_point_decompress(internal),
FUNCTION 3 gist_point_consistent(internal, point, integer, oid, internal),
FUNCTION 4 gist_point_union(internal, internal),
FUNCTION 5 gist_point_same(point, point, internal);
“`

## 1. 函数和操作符的管理

### 1.1 查看函数

“`sql
— 查看所有函数
\df

— 查看特定函数
\df fgedu_add

— 查看函数详细信息
SELECT * FROM pg_proc WHERE proname = ‘fgedu_add’;
“`

### 1.2 查看操作符

“`sql
— 查看所有操作符
\do

— 查看特定操作符
\do ^#

— 查看操作符详细信息
SELECT * FROM pg_operator WHERE oprname = ‘^#’;
“`

### 1.3 修改函数

“`sql
— 修改函数定义
CREATE OR REPLACE FUNCTION fgedu_add(a INTEGER, b INTEGER)
RETURNS INTEGER AS $$
BEGIN
RETURN a + b;
END;
$$ LANGUAGE plpgsql;
“`

### 1.4 删除函数和操作符

“`sql
— 删除函数
DROP FUNCTION IF EXISTS fgedu_add(INTEGER, INTEGER);

— 删除操作符
DROP OPERATOR IF EXISTS ^# (INTEGER);
“`

## 2. 扩展SQL的使用

### 2.1 在SQL语句中使用函数

“`sql
— 在SELECT语句中使用
SELECT fgedu_add(1, 2);

— 在WHERE子句中使用
SELECT * FROM fgedu_student WHERE fgedu_age_check(age);

— 在INSERT语句中使用
INSERT INTO fgedu_student (name, age) VALUES (fgedu_normalize_name(‘John Doe’), 20);
“`

### 2.2 在SQL语句中使用操作符

“`sql
— 使用自定义操作符
SELECT 2 ^#; — 计算2的立方

SELECT 2 ** 3; — 计算2的3次方

SELECT 10 +# 5; — 使用重载的加法操作符
“`

## 3. 扩展SQL的性能优化

### 3.1 函数性能优化

“`sql
— 使用IMMUTABLE标记纯函数
CREATE OR REPLACE FUNCTION fgedu_square(x INTEGER)
RETURNS INTEGER AS $$
SELECT x * x;
$$ LANGUAGE sql IMMUTABLE;

— 使用STABLE标记稳定函数
CREATE OR REPLACE FUNCTION fgedu_get_current_time()
RETURNS TIMESTAMP AS $$
SELECT CURRENT_TIMESTAMP;
$$ LANGUAGE sql STABLE;

— 使用VOLATILE标记易变函数
CREATE OR REPLACE FUNCTION fgedu_generate_random()
RETURNS INTEGER AS $$
SELECT floor(random() * 100);
$$ LANGUAGE sql VOLATILE;
“`

### 3.2 操作符性能优化

“`sql
— 为操作符创建索引支持
CREATE INDEX idx_student_age ON fgedu_student USING btree (age);

— 优化操作符实现函数
CREATE OR REPLACE FUNCTION fgedu_fast_operation(a INTEGER, b INTEGER)
RETURNS INTEGER AS $$
BEGIN
— 优化的实现
RETURN a * b;
END;
$$ LANGUAGE plpgsql IMMUTABLE;
“`

import psycopg2
from psycopg2 import pool

# 数据库连接参数
DB_PARAMS = {
“fgedu.net.cn”: “localfgedu.net.cn”,
“port”: 5432,
“fgedudb”: “fgedudb”,
“fgedu”: “fgedu”,
“password”: “fgedu_password”
}

# 创建连接池
postgres_pool = None

def init_pool():
global postgres_pool
postgres_pool = pool.ThreadedConnectionPool(
minconn=5,
maxconn=20,
**DB_PARAMS
)
print(“连接池初始化成功”)

def get_connection():
“””从连接池获取连接”””
return postgres_pool.getconn()

def release_connection(conn):
“””将连接归还到连接池”””
postgres_pool.putconn(conn)

def execute_query(query, params=None):
“””执行SQL查询”””
conn = None
try:
conn = get_connection()
with conn.cursor() as cur:
cur.execute(query, params)
if query.strip().upper().startswith(“SELECT”):
result = cur.fetchall()
conn.commit()
return result
else:
conn.commit()
return cur.rowcount
except Exception as e:
if conn:
conn.rollback()
print(f”执行查询失败: {e}”)
raise
finally:
if conn:
release_connection(conn)

def create_functions():
“””创建自定义函数”””
# 创建计算学生平均成绩的函数
avg_grade_function = “””
CREATE OR REPLACE FUNCTION fgedu_calculate_avg_grade(student_id INTEGER)
RETURNS NUMERIC AS $$
DECLARE
avg_grade NUMERIC;
BEGIN
SELECT AVG(grade) INTO avg_grade
FROM fgedu_enrollment
WHERE student_id = $1;

RETURN COALESCE(avg_grade, 0);
END;
$$ LANGUAGE plpgsql;
“””
execute_query(avg_grade_function)
print(“计算学生平均成绩的函数创建成功”)

# 创建统计课程选课人数的函数
course_count_function = “””
CREATE OR REPLACE FUNCTION fgedu_count_course_enrollment(course_id INTEGER)
RETURNS INTEGER AS $$
DECLARE
count INTEGER;
BEGIN
SELECT COUNT(*) INTO count
FROM fgedu_enrollment
WHERE course_id = $1;

RETURN count;
END;
$$ LANGUAGE plpgsql;
“””
execute_query(course_count_function)
print(“统计课程选课人数的函数创建成功”)

# 创建检查学生年龄的函数
age_check_function = “””
CREATE OR REPLACE FUNCTION fgedu_age_check(age INTEGER)
RETURNS BOOLEAN AS $$
BEGIN
RETURN age >= 18 AND age <= 30; END; $$ LANGUAGE plpgsql; """ execute_query(age_check_function) print("检查学生年龄的函数创建成功") # 创建格式化学生信息的函数 format_student_function = """ CREATE OR REPLACE FUNCTION fgedu_format_student_info(id INTEGER) RETURNS VARCHAR AS $$ DECLARE student_record RECORD; BEGIN SELECT * INTO student_record FROM fgedu_student WHERE id = $1; IF NOT FOUND THEN RETURN 'Student not found'; END IF; RETURN student_record.name || ', ' || student_record.age || ' years old, ' || student_record.major; END; $$ LANGUAGE plpgsql; """ execute_query(format_student_function) print("格式化学生信息的函数创建成功") def test_functions(): """测试自定义函数""" # 测试计算学生平均成绩的函数 print("\n测试计算学生平均成绩的函数:") result = execute_query("SELECT fgedu_calculate_avg_grade(%s)", (1,)) print(f"学生1的平均成绩: {result[0][0]}") # 测试统计课程选课人数的函数 print("\n测试统计课程选课人数的函数:") result = execute_query("SELECT fgedu_count_course_enrollment(%s)", (1,)) print(f"课程1的选课人数: {result[0][0]}") # 测试检查学生年龄的函数 print("\n测试检查学生年龄的函数:") result = execute_query("SELECT fgedu_age_check(%s)", (20,)) print(f"年龄20是否有效: {result[0][0]}") result = execute_query("SELECT fgedu_age_check(%s)", (17,)) print(f"年龄17是否有效: {result[0][0]}") # 测试格式化学生信息的函数 print("\n测试格式化学生信息的函数:") result = execute_query("SELECT fgedu_format_student_info(%s)", (1,)) print(f"学生1的信息: {result[0][0]}") def main(): """主函数""" # 初始化连接池 init_pool() try: # 创建自定义函数 create_functions() # 测试自定义函数 test_functions() finally: # 关闭连接池 postgres_pool.closeall() print("\n连接池关闭成功") if __name__ == "__main__": main()

测试计算学生平均成绩的函数:
学生1的平均成绩: 90.0

测试统计课程选课人数的函数:
课程1的选课人数: 2

测试检查学生年龄的函数:
年龄20是否有效: True
年龄17是否有效: False

测试格式化学生信息的函数:
学生1的信息: 风哥1号, 21 years old, 人工智能

连接池关闭成功

import psycopg2
from psycopg2 import pool

# 数据库连接参数
DB_PARAMS = {
“fgedu.net.cn”: “localfgedu.net.cn”,
“port”: 5432,
“fgedudb”: “fgedudb”,
“fgedu”: “fgedu”,
“password”: “fgedu_password”
}

# 创建连接池
postgres_pool = None

def init_pool():
global postgres_pool
postgres_pool = pool.ThreadedConnectionPool(
minconn=5,
maxconn=20,
**DB_PARAMS
)
print(“连接池初始化成功”)

def get_connection():
“””从连接池获取连接”””
return postgres_pool.getconn()

def release_connection(conn):
“””将连接归还到连接池”””
postgres_pool.putconn(conn)

def execute_query(query, params=None):
“””执行SQL查询”””
conn = None
try:
conn = get_connection()
with conn.cursor() as cur:
cur.execute(query, params)
if query.strip().upper().startswith(“SELECT”):
result = cur.fetchall()
conn.commit()
return result
else:
conn.commit()
return cur.rowcount
except Exception as e:
if conn:
conn.rollback()
print(f”执行查询失败: {e}”)
raise
finally:
if conn:
release_connection(conn)

def create_operators():
“””创建自定义操作符”””
# 创建立方操作符
cube_function = “””
CREATE OR REPLACE FUNCTION fgedu_cube(n INTEGER)
RETURNS INTEGER AS $$
BEGIN
RETURN n * n * n;
END;
$$ LANGUAGE plpgsql;
“””
execute_query(cube_function)

cube_operator = “””
CREATE OPERATOR ^# (
PROCEDURE = fgedu_cube,
LEFTARG = INTEGER,
RIGHTARG = NONE
);
“””
execute_query(cube_operator)
print(“立方操作符创建成功”)

# 创建幂操作符
power_function = “””
CREATE OR REPLACE FUNCTION fgedu_power(base INTEGER, exponent INTEGER)
RETURNS INTEGER AS $$
DECLARE
result INTEGER := 1;
i INTEGER;
BEGIN
FOR i IN 1..exponent LOOP
result := result * base;
END LOOP;
RETURN result;
END;
$$ LANGUAGE plpgsql;
“””
execute_query(power_function)

power_operator = “””
CREATE OPERATOR ** (
PROCEDURE = fgedu_power,
LEFTARG = INTEGER,
RIGHTARG = INTEGER
);
“””
execute_query(power_operator)
print(“幂操作符创建成功”)

# 创建安全除法操作符
safe_divide_function = “””
CREATE OR REPLACE FUNCTION fgedu_safe_divide(a NUMERIC, b NUMERIC)
RETURNS NUMERIC AS $$
BEGIN
IF b = 0 THEN
RETURN NULL;
END IF;
RETURN a / b;
END;
$$ LANGUAGE plpgsql;
“””
execute_query(safe_divide_function)

safe_divide_operator = “””
CREATE OPERATOR /# (
PROCEDURE = fgedu_safe_divide,
LEFTARG = NUMERIC,
RIGHTARG = NUMERIC
);
“””
execute_query(safe_divide_operator)
print(“安全除法操作符创建成功”)

def test_operators():
“””测试自定义操作符”””
# 测试立方操作符
print(“\n测试立方操作符:”)
result = execute_query(“SELECT 2 ^#”)
print(f”2的立方: {result[0][0]}”)
result = execute_query(“SELECT 3 ^#”)
print(f”3的立方: {result[0][0]}”)

# 测试幂操作符
print(“\n测试幂操作符:”)
result = execute_query(“SELECT 2 ** 3″)
print(f”2的3次方: {result[0][0]}”)
result = execute_query(“SELECT 3 ** 4″)
print(f”3的4次方: {result[0][0]}”)

# 测试安全除法操作符
print(“\n测试安全除法操作符:”)
result = execute_query(“SELECT 10 /# 2″)
print(f”10 /# 2: {result[0][0]}”)
result = execute_query(“SELECT 10 /# 0″)
print(f”10 /# 0: {result[0][0]}”)

def main():
“””主函数”””
# 初始化连接池
init_pool()

try:
# 创建自定义操作符
create_operators()

# 测试自定义操作符
test_operators()

finally:
# 关闭连接池
postgres_pool.closeall()
print(“\n连接池关闭成功”)

if __name__ == “__main__”:
main()

测试立方操作符:
2的立方: 8
3的立方: 27

测试幂操作符:
2的3次方: 8
3的4次方: 81

测试安全除法操作符:
10 /# 2: 5.0
10 /# 0: None

连接池关闭成功

## 1. 案例一：业务逻辑封装

### 1.1 需求
– 计算学生的GPA（平均学分绩点）
– 根据成绩等级计算绩点
– 考虑课程学分权重

### 1.2 实现

“`sql
— 创建计算绩点的函数
CREATE OR REPLACE FUNCTION fgedu_calculate_gpa(student_id INTEGER)
RETURNS NUMERIC AS $$
DECLARE
total_points NUMERIC := 0;
total_credits NUMERIC := 0;
rec RECORD;
BEGIN
— 计算总绩点和总学分
FOR rec IN
SELECT e.grade, c.credit
FROM fgedu_enrollment e
JOIN fgedu_course c ON e.course_id = c.id
WHERE e.student_id = student_id AND e.grade IS NOT NULL
LOOP
— 根据成绩计算绩点
CASE
WHEN rec.grade >= 90 THEN
total_points := total_points + (4.0 * rec.credit);
WHEN rec.grade >= 80 THEN
total_points := total_points + (3.0 * rec.credit);
WHEN rec.grade >= 70 THEN
total_points := total_points + (2.0 * rec.credit);
WHEN rec.grade >= 60 THEN
total_points := total_points + (1.0 * rec.credit);
ELSE
total_points := total_points + (0.0 * rec.credit);
END CASE;
total_credits := total_credits + rec.credit;
END LOOP;

— 计算GPA
IF total_credits > 0 THEN
RETURN total_points / total_credits;
ELSE
RETURN 0;
END IF;
END;
$$ LANGUAGE plpgsql;
“`

### 1.3 使用

“`sql
— 查询学生的GPA
SELECT fgedu_calculate_gpa(1);

— 按GPA排序学生
SELECT s.id, s.name, fgedu_calculate_gpa(s.id) AS gpa
FROM fgedu_student s
ORDER BY gpa DESC;
“`

## 2. 案例二：数据验证

### 2.1 需求
– 验证学生信息的有效性
– 确保年龄在合理范围内
– 确保专业名称符合规范

### 2.2 实现

“`sql
— 创建验证学生信息的函数
CREATE OR REPLACE FUNCTION fgedu_validate_student(name VARCHAR, age INTEGER, major VARCHAR)
RETURNS VARCHAR AS $$
BEGIN
— 验证姓名
IF name IS NULL OR name = ” THEN
RETURN ‘Name cannot be empty’;
END IF;

— 验证年龄
IF age < 16 OR age > 35 THEN
RETURN ‘Age must be between 16 and 35’;
END IF;

— 验证专业
IF major IS NULL OR major = ” THEN
RETURN ‘Major cannot be empty’;
END IF;

RETURN ‘Valid’;
END;
$$ LANGUAGE plpgsql;

— 创建插入学生的函数
CREATE OR REPLACE FUNCTION fgedu_insert_student(name VARCHAR, age INTEGER, gender VARCHAR, major VARCHAR)
RETURNS INTEGER AS $$
DECLARE
validation_result VARCHAR;
student_id INTEGER;
BEGIN
— 验证学生信息
validation_result := fgedu_validate_student(name, age, major);
IF validation_result <> ‘Valid’ THEN
RAISE EXCEPTION ‘Validation failed: %’, validation_result;
END IF;

— 插入学生
INSERT INTO fgedu_student (name, age, gender, major)
VALUES (name, age, gender, major)
RETURNING id INTO student_id;

RETURN student_id;
END;
$$ LANGUAGE plpgsql;
“`

### 2.3 使用

“`sql
— 插入学生
SELECT fgedu_insert_student(‘赵六’, 22, ‘男’, ‘计算机科学’);

— 尝试插入无效学生
SELECT fgedu_insert_student(‘钱七’, 15, ‘女’, ‘软件工程’); — 年龄无效
“`

## 3. 案例三：数据转换

### 3.1 需求
– 将JSON格式的学生信息转换为关系型数据
– 支持批量导入学生数据

### 3.2 实现

“`sql
— 创建从JSON导入学生的函数
CREATE OR REPLACE FUNCTION fgedu_import_students_from_json(json_data JSONB)
RETURNS INTEGER AS $$
DECLARE
item JSONB;
count INTEGER := 0;
BEGIN
— 遍历JSON数组
FOR item IN SELECT * FROM jsonb_array_elements(json_data) LOOP
— 插入学生
INSERT INTO fgedu_student (
name,
age,
gender,
major
) VALUES (
item->>’name’,
(item->>’age’)::INTEGER,
item->>’gender’,
item->>’major’
);
count := count + 1;
END LOOP;

RETURN count;
END;
$$ LANGUAGE plpgsql;
“`

### 3.3 使用

“`sql
— 导入学生数据
SELECT fgedu_import_students_from_json(‘[
{“name”: “孙八”, “age”: 20, “gender”: “男”, “major”: “数据科学”},
{“name”: “周九”, “age”: 21, “gender”: “女”, “major”: “人工智能”}
]’::jsonb);
“`