Support distribution of broadcast table using disk storage in Presto-on-Spark

[pgupta2]

== Test plan ==

• Added unit tests to trigger broadcast join using disk.
• Ran verifier for around 100 broadcast join queries.

== RELEASE NOTES ==
General Changes

• Add support for distributing broadcast table using permanent storage, thereby removing spark driver from the distribution flow
  This feature can be enabled/disabled using 'distribute_broadcast_table_using_disk' session property

Hive Changes

• None

[linux-foundation-easycla]

The committers are authorized under a signed CLA.

• ✅ Arjun Gupta (c477f84eb68e2be06fe5c3b51e99db8b87b946cd, c62a7b323446bcd4af8c9a07392b5c0591cf2e3b, f9d6568e0033836c23df7681ec780cf3c5dc5313, 755d0482f85f3135c4c659eda7995bb1c2956b7c)

[pgupta2]

This will be disabled by default.

[arhimondr]

Generally looks good to me, some comments

[arhimondr]

LocalTempStorage is expected to be used to store files locally, and generally local files are not expected to be remotely accessible and are not expected to survive the restart.

For broadcast only a remote storage service can be used, and files should survive the restart.

What do you think about adding a getCapabilities method to the StorageService interface that would return a Set<StorageCapabilities>, where capabilities could be an enum, e.g.:

StorageCapabilities {
REMOTELY_ACCESSIBLE,
PERSISTANT_BEETWEEN_RESTARTS,
}

Then the broadcast implementation should check if the configured storage provides required capabilities.

CC: @wenleix @viczhang861

[arhimondr]

Originally the api of this interface used to use URI or Path as an identifier. But then we realized that some storages (e.g.: Manifold) may not have primary keys in a form of a URL. That's why we decided to go with an opaque StorageHandle. But as you have noticed the StorageHandle is not serializable. We thought about adding a serialization methods later (as for the purpose of spilling serialization wasn't needed).

What do you think about adding 2 methods to the storage interface:

byte[] serializeHandle(StorageHandle storageHandle)
StorageHandle deserialize(byte[] serializedStorageHandle)

?

[pgupta2]

Sure.. The problem with storage handle was that it cannot be serialized. I fixed it by adding a new API that accepts URI which is serializable (under the assumption that that every storage will have URI). Ideal option was to make StorageHandle serializable which will work for any storage implementations

[arhimondr]

StorageHandle is just a marker interface. It might be hard to enforce that every implementation is Java serializable. Though I suggest to add 2 explicit methods to the StorageService interface to serialize / deserialize handles.

[arhimondr]

Please back that by a configuration property as well

[arhimondr]

nit: extract the RddAndMore into a standalone class, since not it is used not only by the PrestoSparkQueryExecutionFactory

[arhimondr]

Sometimes pages could be very small. From what I remember the storage implementation is not required to buffer. If that's the case - we should buffer up to some amount here. For PageFile format we buffer up to 24mb by default (that's an optimal block size for tempfs, but it is configurable).

CC: @wenleix @viczhang861

[arhimondr]

nit: UncheckedIOException

[arhimondr]

nit: ditto about wrapping

[arhimondr]

Let's make the broadcastId a String and use a UUID, that is simpler and more reliable. After you use a local temporary file as a cache the requirement for the ID to be a long will no longer be there

[pgupta2]

This broadcastId is being used to cache HT in spark's blockmanager. The blockmanager API requires us to pass a BlockID object which could be of different type. BroadcastBlockId class accepts a long identifier. There is another blockId called TestBlockId that accepts a string. We can use it but its sounded counter-intuitive since it is meant for testing purposes.

We cannot extend BlockId class since its sealed and we cannot add a new class since we depend on OSS spark artifacts. So, either we use what we have now or we can use TestBlockId which will accept string as input.

[arhimondr]

let's wrap remove in a try-catch

[arhimondr]

Some comments

[arhimondr]

You can simply do uri.toString().getBytes(UTF_8)

[arhimondr]

In Presto we prefer using airlift.json. You can have a look at JsonMapper. But for the purpose of serialization for this class we can simply use uri.toString()

[arhimondr]

nit: maybe storageBasedBroadcastJoinEnabled (also change the config names)

[arhimondr]

nit: maybe storageBasedBroadcastJoinWriteBufferSize (also change the config names)

[arhimondr]

nit: if you decide to rename config names please don't forget to rename the session properties to keep the naming consistent

[arhimondr]

Since now we have control over caching we need to reserve the memory used by the cache using the Presto memory reservation mechanism. It might be somehow tricky though. Presto offers LocalMemoryContext that is "per operator". There will be several instances of the PrestoSparkRemoteSourceOperator for a single plan node, though the memory for caching should be accounted only once.

[arhimondr]

Ideally it would be best to cache already deserialized pages. Though deserialization should be done outside the lock. Only reading pages from the input stream must be done under the lock.

[arhimondr]

don't forget to change the config here (to use the broadcast one)

[arhimondr]

Make sure it is closed if the execution terminates exceptionally (it may perform some cleanups on close)

[arhimondr]

Generally looks good. A whole bunch of nits though. We should be good to go once the comments are resolved.

[arhimondr]

static final

[pgupta2]

Its a singleton class. Do we still need to make it static?

[arhimondr]

Oh, right. That's a good point. Yeah, we don't have to make it static. Let's make it final though.

[arhimondr]

This might cause the ConcurrentModificationException. Create a copy of the key set before iterating.

[arhimondr]

nit: I would recommend removing this method as it is not very clear what size does it return (e.g.: number of pages, number of cached stages, size in bytes, etc.)

[arhimondr]

The ClassLayout.parseClass(BroadcastTableCacheKey.class).instanceSize() will only include the size of the BroadcastTableCacheKey object itself, it won't include the size of the stageId and planNodeId objects. Since the memory footprint of the cache keys are rather very small I would recommend simply not count for the keys memory usage for simplicity.

[arhimondr]

This method will be called on every page. I would recommend caching the current retained size of a cache in a private static long variable, and recompute the size in the cache method

[arhimondr]

Drop the String.format(, the Airlift logger can do formatting

[arhimondr]

ImmutableList.builder()

[arhimondr]

This could potentially be immutable or shared, I would recommend creating a copy, and then doing shuffle

[arhimondr]

remove the requireNonNull check

[arhimondr]

LGTM % comments

[arhimondr]

It seems like a rebase artifact

[arhimondr]

Check for ioException != exception, as self suppression is not allowed

[arhimondr]

ditto

[arhimondr]

as an afterthought: Although the cache is currently used always from a single thread, since it is a singleton it might be accidentally used from different threads. I would recommend you protecting all the public methods with synchronized to be on the safe side.