Subquery cache & friends #21888

sopel39 · 2024-05-09T09:23:28Z

Implement subquery cache for Hive/Iceberg/Delta

Subquery cache is a lightweight mechanism for caching
source stage computations. It works across queries, but
also within a query if similar subqueries are identified.

Subquery cache works with both streaming and FTE mode. Cache
results are never stalled since data is cached per split. Dedicated
"cache splits ids" include create time and change set
(in case of Delta/Iceberg).

Subquery cache works as follows:
1. During planning, subqueries eligible for caching
   are identified. If there are similar subqueries within
   a query, then common subplan is extracted.
2. Query plan is rewritten using caching plan alternatives
   (fallback to original subquery, cache data, load from cache)
3. SPI PlanSignatures are computed for each cached subquery
4. Splits are scheduled deterministically on nodes based on (PlanSignature, SplitId) pair
5. On the worker cache plugin (currently only memory based) will determine
   if cached data is available for a given split

raunaqmorarka · 2024-05-10T04:47:44Z

core/trino-main/src/main/java/io/trino/execution/scheduler/UniformNodeSelector.java


 List<InternalNode> filteredNodes = filterNodes(nodeMap, includeCoordinator, ImmutableSet.of());
 ResettableRandomizedIterator<InternalNode> randomCandidates = new ResettableRandomizedIterator<>(filteredNodes);
 Set<InternalNode> schedulableNodes = new HashSet<>(filteredNodes);

- // optimizedLocalScheduling enables prioritized assignment of splits to local nodes when splits contain locality information
- if (optimizedLocalScheduling) {


Can we remove optimizedLocalScheduling from NodeSchedulerConfig ?

Can you make a separate PR out of the last commit ? It's unrelated to everything else and can be landed quickly

I think we can also need to change implementation of hive.force-local-scheduling. With current changes we're effectively forcing local scheduling by default in hive (unless node went down). That's probably not the right default as it could create hot spots in the cluster. Instead of flipping isRemotelyAccessible, that flag should now change whether or not host addresses are returned by hive connector and by default we shouldn't provide addresses from HDFS.

deigote · 2024-05-27T13:20:45Z

Hi 👋🏽 maybe a dumb question, but from subquery cache for Hive/Iceberg/Delta I'm not clear if this is about a subquery cache for the Hive/Iceberg/Delta connectors, or rather a cache for any connector that uses Hive/Iceberg/Delta under the hood. I'm hoping the latter but it'd be great if you could clarify 🙏🏽 !

sopel39 · 2024-05-27T14:07:04Z

or rather a cache for any connector that uses Hive/Iceberg/Delta under the hood.

@deigote I'm not sure what you mean by any connector that uses Hive/Iceberg/Delta under the hood. However, this PR makes subquery cache a 1st class citizen, where source of data can be from any connector as long as connector implements getCacheTableId, getCacheColumnId, getCacheSplitId

deigote · 2024-05-27T14:23:27Z

Thanks! That's what I was hoping for. I got confused by the PR's description saying Implement subquery cache for Hive/Iceberg/Delta which seemed to imply the PR was only about the Hive / Iceberg / Delta connectors. But what I get is with this PR:

Trino offers any connector the ability to leverage subquery caching
The Hive / Iceberg / Delta already use said ability to implement subquery caching

sopel39 · 2024-05-29T19:18:51Z

Removed dynamic row filtering from PR as it will be handled separately (#22175 (comment))

kekwan · 2024-06-06T21:44:19Z

Looking forward to this. Would this work also solve CTE #10?

deigote · 2024-06-06T21:52:42Z

@kekwan the way I understood it, it wouldn't "solve" it but it'd contribute to making it a much less severe issue. The CTEs would still execute twice, but their results would be cached on quite a low level. Hopefully the cache hit ratio would be very high but I'm guessing it'd depend on how busy the workers are (I'm assuming the busier they are the more cache evictions).

ChooseAlternativeNode defines alternative sub-plans that can be used to execute given part of the query. The actual sub-plan is then chosen per split during task execution. Alternative sub-plans cannot span multiple stages and are only supported for source stages. Co-authored-by: Assaf Bern <[email protected]>

These methods are required by subquery cache to describe split data for cache key purpose. ConnectorPageSourceProvider#getUnenforcedPredicate is used to describe what unenforced predicate will be applied on split data. ConnectorPageSourceProvider#prunePredicate is used to simplify filter predicates on per split bases (e.g. removing paritioning predicates that fully contain split data) Co-authored-by: Kamil Endruszkiewicz <[email protected]> Co-authored-by: radek <[email protected]>

CacheManager is a set of SPI classes for implementing split level cache storage. MemoryCacheManager is a high-performance implementation of CacheManager that keeps cached data in revocable memory.

Cache table id together with split id and column id represent rows produced by ConnectorPageSource for a given split. Cache ids can also be used to canonicalise query plans for the purpouse of comparison or cache key generation. This commit implements cache ids for Hive, Iceberg, Delta and TPCH connectors. Co-authored-by: Kamil Endruszkiewicz <[email protected]> Co-authored-by: radek <[email protected]> Co-authored-by: lukasz-stec <[email protected]>

Cache hit rate depend on deterministic split generation. Hive connector has a concept of "initial splits" which are smaller and there is a limited of them. Therefore, if deterministic splits are required, then initial splits must be disabled because Hive split generation doesn't have guaranteed ordering.

Dynamic filter id might be registered by both local join and as coming from coordinator.

CanonicalSubplanExtractor creates a canonical representation of a subplan using cache ids provided by the connector. Canonical subplans are used to compare plans against each other and enable extracting of common subplans. Co-authored-by: Kamil Endruszkiewicz <[email protected]>

Subquery cache is a lightweight mechanism for caching source stage computations. It works across queries, but also within a query if similar subqueries are identified. Subquery cache works with both streaming and FTE mode. Cache results are never stalled since data is cached per split. Dedicated "cache splits ids" include create time and change set (in case of Delta/Iceberg). Subquery cache works as follows: 1. During planning, subqueries eligible for caching are identified. If there are similar subqueries within a query, then common subplan is extracted. 2. Query plan is rewritten using caching plan alternatives (fallback to original subquery, cache data, load from cache) 3. SPI PlanSignatures are computed for each cached subquery 4. Splits are scheduled deterministically on nodes based on (PlanSignature, SplitId) pair 5. On the worker cache plugin (currently only memory based) will determine if cached data is available for a given split Co-authored-by: Kamil Endruszkiewicz <[email protected]> Co-authored-by: radek <[email protected]> Co-authored-by: lukasz-stec <[email protected]> Co-authored-by: Raunaq Morarka <[email protected]>

sopel39 · 2024-06-11T13:13:02Z

rebased after #22190

martint

Some initial comments. Still reviewing.

martint · 2024-06-24T21:49:20Z

core/trino-main/src/main/java/io/trino/operator/DriverFactory.java

@@ -28,7 +29,7 @@ public interface DriverFactory

 OptionalInt getDriverInstances();

- Driver createDriver(DriverContext driverContext);
+ Driver createDriver(DriverContext driverContext, Optional<ScheduledSplit> split);


What's the motivation and purpose for this new argument?

This was added to support alternative plans for the source stage. The alternative in this context is a concrete list of operators (a Driver instance) chosen based on the split.

martint · 2024-06-24T23:47:56Z

core/trino-main/src/main/java/io/trino/sql/planner/plan/ChooseAlternativeNode.java

+{
+ private final List<PlanNode> alternatives;
+
+ private final FilteredTableScan originalTableScan;


What's this for? What does "original" mean in this context (e.g. what if the plan is formulated with a set of alternatives from the get go?)

Also, this seems overly specific. What if the original plan had operations other than a table scan and filter?

This was named originalTableScan because ChooseAlternativeNode is ATM created for some existing original sub-plan. The reason we have it though is different. Alternatives work only on a source stage level and are chosen based on a split. This means we need a single source of splits for the ChooseAlternativeNode and originalTableScan is exactly that. The filter part is needed to support dynamic filters at the split source level.

martint · 2024-06-24T23:52:21Z

core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java

@@ -794,7 +812,7 @@ public PartitioningHandleReassigner(PartitioningHandle fragmentPartitioningHandl
 }

 @Override
- public PlanNode visitTableScan(TableScanNode node, RewriteContext<Void> context)
+ public TableScanNode visitTableScan(TableScanNode node, RewriteContext<Void> context)


Unrelated change?

martint · 2024-06-25T00:01:53Z

core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java

+ List<PlanNode> newAlternatives = node.getSources().stream()
+ .map(alternative -> context.defaultRewrite(alternative, context.get()))
+ .toList();
+ TableScanNode newTableScan = visitTableScan(node.getOriginalTableScan().tableScanNode(), context);


Visitor methods are not supposed to be called directly. Use rewriteWith(...).

martint · 2024-06-25T00:07:51Z

core/trino-main/src/main/java/io/trino/sql/planner/planprinter/IoPlanPrinter.java

+ filter);
+ return decomposedPredicate.getTupleDomain().transformKeys(scan.getAssignments()::get);
+ }).orElse(TupleDomain.all());
+ addInputTableConstraints(filterDomain, scan, context);


This isn't necessarily correct. It may apply to some alternatives, but not all. It should either pick the most conservative choice (that which applies to all the alternatives), or list all the possible I/O plans.

martint · 2024-06-28T15:26:39Z

...in/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaPageSourceProvider.java

+ predicate
+ )).isEqualTo(TupleDomain.all());


Formatting:

assertThat(pageSourceProvider.prunePredicate( TEST_SESSION.toConnectorSession(), prepareSplit( TupleDomain.withColumnDomains(ImmutableMap.of(regularColumnHandle, Domain.multipleValues(BIGINT, ImmutableList.of(0L)))), ImmutableMap.of("partitionedColumn", Optional.of("0"))), tableHandle, predicate)) .isEqualTo(TupleDomain.all());

martint · 2024-06-28T15:27:07Z

...in/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaPageSourceProvider.java

+ }
+
+ @Test
+ public void testPrunePredicateWhenSplitIsFilteredOut()


Check formatting in the methods below, too.

martint · 2024-06-28T15:29:56Z

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorPageSourceProvider.java

+ ConnectorSession session,
+ ConnectorSplit split,
+ ConnectorTableHandle table,
+ TupleDomain<ColumnHandle> dynamicFilter)


Why is the argument named "dynamicFilter"? Rename it to "constraint"

martint · 2024-06-28T15:43:22Z

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorPageSourceProvider.java

@@ -13,6 +13,8 @@
 */
 package io.trino.spi.connector;

+import io.trino.spi.predicate.TupleDomain;
+
 import java.util.List;

 public interface ConnectorPageSourceProvider


The new methods seem misplaced in this class. Why are they associated with the PageSourceProvider (and not RecordSetProvider? The should really live outside of either of those, as they have nothing to do with a data stream itself. They are more related to split management.

martint · 2024-06-28T17:43:12Z

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorPageSourceProvider.java

+ * Prunes columns from predicate that are not effective in filtering split data.
+ * If split is completely filtered out by given predicate, then this
+ * method must return {@link TupleDomain#none}.


This is a weird mix of column pruning and constraint pruning (i..e, a column-wise intersection and a all-or-none intersection of the constraint). It would be more general and easier to reason about if it were a pure intersection of the given tupledomain with the constraint guaranteed by the split.

I can't tell yet what's the right abstraction as there are no uses of it up to this commit -- I will revisit once I review the rest of the PR.

martint

Some more comments and questions as I continue perusing the code.

martint · 2024-07-01T22:52:32Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveSplitManager.java

+ ConnectorSession session,
+ ConnectorTableHandle tableHandle,
+ DynamicFilter dynamicFilter,
+ boolean preferDeterministicSplits,


We talked about this offline a few weeks ago. Instead of adding this, we should revisit whether the adaptive split logic is still useful and remove it if not. It's almost certainly not useful for data formats such as ORC and Parquet that cannot be split across row group boundaries.

martint · 2024-07-01T23:10:57Z

core/trino-spi/src/main/java/io/trino/spi/cache/CacheManager.java

+ * @param predicate Predicate that should be enforced on cached rows.
+ * Output of `cachedSplitA` can be used to derive output of matching `cachedSplitB`
+ * (with corresponding {@link PlanSignature}) as long as `cachedSplitB.predicate` is a strict
+ * subset of `cachedSplitA.predicate`. To do so, `cachedSplitB.predicate` must be
+ * applied on output of `cachedSplitA`. Before serialization as a cache key, predicate
+ * needs to be normalized using {@code io.trino.plugin.base.cache.CacheUtils#normalizeTupleDomain(TupleDomain)}.
+ * @param unenforcedPredicate Unenforced (best-effort) predicate that should be applied on cached rows.
+ * Output of `cachedSplitA` can be used to derive output of matching `cachedSplitB`
+ * (with corresponding {@link PlanSignature}) as long as `cachedSplitB.unenforcedPredicate`
+ * is a subset of `cachedSplitA.unenforcedPredicate`. Before serialization as a cache key, predicate
+ * needs to be normalized using {@code io.trino.plugin.base.cache.CacheUtils#normalizeTupleDomain(TupleDomain)}.
+ * @return cached pages for a given split.


Replace markdown code quotes with proper javadoc code tags.

martint · 2024-07-01T23:13:48Z

core/trino-spi/src/main/java/io/trino/spi/cache/CacheManager.java

+ * applied on output of `cachedSplitA`. Before serialization as a cache key, predicate
+ * needs to be normalized using {@code io.trino.plugin.base.cache.CacheUtils#normalizeTupleDomain(TupleDomain)}.


How is this enforced? Who is responsible for the normalization? What happens if it's not normalized?

martint · 2024-07-01T23:14:28Z

core/trino-spi/src/main/java/io/trino/spi/cache/CacheManager.java

+ * subset of `cachedSplitA.predicate`. To do so, `cachedSplitB.predicate` must be
+ * applied on output of `cachedSplitA`. Before serialization as a cache key, predicate
+ * needs to be normalized using {@code io.trino.plugin.base.cache.CacheUtils#normalizeTupleDomain(TupleDomain)}.
+ * @param unenforcedPredicate Unenforced (best-effort) predicate that should be applied on cached rows.


It's not clear what this predicate is for.

martint · 2024-07-01T23:15:25Z

core/trino-spi/src/main/java/io/trino/spi/cache/CacheSplitId.java

+import static io.airlift.slice.SizeOf.instanceSize;
+import static java.util.Objects.requireNonNull;
+
+public class CacheSplitId


This should be a record

martint · 2024-07-01T23:18:31Z

core/trino-spi/src/main/java/io/trino/spi/cache/SignatureKey.java

+import static io.airlift.slice.SizeOf.instanceSize;
+import static java.util.Objects.requireNonNull;
+
+public class SignatureKey


This should be a record

martint · 2024-07-01T23:28:56Z

core/trino-main/src/main/java/io/trino/cache/CanonicalSubplan.java

+ * {@link ColumnHandle}). {@link CacheColumnId} for complex projections will use canonicalized and formatted
+ * version of projection expression.
+ */
+public class CanonicalSubplan


What's the relationship between CanonicalSubplan and PlanSignature?

martint · 2024-07-01T23:31:40Z

core/trino-main/src/main/java/io/trino/cache/CanonicalSubplan.java

+ * {@link ColumnHandle}). {@link CacheColumnId} for complex projections will use canonicalized and formatted
+ * version of projection expression.
+ */
+public class CanonicalSubplan


Why do we need a dedicated class for this? Why isn't a regular PlanNode tree sufficient to represent a canonical structure (after canonicalization, of course)?

martint · 2024-07-01T23:35:07Z

core/trino-spi/src/main/java/io/trino/spi/cache/ConnectorCacheMetadata.java

+ /**
+ * Returns a table identifier for the purpose of caching with {@link CacheManager}.
+ * {@link CacheTableId} together with {@link CacheSplitId} and {@link CacheColumnId}s represents
+ * rows produced by {@link ConnectorPageSource} for a given split. Local table properties
+ * (e.g. rows order) must be part of {@link CacheTableId} if they are present. List of selected
+ * columns should not be part of {@link CacheTableId}. {@link CacheTableId} should not contain
+ * elements that can be derived from {@link CacheSplitId} such as predicate on partition column
+ * which can filter splits entirely.
+ */


This is too complicated. Why are all those conditions required?

What does it mean for "List of selected columns should not be part of ...", especially in the case of a ConnectorTableHandle representing a pushed-down subplan?

martint · 2024-07-01T23:38:43Z

core/trino-spi/src/main/java/io/trino/spi/cache/ConnectorCacheMetadata.java

+ * are eligible for caching with {@link CacheManager}. Connector should convert provided
+ * {@link ConnectorTableHandle} into canonical one by pruning of every non-canonical field.
+ */
+ ConnectorTableHandle getCanonicalTableHandle(ConnectorTableHandle handle);


Not sure I understand what this is. ConnectorTableHandle represents an opaque reference to a table that can be carried in query plans, transmitted to workers, etc. What is a "property of a ConnectorTableHandle"? What does it mean for it to "affect final query results when underlying table is queried"?

Also, what's the purpose of this, given that there's getCacheTableId above?

cla-bot bot added the cla-signed label May 9, 2024

github-actions bot added iceberg Iceberg connector delta-lake Delta Lake connector hive Hive connector labels May 9, 2024

sopel39 force-pushed the ks/subquery_cache branch 2 times, most recently from 9f4aa11 to ad12339 Compare May 9, 2024 16:10

raunaqmorarka reviewed May 10, 2024

View reviewed changes

sopel39 force-pushed the ks/subquery_cache branch 5 times, most recently from c270355 to a72b5df Compare May 15, 2024 12:16

sopel39 force-pushed the ks/subquery_cache branch from a72b5df to 4f474e2 Compare May 21, 2024 14:55

github-actions bot added the ui Web UI label May 21, 2024

sopel39 force-pushed the ks/subquery_cache branch 4 times, most recently from f771ae0 to a0ab7fc Compare May 22, 2024 08:25

sopel39 marked this pull request as ready for review May 22, 2024 08:25

sopel39 changed the title ~~WIP: Subquery cache & friends~~ Subquery cache & friends May 22, 2024

sopel39 force-pushed the ks/subquery_cache branch 5 times, most recently from 865c615 to 74302ec Compare May 23, 2024 12:38

This was referenced May 24, 2024

Reuse table scan results when the same table is used in different parts of query #5880

Closed

Subquery cache roadmap #22114

Open

hackeryang added the performance label May 27, 2024

sopel39 force-pushed the ks/subquery_cache branch from 74302ec to d80cb1d Compare May 27, 2024 21:36

sopel39 mentioned this pull request May 28, 2024

Normalize conjunct order in CanonicalSubplanExtractor #22120

Closed

sopel39 force-pushed the ks/subquery_cache branch from d80cb1d to d7af76b Compare May 28, 2024 13:07

sopel39 mentioned this pull request May 28, 2024

Implement dynamic row filtering in hive, iceberg and delta #22175

Closed

sopel39 requested review from martint May 28, 2024 16:00

sopel39 mentioned this pull request May 29, 2024

Make scheduling of remote accessible splits with addresses more strict #22190

Merged

sopel39 force-pushed the ks/subquery_cache branch from d7af76b to bc4d54b Compare May 29, 2024 19:17

sopel39 force-pushed the ks/subquery_cache branch from bc4d54b to 9e3e422 Compare June 11, 2024 13:03

lukasz-stec and others added 13 commits June 11, 2024 15:03

Rename DriverFactory to OperatorDriverFactory

843a4b2

Introduce DriverFactory interface

1e6096f

Add optional split parameter to DriverFactory#createDriver

8d1fb66

Add CacheManager SPI and MemoryCacheManager implementation

2f386ef

CacheManager is a set of SPI classes for implementing split level cache storage. MemoryCacheManager is a high-performance implementation of CacheManager that keeps cached data in revocable memory.

Simplify LocalExecutionPlanContext#addDriverFactory

839c07d

Allow to register same dynamic filter id multiple times

53b0371

Dynamic filter id might be registered by both local join and as coming from coordinator.

Rename pageSourceProvider to pageSourceManager

a2aa506

sopel39 force-pushed the ks/subquery_cache branch from 9e3e422 to a2aa506 Compare June 11, 2024 13:13

martint reviewed Jun 28, 2024

View reviewed changes

martint reviewed Jul 1, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Subquery cache & friends #21888

Subquery cache & friends #21888

sopel39 commented May 9, 2024 •

edited

Loading

raunaqmorarka May 10, 2024

raunaqmorarka May 10, 2024

raunaqmorarka May 10, 2024

deigote commented May 27, 2024

sopel39 commented May 27, 2024

deigote commented May 27, 2024

sopel39 commented May 29, 2024

kekwan commented Jun 6, 2024 •

edited

Loading

deigote commented Jun 6, 2024

sopel39 commented Jun 11, 2024

martint left a comment

martint Jun 24, 2024

lukasz-stec Jul 5, 2024

martint Jun 24, 2024

lukasz-stec Jul 5, 2024

martint Jun 24, 2024

martint Jun 25, 2024

martint Jun 25, 2024

martint Jun 28, 2024

martint Jun 28, 2024

martint Jun 28, 2024

martint Jun 28, 2024

martint Jun 28, 2024

martint left a comment

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

martint Jul 1, 2024

		* applied on output of `cachedSplitA`. Before serialization as a cache key, predicate
		* needs to be normalized using {@code io.trino.plugin.base.cache.CacheUtils#normalizeTupleDomain(TupleDomain)}.

Subquery cache & friends #21888

Are you sure you want to change the base?

Subquery cache & friends #21888

Conversation

sopel39 commented May 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

deigote commented May 27, 2024

sopel39 commented May 27, 2024

deigote commented May 27, 2024

sopel39 commented May 29, 2024

kekwan commented Jun 6, 2024 • edited Loading

deigote commented Jun 6, 2024

sopel39 commented Jun 11, 2024

martint left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

martint left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sopel39 commented May 9, 2024 •

edited

Loading

kekwan commented Jun 6, 2024 •

edited

Loading