Implement Parallel Partition Pruning for Glue Hive Metastore

[anoopj]

This change parallelizes the partition fetch for the Glue metastore by
splitting the partitions into non-overlapping segments[2]. This can speed
up query planning by upto an order of magnitude.

[1] https://docs.aws.amazon.com/glue/latest/webapi/API_Segment.html

[findepi]

nit: per code style, if there are too many params to put on one line, we put every on a separate line:

public GlueHiveMetastore(
    HdfsEnvironment hdfsEnvironment, 
    GlueHiveMetastoreConfig glueConfig,
    @ForGlueHiveMetastore Executor executor)

[anoopj]

Sure.

[electrum]

This still needs updating

[findepi]

requireNonNull(executor, "executor is null");

[anoopj]

Sure.

[findepi]

we prefer immutable collections, so: toImmutableList()

however, in this case you don't need the list at all:

.forEach(segment -> completionService.submit(() -> ....);

[anoopj]

OK

[findepi]

we don't use abbreviations like "num"

maybe getPartitionThreads ?

[findepi]

• no need for "Glue" here
• add "total" to emphasize this is total cap, not per invocation:

maxGetPartitionTotalThreads ?

Also, what is the rationale for 50 as the default?
What's the default request limit for Glue GetPartition call?
What's the typical duration of a call?

Also, should this be off by default?

[anoopj]

Will change variable names. Typically the call is expected to take hundreds of millis. 5 concurrent segments is conservative enough to be one by default.

[findepi]

This needs to be renamed appropriately, see comment at field name.

[findepi]

When hiveConfig.getMaxGlueGetPartitionThreads() == 1 we could return directExecutor() here.

[anoopj]

Sure

[findepi]

Sort partitions before returning. We want planning to be deterministic (as much as possible).

[findepi]

When sorting, please do this outside of try block, so that try-catch encompasses as little as possible.

[anoopj]

OK. Will do the sorting based on partition values.

[findepi]

Adding this method may change number of test methods effectively run in TestHiveGlueMetastore. Please extract to a separate commit.

[electrum]

This change will cause all of the tests that depend on existing tables to fail (they are currently skipped by the method in the super class). What is the reason for this change?

[anoopj]

I found that the getPartitions tests are getting skipped without this change. What is the best way to test them?

[electrum]

It's complicated. Travis invokes presto-hive-hadoop2/bin/run_hive_tests.sh which

• starts up HDFS and a Hive metastore in Docker
• runs the presto-hive/src/test/sql/create-test.sql Hive script to create test tables
• runs the AbstractTestHive tests against this environment

The purpose is to create tables in various ways using Hive, then make sure that Presto can read them. That's where the "existing tables" part comes from -- they already exist when the tests run.

It's fine to leave this out, since there are plenty of existing tests that exercise the metastore partition calls. I verified this by running TestHiveInMemoryMetastore with logging on the various getPartition* metastore calls.

[findepi]

@anoopj technically, this looks decent.
Instead of one bigger call, we're now making 5 smaller calls in parallel and the PR introduces the necessary plumbing for this.
However, I don't understand yet what problem it is solving.

Are we improving Glue throughput this way?
Why isn't Glue service doing its work in parallel on its own, if this improves its performance?
Is it safe to enable this by default? (eg API rate limits, see https://groups.google.com/d/msg/presto-users/pWFEADyLNUc/pQ70_eDKAAAJ)

[anoopj]

@findepi

@anoopj technically, this looks decent.
Instead of one bigger call, we're now making 5 smaller calls in parallel and the PR introduces the necessary plumbing for this.
However, I don't understand yet what problem it is solving.
Are we improving Glue throughput this way?

This change can improve query planning time on heavily partitioned tables because we can do the scan of the partitions in parallel. For large tables with millions of partitions, a query can be stuck in the planning phase for dozens of minutes. My tests show that the query planning time can be improved by up to an order of magnitude.

Why isn't Glue service doing its work in parallel on its own, if this improves its performance?

Please note that Glue GetPartitions is a paginated API and it returns only a set of partitions. So even if the Glue service did parallel reads across segments, it is not likely to help clients because they would be making the same number of calls anyway to the service.

Is it safe to enable this by default? (eg API rate limits, see https://groups.google.com/d/msg/presto-users/pWFEADyLNUc/pQ70_eDKAAAJ)

I think 5 is a conservative number and should be a safe default. For what it's worth, this is the default followed by Spark and Hive on EMR. (search for segments on the doc page)

We can add some documentation in Presto about the new setting and advise Presto users to either adjust this setting if they run into throttling or contact AWS to raise the throttling limits.

[findepi]

Please note that Glue GetPartitions is a paginated API and it returns only a set of partitions. So even if the Glue service did parallel reads across segments, it is not likely to help clients because they would be making the same number of calls anyway to the service.

good point.

[findepi]

some more minor comments

@anoopj let me know when you AC.

[findepi]

nit

Suggested change

	public @interface ForGlueHiveMetastore
	public @interface ForGlueHiveMetastore {}

[anoopj]

OK

[findepi]

When sorting, please do this outside of try block, so that try-catch encompasses as little as possible.

[findepi]

it's hierarchical: next page (token) within a segment, so move withSegment before withNextToken

[anoopj]

OK

[findepi]

Unrelated, so best to separate commit or drop the change.

[anoopj]

@findepi Updated the PR incorporating the review comments.

[findepi]

I have some doubts about intuitiveness of the configuration.

• for hiveConfig.getMaxGlueGetPartitionThreads() == N, we have at most N threads calling Glue, regardless of number of queries being planned. This is both: parallelization and form of throttling (in case of multiple queries)
• for hiveConfig.getMaxGlueGetPartitionThreads() == 1 it could be expected that we have (at most) 1 thread calling Glue.
  Instead, we have as many threads as the number of queries being planned, because no thread pool is being used.

This may be confusing to administrators. Also, we don't have option to configure "1 thread throttle". At least in theory, this might be an issue in an organization with multiple clusters, hitting Glue API limits.

I think we could use 0 as a special value (no thread pool). And any value >0 to configure the pool.
Any better options?

@electrum what's your thinking?

[anoopj]

Currently when hiveConfig.getMaxGlueGetPartitionThreads() == 1, we are using a direct executor. Why don't we just use a threadpool of size 1 so that the behavior is consistent?

[electrum]

Let's rename the config to GlueMaxParallelPartitionThreads and keep the special behavior for 1 being a direct executor. I notice we bypass the executor for totalSegments == 1, so it's already not a global limit. It's probably confusing no matter what we do, so we can just document the behavior.

[electrum]

A few minor comments. Overall code looks good.

[electrum]

This names seems wrong, since it's not for caching. Could just be createExecutor since the scope is for the Glue metastore module.

[anoopj]

Sure thing.

[electrum]

This still needs updating

[anoopj]

Sorry for missing that one.

[electrum]

I think this would be easier to read as a traditional for loop

[anoopj]

Sure.

[electrum]

We shouldn't need to convert to string here, as List<String> is naturally comparable with the same semantics

[findepi]

Also this should probably be partitions.sort() rather than Collections.sort(partitions)

[anoopj]

Will change to partitions.sort(). Not using the string would require me to write a custom comparator that compares List since List is not Comparable. Something like:

partitions.sort((p1, p2) -> {
        	List<String> values1 = p1.getValues();
        	List<String> values2 = p2.getValues();
        	if (values1.size() != values2.size()) {
        		return values1.size() - values2.size();
                 }
        	for (int i = 0; i < values1.size(); i++) {
                     int c = values1.get(i).compareTo(values2.get(i));
                     if (c != 0) {
                         return c;
                      }
		}
                 return 0;
});

Does that sound reasonable?

[findepi]

partitions.sort(com.google.common.collect.Ordering.natural().lexicographical())

(there should be a way to do this without guava... but i didn't find it.)

[electrum]

This catch block is not needed -- any exception thrown by the test method will fail the test.

[anoopj]

Sure. Had to add throws to the parent class method too.

[electrum]

Yep, throws Exception is very common for test methods (and only for test methods)

[electrum]

This change will cause all of the tests that depend on existing tables to fail (they are currently skipped by the method in the super class). What is the reason for this change?

[findepi]

use ordinary if

if (hiveConfig.getTotalGetPartitionThreads() == 1) {
    return directExecutor();
}

[anoopj]

OK

[findepi]

@electrum how to say that 1 is a special value here?

[electrum]

The difference is subtle and complicated to explain. I think the description here is fine, though we could document it in the main Hive documentation.

[findepi]

Also this should probably be partitions.sort() rather than Collections.sort(partitions)

[findepi]

i don't think "total" should be here.

@electrum, hive.metastore.glue.partitions-segments ?

[anoopj]

Please let me know if this needs a change.

[electrum]

Agreed, let's change this as @findepi suggested. The "total segments" makes sense in terms of the parameter in the API call (since it is saying X of Y), but "partition segments" seems better for a configuration name.

[anoopj]

@electrum @findepi Responded to your feedback and have some questions. Please take a look.

[findepi]

@anoopj a new thought.
I guess that the most common case are queries that hit only 1-2 partitions.
In that case, you will still issue 5 requests even though 1 would be enough.
Maybe we should consider a mixed strategy: request first page of partitions.
If there is a next page, go parallel (possibly discarding the first page).

[anoopj]

@findepi My recommendation is to keep it simple and use parallel calls because a typical query usually spans way more than 5 partitions. Also, the default throttling of Glue is way higher and should allow several concurrent queries (and can be raised by contacting AWS).

[electrum]

I agree with keeping the code simple, but I also agree with @findepi that we could be a bit smarter about parallel reads. One idea is to have a minPartitionsPerSegment, which defaults to a smallish number like 5. Then we could do

total = min(((partitionCount - 1) / minPartitionsPerSegment) + 1, totalSegments);

This would limit parallelism for small numbers of partitions while adding minimal complexity.

[sopel39]

One idea is to have a minPartitionsPerSegment, which defaults to a smallish number like 5

Maybe minPartitionsPerSegment should be equal to max page size (when paginating) of Glue request.

[anoopj]

@electrum @findepi @sopel39

The challenge here is that we don't know what the number of matching partitions a-priori. This could vary based on the number of partitions for the table and how selective the filters were. Please let me know if I misunderstood.

[anoopj]

@electrum @findepi

Ping

[findepi]

Sorry, i missed last update.

a typical query usually spans way more than 5 partitions

If this is an assumption, then we have divergent assumptions here.
If this is a fact, can you please elaborate?

The challenge here is that we don't know what the number of matching partitions a-priori.

I am aware. This is why i proposed to go parallel only after first call.

Maybe we try to impl that and see how much we sacrifice on code simplicity?
e.g. a commit on top of the existing PR would help

[anoopj]

@findepi @electrum @sopel39

I'm more worried about making an unnecessary API call to switch to parallel partition pruning. In my experience using partitioned tables, the tables were typically partitioned using a set of keys and also a time series field and many queries were hitting several partitions with some queries spending more time in query planning than execution and hence this PR.

If there are concerns, we could make this serial by default.

[electrum]

If no one has strong objections, I think this is good to merge with the current behavior. It's a win in the case of many partitions, and for the few partitions case, it likely doesn't add enough overhead to matter. Otherwise, I think we'd need to collect real stats to compare the different cases in various scenarios.

@anoopj can you rebase and address all of the existing comments so that we can get this merged?

[electrum]

Agreed, let's change this as @findepi suggested. The "total segments" makes sense in terms of the parameter in the API call (since it is saying X of Y), but "partition segments" seems better for a configuration name.

[electrum]

Yep, throws Exception is very common for test methods (and only for test methods)

[electrum]

@dain and I were looking at this and noticed a big inefficiency in the interaction between Presto and the Glue API design. Presto accesses partition information in two phases:

1. All matching partition names are fetched during planning. This needs to be fast, since it happens before the execution starts. The TupleDomain is translated in into a metastore filter expression, then the list of partitions names is filtered on the Presto side using a black box Predicate (not limited to what TupleDomain can represent).

2. During execution, as needed, the full partition metadata is fetched for specific partition names. This happens at runtime because the full metadata can be large and fetching it can be slow.

The problem is that the Glue API only allows fetching full partition metadata -- there is no efficient way to fetch just the partition names. So for the Glue metastore implementation, we fetch the full partition metadata during planning, throw away everything but the name, then fetch the metadata again during execution.

If the Glue API had a way to fetch just partition names, we might not need this segmented fetching, since listing names would (hopefully) be significantly faster.

[anoopj]

@david @dain @findepi

That is a good observation and we are aware that this is suboptimal. We currently are working on adding a flag to the Glue GetPartitions API such that it returns only the partition names and the expensive parts like the schema won't have to be read. This would reduce the latency and also allow more partitions to be returned. When this is available in the public AWS APIs and SDK, we can send another PR to make use of that.

I don't think it would obviate the parallel/segmented calls though, since there could be heavily partitioned tables and queries that need to read a lot of partitions. Maybe we could adjust the defaults to be even lower than 5 based on some tests.

I'm a bit overbooked this week and will update the PR with feedback next week hopefully.

[anoopj]

@electrum @findepi

Updated the PR incorporating feedback. I've also tested this on a Presto cluster. On a table with about 2000 partitions, getPartitions used to take 2766 ms. With this change, it drops to 650 ms (with hive.metastore.glue.partitions-segments set to 5, which is the default).

On heavily partitioned tables, this can result in an order of magnitude improvement.

[anoopj]

@electrum @findepi

Ping.

[electrum]

A few minor comments, otherwise looks good. There is one minor fix needed for the Travis build.

[electrum]

Nit: use a method reference and static import

private static final Comparator<Partition> PARTITION_COMPARATOR =
        comparing(Partition::getValues, lexicographical(CASE_INSENSITIVE_ORDER));

[anoopj]

Will change.

[electrum]

This still needs updating

[electrum]

The difference is subtle and complicated to explain. I think the description here is fine, though we could document it in the main Hive documentation.

[electrum]

This still needs updating

[electrum]

Nit: put throws clause on next line

[anoopj]

OK

[electrum]

Nit: use ImmutableList.of()

[anoopj]

Will do.

[electrum]

Create a shared cached thread pool for the test class using @BeforeClass / @AfterClass so that we shut it down, otherwise we can run out of JVM threads when running many tests. See TestThrottledAsyncQueue for an example.

[anoopj]

Will do.

[electrum]

These need to be updated to match the new names. This is the cause of the Travis CI failure.

[anoopj]

Fixed. Sorry for missing that.

[anoopj]

@electrum

Updated the PR with feedback.

[electrum]

Merged, thanks!