Date type 파티션 컬럼에 대한 필터링이 있을 때 quote 를 추가해주는 converter util 함수 및 클래스 추가

[kimtkyeom]

DISCLAIMER

현재 draft pr로 작성한 코드는 동작 테스트는 됐으나 코드 구조나 심미성 측면에서는 부족하여 다른 분들이 확인하여 개선하거나 해야 할 것 같습니다.

DESCRIPTION

문제의 원인

https://www.notion.so/devsisters/ck_metadata_prod-resource_configs-thriftserver-5352af6f19914e95966d90926f0f109c
spark 3.1.2 버전을 spark thrift server에 사용하게 되면서, date type에 대한 partition filter가 pushdown 됨.
이 때 pushdown 되는 형태는 date_col=2021-05-05 같이 quote가 없는 형태여서 glue api가 지원하지 않는다. https://issues.apache.org/jira/browse/SPARK-33477
이러한 패턴의 변화를 aws glue catalog library는 지원하지 않고 있음 Issue

문제 해결의 실마리

다만, 이를 보다보니 hive 에서는 어떻게 partition pushdown을 사용하는지 문득 궁금해져서 코드를 탐색함
spark 내부의 hive client가 hive metastore를 항해 똑같은 partition filter expression을 전달하는데 metastore process 안에서 direct sql 이라는 형태로 partition pruning을 수행함.
https://github.com/apache/hive/blob/cb213d88304034393d68cc31a95be24f5aac62b6/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java#L1229

https://github.com/apache/hive/blob/cb213d88304034393d68cc31a95be24f5aac62b6/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java#L5084

https://github.com/apache/hive/blob/cb213d88304034393d68cc31a95be24f5aac62b6/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java#L586 (이 때 local derby 기준으로 default RawStore는 ObjectStore)

https://github.com/apache/hive/blob/cb213d88304034393d68cc31a95be24f5aac62b6/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java#L2904

https://github.com/apache/hive/blob/cb213d88304034393d68cc31a95be24f5aac62b6/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java#L3227

https://github.com/apache/hive/blob/cb213d88304034393d68cc31a95be24f5aac62b6/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java#L3230

실제 directSql 이 동작하는 패턴을 보면, 내부에서 filter expression을 다시 파싱하여 expression tree를 구성하고, 해당 expression tree를 순회하면서 실제 pruning 할 파티션 목록을 추출하기 위한 SQL text를 구성한다. metastore package에 대한 log level을 debug로 설정하면 아래와 같은 메시지를 확인할 수 있음.

scala> spark.sql("SELECT * FROM temp where b = '2021-05-05'").show()
21/09/15 14:25:07 DEBUG MetaStoreDirectSql: getDatabase: directsql returning db default locn[file:/Users/kimtkyeom/Dev/spark_distributions/spark-3.1.2-devsisters.2/spark-warehouse] desc [Default Hive database] owner [public] ownertype [ROLE]
21/09/15 14:25:07 DEBUG MetaStoreDirectSql: getDatabase: directsql returning db default locn[file:/Users/kimtkyeom/Dev/spark_distributions/sp
ark-3.1.2-devsisters.2/spark-warehouse] desc [Default Hive database] owner [public] ownertype [ROLE]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 1.036926ms + 0.016338ms, the query is [select "PARTITIONS"."PART_ID" from "PARTITIONS"  inner join "TBLS" on "PARTITIONS"."TBL_ID" = "TBLS"."TBL_ID"     and "TBLS"."TBL_NAME" = ?   inner join "DBS" on "TBLS"."DB_ID" = "DBS"."DB_ID"      and "DBS"."NAME" = ? inner join "PARTITION_KEY_VALS" "FILTER0" on "FILTER0"."PART_ID" = "PARTITIONS"."PART_ID" and "FILTER0"."INTEGER_IDX" = 0 where (((case when "FILTER0"."PART_KEY_VAL" <> ? and "TBLS"."TBL_NAME" = ? and "DBS"."NAME" = ? and "FILTER0"."PART_ID"
= "PARTITIONS"."PART_ID" and "FILTER0"."INTEGER_IDX" = 0 then cast("FILTER0"."PART_KEY_VAL" as date) else null end) = ?))]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 1.35914ms + 0.151028ms, the query is [select "PARTITIONS"."PART_ID", "SDS"."SD_ID", "SDS"."CD_ID", "SERDES"."SERDE_ID", "PARTITIONS"."CREATE_TIME", "PARTITIONS"."LAST_ACCESS_TIME", "SDS"."INPUT_FORMAT", "SDS"."IS_COMPRESSED", "SDS"."IS_STOREDASSUBDIRECTORIES", "SDS"."LOCATION", "SDS"."NUM_BUCKETS", "SDS"."OUTPUT_FORMAT", "SERDES"."NAME", "SERDES"."SLIB" from "PARTITIONS"  left outer join "SDS" on "PARTITIONS"."SD_ID" = "SDS"."SD_ID"   left outer join "SERDES" on "SDS"."SERDE_ID" = "SERDES"."SERDE_ID" where "PART_ID" in (1) order by "PART_NAME" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.401488ms + 0.205927ms, the query is [select "PART_ID", "PARAM_KEY", "PARAM_VALUE" from "PARTITION_PARAMS" where "PART_ID" in (1) and "PARAM_KEY" is not null order by "PART_ID" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.451847ms + 0.119447ms, the query is [select "PART_ID", "PART_KEY_VAL" from "PARTITION_KEY_VALS" where "PART_ID" in (1) and "INTEGER_IDX" >= 0 order by "PART_ID" asc, "INTEGER_IDX" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.318219ms + 0.050671ms, the query is [select "SD_ID", "PARAM_KEY", "PARAM_VALUE" from "SD_PARAMS" where "SD_ID" in (7) and "PARAM_KEY" is not null order by "SD_ID" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.316556ms + 0.043167ms, the query is [select "SD_ID", "COLUMN_NAME", "SORT_COLS"."ORDER" from "SORT_COLS" where "SD_ID" in (7) and "INTEGER_IDX" >= 0 order by "SD_ID" asc, "INTEGER_IDX" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.340839ms + 0.04868ms, the query is [select "SD_ID", "BUCKET_COL_NAME" from "BUCKETING_COLS" where "SD_ID" in (7) and "INTEGER_IDX" >= 0 order by "SD_ID" asc, "INTEGER_IDX" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.35307ms + 0.042584ms, the query is [select "SD_ID", "SKEWED_COL_NAME" from "SKEWED_COL_NAMES" where "SD_ID" in (7) and "INTEGER_IDX" >= 0 order by "SD_ID" asc, "INTEGER_IDX" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 1.227652ms + 0.109451ms, the query is [select "CD_ID", "COMMENT", "COLUMN_NAME", "TYPE_NAME" from "COLUMNS_V2" where "CD_ID" in (6) and "INTEGER_IDX" >= 0 order by "CD_ID" asc, "INTEGER_IDX" asc]
21/09/15 14:25:52 DEBUG MetaStoreDirectSql: Direct SQL query in 0.434672ms + 0.14006ms, the query is [select "SERDE_ID", "PARAM_KEY", "PARAM_VALUE" from "SERDE_PARAMS" where "SERDE_ID" in (7) and "PARAM_KEY" is not null order by "SERDE_ID" asc]

위 까지의 내용에 착안하여 expression tree를 직접 순회하면서, date type일 경우 quote를 단 string으로 바꿔주면 되지 않을까? 하는 생각으로 구현을 시작함

구현 세부 사항

기본 JDO metastore 와 glue의 차이점은, glue의 경우 파티션을 pruning할 때 심플한 expression string을 request parameter로 보낸다는 점이다. (JDO metastore의 경우 직접 DB에 저장된 metastore table들에 대한 쿼리를 수행)
따라서 위에 언급했던 구현들 보다는 더욱 간단하게 구현을 진행하였음.

expression tree를 파싱하여 tree를 순회하면서 아래와 같이 filter expression을 단순히 convert 하는 로직만을 구성함

• value type 이 integeral value인 경우 => quote 없이 expression 생성 (eg. a=1)
• value type이 string 일 경우 => quote 추가 (eg. a='1')
• value type이 date 일 경우 => date type 값을 string으로 format 후 quote 삽입 (eg.a='2021-05-05')
  위와 같은 로직을 통해 원래 발생했던 문제 (date type에 대한 expression을 glue가 받아들이지 못하는 이슈) 를 수정할 수 있음
  이를 일종의 Hive aspect => glue catalog aspect의 convert로 보고 해당 클래스 및 메서드를 ConverterUtils 에 추가함
  이후 glue metastore 에서 실제 glue api를 호출하기 직전 convert method를 호출하여 expression을 변환, 만약 변환 과정 중 모종의 이슈가 있는 경우 (파싱에 실패하거나, pushdown할 수 없는 value를 받은 경우) null을 리턴하여 에러는 없으나 모든 파티션을 list 해서 client-side에서 필터링 하도록 수정