mysql

common sense

主从数据库

就是为了提高数据库的并发性能。你想，假设是单机，读写都在一台MySQL上面完成，性能肯定不高。如果有三台MySQL，一台mater只负责写操作，两台salve只负责读操作

为了保持数据一致，可以在同一线程且同一数据库连接内，如有写入操作，以后的读操作均从主库读取，用于保证数据一致性。

主节点：

1、当主节点上进行 insert、update、delete 操作时，会按照时间先后顺序写入到 binlog 中； 2、当从节点连接到主节点时，主节点会创建一个叫做 binlog dump 的线程；

3、一个主节点有多少个从节点，就会创建多少个 binlog dump 线程；

4、当主节点的 binlog 发生变化的时候，也就是进行了更改操作，binlog dump 线程就会通知从节点 (Push模式)，并将相应的 binlog 内容发送给从节点；

从节点：

当开启主从同步的时候，从节点会创建两个线程用来完成数据同步的工作。

I/O线程： 此线程连接到主节点，主节点上的 binlog dump 线程会将 binlog 的内容发送给此线程。此线程接收到 binlog 内容后，再将内容写入到本地的 relay log。

SQL线程： 该线程读取 I/O 线程写入的 relay log，并且根据 relay log 的内容对从数据库做对应的操作

我们的场景：后端服务使用的是主数据库，metabase 的读数据库使用的是 slave 数据库。

优化

explain mysqlcommand: 查看指令执行效率，注意 possible_keys 与 key, 辨别索引未生效情况。

索引

哇索引真的好有用！用 in 查询，本来要经过百万级别的 row，现在只用个位数级别了！快了好多，好神奇！

• 方法：bitree hash（每加一个索引就会创建一颗B+树）

联表查询时，我出现了索引未生效情况（Explain 显示为 Block nested-loop join 而非 Index Nested-Loop Join），具体原因未知，可能是某表 index 字段的 null 值过多的缘故。后通过 "left join my_rooms FORCE INDEX FOR JOIN (idx_rel_room_id) on my_rooms.rel_room_id = rooms.id" 强制使用 index 来解决

数据类型

• TEXT 最多可以存多少字？

TINYBLOB, TINYTEXT       L + 1 bytes, where L < 2^8    (255 Bytes)
BLOB, TEXT               L + 2 bytes, where L < 2^16   (64 Kilobytes)
MEDIUMBLOB, MEDIUMTEXT   L + 3 bytes, where L < 2^24   (16 Megabytes)
LONGBLOB, LONGTEXT       L + 4 bytes, where L < 2^32   (4 Gigabytes)

所以 TEXT 可以存储的位数是 2^16-1. 像 UTF8 1~4 bytes 每个字符，往上取 4 的话大概是 2^14 - 1即16383个字

shell 基操

mysql -u <username> -h <hostname> -P <port> <database> -p

SHOW CREATE TABLE `a`;
RENAME TABLE `a` TO `b`;
DELETE FROM `a`;

## 插入值
INSERT INTO `areas` 
( pid, name, code  )
VALUES
(111, "aaa", "330114" );
## batch update same value
UPDATE dummy SET myfield=1 WHERE id>1;
## 批量去掉某一列的后半段
update areas set code = SUBSTR(code,1, 6);

## 新增 column
alter table areas add `row` bigint(20) first;
## 复制另一 column 的值到该 column
update areas set row = id;
## 更改 column
ALTER TABLE `areas` change column `row` `row` bigint(20) unsigned NOT NULL AUTO_INCREMENT;
## 设置 column 自增
ALTER TABLE areas MODIFY COLUMN row BIGINT auto_increment;
## 删除 column
alter table `areas` drop column `sort`;

## 更改 constraints
ALTER TABLE areas DROP CONSTRAINT primary key;
ALTER TABLE areas ADD CONSTRAINT primary key(row);

查看各表存储

SELECT table_name ,
  round(((data_length + index_length) / 1024 / 1024), 2) as SIZE_MB,
    round(((data_length) / 1024 / 1024), 2) as DATA_LENGTH_MB,
    round(((index_length) / 1024 / 1024), 2) as INDEX_LENGTH_MB
FROM information_schema.TABLES
WHERE table_schema = DATABASE() ORDER BY SIZE_MB DESC;

真的很麻烦！这里要注意有些表虽然 data_length 不大，但是 index_length 或许会比 data 还大。 该值有缓存，若有对表刚进行了修改，需要强制刷新下数据 ANALYZE TABLE table_name;

删除大量数据

最好先新建表，再将表内数据移回来，效率最高。这样可以避免 row-by-row 的 logging 记录 ref

create table temp_t as
    select *
    from table
    where date >= '2018-01-01';

truncate table t;

insert into t
     select *

存储过程与变量使用与临时表

DROP PROCEDURE IF EXISTS xxx; # 移除上一次声明的过程
DROP TEMPORARY TABLE IF EXISTS results; # 移除临时表若有

DELIMITER $$ # 存储过程里起止符一般更改成 $$ 或 \\
CREATE PROCEDURE xxx()
BEGIN
    DECLARE tableName CHAR(100); # 变量需要在开始声明
    DECLARE tableList CURSOR FOR SELECT table_name FROM information_schema.tables where TABLE_SCHEMA = 'byteox_users';

    OPEN tableList;

    tableListLoop: LOOP

        FETCH tableList INTO tableName; # 循环的使用

        # 在 sql 里使用变量的方式
        SET @VarCount = 0;
        SET @VarSQL = CONCAT('SET @VarCount = (SELECT COUNT(*) FROM ', tableName, ' WHERE tenant_id IS NULL OR tenant_id <> "88888888")');
        PREPARE stmt FROM @VarSQL;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;

                IF @VarCount > 0 THEN
                    SET @AllCount = 0;
                    SET @AllCountSQL = CONCAT('SET @AllCount = (SELECT COUNT(*) FROM ', tableName, ')');
                    PREPARE stmt3 FROM @AllCountSQL;
                    EXECUTE stmt3;
                    DEALLOCATE PREPARE stmt3;

                    SET @s = CONCAT('INSERT INTO results
                    (table_name, count, all_count) VALUES ("', tableName,'", ', @VarCount ,',', @AllCount,');');
                    PREPARE stmt2 FROM @s;
                    EXECUTE stmt2;
                    DEALLOCATE PREPARE stmt2;
                END IF;

    END LOOP;
    CLOSE tableList;

END $$
DELIMITER ; # 将 delimiter 重置回 ;

# 创建临时表
CREATE TEMPORARY TABLE IF NOT EXISTS results (
  table_name VARCHAR(100), count INT, all_count INT);

# 执行过程
call xxx();

# 从临时表读取数据
select * from results;

导入 csv

## 设置编码方式
ALTER DATABASE databasename CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
ALTER TABLE tablename CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

## infile 代表本地文件，要注意编码
load data infile '/Users/dearvikki/Downloads/AMap_adcode_citycode/AMap_adcode_citycode_20210406.csv' 
into table `a` CHARACTER SET utf8 
FIELDS TERMINATED BY ','   
OPTIONALLY ENCLOSED BY ''
LINES TERMINATED BY '\r\n'
IGNORE 1 LINES
(name,adcode,citycode);

关于utf8与utf8mb4

导入导出

# 导入 table
mysql --user=root wholedatabase < whole.database.sql
mysql --user=root databasename < single.table.sql

## if alreay running mysql shell 会覆盖掉之前的数据
use databasename;
source data.sql;

# 导出 table
mysqldump --column-statistics=0 ... databasename -p tablename > xxx.sql # --column-statistics=0 关闭了额外导出的分析数据

一个循环插入的例子

DELIMITER $$
CREATE PROCEDURE test()
BEGIN
  DECLARE counts INT DEFAULT 0;
  DECLARE iteration INT DEFAULT 0;
  DECLARE roleId varchar(32);

    SELECT COUNT(*) INTO counts FROM basic_role where tenant_id = "88888888" AND is_delete = 0;

    SET iteration = 0;

    START TRANSACTION;

      WHILE iteration < counts DO
       SELECT `id` INTO roleId FROM basic_role where tenant_id = "88888888" AND is_delete = 0 LIMIT 1 OFFSET iteration; // 需要循环取出对应条的值

       INSERT INTO basic_role_resource_data 
       (id, tenant_id, role_id, resource_id, is_include_children, create_time, update_time ) 
       VALUES 
       (REPLACE(UUID(),"-",""), "88888888", roleId, "c43ea61c383f069bfcac0d20b1278c77", 1, NOW(), NOW());

              SET iteration = iteration + 1;
       END WHILE;

    COMMIT;


END
$$
DELIMITER ;

call test();
drop procedure test;

与统计有关

baseQueryWrapper.select("DATE_FORMAT(gmt_created, '%Y-%m') AS date_desc");  
baseQueryWrapper.groupBy("MONTH(gmt_created)");

baseQueryWrapper.select("DATE_FORMAT(gmt_created, '%Y-%m-%d') AS date_desc");  
baseQueryWrapper.groupBy("DATE(gmt_created)");

需要注意，如果 groupBy 出来的某个字段对应没有统计值，那么结果里是没有这个字段的。因此也会存在如下问题： - MySQL group by date and count including missing dates - Stack Overflow - sql - generate days from date range - Stack Overflow 为了补足没有数据的日期，在 mysql 里做颇为麻烦，还是取完数据后再在代码层面处理为好。

逻辑查询定式

增

• 判断是否已有同名元素：
• firstOrCreate + RowsAffected
• firstrOrCreate + 存储的 car.ID 和 最后存储完成返回回来的 carID 不一致

改

• 要改的 id 是否存在：可以用 Count
• 前端传空值，后端到底是保存该空值，还是置之不管的问题。对于 go 而言，由于收到的数据被 unmarshal 后按照结构体每个字段都一定会有个默认值，所以要使用指针类型，如果是 nil 型证明前端没有传数据，如果不是且取到的值是空值，则代表要保存该空值。由于 gorm 不能通过update的方式保存空值，所以有个取巧的办法就是将字段定义成指针类型，这样前端没传时则为nil不会更新，前端传空时指针有内容，则会更新。

upsert

不使用 upsert，不好排查问题。一般都是先 find 再 cretae / update 打两条日志。

查询

• 联表查询
• 表大的情况下，用 subquery 比 join 要更好，因为 join 确实会造出一个笛卡尔积，一个超大表。

1v1绑定（A v B）

存在一个对象的表上即可。若存在关系表上，有可拓展的优点。

• 修改关系（包括新建关系，或者修改关系上附加的内容）：

• 全面做法：

  • B 有没有被其他 A 绑定，若有，提示已被绑定

  • A 有没有绑定其他 B，若有，解绑 A 与其他 B

  • A 有没有已绑定 B，若有，走 update，否则走 create

• 简单做法：指定 ownerId, []dataId，先 close 掉所有不在 []dataId 里的 ownerId-dataId 关系对，再遍历 []dataIds, 新增新的关系。// 即每个关系对本身不修改，基本是删了再新建

1v多绑定（A v Bs）

• 使用中间表：与1v1表现差不多。除了后端不用走第2步，A 可以绑定其他 B
• 在 B 上显性的存一个 A 字段：方便很多，可以直接改 B 上的 A 值。不用去关系表里 delete、create 或者 update。但要注意，删除某 A 字段时，要去 B 里重置关联上 A 的 B。

raw 查询定式

outer join

原来 mysql 没有 outer join 或 full outer join, 要靠 left join union right join!!

吼吼 写了一个很厉害的查结果 sql！汇集我不太熟悉的 distinct、group by unix 时间戳、将数据汇总！

# 新增用户数
    select tableA.month as '月份', new_user_count as '新增用户数', login_counts as '登录次数', login_counts_distinct as '登录用户数' from (
                select id, count(id) as new_user_count, 
      -- 注意 FROM_UNIXTIME(时间戳） => 时间格式，以及精确到按时按分也是用 DATE_FORMAT 这个函数
      DATE_FORMAT(FROM_UNIXTIME(create_at), '%m-%Y') as month from users 
                where
                id in (select user_id from room_members) and
                create_at >= 1640966400 
                GROUP BY MONTH(FROM_UNIXTIME(create_at)), YEAR(FROM_UNIXTIME(create_at))
        ) as tableA
    left join
    # 登录次数
        (
            select count(operation_records.desc) as login_counts, DATE_FORMAT(FROM_UNIXTIME(create_at), '%m-%Y') as month from operation_records
            where
            operation_records.desc like "%登录" and
            create_at >= 1640966400 
            GROUP BY MONTH(FROM_UNIXTIME(create_at)), YEAR(FROM_UNIXTIME(create_at))
        ) as tableB
    on tableA.month = tableB.month
    # 登录用户数（次数去重）
    left join
    (
            select count(distinct operation_records.desc) as login_counts_distinct, DATE_FORMAT(FROM_UNIXTIME(create_at), '%m-%Y') as month from operation_records
            where
            operation_records.desc like "%登录" and
            create_at >= 1640966400 
            GROUP BY MONTH(FROM_UNIXTIME(create_at)), YEAR(FROM_UNIXTIME(create_at))
        ) as tableC
    on tableA.month = tableC.month
    order by tableA.month desc

# 查看一张表里有而另一张表里无的数据，往往都可以用 join 或者 子查询
select xx from A left join B where A.a = B.b where A.a is not null
select xx from A wehre A.a not in (select B.b from B)

# Group与count
select activity_date as day, count(distinct user_id) as active_users
from activity
group by activity_date

# 查看最大值/个数的数据
# 可以不用 max，而用 order by + limit 1

写了一个修正数据的 mysql! 其中最外层的 join, 是 update the values of one column based on the results of another query 的通用做法。

update data_belong_relations as table0
join (
    select data_belong_relations.owner_id, data_belong_relations.data_id, employee_task_points.id, employee_task_plans.name
    from employee_task_plans 
    LEFT JOIN data_belong_relations on employee_task_plans.id = data_belong_relations.owner_id
    left join employee_task_points on employee_task_plans.name = employee_task_points.name
    WHERE
    employee_task_plans.task_type = 2 and employee_task_plans.proj_id = "524e66b6b2aa4d67b3513bb879e33e6d"
    and data_belong_relations.type = 3332
    and employee_task_points.proj_id = "524e66b6b2aa4d67b3513bb879e33e6d" and employee_task_points.task_type = 2
    ) as table1 
on table0.data_id = table1.data_id
set table0.data_id = table1.id