From 94449f48cc9935f887c2179b901a5da5e2c90f24 Mon Sep 17 00:00:00 2001 From: liugddx Date: Tue, 21 Nov 2023 09:26:17 +0800 Subject: [PATCH 1/4] 1 --- .../routine_load/test_routine_load.groovy | 67 +++++++++++++++---- 1 file changed, 54 insertions(+), 13 deletions(-) diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy index 2829dfc9505f3f..c0b62b0f321e27 100644 --- a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy +++ b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy @@ -106,7 +106,7 @@ suite("test_routine_load","p0") { '[\"$.k00\", \"$.k01\", \"$.k02\", \"$.k03\", \"$.k04\", \"$.k05\", \"$.k06\", \"$.k07\", \"$.k08\", \"$.k09\", \"$.k10\", \"$.k11\", \"$.k12\", \"$.k13\", \"$.k14\", \"$.k15\", \"$.k16\", \"$.k17\"]', ] - def columns = [ + def columns = [ "k00,k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17,k18", "k00,k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17,k18", "k00,k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17,k18", @@ -116,8 +116,8 @@ suite("test_routine_load","p0") { "k00,k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17", ] - def timezoneColumns = - [ + def timezoneColumns = + [ "k00=unix_timestamp('2007-11-30 10:30:19'),k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17,k18", "k00=unix_timestamp('2007-11-30 10:30:19'),k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17,k18", "k00=unix_timestamp('2007-11-30 10:30:19'),k01,k02,k03,k04,k05,k06,k07,k08,k09,k10,k11,k12,k13,k14,k15,k16,k17,k18", @@ -140,7 +140,7 @@ suite("test_routine_load","p0") { def formats = [ "csv", "json", - ] + ] def loadedRows = [0,0,0,0,17,17,17] @@ -216,7 +216,7 @@ suite("test_routine_load","p0") { sleep(5000) count++ } - + if (i <= 3) { qt_sql_exec_mem_limit "select * from ${tableName1} order by k00,k01" } else { @@ -233,6 +233,47 @@ suite("test_routine_load","p0") { } } + i = 0 + if (enabled != null && enabled.equalsIgnoreCase("true")) { + try { + for (String tableName in tables) { + sql new File("""${context.file.parent}/ddl/${tableName}_drop.sql""").text + sql new File("""${context.file.parent}/ddl/${tableName}_create.sql""").text + + def name = "routine_load_" + tableName + try { + sql """ + CREATE ROUTINE LOAD ${jobs[i]} ON ${name} + COLUMNS(${columns[i]}), + COLUMNS TERMINATED BY "|" + PROPERTIES + ( + "exec_mem_limit" = "test", + "max_batch_interval" = "5", + "max_batch_rows" = "300000", + "max_batch_size" = "209715200" + ) + FROM KAFKA + ( + "kafka_broker_list" = "${externalEnvIp}:${kafka_port}", + "kafka_topic" = "${topics[i]}", + "property.kafka_default_offsets" = "OFFSET_BEGINNING" + ); + """ + sql "sync" + i++ + } catch (Exception e) { + log.info("exception: ${e.toString()}".toString()) + assertEquals(e.toString(), "java.sql.SQLException: errCode = 2, detailMessage = exec_mem_limitshould > 0") + } + } + } finally { + for (String tableName in tables) { + sql new File("""${context.file.parent}/ddl/${tableName}_drop.sql""").text + } + } + } + // timezone i = 0 if (enabled != null && enabled.equalsIgnoreCase("true")) { @@ -538,7 +579,7 @@ suite("test_routine_load","p0") { sleep(5000) count++ } - + if (i <= 3) { qt_sql_max_filter_ratio "select * from ${tableName1} order by k00,k01" } else { @@ -603,7 +644,7 @@ suite("test_routine_load","p0") { } else { qt_sql_load_to_single_tablet "select * from ${tableName1} order by k00" } - + sql "stop routine load for ${jobs[i]}" i++ } @@ -829,7 +870,7 @@ suite("test_routine_load","p0") { } } } - + // disable_simdjson_reader and load json i = 0 if (enabled != null && enabled.equalsIgnoreCase("true")) { @@ -927,7 +968,7 @@ suite("test_routine_load","p0") { } } } - + // TODO: need update kafka script // i = 0 @@ -1071,13 +1112,13 @@ suite("test_routine_load","p0") { sleep(5000) count++ } - + if (i <= 3) { qt_sql_multi_table_one_data "select * from ${tableName1} order by k00,k01" } else { qt_sql_multi_table_one_data "select * from ${tableName1} order by k00" } - + i++ } } finally { @@ -1150,13 +1191,13 @@ suite("test_routine_load","p0") { sleep(5000) count++ } - + if (i <= 3) { qt_sql_multi_table "select * from ${tableName1} order by k00,k01" } else { qt_sql_multi_table "select * from ${tableName1} order by k00" } - + i++ } } finally { From 332972c6e5ff82942b89a2ecb6bdbccd15b4bb11 Mon Sep 17 00:00:00 2001 From: liugddx Date: Tue, 21 Nov 2023 10:46:07 +0800 Subject: [PATCH 2/4] 1 --- .../java/org/apache/doris/analysis/CreateRoutineLoadStmt.java | 3 ++- .../suites/load_p0/routine_load/test_routine_load.groovy | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java index 1f49a42995eaab..06ca7ad19dd3a2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java +++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java @@ -488,7 +488,8 @@ private void checkJobProperties() throws UserException { RoutineLoadJob.DEFAULT_STRICT_MODE, LoadStmt.STRICT_MODE + " should be a boolean"); execMemLimit = Util.getLongPropertyOrDefault(jobProperties.get(EXEC_MEM_LIMIT_PROPERTY), - RoutineLoadJob.DEFAULT_EXEC_MEM_LIMIT, EXEC_MEM_LIMIT_PRED, EXEC_MEM_LIMIT_PROPERTY + "should > 0"); + RoutineLoadJob.DEFAULT_EXEC_MEM_LIMIT, EXEC_MEM_LIMIT_PRED, + EXEC_MEM_LIMIT_PROPERTY + " must be greater than > 0"); sendBatchParallelism = ((Long) Util.getLongPropertyOrDefault(jobProperties.get(SEND_BATCH_PARALLELISM), ConnectContext.get().getSessionVariable().getSendBatchParallelism(), SEND_BATCH_PARALLELISM_PRED, diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy index c0b62b0f321e27..7a892066d55bfb 100644 --- a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy +++ b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy @@ -264,7 +264,7 @@ suite("test_routine_load","p0") { i++ } catch (Exception e) { log.info("exception: ${e.toString()}".toString()) - assertEquals(e.toString(), "java.sql.SQLException: errCode = 2, detailMessage = exec_mem_limitshould > 0") + assertEquals(e.toString(), "java.sql.SQLException: errCode = 2, detailMessage = exec_mem_limit must be greater than > 0") } } } finally { From 3caf89ebf10fefafce48f0f398a19d3fa0ae72dc Mon Sep 17 00:00:00 2001 From: Guangdong Liu <804167098@qq.com> Date: Tue, 21 Nov 2023 11:31:17 +0800 Subject: [PATCH 3/4] 1 --- .../java/org/apache/doris/analysis/CreateRoutineLoadStmt.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java index 06ca7ad19dd3a2..ea4dc85435b907 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java +++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java @@ -489,7 +489,7 @@ private void checkJobProperties() throws UserException { LoadStmt.STRICT_MODE + " should be a boolean"); execMemLimit = Util.getLongPropertyOrDefault(jobProperties.get(EXEC_MEM_LIMIT_PROPERTY), RoutineLoadJob.DEFAULT_EXEC_MEM_LIMIT, EXEC_MEM_LIMIT_PRED, - EXEC_MEM_LIMIT_PROPERTY + " must be greater than > 0"); + EXEC_MEM_LIMIT_PROPERTY + " must be greater than 0"); sendBatchParallelism = ((Long) Util.getLongPropertyOrDefault(jobProperties.get(SEND_BATCH_PARALLELISM), ConnectContext.get().getSessionVariable().getSendBatchParallelism(), SEND_BATCH_PARALLELISM_PRED, From 80a413e590d5bed74aedf184c061b593bff72be6 Mon Sep 17 00:00:00 2001 From: Guangdong Liu <804167098@qq.com> Date: Tue, 21 Nov 2023 20:11:45 +0800 Subject: [PATCH 4/4] 1 --- .../suites/load_p0/routine_load/test_routine_load.groovy | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy index 7a892066d55bfb..4633f02722d6ef 100644 --- a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy +++ b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy @@ -264,7 +264,7 @@ suite("test_routine_load","p0") { i++ } catch (Exception e) { log.info("exception: ${e.toString()}".toString()) - assertEquals(e.toString(), "java.sql.SQLException: errCode = 2, detailMessage = exec_mem_limit must be greater than > 0") + assertEquals(e.toString(), "java.sql.SQLException: errCode = 2, detailMessage = exec_mem_limit must be greater than 0") } } } finally {