testStatsBasedRepartitionDataOnInsert flaky

[losipiuk]

[ERROR] io.trino.plugin.iceberg.TestIcebergOrcConnectorTest.testStatsBasedRepartitionDataOnInsert  Time elapsed: 1.576 s  <<< FAILURE!
java.lang.AssertionError: 
[Rows for query [SELECT count(*) FROM "repartition_SELECTDISTINCTorderkeycustkeyorderstatusFROMtpchtinyordersinsert_orderstatus_1ns2vmr9lb$files"]] 
Expecting:
  <(3)>
to contain exactly in any order:
  <[(9)]>
elements not found:
  <(9)>
and elements not expected:
  <(3)>

	at io.trino.sql.query.QueryAssertions$QueryAssert.lambda$matches$2(QueryAssertions.java:346)
	at io.trino.sql.query.QueryAssertions$QueryAssert.matches(QueryAssertions.java:333)
	at io.trino.sql.query.QueryAssertions$QueryAssert.matches(QueryAssertions.java:328)
	at io.trino.plugin.iceberg.BaseIcebergConnectorTest.testRepartitionData(BaseIcebergConnectorTest.java:2874)
	at io.trino.plugin.iceberg.BaseIcebergConnectorTest.lambda$testStatsBasedRepartitionData$27(BaseIcebergConnectorTest.java:2831)
	at io.trino.testing.assertions.Assert.assertEventually(Assert.java:74)
	at io.trino.testing.assertions.Assert.assertEventually(Assert.java:66)
	at io.trino.testing.assertions.Assert.assertEventually(Assert.java:61)
	at io.trino.plugin.iceberg.BaseIcebergConnectorTest.testStatsBasedRepartitionData(BaseIcebergConnectorTest.java:2830)
	at io.trino.plugin.iceberg.BaseIcebergConnectorTest.testStatsBasedRepartitionDataOnInsert(BaseIcebergConnectorTest.java:2809)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:104)
	at org.testng.internal.Invoker.invokeMethod(Invoker.java:645)
	at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:851)
	at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1177)
	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:129)
	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:112)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)

https://github.com/trinodb/trino/runs/5280874831?check_suite_focus=true

[fengguangyuan]

Is this an occasional failure? I could not reproduce it on Master branch. :(

[ebyhr]

@fengguangyuan This is an occasional failure.

[fengguangyuan]

After a lot spare time on this issue, I think the root cause of the failure is being clear.

Basic test info:
    Table:
         name:          tpch.tiny.orders
         files:         3
         part key:      orderstatus
         part values:   ['F', 'O', 'P']
    Cluster:
         workers: 3

    Expected files count:  9

Conclusion

Imbalanced distribution of data pages on 3 output buffer slots in TaskOutputOperator leads to the unexpected number of written files, therefore the number of files could be in range [3, 9].

Details

First of all, we know that INSERT INTO statement will produce a physical plan, which is roughly described as:
scan stage(scan -> hash -> output) -> exchange stage(exchange -> exchange sink) -> write stage

At the scan stage, the last operator TaskOutputOperator is responsible for putting data pages into it's internal buffer, an instance of ArbitraryOutputBuffer which contains 3 ClientBuffers, and then waiting to be consumed by the upstream tasks in parallel, here these tasks are from exchange stage.

ArbitraryOutputBuffer has 3 output client buffers, implying 3 parallelism, because there are three simulated worker nodes created by the test framework.

Ideally, TaskOutputOperator should keep even distribution of all the data pages on 3 buffer channels, for example if there were 9 pages, each buffer should own 3.

But actually, considering all the data pages are not received by ArbitraryOutputBuffer once, ArbitraryOutputBuffer using an improved round-robin strategy to locate buffers for each enqueue. The code below describes the strategy and is what I used to test:

public class ArbitraryOutputBuffer
        implements OutputBuffer
{
    @GuardedBy("this")
    private OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(ARBITRARY);

    // all the pages should be put into this buffer at first, then move to `buffers`
    private final MasterBuffer masterBuffer;

    // the size is determined by the upstream task, here it's 3 corresponding to 3 worker nodes
    @GuardedBy("this")
    private final ConcurrentMap<OutputBufferId, ClientBuffer> buffers = new ConcurrentHashMap<>();

    //  The index of the first client buffer that should be polled
    private final AtomicInteger nextClientBufferIndex = new AtomicInteger(0);

    @Override
    public void enqueue(List<Slice> pages)
    {
        ...
        /* Test code start */
        long beforeCnt = masterBuffer.getBufferedPages();
        /* Test code end */

        // add pages to the buffer (this will increase the reference count by one)
        masterBuffer.addPages(serializedPageReferences);
        
        /* Test code start */
        System.out.printf("ThreadId %s: page size: %s, before master buffer size %s, after master buffer size %s,\n",
                Thread.currentThread().getName(), bytesAdded, beforeCnt, masterBuffer.getBufferedPages());
        /* Test code end */

        // process any pending reads from the client buffers
        List<ClientBuffer> buffers = safeGetBuffersSnapshot();
        if (buffers.isEmpty()) {
            return;
        }
        // handle potential for racy update of next index and client buffers present
        int index = nextClientBufferIndex.get() % buffers.size();
        for (int i = 0; i < buffers.size(); i++) {
            // buffers.get(index).loadPagesIfNecessary(masterBuffer);
            
            /* Test code start */
            ClientBuffer clientBuffer = buffers.get(index);
            clientBuffer.loadPagesIfNecessary(masterBuffer);
            BufferInfo bufferInfo = clientBuffer.getInfo();
            System.out.printf("ThreadId %s: %s now buffered total %s pages on bufferId %s with destroyed %s!\n",
                    Thread.currentThread().getName() + "-" + Thread.currentThread().getId(),
                    this.getClass().getName(),
                    bufferInfo.getBufferedPages(),
                    bufferInfo.getBufferId(),
                    clientBuffer.isDestroyed());
            /* Test code end */

            index = (index + 1) % buffers.size();

            if (masterBuffer.isEmpty()) {
                // Resume from the next client buffer on the next iteration
                nextClientBufferIndex.set(index);
                break;
            }
        }
    }
}

Below logs are from task 20220308_114754_00170_7cd8a.2.1.0 at scan stage, we could see that buffer-1 received the most pages, buffer-2 received fewer pages, while buffer-0 received 0. With no doubts, all the tasks consumed pages from buffer-0 will produce nothing, that is the exchange task reading buffer-0 sink nothing, and finally the write task reading sinked data partition 0 wrote nothing.

This tells the reason why the test could occasionally failed on mismatched files count. In this case, total 6 files were produced by two workers, rather than expected 9.

ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: page size: 5501, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: page size: 5522, before master buffer size 2, after master buffer size 3,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: page size: 2540, before master buffer size 3, after master buffer size 4,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-3-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: page size: 5501, before master buffer size 4, after master buffer size 5,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: page size: 5501, before master buffer size 1, after master buffer size 2,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: page size: 5522, before master buffer size 6, after master buffer size 7,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: page size: 5501, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: page size: 2540, before master buffer size 7, after master buffer size 8,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: page size: 5522, before master buffer size 5, after master buffer size 6,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-0-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 8 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: page size: 5522, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: page size: 2540, before master buffer size 1, after master buffer size 2,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-1-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 8 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 8 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: page size: 2540, before master buffer size 2, after master buffer size 3,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-2-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 8 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: page size: 5501, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 3 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: page size: 5522, before master buffer size 1, after master buffer size 2,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: page size: 5501, before master buffer size 2, after master buffer size 3,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 2 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: page size: 5522, before master buffer size 3, after master buffer size 4,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 2 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: page size: 2540, before master buffer size 4, after master buffer size 5,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-6-66: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: page size: 5501, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: page size: 5501, before master buffer size 1, after master buffer size 2,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: page size: 2540, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-4-63: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: page size: 5522, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: page size: 2540, before master buffer size 1, after master buffer size 2,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-7-64: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: page size: 5522, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 2 pages on bufferId 2 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: page size: 2540, before master buffer size 0, after master buffer size 1,
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 0 pages on bufferId 0 with destroyed false!
ThreadId 20220308_114754_00170_7cd8a.2.1.0-5-65: io.trino.execution.buffer.ArbitraryOutputBuffer now buffered total 1 pages on bufferId 1 with destroyed false!

Discussion

1. Imbalanced distribution of pages by ArbitraryOutputBuffer is acceptable?
2. Imbalanced distribution will decrease the overall performance?
3. It's better to use a more heuristic method to determine the output buffer id, such as the client buffer size, the upstream workers QPS or ... ?

I hope the above explanations would not make you puzzled. :)

[fengguangyuan]

@ebyhr @losipiuk

[findepi]

This tells the reason why the test could occasionally failed on mismatched files count. In this case, total 6 files were produced by two workers, rather than expected 9.

Correct. That's why the test uses assertEventually.
Apparently that's sometimes not enough.

[findepi]

https://github.com/trinodb/trino/runs/6651751956?check_suite_focus=true

[findepi]

https://github.com/trinodb/trino/runs/6899758149?check_suite_focus=true

[findepi]

https://github.com/trinodb/trino/runs/7089990501?check_suite_focus=true

Error:    TestIcebergParquetConnectorTest>BaseIcebergConnectorTest.testStatsBasedRepartitionDataOnInsert:3173->BaseIcebergConnectorTest.testStatsBasedRepartitionData:3195->BaseIcebergConnectorTest.lambda$testStatsBasedRepartitionData$15:3196->BaseIcebergConnectorTest.testRepartitionData:3239 [Rows for query [SELECT count(*) FROM "repartition_SELECTDISTINCTorderkeycustkeyorderstatusFROMtpchtinyordersinsert_orderstatus_ybf5539lvf$files"]] 
Expecting:
  <(6)>
to contain exactly in any order:
  <[(9)]>
elements not found:
  <(9)>
and elements not expected:
  <(6)>

[findepi]

https://github.com/trinodb/trino/actions/runs/3135264357/jobs/5091989947