Use dynamic filter to prune Iceberg splits based on partition values #9193

alexjo2144 · 2021-09-09T22:10:25Z

alexjo2144 · 2021-09-09T22:34:18Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitManager.java

+                session.getTimeZoneKey(),
+                maxSplitsPerSecond,
+                maxOutstandingSplits,
+                new BoundedExecutor(executorService, splitLoaderConcurrency),


I'm not sure this needs to be configurable, might just want to hardcode the thread count.

Maybe iceberg is using multiple threads internally, but isn't the split loading is still single threaded per split source on Trino side ?

Can we hard-code splitLoaderConcurrency to 1 given that we're not parallelising split loading in IcebergSplitSource ?

That, or we could just pass the ExecutorService straight through without wrapping in BoundedExecutor

It's not completely single threaded because the AsyncQueue uses the executor to do some of its operations.

Wouldn't setting splitLoaderConcurrency to 1 result in a deadlock, as we would use one thread for:

addExceptionCallback(ResumableTasks.submit(executor, loaderTask), this::onLoaderFailure);

And then there would be no threads left for splitQueue handling:

new ThrottledAsyncQueue<>(maxSplitsPerSecond, maxOutstandingSplits, executor);

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

.../trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergDynamicPartitionPruning.java

raunaqmorarka · 2021-09-10T05:47:37Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitManager.java

+                session.getTimeZoneKey(),
+                maxSplitsPerSecond,
+                maxOutstandingSplits,
+                new BoundedExecutor(executorService, splitLoaderConcurrency),


Maybe iceberg is using multiple threads internally, but isn't the split loading is still single threaded per split source on Trino side ?

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitManager.java

raunaqmorarka · 2021-09-10T06:13:03Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

+            return TaskStatus.finished();
+        }
+
+        public boolean partitionPassesDynamicFilter(Map<Integer, String> partitionKeys)


Can we have common code for this and TableStatisticsMaker#dataFileMatches ?

Can we also look at file level Domain for the columns for which we have dynamic filter and prune splits based on that ?
We don't necessarily have to restrict ourselves to partitioned columns here.

alexjo2144 · 2021-09-13T17:58:32Z

Comments address, thanks @raunaqmorarka

raunaqmorarka

Please share tpch/tpcds benchmark results after applying current round of comments

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConfig.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

raunaqmorarka · 2021-09-15T08:13:06Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

+    private void checkLoaderException()
+    {
+        if (loaderException != null) {
+            if (loaderException instanceof TrinoException) {


For IOException maybe we should throw TrinoException(ICEBERG_FILESYSTEM_ERROR, e)
Though I don't know what kind of exceptions can be produced here

Maybe. The none of the Iceberg TableScan methods throw a checked IOException, but it seems reasonable that they could.

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergSplitLoader.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitLoader.java

testing/trino-testing/src/main/java/io/trino/testing/AbstractTestDynamicPartitionPruning.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

findepi

(skim)

findepi · 2021-09-17T12:57:42Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergModule.java

+    @Provides
+    public ExecutorService createIcebergExecutor(CatalogName catalogName)
+    {
+        return newCachedThreadPool(daemonThreadsNamed("iceberg-" + catalogName + "-%s"));


we should cleanly forbid catalog names with %s (or sanitize catalogName in places like this)

preexisting (for hive), no change requested here

findepi · 2021-09-17T13:01:38Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

-        }
-        catch (IOException e) {
-            throw new UncheckedIOException(e);
+        checkLoaderException();


any reason to try fail close?

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

electrum

The split queuing adds significant complexity. Why do we need this feature? It was added to Hive because someone wanted to reduce load on HDFS, but I haven’t heard anyone ask about this for Iceberg.

sopel39 · 2021-09-17T18:42:17Z

The split queuing adds significant complexity. Why do we need this feature? It was added to Hive because someone wanted to reduce load on HDFS, but I haven’t heard anyone ask about this for Iceberg.

It's not only about load, but also performance. DF brings huge perf improvements. Therefore if Iceberg is much slower than Hive connector, it won't be a feasible alternative.

sopel39 · 2021-09-17T18:43:01Z

@alexjo2144 would it be possible to get macro benchmarks result?

electrum · 2021-09-17T20:36:46Z

Dynamic filtering is obviously something we want. But I don't see why it is tied with split throttling. If we want to block, we can simply return the DF future from ConnectorSplitSource.getNextBatch():

ConnectorSplitBatch EMPTY_BATCH = new ConnectorSplitBatch(ImmutableList.of(), false);

long startTime; // time of first getNextBatch() call
long elapsed = NANOS.toMillis(System.nanoTime() - startTime);
long remainingWait = waitTime.toMillis() - elapsed;

if ((remainingWait > 0) && dynamicFilter.isBlocked()) {
    return dynamicFilter.isBlocked()
            .thenApply(ignored -> EMPTY_BATCH)
            .completeOnTimeout(EMPTY_BATCH, remainingWait, MILLIS);
}

alexjo2144 · 2021-09-17T22:26:01Z

Dynamic filtering is obviously something we want. But I don't see why it is tied with split throttling

I can take the throttling part out. It only got included because I was trying to mimic some of the Hive pattern and used ThrotledAsyncQueue, but we can use a non-throttled version.

raunaqmorarka · 2021-09-19T14:51:25Z

@electrum In hive connector, although we've implemented the blocking for DF, we keep it turned off by default to avoid any regressions for small queries. If we keep it turned off by default here as well, then would that just result in all the splits getting generated and scheduled up front ? If yes, then we would have to set some non-zero default for the DF blocking timeout and that can potentially slow down some short running queries or queries which waited on DF but the DF wasn't useful.
Also, does throttling of split generation help with reducing memory usage on coordinator or reducing wasted work when a table scan is aborted due to a LIMIT ?

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitManager.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

losipiuk · 2021-10-21T15:41:52Z

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergSplitSource.java

+                    .map(IcebergSplit.class::cast)
+                    .forEach(splits::add);
+        }
+        assertThat(splits.build().size()).isGreaterThan(0);


You can assert that you actually waited ~2s here.
Btw can you make the test without waiting? Maybe using artificial Ticker instead wall time.
You may also verify that future returned from splitSource.getNextBatch gets unblocked when you complete future returned from DynamicFilter.isBlocked

Ignore if too painful

alexjo2144 · 2021-10-21T20:12:08Z

All set, thanks for the comments @losipiuk

findepi · 2021-10-22T12:47:19Z

CI #9620 (reopened)

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

findepi · 2021-10-22T14:15:09Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

+                    column.getType(),
+                    fromByteBuffer(type, lowerBounds.get(fieldId)),
+                    fromByteBuffer(type, upperBounds.get(fieldId)),
+                    mayContainNulls);


we can realize all-null column when scanTask.file().valueCounts() matches scanTask.file().nullValueCounts() at given key. then we could create Domain.onlyNull for such case.

let's maybe do this in follow up

ya, we do that in TupleDomainParquetPredicate as well

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergV2.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java

alexjo2144 · 2021-10-22T17:36:47Z

Discussed offline, but I removed the Executor and any async handling, it does not seem to be useful. SourcePartitionedScheduler only ever has one outstanding batch at a time for non-grouped execution, so doing the partition filtering on an async thread does not allow for parallel execution unless we also have grouped execution.

alexjo2144 · 2021-10-22T20:10:37Z

CI Flake: #7224

losipiuk · 2021-10-25T08:19:40Z

Changes LGTM

cla-bot bot added the cla-signed label Sep 9, 2021

alexjo2144 mentioned this pull request Sep 9, 2021

Support dynamic filtering in Iceberg connector #4115

Closed

alexjo2144 commented Sep 9, 2021

View reviewed changes

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch from 2ba2e75 to 67934a5 Compare September 9, 2021 22:43

alexjo2144 requested review from electrum, raunaqmorarka, sopel39 and findepi September 9, 2021 22:44

raunaqmorarka reviewed Sep 10, 2021

View reviewed changes

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch 3 times, most recently from 497985f to 3d85856 Compare September 13, 2021 17:31

alexjo2144 requested a review from raunaqmorarka September 13, 2021 17:58

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch from 3d85856 to e34a1f8 Compare September 13, 2021 18:32

raunaqmorarka reviewed Sep 15, 2021

View reviewed changes

findepi requested a review from losipiuk September 16, 2021 14:52

losipiuk reviewed Sep 16, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Sep 16, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Sep 16, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

findepi reviewed Sep 17, 2021

View reviewed changes

electrum requested changes Sep 17, 2021

View reviewed changes

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch from e34a1f8 to 3823f26 Compare September 17, 2021 16:22

raunaqmorarka mentioned this pull request Sep 22, 2021

Iceberg: support dynamic filter for page source and get splits #9328

Closed

findepi mentioned this pull request Sep 28, 2021

Background iceberg split loader #9404

Closed

losipiuk reviewed Oct 21, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitManager.java Outdated Show resolved Hide resolved

losipiuk reviewed Oct 21, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Oct 21, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Oct 21, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Oct 21, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Oct 21, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

losipiuk reviewed Oct 21, 2021

View reviewed changes

losipiuk approved these changes Oct 21, 2021

View reviewed changes

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch from d2c9e08 to 103a99d Compare October 21, 2021 19:32

alexjo2144 added 3 commits October 21, 2021 15:52

Verify query correctness in TestHiveDynamicPartitionPruning

237137a

Add AbstractTestDynamicPartitionPruning test class

1b231c7

Add back test for Hive dynamic filtering of partition

0fb2b31

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch from 103a99d to fcc241c Compare October 21, 2021 19:52

findepi approved these changes Oct 22, 2021

View reviewed changes

findepi reviewed Oct 22, 2021

View reviewed changes

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java Outdated Show resolved Hide resolved

alexjo2144 mentioned this pull request Oct 22, 2021

IcebergSplitSource: Support large IN predicates #9743

Closed

Use dynamic filter to prune Iceberg splits

0924252

alexjo2144 force-pushed the iceberg/dynamic-filter-partitions branch from fcc241c to 0924252 Compare October 22, 2021 17:32

empty

2bb6095

findepi merged commit 8f767f2 into trinodb:master Oct 25, 2021

findepi mentioned this pull request Oct 25, 2021

Release notes for 364 #9534

Closed

12 tasks

github-actions bot added this to the 364 milestone Oct 25, 2021

manupatteri mentioned this pull request Mar 9, 2022

Refactor Local Dynamic Filter test cases #11379

Closed

sopel39 mentioned this pull request Mar 14, 2022

Extract abstract dynamic filtering tests for connectors #5776

Closed

alexjo2144 mentioned this pull request Apr 8, 2022

Load Iceberg splits on a background thread using AsyncQueue #11872

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use dynamic filter to prune Iceberg splits based on partition values #9193

Use dynamic filter to prune Iceberg splits based on partition values #9193

alexjo2144 commented Sep 9, 2021 •

edited by findepi

Loading

alexjo2144 Sep 9, 2021

raunaqmorarka Sep 10, 2021

raunaqmorarka Sep 15, 2021

alexjo2144 Sep 16, 2021

alexjo2144 Sep 16, 2021

losipiuk Sep 16, 2021

raunaqmorarka Sep 10, 2021

raunaqmorarka Sep 10, 2021

raunaqmorarka Sep 10, 2021

alexjo2144 commented Sep 13, 2021

raunaqmorarka left a comment

raunaqmorarka Sep 15, 2021

alexjo2144 Sep 16, 2021

findepi left a comment

findepi Sep 17, 2021

findepi Sep 17, 2021

electrum left a comment

sopel39 commented Sep 17, 2021

sopel39 commented Sep 17, 2021

electrum commented Sep 17, 2021 •

edited

Loading

alexjo2144 commented Sep 17, 2021

raunaqmorarka commented Sep 19, 2021

losipiuk Oct 21, 2021

alexjo2144 commented Oct 21, 2021

findepi commented Oct 22, 2021

findepi Oct 22, 2021

findepi Oct 22, 2021

raunaqmorarka Oct 22, 2021

alexjo2144 commented Oct 22, 2021 •

edited

Loading

alexjo2144 commented Oct 22, 2021

losipiuk commented Oct 25, 2021

Use dynamic filter to prune Iceberg splits based on partition values #9193

Use dynamic filter to prune Iceberg splits based on partition values #9193

Conversation

alexjo2144 commented Sep 9, 2021 • edited by findepi Loading

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

alexjo2144 commented Sep 13, 2021

raunaqmorarka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findepi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

electrum left a comment

Choose a reason for hiding this comment

sopel39 commented Sep 17, 2021

sopel39 commented Sep 17, 2021

electrum commented Sep 17, 2021 • edited Loading

alexjo2144 commented Sep 17, 2021

raunaqmorarka commented Sep 19, 2021

Choose a reason for hiding this comment

alexjo2144 commented Oct 21, 2021

findepi commented Oct 22, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alexjo2144 commented Oct 22, 2021 • edited Loading

alexjo2144 commented Oct 22, 2021

losipiuk commented Oct 25, 2021

alexjo2144 commented Sep 9, 2021 •

edited by findepi

Loading

electrum commented Sep 17, 2021 •

edited

Loading

alexjo2144 commented Oct 22, 2021 •

edited

Loading