Use Hive Metastore API for listing partitions that allows range (and other) predicates on the partition keys

Parth-Brahmbhatt commented 5 years ago

See thread https://prestosql.slack.com/archives/CFLB9AMBN/p1554842107205400.

Currently in HivePartitionManager we use getPartitionNamesByParts API from metastore which only allows a single value per partition predicate. This means any non equality partition predicate can not be pushed down to metastore which results in 2 problems:

Even if the user specifies a predicate his query can fail with TOO_MANY_PARTITONS error
Undue load on metastore service for very large tables.

In the newer version of Hive new APIs are introduced to avoid this issue. We should look into moving to get_partitions_by_expr or get_partitions_by_filter for types that are supported by these APIs.

electrum commented 5 years ago

This is a good idea. Do you know what Hive version these were added in? Hopefully they’re old enough to be available everywhere, otherwise we’ll need some fallback code (catch remote “no such method” error and call the old one).

Parth-Brahmbhatt commented 5 years ago

If I just go by when the API was added, then that was 9 years ago.

luohao commented 5 years ago

We have explored this a bit, but I think we should probably avoid using get_partitions_by_expr(PartitionsByExprRequest req). It's very very bad API design. It includes a Kryo serialized object in the request:

struct PartitionsByExprRequest {
...
 3: required binary expr,
...
}

There are a few potential issues:

Kyro doesn't guarantee serialization format. From our experience, if the client(e.g, Presto) and server(e.g., HMS) use different versions of Kyro library, you may get deserialization error. Using Kyro to serialize objects is a bit weired to me.
Having a blob in the interface makes it hard to integrate with non-HMS systems. Presto only uses a few HMS APIs which makes it very easy to integrate with other schema registry. We have tried to implement a equivilant to get_partitions_by_expr in our own data discovery service but it turns out to be non-trivial.

We should probably try get_partitions_by_filter. It uses a filter string, which is more standard.

electrum commented 5 years ago

The problem with get_partitions_by_expr, get_partitions_by_filter, and get_part_specs_by_filter is that they return the full partition metadata objects, which can be huge and take too long to fetch.

With the existing get_partition_names_ps call, we see issues with timeouts and Thrift response size limits on tables with many partitions. And that's just fetching the names, which is a single string, not the huge metadata object.

Presto fetches partition names while planning, then fetches the full partition metadata iteratively during split generation as the query executes. The engine provides the connector with a Constraint containing a black box Predicate that is used to further filter the name list, which allows expressions or functions over partition columns which cannot be expressed via TupleDomain.

I now remember looking into this in the past and coming to the above conclusion that there was no better API that does what we need (list partition names with a filter). However, they recently added get_partition_values in HIVE-17466 which seems to do exactly what we need. The PartitionValuesRequest struct has a filter field.

We should check if this API is available in CDH 5 and HDP 2.

Parth-Brahmbhatt commented 5 years ago

CDH 5 seems to be on Hive 1.1.0 + patches and the Jira you have linked does not seem to be in the list of patches

HDP-2.6.5 which is the most latest release of HDP-2 also only ships HIVE-2.1.0

No matter when we decide to move to a newer API I believe we will have to make it config enabled to support organization that are running on older version of hive servers so I do not see a reason not to implement this right now and make it config enabled.

electrum commented 5 years ago

I agree, there's no reason not to implement it now. We can do it without config if we make the fallback transparent. Have a boolean flag defaulting to true to indicate it is supported, then set it to false if we get a "no such method" error from the remote Thrift server. This means the first request after a server restart will take slightly longer, but this shouldn't cause any problems.

luohao commented 5 years ago

Agreed with @electrum.

As long as we have a fallback mechanism it should not break the compatibility. Hive adopts a similar strategy where a MetaException from HMS will make it falls back to the implementation of client-side filtering.

    430         try {
    431           hasUnknownPartitions = Hive.get().getPartitionsByExpr(
    432               tab, compactExpr, conf, partitions);
    433         } catch (IMetaStoreClient.IncompatibleMetastoreException ime) {
    434           // TODO: backward compat for Hive <= 0.12. Can be removed later.
    435           LOG.warn("Metastore doesn't support getPartitionsByExpr", ime);
    436           doEvalClientSide = true;
    437         } finally {
    438           perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.PARTITION_RETRIEVING);
    439         }

findepi commented 5 years ago

get_partition_values was added in Hive 2.4, but this version is not downloadable (https://archive.apache.org/dist/hive/) and, AFAIK, we're not ready to require Hive 3.1 just yet. I created enhancement proposal to backport get_partition_values to Hive 2.3 branch (https://issues.apache.org/jira/browse/HIVE-21859).

findepi commented 5 years ago

get_partition_values has been merged to Hive 2.3 branch (https://issues.apache.org/jira/browse/HIVE-21859).