PHILO-HE commented on code in PR #9893: URL: https://github.com/apache/incubator-gluten/pull/9893#discussion_r2151258723
########## gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkQ0Test.java: ########## @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.table.runtime.stream.custom; + +import org.apache.gluten.table.runtime.stream.common.Velox4jEnvironment; + +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Paths; + +import static org.assertj.core.api.Assertions.assertThat; + +public class NexmarkQ0Test { + + private static final Logger LOG = LoggerFactory.getLogger(NexmarkQ0Test.class); + + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + @BeforeAll + public static void setupAll() { + LOG.info("NexmarkTest setup"); + Velox4jEnvironment.initializeOnce(); + } + + @BeforeEach + public void setup() { + env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + + EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build(); + tEnv = StreamTableEnvironment.create(env, settings); + } + + @Test + void testNexmarkQ0Query() { Review Comment: It seems this method can be applied for other queries. If true, rename it and this class for future extension. Then, the below table initialization can be also shared for other queries. ########## gluten-flink/ut/pom.xml: ########## @@ -90,6 +90,11 @@ <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> + <dependency> + <groupId>com.github.nexmark</groupId> + <artifactId>nexmark-flink</artifactId> + <version>0.3-SNAPSHOT</version> Review Comment: Better to add test scope for clarity. Please also add it for other dependencies that only required for test. ########## gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkQ0Test.java: ########## @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.table.runtime.stream.custom; + +import org.apache.gluten.table.runtime.stream.common.Velox4jEnvironment; + +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Paths; + +import static org.assertj.core.api.Assertions.assertThat; + +public class NexmarkQ0Test { + + private static final Logger LOG = LoggerFactory.getLogger(NexmarkQ0Test.class); + + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + @BeforeAll + public static void setupAll() { + LOG.info("NexmarkTest setup"); + Velox4jEnvironment.initializeOnce(); + } + + @BeforeEach + public void setup() { + env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + + EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build(); + tEnv = StreamTableEnvironment.create(env, settings); + } + + @Test + void testNexmarkQ0Query() { + String createNexmarkSource = readSqlFromFile("nexmark/ddl_gen.sql"); + tEnv.executeSql(createNexmarkSource); + String createTableView = readSqlFromFile("nexmark/ddl_views.sql"); + String[] sqlTableView = createTableView.split(";"); + for (String sql : sqlTableView) { + String trimmedSql = sql.trim(); + tEnv.executeSql(trimmedSql); + } + + String q0Query = readSqlFromFile("nexmark/q0.sql"); + + String[] sqlStatements = q0Query.split(";"); + String createResultTable = sqlStatements[0].trim(); + TableResult createResult = tEnv.executeSql(createResultTable); + assertThat(createResult.getJobClient().isPresent()).isFalse(); + String insertQuery = sqlStatements[1].trim(); + TableResult insertResult = tEnv.executeSql(insertQuery); + assertThat(insertResult.getJobClient().isPresent()).isTrue(); + try { + Thread.sleep(100000); + } catch (InterruptedException e) { Review Comment: Seems not necessary to handle this exception since it rarely happens. Maybe, just let the test method throw this exception and fail the test if it happens. ########## gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkQ0Test.java: ########## @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.table.runtime.stream.custom; + +import org.apache.gluten.table.runtime.stream.common.Velox4jEnvironment; + +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Paths; + +import static org.assertj.core.api.Assertions.assertThat; + +public class NexmarkQ0Test { + + private static final Logger LOG = LoggerFactory.getLogger(NexmarkQ0Test.class); + + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + @BeforeAll + public static void setupAll() { + LOG.info("NexmarkTest setup"); + Velox4jEnvironment.initializeOnce(); + } + + @BeforeEach + public void setup() { + env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + + EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build(); + tEnv = StreamTableEnvironment.create(env, settings); + } + + @Test + void testNexmarkQ0Query() { + String createNexmarkSource = readSqlFromFile("nexmark/ddl_gen.sql"); + tEnv.executeSql(createNexmarkSource); + String createTableView = readSqlFromFile("nexmark/ddl_views.sql"); + String[] sqlTableView = createTableView.split(";"); + for (String sql : sqlTableView) { + String trimmedSql = sql.trim(); + tEnv.executeSql(trimmedSql); + } + + String q0Query = readSqlFromFile("nexmark/q0.sql"); Review Comment: Maybe, better to obtain all files (name pattern q*.sql) from the directory and handle it in loop, though currently there is only one. ########## gluten-flink/ut/src/test/resources/nexmark/ddl_views.sql: ########## @@ -0,0 +1,36 @@ +CREATE VIEW person AS +SELECT + person.id, + person.name, + person.emailAddress, + person.creditCard, + person.city, + person.state, + `dateTime`, + person.extra +FROM datagen WHERE event_type = 0; + +CREATE VIEW auction AS +SELECT + auction.id, + auction.itemName, + auction.description, + auction.initialBid, + auction.reserve, + `dateTime`, + auction.expires, + auction.seller, + auction.category, + auction.extra +FROM datagen WHERE event_type = 1; + +CREATE VIEW bid AS +SELECT + bid.auction, + bid.bidder, + bid.price, + bid.channel, + bid.url, + `dateTime`, + bid.extra +FROM datagen WHERE event_type = 2; Review Comment: Nit: leave one empty line at the end. ########## gluten-flink/ut/pom.xml: ########## @@ -90,6 +90,11 @@ <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> + <dependency> + <groupId>com.github.nexmark</groupId> + <artifactId>nexmark-flink</artifactId> + <version>0.3-SNAPSHOT</version> Review Comment: Please clarify which code needs this dependency. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
