You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/02/20 08:24:21 UTC
(doris) branch branch-2.1 updated (24fb24db3eb -> 2af920c6f89)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a change to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
from 24fb24db3eb [refactor](catalog) catalog interface refactor #31047 (#31135)
new b69acbf98bf [Improvement](group commit) FE should throw NotFoundException if database or table not exist when loading data (#31113)
new 15964ee3d84 [opt](deps) reduce the size of FE dependencies (#31129)
new a7a530189e2 [regression-test](decimalv3) Adding numerical precision exceeding the data type will error case (#31068)
new fd453ace383 [enhancement](sc-test) Optimize waitForSchemeChangeDone (#31002)
new 493385c2c71 [Bug](AggState) fix not match function when agg combinator function has alias (#31101)
new 54e8354fc69 [doc](mtmv) Update the mtmv documentation to make it consistent with the code (#30899)
new 4aaab6fb441 [fix](statistics)Refresh follower FE cache after alter column stats. Support alter index column stats (#31108)
new 95a1e1c0bfb [Enhancement](tools) Support transaction for benchmarks loading (#31126)
new 92518166738 [fix](Nereids): log constraint editlog in lock scope (#30630)
new 6d7662e8ac5 [fix](tools) Fix non-standard code in tpch/tpcds dbgen tools leading to compilation failure (#31124)
new 97c9d75af34 [Feature](executor)Add scan_thread_num property for workload group (#31106)
new 2af920c6f89 [chore](regression) Update q03.sql (#31063)
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/runtime/task_group/task_group.cpp | 15 +-
be/src/runtime/task_group/task_group.h | 2 +
be/src/runtime/task_group/task_group_manager.cpp | 4 +
be/src/vec/data_types/data_type_agg_state.h | 28 +---
be/src/vec/exec/scan/scanner_scheduler.cpp | 2 +-
be/src/vec/exec/scan/scanner_scheduler.h | 35 ++---
be/src/vec/exprs/vectorized_agg_fn.cpp | 4 +-
docs/en/docs/admin-manual/workload-group.md | 1 +
.../async-materialized-view.md | 5 +
.../Create/CREATE-ASYNC-MATERIALIZED-VIEW.md | 9 +-
docs/zh-CN/docs/admin-manual/workload-group.md | 1 +
.../async-materialized-view.md | 5 +
.../Create/CREATE-ASYNC-MATERIALIZED-VIEW.md | 8 +-
fe/fe-core/pom.xml | 4 +
fe/fe-core/src/main/cup/sql_parser.cup | 7 +-
.../doris/analysis/AlterColumnStatsStmt.java | 27 +++-
.../java/org/apache/doris/catalog/TableIf.java | 113 ++++++++------
.../trees/plans/commands/AddConstraintCommand.java | 13 +-
.../plans/commands/DropConstraintCommand.java | 5 +-
.../java/org/apache/doris/persist/EditLog.java | 2 +-
.../resource/workloadgroup/WorkloadGroup.java | 24 ++-
.../resource/workloadgroup/WorkloadGroupMgr.java | 1 +
.../apache/doris/service/FrontendServiceImpl.java | 13 +-
.../org/apache/doris/statistics/ColStatsData.java | 12 ++
.../doris/statistics/StatisticsRepository.java | 11 +-
.../doris/tablefunction/MetadataGenerator.java | 5 +-
.../WorkloadGroupsTableValuedFunction.java | 1 +
.../catalog/constraint/ConstraintPersistTest.java | 55 ++++++-
fe/pom.xml | 22 ++-
gensrc/thrift/BackendService.thrift | 1 +
.../data/datatype_p0/agg_state/test_agg_state.out | 6 +
.../decimalv3/test_decimalv3_string.out} | 11 +-
.../data/workload_manager_p0/test_curd_wlg.out | 24 +--
.../org/apache/doris/regression/suite/Suite.groovy | 11 +-
.../datatype_p0/agg_state/test_agg_state.groovy | 3 +
.../decimalv3/test_decimalv3_string.groovy | 28 ++++
.../sql_functions/array_functions/sql/q03.sql | 2 +-
.../suites/statistics/test_analyze_mv.groovy | 42 +++++
.../workload_manager_p0/test_curd_wlg.groovy | 20 ++-
tools/clickbench-tools/load-clickbench-data.sh | 37 +++--
tools/ssb-tools/bin/load-ssb-data.sh | 169 ++++++++++++++++++---
tools/tpcds-tools/README.md | 3 +-
tools/tpcds-tools/bin/build-tpcds-tools.sh | 10 +-
tools/tpcds-tools/bin/load-tpcds-data.sh | 40 +++--
tools/tpch-tools/README.md | 2 +
tools/tpch-tools/bin/build-tpch-dbgen.sh | 8 +-
tools/tpch-tools/bin/load-tpch-data.sh | 123 +++++++++++----
47 files changed, 732 insertions(+), 242 deletions(-)
copy regression-test/data/{compaction/test_single_replica_compaction.out => datatype_p0/decimalv3/test_decimalv3_string.out} (57%)
create mode 100644 regression-test/suites/datatype_p0/decimalv3/test_decimalv3_string.groovy
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/12: [opt](deps) reduce the size of FE dependencies (#31129)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 15964ee3d845fbdc65c8fb4133a9215e9a50b519
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Tue Feb 20 09:27:07 2024 +0800
[opt](deps) reduce the size of FE dependencies (#31129)
---
fe/fe-core/pom.xml | 4 ++++
fe/pom.xml | 22 ++++++++++++++--------
2 files changed, 18 insertions(+), 8 deletions(-)
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
index 5714a0ea657..db766cff351 100644
--- a/fe/fe-core/pom.xml
+++ b/fe/fe-core/pom.xml
@@ -398,6 +398,10 @@ under the License.
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-dynamodb</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-logs</artifactId>
+ </dependency>
<dependency>
<groupId>com.huaweicloud</groupId>
<artifactId>hadoop-huaweicloud</artifactId>
diff --git a/fe/pom.xml b/fe/pom.xml
index ebbbf010f1b..3fd1bfc751f 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -402,11 +402,6 @@ under the License.
<artifactId>nimbus-jose-jwt</artifactId>
<version>${nimbusds.version}</version>
</dependency>
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>aws-java-sdk-bundle</artifactId>
- <version>${aws-java-sdk.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.orc</groupId>
<artifactId>orc-core</artifactId>
@@ -908,12 +903,17 @@ under the License.
<artifactId>logback-classic</artifactId>
</exclusion>
<exclusion>
- <artifactId>elasticsearch-rest-high-level-client</artifactId>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-storage-api</artifactId>
+ </exclusion>
+ <exclusion>
<groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-high-level-client</artifactId>
</exclusion>
+ <!--ranger audit only depends on aws logs, which is provided alone-->
<exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-storage-api</artifactId>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
@@ -1416,6 +1416,12 @@ under the License.
<artifactId>aws-java-sdk-dynamodb</artifactId>
<version>${aws-java-sdk.version}</version>
</dependency>
+ <!--only for apache ranger audit-->
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-logs</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
<dependency>
<groupId>org.mariadb.jdbc</groupId>
<artifactId>mariadb-java-client</artifactId>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/12: [Improvement](group commit) FE should throw NotFoundException if database or table not exist when loading data (#31113)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit b69acbf98bf8395dd2b4477ce1362f20b408a9ba
Author: huanghaibin <28...@qq.com>
AuthorDate: Tue Feb 20 09:21:31 2024 +0800
[Improvement](group commit) FE should throw NotFoundException if database or table not exist when loading data (#31113)
* [Improvement](group commit) FE should throw NotFoundException if database or table not exist when loading data
* edit
---
.../java/org/apache/doris/service/FrontendServiceImpl.java | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index bfce1a8f7ae..34050c2dced 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -1108,6 +1108,10 @@ public class FrontendServiceImpl implements FrontendService.Iface {
status.setStatusCode(TStatusCode.LABEL_ALREADY_EXISTS);
status.addToErrorMsgs(e.getMessage());
result.setJobStatus(e.getJobStatus());
+ } catch (MetaNotFoundException e) {
+ LOG.warn("failed to begin", e);
+ status.setStatusCode(TStatusCode.NOT_FOUND);
+ status.addToErrorMsgs(e.getMessage());
} catch (UserException e) {
LOG.warn("failed to begin: {}", e.getMessage());
status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
@@ -1145,7 +1149,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (Strings.isNullOrEmpty(request.getCluster())) {
dbName = request.getDb();
}
- throw new UserException("unknown database, database=" + dbName);
+ throw new MetaNotFoundException("unknown database, database=" + dbName);
}
OlapTable table = (OlapTable) db.getTableOrMetaException(request.tbl, TableType.OLAP);
@@ -1249,7 +1253,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (Strings.isNullOrEmpty(request.getCluster())) {
dbName = request.getDb();
}
- throw new UserException("unknown database, database=" + dbName);
+ throw new MetaNotFoundException("unknown database, database=" + dbName);
}
// step 4: fetch all tableIds
@@ -1679,6 +1683,11 @@ public class FrontendServiceImpl implements FrontendService.Iface {
}
try {
loadTxnRollbackImpl(request);
+ } catch (MetaNotFoundException e) {
+ String msg = "failed to rollback txn" + request.getTxnId();
+ LOG.warn(msg, e);
+ status.setStatusCode(TStatusCode.NOT_FOUND);
+ status.addToErrorMsgs(e.getMessage());
} catch (UserException e) {
LOG.warn("failed to rollback txn {}: {}", request.getTxnId(), e.getMessage());
status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 10/12: [fix](tools) Fix non-standard code in tpch/tpcds dbgen tools leading to compilation failure (#31124)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 6d7662e8ac5784ee56e54333d50f0cc72534c2ca
Author: zclllyybb <zh...@selectdb.com>
AuthorDate: Tue Feb 20 15:15:14 2024 +0800
[fix](tools) Fix non-standard code in tpch/tpcds dbgen tools leading to compilation failure (#31124)
Fix non-standard code in tpch/tpcds dbgen tools leading to compilation failure
---
tools/tpcds-tools/README.md | 3 ++-
tools/tpcds-tools/bin/build-tpcds-tools.sh | 10 +++++-----
tools/tpch-tools/README.md | 2 ++
tools/tpch-tools/bin/build-tpch-dbgen.sh | 8 ++++----
4 files changed, 13 insertions(+), 10 deletions(-)
diff --git a/tools/tpcds-tools/README.md b/tools/tpcds-tools/README.md
index e4cf768d3c6..d9b6aa7e738 100644
--- a/tools/tpcds-tools/README.md
+++ b/tools/tpcds-tools/README.md
@@ -24,9 +24,10 @@ follow the steps below:
### 1. build tpc-ds dsdgen dsqgen tool.
- export PATH=/usr/bin/:$PATH
./bin/build-tpcds-tools.sh
+ If the build failed in dbgen tools' compilation, update your GCC version or change all "TPC-DS_Tools_v3.2.0new.zip" in build-tpcds-dbgen.sh to "TPC-DS_Tools_v3.2.0.zip"
+
### 2. generate tpc-ds data. use -h for more infomations.
./bin/gen-tpcds-data.sh -s 1
diff --git a/tools/tpcds-tools/bin/build-tpcds-tools.sh b/tools/tpcds-tools/bin/build-tpcds-tools.sh
index fd6a486f4c3..f765b0a4aa0 100755
--- a/tools/tpcds-tools/bin/build-tpcds-tools.sh
+++ b/tools/tpcds-tools/bin/build-tpcds-tools.sh
@@ -47,12 +47,12 @@ check_prerequest "unzip -h" "unzip"
# download tpcds tools package first
if [[ -d "${CURDIR}/DSGen-software-code-3.2.0rc1" ]]; then
- echo "If you want to rebuild TPC-DS_Tools_v3.2.0 again, please delete ${CURDIR}/DSGen-software-code-3.2.0rc1 first." && exit 1
-elif [[ -f "${CURDIR}/TPC-DS_Tools_v3.2.0.zip" ]]; then
- unzip TPC-DS_Tools_v3.2.0.zip -d "${CURDIR}/"
+ echo "If you want to rebuild TPC-DS_Tools_v3.2.0 again, please delete ${CURDIR}/DSGen-software-code-3.2.0rc1 first."
+elif [[ -f "${CURDIR}/TPC-DS_Tools_v3.2.0new.zip" ]]; then
+ unzip TPC-DS_Tools_v3.2.0new.zip -d "${CURDIR}/"
else
- wget "https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/tools/TPC-DS_Tools_v3.2.0.zip"
- unzip TPC-DS_Tools_v3.2.0.zip -d "${CURDIR}/"
+ wget "https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/tools/TPC-DS_Tools_v3.2.0new.zip"
+ unzip TPC-DS_Tools_v3.2.0new.zip -d "${CURDIR}/"
fi
# compile tpcds-dsdgen
diff --git a/tools/tpch-tools/README.md b/tools/tpch-tools/README.md
index d1b592bc673..a7deac387ba 100644
--- a/tools/tpch-tools/README.md
+++ b/tools/tpch-tools/README.md
@@ -26,6 +26,8 @@ follow the steps below:
./bin/build-tpch-dbgen.sh
+ If the build failed in dbgen tools' compilation, update your GCC version or change download link used by wget in build-tpch-dbgen.sh from ".../TPC-H_Tools_v3.0.0new.zip" to ".../TPC-H_Tools_v3.0.0.zip"
+
### 2. generate tpc-h data. use -h for more infomations.
./bin/gen-tpch-data.sh -s 1
diff --git a/tools/tpch-tools/bin/build-tpch-dbgen.sh b/tools/tpch-tools/bin/build-tpch-dbgen.sh
index eab1b7d591e..9499ae5d74e 100755
--- a/tools/tpch-tools/bin/build-tpch-dbgen.sh
+++ b/tools/tpch-tools/bin/build-tpch-dbgen.sh
@@ -38,7 +38,7 @@ check_prerequest() {
local CMD=$1
local NAME=$2
if ! ${CMD}; then
- echo "${NAME} is missing. This script depends on unzip to extract files from TPC-H_Tools_v3.0.0.zip"
+ echo "${NAME} is missing. This script depends on unzip to extract files from TPC-H_Tools_v3.0.0new.zip"
exit 1
fi
}
@@ -48,10 +48,10 @@ check_prerequest "unzip -h" "unzip"
# download tpch tools pacage first
if [[ -d ${TPCH_DBGEN_DIR} ]]; then
echo "Dir ${TPCH_DBGEN_DIR} already exists. No need to download."
- echo "If you want to download TPC-H_Tools_v3.0.0 again, please delete this dir first."
+ echo "If you want to download TPC-H_Tools_v3.0.0new again, please delete this dir first."
else
- wget "https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/tools/TPC-H_Tools_v3.0.0.zip"
- unzip TPC-H_Tools_v3.0.0.zip -d "${CURDIR}/"
+ wget "https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/tools/TPC-H_Tools_v3.0.0new.zip"
+ unzip TPC-H_Tools_v3.0.0new.zip -d "${CURDIR}/"
fi
# modify tpcd.h
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 08/12: [Enhancement](tools) Support transaction for benchmarks loading (#31126)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 95a1e1c0bfbbbde79152311f3db38e9e6bc3c20c
Author: zclllyybb <zh...@selectdb.com>
AuthorDate: Tue Feb 20 14:29:34 2024 +0800
[Enhancement](tools) Support transaction for benchmarks loading (#31126)
---
tools/clickbench-tools/load-clickbench-data.sh | 37 ++++--
tools/ssb-tools/bin/load-ssb-data.sh | 169 +++++++++++++++++++++----
tools/tpcds-tools/bin/load-tpcds-data.sh | 40 ++++--
tools/tpch-tools/bin/load-tpch-data.sh | 123 ++++++++++++++----
4 files changed, 299 insertions(+), 70 deletions(-)
diff --git a/tools/clickbench-tools/load-clickbench-data.sh b/tools/clickbench-tools/load-clickbench-data.sh
index aff122edff4..41d17a8a76e 100755
--- a/tools/clickbench-tools/load-clickbench-data.sh
+++ b/tools/clickbench-tools/load-clickbench-data.sh
@@ -30,13 +30,18 @@ ROOT=$(
CURDIR=${ROOT}
DATA_DIR=$CURDIR/
-# DATA_DIR=/mnt/disk1/stephen/data/clickbench
usage() {
echo "
This script is used to load ClickBench data,
will use mysql client to connect Doris server which is specified in conf/doris-cluster.conf file.
-Usage: $0
+Usage: $0 <options>
+ Optional options:
+ -x use transaction id. multi times of loading with the same id won't load duplicate data.
+
+ Eg.
+ $0 load data using default value.
+ $0 -x blabla use transaction id \"blabla\".
"
exit 1
}
@@ -44,11 +49,13 @@ Usage: $0
OPTS=$(getopt \
-n $0 \
-o '' \
- -o 'h' \
+ -o 'hx:' \
-- "$@")
eval set -- "$OPTS"
HELP=0
+TXN_ID=""
+
while true; do
case "$1" in
-h)
@@ -59,6 +66,10 @@ while true; do
shift
break
;;
+ -x)
+ TXN_ID=$2
+ shift 2
+ ;;
*)
echo "Internal error"
exit 1
@@ -113,7 +124,6 @@ function load() {
if [ ! -f "$DATA_DIR/hits_split${i}" ]; then
echo "will download hits_split${i} to $DATA_DIR"
wget --continue "https://doris-test-data.oss-cn-hongkong.aliyuncs.com/ClickBench/hits_split${i}" &
- # wget --continue "https://doris-test-data.oss-cn-hongkong-internal.aliyuncs.com/ClickBench/hits_split${i}" &
PID=$!
wget_pids[${#wget_pids[@]}]=$PID
fi
@@ -127,11 +137,20 @@ function load() {
for i in $(seq 0 9); do
echo -e "
start loading hits_split${i}"
- curl --location-trusted \
- -u $USER:$PASSWORD \
- -T "$DATA_DIR/hits_split${i}" \
- -H "columns:WatchID,JavaEnable,Title,GoodEvent,EventTime,EventDate,CounterID,ClientIP,RegionID,UserID,CounterClass,OS,UserAgent,URL,Referer,IsRefresh,RefererCategoryID,RefererRegionID,URLCategoryID,URLRegionID,ResolutionWidth,ResolutionHeight,ResolutionDepth,FlashMajor,FlashMinor,FlashMinor2,NetMajor,NetMinor,UserAgentMajor,UserAgentMinor,CookieEnable,JavascriptEnable,IsMobile,MobilePhone,MobilePhoneModel,Params,IPNetworkID,TraficSourceID,SearchEngineID,SearchPhrase,AdvEngine [...]
- http://$FE_HOST:$FE_HTTP_PORT/api/$DB/hits/_stream_load
+ if [[ -z ${TXN_ID} ]]; then
+ curl --location-trusted \
+ -u $USER:$PASSWORD \
+ -T "$DATA_DIR/hits_split${i}" \
+ -H "columns:WatchID,JavaEnable,Title,GoodEvent,EventTime,EventDate,CounterID,ClientIP,RegionID,UserID,CounterClass,OS,UserAgent,URL,Referer,IsRefresh,RefererCategoryID,RefererRegionID,URLCategoryID,URLRegionID,ResolutionWidth,ResolutionHeight,ResolutionDepth,FlashMajor,FlashMinor,FlashMinor2,NetMajor,NetMinor,UserAgentMajor,UserAgentMinor,CookieEnable,JavascriptEnable,IsMobile,MobilePhone,MobilePhoneModel,Params,IPNetworkID,TraficSourceID,SearchEngineID,SearchPhrase,AdvEn [...]
+ http://$FE_HOST:$FE_HTTP_PORT/api/$DB/hits/_stream_load
+ else
+ curl --location-trusted \
+ -u $USER:$PASSWORD \
+ -T "$DATA_DIR/hits_split${i}" \
+ -H "label:${TXN_ID}_${i}" \
+ -H "columns:WatchID,JavaEnable,Title,GoodEvent,EventTime,EventDate,CounterID,ClientIP,RegionID,UserID,CounterClass,OS,UserAgent,URL,Referer,IsRefresh,RefererCategoryID,RefererRegionID,URLCategoryID,URLRegionID,ResolutionWidth,ResolutionHeight,ResolutionDepth,FlashMajor,FlashMinor,FlashMinor2,NetMajor,NetMinor,UserAgentMajor,UserAgentMinor,CookieEnable,JavascriptEnable,IsMobile,MobilePhone,MobilePhoneModel,Params,IPNetworkID,TraficSourceID,SearchEngineID,SearchPhrase,AdvEn [...]
+ http://$FE_HOST:$FE_HTTP_PORT/api/$DB/hits/_stream_load
+ fi
done
}
diff --git a/tools/ssb-tools/bin/load-ssb-data.sh b/tools/ssb-tools/bin/load-ssb-data.sh
index efd3bf3109f..d30720d7f1e 100755
--- a/tools/ssb-tools/bin/load-ssb-data.sh
+++ b/tools/ssb-tools/bin/load-ssb-data.sh
@@ -36,11 +36,13 @@ usage() {
echo "
Usage: $0 <options>
Optional options:
- -c parallelism to load data of lineorder table, default is 5.
+ -c parallelism to load data of lineorder table, default is 5.
+ -x use transaction id. multi times of loading with the same id won't load duplicate data.
Eg.
$0 load data using default value.
- $0 -c 10 load lineorder table data using parallelism 10.
+ $0 -c 10 load lineorder table data using parallelism 10.
+ $0 -x blabla use transaction id \"blabla\".
"
exit 1
}
@@ -48,13 +50,14 @@ Usage: $0 <options>
OPTS=$(getopt \
-n "$0" \
-o '' \
- -o 'hc:' \
+ -o 'hc:x:' \
-- "$@")
eval set -- "${OPTS}"
PARALLEL=5
HELP=0
+TXN_ID=""
if [[ $# == 0 ]]; then
usage
@@ -70,6 +73,10 @@ while true; do
PARALLEL=$2
shift 2
;;
+ -x)
+ TXN_ID=$2
+ shift 2
+ ;;
--)
shift
break
@@ -110,9 +117,13 @@ run_sql() {
load_lineitem_flat() {
# Loading data in batches by year.
+ local flat_con_idx=0
for con in 'lo_orderdate<19930101' 'lo_orderdate>=19930101 and lo_orderdate<19940101' 'lo_orderdate>=19940101 and lo_orderdate<19950101' 'lo_orderdate>=19950101 and lo_orderdate<19960101' 'lo_orderdate>=19960101 and lo_orderdate<19970101' 'lo_orderdate>=19970101 and lo_orderdate<19980101' 'lo_orderdate>=19980101'; do
echo -e "\n${con}"
- run_sql "
+ flat_con_idx=$((flat_con_idx + 1))
+
+ if [[ -z ${TXN_ID} ]]; then
+ run_sql "
INSERT INTO lineorder_flat
SELECT
LO_ORDERDATE,
@@ -182,6 +193,79 @@ ON (s.s_suppkey = l.lo_suppkey)
INNER JOIN part p
ON (p.p_partkey = l.lo_partkey);
"
+ else
+ run_sql "
+INSERT INTO lineorder_flat
+WITH LABEL \`${TXN_ID}_flat_${flat_con_idx}\`
+SELECT
+ LO_ORDERDATE,
+ LO_ORDERKEY,
+ LO_LINENUMBER,
+ LO_CUSTKEY,
+ LO_PARTKEY,
+ LO_SUPPKEY,
+ LO_ORDERPRIORITY,
+ LO_SHIPPRIORITY,
+ LO_QUANTITY,
+ LO_EXTENDEDPRICE,
+ LO_ORDTOTALPRICE,
+ LO_DISCOUNT,
+ LO_REVENUE,
+ LO_SUPPLYCOST,
+ LO_TAX,
+ LO_COMMITDATE,
+ LO_SHIPMODE,
+ C_NAME,
+ C_ADDRESS,
+ C_CITY,
+ C_NATION,
+ C_REGION,
+ C_PHONE,
+ C_MKTSEGMENT,
+ S_NAME,
+ S_ADDRESS,
+ S_CITY,
+ S_NATION,
+ S_REGION,
+ S_PHONE,
+ P_NAME,
+ P_MFGR,
+ P_CATEGORY,
+ P_BRAND,
+ P_COLOR,
+ P_TYPE,
+ P_SIZE,
+ P_CONTAINER
+FROM (
+ SELECT
+ lo_orderkey,
+ lo_linenumber,
+ lo_custkey,
+ lo_partkey,
+ lo_suppkey,
+ lo_orderdate,
+ lo_orderpriority,
+ lo_shippriority,
+ lo_quantity,
+ lo_extendedprice,
+ lo_ordtotalprice,
+ lo_discount,
+ lo_revenue,
+ lo_supplycost,
+ lo_tax,
+ lo_commitdate,
+ lo_shipmode
+ FROM lineorder
+ WHERE ${con}
+) l
+INNER JOIN customer c
+ON (c.c_custkey = l.lo_custkey)
+INNER JOIN supplier s
+ON (s.s_suppkey = l.lo_suppkey)
+INNER JOIN part p
+ON (p.p_partkey = l.lo_partkey);
+"
+ fi
done
}
@@ -200,36 +284,73 @@ echo "Start time: $(date)"
echo "==========Start to load data into ssb tables=========="
echo 'Loading data for table: part'
-curl --location-trusted -u "${USER}":"${PASSWORD}" \
- -H "column_separator:|" \
- -H "columns:p_partkey,p_name,p_mfgr,p_category,p_brand,p_color,p_type,p_size,p_container,p_dummy" \
- -T "${SSB_DATA_DIR}"/part.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/part/_stream_load
+if [[ -z ${TXN_ID} ]]; then
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "column_separator:|" \
+ -H "columns:p_partkey,p_name,p_mfgr,p_category,p_brand,p_color,p_type,p_size,p_container,p_dummy" \
+ -T "${SSB_DATA_DIR}"/part.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/part/_stream_load
+else
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "label:${TXN_ID}_part" -H "column_separator:|" \
+ -H "columns:p_partkey,p_name,p_mfgr,p_category,p_brand,p_color,p_type,p_size,p_container,p_dummy" \
+ -T "${SSB_DATA_DIR}"/part.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/part/_stream_load
+fi
echo 'Loading data for table: date'
-curl --location-trusted -u "${USER}":"${PASSWORD}" \
- -H "column_separator:|" \
- -H "columns:d_datekey,d_date,d_dayofweek,d_month,d_year,d_yearmonthnum,d_yearmonth,d_daynuminweek,d_daynuminmonth,d_daynuminyear,d_monthnuminyear,d_weeknuminyear,d_sellingseason,d_lastdayinweekfl,d_lastdayinmonthfl,d_holidayfl,d_weekdayfl,d_dummy" \
- -T "${SSB_DATA_DIR}"/date.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/dates/_stream_load
+if [[ -z ${TXN_ID} ]]; then
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "label:${TXN_ID}_date" -H "column_separator:|" \
+ -H "columns:d_datekey,d_date,d_dayofweek,d_month,d_year,d_yearmonthnum,d_yearmonth,d_daynuminweek,d_daynuminmonth,d_daynuminyear,d_monthnuminyear,d_weeknuminyear,d_sellingseason,d_lastdayinweekfl,d_lastdayinmonthfl,d_holidayfl,d_weekdayfl,d_dummy" \
+ -T "${SSB_DATA_DIR}"/date.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/dates/_stream_load
+else
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "column_separator:|" \
+ -H "columns:d_datekey,d_date,d_dayofweek,d_month,d_year,d_yearmonthnum,d_yearmonth,d_daynuminweek,d_daynuminmonth,d_daynuminyear,d_monthnuminyear,d_weeknuminyear,d_sellingseason,d_lastdayinweekfl,d_lastdayinmonthfl,d_holidayfl,d_weekdayfl,d_dummy" \
+ -T "${SSB_DATA_DIR}"/date.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/dates/_stream_load
+fi
echo 'Loading data for table: supplier'
-curl --location-trusted -u "${USER}":"${PASSWORD}" \
- -H "column_separator:|" \
- -H "columns:s_suppkey,s_name,s_address,s_city,s_nation,s_region,s_phone,s_dummy" \
- -T "${SSB_DATA_DIR}"/supplier.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/supplier/_stream_load
+if [[ -z ${TXN_ID} ]]; then
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "column_separator:|" \
+ -H "columns:s_suppkey,s_name,s_address,s_city,s_nation,s_region,s_phone,s_dummy" \
+ -T "${SSB_DATA_DIR}"/supplier.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/supplier/_stream_load
+else
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "label:${TXN_ID}_supplier" -H "column_separator:|" \
+ -H "columns:s_suppkey,s_name,s_address,s_city,s_nation,s_region,s_phone,s_dummy" \
+ -T "${SSB_DATA_DIR}"/supplier.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/supplier/_stream_load
+fi
echo 'Loading data for table: customer'
-curl --location-trusted -u "${USER}":"${PASSWORD}" \
- -H "column_separator:|" \
- -H "columns:c_custkey,c_name,c_address,c_city,c_nation,c_region,c_phone,c_mktsegment,no_use" \
- -T "${SSB_DATA_DIR}"/customer.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/customer/_stream_load
+if [[ -z ${TXN_ID} ]]; then
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "column_separator:|" \
+ -H "columns:c_custkey,c_name,c_address,c_city,c_nation,c_region,c_phone,c_mktsegment,no_use" \
+ -T "${SSB_DATA_DIR}"/customer.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/customer/_stream_load
+else
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "label:${TXN_ID}_customer" -H "column_separator:|" \
+ -H "columns:c_custkey,c_name,c_address,c_city,c_nation,c_region,c_phone,c_mktsegment,no_use" \
+ -T "${SSB_DATA_DIR}"/customer.tbl http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/customer/_stream_load
+fi
echo "Loading data for table: lineorder, with ${PARALLEL} parallel"
function load() {
echo "$@"
- curl --location-trusted -u "${USER}":"${PASSWORD}" \
- -H "column_separator:|" \
- -H "columns:lo_orderkey,lo_linenumber,lo_custkey,lo_partkey,lo_suppkey,lo_orderdate,lo_orderpriority,lo_shippriority,lo_quantity,lo_extendedprice,lo_ordtotalprice,lo_discount,lo_revenue,lo_supplycost,lo_tax,lo_commitdate,lo_shipmode,lo_dummy" \
- -T "$@" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/lineorder/_stream_load
+ # shellcheck disable=SC2016,SC2124
+ local FILE_ID="${@//*./}"
+ if [[ -z ${TXN_ID} ]]; then
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "column_separator:|" \
+ -H "columns:lo_orderkey,lo_linenumber,lo_custkey,lo_partkey,lo_suppkey,lo_orderdate,lo_orderpriority,lo_shippriority,lo_quantity,lo_extendedprice,lo_ordtotalprice,lo_discount,lo_revenue,lo_supplycost,lo_tax,lo_commitdate,lo_shipmode,lo_dummy" \
+ -T "$@" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/lineorder/_stream_load
+ else
+ curl --location-trusted -u "${USER}":"${PASSWORD}" \
+ -H "label:${TXN_ID}_lineorder_${FILE_ID}" -H "column_separator:|" \
+ -H "columns:lo_orderkey,lo_linenumber,lo_custkey,lo_partkey,lo_suppkey,lo_orderdate,lo_orderpriority,lo_shippriority,lo_quantity,lo_extendedprice,lo_ordtotalprice,lo_discount,lo_revenue,lo_supplycost,lo_tax,lo_commitdate,lo_shipmode,lo_dummy" \
+ -T "$@" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/lineorder/_stream_load
+ fi
}
# set parallelism
diff --git a/tools/tpcds-tools/bin/load-tpcds-data.sh b/tools/tpcds-tools/bin/load-tpcds-data.sh
index d34021076d6..ab9fb9d3361 100755
--- a/tools/tpcds-tools/bin/load-tpcds-data.sh
+++ b/tools/tpcds-tools/bin/load-tpcds-data.sh
@@ -35,11 +35,13 @@ usage() {
echo "
Usage: $0 <options>
Optional options:
- -c parallelism to load data of lineitem, orders, partsupp, default is 5.
+ -c parallelism to load data of lineitem, orders, partsupp, default is 5.
+ -x use transaction id. multi times of loading with the same id won't load duplicate data.
Eg.
$0 load data using default value.
- $0 -c 10 load lineitem, orders, partsupp table data using parallelism 10.
+ $0 -c 10 load lineitem, orders, partsupp table data using parallelism 10.
+ $0 -x blabla use transaction id \"blabla\".
"
exit 1
}
@@ -47,13 +49,14 @@ Usage: $0 <options>
OPTS=$(getopt \
-n "$0" \
-o '' \
- -o 'hc:' \
+ -o 'hc:x:' \
-- "$@")
eval set -- "${OPTS}"
PARALLEL=5
HELP=0
+TXN_ID=""
if [[ $# == 0 ]]; then
usage
@@ -69,6 +72,10 @@ while true; do
PARALLEL=$2
shift 2
;;
+ -x)
+ TXN_ID=$2
+ shift 2
+ ;;
--)
shift
break
@@ -170,13 +177,26 @@ for table_name in ${!table_columns[*]}; do
{
for file in "${TPCDS_DATA_DIR}/${table_name}"_{1..100}_*.dat; do
if ! [[ -f "${file}" ]]; then continue; fi
- ret=$(curl \
- --location-trusted \
- -u "${USER}":"${PASSWORD:=}" \
- -H "column_separator:|" \
- -H "columns: ${table_columns[${table_name}]}" \
- -T "${file}" \
- http://"${FE_HOST}":"${FE_HTTP_PORT:=8030}"/api/"${DB}"/"${table_name}"/_stream_load 2>/dev/null)
+ FILE_ID=$(echo "${file}" | awk -F'/' '{print $(NF)}' | awk -F'.' '{print $(1)}')
+ if [[ -z ${TXN_ID} ]]; then
+ ret=$(curl \
+ --location-trusted \
+ -u "${USER}":"${PASSWORD:-}" \
+ -H "column_separator:|" \
+ -H "columns: ${table_columns[${table_name}]}" \
+ -T "${file}" \
+ http://"${FE_HOST}":"${FE_HTTP_PORT:-8030}"/api/"${DB}"/"${table_name}"/_stream_load 2>/dev/null)
+ else
+ ret=$(curl \
+ --location-trusted \
+ -u "${USER}":"${PASSWORD:-}" \
+ -H "label:${TXN_ID}_${FILE_ID}" \
+ -H "column_separator:|" \
+ -H "columns: ${table_columns[${table_name}]}" \
+ -T "${file}" \
+ http://"${FE_HOST}":"${FE_HTTP_PORT:-8030}"/api/"${DB}"/"${table_name}"/_stream_load 2>/dev/null)
+ fi
+
if [[ $(echo "${ret}" | jq ".Status") == '"Success"' ]]; then
echo "----loaded ${file}"
else
diff --git a/tools/tpch-tools/bin/load-tpch-data.sh b/tools/tpch-tools/bin/load-tpch-data.sh
index c56d2ea71ac..fa21028f2e9 100755
--- a/tools/tpch-tools/bin/load-tpch-data.sh
+++ b/tools/tpch-tools/bin/load-tpch-data.sh
@@ -36,11 +36,13 @@ usage() {
echo "
Usage: $0 <options>
Optional options:
- -c parallelism to load data of lineitem, orders, partsupp, default is 5.
+ -c parallelism to load data of lineitem, orders, partsupp, default is 5.
+ -x use transaction id. multi times of loading with the same id won't load duplicate data.
Eg.
$0 load data using default value.
- $0 -c 10 load lineitem, orders, partsupp table data using parallelism 10.
+ $0 -c 10 load lineitem, orders, partsupp table data using parallelism 10.
+ $0 -x blabla use transaction id \"blabla\".
"
exit 1
}
@@ -48,13 +50,14 @@ Usage: $0 <options>
OPTS=$(getopt \
-n "$0" \
-o '' \
- -o 'hc:' \
+ -o 'hc:x:' \
-- "$@")
eval set -- "${OPTS}"
PARALLEL=5
HELP=0
+TXN_ID=""
if [[ $# == 0 ]]; then
usage
@@ -70,6 +73,10 @@ while true; do
PARALLEL=$2
shift 2
;;
+ -x)
+ TXN_ID=$2
+ shift 2
+ ;;
--)
shift
break
@@ -116,51 +123,113 @@ echo "DB: ${DB}"
function load_region() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: r_regionkey, r_name, r_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/region/_stream_load
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: r_regionkey, r_name, r_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/region/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_region" \
+ -H "columns: r_regionkey, r_name, r_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/region/_stream_load
+ fi
}
function load_nation() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: n_nationkey, n_name, n_regionkey, n_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/nation/_stream_load
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: n_nationkey, n_name, n_regionkey, n_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/nation/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_nation" \
+ -H "columns: n_nationkey, n_name, n_regionkey, n_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/nation/_stream_load
+ fi
}
function load_supplier() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/supplier/_stream_load
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/supplier/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_supplier" \
+ -H "columns: s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/supplier/_stream_load
+ fi
}
function load_customer() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/customer/_stream_load
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/customer/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_customer" \
+ -H "columns: c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/customer/_stream_load
+ fi
}
function load_part() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: p_partkey, p_name, p_mfgr, p_brand, p_type, p_size, p_container, p_retailprice, p_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/part/_stream_load
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: p_partkey, p_name, p_mfgr, p_brand, p_type, p_size, p_container, p_retailprice, p_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/part/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_part" \
+ -H "columns: p_partkey, p_name, p_mfgr, p_brand, p_type, p_size, p_container, p_retailprice, p_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/part/_stream_load
+ fi
}
function load_partsupp() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: ps_partkey, ps_suppkey, ps_availqty, ps_supplycost, ps_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/partsupp/_stream_load
+ # shellcheck disable=SC2016,SC2124
+ local FILE_ID="${@//*./}"
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: ps_partkey, ps_suppkey, ps_availqty, ps_supplycost, ps_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/partsupp/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_partsupp_${FILE_ID}" \
+ -H "columns: ps_partkey, ps_suppkey, ps_availqty, ps_supplycost, ps_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/partsupp/_stream_load
+ fi
}
function load_orders() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment, temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/orders/_stream_load
+ # shellcheck disable=SC2016,SC2124
+ local FILE_ID="${@//*./}"
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/orders/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_orders_${FILE_ID}" \
+ -H "columns: o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment, temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/orders/_stream_load
+ fi
}
function load_lineitem() {
echo "$*"
- curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
- -H "columns: l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag,l_linestatus, l_shipdate,l_commitdate,l_receiptdate,l_shipinstruct,l_shipmode,l_comment,temp" \
- -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/lineitem/_stream_load
+ # shellcheck disable=SC2016,SC2124
+ local FILE_ID="${@//*./}"
+ if [[ -z ${TXN_ID} ]]; then
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "columns: l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag,l_linestatus, l_shipdate,l_commitdate,l_receiptdate,l_shipinstruct,l_shipmode,l_comment,temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/lineitem/_stream_load
+ else
+ curl -s --location-trusted -u "${USER}":"${PASSWORD}" -H "column_separator:|" \
+ -H "label:${TXN_ID}_lineitem_${FILE_ID}" \
+ -H "columns: l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag,l_linestatus, l_shipdate,l_commitdate,l_receiptdate,l_shipinstruct,l_shipmode,l_comment,temp" \
+ -T "$*" http://"${FE_HOST}":"${FE_HTTP_PORT}"/api/"${DB}"/lineitem/_stream_load
+ fi
}
# start load
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 12/12: [chore](regression) Update q03.sql (#31063)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 2af920c6f89d2b75eda599be1ac4a2795efee118
Author: Dongyang Li <he...@qq.com>
AuthorDate: Tue Feb 20 15:51:50 2024 +0800
[chore](regression) Update q03.sql (#31063)
add 'order by' of case
---
.../suites/query_p0/sql_functions/array_functions/sql/q03.sql | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/regression-test/suites/query_p0/sql_functions/array_functions/sql/q03.sql b/regression-test/suites/query_p0/sql_functions/array_functions/sql/q03.sql
index ed765d235f0..90b9eb8abbe 100644
--- a/regression-test/suites/query_p0/sql_functions/array_functions/sql/q03.sql
+++ b/regression-test/suites/query_p0/sql_functions/array_functions/sql/q03.sql
@@ -1,4 +1,4 @@
DROP TABLE IF EXISTS array_insert_select_test;
CREATE TABLE IF NOT EXISTS array_insert_select_test (id int, c_array array<int(11)>) ENGINE = Olap DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES('replication_num' = '1');
insert into array_insert_select_test select k1, collect_list(k3) from test_query_db.test group by k1;
-select c_array from array_insert_select_test;
+select c_array from array_insert_select_test order by id;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 04/12: [enhancement](sc-test) Optimize waitForSchemeChangeDone (#31002)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit fd453ace3835ccb971c1a48ace903c4de860a8ff
Author: kkop <45...@users.noreply.github.com>
AuthorDate: Tue Feb 20 10:02:21 2024 +0800
[enhancement](sc-test) Optimize waitForSchemeChangeDone (#31002)
---
.../groovy/org/apache/doris/regression/suite/Suite.groovy | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
diff --git a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index 84a84fa7a54..7d8a388b5ee 100644
--- a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++ b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -24,6 +24,7 @@ import com.google.common.util.concurrent.MoreExecutors
import com.google.gson.Gson
import groovy.json.JsonSlurper
import com.google.common.collect.ImmutableList
+import org.apache.commons.lang3.ObjectUtils
import org.apache.doris.regression.Config
import org.apache.doris.regression.action.BenchmarkAction
import org.apache.doris.regression.action.WaitForAction
@@ -522,8 +523,16 @@ class Suite implements GroovyInterceptable {
runAction(new BenchmarkAction(context), actionSupplier)
}
- void waitForSchemaChangeDone(Closure actionSupplier) {
+ void waitForSchemaChangeDone(Closure actionSupplier, String insertSql = null, boolean cleanOperator = false,String tbName=null) {
runAction(new WaitForAction(context), actionSupplier)
+ if (ObjectUtils.isNotEmpty(insertSql)){
+ sql insertSql
+ }
+ if (cleanOperator==true){
+ if (ObjectUtils.isEmpty(tbName)) throw new RuntimeException("tbName cloud not be null")
+ quickTest("", """ SELECT * FROM ${tbName} """)
+ sql """ DROP TABLE ${tbName} """
+ }
}
String getBrokerName() {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 06/12: [doc](mtmv) Update the mtmv documentation to make it consistent with the code (#30899)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 54e8354fc69e69195f2d4c5f3eb2f2f0c0180daf
Author: zhangdong <49...@qq.com>
AuthorDate: Tue Feb 20 10:59:13 2024 +0800
[doc](mtmv) Update the mtmv documentation to make it consistent with the code (#30899)
---
.../async-materialized-view/async-materialized-view.md | 5 +++++
.../Create/CREATE-ASYNC-MATERIALIZED-VIEW.md | 9 +++++----
.../async-materialized-view/async-materialized-view.md | 5 +++++
.../Create/CREATE-ASYNC-MATERIALIZED-VIEW.md | 8 +++++---
4 files changed, 20 insertions(+), 7 deletions(-)
diff --git a/docs/en/docs/query-acceleration/async-materialized-view/async-materialized-view.md b/docs/en/docs/query-acceleration/async-materialized-view/async-materialized-view.md
index a3cbf880051..e2f35c0e3f3 100644
--- a/docs/en/docs/query-acceleration/async-materialized-view/async-materialized-view.md
+++ b/docs/en/docs/query-acceleration/async-materialized-view/async-materialized-view.md
@@ -206,3 +206,8 @@ Specific syntax can be viewed [DROP MATERIALIZED VIEW](../../sql-manual/sql-refe
## The use of materialized views
can be viewed [Query async materialized view](./query-async-materialized-view.md)
+
+## Notice
+
+- Asynchronous materialized views are only supported for use in the Nereids optimizer, [Nereids optimizer](../nereids.md)
+- Currently, determining the synchronization between materialized views and base tables is only supported for `OlapTable`. For other types of external tables, they are directly considered to be synchronized. For instance, if the base tables of a materialized view are all external tables, they are assumed to be synchronized. When querying `mv_infos()`, the SyncWithBaseTables flag will always return 1 (true) for these external tables. When refreshing a materialized view, it is necessary to [...]
\ No newline at end of file
diff --git a/docs/en/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md b/docs/en/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md
index 31450326d7a..37156bc7011 100644
--- a/docs/en/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md
+++ b/docs/en/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md
@@ -160,8 +160,7 @@ KEY(k1,k2)
```
##### partition
-There are two partition methods for materialized views. If no partition is specified, there is only one partition by default. If a partition field is specified,
-it will automatically deduce which base table the field comes from and synchronize all partitions of the base table (constraint: the base table can only have one partition field and cannot allow null values)
+There are two types of partitioning methods for materialized views. If no partitioning is specified, there will be a default single partition. If a partitioning field is specified, the system will automatically deduce the source base table of that field and synchronize all partitions of the base table (currently supporting `OlapTable` and `hive`). (Limitation: the current base table can only have one partitioning field.)
For example, if the base table is a range partition with a partition field of `create_time` and partitioning by day, and `partition by(ct) as select create_time as ct from t1` is specified when creating a materialized view,
then the materialized view will also be a range partition with a partition field of 'ct' and partitioning by day
@@ -171,11 +170,13 @@ The materialized view can specify both the properties of the table and the prope
The properties unique to materialized views include:
-`grace_period`: Maximum delay time allowed for materialized view data during query rewriting
+`grace_period`: When performing query rewrites, there is a maximum allowed delay time (measured in seconds) for the data of the materialized view. If there is a discrepancy between the data of partition A and the base table, and the last refresh time of partition A of the materialized view was 1, while the current system time is 2, then this partition will not undergo transparent rewriting. However, if the grace_period is greater than or equal to 1, this partition will be used for transp [...]
`excluded_trigger_tables`: Table names ignored during data refresh, separated by commas. For example, ` table1, table2`
-`refresh_partition_num`: The number of partitions refreshed in a single insert statement, defaults to 1
+`refresh_partition_num`: The number of partitions refreshed by a single insert statement is set to 1 by default. When refreshing a materialized view, the system first calculates the list of partitions to be refreshed and then splits it into multiple insert statements that are executed in sequence according to this configuration. If any insert statement fails, the entire task will stop executing. The materialized view ensures the transactionality of individual insert statements, meaning t [...]
+
+`workload_group`: The name of the workload_group used by the materialized view when performing refresh tasks. This is used to limit the resources used for refreshing data in the materialized view, in order to avoid affecting the operation of other business processes. For details on how to create and use workload_group, refer to [WORKLOAD-GROUP](../../../../admin-manual/workload-group.md)
##### query
diff --git a/docs/zh-CN/docs/query-acceleration/async-materialized-view/async-materialized-view.md b/docs/zh-CN/docs/query-acceleration/async-materialized-view/async-materialized-view.md
index 95a684df070..6f21206f02b 100644
--- a/docs/zh-CN/docs/query-acceleration/async-materialized-view/async-materialized-view.md
+++ b/docs/zh-CN/docs/query-acceleration/async-materialized-view/async-materialized-view.md
@@ -207,3 +207,8 @@ DROP MATERIALIZED VIEW mv1;
请参阅 [查询异步物化视图](./query-async-materialized-view.md)
+## 注意事项
+
+- 异步物化视图仅支持在Nereids优化器使用，[Nereids优化器](../nereids.md)
+- 当前判断物化视图和基表是否同步仅支持`OlapTable`。对于其它外表，会直接认为是同步的。例如，物化视图的基表全是外表。在查询`mv_infos()`时，SyncWithBaseTables会永远为1（true）。在刷新物化视图时需要手动刷新指定的分区或指定`complete`刷新全部分区
+
diff --git a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md
index 7be195589a8..8add74079bf 100644
--- a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md
+++ b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-ASYNC-MATERIALIZED-VIEW.md
@@ -160,7 +160,7 @@ KEY(k1,k2)
```
##### partition
-物化视图有两种分区方式，如果不指定分区，默认只有一个分区，如果指定分区字段，会自动推导出字段来自哪个基表并同步基表的所有分区（限制条件：基表只能有一个分区字段且不能允许空值）
+物化视图有两种分区方式，如果不指定分区，默认只有一个分区，如果指定分区字段，会自动推导出字段来自哪个基表并同步基表(当前支持`OlapTable`和`hive`)的所有分区（限制条件：当前基表只能有一个分区字段）
例如：基表是range分区，分区字段为`create_time`并按天分区，创建物化视图时指定`partition by(ct) as select create_time as ct from t1`
那么物化视图也会是range分区，分区字段为`ct`,并且按天分区
@@ -170,11 +170,13 @@ KEY(k1,k2)
物化视图特有的property包括：
-`grace_period`：查询改写时允许物化视图数据的最大延迟时间
+`grace_period`：查询改写时允许物化视图数据的最大延迟时间（单位：秒）。如果分区A和基表的数据不一致，物化视图的分区A上次刷新时间为1，系统当前时间为2，那么该分区不会被透明改写。但是如果grace_period大于等于1，该分区就会被用于透明改写
`excluded_trigger_tables`：数据刷新时忽略的表名，逗号分割。例如`table1,table2`
-`refresh_partition_num`：单次insert语句刷新的分区数量，默认为1
+`refresh_partition_num`：单次insert语句刷新的分区数量，默认为1。物化视图刷新时会先计算要刷新的分区列表，然后根据该配置拆分成多个insert语句顺序执行。遇到失败的insert语句，整个任务将停止执行。物化视图保证单个insert语句的事务性，失败的insert语句不会影响到已经刷新成功的分区。
+
+`workload_group`：物化视图执行刷新任务时使用的workload_group名称。用来限制物化视图刷新数据使用的资源，避免影响到其它业务的运行。workload_group创建及使用 [WORKLOAD-GROUP](../../../../admin-manual/workload-group.md)
##### query
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 03/12: [regression-test](decimalv3) Adding numerical precision exceeding the data type will error case (#31068)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit a7a530189e2f294209ce24f4a012611d6597f06a
Author: Hong Liu <84...@qq.com>
AuthorDate: Tue Feb 20 09:58:25 2024 +0800
[regression-test](decimalv3) Adding numerical precision exceeding the data type will error case (#31068)
* Adding numerical precision exceeding the data type will result in an error case
* add order by
---------
Co-authored-by: smallhibiscus <844981280>
---
.../decimalv3/test_decimalv3_string.out | 5 ++++
.../decimalv3/test_decimalv3_string.groovy | 28 ++++++++++++++++++++++
2 files changed, 33 insertions(+)
diff --git a/regression-test/data/datatype_p0/decimalv3/test_decimalv3_string.out b/regression-test/data/datatype_p0/decimalv3/test_decimalv3_string.out
new file mode 100644
index 00000000000..2c45c69133e
--- /dev/null
+++ b/regression-test/data/datatype_p0/decimalv3/test_decimalv3_string.out
@@ -0,0 +1,5 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !decimalv3_string --
+aaa 10 22500000900.0000
+bbb 20 22500000900.0000
+
diff --git a/regression-test/suites/datatype_p0/decimalv3/test_decimalv3_string.groovy b/regression-test/suites/datatype_p0/decimalv3/test_decimalv3_string.groovy
new file mode 100644
index 00000000000..0681ae6d301
--- /dev/null
+++ b/regression-test/suites/datatype_p0/decimalv3/test_decimalv3_string.groovy
@@ -0,0 +1,28 @@
+// 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.
+
+suite("test_decimalv3_string") {
+ def db = "test_decimalv3_string"
+ sql "CREATE DATABASE IF NOT EXISTS ${db}"
+ sql "use ${db}"
+ sql "drop table if exists replicationtesttable"
+ sql '''CREATE TABLE `replicationtesttable` ( `k1` varchar(144) NOT NULL COMMENT '_key__', `k2` int(11) NOT NULL COMMENT '_key__', v4 decimal(18,4) null, ) ENGINE=OLAP UNIQUE KEY(`k1`, `k2`) COMMENT 'OLAP' DISTRIBUTED BY HASH(`k1`) BUCKETS 32 PROPERTIES ( "replication_allocation" = "tag.location.default: 1", "is_being_synced" = "false", "storage_format" = "V2", "light_schema_change" = "true", "disable_auto_compaction" = "false", "enable_single_replica_compaction" = "false" ); '''
+ sql "insert into replicationtesttable (k1,k2,v4) select 'aaa',10,22500000900.000000000000000000000000000000"
+ sql "insert into replicationtesttable (k1,k2,v4) select 'bbb',20,'22500000900.000000000000000000000000000000'"
+
+ qt_decimalv3_string "select * from replicationtesttable order by 1"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 07/12: [fix](statistics)Refresh follower FE cache after alter column stats. Support alter index column stats (#31108)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 4aaab6fb441433c2db0ace7564e2c06aa732f906
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Tue Feb 20 11:52:11 2024 +0800
[fix](statistics)Refresh follower FE cache after alter column stats. Support alter index column stats (#31108)
1. Refresh follower FE cache after alter column stats. So that follower could update the cached stats too.
2. Support alter index column stats.
---
fe/fe-core/src/main/cup/sql_parser.cup | 7 +++-
.../doris/analysis/AlterColumnStatsStmt.java | 27 +++++++++++++-
.../org/apache/doris/statistics/ColStatsData.java | 12 +++++++
.../doris/statistics/StatisticsRepository.java | 11 +++---
.../suites/statistics/test_analyze_mv.groovy | 42 ++++++++++++++++++++++
5 files changed, 93 insertions(+), 6 deletions(-)
diff --git a/fe/fe-core/src/main/cup/sql_parser.cup b/fe/fe-core/src/main/cup/sql_parser.cup
index ec3bba7996b..798960387cd 100644
--- a/fe/fe-core/src/main/cup/sql_parser.cup
+++ b/fe/fe-core/src/main/cup/sql_parser.cup
@@ -1403,7 +1403,12 @@ alter_stmt ::=
| KW_ALTER KW_TABLE table_name:tbl KW_MODIFY KW_COLUMN ident:columnName
KW_SET KW_STATS LPAREN key_value_map:map RPAREN opt_partition_names:partitionNames
{:
- RESULT = new AlterColumnStatsStmt(tbl, columnName, map, partitionNames);
+ RESULT = new AlterColumnStatsStmt(tbl, null, columnName, map, partitionNames);
+ :}
+ | KW_ALTER KW_TABLE table_name:tbl KW_INDEX ident:idx KW_MODIFY KW_COLUMN ident:columnName
+ KW_SET KW_STATS LPAREN key_value_map:map RPAREN opt_partition_names:partitionNames
+ {:
+ RESULT = new AlterColumnStatsStmt(tbl, idx, columnName, map, partitionNames);
:}
| KW_ALTER KW_TABLE table_name:tbl KW_SET LPAREN key_value_map:properties RPAREN
{:
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterColumnStatsStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterColumnStatsStmt.java
index 7780f7833f7..ef4357dc285 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterColumnStatsStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterColumnStatsStmt.java
@@ -73,6 +73,7 @@ public class AlterColumnStatsStmt extends DdlStmt {
.build();
private final TableName tableName;
+ private final String indexName;
private final String columnName;
private final Map<String, String> properties;
private final PartitionNames optPartitionNames;
@@ -80,9 +81,12 @@ public class AlterColumnStatsStmt extends DdlStmt {
private final List<Long> partitionIds = Lists.newArrayList();
private final Map<StatsType, String> statsTypeToValue = Maps.newHashMap();
- public AlterColumnStatsStmt(TableName tableName, String columnName,
+ private long indexId = -1;
+
+ public AlterColumnStatsStmt(TableName tableName, String indexName, String columnName,
Map<String, String> properties, PartitionNames optPartitionNames) {
this.tableName = tableName;
+ this.indexName = indexName;
this.columnName = columnName;
this.properties = properties == null ? Collections.emptyMap() : properties;
this.optPartitionNames = optPartitionNames;
@@ -96,6 +100,10 @@ public class AlterColumnStatsStmt extends DdlStmt {
return columnName;
}
+ public long getIndexId() {
+ return indexId;
+ }
+
public List<Long> getPartitionIds() {
return partitionIds;
}
@@ -148,6 +156,19 @@ public class AlterColumnStatsStmt extends DdlStmt {
DatabaseIf db = catalog.getDbOrAnalysisException(tableName.getDb());
TableIf table = db.getTableOrAnalysisException(tableName.getTbl());
+ if (indexName != null) {
+ if (!(table instanceof OlapTable)) {
+ throw new AnalysisException("Only OlapTable support alter index stats. "
+ + "Table " + table.getName() + " is not OlapTable.");
+ }
+ OlapTable olapTable = (OlapTable) table;
+ Long idxId = olapTable.getIndexIdByName(indexName);
+ if (idxId == null) {
+ throw new AnalysisException("Index " + indexName + " not exist in table " + table.getName());
+ }
+ indexId = idxId;
+ }
+
if (table.getColumn(columnName) == null) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_WRONG_COLUMN_NAME,
columnName, FeNameFormat.getColumnNameRegex());
@@ -176,6 +197,10 @@ public class AlterColumnStatsStmt extends DdlStmt {
StringBuilder sb = new StringBuilder();
sb.append("ALTER TABLE ");
sb.append(tableName.toSql());
+ if (indexName != null) {
+ sb.append(" INDEX ");
+ sb.append(indexName);
+ }
sb.append(" MODIFY COLUMN ");
sb.append(columnName);
sb.append(" SET STATS ");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsData.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsData.java
index bdc600987f4..c90b3dd8e1d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsData.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsData.java
@@ -101,6 +101,18 @@ public class ColStatsData {
this.updateTime = row.get(13);
}
+ public ColStatsData(String id, long catalogId, long dbId, long tblId, long idxId, String colId, String partId,
+ ColumnStatistic columnStatistic) {
+ this.statsId = new StatsId(id, catalogId, dbId, tblId, idxId, colId, partId);
+ this.count = Math.round(columnStatistic.count);
+ this.ndv = Math.round(columnStatistic.ndv);
+ this.nullCount = Math.round(columnStatistic.numNulls);
+ this.minLit = columnStatistic.minExpr == null ? null : columnStatistic.minExpr.getStringValue();
+ this.maxLit = columnStatistic.maxExpr == null ? null : columnStatistic.maxExpr.getStringValue();
+ this.dataSizeInBytes = Math.round(columnStatistic.dataSize);
+ this.updateTime = columnStatistic.updatedTime;
+ }
+
public String toSQL(boolean roundByParentheses) {
StringJoiner sj = null;
if (roundByParentheses) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsRepository.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsRepository.java
index 92ff3cd42b4..c8db624eeb3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsRepository.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsRepository.java
@@ -247,6 +247,7 @@ public class StatisticsRepository {
String min = alterColumnStatsStmt.getValue(StatsType.MIN_VALUE);
String max = alterColumnStatsStmt.getValue(StatsType.MAX_VALUE);
String dataSize = alterColumnStatsStmt.getValue(StatsType.DATA_SIZE);
+ long indexId = alterColumnStatsStmt.getIndexId();
ColumnStatisticBuilder builder = new ColumnStatisticBuilder();
String colName = alterColumnStatsStmt.getColumnName();
Column column = objects.table.getColumn(colName);
@@ -282,10 +283,10 @@ public class StatisticsRepository {
ColumnStatistic columnStatistic = builder.build();
Map<String, String> params = new HashMap<>();
- params.put("id", constructId(objects.table.getId(), -1, colName));
+ params.put("id", constructId(objects.table.getId(), indexId, colName));
params.put("catalogId", String.valueOf(objects.catalog.getId()));
params.put("dbId", String.valueOf(objects.db.getId()));
- params.put("idxId", "-1");
+ params.put("idxId", String.valueOf(indexId));
params.put("tblId", String.valueOf(objects.table.getId()));
params.put("colId", String.valueOf(colName));
params.put("count", String.valueOf(columnStatistic.count));
@@ -299,8 +300,10 @@ public class StatisticsRepository {
// update table granularity statistics
params.put("partId", "NULL");
StatisticsUtil.execUpdate(INSERT_INTO_COLUMN_STATISTICS, params);
- Env.getCurrentEnv().getStatisticsCache()
- .updateColStatsCache(objects.table.getId(), -1, colName, columnStatistic);
+ ColStatsData data = new ColStatsData(constructId(objects.table.getId(), indexId, colName),
+ objects.catalog.getId(), objects.db.getId(), objects.table.getId(), indexId, colName,
+ null, columnStatistic);
+ Env.getCurrentEnv().getStatisticsCache().syncColStats(data);
AnalysisInfo mockedJobInfo = new AnalysisInfoBuilder()
.setTblUpdateTime(System.currentTimeMillis())
.setColName("")
diff --git a/regression-test/suites/statistics/test_analyze_mv.groovy b/regression-test/suites/statistics/test_analyze_mv.groovy
index 3a29cafafbd..d868b02768d 100644
--- a/regression-test/suites/statistics/test_analyze_mv.groovy
+++ b/regression-test/suites/statistics/test_analyze_mv.groovy
@@ -410,6 +410,48 @@ suite("test_analyze_mv") {
verifyTaskStatus(result_sample, "mva_MIN__`value3`", "mv3")
verifyTaskStatus(result_sample, "mva_SUM__CAST(`value1` AS BIGINT)", "mv3")
+ // Test alter column stats
+ sql """drop stats mvTestDup"""
+ sql """alter table mvTestDup modify column key1 set stats ('ndv'='1', 'num_nulls'='1', 'min_value'='10', 'max_value'='40', 'row_count'='50');"""
+ sql """alter table mvTestDup index mv3 modify column mv_key1 set stats ('ndv'='5', 'num_nulls'='0', 'min_value'='0', 'max_value'='4', 'row_count'='5');"""
+ sql """alter table mvTestDup index mv3 modify column `mva_SUM__CAST(``value1`` AS BIGINT)` set stats ('ndv'='10', 'num_nulls'='2', 'min_value'='1', 'max_value'='5', 'row_count'='11');"""
+
+ def result = sql """show column cached stats mvTestDup(key1)"""
+ assertEquals(1, result.size())
+ assertEquals("key1", result[0][0])
+ assertEquals("N/A", result[0][1])
+ assertEquals("50.0", result[0][2])
+ assertEquals("1.0", result[0][3])
+ assertEquals("1.0", result[0][4])
+ assertEquals("0.0", result[0][5])
+ assertEquals("0.0", result[0][6])
+ assertEquals("10", result[0][7])
+ assertEquals("40", result[0][8])
+
+ result = sql """show column cached stats mvTestDup(mv_key1)"""
+ assertEquals(1, result.size())
+ assertEquals("mv_key1", result[0][0])
+ assertEquals("mv3", result[0][1])
+ assertEquals("5.0", result[0][2])
+ assertEquals("5.0", result[0][3])
+ assertEquals("0.0", result[0][4])
+ assertEquals("0.0", result[0][5])
+ assertEquals("0.0", result[0][6])
+ assertEquals("0", result[0][7])
+ assertEquals("4", result[0][8])
+
+ result = sql """show column cached stats mvTestDup(`mva_SUM__CAST(``value1`` AS BIGINT)`)"""
+ assertEquals(1, result.size())
+ assertEquals("mva_SUM__CAST(`value1` AS BIGINT)", result[0][0])
+ assertEquals("mv3", result[0][1])
+ assertEquals("11.0", result[0][2])
+ assertEquals("10.0", result[0][3])
+ assertEquals("2.0", result[0][4])
+ assertEquals("0.0", result[0][5])
+ assertEquals("0.0", result[0][6])
+ assertEquals("1", result[0][7])
+ assertEquals("5", result[0][8])
+
sql """drop database if exists test_analyze_mv"""
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 09/12: [fix](Nereids): log constraint editlog in lock scope (#30630)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 92518166738158d689df2cc072ac32343970c655
Author: 谢健 <ji...@gmail.com>
AuthorDate: Tue Feb 20 14:39:39 2024 +0800
[fix](Nereids): log constraint editlog in lock scope (#30630)
---
.../java/org/apache/doris/catalog/TableIf.java | 113 +++++++++++++--------
.../trees/plans/commands/AddConstraintCommand.java | 13 +--
.../plans/commands/DropConstraintCommand.java | 5 +-
.../java/org/apache/doris/persist/EditLog.java | 2 +-
.../catalog/constraint/ConstraintPersistTest.java | 55 +++++++++-
5 files changed, 127 insertions(+), 61 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java
index 7c1943c69a6..29c7d6b83e2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java
@@ -26,6 +26,7 @@ import org.apache.doris.cluster.ClusterNamespace;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.persist.AlterConstraintLog;
import org.apache.doris.statistics.AnalysisInfo;
import org.apache.doris.statistics.BaseAnalysisTask;
import org.apache.doris.statistics.ColumnStatistic;
@@ -227,7 +228,7 @@ public interface TableIf {
}
// Note this function is not thread safe
- default void checkConstraintNotExistence(String name, Constraint primaryKeyConstraint,
+ default void checkConstraintNotExistenceUnsafe(String name, Constraint primaryKeyConstraint,
Map<String, Constraint> constraintMap) {
if (constraintMap.containsKey(name)) {
throw new RuntimeException(String.format("Constraint name %s has existed", name));
@@ -240,87 +241,108 @@ public interface TableIf {
}
}
- default Constraint addUniqueConstraint(String name, ImmutableList<String> columns) {
+ default void addUniqueConstraint(String name, ImmutableList<String> columns, boolean replay) {
writeLock();
try {
Map<String, Constraint> constraintMap = getConstraintsMapUnsafe();
UniqueConstraint uniqueConstraint = new UniqueConstraint(name, ImmutableSet.copyOf(columns));
- checkConstraintNotExistence(name, uniqueConstraint, constraintMap);
+ checkConstraintNotExistenceUnsafe(name, uniqueConstraint, constraintMap);
constraintMap.put(name, uniqueConstraint);
- return uniqueConstraint;
+ if (!replay) {
+ Env.getCurrentEnv().getEditLog().logAddConstraint(
+ new AlterConstraintLog(uniqueConstraint, this));
+ }
} finally {
writeUnlock();
}
}
- default Constraint addPrimaryKeyConstraint(String name, ImmutableList<String> columns) {
+ default void addPrimaryKeyConstraint(String name, ImmutableList<String> columns, boolean replay) {
writeLock();
try {
Map<String, Constraint> constraintMap = getConstraintsMapUnsafe();
PrimaryKeyConstraint primaryKeyConstraint = new PrimaryKeyConstraint(name, ImmutableSet.copyOf(columns));
- checkConstraintNotExistence(name, primaryKeyConstraint, constraintMap);
+ checkConstraintNotExistenceUnsafe(name, primaryKeyConstraint, constraintMap);
constraintMap.put(name, primaryKeyConstraint);
- return primaryKeyConstraint;
+ if (!replay) {
+ Env.getCurrentEnv().getEditLog().logAddConstraint(
+ new AlterConstraintLog(primaryKeyConstraint, this));
+ }
} finally {
writeUnlock();
}
}
- default void updatePrimaryKeyForForeignKey(PrimaryKeyConstraint requirePrimaryKey, TableIf referencedTable) {
- referencedTable.writeLock();
- try {
- Optional<Constraint> primaryKeyConstraint = referencedTable.getConstraintsMapUnsafe().values().stream()
- .filter(requirePrimaryKey::equals)
- .findFirst();
- if (!primaryKeyConstraint.isPresent()) {
- throw new AnalysisException(String.format(
- "Foreign key constraint requires a primary key constraint %s in %s",
- requirePrimaryKey.getPrimaryKeyNames(), referencedTable.getName()));
- }
- ((PrimaryKeyConstraint) (primaryKeyConstraint.get())).addForeignTable(this);
- } finally {
- referencedTable.writeUnlock();
+ default PrimaryKeyConstraint tryGetPrimaryKeyForForeignKeyUnsafe(
+ PrimaryKeyConstraint requirePrimaryKey, TableIf referencedTable) {
+ Optional<Constraint> primaryKeyConstraint = referencedTable.getConstraintsMapUnsafe().values().stream()
+ .filter(requirePrimaryKey::equals)
+ .findFirst();
+ if (!primaryKeyConstraint.isPresent()) {
+ throw new AnalysisException(String.format(
+ "Foreign key constraint requires a primary key constraint %s in %s",
+ requirePrimaryKey.getPrimaryKeyNames(), referencedTable.getName()));
}
+ return ((PrimaryKeyConstraint) (primaryKeyConstraint.get()));
}
- default Constraint addForeignConstraint(String name, ImmutableList<String> columns,
- TableIf referencedTable, ImmutableList<String> referencedColumns) {
+ default void addForeignConstraint(String name, ImmutableList<String> columns,
+ TableIf referencedTable, ImmutableList<String> referencedColumns, boolean replay) {
writeLock();
+ referencedTable.writeLock();
try {
Map<String, Constraint> constraintMap = getConstraintsMapUnsafe();
ForeignKeyConstraint foreignKeyConstraint =
new ForeignKeyConstraint(name, columns, referencedTable, referencedColumns);
- checkConstraintNotExistence(name, foreignKeyConstraint, constraintMap);
- PrimaryKeyConstraint requirePrimaryKey = new PrimaryKeyConstraint(name,
+ checkConstraintNotExistenceUnsafe(name, foreignKeyConstraint, constraintMap);
+ PrimaryKeyConstraint requirePrimaryKeyName = new PrimaryKeyConstraint(name,
foreignKeyConstraint.getReferencedColumnNames());
- updatePrimaryKeyForForeignKey(requirePrimaryKey, referencedTable);
+ PrimaryKeyConstraint primaryKeyConstraint =
+ tryGetPrimaryKeyForForeignKeyUnsafe(requirePrimaryKeyName, referencedTable);
+ primaryKeyConstraint.addForeignTable(this);
constraintMap.put(name, foreignKeyConstraint);
- return foreignKeyConstraint;
+ if (!replay) {
+ Env.getCurrentEnv().getEditLog().logAddConstraint(
+ new AlterConstraintLog(foreignKeyConstraint, this));
+ }
} finally {
+ referencedTable.writeUnlock();
writeUnlock();
}
}
default void replayAddConstraint(Constraint constraint) {
- if (constraint instanceof UniqueConstraint) {
- UniqueConstraint uniqueConstraint = (UniqueConstraint) constraint;
- this.addUniqueConstraint(constraint.getName(),
- ImmutableList.copyOf(uniqueConstraint.getUniqueColumnNames()));
- } else if (constraint instanceof PrimaryKeyConstraint) {
- PrimaryKeyConstraint primaryKeyConstraint = (PrimaryKeyConstraint) constraint;
- this.addPrimaryKeyConstraint(primaryKeyConstraint.getName(),
- ImmutableList.copyOf(primaryKeyConstraint.getPrimaryKeyNames()));
- } else if (constraint instanceof ForeignKeyConstraint) {
- ForeignKeyConstraint foreignKey = (ForeignKeyConstraint) constraint;
- this.addForeignConstraint(foreignKey.getName(),
- ImmutableList.copyOf(foreignKey.getForeignKeyNames()),
- foreignKey.getReferencedTable(),
- ImmutableList.copyOf(foreignKey.getReferencedColumnNames()));
+ // Since constraints are not indispensable, we only log when replay fails
+ try {
+ if (constraint instanceof UniqueConstraint) {
+ UniqueConstraint uniqueConstraint = (UniqueConstraint) constraint;
+ this.addUniqueConstraint(constraint.getName(),
+ ImmutableList.copyOf(uniqueConstraint.getUniqueColumnNames()), true);
+ } else if (constraint instanceof PrimaryKeyConstraint) {
+ PrimaryKeyConstraint primaryKeyConstraint = (PrimaryKeyConstraint) constraint;
+ this.addPrimaryKeyConstraint(primaryKeyConstraint.getName(),
+ ImmutableList.copyOf(primaryKeyConstraint.getPrimaryKeyNames()), true);
+ } else if (constraint instanceof ForeignKeyConstraint) {
+ ForeignKeyConstraint foreignKey = (ForeignKeyConstraint) constraint;
+ this.addForeignConstraint(foreignKey.getName(),
+ ImmutableList.copyOf(foreignKey.getForeignKeyNames()),
+ foreignKey.getReferencedTable(),
+ ImmutableList.copyOf(foreignKey.getReferencedColumnNames()), true);
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage());
}
}
- default Constraint dropConstraint(String name) {
- Constraint dropConstraint;
+ default void replayDropConstraint(String name) {
+ try {
+ dropConstraint(name, true);
+ } catch (Exception e) {
+ LOG.error(e.getMessage());
+ }
+ }
+
+ default void dropConstraint(String name, boolean replay) {
writeLock();
try {
Map<String, Constraint> constraintMap = getConstraintsMapUnsafe();
@@ -334,11 +356,12 @@ public interface TableIf {
((PrimaryKeyConstraint) constraint).getForeignTables()
.forEach(t -> t.dropFKReferringPK(this, (PrimaryKeyConstraint) constraint));
}
- dropConstraint = constraint;
+ if (!replay) {
+ Env.getCurrentEnv().getEditLog().logDropConstraint(new AlterConstraintLog(constraint, this));
+ }
} finally {
writeUnlock();
}
- return dropConstraint;
}
default void dropFKReferringPK(TableIf table, PrimaryKeyConstraint constraint) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java
index 8c90bc0f914..38bf71133c0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java
@@ -17,7 +17,6 @@
package org.apache.doris.nereids.trees.plans.commands;
-import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Pair;
import org.apache.doris.nereids.NereidsPlanner;
@@ -30,7 +29,6 @@ import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel
import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
-import org.apache.doris.persist.AlterConstraintLog;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
@@ -63,19 +61,16 @@ public class AddConstraintCommand extends Command implements ForwardWithSync {
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
Pair<ImmutableList<String>, TableIf> columnsAndTable = extractColumnsAndTable(ctx, constraint.toProject());
- org.apache.doris.catalog.constraint.Constraint catalogConstraint = null;
if (constraint.isForeignKey()) {
Pair<ImmutableList<String>, TableIf> referencedColumnsAndTable
= extractColumnsAndTable(ctx, constraint.toReferenceProject());
- catalogConstraint = columnsAndTable.second.addForeignConstraint(name, columnsAndTable.first,
- referencedColumnsAndTable.second, referencedColumnsAndTable.first);
+ columnsAndTable.second.addForeignConstraint(name, columnsAndTable.first,
+ referencedColumnsAndTable.second, referencedColumnsAndTable.first, false);
} else if (constraint.isPrimaryKey()) {
- catalogConstraint = columnsAndTable.second.addPrimaryKeyConstraint(name, columnsAndTable.first);
+ columnsAndTable.second.addPrimaryKeyConstraint(name, columnsAndTable.first, false);
} else if (constraint.isUnique()) {
- catalogConstraint = columnsAndTable.second.addUniqueConstraint(name, columnsAndTable.first);
+ columnsAndTable.second.addUniqueConstraint(name, columnsAndTable.first, false);
}
- Env.getCurrentEnv().getEditLog().logAddConstraint(
- new AlterConstraintLog(catalogConstraint, columnsAndTable.second));
}
private Pair<ImmutableList<String>, TableIf> extractColumnsAndTable(ConnectContext ctx, LogicalPlan plan) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java
index 84143f234c3..9878b0e9263 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java
@@ -17,7 +17,6 @@
package org.apache.doris.nereids.trees.plans.commands;
-import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.exceptions.AnalysisException;
@@ -28,7 +27,6 @@ import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel
import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
-import org.apache.doris.persist.AlterConstraintLog;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
@@ -58,8 +56,7 @@ public class DropConstraintCommand extends Command implements ForwardWithSync {
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
TableIf table = extractTable(ctx, plan);
- org.apache.doris.catalog.constraint.Constraint catalogConstraint = table.dropConstraint(name);
- Env.getCurrentEnv().getEditLog().logDropConstraint(new AlterConstraintLog(catalogConstraint, table));
+ table.dropConstraint(name, false);
}
private TableIf extractTable(ConnectContext ctx, LogicalPlan plan) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
index 7f343f94692..88675d5cca3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
@@ -992,7 +992,7 @@ public class EditLog {
}
case OperationType.OP_DROP_CONSTRAINT: {
final AlterConstraintLog log = (AlterConstraintLog) journal.getData();
- log.getTableIf().dropConstraint(log.getConstraint().getName());
+ log.getTableIf().replayDropConstraint(log.getConstraint().getName());
break;
}
case OperationType.OP_ALTER_USER: {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/constraint/ConstraintPersistTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/constraint/ConstraintPersistTest.java
index 782073d4ac3..07c7abf7ce0 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/constraint/ConstraintPersistTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/constraint/ConstraintPersistTest.java
@@ -24,6 +24,7 @@ import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Config;
+import org.apache.doris.common.Pair;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.es.EsExternalCatalog;
import org.apache.doris.datasource.es.EsExternalDatabase;
@@ -53,6 +54,7 @@ import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
class ConstraintPersistTest extends TestWithFeService implements PlanPatternMatchSupported {
@@ -144,6 +146,52 @@ class ConstraintPersistTest extends TestWithFeService implements PlanPatternMatc
Assertions.assertTrue(tableIf.getConstraintsMap().isEmpty());
}
+ @Test
+ void replayDropConstraintLogTest() throws Exception {
+ Config.edit_log_type = "local";
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ DataOutput output = new DataOutputStream(outputStream);
+ List<Pair<Short, AlterConstraintLog>> logs = new CopyOnWriteArrayList<>();
+ EditLog editLog = new EditLog("");
+ new MockUp<EditLog>() {
+ @Mock
+ public void logAddConstraint(AlterConstraintLog log) {
+ logs.add(Pair.of(OperationType.OP_ADD_CONSTRAINT, log));
+ }
+
+ @Mock
+ public void logDropConstraint(AlterConstraintLog log) {
+ logs.add(Pair.of(OperationType.OP_DROP_CONSTRAINT, log));
+ }
+ };
+ new MockUp<Env>() {
+ @Mock
+ public EditLog getEditLog() {
+ return editLog;
+ }
+ };
+ addConstraint("alter table t1 add constraint pk primary key (k1)");
+ addConstraint("alter table t2 add constraint pk primary key (k1)");
+ addConstraint("alter table t1 add constraint uk unique (k1)");
+ addConstraint("alter table t1 add constraint fk foreign key (k1) references t2(k1)");
+ TableIf tableIf = RelationUtil.getTable(
+ RelationUtil.getQualifierName(connectContext, Lists.newArrayList("test", "t1")),
+ connectContext.getEnv());
+ Assertions.assertEquals(3, tableIf.getConstraintsMap().size());
+ dropConstraint("alter table t1 drop constraint uk");
+ dropConstraint("alter table t1 drop constraint pk");
+ dropConstraint("alter table t2 drop constraint pk");
+ Assertions.assertEquals(0, tableIf.getConstraintsMap().size());
+ for (Pair<Short, AlterConstraintLog> log : logs) {
+ JournalEntity journalEntity = new JournalEntity();
+ journalEntity.setData(log.second);
+ journalEntity.setOpCode(log.first);
+ journalEntity.write(output);
+ }
+ Assertions.assertEquals(0, tableIf.getConstraintsMap().size());
+ Assertions.assertEquals(0, tableIf.getConstraintsMap().size());
+ }
+
@Test
void constraintWithTablePersistTest() throws Exception {
addConstraint("alter table t1 add constraint pk primary key (k1)");
@@ -169,8 +217,11 @@ class ConstraintPersistTest extends TestWithFeService implements PlanPatternMatc
@Test
void externalTableTest() throws Exception {
ExternalTable externalTable = new ExternalTable();
- externalTable.addPrimaryKeyConstraint("pk", ImmutableList.of("col"));
-
+ try {
+ externalTable.addPrimaryKeyConstraint("pk", ImmutableList.of("col"), false);
+ } catch (Exception ignore) {
+ // ignore
+ }
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
DataOutput output = new DataOutputStream(outputStream);
externalTable.write(output);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 11/12: [Feature](executor)Add scan_thread_num property for workload group (#31106)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 97c9d75af346532314c207beb11f24259d0a4884
Author: wangbo <wa...@apache.org>
AuthorDate: Tue Feb 20 15:16:30 2024 +0800
[Feature](executor)Add scan_thread_num property for workload group (#31106)
---
be/src/runtime/task_group/task_group.cpp | 15 ++++++++--
be/src/runtime/task_group/task_group.h | 2 ++
be/src/runtime/task_group/task_group_manager.cpp | 4 +++
be/src/vec/exec/scan/scanner_scheduler.cpp | 2 +-
be/src/vec/exec/scan/scanner_scheduler.h | 35 +++++++++++-----------
docs/en/docs/admin-manual/workload-group.md | 1 +
docs/zh-CN/docs/admin-manual/workload-group.md | 1 +
.../resource/workloadgroup/WorkloadGroup.java | 24 ++++++++++++++-
.../resource/workloadgroup/WorkloadGroupMgr.java | 1 +
.../doris/tablefunction/MetadataGenerator.java | 5 ++--
.../WorkloadGroupsTableValuedFunction.java | 1 +
gensrc/thrift/BackendService.thrift | 1 +
.../data/workload_manager_p0/test_curd_wlg.out | 24 +++++++--------
.../workload_manager_p0/test_curd_wlg.groovy | 20 +++++++++----
14 files changed, 94 insertions(+), 42 deletions(-)
diff --git a/be/src/runtime/task_group/task_group.cpp b/be/src/runtime/task_group/task_group.cpp
index e0b0dc1fb5e..ee1be702768 100644
--- a/be/src/runtime/task_group/task_group.cpp
+++ b/be/src/runtime/task_group/task_group.cpp
@@ -51,15 +51,17 @@ TaskGroup::TaskGroup(const TaskGroupInfo& tg_info)
_enable_memory_overcommit(tg_info.enable_memory_overcommit),
_cpu_share(tg_info.cpu_share),
_mem_tracker_limiter_pool(MEM_TRACKER_GROUP_NUM),
- _cpu_hard_limit(tg_info.cpu_hard_limit) {}
+ _cpu_hard_limit(tg_info.cpu_hard_limit),
+ _scan_thread_num(tg_info.scan_thread_num) {}
std::string TaskGroup::debug_string() const {
std::shared_lock<std::shared_mutex> rl {_mutex};
return fmt::format(
"TG[id = {}, name = {}, cpu_share = {}, memory_limit = {}, enable_memory_overcommit = "
- "{}, version = {}, cpu_hard_limit = {}]",
+ "{}, version = {}, cpu_hard_limit = {}, scan_thread_num = {}]",
_id, _name, cpu_share(), PrettyPrinter::print(_memory_limit, TUnit::BYTES),
- _enable_memory_overcommit ? "true" : "false", _version, cpu_hard_limit());
+ _enable_memory_overcommit ? "true" : "false", _version, cpu_hard_limit(),
+ _scan_thread_num);
}
void TaskGroup::check_and_update(const TaskGroupInfo& tg_info) {
@@ -81,6 +83,7 @@ void TaskGroup::check_and_update(const TaskGroupInfo& tg_info) {
_enable_memory_overcommit = tg_info.enable_memory_overcommit;
_cpu_share = tg_info.cpu_share;
_cpu_hard_limit = tg_info.cpu_hard_limit;
+ _scan_thread_num = tg_info.scan_thread_num;
} else {
return;
}
@@ -185,6 +188,12 @@ Status TaskGroupInfo::parse_topic_info(const TWorkloadGroupInfo& workload_group_
}
task_group_info->enable_cpu_hard_limit = enable_cpu_hard_limit;
+ // 9 scan thread num
+ task_group_info->scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.scan_thread_num && workload_group_info.scan_thread_num > 0) {
+ task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
+ }
+
return Status::OK();
}
diff --git a/be/src/runtime/task_group/task_group.h b/be/src/runtime/task_group/task_group.h
index 647d088e817..3767731435a 100644
--- a/be/src/runtime/task_group/task_group.h
+++ b/be/src/runtime/task_group/task_group.h
@@ -134,6 +134,7 @@ private:
std::atomic<uint64_t> _cpu_share;
std::vector<TgTrackerLimiterGroup> _mem_tracker_limiter_pool;
std::atomic<int> _cpu_hard_limit;
+ std::atomic<int> _scan_thread_num;
// means task group is mark dropped
// new query can not submit
@@ -153,6 +154,7 @@ struct TaskGroupInfo {
int64_t version;
int cpu_hard_limit;
bool enable_cpu_hard_limit;
+ int scan_thread_num;
// log cgroup cpu info
uint64_t cgroup_cpu_shares = 0;
int cgroup_cpu_hard_limit = 0;
diff --git a/be/src/runtime/task_group/task_group_manager.cpp b/be/src/runtime/task_group/task_group_manager.cpp
index 05be653747d..18e446295cb 100644
--- a/be/src/runtime/task_group/task_group_manager.cpp
+++ b/be/src/runtime/task_group/task_group_manager.cpp
@@ -96,6 +96,7 @@ Status TaskGroupManager::upsert_cg_task_scheduler(taskgroup::TaskGroupInfo* tg_i
int cpu_hard_limit = tg_info->cpu_hard_limit;
uint64_t cpu_shares = tg_info->cpu_share;
bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
std::lock_guard<std::shared_mutex> write_lock(_task_scheduler_lock);
// step 1: init cgroup cpu controller
@@ -142,6 +143,9 @@ Status TaskGroupManager::upsert_cg_task_scheduler(taskgroup::TaskGroupInfo* tg_i
return Status::InternalError<false>("scan scheduler start failed, gid={}", tg_id);
}
}
+ if (scan_thread_num > 0 && _tg_scan_sche_map.find(tg_id) != _tg_scan_sche_map.end()) {
+ _tg_scan_sche_map.at(tg_id)->reset_thread_num(scan_thread_num);
+ }
// step 4: init non-pipe scheduler
if (_non_pipe_thread_pool_map.find(tg_id) == _non_pipe_thread_pool_map.end()) {
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
index 40fff7ed70c..d3fd740b437 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -162,7 +162,7 @@ void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
this->_scanner_scan(ctx, scanner_ref);
};
SimplifiedScanTask simple_scan_task = {work_func, ctx};
- ret = scan_sche->get_scan_queue()->try_put(simple_scan_task);
+ ret = scan_sche->submit_scan_task(simple_scan_task);
} else {
PriorityThreadPool::Task task;
task.work_function = [this, scanner_ref = scan_task, ctx]() {
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h
index 7a602038956..746aa34ff9a 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -99,12 +99,9 @@ struct SimplifiedScanTask {
std::shared_ptr<vectorized::ScannerContext> scanner_context = nullptr;
};
-// used for cpu hard limit
class SimplifiedScanScheduler {
public:
SimplifiedScanScheduler(std::string wg_name, CgroupCpuCtl* cgroup_cpu_ctl) {
- _scan_task_queue = std::make_unique<BlockingQueue<SimplifiedScanTask>>(
- config::doris_scanner_thread_pool_queue_size);
_is_stop.store(false);
_cgroup_cpu_ctl = cgroup_cpu_ctl;
_wg_name = wg_name;
@@ -117,7 +114,6 @@ public:
void stop() {
_is_stop.store(true);
- _scan_task_queue->shutdown();
_scan_thread_pool->shutdown();
_scan_thread_pool->wait();
}
@@ -128,27 +124,32 @@ public:
.set_max_threads(config::doris_scanner_thread_pool_thread_num)
.set_cgroup_cpu_ctl(_cgroup_cpu_ctl)
.build(&_scan_thread_pool));
-
- for (int i = 0; i < config::doris_scanner_thread_pool_thread_num; i++) {
- RETURN_IF_ERROR(_scan_thread_pool->submit_func([this] { this->_work(); }));
- }
return Status::OK();
}
- BlockingQueue<SimplifiedScanTask>* get_scan_queue() { return _scan_task_queue.get(); }
+ Status submit_scan_task(SimplifiedScanTask scan_task) {
+ if (!_is_stop) {
+ return _scan_thread_pool->submit_func([scan_task] { scan_task.scan_func(); });
+ } else {
+ return Status::InternalError<false>("scanner pool {} is shutdown.", _wg_name);
+ }
+ }
-private:
- void _work() {
- while (!_is_stop.load()) {
- SimplifiedScanTask scan_task;
- if (_scan_task_queue->blocking_get(&scan_task)) {
- scan_task.scan_func();
- };
+ void reset_thread_num(int thread_num) {
+ int max_thread_num = _scan_thread_pool->max_threads();
+ if (max_thread_num != thread_num) {
+ if (thread_num > max_thread_num) {
+ static_cast<void>(_scan_thread_pool->set_max_threads(thread_num));
+ static_cast<void>(_scan_thread_pool->set_min_threads(thread_num));
+ } else {
+ static_cast<void>(_scan_thread_pool->set_min_threads(thread_num));
+ static_cast<void>(_scan_thread_pool->set_max_threads(thread_num));
+ }
}
}
+private:
std::unique_ptr<ThreadPool> _scan_thread_pool;
- std::unique_ptr<BlockingQueue<SimplifiedScanTask>> _scan_task_queue;
std::atomic<bool> _is_stop;
CgroupCpuCtl* _cgroup_cpu_ctl = nullptr;
std::string _wg_name;
diff --git a/docs/en/docs/admin-manual/workload-group.md b/docs/en/docs/admin-manual/workload-group.md
index 07db184c4e3..93216a91fb9 100644
--- a/docs/en/docs/admin-manual/workload-group.md
+++ b/docs/en/docs/admin-manual/workload-group.md
@@ -58,6 +58,7 @@ Scenario 2: If the Workload Group is not used in version 2.0, it is also necessa
* max_concurrency: Optional, maximum query concurrency, default value is the maximum integer value, which means there is no concurrency limit. When the number of running queries reaches this value, new queries will being queued.
* max_queue_size: Optional, length of the query queue. When the queue is full, new queries will be rejected. The default value is 0, which means no queuing.
* queue_timeout: Optional, query the timeout time in the queue, measured in milliseconds. If the query exceeds this value, an exception will be thrown directly to the client. The default value is 0, which means no queuing.
+* scan_thread_num: Optional, the number of threads used for scanning in the current workload group. The default value is -1, which means it does not take effect, the number of scan threads in the be configuration shall prevail. The value is an integer greater than 0.
Notes:
diff --git a/docs/zh-CN/docs/admin-manual/workload-group.md b/docs/zh-CN/docs/admin-manual/workload-group.md
index 62d6cc61c49..d25fd26212e 100644
--- a/docs/zh-CN/docs/admin-manual/workload-group.md
+++ b/docs/zh-CN/docs/admin-manual/workload-group.md
@@ -58,6 +58,7 @@ Workload Group是从2.0版本开始支持的功能，Workload Group在2.0版本
* max_concurrency：可选，最大查询并发数，默认值为整型最大值，也就是不做并发的限制。运行中的查询数量达到该值时，新来的查询会进入排队的逻辑。
* max_queue_size：可选，查询排队队列的长度，当排队队列已满时，新来的查询会被拒绝。默认值为0，含义是不排队。
* queue_timeout：可选，查询在排队队列中的超时时间，单位为毫秒，如果查询在队列中的排队时间超过这个值，那么就会直接抛出异常给客户端。默认值为0，含义是不排队。
+* scan_thread_num：可选，当前workload group用于scan的线程个数，默认值为-1，含义是不生效，此时以be配置中的scan线程数为准。取值为大于0的整数。
注意事项：
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java
index b998eb877f2..e624d29f7f1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java
@@ -61,12 +61,14 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
public static final String QUEUE_TIMEOUT = "queue_timeout";
+ public static final String SCAN_THREAD_NUM = "scan_thread_num";
+
// NOTE(wb): all property is not required, some properties default value is set in be
// default value is as followed
// cpu_share=1024, memory_limit=0%(0 means not limit), enable_memory_overcommit=true
private static final ImmutableSet<String> ALL_PROPERTIES_NAME = new ImmutableSet.Builder<String>()
.add(CPU_SHARE).add(MEMORY_LIMIT).add(ENABLE_MEMORY_OVERCOMMIT).add(MAX_CONCURRENCY)
- .add(MAX_QUEUE_SIZE).add(QUEUE_TIMEOUT).add(CPU_HARD_LIMIT).build();
+ .add(MAX_QUEUE_SIZE).add(QUEUE_TIMEOUT).add(CPU_HARD_LIMIT).add(SCAN_THREAD_NUM).build();
@SerializedName(value = "id")
private long id;
@@ -210,6 +212,19 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
}
}
+ if (properties.containsKey(SCAN_THREAD_NUM)) {
+ String value = properties.get(SCAN_THREAD_NUM);
+ try {
+ int intValue = Integer.parseInt(value);
+ if (intValue <= 0 && intValue != -1) {
+ throw new NumberFormatException();
+ }
+ } catch (NumberFormatException e) {
+ throw new DdlException(
+ SCAN_THREAD_NUM + " must be a positive integer or -1. but input value is " + value);
+ }
+ }
+
// check queue property
if (properties.containsKey(MAX_CONCURRENCY)) {
try {
@@ -292,6 +307,8 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
row.add("0%");
} else if (ENABLE_MEMORY_OVERCOMMIT.equals(key) && !properties.containsKey(key)) {
row.add("true");
+ } else if (SCAN_THREAD_NUM.equals(key) && !properties.containsKey(key)) {
+ row.add("-1");
} else {
row.add(properties.get(key));
}
@@ -347,6 +364,11 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
+ "id=" + id + ",name=" + name);
}
+ String scanThreadNumStr = properties.get(SCAN_THREAD_NUM);
+ if (scanThreadNumStr != null) {
+ tWorkloadGroupInfo.setScanThreadNum(Integer.parseInt(scanThreadNumStr));
+ }
+
TopicInfo topicInfo = new TopicInfo();
topicInfo.setWorkloadGroupInfo(tWorkloadGroupInfo);
return topicInfo;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
index 333ba6d98f3..82151d07a8b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
@@ -70,6 +70,7 @@ public class WorkloadGroupMgr implements Writable, GsonPostProcessable {
.add(WorkloadGroup.ENABLE_MEMORY_OVERCOMMIT)
.add(WorkloadGroup.MAX_CONCURRENCY).add(WorkloadGroup.MAX_QUEUE_SIZE)
.add(WorkloadGroup.QUEUE_TIMEOUT).add(WorkloadGroup.CPU_HARD_LIMIT)
+ .add(WorkloadGroup.SCAN_THREAD_NUM)
.add(QueryQueue.RUNNING_QUERY_NUM).add(QueryQueue.WAITING_QUERY_NUM)
.build();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
index 06688b3f46b..6feb8fa94f4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
@@ -383,8 +383,9 @@ public class MetadataGenerator {
trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(6)))); // max queue size
trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(7)))); // queue timeout
trow.addToColumnValue(new TCell().setStringVal(rGroupsInfo.get(8))); // cpu hard limit
- trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(9)))); // running query num
- trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(10)))); // waiting query num
+ trow.addToColumnValue(new TCell().setIntVal(Integer.parseInt(rGroupsInfo.get(9)))); // scan thread num
+ trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(10)))); // running query num
+ trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(11)))); // waiting query num
dataBatch.add(trow);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/WorkloadGroupsTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/WorkloadGroupsTableValuedFunction.java
index 5b34db5a684..a50fb7ca853 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/WorkloadGroupsTableValuedFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/WorkloadGroupsTableValuedFunction.java
@@ -49,6 +49,7 @@ public class WorkloadGroupsTableValuedFunction extends MetadataTableValuedFuncti
new Column(WorkloadGroup.MAX_QUEUE_SIZE, ScalarType.createType(PrimitiveType.BIGINT)),
new Column(WorkloadGroup.QUEUE_TIMEOUT, ScalarType.createType(PrimitiveType.BIGINT)),
new Column(WorkloadGroup.CPU_HARD_LIMIT, ScalarType.createStringType()),
+ new Column(WorkloadGroup.SCAN_THREAD_NUM, ScalarType.createType(PrimitiveType.INT)),
new Column(QueryQueue.RUNNING_QUERY_NUM, ScalarType.createType(PrimitiveType.BIGINT)),
new Column(QueryQueue.WAITING_QUERY_NUM, ScalarType.createType(PrimitiveType.BIGINT)));
diff --git a/gensrc/thrift/BackendService.thrift b/gensrc/thrift/BackendService.thrift
index dab0b860677..c59abd65d3b 100644
--- a/gensrc/thrift/BackendService.thrift
+++ b/gensrc/thrift/BackendService.thrift
@@ -176,6 +176,7 @@ struct TWorkloadGroupInfo {
6: optional string mem_limit
7: optional bool enable_memory_overcommit
8: optional bool enable_cpu_hard_limit
+ 9: optional i32 scan_thread_num
}
struct TWorkloadMoveQueryToGroupAction {
diff --git a/regression-test/data/workload_manager_p0/test_curd_wlg.out b/regression-test/data/workload_manager_p0/test_curd_wlg.out
index 1f3fbaf9e74..77b9fa75be8 100644
--- a/regression-test/data/workload_manager_p0/test_curd_wlg.out
+++ b/regression-test/data/workload_manager_p0/test_curd_wlg.out
@@ -6,15 +6,15 @@
2
-- !show_1 --
-normal 20 50% true 2147483647 0 0 1%
-test_group 10 10% true 2147483647 0 0 0%
+normal 20 50% true 2147483647 0 0 1% 16
+test_group 10 10% true 2147483647 0 0 0% -1
-- !mem_limit_1 --
2
-- !mem_limit_2 --
-normal 20 50% true 2147483647 0 0 1%
-test_group 10 11% true 2147483647 0 0 0%
+normal 20 50% true 2147483647 0 0 1% 16
+test_group 10 11% true 2147483647 0 0 0% -1
-- !mem_overcommit_1 --
2
@@ -23,24 +23,24 @@ test_group 10 11% true 2147483647 0 0 0%
2
-- !mem_overcommit_3 --
-normal 20 50% true 2147483647 0 0 1%
-test_group 10 11% false 2147483647 0 0 0%
+normal 20 50% true 2147483647 0 0 1% 16
+test_group 10 11% false 2147483647 0 0 0% -1
-- !cpu_hard_limit_1 --
2
-- !cpu_hard_limit_2 --
-normal 20 50% true 2147483647 0 0 1%
-test_group 10 11% false 2147483647 0 0 20%
+normal 20 50% true 2147483647 0 0 1% 16
+test_group 10 11% false 2147483647 0 0 20% -1
-- !queue_1 --
2
-- !show_queue --
-normal 20 50% true 2147483647 0 0 1%
-test_group 10 11% false 100 0 0 20%
+normal 20 50% true 2147483647 0 0 1% 16
+test_group 10 11% false 100 0 0 20% -1
-- !select_tvf_1 --
-normal 20 50% true 2147483647 0 0 1%
-test_group 10 11% false 100 0 0 20%
+normal 20 50% true 2147483647 0 0 1% 16
+test_group 10 11% false 100 0 0 20% -1
diff --git a/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy b/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
index 3e56525acca..1c3536eefc5 100644
--- a/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
+++ b/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
@@ -80,12 +80,14 @@ suite("test_crud_wlg") {
sql "alter workload group normal properties ( 'max_queue_size'='0' );"
sql "alter workload group normal properties ( 'queue_timeout'='0' );"
sql "alter workload group normal properties ( 'cpu_hard_limit'='1%' );"
+ sql "alter workload group normal properties ( 'scan_thread_num'='-1' );"
sql "set workload_group=normal;"
// test cpu_share
qt_cpu_share """ select count(1) from ${table_name} """
+ sql "alter workload group normal properties ( 'scan_thread_num'='16' );"
sql "alter workload group normal properties ( 'cpu_share'='20' );"
qt_cpu_share_2 """ select count(1) from ${table_name} """
@@ -96,6 +98,12 @@ suite("test_crud_wlg") {
exception "requires a positive integer"
}
+ test {
+ sql "alter workload group normal properties ( 'scan_thread_num'='0' );"
+
+ exception "scan_thread_num must be a positive integer or -1"
+ }
+
sql "drop workload group if exists test_group;"
// test create group
@@ -107,7 +115,7 @@ suite("test_crud_wlg") {
");"
sql "set workload_group=test_group;"
- qt_show_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit from workload_groups() order by name;"
+ qt_show_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
// test memory_limit
test {
@@ -118,7 +126,7 @@ suite("test_crud_wlg") {
sql "alter workload group test_group properties ( 'memory_limit'='11%' );"
qt_mem_limit_1 """ select count(1) from ${table_name} """
- qt_mem_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit from workload_groups() order by name;"
+ qt_mem_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
// test enable_memory_overcommit
test {
@@ -131,7 +139,7 @@ suite("test_crud_wlg") {
qt_mem_overcommit_1 """ select count(1) from ${table_name} """
sql "alter workload group test_group properties ( 'enable_memory_overcommit'='false' );"
qt_mem_overcommit_2 """ select count(1) from ${table_name} """
- qt_mem_overcommit_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit from workload_groups() order by name;"
+ qt_mem_overcommit_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
// test cpu_hard_limit
test {
@@ -150,7 +158,7 @@ suite("test_crud_wlg") {
sql "alter workload group test_group properties ( 'cpu_hard_limit'='20%' );"
qt_cpu_hard_limit_1 """ select count(1) from ${table_name} """
- qt_cpu_hard_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit from workload_groups() order by name;"
+ qt_cpu_hard_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
// test query queue
test {
@@ -173,7 +181,7 @@ suite("test_crud_wlg") {
sql "alter workload group test_group properties ( 'max_concurrency'='100' );"
qt_queue_1 """ select count(1) from ${table_name} """
- qt_show_queue "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit from workload_groups() order by name;"
+ qt_show_queue "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
// test create group failed
// failed for cpu_share
@@ -251,7 +259,7 @@ suite("test_crud_wlg") {
}
// test show workload groups
- qt_select_tvf_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit from workload_groups() order by name;"
+ qt_select_tvf_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
// test auth
sql """drop user if exists test_wlg_user"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 05/12: [Bug](AggState) fix not match function when agg combinator function has alias (#31101)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 493385c2c71e22670175426f3ba6a1c38745cb06
Author: Pxl <px...@qq.com>
AuthorDate: Tue Feb 20 10:47:51 2024 +0800
[Bug](AggState) fix not match function when agg combinator function has alias (#31101)
---
be/src/vec/data_types/data_type_agg_state.h | 28 ++++------------------
be/src/vec/exprs/vectorized_agg_fn.cpp | 4 +---
.../data/datatype_p0/agg_state/test_agg_state.out | 6 +++++
.../datatype_p0/agg_state/test_agg_state.groovy | 3 +++
4 files changed, 15 insertions(+), 26 deletions(-)
diff --git a/be/src/vec/data_types/data_type_agg_state.h b/be/src/vec/data_types/data_type_agg_state.h
index 264f3008979..7531adc106b 100644
--- a/be/src/vec/data_types/data_type_agg_state.h
+++ b/be/src/vec/data_types/data_type_agg_state.h
@@ -14,20 +14,10 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
-// This file is copied from
-// https://github.com/ClickHouse/ClickHouse/blob/master/src/DataTypes/DataTypeString.h
-// and modified by Doris
#pragma once
-#include <gen_cpp/Types_types.h>
#include <gen_cpp/data.pb.h>
-#include <glog/logging.h>
-#include <stddef.h>
-#include <stdint.h>
-
-#include <memory>
-#include <string>
#include "common/exception.h"
#include "common/status.h"
@@ -41,22 +31,14 @@
#include "vec/data_types/data_type_string.h"
#include "vec/data_types/serde/data_type_fixedlengthobject_serde.h"
-namespace doris {
-namespace vectorized {
-class BufferWritable;
-class IColumn;
-class ReadBuffer;
-} // namespace vectorized
-} // namespace doris
-
namespace doris::vectorized {
class DataTypeAggState : public DataTypeString {
public:
DataTypeAggState(DataTypes sub_types, bool result_is_nullable, std::string function_name)
: _result_is_nullable(result_is_nullable),
- _sub_types(sub_types),
- _function_name(function_name) {
+ _sub_types(std::move(sub_types)),
+ _function_name(std::move(function_name)) {
_agg_function = AggregateFunctionSimpleFactory::instance().get(_function_name, _sub_types,
_result_is_nullable);
if (_agg_function == nullptr) {
@@ -73,11 +55,11 @@ public:
_function_name, _result_is_nullable, get_types_string());
}
+ std::string get_function_name() const { return _function_name; }
+
TypeIndex get_type_id() const override { return TypeIndex::AggState; }
- TypeDescriptor get_type_as_type_descriptor() const override {
- return TypeDescriptor(TYPE_AGG_STATE);
- }
+ TypeDescriptor get_type_as_type_descriptor() const override { return {TYPE_AGG_STATE}; }
doris::FieldType get_storage_field_type() const override {
return doris::FieldType::OLAP_FIELD_TYPE_AGG_STATE;
diff --git a/be/src/vec/exprs/vectorized_agg_fn.cpp b/be/src/vec/exprs/vectorized_agg_fn.cpp
index 166ad9bc2b2..be771523505 100644
--- a/be/src/vec/exprs/vectorized_agg_fn.cpp
+++ b/be/src/vec/exprs/vectorized_agg_fn.cpp
@@ -166,9 +166,7 @@ Status AggFnEvaluator::prepare(RuntimeState* state, const RowDescriptor& desc,
}
std::string type_function_name =
- assert_cast<const DataTypeAggState*>(argument_types[0].get())
- ->get_nested_function()
- ->get_name();
+ assert_cast<const DataTypeAggState*>(argument_types[0].get())->get_function_name();
if (type_function_name + AGG_UNION_SUFFIX == _fn.name.function_name) {
if (_data_type->is_nullable()) {
return Status::InternalError(
diff --git a/regression-test/data/datatype_p0/agg_state/test_agg_state.out b/regression-test/data/datatype_p0/agg_state/test_agg_state.out
index 03b46f74a49..120a9844f64 100644
--- a/regression-test/data/datatype_p0/agg_state/test_agg_state.out
+++ b/regression-test/data/datatype_p0/agg_state/test_agg_state.out
@@ -44,3 +44,9 @@
-- !max_by_null --
\N \N
+-- !ndv --
+1
+
+-- !approx_count_distinct --
+1
+
diff --git a/regression-test/suites/datatype_p0/agg_state/test_agg_state.groovy b/regression-test/suites/datatype_p0/agg_state/test_agg_state.groovy
index 3930b0221ff..0b05d4312f9 100644
--- a/regression-test/suites/datatype_p0/agg_state/test_agg_state.groovy
+++ b/regression-test/suites/datatype_p0/agg_state/test_agg_state.groovy
@@ -73,4 +73,7 @@ suite("test_agg_state") {
sql "select avg_state(1) from d_table;"
exception "write_column_to_pb with type ColumnFixedLengthObject"
}
+
+ qt_ndv """select ndv_merge(t) from (select ndv_union(ndv_state(1)) as t from d_table group by k1)p;"""
+ qt_approx_count_distinct """select approx_count_distinct_merge(t) from (select approx_count_distinct_union(approx_count_distinct_state(1)) as t from d_table group by k1)p;"""
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org