{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/14690","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/14690/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/14690/comments","events_url":"https://api.github.com/repos/apache/spark/issues/14690/events","html_url":"https://github.com/apache/spark/pull/14690","id":171724517,"number":14690,"title":"[SPARK-16980][SQL] Load only catalog table partition metadata required to answer a query","user":{"login":"mallman","id":833693,"avatar_url":"https://avatars.githubusercontent.com/u/833693?v=3","gravatar_id":"","url":"https://api.github.com/users/mallman","html_url":"https://github.com/mallman","followers_url":"https://api.github.com/users/mallman/followers","following_url":"https://api.github.com/users/mallman/following{/other_user}","gists_url":"https://api.github.com/users/mallman/gists{/gist_id}","starred_url":"https://api.github.com/users/mallman/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mallman/subscriptions","organizations_url":"https://api.github.com/users/mallman/orgs","repos_url":"https://api.github.com/users/mallman/repos","events_url":"https://api.github.com/users/mallman/events{/privacy}","received_events_url":"https://api.github.com/users/mallman/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2016-08-17T18:13:53Z","updated_at":"2016-08-18T18:09:35Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/14690","html_url":"https://github.com/apache/spark/pull/14690","diff_url":"https://github.com/apache/spark/pull/14690.diff","patch_url":"https://github.com/apache/spark/pull/14690.patch"},"body":"(This PR addresses https://issues.apache.org/jira/browse/SPARK-16980.)\r\n\r\n(N.B. I'm submitting this PR as an enhanced version of an internal POC I wrote. I'm looking for preliminary feedback on what I have so far and to discuss some design and implementation issues. This PR is not currently a candidate for merging into master.)\r\n\r\n(N.B. This PR is known to fail several unit tests related to Hive/Parquet conversion. Obviously, these failures will be addressed before this PR is submitted for merging into master.)\r\n\r\n## What changes were proposed in this pull request?\r\n\r\nIn a new Spark application, when a partitioned Hive table is converted to use Spark's `HadoopFsRelation` in `HiveMetastoreCatalog`, metadata for every partition of that table are retrieved from the metastore and loaded into driver memory. In addition, every partition's metadata files are read from the filesystem to perform schema inference.\r\n\r\nIf a user queries such a table with predicates which prune that table's partitions, we would like to be able to answer that query without consulting partition metadata which are not involved in the query. When querying a table with a large number of partitions for some data from a small number of partitions (maybe even a single partition), the current conversion strategy is highly inefficient. I suspect this scenario is not uncommon in the wild.\r\n\r\nIn addition to being inefficient in running time, the current strategy is inefficient in its use of driver memory. When the sum of the number of partitions of all tables loaded in a driver reaches a certain level (somewhere in the tens of thousands), their cached data exhaust all driver heap memory in the default configuration. I suspect this scenario is less common (in that not too many deployments work with tables with tens of thousands of partitions), however this does illustrate how large the memory footprint of this metadata can be. With tables with hundreds or thousands of partitions, I would expect the `HiveMetastoreCatalog` table cache to represent a significant portion of the driver's heap space.\r\n\r\nThis PR proposes an alternative approach. Basically, it makes three changes:\r\n\r\n1. It adds a new method, `listPartitionsByFilter` to the Catalyst `ExternalCatalog` trait which returns the partition metadata for a given sequence of partition pruning predicates.\r\n1. It refactors the `FileCatalog` type hierarchy to include a new `TableFileCatalog` to efficiently return files only for partitions matching a sequence of partition pruning predicates.\r\n1. It removes partition loading and caching from `HiveMetastoreCatalog`.\r\n\r\nThe net effect is that when a query over a partitioned Hive table is planned, the analyzer retrieves the table metadata from `HiveMetastoreCatalog`. As part of this operation, the `HiveMetastoreCatalog` builds a `HadoopFsRelation` with a `TableFileCatalog`. It does not load any partition metadata or scan any files. The physical planner identifies the data files the query needs by asking the relation's `TableFileCatalog` for the files matching any predicate pruning predicates. The `TableFileCatalog` in turn calls the `listPartitionsByFilter` method on its external catalog. This queries the Hive metastore, passing along those filters.\r\n\r\n## Open Issues\r\n\r\n1. This PR omits partition metadata caching. I'm not sure if this is even needed if we're only loading partition metadata for a given query. However, it may not be that tricky to implement this effectively.\r\n1. This PR removes and omits partitioned Hive table schema reconciliation. As a result, it fails to find Parquet schema columns with upper case letters because of the Hive metastore's case-insensitivity. I think this is the most significant hurdle for this PR. It just occurred to me that we might be able to do just-in-time schema reconciliation using the partitions that are used in a query. I haven't tried this, but I would attempt this by adding a method to `HadoopFsRelation` or `BasicFileCatalog` which returns a SQL schema for a given sequence of partition pruning predicates (or partitions). I'll give this a try and report back. Another idea would be to use the current strategy of merging schema from all table files unless the user sets a boolean SQL configuration parameter like `spark.sql.assumeLowerCaseColumnNames`. If the user's tables have only lower-case column names, then it's safe to use this PR's optimizations. I don't think this is an entirely unrealistic scenario as we have enforced all lower-case column names from the beginning because of case-sensitivity issues. Maybe we're not the only ones?\r\n1. This PR omits an implementation of `listPartitionsByFilter` for the `InMemoryCatalog`.\r\n1. Should we use `TableFileCatalog` for non-partitioned tables, too (to unify the partitioned and non-partitioned table access code paths)? This is probably something to explore in a future PR.\r\n1. This PR breaks parquet log output redirection. I can work around this by running `Class.forName(\"org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat$\")` first thing in a Spark shell session, however we need to figure out how to fix this properly.\r\n1. This PR includes no new unit tests. We should probably have some. :)\r\n\r\n## How was this patch tested?\r\n\r\nThe current Spark unit tests were run, and some ad-hoc tests were performed to validate that only the necessary partition metadata is loaded.\r\n\r\nN.B. Several tests in the hive subproject's `ParquetMetastoreSuite` fail. All of these tests fail because of the use of upper case characters in the test parquet table column names."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/240807594","html_url":"https://github.com/apache/spark/pull/14690#issuecomment-240807594","issue_url":"https://api.github.com/repos/apache/spark/issues/14690","id":240807594,"user":{"login":"mallman","id":833693,"avatar_url":"https://avatars.githubusercontent.com/u/833693?v=3","gravatar_id":"","url":"https://api.github.com/users/mallman","html_url":"https://github.com/mallman","followers_url":"https://api.github.com/users/mallman/followers","following_url":"https://api.github.com/users/mallman/following{/other_user}","gists_url":"https://api.github.com/users/mallman/gists{/gist_id}","starred_url":"https://api.github.com/users/mallman/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mallman/subscriptions","organizations_url":"https://api.github.com/users/mallman/orgs","repos_url":"https://api.github.com/users/mallman/repos","events_url":"https://api.github.com/users/mallman/events{/privacy}","received_events_url":"https://api.github.com/users/mallman/received_events","type":"User","site_admin":false},"created_at":"2016-08-18T18:09:35Z","updated_at":"2016-08-18T18:09:35Z","body":"I've been thinking some more about the metastore/file schema reconciliation. As I mentioned in the PR description, this patch omits this reconciliation. This causes failures when the parquet files have columns with upper case characters, among other cases.\r\n\r\nI mentioned trying to add a new schema method to the relation or its file catalog which would read the file schema for only a subset of the partitions. However, I quickly determined that would require a fundamental change in the way `BaseRelation` instances are handled throughout the codebase and decided that was a nonstarter.\r\n\r\nI have another idea I'd like your thoughts on. Rather than defer partition pruning to physical planning, what if we did that in the optimization phase and performed the hive `MetastoreRelation` conversion in an optimization rule? This rule would build a `HadoopFsRelation` based on the partition pruning predicates. We could add it to a batch after all of the pruning predicates have been pushed down and set it to run once. In that kind of scenario, I believe we could leave the `HadoopFsRelation` and `FileCatalog` types as they are in the current codebase. We'd just be building a `FileCatalog` restricted to the partitions we identify in the optimization phase. Physical planning would proceed as usual.\r\n\r\nIn this approach, we should be able to do \"just in time\" schema reconciliation/merging based only on the selected partitions. *And* we could address an issue I've long been frustrated by—the fact that a `BaseRelation`'s size is defined irregardless of the partitions we're using. By creating a `HadoopFsRelation` restricted to the partitions we care about, we can compute a precise size for the relation tailored to our query. This would greatly help the automatic broadcast join conversion heuristic.\r\n\r\nThoughts? I may start fiddling around with this when I have time today. It's possible I'm missing a crucial detail that makes this approach impractical."}}
{"actor":{"display_login":"mallman"}}