Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix spooling output buffer unit test #15905

Merged
merged 1 commit into from
Apr 8, 2021
Merged

Fix spooling output buffer unit test #15905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...-main/src/main/java/com/facebook/presto/execution/buffer/SpoolingOutputBufferFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ public class SpoolingOutputBufferFactory

private final Closer closer = Closer.create();

private static ExecutorService coreExecutor = newCachedThreadPool(daemonThreadsNamed("spooling-outputbuffer-%s"));
private final ExecutorService coreExecutor = newCachedThreadPool(daemonThreadsNamed("spooling-outputbuffer-%s"));

@Inject
public SpoolingOutputBufferFactory(FeaturesConfig featuresConfig, TempStorageManager tempStorageManager, FinalizerService finalizerService)
Expand Down
116 changes: 51 additions & 65 deletions presto-main/src/test/java/com/facebook/presto/execution/buffer/TestSpoolingOutputBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
Expand Down Expand Up @@ -61,60 +62,65 @@ public class TestSpoolingOutputBuffer
{
private static final String TASK_INSTANCE_ID = "task-instance-id";

private static final DataSize THRESHOLD = sizeOfPages(3);
private static final List<BigintType> TYPES = ImmutableList.of(BIGINT);
private static final OutputBufferId BUFFER_ID = new OutputBufferId(0);
private static final OutputBufferId INVALID_BUFFER_ID = new OutputBufferId(1);
private static final OutputBuffers OUTPUT_BUFFERS = OutputBuffers.createSpoolingOutputBuffers();

private static final QueryIdGenerator queryIdGenerator = new QueryIdGenerator();

private static SpoolingOutputBufferFactory spoolingOutputBufferFactory;

private ScheduledExecutorService stateNotificationExecutor;

@BeforeClass
public void setUp()
{
stateNotificationExecutor = newScheduledThreadPool(5, daemonThreadsNamed("test-%s"));

tsaichien marked this conversation as resolved.
Show resolved Hide resolved
FeaturesConfig featuresConfig = new FeaturesConfig();
featuresConfig.setSpoolingOutputBufferThreshold(THRESHOLD);
spoolingOutputBufferFactory = new SpoolingOutputBufferFactory(featuresConfig);
}

@AfterClass(alwaysRun = true)
public void tearDown()
throws IOException
{
if (stateNotificationExecutor != null) {
stateNotificationExecutor.shutdownNow();
stateNotificationExecutor = null;
}
spoolingOutputBufferFactory.shutdown();
}

@Test(enabled = false)
@Test
public void testSimpleInMemory()
{
DataSize threshold = sizeOfPages(5);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

// add three pages
for (int i = 0; i < 3; i++) {
for (int i = 0; i < 2; i++) {
addPage(buffer, createPage(i));
}
compareTotalBuffered(buffer, 3);
compareTotalBuffered(buffer, 2);

assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 0, sizeOfPages(1), MAX_WAIT), bufferResult(0, createPage(0)));
compareTotalBuffered(buffer, 3);

assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 1, sizeOfPages(1), MAX_WAIT), bufferResult(1, createPage(1)));
compareTotalBuffered(buffer, 2);

assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 2, sizeOfPages(1), MAX_WAIT), bufferResult(2, createPage(2)));
assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 1, sizeOfPages(1), MAX_WAIT), bufferResult(1, createPage(1)));
compareTotalBuffered(buffer, 1);

buffer.setNoMorePages();
assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 3, sizeOfPages(1), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 3, true));
assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 2, sizeOfPages(1), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 2, true));
compareTotalBuffered(buffer, 0);
}

@Test(enabled = false)
@Test
public void testSimple()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

List<Page> pages = new LinkedList<>();
// add five pages to storage
Expand Down Expand Up @@ -171,11 +177,10 @@ public void testSimple()
assertEquals(buffer.getInfo().getTotalPagesSent(), 9);
}

@Test(enabled = false)
@Test
void testUnevenMaxSize()
{
DataSize threshold = sizeOfPages(5);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

List<Page> pages = new LinkedList<>();
// add five pages to storage
Expand All @@ -196,11 +201,10 @@ void testUnevenMaxSize()
compareTotalBuffered(buffer, 8);
}

@Test(enabled = false)
@Test
void testGetOutOfOrder()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

List<Page> pages = new LinkedList<>();
// add five pages to storage
Expand All @@ -219,11 +223,10 @@ void testGetOutOfOrder()
compareTotalBuffered(buffer, 5);
}

@Test(enabled = false)
@Test
public void testSimplePendingRead()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

// attempt to get a page
ListenableFuture<BufferResult> future = buffer.get(BUFFER_ID, 0, sizeOfPages(2));
Expand Down Expand Up @@ -256,11 +259,10 @@ public void testSimplePendingRead()
assertFalse(future.isDone());
}

@Test(enabled = false)
@Test
public void testMultiplePendingReads()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

// attempt to get a page
ListenableFuture<BufferResult> oldPendingRead = buffer.get(BUFFER_ID, 0, sizeOfPages(3));
Expand All @@ -281,11 +283,10 @@ public void testMultiplePendingReads()
assertBufferResultEquals(TYPES, getFuture(newPendingRead, MAX_WAIT), createBufferResult(TASK_INSTANCE_ID, 0, pages));
}

@Test(enabled = false)
@Test
public void testAddAfterPendingRead()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

ListenableFuture<BufferResult> pendingRead = buffer.get(BUFFER_ID, 0, sizeOfPages(5));
assertFalse(pendingRead.isDone());
Expand All @@ -309,11 +310,10 @@ public void testAddAfterPendingRead()
assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 5, sizeOfPages(1), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 5, true));
}

@Test(enabled = false)
@Test
public void testNoMorePagesAfterPendingRead()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

ListenableFuture<BufferResult> pendingRead = buffer.get(BUFFER_ID, 0, sizeOfPages(5));
assertFalse(pendingRead.isDone());
Expand All @@ -323,11 +323,10 @@ public void testNoMorePagesAfterPendingRead()
assertBufferResultEquals(TYPES, getFuture(pendingRead, NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, true));
}

@Test(enabled = false)
@Test
public void testDestroyAfterPendingRead()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

ListenableFuture<BufferResult> pendingRead = buffer.get(BUFFER_ID, 0, sizeOfPages(5));
assertFalse(pendingRead.isDone());
Expand All @@ -338,11 +337,10 @@ public void testDestroyAfterPendingRead()
assertBufferResultEquals(TYPES, getFuture(pendingRead, NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, false));
}

@Test(enabled = false)
@Test
public void testAcknowledgeSimple()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

for (int i = 0; i < 3; i++) {
addPage(buffer, createPage(i));
Expand Down Expand Up @@ -378,11 +376,10 @@ public void testAcknowledgeSimple()
compareTotalBuffered(buffer, 3);
}

@Test(enabled = false)
@Test
public void testAcknowledgeStorageAndMemory()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

// add three pages to file
for (int i = 0; i < 3; i++) {
Expand All @@ -399,11 +396,10 @@ public void testAcknowledgeStorageAndMemory()
compareTotalBuffered(buffer, 1);
}

@Test(enabled = false)
@Test
public void testDuplicateGet()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

// add three pages
for (int i = 0; i < 3; i++) {
Expand All @@ -427,11 +423,10 @@ public void testDuplicateGet()
compareTotalBuffered(buffer, 0);
}

@Test(enabled = false)
@Test
public void testAddAfterNoMorePages()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

for (int i = 0; i < 2; i++) {
addPage(buffer, createPage(i));
Expand All @@ -450,11 +445,10 @@ public void testAddAfterNoMorePages()
assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 2, sizeOfPages(3), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 2, true));
}

@Test(enabled = false)
@Test
public void testAddAfterDestroy()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

for (int i = 0; i < 2; i++) {
addPage(buffer, createPage(i));
Expand All @@ -473,11 +467,10 @@ public void testAddAfterDestroy()
assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 0, sizeOfPages(3), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, true));
}

@Test(enabled = false)
@Test
public void testAbort()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

// add three pages into a file
for (int i = 0; i < 3; i++) {
Expand All @@ -501,11 +494,10 @@ public void testAbort()
compareTotalBuffered(buffer, 0);
}

@Test(enabled = false)
@Test
public void testSetOutputBuffers()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

OutputBuffers newBuffers = new OutputBuffers(SPOOLING, 1, true, ImmutableMap.of());
buffer.setOutputBuffers(newBuffers);
Expand All @@ -519,11 +511,10 @@ public void testSetOutputBuffers()
}
}

@Test(enabled = false)
@Test
public void testBufferCompletion()
{
DataSize threshold = sizeOfPages(3);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer(threshold);
SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

assertFalse(buffer.isFinished());

Expand Down Expand Up @@ -558,18 +549,13 @@ public void testBufferCompletion()
assertTrue(buffer.isFinished());
}

private SpoolingOutputBuffer createSpoolingOutputBuffer(DataSize threshold)
private SpoolingOutputBuffer createSpoolingOutputBuffer()
{
FeaturesConfig featuresConfig = new FeaturesConfig();
featuresConfig.setSpoolingOutputBufferThreshold(threshold);

SpoolingOutputBufferFactory spoolingOutputBufferFactory = new SpoolingOutputBufferFactory(featuresConfig);
OutputBuffers outputBuffers = OutputBuffers.createSpoolingOutputBuffers();
TaskId taskId = new TaskId(queryIdGenerator.createNextQueryId().toString(), 0, 0, 0);
return spoolingOutputBufferFactory.createSpoolingOutputBuffer(
taskId,
TASK_INSTANCE_ID,
outputBuffers,
OUTPUT_BUFFERS,
new StateMachine<>("bufferState", stateNotificationExecutor, OPEN, TERMINAL_BUFFER_STATES));
}

Expand Down