Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3654222271
##########
gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java:
##########
@@ -147,32 +154,40 @@ void testKafkaSourceSqlPushesDownWatermark() {
@Test
void testAllNexmarkSourceQueries()
throws ExecutionException, InterruptedException, TimeoutException {
- setupNexmarkEnvironment(tEnv, "ddl_gen.sql", NEXMARK_VARIABLES);
- List<String> queryFiles = getQueries();
- assertThat(queryFiles).isNotEmpty();
- LOG.warn("Found {} Nexmark query files: {}", queryFiles.size(),
queryFiles);
-
- for (String queryFile : queryFiles) {
- LOG.warn("Executing nextmark query from file: {}", queryFile);
- executeQuery(tEnv, queryFile, false);
+ try {
+ setupNexmarkEnvironment(tEnv, "ddl_gen.sql", NEXMARK_VARIABLES);
+ List<String> queryFiles = getQueries();
+ assertThat(queryFiles).isNotEmpty();
+ LOG.warn("Found {} Nexmark query files: {}", queryFiles.size(),
queryFiles);
+
+ for (String queryFile : queryFiles) {
+ LOG.warn("Executing nextmark query from file: {}", queryFile);
+ executeQuery(tEnv, queryFile, false);
+ }
+ } finally {
+ clearEnvironment(tEnv);
}
- clearEnvironment(tEnv);
}
@Test
void testAllKafkaSourceQueries()
throws ExecutionException, InterruptedException, TimeoutException {
- kafkaInstance.getKafkaTestUtils().createTopic(topicName, 1, (short) 1);
- setupNexmarkEnvironment(tEnv, "ddl_kafka.sql", KAFKA_VARIABLES);
- List<String> queryFiles = getQueries();
- assertThat(queryFiles).isNotEmpty();
- LOG.warn("Found {} Nexmark query files: {}", queryFiles.size(),
queryFiles);
-
- for (String queryFile : queryFiles) {
- LOG.warn("Executing kafka query from file:{}", queryFile);
- executeQuery(tEnv, queryFile, true);
+ try {
+ kafkaInstance.getKafkaTestUtils().createTopic(topicName, 1, (short) 1);
+ setupNexmarkEnvironment(tEnv, "ddl_kafka.sql", KAFKA_VARIABLES);
+ List<String> queryFiles = getQueries();
+ assertThat(queryFiles).isNotEmpty();
+ LOG.warn("Found {} Nexmark query files: {}", queryFiles.size(),
queryFiles);
+
+ for (String queryFile : queryFiles) {
+ LOG.warn("Executing kafka query from file:{}", queryFile);
+ if (!"q10_orc.sql".equals(queryFile)) {
+ executeQuery(tEnv, queryFile, true);
+ }
+ }
Review Comment:
Skipping q10_orc.sql inside the loop means this test can end up executing
zero queries (e.g., when -Dnexmark.queries=q10_orc.sql is used). Filter it out
up-front and keep an assertion that at least one query will run.
This issue also appears on line 246 of the same file.
##########
gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java:
##########
@@ -242,11 +268,142 @@ private void executeQuery(StreamTableEnvironment tEnv,
String queryFileName, boo
assertThat(checkJobRunningStatus(insertResult, 30000) == true);
} else {
waitForJobCompletion(insertResult, 30000);
+ if ("q10_orc.sql".equals(queryFileName)) {
+ verifyQ10OrcOutput();
+ }
}
}
assertTrue(sqlStatements[sqlStatements.length - 1].trim().isEmpty());
}
+ private void executeQ10OrcBatchQuery()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ cleanQ10OrcOutput();
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+
+ EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
+ StreamTableEnvironment batchTEnv = StreamTableEnvironment.create(env,
settings);
+ try {
+ createQ10OrcBidView(batchTEnv);
+ String queryContent = readSqlFromFile(NEXMARK_RESOURCE_DIR +
"/q10_orc.sql");
+ String[] sqlStatements = queryContent.split(";");
+ assertThat(sqlStatements.length).isEqualTo(3);
+
+ TableResult createResult = batchTEnv.executeSql(sqlStatements[0].trim());
+ assertFalse(createResult.getJobClient().isPresent());
+
+ TableResult insertResult = batchTEnv.executeSql(sqlStatements[1].trim());
+ waitForJobCompletion(insertResult, 30000);
+ verifyQ10OrcOutput();
+ assertTrue(sqlStatements[2].trim().isEmpty());
+ } finally {
+ clearEnvironment(batchTEnv);
+ }
+ }
+
+ private static void createQ10OrcBidView(StreamTableEnvironment tEnv) {
+ tEnv.executeSql(
+ "CREATE TEMPORARY VIEW bid AS "
+ + "SELECT "
+ + "CAST(1 AS BIGINT) AS auction, "
+ + "CAST(2 AS BIGINT) AS bidder, "
+ + "CAST(100 AS BIGINT) AS price, "
+ + "CAST('channel' AS STRING) AS channel, "
+ + "CAST('url' AS STRING) AS url, "
+ + "TIMESTAMP '2026-07-21 07:30:00' AS `dateTime`, "
+ + "CAST('extra' AS STRING) AS extra");
+ }
+
+ private void cleanQ10OrcOutput() {
+ Path outputDir = Paths.get("/tmp/data/output/bid_orc");
+ if (!Files.exists(outputDir)) {
+ return;
+ }
+ try (java.util.stream.Stream<Path> files = Files.walk(outputDir)) {
+ files
+ .sorted(Comparator.reverseOrder())
+ .forEach(
+ path -> {
+ try {
+ Files.deleteIfExists(path);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to delete " + path, e);
+ }
+ });
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to clean Q10 ORC output directory",
e);
+ }
+ }
+
+ private void verifyQ10OrcOutput() throws InterruptedException {
+ Path outputDir = Paths.get("/tmp/data/output/bid_orc");
+ assertTrue("Q10 ORC output directory should exist",
Files.exists(outputDir));
+
+ List<Path> partFiles = waitForFinalQ10OrcPartFiles(outputDir);
+ long rowCount = 0L;
+ for (Path partFile : partFiles) {
+ try {
+ rowCount += readAndVerifyQ10OrcFile(partFile);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read Q10 ORC output file " +
partFile, e);
+ }
+ }
+ assertThat(rowCount).isGreaterThan(0L);
Review Comment:
verifyQ10OrcOutput() only asserts rowCount > 0, but the test input view
produces exactly one row (SELECT of constants with no FROM). Using an exact
row-count assertion will make this test stricter and better at catching
partial/duplicate writes.
--
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]
