Skip to content

Commit 0ad4af9

Browse files
Joe-AbrahamJoe-Abraham
committed
feat(functions) : Add e2e test for remote functions
1 parent 5534a76 commit 0ad4af9

File tree

8 files changed

+258
-39
lines changed

8 files changed

+258
-39
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,3 +66,4 @@ presto-native-execution/deps-install
6666
# Compiled executables used for docker build
6767
/docker/presto-cli-*-executable.jar
6868
/docker/presto-server-*.tar.gz
69+
/docker/presto-function-server-executable.jar

‎Jenkinsfile‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,7 @@ pipeline {
133133
returnStdout: true).trim()
134134
env.PRESTO_PKG = "presto-server-${PRESTO_VERSION}.tar.gz"
135135
env.PRESTO_CLI_JAR = "presto-cli-${PRESTO_VERSION}-executable.jar"
136+
env.PRESTO_REMOTE_SERVER_JAR = "presto-function-server-executable.jar"
136137
env.PRESTO_BUILD_VERSION = env.PRESTO_VERSION + '-' +
137138
sh(script: "git show -s --format=%cd --date=format:'%Y%m%d%H%M%S'", returnStdout: true).trim() + "-" +
138139
env.PRESTO_COMMIT_SHA.substring(0, 7)
@@ -162,6 +163,7 @@ pipeline {
162163
aws s3 cp index.txt ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/ --no-progress
163164
aws s3 cp presto-server/target/${PRESTO_PKG} ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/ --no-progress
164165
aws s3 cp presto-cli/target/${PRESTO_CLI_JAR} ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/ --no-progress
166+
aws s3 cp presto-function-server/target/${PRESTO_REMOTE_SERVER_JAR} ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/ --no-progress
165167
'''
166168
}
167169
}
@@ -205,6 +207,7 @@ pipeline {
205207
cd docker/
206208
aws s3 cp ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/${PRESTO_PKG} . --no-progress
207209
aws s3 cp ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/${PRESTO_CLI_JAR} . --no-progress
210+
aws s3 cp ${AWS_S3_PREFIX}/${PRESTO_BUILD_VERSION}/${PRESTO_REMOTE_SERVER_JAR} . --no-progress
208211
209212
echo "Building ${DOCKER_IMAGE}"
210213
REG_ORG=${AWS_ECR} IMAGE_NAME=${IMG_NAME} TAG=${PRESTO_BUILD_VERSION} ./build.sh ${PRESTO_VERSION}

‎docker/Dockerfile‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ FROM quay.io/centos/centos:stream9
33
ARG PRESTO_VERSION
44
ARG PRESTO_PKG=presto-server-$PRESTO_VERSION.tar.gz
55
ARG PRESTO_CLI_JAR=presto-cli-$PRESTO_VERSION-executable.jar
6+
ARG PRESTO_REMOTE_SERVER_JAR=presto-function-server-executable.jar
67
ARG JMX_PROMETHEUS_JAVAAGENT_VERSION=0.20.0
78

89
ENV PRESTO_HOME="/opt/presto-server"
@@ -14,6 +15,7 @@ RUN --mount=type=cache,target=/var/cache/dnf,sharing=locked \
1415
&& mv /etc/yum/protected.d/systemd.conf /etc/yum/protected.d/systemd.conf.bak
1516

1617
COPY --chmod=755 $PRESTO_CLI_JAR /opt/presto-cli
18+
COPY --chmod=755 $PRESTO_REMOTE_SERVER_JAR /opt/presto-function-server-executable.jar
1719

1820
RUN --mount=type=bind,source=$PRESTO_PKG,target=/$PRESTO_PKG \
1921
# Download Presto and move \

‎presto-native-execution/.dockerignore‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
# Ignore build directories
2+
_build/
3+
cmake-build-debug/
4+
cmake-build-release/

‎presto-native-execution/pom.xml‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -503,6 +503,12 @@
503503
<include>presto-cli-*-executable.jar</include>
504504
</includes>
505505
</resource>
506+
<resource>
507+
<directory>${project.parent.basedir}/presto-function-server/target</directory>
508+
<includes>
509+
<include>presto-function-server-executable.jar</include>
510+
</includes>
511+
</resource>
506512
<resource>
507513
<directory>${project.parent.basedir}/presto-server/target</directory>
508514
<includes>
@@ -574,7 +580,7 @@
574580
<args>
575581
<BUILD_TYPE>Release</BUILD_TYPE>
576582
<DEPENDENCY_IMAGE>presto-native-dependency:latest</DEPENDENCY_IMAGE>
577-
<EXTRA_CMAKE_FLAGS>-DPRESTO_ENABLE_TESTING=OFF</EXTRA_CMAKE_FLAGS>
583+
<EXTRA_CMAKE_FLAGS>"-DPRESTO_ENABLE_TESTING=OFF -DPRESTO_ENABLE_REMOTE_FUNCTIONS=ON"</EXTRA_CMAKE_FLAGS>
578584
<NUM_THREADS>2</NUM_THREADS>
579585
<BASE_IMAGE>ubuntu:22.04</BASE_IMAGE>
580586
<OSNAME>ubuntu</OSNAME>

‎presto-native-execution/src/test/java/com/facebook/presto/nativeworker/ContainerQueryRunner.java‎

Lines changed: 100 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import com.facebook.presto.sql.planner.NodePartitioningManager;
2929
import com.facebook.presto.sql.planner.sanity.PlanCheckerProviderManager;
3030
import com.facebook.presto.testing.MaterializedResult;
31+
import com.facebook.presto.testing.MaterializedRow;
3132
import com.facebook.presto.testing.QueryRunner;
3233
import com.facebook.presto.testing.TestingAccessControlManager;
3334
import com.facebook.presto.transaction.TransactionManager;
@@ -38,6 +39,7 @@
3839

3940
import java.io.IOException;
4041
import java.sql.Connection;
42+
import java.sql.DriverManager;
4143
import java.sql.ResultSet;
4244
import java.sql.SQLException;
4345
import java.sql.Statement;
@@ -50,46 +52,57 @@
5052
import java.util.logging.Logger;
5153

5254
import static com.facebook.presto.testing.TestingSession.testSessionBuilder;
53-
import static java.sql.DriverManager.getConnection;
5455

5556
public class ContainerQueryRunner
5657
implements QueryRunner
5758
{
58-
private static final Network network = Network.newNetwork();
59-
private static final String PRESTO_COORDINATOR_IMAGE = System.getProperty("coordinatorImage", "presto-coordinator:latest");
60-
private static final String PRESTO_WORKER_IMAGE = System.getProperty("workerImage", "presto-worker:latest");
61-
private static final String CONTAINER_TIMEOUT = System.getProperty("containerTimeout", "120");
62-
private static final String CLUSTER_SHUTDOWN_TIMEOUT = System.getProperty("clusterShutDownTimeout", "10");
63-
private static final String BASE_DIR = System.getProperty("user.dir");
64-
private static final int DEFAULT_COORDINATOR_PORT = 8080;
65-
private static final String TPCH_CATALOG = "tpch";
66-
private static final String TINY_SCHEMA = "tiny";
67-
private static final int DEFAULT_NUMBER_OF_WORKERS = 4;
68-
private static final Logger logger = Logger.getLogger(ContainerQueryRunner.class.getName());
69-
private final GenericContainer<?> coordinator;
70-
private final List<GenericContainer<?>> workers = new ArrayList<>();
71-
private final int coordinatorPort;
72-
private final String catalog;
73-
private final String schema;
74-
private final int numberOfWorkers;
75-
private Connection connection;
59+
protected static final Network network = Network.newNetwork();
60+
protected static final String PRESTO_COORDINATOR_IMAGE = System.getProperty("coordinatorImage", "presto-coordinator:latest");
61+
protected static final String PRESTO_WORKER_IMAGE = System.getProperty("workerImage", "presto-worker:latest");
62+
protected static final String CONTAINER_TIMEOUT = System.getProperty("containerTimeout", "120");
63+
protected static final String CLUSTER_SHUTDOWN_TIMEOUT = System.getProperty("clusterShutDownTimeout", "10");
64+
protected static final String BASE_DIR = System.getProperty("user.dir");
65+
protected static final int DEFAULT_COORDINATOR_PORT = 8080;
66+
protected static final int DEFAULT_FUNCTION_SERVER_PORT = 1122;
67+
protected static final String TPCH_CATALOG = "tpch";
68+
protected static final String TINY_SCHEMA = "tiny";
69+
protected static final int DEFAULT_NUMBER_OF_WORKERS = 4;
70+
71+
protected static final Logger logger = Logger.getLogger(ContainerQueryRunner.class.getName());
72+
73+
protected final GenericContainer<?> coordinator;
74+
protected final List<GenericContainer<?>> workers = new ArrayList<>();
75+
protected final int coordinatorPort;
76+
protected final String catalog;
77+
protected final String schema;
78+
protected GenericContainer<?> functionServer;
79+
protected int functionServerPort;
80+
protected boolean enableFunctionServer;
81+
protected Connection connection;
7682

7783
public ContainerQueryRunner()
7884
throws InterruptedException, IOException
7985
{
80-
this(DEFAULT_COORDINATOR_PORT, TPCH_CATALOG, TINY_SCHEMA, DEFAULT_NUMBER_OF_WORKERS);
86+
this(DEFAULT_COORDINATOR_PORT, TPCH_CATALOG, TINY_SCHEMA, DEFAULT_NUMBER_OF_WORKERS, DEFAULT_FUNCTION_SERVER_PORT, false);
8187
}
8288

83-
public ContainerQueryRunner(int coordinatorPort, String catalog, String schema, int numberOfWorkers)
89+
public ContainerQueryRunner(int coordinatorPort, String catalog, String schema, int numberOfWorkers, int functionServerPort, boolean enableFunctionServer)
8490
throws InterruptedException, IOException
8591
{
8692
this.coordinatorPort = coordinatorPort;
8793
this.catalog = catalog;
8894
this.schema = schema;
89-
this.numberOfWorkers = numberOfWorkers;
95+
this.functionServerPort = functionServerPort;
96+
this.enableFunctionServer = enableFunctionServer;
97+
98+
// Start function server first if enabled
99+
if (enableFunctionServer) {
100+
this.functionServer = createFunctionServer();
101+
this.functionServer.start();
102+
logger.info("Presto function server is deployed at http://" + functionServer.getHost() + ":" + functionServer.getMappedPort(functionServerPort));
103+
}
90104

91-
// The container details can be added as properties in VM options for testing in IntelliJ.
92-
coordinator = createCoordinator();
105+
this.coordinator = createCoordinator();
93106
for (int i = 0; i < numberOfWorkers; i++) {
94107
workers.add(createNativeWorker(7777 + i, "native-worker-" + i));
95108
}
@@ -107,23 +120,32 @@ public ContainerQueryRunner(int coordinatorPort, String catalog, String schema,
107120
coordinator.getMappedPort(coordinatorPort),
108121
catalog,
109122
schema,
110-
"timeZoneId=UTC");
123+
enableFunctionServer ? "timeZoneId=UTC&sessionProperties=remote_functions_enabled:true" : "timeZoneId=UTC");
124+
125+
postStartContainers(url);
126+
}
111127

128+
protected void postStartContainers(String url)
129+
{
112130
try {
113-
connection = getConnection(url, "test", null);
131+
this.connection = DriverManager.getConnection(url, "test", null);
114132
}
115133
catch (SQLException e) {
116134
throw new RuntimeException(e);
117135
}
118136

119137
// Delete the temporary files once the containers are started.
120138
ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/coordinator");
121-
for (int i = 0; i < numberOfWorkers; i++) {
122-
ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/native-worker-" + i);
139+
for (GenericContainer<?> worker : workers) {
140+
String alias = worker.getNetworkAliases().get(1);
141+
ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/" + alias);
142+
}
143+
if (enableFunctionServer) {
144+
ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/function-server");
123145
}
124146
}
125147

126-
private GenericContainer<?> createCoordinator()
148+
protected GenericContainer<?> createCoordinator()
127149
throws IOException
128150
{
129151
ContainerQueryRunnerUtils.createCoordinatorTpchProperties();
@@ -132,22 +154,25 @@ private GenericContainer<?> createCoordinator()
132154
ContainerQueryRunnerUtils.createCoordinatorJvmConfig();
133155
ContainerQueryRunnerUtils.createCoordinatorLogProperties();
134156
ContainerQueryRunnerUtils.createCoordinatorNodeProperties();
135-
ContainerQueryRunnerUtils.createCoordinatorEntryPointScript();
157+
ContainerQueryRunnerUtils.createCoordinatorEntryPointScript(); // Never run function server in coordinator
158+
if (enableFunctionServer) {
159+
ContainerQueryRunnerUtils.createRestRemoteProperties(functionServerPort);
160+
}
136161

137162
return new GenericContainer<>(PRESTO_COORDINATOR_IMAGE)
138-
.withExposedPorts(coordinatorPort)
139163
.withNetwork(network)
140164
.withNetworkAliases("presto-coordinator")
141165
.withCopyFileToContainer(MountableFile.forHostPath(BASE_DIR + "/testcontainers/coordinator/etc"), "/opt/presto-server/etc")
142166
.withCopyFileToContainer(MountableFile.forHostPath(BASE_DIR + "/testcontainers/coordinator/entrypoint.sh"), "/opt/entrypoint.sh")
143167
.waitingFor(Wait.forLogMessage(".*======== SERVER STARTED ========.*", 1))
144-
.withStartupTimeout(Duration.ofSeconds(Long.parseLong(CONTAINER_TIMEOUT)));
168+
.withStartupTimeout(Duration.ofSeconds(Long.parseLong(CONTAINER_TIMEOUT)))
169+
.withExposedPorts(coordinatorPort);
145170
}
146171

147-
private GenericContainer<?> createNativeWorker(int port, String nodeId)
172+
protected GenericContainer<?> createNativeWorker(int port, String nodeId)
148173
throws IOException
149174
{
150-
ContainerQueryRunnerUtils.createNativeWorkerConfigProperties(coordinatorPort, nodeId);
175+
ContainerQueryRunnerUtils.createNativeWorkerConfigPropertiesWithFunctionServer(coordinatorPort, functionServerPort, nodeId);
151176
ContainerQueryRunnerUtils.createNativeWorkerTpchProperties(nodeId);
152177
ContainerQueryRunnerUtils.createNativeWorkerEntryPointScript(nodeId);
153178
ContainerQueryRunnerUtils.createNativeWorkerNodeProperties(nodeId);
@@ -160,6 +185,23 @@ private GenericContainer<?> createNativeWorker(int port, String nodeId)
160185
.waitingFor(Wait.forLogMessage(".*Announcement succeeded: HTTP 202.*", 1));
161186
}
162187

188+
protected GenericContainer<?> createFunctionServer()
189+
throws IOException
190+
{
191+
ContainerQueryRunnerUtils.createFunctionServerConfigProperties(functionServerPort);
192+
ContainerQueryRunnerUtils.createFunctionServerEntryPointScript();
193+
194+
// Reuse the coordinator image since it already contains the function server jar
195+
return new GenericContainer<>(PRESTO_COORDINATOR_IMAGE)
196+
.withNetwork(network)
197+
.withNetworkAliases("presto-function-server")
198+
.withCopyFileToContainer(MountableFile.forHostPath(BASE_DIR + "/testcontainers/function-server/etc"), "/opt/function-server/etc")
199+
.withCopyFileToContainer(MountableFile.forHostPath(BASE_DIR + "/testcontainers/function-server/entrypoint.sh"), "/opt/entrypoint.sh")
200+
.waitingFor(Wait.forLogMessage(".* STARTED .*", 1))
201+
.withStartupTimeout(Duration.ofSeconds(Long.parseLong(CONTAINER_TIMEOUT)))
202+
.withExposedPorts(functionServerPort);
203+
}
204+
163205
@Override
164206
public void close()
165207
{
@@ -171,6 +213,9 @@ public void close()
171213
}
172214
coordinator.stop();
173215
workers.forEach(GenericContainer::stop);
216+
if (functionServer != null) {
217+
functionServer.stop();
218+
}
174219
}
175220

176221
@Override
@@ -248,7 +293,27 @@ public MaterializedResult execute(String sql)
248293
@Override
249294
public MaterializedResult execute(Session session, String sql, List<? extends Type> resultTypes)
250295
{
251-
throw new UnsupportedOperationException();
296+
// Added logic similar to H2QueryRunner.
297+
try {
298+
Statement statement = connection.createStatement();
299+
ResultSet resultSet = statement.executeQuery(sql);
300+
MaterializedResult rawResult = ContainerQueryRunnerUtils.toMaterializedResult(resultSet);
301+
302+
// Coerce the raw result to the requested resultTypes
303+
List<MaterializedRow> coercedRows = new ArrayList<>();
304+
for (MaterializedRow row : rawResult.getMaterializedRows()) {
305+
List<Object> coercedValues = new ArrayList<>();
306+
for (int i = 0; i < resultTypes.size(); i++) {
307+
Object value = row.getField(i);
308+
coercedValues.add(value);
309+
}
310+
coercedRows.add(new MaterializedRow(MaterializedResult.DEFAULT_PRECISION, coercedValues));
311+
}
312+
return new MaterializedResult(coercedRows, resultTypes);
313+
}
314+
catch (SQLException e) {
315+
throw new RuntimeException("Error executing query: " + sql, e);
316+
}
252317
}
253318

254319
@Override

0 commit comments

Comments
 (0)