1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# 连接 pgsql 并创建数据库
psql -h localhost -U username -W -c "CREATE DATABASE game_community_admin"

# 连接某一个数据库
psql -h 172.16.6.41 -p 5432 -U postgres -d game_community_admin -W

# 查询当前数据库连接数，datname：数据库名称
SELECT pid, usename, application_name, state, count(*) FROM pg_stat_activity WHERE datname = 'game_community_admin';

# 释放当前数据库全部连接，才能删库
SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = 'game_community_admin' AND pid<>pg_backend_pid();




\l	# 查看所有库
\x	# 扩展显示
\c <database>		# 切换数据库
\q	# 退出 sql 命令行
\dt	# 查看此数据库下所有表
\d <tablename>	# 查看表结构

1
2
3
4
5
6
7

CREATE TABLE ROLE (
  -- GENERATED BY DEFAULT AS IDENTITY(START 20000000) 指定起始值
	id INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
);


"id" serial PRIMARY KEY

1

SELECT name, age FROM users WHERE id = 1;

1
2
3
4
5

# 创建覆盖索引
CREATE INDEX idx_name_age ON students (name, age);

# 查询语句，利用覆盖索引
SELECT name, age FROM users WHERE name = 'John';

1
2
3
4
5
6
7
8
9

class User(BaseModel):
    __tablename__ = 'user'
    username = db.Column(db.string(64), nullable=False, index=True)
    org_id = db.Column(db.SMALLINT, nullable=False, comment='组织id')
    delete_id = db.Column(db.INTEGER, nullable=False, default=0)

    __table_args__ = (
        db.UniqueConstraint('username', 'delete_id', name='_username_delete_id_uc'),
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

user_roles = db.Table('sys_user_role',
    db.Column('id', db.Integer, primary_key=True),
    db.Column('user_id', db.Integer, nullable=False),
    db.Column('role_id', db.Integer, nullable=False)
)


def set_user_role(user, args):
    stmt = user_roles.insert().values(**args).returning(user_roles)
    ret = db.session.execute(stmt)
    ret = ret.mappings().one_or_none()
    db.session.commit()

1
2
3
4
5
6
7
8

- 查询表 arguments 字段中的 name 属性
SELECT arguments->>'name' FROM table1;

- 查询 arguments 字段中的 name 属性，其中 arguments 值为数组嵌套字典格式，如 [{}, {}]
select * from table1
where exists 
  (select 1 from jsonb_array_elements(arguments) as nested_data
   where nested_data->>'name' like '%原游戏%');

1
2
3
4
5

SELECT title, content
FROM post a
JOIN 
  (SELECT id FROM post ORDER BY title LIMIT 10000, 20) b
ON a.id = b.id;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

SELECT * 
FROM fx_sharing_cp
WHERE daterange(start_dt, end_dt, '[]') && daterange('2023-12-21', '2023-12-22', '[]');


-- 为了保证高并发场景下数据不一致的问题，读和写最好用一条 sql 语句实现
WITH overlap_check AS (
  SELECT * 
  FROM fx_sharing_cp
  WHERE daterange(start_dt, end_dt, '[]') 
  && daterange('2023-12-21', '2023-12-22', '[]')
)
INSERT INTO fx_sharing_cp (org_id, game_id, pattern, start_dt, end_dt, create_by)
SELECT 1, 1, 1, '2023-12-21', '2023-12-22', 0
WHERE (SELECT COUNT(*) FROM overlap_check) = 0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

SELECT
  tj_date,
  game_id,
  chl_id,
  plat,
  SUM(role_numb) AS role_numb,
  SUM(reg_numb)  AS reg_numb
FROM
  business_day
WHERE
  game_id = 2
GROUP BY
  GROUPING SETS (
    (tj_date, game_id),
    (tj_date, game_id, chl_id),
    (tj_date, game_id, chl_id, plat)
  )

1
2
3
4
5
6
7
8

UPDATE_POST_SCORE = """
    UPDATE post_community SET score = tmp.score
    FROM (VALUES (%s, %s)) AS tmp(id, score)
    WHERE post_community.post_id=tmp.id;
"""

from psycopg2.extras import execute_batch
execute_batch(cursor, UPDATE_POST_SCORE, result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

cursor = self.conn.connection.cursor(name="named_cursor")
cursor.itersize = 5000
cursor.execute(SELECT_POST_COUNTS)
data = cursor.fetchmany(20000)

now = DT.now_time()
while data:
    result = map(lambda x: (
        x[0],
        (x[1] + 2 * x[2] + 3 * x[3]) / (6 * ((round((now - x[-1].replace(tzinfo=now.tzinfo)).total_seconds() / 3600 + 2)) ** 1.8))
    ), data)
    execute_batch(self.cursor, UPDATE_POST_SCORE, result)
    data = cursor.fetchmany(20000)

1
2
3
4
5
6

- 清空表数据并重置主键 id
TRUNCATE TABLE post_community RESTART IDENTITY;


- 删除列后重建表，把仍保留的列重新紧凑写入新文件，然后替换旧文件，清除物理数据
VACUUM FULL qbank_inline_label_repair_backups;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 让 rtime 分区

CREATE TABLE IF NOT EXISTS "event_log" (
	event VARCHAR(50),
	rtime INT NOT NULL DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)::INT,
	distinct_id VARCHAR NOT NULL,
	user_id INT DEFAULT -1,
	args json
) PARTITION BY range (rtime);

-- 不属于任何分区的数据将会插入到默认表中
CREATE TABLE event_log_default PARTITION OF event_log DEFAULT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

from dateutil.relativedelta import relativedelta


def part_table_by_month(sub_table):

    """按月分区表"""
    now = DT.cur_dt()
    next = now + relativedelta(months=1)  # 下一个月
    next_next = next + relativedelta(months=1)  # 下下个月

    part_name = f"{sub_table}_{next.year}{next.month:02d}"
    start = DT.str2ts(f"{next.year}-{next.month}-01", "%Y-%m-%d")
    end = DT.str2ts(f"{next_next.year}-{next_next.month}-01", "%Y-%m-%d") - 1
    try:
        part_ret = part_table(sub_table, start, end)
        logger.info(f"创建分区表成功: {part_ret}")
    except Exception as e:
        logger.critical(f"创建分区表失败: {part_name}, 失败原因：{str(e)}")


def part_table(sub_table: str, start, end) -> str:
    """
    按月创建分区表（当月触发，创建下个月的）
    :param end: 分区结束范围
    :param start: 分区起始范围
    :param sub_table: 主表名
    :return: 分区表名
    """

    sql_command = f"""CREATE TABLE IF NOT EXISTS {sub_table} PARTITION OF {sub_table} FOR VALUES FROM (:start) TO (:end);"""
    try:
        db.session.execute(text(sql_command), {"start": start, "end": end})
        db.session.commit()
        return sub_table
    except Exception as e:
        db.session.rollback()
        raise e

1

SELECT pg_size_pretty(pg_total_relation_size('你的模式名.你的表名')) AS 总大小;

1

SELECT pg_size_pretty(pg_relation_size('你的模式名.你的表名')) AS 数据大小;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

SELECT
    schemaname AS 模式,
    tablename AS 表名,
    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS 总大小
FROM
    pg_tables
WHERE
    schemaname NOT IN ('information_schema', 'pg_catalog')
ORDER BY
    pg_total_relation_size(schemaname||'.'||tablename) DESC;

1

SELECT pg_size_pretty(pg_database_size('你的数据库名')) AS 数据库大小;

1
2
3
4
5
6
7

SELECT
    datname AS 数据库名,
    pg_size_pretty(pg_database_size(datname)) AS 大小
FROM
    pg_database
ORDER BY
    pg_database_size(datname) DESC;

1
2
3
4
5
6
7
8

# 备份数据库，排除`community_log`和 `users`表，仅导出数据不包括结构
pg_dump -h <host> -p <port> -U <user> -d <database> -F p --data-only -T community_log -T users > /data/backup.sql

# 导出某张表
pg_dump -h host -p port -U username -s -t tablename dbname > struct.sql

# 目标数据库上执行 sql 脚本
psql -h <host> -p <port> -U <user> -d <database> -W -f /data/backup.sql

1
2
3
4
5
6
7

# 生成 10 位 base64 编码的密码并以标准字符集输出
openssl rand 10 | openssl base64 -A
vim /etc/redis.conf, 取消 requirepass 行的注释，修改后面的值
sudo systemctl restart redis

# 开启过期 key 事件监听（会有一定的额外消耗）
config set notify-keyspace-events Ex

1
2
3

save 900 1     # 900秒（15分钟）内至少1个key变化
save 300 10    # 300秒（5分钟）内至少10个key变化
save 60 10000  # 60秒内至少10000个key变化

1
2
3
4
5
6

# 启用 RDB（默认已启用）
save 900 1

# 启用 AOF
appendonly yes
appendfsync everysec  # 每秒同步一次（平衡性能与安全）

1
2
3
4
5
6
7

def scan_uk(pattern, count=None):
    """使用 scan command 匹配 keys 并去重"""
    uq_keys = set()
    for key in client.scan_iter(match=pattern, count=count):
        if key not in uq_keys:
            uq_keys.add(key)
            yield key

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def find_order(self, n1: QueryT, n2: QueryT, filters: list = None):
    """
    比较两个节点的顺序
    :param n1:
    :param n2:
    :param filters:
    :return:
        1 n1在前
        -1 n1在后
        None 节点不满足链表关系
    """

    if filters is None:
        filters = []

    data = self.model.query.filter(*filters).all()
    nodes = {self.get_cur(item): item for item in data}

    cur = n1
    while cur:
        if self.get_cur(cur) == self.get_cur(n2):
            return 1
        cur = nodes.get(self.get_next(cur))

    cur = n2
    while cur:
        if self.get_cur(cur) == self.get_cur(n1):
            return -1
        cur = nodes.get(self.get_next(cur))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

WITH RECURSIVE cte AS (
    SELECT * FROM avatar_management
    WHERE pre_id IS NULL and community_id=8  -- 找到链表的起始节点，即 pre_id 为空的节点
    UNION
		SELECT am.*
		FROM avatar_management am 
		JOIN cte ON am.pre_id = cte.id  -- 通过递归连接找到下一个节点
		where am.community_id=8
)
SELECT * FROM cte;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

ZREVRANGE 按分数值递减返回指定区间内的成员，相同 score 的 member 按字典序的逆序排列
>> ZREVRANGE key start stop [WITHSCORES]

向集合中插入元素，评论 id 作为 member，热度值作为 score
>> ZADD key score1 member1 score2 member2 ...

删除元素
>> ZREM key member

查询
>> zrevrange key last last + limit - 1

获取总数
>> zcard key

1
2
3
4

# 获取最新 key
max(keys, key=lambda x: int(x.split("_")[-1]))

last = last + limit if last + limit < total else None