Vou falar um pouco sobre particionamento de tabelas no MySQL e compartilhar algumas procedures que desenvolvi durante uma projeto com a PerformanceDB para facilitar o processo de criação, adição e remoção de partições.

Em alguns ambientes encontramos casos de tabelas que crescem exponencialmente, gerando assim problemas de performance e dificultando a manutenção do banco de dados, pois qualquer intervenção se torna muito custosa em termos de utilização de recursos do servidor (processamento, memória e i/o de disco).

Para esses casos podemos utilizar a técnica de particionamento, permitindo assim a quebra desses dados em arquivos físicos menores o que traz ganhos de performance e facilita a manutenção.

Imagine que temos uma tabela com 1 milhão de registros, o particionamento consiste em identificar uma coluna comumente usada em suas consultas e defini-la como chave de particionamento, o objetivo é distribuir os dados em arquivos físicos separados o mais uniformemente possível e garantir que quando for acessar um dado, o mysql saiba exatamente a partição em que ele se encontra, evitando assim uma varredura de toda a tabela.

Então se considerarmos o exemplo da imagem acima, foi feita uma distribuição dos dados igualmente e cada partição contém apenas 250 mil registros. Na prática, cada arquivo físico das partições seria como uma tabela menor, mas de forma transparente para o usuário.

Principais critérios de particionamento:

RANGE Partitioning
LIST Partitioning
HASH Partitioning
KEY Partitioning

IMPORTANTE! Por simplicidade, as tabelas nos exemplos utilizados não usam nenhuma chave. Na pratica as tabelas particionadas provavelmente terão chaves exclusivas (PRIMARY KEY E UNIQUE KEY) e a regra é que TODAS as colunas usadas na expressão de particionamento devem fazer parte de cada chave exclusiva que a tabela possa ter.

A seguir abordarei de forma prática exemplos de como particionar uma tabela de acordo com cada critério.

RANGE Partitioning

No exemplo abaixo temos a tabela employees com a seguinte estrutura:

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
    , hired DATE NOT NULL DEFAULT '1970-01-01'
    , separated DATE NOT NULL DEFAULT '9999-12-31'
    , job_code INT NOT NULL
	, store_id INT NOT NULL 
);

Vamos utilizar o particionamento por intervalo (RANGE) para separar os empregados pela loja (store) em que trabalham utilizando a coluna store_id:

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
	, hired DATE NOT NULL DEFAULT '1970-01-01'
	, separated DATE NOT NULL DEFAULT '9999-12-31'
	, job_code INT NOT NULL
	, store_id INT NOT NULL 
) 
PARTITION BY RANGE (store_id) 
(    
	PARTITION p0 VALUES LESS THAN (6)
	, PARTITION p1 VALUES LESS THAN (11)
	, PARTITION p2 VALUES LESS THAN (16)
	, PARTITION p3 VALUES LESS THAN (21) 
);

Entendendo a instrução:

PARTITION p0 VALUES LESS THAN (6)

Valores menores que 6

Desta forma os empregados que trabalham nas lojas de store_id de 1 a 5 estarão na partição p0.

PARTITION p1 VALUES LESS THAN (11)

Valores menores que 11

Logo os que trabalham nas lojas de store_id de 6 a 10 estarão na partição p1 e assim por diante.

O particionamento por intervalo (by RANGE) também nos permite utilizar algumas funções para distribuição dos dados, dentre as mais comuns são a funções que trabalham com data.

No exemplo abaixo iremos particionar a mesma tabela desta vez utilizando a coluna hired (contratado) e usaremos a função “year()” de forma a termos uma separação de empregados de acordo com o ano em que foram contratados.

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
	, hired DATE NOT NULL DEFAULT '1970-01-01'
	, separated DATE NOT NULL DEFAULT '9999-12-31'
	, job_code INT
	, store_id INT 
) 
PARTITION BY RANGE ( YEAR(hired) ) 
(    
	PARTITION p0 VALUES LESS THAN (1991)
	, PARTITION p1 VALUES LESS THAN (1996)
	, PARTITION p2 VALUES LESS THAN (2001)
	, PARTITION p3 VALUES LESS THAN MAXVALUE 
);

Entendendo a instrução:

PARTITION p0 VALUES LESS THAN (1991)

Valores menores que 1991

Desta forma os empregados que foram contratados antes do ano 1991 estarão na partição p0. Seguindo a mesma lógica do particionamento anterior.

PARTITION p3 VALUES LESS THAN MAXVALUE

Valores menores que o maior valor da coluna

O empregados contratados no ano atual (sendo maior que 2000) até o ano 2001 estarão na partição p3 .

LIST Partitioning

O particionamento por lista (by LIST) é semelhante ao exemplo anterior, mas ao invés de selecionarmos um intervalo de valores iremos definir uma lista de valores para cada partição:

Utilizaremos a mesma tabela employees:

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
	, hired DATE NOT NULL DEFAULT '1970-01-01'
	, separated DATE NOT NULL DEFAULT '9999-12-31'
	, job_code INT
	, store_id INT 
);

Suponha que existam 20 locadoras de vídeo distribuídas em 4 franquias, conforme mostrado na tabela a seguir.

Região	Número store_id
Norte	3, 5, 6, 9, 17
Leste	1, 2, 10, 11, 19, 20
Oeste	4, 12, 13, 14, 18
Central	7, 8, 15, 16

Na instrução abaixo iremos particionar a tabela de forma que os registros dos empregados que pertençam a uma mesma região sejam armazenados em uma mesma partição:

Usaremos a coluna store_id como chave de particionamento.

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
	, hired DATE NOT NULL DEFAULT '1970-01-01'
	, separated DATE NOT NULL DEFAULT '9999-12-31'
	, job_code INT
	, store_id INT 
) 
PARTITION BY LIST(store_id) 
(    
	PARTITION pNorth VALUES IN (3,5,6,9,17)
	, PARTITION pEast VALUES IN (1,2,10,11,19,20)
	, PARTITION pWest VALUES IN (4,12,13,14,18)
	, PARTITION pCentral VALUES IN (7,8,15,16) 
);

ATENÇÃO! Usando este tipo de particionamento, não serão permitidos a inserção de registros cujo store_id não estejam contemplados em nenhuma lista.

Supondo que foi aberta uma nova franquia na região norte com store_id igual a 21, ao tentarmos inserir o registro de um empregado nesta loja ocorrerá um erro, pois o mysql não saberá onde armazenar esse valor.

Ao tentar inserir o registro abaixo irá ocorrer um erro:

INSERT INTO employees (id,fname,lname,hired,separated,job_code,store_id)
VALUES (345,'Joseph','Smith','2019-05-08','9999-12-31',32,21);

Error Code: 1526. Table has no partition for value 21

Para resolvermos isso precisamos reorganizar a partição Norte e incluir o valor da nova store_id:

Copy to Clipboard
Syntax HighlighterALTER TABLE employees 
REORGANIZE PARTITION pNorth INTO 
(     
	PARTITION pNorth VALUES IN (3,5,6,9,17,21) 
);

Com isso, ao tentar inserir novamente o registro anterior não teremos mais o erro.

HASH Partitioning

O particionamento por HASH é usado principalmente para garantir uma distribuição uniforme dos dados entre um número de partições predeterminadas.
Seguindo com os nossos exemplos, vamos particionar a tabela employees por HASH.
A instrução a seguir particiona a tabela employees gerando um hash da coluna store_id dividindo os dados em 4 partições:

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
	, hired DATE NOT NULL DEFAULT '1970-01-01'
	, separated DATE NOT NULL DEFAULT '9999-12-31'
	, job_code INT
	, store_id INT 
) 
PARTITION BY HASH(store_id) PARTITIONS 4;

Obs. Se não incluir o “PARTITIONS 4” será criada apenas uma partição.

Você pode consultar as informações das partições criadas na tabela INFORMATION_SCHEMA.PARTITIONS usando a “query” abaixo:

Copy to Clipboard
Syntax HighlighterSELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, PARTITION_METHOD, PARTITION_EXPRESSION FROM INFORMATION_SCHEMA.PARTITIONS WHERE TABLE_SCHEMA = 'teste' AND TABLE_NAME = 'employees' ORDER BY PARTITION_ORDINAL_POSITION;

TABLE_SCHEMA	TABLE_NAME	PARTITION_NAME	PARTITION_METHOD	PARTITION_EXPRESSION
teste	employees	p0	HASH	`store_id`
teste	employees	p1	HASH	`store_id`
teste	employees	p2	HASH	`store_id`
teste	employees	p3	HASH	`store_id`

Também é possível usar uma função SQL que retorne um número inteiro.

No exemplo abaixo estamos particionando a tabela com base no ano que o funcionário foi contratado:

Copy to Clipboard
Syntax HighlighterCREATE TABLE employees 
(    
	id INT NOT NULL
	, fname VARCHAR(30)
	, lname VARCHAR(30)
	, hired DATE NOT NULL DEFAULT '1970-01-01'
	, separated DATE NOT NULL DEFAULT '9999-12-31'
	, job_code INT
	, store_id INT 
) 
PARTITION BY HASH( YEAR(hired) ) PARTITIONS 4;

A função de hash mais eficiente é aquela que opera em uma única coluna da tabela e cujo valor aumenta ou diminui consistentemente com o valor da coluna, pois isso permite “podar” os intervalos das partições.
Por exemplo, a função year() é uma boa opção, porém nem toda mudança na coluna hired ira causar uma variação na expressão year(hired), uma opção que obteríamos uma boa variação seria usando a função to_days() pois desta forma obteríamos uma boa variação da expressão to_days(hired) com as mudanças de valores na coluna hired.

KEY Partitioning

O particionamento por chave (BY KEY) é semelhante ao particionamento por HASH, a diferença é que no particionamento por HASH o usuário define a expressão a ser utilizada já no particionamento por chave o próprio Mysql fornece a expressão. No NDB Cluster é utilizado MD5(), para as tabelas que usam outros mecanismos de armazenamento como o INNODB a função de particionamento empregada pelo Mysql é baseada no algoritmo PASSWORD().

As regras de sintaxe seguem os mesmos moldes que já utilizamos até o momento nos exemplos anteriores.

Observe o comando a seguir:

Copy to Clipboard
Syntax HighlighterCREATE TABLE k1 (
   id INT NOT NULL PRIMARY KEY,
   name VARCHAR(20)
)
PARTITION BY KEY()
PARTITIONS 2;

Note que não foi referenciada nenhuma coluna na instrução PARTITION BY KEY(), de modo que o mysql irá particionar a tabela pela sua chave primária, neste caso a coluna id. Se a chave primária possuir mais colunas, todas farão parte da chave de particionamento.

Se não houver chave primária, mas houver uma chave exclusiva, a chave exclusiva será usada para a chave de particionamento:

Copy to Clipboard
Syntax HighlighterCREATE TABLE k1 
(    
	id INT NOT NULL
	, name VARCHAR(20)
	, UNIQUE KEY (id) 
) 
PARTITION BY KEY() PARTITIONS 2;

No exemplo abaixo ocorrerá um erro ao tentarmos criar a tabela tk, pois ela não possui chaves exclusivas:

Copy to Clipboard
Syntax HighlighterCREATE TABLE tk 
(
	col1 INT
	, col2 CHAR(5)
	, col3 DATE
)    
PARTITION BY KEY() PARTITIONS 4;

Podemos particionar uma tabela que não possui chaves exclusivas mas para isso temos que definir explicitamente a coluna a ser utilizada como chave de particionamento:

Copy to Clipboard

A seguir abordaremos os comandos de gerenciamento de suas partições.

Gerenciamento das partições RANGE e LIST:

Agora que já aprendeu os conceitos básicos de como criar uma tabela particionada, irei abordar como manipular as partições.
Vamos à prática!
Abaixo seguem os comandos de criação da tabela tr e inserção de 10 registros nela:

Copy to Clipboard
Syntax HighlighterCREATE TABLE tr 
(
	id INT
	, name VARCHAR(50)
	, purchased DATE
)         
PARTITION BY RANGE( YEAR(purchased) ) 
(             
	PARTITION p0 VALUES LESS THAN (1990)
	, PARTITION p1 VALUES LESS THAN (1995)
	, PARTITION p2 VALUES LESS THAN (2000)
	, PARTITION p3 VALUES LESS THAN (2005)
	, PARTITION p4 VALUES LESS THAN (2010)
	, PARTITION p5 VALUES LESS THAN (2015)          
); 

INSERT INTO tr VALUES         
(1, 'desk organiser', '2003-10-15')
, (2, 'alarm clock', '1997-11-05')
, (3, 'chair', '2009-03-10')
, (4, 'bookcase', '1989-01-10')
, (5, 'exercise bike', '2014-05-09')
, (6, 'sofa', '1987-06-05')
, (7, 'espresso maker', '2011-11-22')
, (8, 'aquarium', '1992-08-04')
, (9, 'study desk', '2006-09-16')
, (10, 'lava lamp', '1998-12-25');

Os dados foram distribuídos de acordo com o ano de compra (year(purchased)).
Podemos observar que ao executarmos a consulta abaixo o mysql varre apenas uma partição, como demonstrado no explain:

Copy to Clipboard
Syntax HighlighterEXPLAIN
SELECT * FROM tr
        WHERE purchased BETWEEN '1995-01-01' AND '1999-12-31';




+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+-------------+
| id | select_type | table | partitions | type | possible_keys | key  | key_len | ref  | rows | filtered | Extra       |
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+-------------+
|  1 | SIMPLE      | tr    | p2         | ALL  | NULL          | NULL | NULL    | NULL |    2 |    50.00 | Using where |
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+-------------+

Você pode fazer uma consulta diretamente na partição p2:

Copy to Clipboard
Syntax HighlighterSELECT * FROM tr PARTITION (p2);

+------+-------------+------------+
| id   | name        | purchased  |
+------+-------------+------------+
|    2 | alarm clock | 1997-11-05 |
|   10 | lava lamp   | 1998-12-25 |
+------+-------------+------------+

Para expurgo de dados essa é uma funcionalidade extremamente útil e performática pois o impacto ser torna muito baixo, caso seja necessário remover os registros dos anos correspondentes a partição p2 (1995 a 1999), podemos excluir apenas a partição p2 com o comando abaixo:

Copy to Clipboard

Muito cuidado! O comando acima irá apagar todos os dados armazenados na partição correspondente!

Caso queira excluir mais de uma partição basta colocar mais de um nome separado por virgula:

Copy to Clipboard

Agora surgiu a necessidade de adicionar uma nova partição para armazenar os produtos comprados entre 2015 e 2019, você pode adicionar a nova partição com o comando abaixo:

Copy to Clipboard

Note que ao criar as partições por intervalo seguimos uma ordem crescente para cada valor.

No exemplo abaixo os anos são postos de forma crescente e isto é mandatório:

Copy to Clipboard

Mas como seria caso identificasse a necessidade de criar um valor menor que o primeiro ano, por exemplo 1985?

Copy to Clipboard

O mysql nos retornaria o erro abaixo:

Copy to Clipboard

Nesses casos vamos precisar usar a instrução REORGANIZE PARTITION para conseguirmos organizar de forma adequada os intervalos mantendo assim a ordem crescentes dos valores:

Copy to Clipboard

O comando reorganizou a partição p0 em dois novos intervalos:

n0 à valores menores que 1985

p0 à valores menores que 1990 (mantemos o nome que era anteriormente)

De forma que a nova estrutura de partições passe a ficar assim:

PARTITION n0 VALUES LESS THAN (1985)

PARTITION p0 VALUES LESS THAN (1990)

PARTITION p1 VALUES LESS THAN (1995)

PARTITION p2 VALUES LESS THAN (2000)

PARTITION p3 VALUES LESS THAN (2005)

PARTITION p4 VALUES LESS THAN (2010)

PARTITION p5 VALUES LESS THAN (2015)

PARTITION p6 VALUES LESS THAN (2020)

Todos os comandos abordados servem tanto para partições do tipo RANGE como para as do tipo LIST.

Gerenciamento das partições HASH e KEY:

Suponha que você tenha uma tabela contendo dados sobre clientes, que é dividida em doze partições de acordo com o mês de assinatura conforme definido a seguir:

Copy to Clipboard

Para reduzir o número de partições de doze para oito executamos o seguinte comando:

Copy to Clipboard

A clausula COALESCE funciona tanto em tabelas particionadas por HASH como para tabelas particionada por KEY.

Copy to Clipboard

O número informado para clausula COALESCE PARTITION é o numero de partições a serem mescladas nas demais, ou seja, as que serão removidas da tabela.

Não é possível remover essas partições, se tentar ocorrerá um erro:

Copy to Clipboard

Se quiser aumentar o número de partições da tabela clients de 12 para 18 execute o comando para adicionar 6 novas partições:

Copy to Clipboard

Procedures de particionamento

Se você utiliza ou pretende utilizar particionamento “by RANGE” dividindo seus dados por dia, mês ou ano, abaixo deixo a documentações de procudres elaboradas por mim(e anexadas ao final desse post) para facilitar o trabalho de inclusão e exclusão de partições.

A seguir abordarei a utilização dessas procedures bem como ao final irei disponibilizar os códigos caso queira usar em seu ambiente:

A estrutura de procedures está dividida da seguinte forma:

create_partition()
add_partition()
drop_partition()

Criando novas partições com create_partition():

Parâmetros:

Nome_database
Nome_tabela
Nome_coluna (“data types” válidos, ‘date’, ‘datetime’, ‘timestamp’)
Tipo_periodo (‘day’, ‘month’, ‘year’)
Set_bin_log (0, 1) caso queira desabilitar a escrita no log binário

Supondo que queira particionar a tabela clients que pertence ao database teste por mês usando a coluna signed:

Copy to Clipboard

Execute a chamada da procedure conforme abaixo:

Copy to Clipboard

Para o exemplo acima, levando em consideração a data atual 07/08/2019 será criada a partição abaixo:

p201908

Caso a tabela possua registros de datas anteriores, serão adicionadas as partições correspondentes dos registros anteriores até a data atual.

Adicionando novas partições com add_partition():

Parâmetros:

Nome_database
Nome_tabela
Nome_coluna (“data types” válidos, ‘date’, ‘datetime’, ‘timestamp’)
Tipo_periodo (‘day’, ‘month’, ‘year’)
Qtd_novas_partições
Set_bin_log (0, 1) caso queira desabilitar a escrita no log binário

Supondo que queira adicionar a tabela clients que uma nova partição.

Execute a chamada da procedure conforme abaixo:

Copy to Clipboard

Caso a tabela não possua partições a procedure add_partition() irá chamar a procudure create_partition() internamente. No exemplo acima se a tabela ainda não for particionada será criada uma partição para o ano/mês corrente e como deseja adicionar uma nova partição conforme informado no parâmetro de quantidade, também será adicionada a partição do ano/mês para 1 mês seguinte.

Para o exemplo acima, levando em consideração a data atual 07/08/2019 serão criadas as duas partições abaixo:

p201908
p201909

Se chamar o comando novamente passando o valor de quantidade para duas novas partições serão criadas as partições conforme abaixo:

Copy to Clipboard

p201910
p201911

Excluindo partições com drop_partition():

Parâmetros:

Nome_database
Nome_tabela
Tipo_periodo (‘day’, ‘month’, ‘year’)
Periodo_mantido (3 serão mantidos dados dos últimos 3 meses)
Set_bin_log (0, 1) caso queira desabilitar a escrita no log binário

Supondo que queira excluir as partições acima de 3 meses da tabela clients que foi particionada por mês.

Levando em consideração o mês atual 201908.

A tabela possui as seguintes partições:

p201809
p201810
p201811
p201812
p201901
p201902
p201903
p201904
p201905
p201906
p201907
p201908
p201909

Execute a chamada da procedure conforme abaixo:

Copy to Clipboard

Serão excluídas todas as partições mais antigas que 201905 ficando apenas:

p201905
p201906
p201907
p201908
p201909

Espero ter contribuído para o entendimento de como particionar as tabelas em seu banco de dados. A intenção foi abordar o assunto de uma forma mais objetiva, tenha em mente que existem vários aspectos que não foram abordados nesse artigo. Se precisar se aprofundar mais no assunto consulte a documentação online.

https://dev.mysql.com/doc/refman/5.7/en/partitioning.html

Caso tenha alguma dúvida ou sugestão sobre o artigo, não deixe de nos escrever, estaremos sempre dispostos a ajudar.

Procedure create_partition

Copy to Clipboard

DELIMITER $$
CREATE PROCEDURE `create_partition`(db_name VARCHAR(100), tb_name VARCHAR(100), col_name VARCHAR(100), partition_by enum('day','month','year'), set_bin_log int)
proc_Exit:BEGIN

DECLARE v_database_name varchar(100);
DECLARE v_table_name varchar(100);
DECLARE v_column_name varchar(100);
DECLARE v_partition_name varchar(100);
DECLARE v_partition_by enum('day','month','year');
DECLARE v_set_bin_log int;

DECLARE v_data_recente date;
DECLARE v_data_passado date;
DECLARE v_data_aux date;
DECLARE v_total_dias int;

DECLARE v_database_count int;
DECLARE v_table_count int;
DECLARE v_column_count int;
DECLARE v_column_datatype varchar(50);
DECLARE v_column_datatype_count int;
DECLARE v_column_on_pk_count int;
DECLARE v_column_on_uk_count int;
DECLARE v_table_pk_count int;
DECLARE v_table_uk_count int;
DECLARE v_partition_count int;
DECLARE v_table_rows_count int;
DECLARE v_count int;

DECLARE v_check_create_partition INT;
DECLARE v_check_create_partition_pk INT;
DECLARE v_check_create_partition_uk INT;
DECLARE v_check_create_partition_pk_uk INT;

-- ---------------------------------------------
 -- SELECT INTO VARIABLES
 -- ---------------------------------------------
 set v_database_name = db_name;
 set v_table_name = tb_name;
 set v_column_name = col_name;
 set v_partition_by = partition_by;
 set v_set_bin_log = set_bin_log;
 
 
set v_data_recente = now();
set v_data_passado = now();
set v_data_aux = v_data_passado;
set v_total_dias = 0;
 
 if v_set_bin_log = 0 then
	set sql_log_bin = 0;
 else
	set sql_log_bin = 1;
 end if;
 
set v_database_count = 0;
set v_table_count = 0;
set v_column_count = 0;
set v_column_datatype_count = 0;
set v_column_on_pk_count = 0;
set v_column_on_uk_count = 0;
set v_table_pk_count = 0;
set v_table_uk_count = 0;
set v_table_rows_count = 0;
set v_check_create_partition = 0;
set v_check_create_partition_pk = 0;
set v_check_create_partition_uk = 0;
set v_check_create_partition_pk_uk = 0;

-- ---------------------------------------------
-- DATABASE EXISTE ?
-- ---------------------------------------------
select count(*) into v_database_count
   from information_schema.schemata
where 
   schema_name = v_database_name;
   
-- ---------------------------------------------
-- TABELA EXISTE ?
-- ---------------------------------------------

select count(*) into v_table_count
   from information_schema.tables
where 
   table_schema = v_database_name
   and table_name = v_table_name;

-- ---------------------------------------------
-- COLUNA EXISTE E ATENDE CRITERIO ?
-- ---------------------------------------------

select count(*) into v_column_count
   from information_schema.COLUMNS
where table_schema = v_database_name
and table_name = v_table_name
and column_name = v_column_name
and data_type in ('date','datetime','timestamp');

-- ---------------------------------------------
-- CHECK TABELA vs. (PRINMARY KEY)
-- ---------------------------------------------                         
 select 
     count(1) into v_table_pk_count
 from information_schema.KEY_COLUMN_USAGE kcu
 inner join information_schema.TABLE_CONSTRAINTS tc on kcu.table_schema = tc.table_schema
     and kcu.table_name = tc.table_name 
     and kcu.constraint_name = tc.constraint_name
 where kcu.table_schema = v_database_name
 and kcu.table_name = v_table_name
 and tc.constraint_type in ('PRIMARY KEY');

-- ---------------------------------------------
-- CHECK TABELA vs. (UNIQUE KEY)
-- ---------------------------------------------                         
 select 
     count(1) into v_table_uk_count
 from information_schema.KEY_COLUMN_USAGE kcu
 inner join information_schema.TABLE_CONSTRAINTS tc on kcu.table_schema = tc.table_schema
     and kcu.table_name = tc.table_name 
     and kcu.constraint_name = tc.constraint_name
 where kcu.table_schema = v_database_name
 and kcu.table_name = v_table_name
 and tc.constraint_type in ('UNIQUE');

-- ---------------------------------------------
-- CHECK COLUNA vs. (PRINMARY KEY)
-- ---------------------------------------------

select 
     count(1) into v_column_on_pk_count
 from information_schema.KEY_COLUMN_USAGE kcu
 inner join information_schema.TABLE_CONSTRAINTS tc on kcu.table_schema = tc.table_schema
     and kcu.table_name = tc.table_name 
     and kcu.constraint_name = tc.constraint_name
 where kcu.table_schema = v_database_name
 and kcu.table_name = v_table_name
 and kcu.column_name = v_column_name
 and tc.constraint_type in ('PRIMARY KEY');   
 
-- ---------------------------------------------
-- CHECK COLUNA vs. (UNIQUE KEY)
-- ---------------------------------------------   
 
select 
    count(1) into v_column_on_uk_count
from information_schema.KEY_COLUMN_USAGE kcu
inner join information_schema.TABLE_CONSTRAINTS tc on kcu.table_schema = tc.table_schema
    and kcu.table_name = tc.table_name 
    and kcu.constraint_name = tc.constraint_name
where kcu.table_schema = v_database_name
and kcu.table_name = v_table_name
and kcu.column_name = v_column_name
and tc.constraint_type in ('UNIQUE');

-- ---------------------------------------------
-- TABELA POSSUI PARTIÇÕES
-- ---------------------------------------------

SELECT count(*) into v_partition_count
	FROM INFORMATION_SCHEMA.PARTITIONS 
WHERE TABLE_SCHEMA = v_database_name
AND TABLE_NAME = v_table_name 
AND PARTITION_NAME IS NOT NULL;

-- ---------------------------------------------
-- TABELA POSSUI REGISTROS?
-- ---------------------------------------------

set @v_table_rows_count_cmd = concat('select count(*) into @v_table_rows_count from ',v_database_name,'.',v_table_name,';');

prepare stmt from @v_table_rows_count_cmd;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

case 
when v_database_count = 0 then select concat(now(),' [ERRO]: Database ',v_database_name ,' não existe ') as menssage;leave  proc_Exit;
when v_table_count = 0 then select concat(now(),' [ERRO]: A tabela ',v_table_name ,' não existe ') as menssage;leave  proc_Exit;
when v_column_count = 0 then select concat(now(),' [ERRO]: A coluna ',v_column_name, ' não existe ou não atende critérios de datatype ') as menssage;leave  proc_Exit;
when v_table_pk_count = 0 and v_table_uk_count = 0 then select concat(now(),' [INFO]: A tabela \"', v_table_name, '\" não possui Primary Key nem Unique Key.') as menssage;set v_check_create_partition=1;
when v_table_pk_count >= 1 and v_table_uk_count >= 1 then select concat(now(),' [INFO]: A tabela \"', v_table_name, '\" possui Primary Key e Unique Key.') as menssage;set v_check_create_partition_pk_uk=1;
when v_table_pk_count = 0 and v_table_uk_count >= 1 then select concat(now(),' [INFO]: A tabela \"', v_table_name, '\" possui apenas Unique Key.')as menssage;set v_check_create_partition_uk=1;
when v_table_pk_count >= 1 and v_table_uk_count = 0 then select concat(now(),' [INFO]: A tabela \"', v_table_name, '\" possui apenas Primary Key.')as menssage;set v_check_create_partition_pk=1;
else begin end;
end case;

case 
when v_check_create_partition_pk_uk = 1 and ( v_column_on_pk_count = 0 or v_column_on_uk_count = 0 ) then select concat(now(),' [ERRO]: A coluna ',v_column_name, ' precisa fazer parte da Primary Key e da Unique Key favor adiciona-la') as menssage;leave  proc_Exit; 
when v_check_create_partition_pk = 1 and v_column_on_pk_count = 0  then select concat(now(),' [ERRO]: A coluna ',v_column_name, ' não faz parte da Primary Key favor adiciona-la') as menssage;leave  proc_Exit; 
when v_check_create_partition_uk = 1 and v_column_on_uk_count = 0 then select concat(now(),' [ERRO]: A coluna ',v_column_name, ' não faz parte da Unique Key favor adiciona-la') as menssage;leave  proc_Exit; 
else begin end;
end case;

if (v_check_create_partition_pk_uk = 1 or v_check_create_partition_pk = 1 or v_check_create_partition_uk = 1 or v_check_create_partition = 1)  then
 
 select concat(now(),' [INFO]: Iniciando o processo de particionamento da \"', v_table_name, '\"')as menssage;
 
	if v_partition_count = 0 then
    
		case 
			when v_partition_by = 'day' then select concat(now(),' [INFO]: A tabela ',v_table_name, ' será particionada por dia'); 
				if @v_table_rows_count > 0 then
					select concat(now(),' [INFO]: A tabela ',v_table_name, ' possui registros, serão adicionadas as partiçoes dos dias correspondentes até o dia atual') as menssage; 
                
					set @v_get_min_date_cmd = concat('select min(',v_column_name,') into @v_get_min_date from ',v_database_name,'.',v_table_name,';');
                    
                    prepare stmt from @v_get_min_date_cmd;
					EXECUTE stmt;
                    DEALLOCATE PREPARE stmt;
					
                    set v_data_passado = @v_get_min_date;
                    set v_data_aux = v_data_passado;
                    
                    select datediff(v_data_recente, v_data_passado) into v_total_dias;
                    
					set @v_control_loop =1;
                    
                    set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' partition by range (to_days(',v_column_name,')) (' );
                    select date_add(v_data_passado, interval 1 day) into @partition_value;
                    select concat('p',date_format(v_data_passado,"%Y%m%d")) into @partition_name;
                    set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (to_days(\'',@partition_value);
                    set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'\')),');
                    
                    
					add_partition_loop: LOOP

select date_add(v_data_passado, interval @v_control_loop day) into v_data_aux;
						select date_add(v_data_aux, interval 1 day) into @partition_value;
						select concat('p',date_format(v_data_aux,"%Y%m%d")) into @partition_name;	
                        set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (to_days(\'',@partition_value);

SET @v_control_loop=@v_control_loop+1;
						IF @v_control_loop=v_total_dias+1 THEN
							set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'\')));');
							LEAVE add_partition_loop;
						END IF;
                        
                        set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'\')),');
                        
					END LOOP add_partition_loop;
                    
                    select concat(now(),' [INFO]: Criando partições para os registros existentes até o dia atual na tabela ',v_table_name) as menssage;
                   
                   -- select @v_create_partition_cmd;
                   
                   prepare stmt from @v_create_partition_cmd;
				   EXECUTE stmt;
				   DEALLOCATE PREPARE stmt;
                    
				else
					select concat(now(),' [INFO]: A tabela ',v_table_name, ' não possui registros, será adicionada a partição do dia atual') as menssage; 
                    
                    
                    
                    set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' partition by range (to_days(',v_column_name,')) (' );
                    select date_add(v_data_passado, interval 1 day) into @partition_value;
                    select concat('p',date_format(v_data_passado,"%Y%m%d")) into @partition_name;
                    set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (to_days(\'',@partition_value);
                    set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'\')));');
                    
                    
                     prepare stmt from @v_create_partition_cmd;
				     EXECUTE stmt;
				     DEALLOCATE PREPARE stmt;
                    
				end if;
            
            
			when v_partition_by = 'month' then 
				if @v_table_rows_count > 0 then
					select concat(now(),' [INFO]: A tabela ',v_table_name, ' possui registros, serão adicionadas as partiçoes dos meses correspondentes até o mês atual') as menssage; 
												
					set @v_get_min_date_cmd = concat('select min(',v_column_name,') into @v_get_min_date from ',v_database_name,'.',v_table_name,';');
                    
                    prepare stmt from @v_get_min_date_cmd;
					EXECUTE stmt;
                    DEALLOCATE PREPARE stmt;
					
                    set v_data_passado = @v_get_min_date;
                    select str_to_date(concat(EXTRACT(YEAR_MONTH FROM v_data_recente),'01'),"%Y%m%d"), str_to_date(concat(EXTRACT(YEAR_MONTH FROM v_data_passado),'01'),"%Y%m%d") into v_data_recente, v_data_passado;
                    set v_data_aux = v_data_passado;
                    
                    SELECT TIMESTAMPDIFF(MONTH,v_data_passado,v_data_recente) into v_total_dias;
                    
					set @v_control_loop =1;
					set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' partition by range (extract(year_month from ',v_column_name,')) (' );
					select EXTRACT(YEAR_MONTH FROM date_add(v_data_passado, interval 1 month)) into @partition_value;
                    select concat('p',EXTRACT(YEAR_MONTH FROM v_data_passado)) into @partition_name;
                    set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);
                    set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'),');						
                     
                     
                     add_partition_loop: LOOP

select date_add(v_data_passado, interval @v_control_loop month) into v_data_aux;
						select extract(year_month from date_add(v_data_aux, interval 1 month)) into @partition_value;
						select concat('p',extract(year_month from v_data_aux)) into @partition_name;	
                        set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);

SET @v_control_loop=@v_control_loop+1;
						IF @v_control_loop=v_total_dias+1 THEN
							set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'));');
							LEAVE add_partition_loop;
						END IF;
                        
                        set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'),');
                        
					END LOOP add_partition_loop;
                                            
					
                                               
                    select concat(now(),' [INFO]: Criando partições para os registros existentes na tabela ',v_table_name) as menssage; 
                     
                     -- select @v_create_partition_cmd;
                     
					prepare stmt from @v_create_partition_cmd;
					EXECUTE stmt;
					DEALLOCATE PREPARE stmt;
                                                
				else
				    					
                    select concat(now(),' [INFO]: A tabela ',v_table_name, ' não possui registros, será adicionada a partição do mês atual') as menssage; 
                                             
                    set @v_table_partition_get_name_and_value_cmd = concat('select distinct concat(\'p\',EXTRACT(YEAR_MONTH FROM now())) as partition_name, EXTRACT(YEAR_MONTH FROM date_add(now(), interval 1 month)) as values_partition into @partition_name, @partition_value;');
                    prepare stmt from @v_table_partition_get_name_and_value_cmd;
					EXECUTE stmt;
					DEALLOCATE PREPARE stmt;
                                                
                    set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' partition by range (extract(year_month from ',v_column_name,')) (' );
                    set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);
					set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'));');
                                            
                                                
                    select concat(now(),' [INFO]: Criando as novas partições na tabela ',v_table_name) as menssage; 
                                                
                                                
					prepare stmt from @v_create_partition_cmd;
					EXECUTE stmt;
					DEALLOCATE PREPARE stmt;
                                                
                    end if;
            
            
			when v_partition_by = 'year' then select concat(now(),' [INFO]: A tabela ',v_table_name, ' será particionada por ano') as menssage; 
            
				if @v_table_rows_count > 0 then
					select concat(now(),' [INFO]: A tabela ',v_table_name, ' possui registros, serão adicionadas as partições dos anos correspondentes até o ano atual') as menssage; 
                    
                    set @v_get_min_date_cmd = concat('select min(',v_column_name,') into @v_get_min_date from ',v_database_name,'.',v_table_name,';');
                    
                    prepare stmt from @v_get_min_date_cmd;
					EXECUTE stmt;
                    DEALLOCATE PREPARE stmt;
					
                    set v_data_passado = @v_get_min_date;
                    set v_data_aux = v_data_passado;
                    
                    if  year(v_data_recente) > year(v_data_passado) then
                    select year(v_data_recente)-year(v_data_passado) as year_diff into v_total_dias;
                    else
						select concat(now(),' [AVISO]: Ano minimo da tabela está no futuro | ano minimo: ',year(v_data_passado),' ano atual: ', year(v_data_recente)) as menssage; 
                        leave  proc_Exit;
                    end if;
                    
                    
					set @v_control_loop =1;
                    
                    set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' partition by range (year(',v_column_name,')) (' );
                    select year(date_add(v_data_passado, interval 1 year)) into @partition_value;
                    select concat('p',year(v_data_passado)) into @partition_name;
                    set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);
                    set @v_create_partition_cmd = concat(@v_create_partition_cmd,@v_create_partition_body,'),');
                    
                    
					add_partition_loop: LOOP

select date_add(v_data_passado, interval @v_control_loop year) into v_data_aux;
						select year(date_add(v_data_aux, interval 1 year)) into @partition_value;
						select concat('p',year(v_data_aux)) into @partition_name;	
                        set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);

select concat(now(),' [ERRO]: Tabela \"', v_table_name, '\" não foi validada para particionamento')as menssage;

end if;

END$$

DELIMITER ;

Procedure add_partition

Copy to Clipboard

DELIMITER $$
CREATE PROCEDURE `add_partition`(db_name VARCHAR(100), tb_name VARCHAR(100), col_name VARCHAR(100), partition_by enum('day','month','year'), partition_qtd int, set_bin_log int)
proc_Exit:BEGIN

DECLARE v_pmax int;
DECLARE v_database_count int;
DECLARE v_table_count int;
DECLARE v_column_count int;
DECLARE v_column_count_partition_key int;
DECLARE v_column_datatype varchar(50);
DECLARE v_column_datatype_count int;
DECLARE v_column_on_pk_count int;
DECLARE v_column_on_uk_count int;
DECLARE v_table_pk_count int;
DECLARE v_table_uk_count int;
DECLARE v_partition_count int;
DECLARE v_table_rows_count int;
DECLARE v_count int;

DECLARE v_data_recente date;
DECLARE v_data_passado date;
DECLARE v_data_aux date;
DECLARE v_total_dias int;

DECLARE v_check_create_partition INT;
DECLARE v_check_create_partition_pk INT;
DECLARE v_check_create_partition_uk INT;
DECLARE v_check_create_partition_pk_uk INT;

-- ---------------------------------------------
 -- SELECT INTO VARIABLES
 -- ---------------------------------------------
 set v_database_name = db_name;
 set v_table_name = tb_name;
 set v_column_name = col_name;
 set v_partition_by = partition_by;
 set v_partition_qtd = partition_qtd;
 set v_set_bin_log = set_bin_log;
 
 set v_data_recente = now();
 set v_data_passado = now();
 set v_data_aux = v_data_passado;
 set v_total_dias = 0;
 
 if v_set_bin_log = 0 then
	set sql_log_bin = 0;
 else
	set sql_log_bin = 1;
 end if;
 
set v_pmax = 0;
set v_database_count = 0;
set v_table_count = 0;
set v_column_count = 0;
set v_column_count_partition_key = 0;
set v_column_datatype_count = 0;
set v_column_on_pk_count = 0;
set v_column_on_uk_count = 0;
set v_table_pk_count = 0;
set v_table_uk_count = 0;
set v_table_rows_count = 0;
set v_check_create_partition = 0;
set v_check_create_partition_pk = 0;
set v_check_create_partition_uk = 0;
set v_check_create_partition_pk_uk = 0;

select count(*) into v_table_count
   from information_schema.tables
where 
   table_schema = v_database_name
   and table_name = v_table_name;

-- ---------------------------------------------
-- COLUNA EXISTE E ATENDE CRITERIO ?
-- ---------------------------------------------

-- ---------------------------------------------
-- TABELA POSSUI PARTIÇÕES ?
-- ---------------------------------------------

SELECT count(*) into v_partition_count
	FROM INFORMATION_SCHEMA.PARTITIONS 
WHERE TABLE_SCHEMA = v_database_name
AND TABLE_NAME = v_table_name 
AND PARTITION_NAME IS NOT NULL;

-- ---------------------------------------------
-- COLUNA INFORMADA JÁ É CHAVE DE PARTICIONAMENTO?
-- ---------------------------------------------

set @v_column_count_partition_key_cmd = concat('select count(*) into @v_column_count_partition_key from INFORMATION_SCHEMA.PARTITIONS WHERE TABLE_SCHEMA = \'',v_database_name,'\' AND TABLE_NAME = \'',v_table_name,'\' AND PARTITION_EXPRESSION LIKE \'%',v_column_name,'%\' AND PARTITION_NAME IS NOT NULL;');

prepare stmt from @v_column_count_partition_key_cmd;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

set v_column_count_partition_key = @v_column_count_partition_key;

if v_partition_count = 0 then
select concat(now(),' [INFO]: A tabela ',v_table_name ,' não possui partições, serão processadas as partiçoes iniciais com a procudures create_partition() ') as menssage;
 call create_partition(v_database_name, v_table_name, v_column_name, v_partition_by,v_set_bin_log);
end if;

if v_partition_qtd > 0 then

case 
		when v_partition_by = 'day' then select concat(now(),' [INFO]: Adição de partição para novos dias à tabela ',v_table_name)as menssage; 
        
        select from_days(max(partition_description)) into v_data_passado
        from information_schema.partitions 
        where table_schema=v_database_name 
        and table_name=v_table_name;
        
        
        set v_total_dias= v_partition_qtd;
        set v_data_aux = v_data_passado;
        
        
        set @v_control_loop =1;
                    
		set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' add partition (' );
                  
		add_partition_loop: LOOP

select date_add(v_data_passado, interval @v_control_loop day) into v_data_aux;
			select v_data_aux into @partition_value;
			select concat('p',date_format(date_sub(v_data_aux, interval 1 day),"%Y%m%d")) into @partition_name;	
            set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (to_days(\'',@partition_value);

when v_partition_by = 'month' then 
			select concat(now(),' [INFO]: Serão adicionadas partições para ',v_partition_qtd,' novo(s) mês(es) à tabela ',v_table_name) as menssage; 
            
            
            
			-- -----------------------------------------------
            -- VALOR DA ULTIMA PARTIÇÃO CRIADA
            -- -----------------------------------------------
            
            select max(partition_description) into v_pmax 
            from information_schema.partitions 
            where table_schema= v_database_name
            and table_name=v_table_name;
            
            select str_to_date(concat(v_pmax,'01'),"%Y%m%d") into v_data_passado;
            set v_total_dias= v_partition_qtd;
			set v_data_aux = v_data_passado;
            
            set @v_create_partition_cmd = concat('alter table ',v_database_name,'.',v_table_name,' add partition (' );
            
            SET @v_control_loop = 1;
            
            add_partition_loop: LOOP         
            
			select concat('p',EXTRACT(YEAR_MONTH FROM v_data_aux)) into @partition_name;
			select date_add(v_data_passado, interval @v_control_loop month) into v_data_aux;
			select EXTRACT(YEAR_MONTH FROM v_data_aux) into @partition_value;
				
            set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);

select @v_data_passado+@v_control_loop into @v_data_aux;
			select @v_data_aux into @partition_value;
			select concat('p',(@v_data_aux-1)) into @partition_name;	
            set @v_create_partition_body = concat(' partition ',@partition_name, ' values less than (',@partition_value);

DELIMITER ;

Procedure drop_partition

Copy to Clipboard

DELIMITER $$
CREATE PROCEDURE `drop_partition`(db_name VARCHAR(100), tb_name VARCHAR(100), partition_by enum('day','month','year'), partition_qtd int, set_bin_log int)
proc_Exit:BEGIN

DECLARE v_database_name varchar(100);
DECLARE v_table_name varchar(100);
DECLARE v_partition_name varchar(100);
DECLARE v_partition_by enum('day','month','year');
DECLARE v_partition_qtd int;
DECLARE v_set_bin_log int;

DECLARE v_database_count int;
DECLARE v_table_count int;
DECLARE v_partition_count int;
DECLARE v_date_partition_today date;
DECLARE v_date_partition_today_count int;

select count(*) into v_table_count
   from information_schema.tables
where 
   table_schema = v_database_name
   and table_name = v_table_name;

-- ---------------------------------------------
-- TABELA POSSUI PARTIÇÕES ?
-- ---------------------------------------------

SELECT count(*) into v_partition_count
	FROM INFORMATION_SCHEMA.PARTITIONS 
WHERE TABLE_SCHEMA = v_database_name
AND TABLE_NAME = v_table_name 
AND PARTITION_NAME IS NOT NULL;

if v_partition_count = 0 then
select concat(now(),' [AVISO]: A tabela ',v_table_name ,' não possui partições!') as menssage;
leave  proc_Exit;
end if;

if v_partition_qtd > 0 then
	case 
		when v_partition_by = 'day' then select concat(now(),' [INFO]: Exclusão de partições acima de ',v_partition_qtd,' dia(s) da tabela ',v_table_name)as menssage; 
        
        
        SELECT date_add(from_days(partition_description), interval 1 day), count(partition_description) into v_date_partition_today, v_date_partition_today_count
			FROM INFORMATION_SCHEMA.PARTITIONS 
		WHERE TABLE_SCHEMA = v_database_name
		AND TABLE_NAME = v_table_name
		AND PARTITION_NAME IS NOT NULL
		AND partition_description = to_days(now());
        
        if v_date_partition_today_count = 1 then
        
			drop temporary table if exists temp_drop_partition;

create temporary table temp_drop_partition 
				select partition_name 
					FROM INFORMATION_SCHEMA.PARTITIONS 
				WHERE TABLE_SCHEMA = v_database_name
				AND TABLE_NAME = v_table_name
				AND PARTITION_NAME IS NOT NULL
				AND (datediff(v_date_partition_today,from_days(partition_description))) > v_partition_qtd;
		else 
			select concat(now(),' [ERRO]: Não foi encontrada a partição do dia atual')as menssage;
			leave  proc_Exit;
		end if;
        
        
	   select date_sub(v_date_partition_today, interval v_partition_qtd day) into @date_sub_qtd_drop;
       select count(*) into @count_partitions_drop from temp_drop_partition;
       
       if @count_partitions_drop > 0 then
       
       select concat(now(),' [INFO]: Serão excluidas todas as patições mais antigas que ',@date_sub_qtd_drop, ' total de ',@count_partitions_drop,' partições' )as menssage;
        
			select group_concat(partition_name separator ',') into @partition_name 
            from temp_drop_partition;
            
			set @drop_partition_cmd = concat('ALTER TABLE ',v_database_name,'.',v_table_name,' DROP PARTITION ',@partition_name,';');
            
            -- select @drop_partition_cmd;
            
            
            prepare stmt from @drop_partition_cmd;
			EXECUTE stmt;
			DEALLOCATE PREPARE stmt;
            
            else 
            
            select concat(now(),' [INFO]: Não existem partições mais antigas que ',v_partition_qtd, ' dia(s) para serem excluidas' )as menssage;
            
            end if;
            
            
            
		when v_partition_by = 'month' then select concat(now(),' [INFO]: Exclusão de partições acima de ',v_partition_qtd,' mês(es) da tabela ',v_table_name) as menssage; 
			
            select EXTRACT(YEAR_MONTH FROM date_add(now(), interval 1 month)) into @v_date_partition_today;
			select str_to_date(concat(@v_date_partition_today,'01'),"%Y%m%d") into @v_patition_atual_aux;
			select EXTRACT(YEAR_MONTH FROM date_sub(@v_patition_atual_aux, interval v_partition_qtd month))  into @v_partition_datesub;
            
            SELECT count(partition_description) into v_date_partition_today_count
			FROM INFORMATION_SCHEMA.PARTITIONS 
			WHERE TABLE_SCHEMA = v_database_name
			AND TABLE_NAME = v_table_name 
			AND PARTITION_NAME IS NOT NULL
			AND partition_description = @v_date_partition_today;
            
            if v_date_partition_today_count = 1 then
        
			drop temporary table if exists temp_drop_partition;
			create temporary table temp_drop_partition 
				select partition_name 
					FROM INFORMATION_SCHEMA.PARTITIONS 
				WHERE TABLE_SCHEMA = v_database_name
				AND TABLE_NAME = v_table_name
				AND PARTITION_NAME IS NOT NULL
				AND partition_description < @v_partition_datesub;
		else 
			select concat(now(),' [ERRO]: Não foi encontrada a partição do mês atual')as menssage;
			leave  proc_Exit;
		end if;
            
            
          select EXTRACT(YEAR_MONTH FROM date_sub(@v_patition_atual_aux, interval v_partition_qtd+1 month)) into @date_sub_drop;
		  select count(*) into @count_partitions_drop from temp_drop_partition;  
            
        
        if @count_partitions_drop > 0 then
       
       select concat(now(),' [INFO]: Serão excluidas todas as patições mais antigas que ',@date_sub_drop, ' total de ',@count_partitions_drop,' partições' )as menssage;
        
			select group_concat(partition_name separator ',') into @partition_name 
            from temp_drop_partition;
            
			set @drop_partition_cmd = concat('ALTER TABLE ',v_database_name,'.',v_table_name,' DROP PARTITION ',@partition_name,';');
            
            -- select @drop_partition_cmd;
            
            
            prepare stmt from @drop_partition_cmd;
			EXECUTE stmt;
			DEALLOCATE PREPARE stmt;
            
            else 
            
            select concat(now(),' [INFO]: Não existem partições mais antigas que ',v_partition_qtd, ' mês(s) para serem excluidas' )as menssage;
            
            end if;
            
            
        
		when v_partition_by = 'year' then select concat(now(),' [INFO]: Exclusão de partições acima de ',v_partition_qtd,' ano(s) da tabela ',v_table_name)as menssage; 
        
        
        select year(date_add(now(), interval 1 year)), date_add(now(), interval 1 year)  into @v_date_partition_today_aux, v_date_partition_today;
        
        select year(date_sub(v_date_partition_today, interval v_partition_qtd year)) into @v_partition_datesub;
        
        SELECT count(partition_description) into v_date_partition_today_count
			FROM INFORMATION_SCHEMA.PARTITIONS 
		WHERE TABLE_SCHEMA = v_database_name
		AND TABLE_NAME = v_table_name
		AND PARTITION_NAME IS NOT NULL
		AND partition_description = @v_date_partition_today_aux;
        
        if v_date_partition_today_count = 1 then
        
			drop temporary table if exists temp_drop_partition;

create temporary table temp_drop_partition 
				select partition_name 
					FROM INFORMATION_SCHEMA.PARTITIONS 
				WHERE TABLE_SCHEMA = v_database_name
				AND TABLE_NAME = v_table_name
				AND PARTITION_NAME IS NOT NULL
				AND partition_description < @v_partition_datesub;
		else 
			select concat(now(),' [ERRO]: Não foi encontrada a partição do ano atual')as menssage;
			leave  proc_Exit;
		end if;
        
        
	   
       select count(*) into @count_partitions_drop from temp_drop_partition;
       
       if @count_partitions_drop > 0 then
       
       select concat(now(),' [INFO]: Serão excluidas todas as patições mais antigas que ',@v_partition_datesub-1, ' total de ',@count_partitions_drop,' partições' )as menssage;
        
			select group_concat(partition_name separator ',') into @partition_name 
            from temp_drop_partition;
            
			set @drop_partition_cmd = concat('ALTER TABLE ',v_database_name,'.',v_table_name,' DROP PARTITION ',@partition_name,';');
            
            -- select @drop_partition_cmd;
            
            
            prepare stmt from @drop_partition_cmd;
			EXECUTE stmt;
			DEALLOCATE PREPARE stmt;
            
            else 
            
            select concat(now(),' [INFO]: Não existem partições mais antigas que ',v_partition_qtd, ' ano(s) para serem excluidas' )as menssage;
            
            end if;
	else begin end;
end case;
else
	select concat(now(),' [INFO]: O valor do paramêtro v_partition_qtd tem de ser maior que 0') as menssage;
end if;

drop temporary table if exists temp_drop_partition;
set sql_log_bin = 1;
END$$

DELIMITER ;

Particionamento de Tabelas no Mysql

Principais critérios de particionamento:

RANGE Partitioning

LIST Partitioning

HASH Partitioning

KEY Partitioning

Gerenciamento das partições HASH e KEY:

Procedures de particionamento

A estrutura de procedures está dividida da seguinte forma:

Criando novas partições com create_partition():

Adicionando novas partições com add_partition():

Excluindo partições com drop_partition():

Procedure create_partition

Procedure add_partition

Procedure drop_partition

Compartilhe

Sobre o Autor: Guilherme Prado