5 – 教你如何实现 JDBC 列权限控制
在上一篇文章(
4 – 教你如何实现 Hive 列权限控制
)中详细的讲述了,Spark SQL查询Hive是如何解析库表列信息的。今天笔者通过MLSQL源码演示,是如何使用Druid SQL Parser解析SQL中的表列信息。
Druid SQL Parser分三个模块:
- Parser:parser是将输入文本转换为ast(抽象语法树),parser有包括两个部分,Parser和Lexer,其中Lexer实现词法分析,Parser实现语法分析;
- AST:AST是Abstract Syntax Tree的缩写,也就是抽象语法树。AST是parser输出的结果;
- Visitor:Visitor是遍历AST的手段,是处理AST最方便的模式,Visitor是一个接口,有缺省什么都没做的实现VistorAdapter。
是不是跟第三篇文章(
3 – MLSQL DSL-你准备好搞自己的DSL了吗
)的Antlr很像啊?
下面以Mysql SQL为例演示MLSQL是如何进行JDBC表列解析的。 首先在Mysql的test库下建如下两张表:
create table c5_user( id varchar(255) , name varchar(255) ) create table c5_country( id varchar(255) , country varchar(255) )
然后构造测试SQL:
select name , country from c5_user a left join c5_country b on a.id = b.id
通过分析,在不确定c5_user和c5_country表结构的情况下,是无法分析出name是属于c5_user还是属于c5_country的。因此要准确判断列归属的表就需要表的schema信息。
下面讲一下MLSQL的处理逻辑:
- 解析传入SQL中的所有表(不支持跨库SQL查询,比如:Hive)
- 根据表,从Mysql库中查询表的schema信息
- 根据表的schema信息构造Druid中SchemaRepository的表信息
- 解析SQL为语法树,通过SchemaVisitor补充schema信息(SchemaRepository),遍历语法树解析出列信息
从MLSQL中提取出如下几个方法:
1. 从SQL中解析表,对源码有部分修改
//streaming.core.datasource.impl.MLSQLDirectJDBC def extractTablesFromSQL(sql: String, dbType: String = JdbcConstants.MYSQL) = { val stmt = SQLUtils.parseSingleStatement(sql, dbType) val visitor = SQLUtils.createSchemaStatVisitor(dbType) stmt.accept(visitor) visitor.getTables().asScala.map { f => val dbAndTable = f._1.getName if (dbAndTable.contains(".")) { val Array(db, table) = dbAndTable.split("\\.", 2) (db ,table) } else (dbAndTable, None) }.toList }
2. 查询表的元数据信息
//streaming.core.datasource.JDBCUtils def queryTableWithColumnsInDriver(options: Map[String, String] ,tableList: List[String]) = { val tableAndCols = mutable.HashMap.empty[String, mutable.HashMap[String ,String]] val driver = options("driver") val url = options("url") Class.forName(driver) val connection = java.sql.DriverManager.getConnection(url, options("user"), options("password")) try { val dbMetaData = connection.getMetaData() tableList.foreach(table => { val rs = dbMetaData.getColumns(null, null, table, "%") val value = tableAndCols.getOrElse(table, mutable.HashMap.empty[String ,String])
while(rs.next()){ value += (rs.getString("COLUMN_NAME") -> rs.getString("TYPE_NAME")) }
tableAndCols.update(table, value) rs.close() }) } finally { if (connection != null) connection.close() } tableAndCols }
3. 根据表的元数据信息,构造表的创建语句,用于构造SchemaRepository
//streaming.core.datasource.JDBCUtils def tableColumnsToCreateSql(tableClos: mutable.HashMap[String, mutable.HashMap[String, String]]) = { val createSqlList = mutable.ArrayBuffer.empty[String] tableClos.foreach(table => { var createSql = "create table " + table._1 + " (" + table._2.map(m => m._1 + " " + m._2) .mkString(",") + " )" createSqlList += createSql }) createSqlList.toList }
4. 解析SQL的表和列
//tech.mlsql.sql.MLSQLSQLParser def extractTableWithColumns(dbType :String ,sql :String ,createSchemaList :List[String]) = { val tableAndCols = mutable.HashMap.empty[String, mutable.HashSet[String]]
val repository = new SchemaRepository(dbType)
createSchemaList.foreach(repository.console(_))
val stmtList = SQLUtils.parseStatements(sql, dbType) val stmt = stmtList.get(0).asInstanceOf[SQLSelectStatement] repository.resolve(stmt)
val statVisitor = SQLUtils.createSchemaStatVisitor(dbType) stmt.accept(statVisitor)
val iter = statVisitor.getColumns().iterator()
while(iter.hasNext()){ val c = iter.next() if(c.isSelect()){ val value = tableAndCols.getOrElse(c.getTable, mutable.HashSet.empty[String]) value.add(c.getName) tableAndCols.update(c.getTable, value) } }
tableAndCols }
下面来看一下整个流程:
val sql = """ |select name ,country | from c5_user a | left join c5_country b | on a.id = b.id """.stripMargin
val params = Map("driver"->"com.mysql.jdbc.Driver" , "url"->"jdbc:mysql://127.0.0.1:3306/test" , "user"->"root" , "password"->"mlsql")
val rest = extractTablesFromSQL(sql)
val tableList = extractTablesFromSQL(sql).map(_._1)
val tableColsMap = queryTableWithColumnsInDriver(params, tableList)
val createSqlList = tableColumnsToCreateSql(tableColsMap) println("创建SQL:") createSqlList.foreach(println(_))
val tableAndCols = extractTableWithColumns(JdbcConstants.MYSQL, sql, createSqlList)
println("表列信息:") println(tableAndCols) -------------------------------------------------------------------------------------- 输出结果: 创建SQL: create table c5_country (country VARCHAR,id VARCHAR ) create table c5_user (name VARCHAR,id VARCHAR ) 表列信息: Map(c5_country -> Set(country), c5_user -> Set(name))
对于MLSQL JDBC的表列信息解析依赖于Druid,因此语法支持受限于Druid,目前Druid的支持如下:
数据库 | DML | DDL |
odps | 完全支持 | 完全支持 |
mysql | 完全支持 | 完全支持 |
postgresql | 完全支持 | 完全支持 |
oracle | 支持大部分 | 支持大部分 |
sql server | 支持常用的 | 支持常用的 |
db2 | 支持常用的 | 支持常用的 |
hive | 支持常用的 | 支持常用的 |
可以看出大部分常用的语法是满足的,在笔者的使用过程中,暂时未发现不满足的语法(除了hive),也可能是业务场景还不够复杂。但是想想,MLSQL只是读取JDBC数据,然后分析,也不需要太复杂的SQL,除非把MLSQL当数据查询服务来用,这样就会遇到很多复杂的SQL。对于MLSQL Hive JDBC库表列解析,可以把SQL当成Spark SQL,通过分析查询计划解析库表列,这样就能支持Hive的全部语法,而真正的执行是通过Hive JDBC(在没有用Ranger的情况下)。
喜欢就点击最上方的[ MLSQL之道 ]关注下吧,后面精彩不断!
源码地址:
https://github.com/latincross/mlsqlwechat(c5-druid)
更多介绍请访问:
http://docs.mlsql.tech/zh/