package org.springblade.sqlfly.opt;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.springblade.sqlfly.config.Column;
import org.springblade.sqlfly.config.Table;
import org.springblade.sqlfly.type.ClickhouseType;
import org.springblade.sqlfly.type.JavaType;

import cn.hutool.core.util.StrUtil;

/**
 * clickhouse数据库操作
 * 
 * @author William
 * @date 2022年10月28日 上午11:29:37
 */
public class FlyClickhouse extends IFly {

	protected FlyClickhouse() {
	}

	@Override
	public List<Table> getTableList(String schema) {
		List<Table> tables = new ArrayList<>();
		String sql = "select table  from system.parts where database = '" + db.getDbName() + "' group by table";
		List<String> retList = fly.getList(String.class, sql);
		retList.forEach(item -> tables.add(new Table(item)));
		return tables;
	}

	@Override
	public Table getTable(String schema, String tableName) {
		return new Table(tableName);
	}

	@Override
	public List<Column> getColumnList(String tableName) {
		List<Column> list = new ArrayList<>();
		String sql = "DESCRIBE " + db.getDbName() + "." + tableName;
		List<Map<String, Object>> retList = fly.getMapList(sql);
		if (retList != null) {
			JavaType javaType = new ClickhouseType();
			retList.forEach(map -> {
				Column dataColumn = new Column();
				dataColumn.setColumnName(map.get("name").toString());
				dataColumn.setColumnComment(map.get("comment") != null ? map.get("comment").toString() : "");
				String type = map.get("type").toString();

				if (type.contains("Nullable")) {
					dataColumn.setIsNullable("YES");
				} else {
					dataColumn.setIsNullable("NO");
				}
				type = type.replace("Nullable", "").replace("(", "").replace(")", "");
				dataColumn.setDataType(type);
				dataColumn.setJavaType(javaType.get(dataColumn.getDataType()));
				list.add(dataColumn);
			});
			findPrimaryKey(tableName, list);
		}
		return list;
	}

	private void findPrimaryKey(String tableName, List<Column> list) {
		String primaryColumn = "";
		StringBuffer sb = new StringBuffer();
		sb.append("SELECT * FROM system.columns ");
		sb.append("WHERE database = '" + db.getDbName() + "' ");
		sb.append("AND table = '" + tableName + "'");
		String sql = sb.toString();
		List<Map<String, Object>> retList = fly.getMapList(sql);
		if (retList == null || retList.isEmpty()) {
			return;
		}

		for (Map<String, Object> map : retList) {
			if (map.get("is_in_primary_key") != null && map.get("is_in_primary_key").toString().equalsIgnoreCase("1")) {
				primaryColumn = map.get("name").toString();
				break;
			}
		}
		if (StrUtil.isNotBlank(primaryColumn)) {
			// 有主键
			for (Column col : list) {
				if (col.getColumnName().equalsIgnoreCase(primaryColumn)) {
					col.setPrimaryKey(true);
					break;
				}
			}
		}
	}

}
