You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2022/12/19 05:10:34 UTC
[doris] branch branch-1.2-lts updated (ae084be649 -> 95aab248b0)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a change to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
from ae084be649 [Bug](function) fix overflow on concat_ws (#15043)
new b027ae3490 [fix](merge-on-write) calc delete bitmap need all segments which _do_flush in one memtable (#15018)
new 5fce045f51 [Enhancement](partition prune): calculate the column ranges of compound predicates (#14886)
new a282d720c9 [Bug](datev2) Fix wrong result when use datev2 as partition key (#15094)
new 12192e3f13 [fix](memtracker) Fix load channel memory tracker are not refreshed in time (#15048)
new 315f039685 [fix](explain) fix explain output format problem (#15019)
new e68a49f3fe [Bug](case function) do not crash if prepare failed (#15113)
new d41a963c8f [fix](multi catalog)Return emtpy block while external table scanner couldn't find the file (#14997)
new 3fe708cf56 [config](load) enable new load scan node by default (#14808)
new 8853d515ae [vectorized](join) add try catch in create thread (#15065)
new d23e7905b0 [fix](iceberg-v2) icebergv2 filter data path (#14470)
new 2baa0e8891 [improvement](multi-catalog)Support invalid/not invalid option for refresh catalog and db. (#14922)
new d6f4f08dcc [Bug](datev2) Fix compatible problems caused by datev2 (#15131)
new f6936f256d [fix](type) forbid time type when creating table (#15093)
new cc13983766 [vectorized](jdbc) fix external table of oracle with condition about … (#15092)
new 4a1bcebf41 [fix](subquery)fix bug of using constexpr as subquery's output (#15119)
new 22aacd849c [fix](load) delta writer init failed might cause data inconsistency between multiple replicas (#15058)
new afbe5ba198 [fix](multi-catalog) hidden password for show create jdbc catalog (#15145)
new 6234291506 [fix](schema) Fix create table error if Colocate tables not equal to bucket num (#15071)
new d79b601796 [enhancement](signal) output query_id when 'be' core dumped #15080
new 6766c51353 [bug](udaf) fix java udaf incorrect get null value with row (#15151)
new 95aab248b0 [fix](merge-on-write) delete all rows with same key in all pre segments (#14995)
The 21 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/common/daemon.cpp | 9 +-
be/src/http/action/stream_load.cpp | 1 +
be/src/io/hdfs_file_reader.cpp | 3 +
be/src/olap/delta_writer.cpp | 4 +-
be/src/olap/memtable.cpp | 19 ++--
be/src/olap/memtable.h | 3 +-
be/src/olap/rowset/beta_rowset.cpp | 35 ++++---
be/src/olap/rowset/beta_rowset.h | 3 +-
be/src/olap/rowset/beta_rowset_writer.h | 2 +
be/src/olap/rowset/rowset_writer.h | 2 +
be/src/olap/tablet.cpp | 51 +++++-----
be/src/olap/tablet.h | 3 +-
be/src/runtime/fragment_mgr.cpp | 5 -
be/src/runtime/raw_value.h | 11 +--
be/src/runtime/tablets_channel.cpp | 17 +++-
be/src/runtime/thread_context.cpp | 3 +
be/src/util/jsonb_parser.h | 25 +++--
be/src/vec/exec/format/parquet/parquet_common.h | 8 ++
be/src/vec/exec/format/parquet/schema_desc.cpp | 14 ++-
be/src/vec/exec/format/parquet/schema_desc.h | 4 +-
.../exec/format/parquet/vparquet_column_reader.cpp | 10 +-
.../vec/exec/format/parquet/vparquet_page_index.h | 4 +-
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 58 +++++++-----
be/src/vec/exec/format/parquet/vparquet_reader.h | 3 +-
be/src/vec/exec/format/table/iceberg_reader.cpp | 104 ++++++++++++++++-----
be/src/vec/exec/format/table/iceberg_reader.h | 26 ++++--
be/src/vec/exec/format/table/table_format_reader.h | 3 +-
be/src/vec/exec/join/vjoin_node_base.cpp | 15 ++-
be/src/vec/exec/scan/new_jdbc_scan_node.cpp | 2 +-
be/src/vec/exec/scan/new_jdbc_scanner.cpp | 7 +-
be/src/vec/exec/scan/new_jdbc_scanner.h | 2 +-
be/src/vec/exec/scan/new_odbc_scan_node.cpp | 2 +-
be/src/vec/exec/scan/new_odbc_scanner.cpp | 6 +-
be/src/vec/exec/scan/new_odbc_scanner.h | 2 +-
be/src/vec/exec/scan/scanner_scheduler.cpp | 12 ++-
be/src/vec/exec/scan/vfile_scanner.cpp | 15 ++-
be/src/vec/exprs/vcase_expr.cpp | 12 ---
be/src/vec/functions/function_case.h | 9 +-
fe/fe-core/src/main/cup/sql_parser.cup | 12 +--
.../org/apache/doris/analysis/CreateTableStmt.java | 4 +
.../org/apache/doris/analysis/DataDescription.java | 27 +++++-
.../org/apache/doris/analysis/DateLiteral.java | 89 ++++++++++++++++--
.../apache/doris/analysis/RefreshCatalogStmt.java | 15 ++-
.../org/apache/doris/analysis/RefreshDbStmt.java | 18 +++-
.../apache/doris/catalog/ColocateGroupSchema.java | 12 +++
.../org/apache/doris/catalog/PartitionKey.java | 22 +++--
.../org/apache/doris/catalog/RefreshManager.java | 7 +-
.../doris/catalog/external/ExternalDatabase.java | 8 +-
.../main/java/org/apache/doris/common/Config.java | 2 +-
.../java/org/apache/doris/common/ErrorCode.java | 2 +
.../org/apache/doris/common/util/PrintableMap.java | 1 +
.../apache/doris/datasource/CatalogFactory.java | 1 +
.../org/apache/doris/datasource/CatalogLog.java | 3 +
.../org/apache/doris/datasource/CatalogMgr.java | 10 +-
.../apache/doris/datasource/EsExternalCatalog.java | 2 +-
.../apache/doris/datasource/ExternalCatalog.java | 12 ++-
.../apache/doris/datasource/ExternalObjectLog.java | 3 +
.../doris/datasource/HMSExternalCatalog.java | 2 +-
.../apache/doris/datasource/InternalCatalog.java | 1 +
.../doris/datasource/JdbcExternalCatalog.java | 2 +-
.../org/apache/doris/load/BrokerFileGroup.java | 2 +
.../java/org/apache/doris/load/DeleteHandler.java | 6 +-
.../org/apache/doris/planner/LoadScanNode.java | 5 +-
.../apache/doris/planner/NestedLoopJoinNode.java | 2 +-
.../org/apache/doris/planner/OdbcScanNode.java | 8 ++
.../java/org/apache/doris/planner/ScanNode.java | 83 ++++++++++++++++
.../apache/doris/planner/SingleNodePlanner.java | 4 +-
.../apache/doris/planner/StreamLoadPlanner.java | 14 ++-
.../planner/external/ExternalFileScanNode.java | 23 ++++-
.../doris/planner/external/FileGroupInfo.java | 15 +++
.../apache/doris/planner/external/HiveSplit.java | 1 +
.../planner/external/IcebergScanProvider.java | 86 +++++++++++++++--
.../doris/planner/external/IcebergSplit.java | 5 +
.../doris/planner/external/LoadScanProvider.java | 3 +-
.../doris/planner/external/QueryScanProvider.java | 9 +-
.../apache/doris/service/FrontendServiceImpl.java | 3 +-
.../java/org/apache/doris/task/LoadTaskInfo.java | 5 +
.../java/org/apache/doris/task/StreamLoadTask.java | 11 +++
.../java/org/apache/doris/alter/AlterTest.java | 5 +-
.../doris/analysis/RangePartitionPruneTest.java | 8 ++
.../doris/planner/StreamLoadPlannerTest.java | 71 --------------
.../apache/doris/utframe/TestWithFeService.java | 8 +-
.../java/org/apache/doris/udf/UdafExecutor.java | 3 +-
gensrc/thrift/FrontendService.thrift | 1 +
gensrc/thrift/PlanNodes.thrift | 7 +-
.../test_outer_join_with_subquery.out | 3 +
regression-test/data/delete_p0/test_delete.out | 16 ++--
.../data/partition_p0/test_datev2_partition.out | 39 ++++++++
.../query_p0/show/test_show_create_catalog.out | 4 +
.../table_valued_function/test_hdfs_tvf.groovy | 2 -
.../test_outer_join_with_subquery.groovy | 25 +++++
.../suites/delete_p0/test_delete.groovy | 22 ++---
.../suites/export_p0/test_outfile_parquet.groovy | 2 +-
.../external_catalog_p0/hive/test_hive_orc.groovy | 27 ------
.../hive/test_hive_other.groovy | 25 -----
.../hive/test_hive_parquet.groovy | 28 ------
.../jsonb_p0/test_jsonb_load_and_function.groovy | 26 +++++-
.../test_jsonb_load_unique_key_and_function.groovy | 2 +
.../load_p0/broker_load/test_array_load.groovy | 8 --
.../load_p0/broker_load/test_broker_load.groovy | 21 -----
...n_column_exclude_schema_without_jsonpath.groovy | 2 -
.../stream_load/load_json_null_to_nullable.groovy | 6 --
.../stream_load/load_json_with_jsonpath.groovy | 7 --
.../load_p0/stream_load/test_hdfs_json_load.groovy | 6 --
.../load_p0/stream_load/test_json_load.groovy | 13 ---
.../stream_load/test_txt_special_delimiter.groovy | 7 --
.../partition_p0/test_datev2_partition.groovy | 76 +++++++++++++++
.../test_streamload_perfomance.groovy | 2 +-
.../query_p0/show/test_show_create_catalog.groovy | 46 +++++++++
.../multi_catalog_query/hive_catalog_orc.groovy | 25 -----
.../hive_catalog_parquet.groovy | 25 -----
111 files changed, 1066 insertions(+), 556 deletions(-)
create mode 100644 regression-test/data/partition_p0/test_datev2_partition.out
create mode 100644 regression-test/data/query_p0/show/test_show_create_catalog.out
create mode 100644 regression-test/suites/partition_p0/test_datev2_partition.groovy
create mode 100644 regression-test/suites/query_p0/show/test_show_create_catalog.groovy
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 11/21: [improvement](multi-catalog)Support invalid/not invalid option for refresh catalog and db. (#14922)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 2baa0e8891dfb9e1c90b6c546fe38191bbd71986
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Thu Dec 8 20:38:07 2022 +0800
[improvement](multi-catalog)Support invalid/not invalid option for refresh catalog and db. (#14922)
Current refresh catalog/db operation always invalid all the related cache. In some cases, it is not necessary,
for example, create a new db in external data source. This pr is to support refresh without invalidate cache.
refresh catalog hive properties("invalid_cache" = "false");
refresh database hive.db1 properties("invalid_cache" = "false");
---
fe/fe-core/src/main/cup/sql_parser.cup | 12 ++++++------
.../org/apache/doris/analysis/RefreshCatalogStmt.java | 15 ++++++++++++++-
.../java/org/apache/doris/analysis/RefreshDbStmt.java | 18 ++++++++++++++++--
.../java/org/apache/doris/catalog/RefreshManager.java | 7 ++++---
.../doris/catalog/external/ExternalDatabase.java | 8 ++++++--
.../org/apache/doris/datasource/CatalogFactory.java | 1 +
.../java/org/apache/doris/datasource/CatalogLog.java | 3 +++
.../java/org/apache/doris/datasource/CatalogMgr.java | 8 ++++----
.../org/apache/doris/datasource/EsExternalCatalog.java | 2 +-
.../org/apache/doris/datasource/ExternalCatalog.java | 12 ++++++++++--
.../org/apache/doris/datasource/ExternalObjectLog.java | 3 +++
.../apache/doris/datasource/HMSExternalCatalog.java | 2 +-
.../apache/doris/datasource/JdbcExternalCatalog.java | 2 +-
13 files changed, 70 insertions(+), 23 deletions(-)
diff --git a/fe/fe-core/src/main/cup/sql_parser.cup b/fe/fe-core/src/main/cup/sql_parser.cup
index 6b3fff4bd5..5c209fe478 100644
--- a/fe/fe-core/src/main/cup/sql_parser.cup
+++ b/fe/fe-core/src/main/cup/sql_parser.cup
@@ -1137,13 +1137,13 @@ refresh_stmt ::=
{:
RESULT = new RefreshTableStmt(tbl);
:}
- | KW_REFRESH KW_DATABASE ident:db
+ | KW_REFRESH KW_DATABASE ident:db opt_properties:properties
{:
- RESULT = new RefreshDbStmt(db);
+ RESULT = new RefreshDbStmt(db, properties);
:}
- | KW_REFRESH KW_DATABASE ident:ctl DOT ident:db
+ | KW_REFRESH KW_DATABASE ident:ctl DOT ident:db opt_properties:properties
{:
- RESULT = new RefreshDbStmt(ctl, db);
+ RESULT = new RefreshDbStmt(ctl, db, properties);
:}
| KW_REFRESH KW_MATERIALIZED KW_VIEW table_name:mv
{:
@@ -1153,9 +1153,9 @@ refresh_stmt ::=
{:
RESULT = new RefreshMaterializedViewStmt(mv, MVRefreshInfo.RefreshMethod.COMPLETE);
:}
- | KW_REFRESH KW_CATALOG ident:catalogName
+ | KW_REFRESH KW_CATALOG ident:catalogName opt_properties:properties
{:
- RESULT = new RefreshCatalogStmt(catalogName);
+ RESULT = new RefreshCatalogStmt(catalogName, properties);
:}
;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshCatalogStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshCatalogStmt.java
index 481ebf3b9f..e054dca050 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshCatalogStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshCatalogStmt.java
@@ -27,22 +27,32 @@ import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
+import java.util.Map;
+
/**
* RefreshCatalogStmt
* Manually refresh the catalog metadata.
*/
public class RefreshCatalogStmt extends DdlStmt {
+ private static final String INVALID_CACHE = "invalid_cache";
private final String catalogName;
+ private Map<String, String> properties;
+ private boolean invalidCache = false;
- public RefreshCatalogStmt(String catalogName) {
+ public RefreshCatalogStmt(String catalogName, Map<String, String> properties) {
this.catalogName = catalogName;
+ this.properties = properties;
}
public String getCatalogName() {
return catalogName;
}
+ public boolean isInvalidCache() {
+ return invalidCache;
+ }
+
@Override
public void analyze(Analyzer analyzer) throws UserException {
super.analyze(analyzer);
@@ -56,6 +66,9 @@ public class RefreshCatalogStmt extends DdlStmt {
ErrorReport.reportAnalysisException(ErrorCode.ERR_CATALOG_ACCESS_DENIED,
analyzer.getQualifiedUser(), catalogName);
}
+ String invalidConfig = properties == null ? null : properties.get(INVALID_CACHE);
+ // Default is to invalid cache.
+ invalidCache = invalidConfig == null ? true : invalidConfig.equalsIgnoreCase("true");
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshDbStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshDbStmt.java
index 5118088916..ccd05a0524 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshDbStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/RefreshDbStmt.java
@@ -31,19 +31,26 @@ import com.google.common.base.Strings;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.util.Map;
+
public class RefreshDbStmt extends DdlStmt {
private static final Logger LOG = LogManager.getLogger(RefreshDbStmt.class);
+ private static final String INVALID_CACHE = "invalid_cache";
private String catalogName;
private String dbName;
+ private Map<String, String> properties;
+ private boolean invalidCache = false;
- public RefreshDbStmt(String dbName) {
+ public RefreshDbStmt(String dbName, Map<String, String> properties) {
this.dbName = dbName;
+ this.properties = properties;
}
- public RefreshDbStmt(String catalogName, String dbName) {
+ public RefreshDbStmt(String catalogName, String dbName, Map<String, String> properties) {
this.catalogName = catalogName;
this.dbName = dbName;
+ this.properties = properties;
}
public String getDbName() {
@@ -54,6 +61,10 @@ public class RefreshDbStmt extends DdlStmt {
return catalogName;
}
+ public boolean isInvalidCache() {
+ return invalidCache;
+ }
+
@Override
public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
super.analyze(analyzer);
@@ -82,6 +93,9 @@ public class RefreshDbStmt extends DdlStmt {
ErrorReport.reportAnalysisException(
ErrorCode.ERR_DBACCESS_DENIED_ERROR, analyzer.getQualifiedUser(), dbName);
}
+ String invalidConfig = properties == null ? null : properties.get(INVALID_CACHE);
+ // Default is to invalid cache.
+ invalidCache = invalidConfig == null ? true : invalidConfig.equalsIgnoreCase("true");
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
index 558b133f8c..b8d6929994 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
@@ -76,7 +76,7 @@ public class RefreshManager {
refreshInternalCtlIcebergDb(dbName, env);
} else {
// Process external catalog db refresh
- refreshExternalCtlDb(dbName, catalog);
+ refreshExternalCtlDb(dbName, catalog, stmt.isInvalidCache());
}
LOG.info("Successfully refresh db: {}", dbName);
}
@@ -107,7 +107,7 @@ public class RefreshManager {
env.getIcebergTableCreationRecordMgr().registerDb(db);
}
- private void refreshExternalCtlDb(String dbName, CatalogIf catalog) throws DdlException {
+ private void refreshExternalCtlDb(String dbName, CatalogIf catalog, boolean invalidCache) throws DdlException {
if (!(catalog instanceof ExternalCatalog)) {
throw new DdlException("Only support refresh ExternalCatalog Database");
}
@@ -116,10 +116,11 @@ public class RefreshManager {
if (db == null) {
throw new DdlException("Database " + dbName + " does not exist in catalog " + catalog.getName());
}
- ((ExternalDatabase) db).setUnInitialized();
+ ((ExternalDatabase) db).setUnInitialized(invalidCache);
ExternalObjectLog log = new ExternalObjectLog();
log.setCatalogId(catalog.getId());
log.setDbId(db.getId());
+ log.setInvalidCache(invalidCache);
Env.getCurrentEnv().getEditLog().logRefreshExternalDb(log);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalDatabase.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalDatabase.java
index 820360c21d..6ae8594c07 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalDatabase.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalDatabase.java
@@ -65,6 +65,7 @@ public class ExternalDatabase<T extends ExternalTable> implements DatabaseIf<T>,
@SerializedName(value = "initialized")
protected boolean initialized = false;
protected ExternalCatalog extCatalog;
+ protected boolean invalidCacheInInit = true;
/**
* No args constructor for persist.
@@ -93,9 +94,12 @@ public class ExternalDatabase<T extends ExternalTable> implements DatabaseIf<T>,
public void setTableExtCatalog(ExternalCatalog extCatalog) {
}
- public void setUnInitialized() {
+ public void setUnInitialized(boolean invalidCache) {
this.initialized = false;
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDbCache(extCatalog.getId(), name);
+ this.invalidCacheInInit = invalidCache;
+ if (invalidCache) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDbCache(extCatalog.getId(), name);
+ }
}
public boolean isInitialized() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java
index f5704e909d..c5599ca411 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java
@@ -54,6 +54,7 @@ public class CatalogFactory {
log.setNewCatalogName(((AlterCatalogNameStmt) stmt).getNewCatalogName());
} else if (stmt instanceof RefreshCatalogStmt) {
log.setCatalogId(catalogId);
+ log.setInvalidCache(((RefreshCatalogStmt) stmt).isInvalidCache());
} else {
throw new RuntimeException("Unknown stmt for catalog manager " + stmt.getClass().getSimpleName());
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogLog.java
index 0fc23af7b9..7e54e18386 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogLog.java
@@ -53,6 +53,9 @@ public class CatalogLog implements Writable {
@SerializedName(value = "newProps")
private Map<String, String> newProps;
+ @SerializedName(value = "invalidCache")
+ private boolean invalidCache;
+
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
index 854ee223d1..f3b9299bc8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
@@ -114,12 +114,12 @@ public class CatalogMgr implements Writable, GsonPostProcessable {
return catalog;
}
- private void unprotectedRefreshCatalog(long catalogId) {
+ private void unprotectedRefreshCatalog(long catalogId, boolean invalidCache) {
CatalogIf catalog = idToCatalog.get(catalogId);
if (catalog != null) {
String catalogName = catalog.getName();
if (!catalogName.equals(InternalCatalog.INTERNAL_CATALOG_NAME)) {
- ((ExternalCatalog) catalog).setUninitialized();
+ ((ExternalCatalog) catalog).setUninitialized(invalidCache);
}
}
}
@@ -448,7 +448,7 @@ public class CatalogMgr implements Writable, GsonPostProcessable {
public void replayRefreshCatalog(CatalogLog log) throws DdlException {
writeLock();
try {
- unprotectedRefreshCatalog(log.getCatalogId());
+ unprotectedRefreshCatalog(log.getCatalogId(), log.isInvalidCache());
} finally {
writeUnlock();
}
@@ -504,7 +504,7 @@ public class CatalogMgr implements Writable, GsonPostProcessable {
try {
ExternalCatalog catalog = (ExternalCatalog) idToCatalog.get(log.getCatalogId());
ExternalDatabase db = catalog.getDbForReplay(log.getDbId());
- db.setUnInitialized();
+ db.setUnInitialized(log.isInvalidCache());
} finally {
writeUnlock();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/EsExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/EsExternalCatalog.java
index 496eabfc8c..f078998fc8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/EsExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/EsExternalCatalog.java
@@ -134,7 +134,7 @@ public class EsExternalCatalog extends ExternalCatalog {
initCatalogLog.setCatalogId(id);
initCatalogLog.setType(InitCatalogLog.Type.ES);
if (dbNameToId != null && dbNameToId.containsKey(DEFAULT_DB)) {
- idToDb.get(dbNameToId.get(DEFAULT_DB)).setUnInitialized();
+ idToDb.get(dbNameToId.get(DEFAULT_DB)).setUnInitialized(invalidCacheInInit);
initCatalogLog.addRefreshDb(dbNameToId.get(DEFAULT_DB));
} else {
dbNameToId = Maps.newConcurrentMap();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
index 2369e6afd4..cc0c2f0b2d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
@@ -69,6 +69,7 @@ public abstract class ExternalCatalog implements CatalogIf<ExternalDatabase>, Wr
// db name does not contains "default_cluster"
protected Map<String, Long> dbNameToId = Maps.newConcurrentMap();
private boolean objectCreated = false;
+ protected boolean invalidCacheInInit = true;
private ExternalSchemaCache schemaCache;
@@ -129,8 +130,15 @@ public abstract class ExternalCatalog implements CatalogIf<ExternalDatabase>, Wr
// init schema related objects
protected abstract void init();
- public void setUninitialized() {
+ public void setUninitialized(boolean invalidCache) {
this.initialized = false;
+ this.invalidCacheInInit = invalidCache;
+ if (invalidCache) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateCatalogCache(id);
+ }
+ }
+
+ public void updateDbList() {
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateCatalogCache(id);
}
@@ -209,7 +217,7 @@ public abstract class ExternalCatalog implements CatalogIf<ExternalDatabase>, Wr
Map<Long, ExternalDatabase> tmpIdToDb = Maps.newConcurrentMap();
for (int i = 0; i < log.getRefreshCount(); i++) {
ExternalDatabase db = getDbForReplay(log.getRefreshDbIds().get(i));
- db.setUnInitialized();
+ db.setUnInitialized(invalidCacheInInit);
tmpDbNameToId.put(db.getFullName(), db.getId());
tmpIdToDb.put(db.getId(), db);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalObjectLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalObjectLog.java
index cff446657e..030f981382 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalObjectLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalObjectLog.java
@@ -43,6 +43,9 @@ public class ExternalObjectLog implements Writable {
@SerializedName(value = "tableId")
private long tableId;
+ @SerializedName(value = "invalidCache")
+ private boolean invalidCache;
+
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/HMSExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/HMSExternalCatalog.java
index 0dfffe19fb..8b79bf97f2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/HMSExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/HMSExternalCatalog.java
@@ -78,7 +78,7 @@ public class HMSExternalCatalog extends ExternalCatalog {
dbId = dbNameToId.get(dbName);
tmpDbNameToId.put(dbName, dbId);
ExternalDatabase db = idToDb.get(dbId);
- db.setUnInitialized();
+ db.setUnInitialized(invalidCacheInInit);
tmpIdToDb.put(dbId, db);
initCatalogLog.addRefreshDb(dbId);
} else {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/JdbcExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/JdbcExternalCatalog.java
index bcae5988d9..713c8176d0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/JdbcExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/JdbcExternalCatalog.java
@@ -122,7 +122,7 @@ public class JdbcExternalCatalog extends ExternalCatalog {
dbId = dbNameToId.get(dbName);
tmpDbNameToId.put(dbName, dbId);
ExternalDatabase db = idToDb.get(dbId);
- db.setUnInitialized();
+ db.setUnInitialized(invalidCacheInInit);
tmpIdToDb.put(dbId, db);
initCatalogLog.addRefreshDb(dbId);
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 10/21: [fix](iceberg-v2) icebergv2 filter data path (#14470)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit d23e7905b09c1d1f4dfe4849d21e5a79034dbf4e
Author: slothever <18...@users.noreply.github.com>
AuthorDate: Thu Dec 15 10:18:12 2022 +0800
[fix](iceberg-v2) icebergv2 filter data path (#14470)
1. a icebergv2 delete file may cross many data paths, so the path of a file split is required as a predicate to filter rows of delete file
- create delete file structure to save predicate parameters
- create predicate for file path
2. add some log to print row range
3. fix bug when create file metadata
---
be/src/vec/exec/format/parquet/parquet_common.h | 8 ++
be/src/vec/exec/format/parquet/schema_desc.h | 4 +-
.../exec/format/parquet/vparquet_column_reader.cpp | 10 +-
.../vec/exec/format/parquet/vparquet_page_index.h | 4 +-
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 58 +++++++-----
be/src/vec/exec/format/parquet/vparquet_reader.h | 3 +-
be/src/vec/exec/format/table/iceberg_reader.cpp | 104 ++++++++++++++++-----
be/src/vec/exec/format/table/iceberg_reader.h | 26 ++++--
be/src/vec/exec/format/table/table_format_reader.h | 3 +-
be/src/vec/exec/scan/vfile_scanner.cpp | 6 +-
.../planner/external/ExternalFileScanNode.java | 2 +-
.../apache/doris/planner/external/HiveSplit.java | 1 +
.../planner/external/IcebergScanProvider.java | 86 +++++++++++++++--
.../doris/planner/external/IcebergSplit.java | 5 +
.../doris/planner/external/QueryScanProvider.java | 9 +-
gensrc/thrift/PlanNodes.thrift | 7 +-
16 files changed, 252 insertions(+), 84 deletions(-)
diff --git a/be/src/vec/exec/format/parquet/parquet_common.h b/be/src/vec/exec/format/parquet/parquet_common.h
index d124c2fb9a..611ce969f1 100644
--- a/be/src/vec/exec/format/parquet/parquet_common.h
+++ b/be/src/vec/exec/format/parquet/parquet_common.h
@@ -46,6 +46,14 @@ struct RowRange {
int64_t first_row;
int64_t last_row;
+
+ bool operator<(const RowRange& range) const { return first_row < range.first_row; }
+
+ std::string debug_string() const {
+ std::stringstream ss;
+ ss << "[" << first_row << "," << last_row << ")";
+ return ss.str();
+ }
};
struct ParquetReadColumn {
diff --git a/be/src/vec/exec/format/parquet/schema_desc.h b/be/src/vec/exec/format/parquet/schema_desc.h
index 73f9f97d97..c45e96f236 100644
--- a/be/src/vec/exec/format/parquet/schema_desc.h
+++ b/be/src/vec/exec/format/parquet/schema_desc.h
@@ -40,11 +40,13 @@ struct FieldSchema {
tparquet::Type::type physical_type;
// The index order in FieldDescriptor._physical_fields
int physical_column_index = -1;
-
int16_t definition_level = 0;
int16_t repetition_level = 0;
std::vector<FieldSchema> children;
+ FieldSchema() = default;
+ ~FieldSchema() = default;
+ FieldSchema(const FieldSchema& fieldSchema) = default;
std::string debug_string() const;
};
diff --git a/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp
index 235a5f3e3b..c0557c482a 100644
--- a/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp
@@ -67,17 +67,17 @@ void ParquetColumnReader::_generate_read_ranges(int64_t start_index, int64_t end
}
int index = _row_range_index;
while (index < _row_ranges->size()) {
- const RowRange& read_range = (*_row_ranges)[index];
- if (read_range.last_row <= start_index) {
+ const RowRange& row_range = (*_row_ranges)[index];
+ if (row_range.last_row <= start_index) {
index++;
_row_range_index++;
continue;
}
- if (read_range.first_row >= end_index) {
+ if (row_range.first_row >= end_index) {
break;
}
- int64_t start = read_range.first_row < start_index ? start_index : read_range.first_row;
- int64_t end = read_range.last_row < end_index ? read_range.last_row : end_index;
+ int64_t start = row_range.first_row < start_index ? start_index : row_range.first_row;
+ int64_t end = row_range.last_row < end_index ? row_range.last_row : end_index;
read_ranges.emplace_back(start, end);
index++;
}
diff --git a/be/src/vec/exec/format/parquet/vparquet_page_index.h b/be/src/vec/exec/format/parquet/vparquet_page_index.h
index 978a798bf4..4a27593abf 100644
--- a/be/src/vec/exec/format/parquet/vparquet_page_index.h
+++ b/be/src/vec/exec/format/parquet/vparquet_page_index.h
@@ -36,9 +36,9 @@ public:
std::vector<int>& skipped_ranges);
bool check_and_get_page_index_ranges(const std::vector<tparquet::ColumnChunk>& columns);
Status parse_column_index(const tparquet::ColumnChunk& chunk, const uint8_t* buff,
- tparquet::ColumnIndex* _column_index);
+ tparquet::ColumnIndex* column_index);
Status parse_offset_index(const tparquet::ColumnChunk& chunk, const uint8_t* buff,
- tparquet::OffsetIndex* _offset_index);
+ tparquet::OffsetIndex* offset_index);
private:
friend class ParquetReader;
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index 01c226b1e4..c8d4f6b3c4 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -324,11 +324,13 @@ Status ParquetReader::get_parsered_schema(std::vector<std::string>* col_names,
_scan_range.start_offset,
_scan_range.file_size, 0, _file_reader));
}
- RETURN_IF_ERROR(_file_reader->open());
- if (_file_reader->size() == 0) {
- return Status::EndOfFile("Empty Parquet File");
+ if (_file_metadata == nullptr) {
+ RETURN_IF_ERROR(_file_reader->open());
+ if (_file_reader->size() == 0) {
+ return Status::EndOfFile("Empty Parquet File");
+ }
+ RETURN_IF_ERROR(parse_thrift_footer(_file_reader.get(), _file_metadata));
}
- RETURN_IF_ERROR(parse_thrift_footer(_file_reader.get(), _file_metadata));
_t_metadata = &_file_metadata->to_thrift();
_total_groups = _t_metadata->row_groups.size();
@@ -360,57 +362,64 @@ Status ParquetReader::get_columns(std::unordered_map<std::string, TypeDescriptor
return Status::OK();
}
-void ParquetReader::merge_delete_row_ranges(const std::vector<RowRange>& delete_row_ranges) {
+void ParquetReader::merge_delete_row_ranges(const std::set<RowRange>& delete_row_ranges) {
if (_row_ranges.empty()) {
+ _current_group_reader->set_row_ranges(_row_ranges);
return;
}
- std::vector<RowRange> candidate_ranges;
- auto start_range = _row_ranges.begin();
if (!delete_row_ranges.empty()) {
+ std::vector<RowRange> candidate_ranges;
+ auto start_range = _row_ranges.begin();
auto delete_range = delete_row_ranges.begin();
- int64_t range_start_idx = start_range->first_row;
+ int64_t processed_range_start_idx = start_range->first_row;
while (start_range != _row_ranges.end() && delete_range != delete_row_ranges.end()) {
int64_t delete_start = delete_range->first_row;
int64_t delete_end = delete_range->last_row;
int64_t range_start = start_range->first_row;
int64_t range_end = start_range->last_row;
-
- if (delete_end >= range_end) {
- if (range_start < delete_start) {
+ if (delete_end > range_end) {
+ if (range_start < processed_range_start_idx) {
+ // rows before processed_range_start_idx have been processed
+ range_start = processed_range_start_idx;
+ }
+ if (range_end < delete_start) {
+ /**
+ * start_range
+ * || --------- || - |--------- |
+ * delete_range
+ */
+ candidate_ranges.emplace_back(range_start, range_end);
+ } else if (range_start < delete_start) {
/**
* row_range
* || --------|-------- || ----- |
* delete_start delete_end
*/
candidate_ranges.emplace_back(range_start, delete_start);
- } else if (range_end <= delete_start) {
- /**
- * start_range
- * || --------- || ----------- |
- * delete_range
- */
- candidate_ranges.emplace_back(range_start, range_end);
}
+ // range_end > delete_end && range_start > delete_start
start_range++;
} else {
// delete_end < range_end，most of the time, we will use this branch
- if (range_start <= delete_start) {
+ if (processed_range_start_idx < delete_start) {
/**
* row_range_start row_range_end
* || --- | --------- | --- ||
* delete_range
*/
- candidate_ranges.emplace_back(range_start_idx, delete_start);
- range_start_idx = delete_end + 1;
+ candidate_ranges.emplace_back(processed_range_start_idx, delete_start);
}
+ // delete_end is in row_range, so it can assign to processed_range_start_idx
+ processed_range_start_idx = delete_end;
delete_range++;
if (delete_range == delete_row_ranges.end()) {
- candidate_ranges.emplace_back(delete_end + 1, range_end);
+ range_end = _row_ranges[_row_ranges.size() - 1].last_row;
+ if (processed_range_start_idx != range_end) {
+ candidate_ranges.emplace_back(processed_range_start_idx, range_end);
+ }
}
}
}
- }
- if (!candidate_ranges.empty()) {
_row_ranges.assign(candidate_ranges.begin(), candidate_ranges.end());
}
_current_group_reader->set_row_ranges(_row_ranges);
@@ -545,7 +554,6 @@ Status ParquetReader::_process_page_index(const tparquet::RowGroup& row_group) {
if (!_has_page_index(row_group.columns, page_index)) {
return Status::OK();
}
- // int64_t buffer_size = page_index._column_index_size;
uint8_t col_index_buff[page_index._column_index_size];
int64_t bytes_read = 0;
RETURN_IF_ERROR(_file_reader->readat(page_index._column_index_start,
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.h b/be/src/vec/exec/format/parquet/vparquet_reader.h
index aaae115bd3..f5891754e7 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.h
@@ -79,7 +79,7 @@ public:
Status file_metadata(FileMetaData** metadata);
- void merge_delete_row_ranges(const std::vector<RowRange>& delete_row_ranges);
+ void merge_delete_row_ranges(const std::set<RowRange>& delete_row_ranges);
int64_t size() const { return _file_reader->size(); }
@@ -149,7 +149,6 @@ private:
const TFileScanRangeParams& _scan_params;
const TFileRangeDesc& _scan_range;
std::unique_ptr<FileReader> _file_reader = nullptr;
- std::vector<RowRange> _delete_row_ranges;
std::vector<RowRange> _row_ranges;
std::shared_ptr<FileMetaData> _file_metadata;
const tparquet::FileMetaData* _t_metadata;
diff --git a/be/src/vec/exec/format/table/iceberg_reader.cpp b/be/src/vec/exec/format/table/iceberg_reader.cpp
index 041c70438f..9e2f659772 100644
--- a/be/src/vec/exec/format/table/iceberg_reader.cpp
+++ b/be/src/vec/exec/format/table/iceberg_reader.cpp
@@ -29,38 +29,64 @@ namespace doris::vectorized {
const int64_t MIN_SUPPORT_DELETE_FILES_VERSION = 2;
const std::string ICEBERG_ROW_POS = "pos";
+IcebergTableReader::IcebergTableReader(GenericReader* file_format_reader, RuntimeProfile* profile,
+ RuntimeState* state, const TFileScanRangeParams& params)
+ : TableFormatReader(file_format_reader), _profile(profile), _state(state), _params(params) {
+ static const char* iceberg_profile = "IcebergProfile";
+ ADD_TIMER(_profile, iceberg_profile);
+ _iceberg_profile._delete_files_init_time =
+ ADD_CHILD_TIMER(_profile, "DeleteFileInitTime", iceberg_profile);
+ _iceberg_profile._delete_files_read_total_time =
+ ADD_CHILD_TIMER(_profile, "DeleteFileReadTime", iceberg_profile);
+}
+
+IcebergTableReader::~IcebergTableReader() {
+ if (_data_path_conjunct_ctx != nullptr) {
+ _data_path_conjunct_ctx->close(_state);
+ }
+}
+
Status IcebergTableReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
return _file_format_reader->get_next_block(block, read_rows, eof);
}
+Status IcebergTableReader::set_fill_columns(
+ const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
+ partition_columns,
+ const std::unordered_map<std::string, VExprContext*>& missing_columns) {
+ return _file_format_reader->set_fill_columns(partition_columns, missing_columns);
+}
+
Status IcebergTableReader::get_columns(
std::unordered_map<std::string, TypeDescriptor>* name_to_type,
std::unordered_set<std::string>* missing_cols) {
return _file_format_reader->get_columns(name_to_type, missing_cols);
}
-void IcebergTableReader::filter_rows() {
+void IcebergTableReader::filter_rows(const TFileRangeDesc& range) {
if (_cur_delete_file_reader == nullptr) {
return;
}
- auto& table_desc = _params.table_format_params.iceberg_params;
+ SCOPED_TIMER(_iceberg_profile._delete_files_read_total_time);
+ auto& table_desc = range.table_format_params.iceberg_params;
auto& version = table_desc.format_version;
if (version < MIN_SUPPORT_DELETE_FILES_VERSION) {
return;
}
bool eof = false;
- std::vector<RowRange> delete_row_ranges;
+ std::set<RowRange> delete_row_ranges;
while (!eof) {
size_t read_rows = 0;
Block block = Block();
- for (const FieldSchema* field : _column_schemas) {
- DataTypePtr data_type = DataTypeFactory::instance().create_data_type(field->type, true);
+ for (const FieldSchema& field : _column_schemas) {
+ DataTypePtr data_type = DataTypeFactory::instance().create_data_type(field.type, true);
MutableColumnPtr data_column = data_type->create_column();
- block.insert(ColumnWithTypeAndName(std::move(data_column), data_type, field->name));
+ block.insert(ColumnWithTypeAndName(std::move(data_column), data_type, field.name));
}
Status st = _cur_delete_file_reader->get_next_block(&block, &read_rows, &eof);
if (!st.ok() || eof) {
if (!_delete_file_readers.empty()) {
+ eof = false;
_cur_delete_file_reader = std::move(_delete_file_readers.front());
_delete_file_readers.pop_front();
}
@@ -88,40 +114,58 @@ void IcebergTableReader::filter_rows() {
int64_t row_id = delete_row_ids[i];
int64_t row_range_start = row_id;
int64_t row_range_end = row_id;
- // todo: add debug info
- // todo: asure reading delete file data in file_range only
while (i + 1 < num_deleted_ids) {
if (delete_row_ids[i + 1] == delete_row_ids[i] + 1) {
row_range_end = delete_row_ids[i + 1];
i++;
continue;
} else {
- delete_row_ranges.emplace_back(row_range_start, row_range_end);
+ delete_row_ranges.emplace(row_range_start, row_range_end + 1);
row_range_start = ++row_range_end;
break;
}
}
if (i == num_deleted_ids - 1) {
- delete_row_ranges.emplace_back(row_range_start,
- delete_row_ids[num_deleted_ids - 1]);
+ delete_row_ranges.emplace(row_range_start,
+ delete_row_ids[num_deleted_ids - 1] + 1);
}
row_range_start = delete_row_ids[i + 1];
i++;
}
}
}
+ if (VLOG_IS_ON(3)) {
+ if (!delete_row_ranges.empty()) {
+ std::stringstream out;
+ out << "[";
+ for (const RowRange& delete_row_range : delete_row_ranges) {
+ out << " " << delete_row_range.debug_string();
+ }
+ out << " ]";
+ VLOG_NOTICE << "Delete row range info: " << out.str();
+ }
+ }
ParquetReader* parquet_reader = (ParquetReader*)(_file_format_reader.get());
parquet_reader->merge_delete_row_ranges(delete_row_ranges);
}
-Status IcebergTableReader::init_row_filters() {
- auto& table_desc = _params.table_format_params.iceberg_params;
+Status IcebergTableReader::init_row_filters(const TFileRangeDesc& range) {
+ auto& table_desc = range.table_format_params.iceberg_params;
auto& version = table_desc.format_version;
if (version >= MIN_SUPPORT_DELETE_FILES_VERSION) {
+ SCOPED_TIMER(_iceberg_profile._delete_files_init_time);
auto& delete_file_type = table_desc.content;
auto files = table_desc.delete_files;
if (delete_file_type == POSITON_DELELE) {
// position delete
+ auto row_desc = RowDescriptor(_state->desc_tbl(),
+ std::vector<TupleId>({table_desc.delete_table_tuple_id}),
+ std::vector<bool>({false}));
+ RETURN_IF_ERROR(VExpr::create_expr_tree(
+ _state->obj_pool(), table_desc.file_select_conjunct, &_data_path_conjunct_ctx));
+ RETURN_IF_ERROR(_data_path_conjunct_ctx->prepare(_state, row_desc));
+ RETURN_IF_ERROR(_data_path_conjunct_ctx->open(_state));
+ vector<std::string> names;
for (auto& delete_file : files) {
_position_delete_params.low_bound_index = delete_file.position_lower_bound;
_position_delete_params.upper_bound_index = delete_file.position_upper_bound;
@@ -134,17 +178,29 @@ Status IcebergTableReader::init_row_filters() {
ParquetReader* delete_reader = new ParquetReader(
_profile, _params, delete_range, _state->query_options().batch_size,
const_cast<cctz::time_zone*>(&_state->timezone_obj()));
- FileMetaData* metadata = nullptr;
- RETURN_IF_ERROR(delete_reader->file_metadata(&metadata));
-
- auto& delete_file_schema = metadata->schema();
- vector<std::string> names;
- for (auto i = 0; i < delete_file_schema.size(); ++i) {
- const FieldSchema* field = delete_file_schema.get_column(i);
- _column_schemas.emplace_back(field);
- names.emplace_back(field->name);
+ if (_delete_file_schema == nullptr) {
+ FileMetaData* metadata = nullptr;
+ RETURN_IF_ERROR(delete_reader->file_metadata(&metadata));
+ if (metadata == nullptr) {
+ break;
+ }
+ _delete_file_schema = &metadata->schema();
+ int num_of_col = _delete_file_schema->size();
+ for (auto i = 0; i < num_of_col; ++i) {
+ const FieldSchema* field = _delete_file_schema->get_column(i);
+ _column_schemas.emplace_back(*field);
+ names.emplace_back(field->name);
+ }
}
- Status d_st = delete_reader->init_reader(names, false);
+ DCHECK_EQ(_column_schemas.size(), _delete_file_schema->size());
+ // The expr assure reading delete file data from current file range only
+ Status d_st =
+ delete_reader->init_reader(names, nullptr, _data_path_conjunct_ctx, false);
+ std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
+ partition_columns;
+ std::unordered_map<std::string, VExprContext*> missing_columns;
+ delete_reader->set_fill_columns(partition_columns, missing_columns);
+
_delete_file_readers.emplace_back((GenericReader*)delete_reader);
ParquetReader* parquet_reader = (ParquetReader*)(_file_format_reader.get());
@@ -161,7 +217,7 @@ Status IcebergTableReader::init_row_filters() {
}
}
// todo: equality delete
- filter_rows();
+ filter_rows(range);
return Status::OK();
}
diff --git a/be/src/vec/exec/format/table/iceberg_reader.h b/be/src/vec/exec/format/table/iceberg_reader.h
index e862fe6171..c71ffdaa9c 100644
--- a/be/src/vec/exec/format/table/iceberg_reader.h
+++ b/be/src/vec/exec/format/table/iceberg_reader.h
@@ -23,22 +23,25 @@
#include "table_format_reader.h"
#include "vec/exec/format/generic_reader.h"
+#include "vec/exprs/vexpr.h"
namespace doris::vectorized {
class IcebergTableReader : public TableFormatReader {
public:
IcebergTableReader(GenericReader* file_format_reader, RuntimeProfile* profile,
- RuntimeState* state, const TFileScanRangeParams& params)
- : TableFormatReader(file_format_reader),
- _profile(profile),
- _state(state),
- _params(params) {}
- Status init_row_filters();
- void filter_rows() override;
+ RuntimeState* state, const TFileScanRangeParams& params);
+ ~IcebergTableReader() override;
+ Status init_row_filters(const TFileRangeDesc& range);
+ void filter_rows(const TFileRangeDesc& range) override;
Status get_next_block(Block* block, size_t* read_rows, bool* eof) override;
+ Status set_fill_columns(
+ const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
+ partition_columns,
+ const std::unordered_map<std::string, VExprContext*>& missing_columns) override;
+
Status get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
std::unordered_set<std::string>* missing_cols) override;
@@ -52,13 +55,20 @@ public:
};
private:
+ struct IcebergProfile {
+ RuntimeProfile::Counter* _delete_files_init_time;
+ RuntimeProfile::Counter* _delete_files_read_total_time;
+ };
RuntimeProfile* _profile;
RuntimeState* _state;
const TFileScanRangeParams& _params;
- std::vector<const FieldSchema*> _column_schemas;
+ std::vector<FieldSchema> _column_schemas;
std::deque<std::unique_ptr<GenericReader>> _delete_file_readers;
std::unique_ptr<GenericReader> _cur_delete_file_reader;
PositionDeleteParams _position_delete_params;
+ const FieldDescriptor* _delete_file_schema = nullptr;
+ VExprContext* _data_path_conjunct_ctx = nullptr;
+ IcebergProfile _iceberg_profile;
};
} // namespace doris::vectorized
diff --git a/be/src/vec/exec/format/table/table_format_reader.h b/be/src/vec/exec/format/table/table_format_reader.h
index 3409135dee..188d41a2b9 100644
--- a/be/src/vec/exec/format/table/table_format_reader.h
+++ b/be/src/vec/exec/format/table/table_format_reader.h
@@ -29,6 +29,7 @@ namespace doris::vectorized {
class TableFormatReader : public GenericReader {
public:
TableFormatReader(GenericReader* file_format_reader);
+ ~TableFormatReader() override = default;
Status get_next_block(Block* block, size_t* read_rows, bool* eof) override {
return _file_format_reader->get_next_block(block, read_rows, eof);
}
@@ -37,7 +38,7 @@ public:
return _file_format_reader->get_columns(name_to_type, missing_cols);
}
- virtual void filter_rows() = 0;
+ virtual void filter_rows(const TFileRangeDesc& range) = 0;
protected:
std::string _table_format; // hudi, iceberg
diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp b/be/src/vec/exec/scan/vfile_scanner.cpp
index 6d170a6130..cf61384e81 100644
--- a/be/src/vec/exec/scan/vfile_scanner.cpp
+++ b/be/src/vec/exec/scan/vfile_scanner.cpp
@@ -491,11 +491,11 @@ Status VFileScanner::_get_next_reader() {
}
init_status = parquet_reader->init_reader(_file_col_names, _colname_to_value_range,
_push_down_expr);
- if (_params.__isset.table_format_params &&
- _params.table_format_params.table_format_type == "iceberg") {
+ if (range.__isset.table_format_params &&
+ range.table_format_params.table_format_type == "iceberg") {
IcebergTableReader* iceberg_reader = new IcebergTableReader(
(GenericReader*)parquet_reader, _profile, _state, _params);
- iceberg_reader->init_row_filters();
+ iceberg_reader->init_row_filters(range);
_cur_reader.reset((GenericReader*)iceberg_reader);
} else {
_cur_reader.reset((GenericReader*)parquet_reader);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
index c42c97041b..4117f20539 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
@@ -207,7 +207,7 @@ public class ExternalFileScanNode extends ExternalScanNode {
scanProvider = new HudiScanProvider(hmsTable, desc, columnNameToRange);
break;
case ICEBERG:
- scanProvider = new IcebergScanProvider(hmsTable, desc, columnNameToRange);
+ scanProvider = new IcebergScanProvider(hmsTable, analyzer, desc, columnNameToRange);
break;
case HIVE:
scanProvider = new HiveScanProvider(hmsTable, desc, columnNameToRange);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/HiveSplit.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/HiveSplit.java
index a1a1749c2f..6c8f916a5e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/HiveSplit.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/HiveSplit.java
@@ -23,6 +23,7 @@ import org.apache.hadoop.mapred.FileSplit;
@Data
public class HiveSplit extends FileSplit {
+ public HiveSplit() {}
public HiveSplit(Path file, long start, long length, String[] hosts) {
super(file, start, length, hosts);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergScanProvider.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergScanProvider.java
index 30b8888fd7..7681d6a66d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergScanProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergScanProvider.java
@@ -17,8 +17,19 @@
package org.apache.doris.planner.external;
+import org.apache.doris.analysis.Analyzer;
+import org.apache.doris.analysis.BaseTableRef;
+import org.apache.doris.analysis.BinaryPredicate;
import org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.analysis.StringLiteral;
+import org.apache.doris.analysis.TableName;
+import org.apache.doris.analysis.TableRef;
import org.apache.doris.analysis.TupleDescriptor;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.external.ExternalTable;
import org.apache.doris.catalog.external.HMSExternalTable;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
@@ -26,10 +37,15 @@ import org.apache.doris.common.UserException;
import org.apache.doris.external.iceberg.util.IcebergUtils;
import org.apache.doris.planner.ColumnRange;
import org.apache.doris.thrift.TFileFormatType;
+import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TIcebergDeleteFileDesc;
import org.apache.doris.thrift.TIcebergFileDesc;
+import org.apache.doris.thrift.TIcebergTable;
+import org.apache.doris.thrift.TTableDescriptor;
import org.apache.doris.thrift.TTableFormatFileDesc;
+import org.apache.doris.thrift.TTableType;
+import lombok.Data;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.InputSplit;
@@ -59,13 +75,20 @@ import java.util.OptionalLong;
public class IcebergScanProvider extends HiveScanProvider {
private static final int MIN_DELETE_FILE_SUPPORT_VERSION = 2;
+ public static final String V2_DELETE_TBL = "iceberg#delete#tbl";
+ public static final String V2_DELETE_DB = "iceberg#delete#db";
+ private static final DeleteFileTempTable scanDeleteTable =
+ new DeleteFileTempTable(TableIf.TableType.HMS_EXTERNAL_TABLE);
+ private final Analyzer analyzer;
- public IcebergScanProvider(HMSExternalTable hmsTable, TupleDescriptor desc,
- Map<String, ColumnRange> columnNameToRange) {
+ public IcebergScanProvider(HMSExternalTable hmsTable, Analyzer analyzer, TupleDescriptor desc,
+ Map<String, ColumnRange> columnNameToRange) {
super(hmsTable, desc, columnNameToRange);
+ this.analyzer = analyzer;
}
- public static void setIcebergParams(ExternalFileScanNode.ParamCreateContext context, IcebergSplit icebergSplit) {
+ public static void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSplit)
+ throws UserException {
TTableFormatFileDesc tableFormatFileDesc = new TTableFormatFileDesc();
tableFormatFileDesc.setTableFormatType(icebergSplit.getTableFormatType().value());
TIcebergFileDesc fileDesc = new TIcebergFileDesc();
@@ -74,6 +97,7 @@ public class IcebergScanProvider extends HiveScanProvider {
if (formatVersion < MIN_DELETE_FILE_SUPPORT_VERSION) {
fileDesc.setContent(FileContent.DATA.id());
} else {
+ setPathSelectConjunct(fileDesc, icebergSplit);
for (IcebergDeleteFileFilter filter : icebergSplit.getDeleteFileFilters()) {
TIcebergDeleteFileDesc deleteFileDesc = new TIcebergDeleteFileDesc();
deleteFileDesc.setPath(filter.getDeleteFilePath());
@@ -99,7 +123,20 @@ public class IcebergScanProvider extends HiveScanProvider {
}
}
tableFormatFileDesc.setIcebergParams(fileDesc);
- context.params.setTableFormatParams(tableFormatFileDesc);
+ rangeDesc.setTableFormatParams(tableFormatFileDesc);
+ }
+
+ private static void setPathSelectConjunct(TIcebergFileDesc fileDesc, IcebergSplit icebergSplit)
+ throws UserException {
+ BaseTableRef tableRef = icebergSplit.getDeleteTableRef();
+ fileDesc.setDeleteTableTupleId(tableRef.getDesc().getId().asInt());
+ SlotRef lhs = new SlotRef(tableRef.getName(), DeleteFileTempTable.DATA_FILE_PATH);
+ lhs.analyze(icebergSplit.getAnalyzer());
+ lhs.getDesc().setIsMaterialized(true);
+ StringLiteral rhs = new StringLiteral(icebergSplit.getPath().toUri().toString());
+ BinaryPredicate pathSelectConjunct = new BinaryPredicate(BinaryPredicate.Operator.EQ, lhs, rhs);
+ pathSelectConjunct.analyze(icebergSplit.getAnalyzer());
+ fileDesc.setFileSelectConjunct(pathSelectConjunct.treeToThrift());
}
@Override
@@ -134,18 +171,27 @@ public class IcebergScanProvider extends HiveScanProvider {
scan = scan.filter(predicate);
}
List<InputSplit> splits = new ArrayList<>();
-
int formatVersion = ((BaseTable) table).operations().current().formatVersion();
+ BaseTableRef tableRef = null;
+ if (formatVersion >= MIN_DELETE_FILE_SUPPORT_VERSION) {
+ TableName fullName = analyzer.getFqTableName(scanDeleteTable.getTableName());
+ fullName.analyze(analyzer);
+ TableRef ref = new TableRef(fullName, fullName.toString(), null);
+ tableRef = new BaseTableRef(ref, scanDeleteTable, scanDeleteTable.getTableName());
+ tableRef.analyze(analyzer);
+ }
for (FileScanTask task : scan.planFiles()) {
for (FileScanTask spitTask : task.split(128 * 1024 * 1024)) {
String dataFilePath = spitTask.file().path().toString();
IcebergSplit split = new IcebergSplit(new Path(dataFilePath), spitTask.start(),
spitTask.length(), new String[0]);
split.setFormatVersion(formatVersion);
- if (formatVersion == 2) {
+ if (formatVersion >= MIN_DELETE_FILE_SUPPORT_VERSION) {
split.setDeleteFileFilters(getDeleteFileFilters(spitTask));
+ split.setDeleteTableRef(tableRef);
}
split.setTableFormatType(TableFormatType.ICEBERG);
+ split.setAnalyzer(analyzer);
splits.add(split);
}
}
@@ -193,4 +239,32 @@ public class IcebergScanProvider extends HiveScanProvider {
public List<String> getPathPartitionKeys() throws DdlException, MetaNotFoundException {
return Collections.emptyList();
}
+
+ @Data
+ static class DeleteFileTempTable extends ExternalTable {
+ public static final String DATA_FILE_PATH = "file_path";
+ private final TableName tableName;
+ private final List<Column> fullSchema = new ArrayList<>();
+
+ public DeleteFileTempTable(TableType type) {
+ super(0, V2_DELETE_TBL, null, V2_DELETE_DB, type);
+ this.tableName = new TableName(null, V2_DELETE_DB, V2_DELETE_TBL);
+ Column dataFilePathCol = new Column(DATA_FILE_PATH, PrimitiveType.STRING, true);
+ this.fullSchema.add(dataFilePathCol);
+ }
+
+ @Override
+ public List<Column> getFullSchema() {
+ return fullSchema;
+ }
+
+ @Override
+ public TTableDescriptor toThrift() {
+ TIcebergTable tIcebergTable = new TIcebergTable(V2_DELETE_DB, V2_DELETE_TBL, new HashMap<>());
+ TTableDescriptor tTableDescriptor = new TTableDescriptor(getId(), TTableType.ICEBERG_TABLE,
+ fullSchema.size(), 0, getName(), "");
+ tTableDescriptor.setIcebergTable(tIcebergTable);
+ return tTableDescriptor;
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergSplit.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergSplit.java
index 05ae3c69d3..14fce6caf8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergSplit.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/IcebergSplit.java
@@ -17,6 +17,9 @@
package org.apache.doris.planner.external;
+import org.apache.doris.analysis.Analyzer;
+import org.apache.doris.analysis.BaseTableRef;
+
import lombok.Data;
import org.apache.hadoop.fs.Path;
@@ -28,8 +31,10 @@ public class IcebergSplit extends HiveSplit {
super(file, start, length, hosts);
}
+ private Analyzer analyzer;
private String dataFilePath;
private Integer formatVersion;
+ private BaseTableRef deleteTableRef;
private List<IcebergDeleteFileFilter> deleteFileFilters;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java
index 3066b7354a..a2ce044bb7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java
@@ -73,9 +73,6 @@ public abstract class QueryScanProvider implements FileScanProviderIf {
context.params.setFileAttributes(getFileAttributes());
}
- if (inputSplit instanceof IcebergSplit) {
- IcebergScanProvider.setIcebergParams(context, (IcebergSplit) inputSplit);
- }
// set hdfs params for hdfs file type.
Map<String, String> locationProperties = getLocationProperties();
if (locationType == TFileType.FILE_HDFS) {
@@ -96,7 +93,6 @@ public abstract class QueryScanProvider implements FileScanProviderIf {
} else if (locationType == TFileType.FILE_S3) {
context.params.setProperties(locationProperties);
}
-
TScanRangeLocations curLocations = newLocations(context.params, backendPolicy);
FileSplitStrategy fileSplitStrategy = new FileSplitStrategy();
@@ -108,7 +104,10 @@ public abstract class QueryScanProvider implements FileScanProviderIf {
pathPartitionKeys, false);
TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys);
-
+ // external data lake table
+ if (split instanceof IcebergSplit) {
+ IcebergScanProvider.setIcebergParams(rangeDesc, (IcebergSplit) inputSplit);
+ }
curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc);
LOG.debug("assign to backend {} with table split: {} ({}, {}), location: {}",
curLocations.getLocations().get(0).getBackendId(), fileSplit.getPath(), fileSplit.getStart(),
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 2725c67acc..0dd3373b74 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -268,7 +268,10 @@ struct TIcebergFileDesc {
1: optional i32 format_version;
// Iceberg file type, 0: data, 1: position delete, 2: equality delete.
2: optional i32 content;
+ // When open a delete file, filter the data file path with the 'file_path' property
3: optional list<TIcebergDeleteFileDesc> delete_files;
+ 4: optional Types.TTupleId delete_table_tuple_id;
+ 5: optional Exprs.TExpr file_select_conjunct;
}
struct TTableFormatFileDesc {
@@ -308,7 +311,7 @@ struct TFileScanRangeParams {
14: optional list<Types.TNetworkAddress> broker_addresses
15: optional TFileAttributes file_attributes
16: optional Exprs.TExpr pre_filter_exprs
- // For data lake table format
+ // Deprecated, For data lake table format
17: optional TTableFormatFileDesc table_format_params
// For csv query task, same the column index in file, order by dest_tuple
18: optional list<i32> column_idxs
@@ -329,6 +332,8 @@ struct TFileRangeDesc {
6: optional list<string> columns_from_path;
// column names from file path, in the same order with columns_from_path
7: optional list<string> columns_from_path_keys;
+ // For data lake table format
+ 8: optional TTableFormatFileDesc table_format_params
}
// TFileScanRange represents a set of descriptions of a file and the rules for reading and converting it.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 16/21: [fix](load) delta writer init failed might cause data inconsistency between multiple replicas (#15058)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 22aacd849c1638f6fcd87cd1ad58f70eed5fbe8f
Author: zhannngchen <48...@users.noreply.github.com>
AuthorDate: Fri Dec 16 22:07:00 2022 +0800
[fix](load) delta writer init failed might cause data inconsistency between multiple replicas (#15058)
In the following case, data inconsistency would happen between multiple replicas
current delta writer only writes a few lines of data (which meas the write() method only called once)
writer failed when init()(which is called at the fist time we call write()), and current tablet is recorded in _broken_tablets
delta writer closed, and in the close() method, delta writer found it's not inited, treat such case as an empty load, it will try to init again, which would create an empty rowset.
tablet sink received the error report in rpc response, marked the replica as failed, but since the quorum replicas are succeed, so the following load commit operation will succeed.
FE send publish version task to each be, the one with empty rowset will publish version successfully.
We got 2 replica with data and 1 empty replica.
---
be/src/olap/delta_writer.cpp | 4 ++--
be/src/runtime/tablets_channel.cpp | 17 ++++++++++++-----
2 files changed, 14 insertions(+), 7 deletions(-)
diff --git a/be/src/olap/delta_writer.cpp b/be/src/olap/delta_writer.cpp
index d330e430fe..75f402f202 100644
--- a/be/src/olap/delta_writer.cpp
+++ b/be/src/olap/delta_writer.cpp
@@ -318,7 +318,7 @@ Status DeltaWriter::close() {
// if this delta writer is not initialized, but close() is called.
// which means this tablet has no data loaded, but at least one tablet
// in same partition has data loaded.
- // so we have to also init this DeltaWriter, so that it can create a empty rowset
+ // so we have to also init this DeltaWriter, so that it can create an empty rowset
// for this tablet when being closed.
RETURN_NOT_OK(init());
}
@@ -409,7 +409,7 @@ Status DeltaWriter::cancel() {
Status DeltaWriter::cancel_with_status(const Status& st) {
std::lock_guard<std::mutex> l(_lock);
- if (!_is_init || _is_cancelled) {
+ if (_is_cancelled) {
return Status::OK();
}
_mem_table.reset();
diff --git a/be/src/runtime/tablets_channel.cpp b/be/src/runtime/tablets_channel.cpp
index d6c81fb845..5bee7e2050 100644
--- a/be/src/runtime/tablets_channel.cpp
+++ b/be/src/runtime/tablets_channel.cpp
@@ -121,6 +121,14 @@ Status TabletsChannel::close(
// just skip this tablet(writer) and continue to close others
continue;
}
+ // to make sure tablet writer in `_broken_tablets` won't call `close_wait` method.
+ // `close_wait` might create the rowset and commit txn directly, and the subsequent
+ // publish version task will success, which can cause the replica inconsistency.
+ if (_broken_tablets.find(it.second->tablet_id()) != _broken_tablets.end()) {
+ LOG(WARNING) << "SHOULD NOT HAPPEN, tablet writer is broken but not cancelled"
+ << ", tablet_id=" << it.first << ", transaction_id=" << _txn_id;
+ continue;
+ }
need_wait_writers.insert(it.second);
} else {
auto st = it.second->cancel();
@@ -182,11 +190,9 @@ void TabletsChannel::_close_wait(DeltaWriter* writer,
const bool write_single_replica) {
Status st = writer->close_wait(slave_tablet_nodes, write_single_replica);
if (st.ok()) {
- if (_broken_tablets.find(writer->tablet_id()) == _broken_tablets.end()) {
- PTabletInfo* tablet_info = tablet_vec->Add();
- tablet_info->set_tablet_id(writer->tablet_id());
- tablet_info->set_schema_hash(writer->schema_hash());
- }
+ PTabletInfo* tablet_info = tablet_vec->Add();
+ tablet_info->set_tablet_id(writer->tablet_id());
+ tablet_info->set_schema_hash(writer->schema_hash());
} else {
PTabletError* tablet_error = tablet_errors->Add();
tablet_error->set_tablet_id(writer->tablet_id());
@@ -487,6 +493,7 @@ Status TabletsChannel::add_batch(const TabletWriterAddRequest& request,
PTabletError* error = tablet_errors->Add();
error->set_tablet_id(tablet_to_rowidxs_it.first);
error->set_msg(err_msg);
+ tablet_writer_it->second->cancel_with_status(st);
_broken_tablets.insert(tablet_to_rowidxs_it.first);
// continue write to other tablet.
// the error will return back to sender.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 07/21: [fix](multi catalog)Return emtpy block while external table scanner couldn't find the file (#14997)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit d41a963c8ff0c971baa9080417b343601f8a8ade
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Fri Dec 16 09:36:35 2022 +0800
[fix](multi catalog)Return emtpy block while external table scanner couldn't find the file (#14997)
FE file path cache for external table may out of date. In this case, BE may fail to find the not exist file from FE cache.
This pr is to handle this case: instead of throw an error message to the user, we return empty result set to the user.
---
be/src/io/hdfs_file_reader.cpp | 3 +++
be/src/vec/exec/scan/scanner_scheduler.cpp | 12 +++++++++++-
be/src/vec/exec/scan/vfile_scanner.cpp | 3 +++
3 files changed, 17 insertions(+), 1 deletion(-)
diff --git a/be/src/io/hdfs_file_reader.cpp b/be/src/io/hdfs_file_reader.cpp
index d57c489691..78a21ac477 100644
--- a/be/src/io/hdfs_file_reader.cpp
+++ b/be/src/io/hdfs_file_reader.cpp
@@ -72,6 +72,9 @@ Status HdfsFileReader::open() {
RETURN_IF_ERROR(HdfsFsCache::instance()->get_connection(_hdfs_params, &_fs_handle));
_hdfs_fs = _fs_handle->hdfs_fs;
+ if (hdfsExists(_hdfs_fs, _path.c_str()) != 0) {
+ return Status::NotFound("{} not exists!", _path);
+ }
_hdfs_file = hdfsOpenFile(_hdfs_fs, _path.c_str(), O_RDONLY, 0, 0, 0);
if (_hdfs_file == nullptr) {
if (_fs_handle->from_cache) {
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
index f2170a9599..d45adc13b9 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -26,6 +26,7 @@
#include "vec/core/block.h"
#include "vec/exec/scan/vscanner.h"
#include "vec/exprs/vexpr.h"
+#include "vfile_scanner.h"
namespace doris::vectorized {
@@ -232,12 +233,21 @@ void ScannerScheduler::_scanner_scan(ScannerScheduler* scheduler, ScannerContext
auto block = ctx->get_free_block(&get_free_block);
status = scanner->get_block(state, block, &eos);
VLOG_ROW << "VOlapScanNode input rows: " << block->rows() << ", eos: " << eos;
- if (!status.ok()) {
+ // The VFileScanner for external table may try to open not exist files,
+ // Because FE file cache for external table may out of date.
+ if (!status.ok() && (typeid(*scanner) == typeid(doris::vectorized::VFileScanner) &&
+ !status.is<ErrorCode::NOT_FOUND>())) {
LOG(WARNING) << "Scan thread read VOlapScanner failed: " << status.to_string();
// Add block ptr in blocks, prevent mem leak in read failed
blocks.push_back(block);
break;
}
+ if (status.is<ErrorCode::NOT_FOUND>()) {
+ // The only case in this if branch is external table file delete and fe cache has not been updated yet.
+ // Set status to OK.
+ status = Status::OK();
+ eos = true;
+ }
raw_bytes_read += block->bytes();
num_rows_in_block += block->rows();
diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp b/be/src/vec/exec/scan/vfile_scanner.cpp
index ee1605a6e0..b478d5c96e 100644
--- a/be/src/vec/exec/scan/vfile_scanner.cpp
+++ b/be/src/vec/exec/scan/vfile_scanner.cpp
@@ -530,6 +530,9 @@ Status VFileScanner::_get_next_reader() {
if (init_status.is_end_of_file()) {
continue;
} else if (!init_status.ok()) {
+ if (init_status.is<ErrorCode::NOT_FOUND>()) {
+ return init_status;
+ }
return Status::InternalError("failed to init reader for file {}, err: {}", range.path,
init_status.get_error_msg());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 15/21: [fix](subquery)fix bug of using constexpr as subquery's output (#15119)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 4a1bcebf411cba02228ea35347e5d39dd7569fba
Author: starocean999 <40...@users.noreply.github.com>
AuthorDate: Fri Dec 16 21:58:58 2022 +0800
[fix](subquery)fix bug of using constexpr as subquery's output (#15119)
---
.../apache/doris/planner/SingleNodePlanner.java | 4 +++-
.../test_outer_join_with_subquery.out | 3 +++
.../test_outer_join_with_subquery.groovy | 25 ++++++++++++++++++++++
3 files changed, 31 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java
index 5e6b779514..5e1973aaed 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java
@@ -1608,7 +1608,9 @@ public class SingleNodePlanner {
}
// Set output smap of rootNode *before* creating a SelectNode for proper resolution.
rootNode.setOutputSmap(outputSmap);
-
+ if (rootNode instanceof UnionNode && ((UnionNode) rootNode).isConstantUnion()) {
+ rootNode.setWithoutTupleIsNullOutputSmap(outputSmap);
+ }
// rootNode.setOutputSmap(ExprSubstitutionMap.compose(inlineViewRef.getBaseTblSmap(),
// rootNode.getOutputSmap(), analyzer));
// Expr.substituteList(inlineViewRef.getViewStmt().getResultExprs(), analyzer.getChangeResSmap());
diff --git a/regression-test/data/correctness_p0/test_outer_join_with_subquery.out b/regression-test/data/correctness_p0/test_outer_join_with_subquery.out
index 72d126351a..875d88ccc5 100644
--- a/regression-test/data/correctness_p0/test_outer_join_with_subquery.out
+++ b/regression-test/data/correctness_p0/test_outer_join_with_subquery.out
@@ -2,3 +2,6 @@
-- !select --
1
+-- !select --
+1
+
diff --git a/regression-test/suites/correctness_p0/test_outer_join_with_subquery.groovy b/regression-test/suites/correctness_p0/test_outer_join_with_subquery.groovy
index 59bb822e3b..97de116ca1 100644
--- a/regression-test/suites/correctness_p0/test_outer_join_with_subquery.groovy
+++ b/regression-test/suites/correctness_p0/test_outer_join_with_subquery.groovy
@@ -78,6 +78,31 @@ suite("test_outer_join_with_subquery") {
) as subq_2 on (subq_1.c1 = subq_2.c2);
"""
+ qt_select"""
+ with idm_org_table as (
+ SELECT
+ 1 a
+ )
+ select
+ a
+ from
+ test_outer_join_with_subquery_outerjoin_A
+ where
+ a in (
+ SELECT
+ a
+ from
+ test_outer_join_with_subquery_outerjoin_B
+ where
+ a = (
+ select
+ a
+ from
+ idm_org_table
+ )
+ );
+ """
+
sql """
drop table if exists test_outer_join_with_subquery_outerjoin_A;
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 12/21: [Bug](datev2) Fix compatible problems caused by datev2 (#15131)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit d6f4f08dcc14c435cb721e5009c162478a1ab25e
Author: Gabriel <ga...@gmail.com>
AuthorDate: Fri Dec 16 21:52:39 2022 +0800
[Bug](datev2) Fix compatible problems caused by datev2 (#15131)
This bug is introduced by #15094
---
.../org/apache/doris/analysis/DateLiteral.java | 37 ++++++++++++++++++++++
.../org/apache/doris/catalog/PartitionKey.java | 10 ++++++
2 files changed, 47 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java
index 2a6efebd40..2377351822 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java
@@ -769,6 +769,33 @@ public class DateLiteral extends LiteralExpr {
}
}
+ @Override
+ public void checkValueValid() throws AnalysisException {
+ if (year < 0 || year > 9999) {
+ throw new AnalysisException("DateLiteral has invalid year value: " + year);
+ }
+ if (month < 1 || month > 12) {
+ throw new AnalysisException("DateLiteral has invalid month value: " + month);
+ }
+ if (day < 1 || day > 31) {
+ throw new AnalysisException("DateLiteral has invalid day value: " + day);
+ }
+ if (type.isDatetimeV2() || type.isDatetime()) {
+ if (hour < 0 || hour > 24) {
+ throw new AnalysisException("DateLiteral has invalid hour value: " + hour);
+ }
+ if (minute < 0 || minute > 60) {
+ throw new AnalysisException("DateLiteral has invalid minute value: " + minute);
+ }
+ if (second < 0 || second > 60) {
+ throw new AnalysisException("DateLiteral has invalid second value: " + second);
+ }
+ if (type.isDatetimeV2() && (microsecond < 0 || microsecond > 999999)) {
+ throw new AnalysisException("DateLiteral has invalid microsecond value: " + microsecond);
+ }
+ }
+ }
+
public static DateLiteral read(DataInput in) throws IOException {
DateLiteral literal = new DateLiteral();
literal.readFields(in);
@@ -1621,4 +1648,14 @@ public class DateLiteral extends LiteralExpr {
throw new AnalysisException("Datetime value is out of range: " + dateStr);
}
}
+
+ public void setMinValue() {
+ year = 0;
+ month = 1;
+ day = 1;
+ hour = 0;
+ minute = 0;
+ second = 0;
+ microsecond = 0;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
index 68b49c2974..6f26d4ff3a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
@@ -338,6 +338,16 @@ public class PartitionKey implements Comparable<PartitionKey>, Writable {
if (type != PrimitiveType.DATETIMEV2) {
literal.setType(Type.fromPrimitiveType(type));
}
+ if (type.isDateV2Type()) {
+ try {
+ literal.checkValueValid();
+ } catch (AnalysisException e) {
+ LOG.warn("Value {} for partition key [type = {}] is invalid! This is a bug exists in Doris "
+ + "1.2.0 and fixed since Doris 1.2.1. You should create this table again using Doris "
+ + "1.2.1+ .", literal.getStringValue(), type);
+ ((DateLiteral) literal).setMinValue();
+ }
+ }
keys.add(literal);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 02/21: [Enhancement](partition prune): calculate the column ranges of compound predicates (#14886)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 5fce045f51a69d28e8824f3fc475b862f83821e4
Author: spaces-x <we...@gmail.com>
AuthorDate: Thu Dec 15 20:47:44 2022 +0800
[Enhancement](partition prune): calculate the column ranges of compound predicates (#14886)
Doris does not support disjunctive predicates very well, which causes some problems in partition prune.
For example, sqls like the followings will trigger a full table scan without partition pruning
select * from test.t1
where (dt between 20211121 and 20211122) or (dt between 20211125 and 20211126)
---
.../java/org/apache/doris/planner/ScanNode.java | 83 ++++++++++++++++++++++
.../doris/analysis/RangePartitionPruneTest.java | 8 +++
2 files changed, 91 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
index 8523830c66..460ea01b7b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
@@ -43,7 +43,9 @@ import com.google.common.base.MoreObjects;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Range;
+import com.google.common.collect.RangeSet;
import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeSet;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -260,6 +262,26 @@ public abstract class ScanNode extends PlanNode {
ColumnBound bound = ColumnBound.of((LiteralExpr) inPredicate.getChild(i));
result.add(Range.closed(bound, bound));
}
+ } else if (expr instanceof CompoundPredicate) {
+ CompoundPredicate compoundPredicate = (CompoundPredicate) expr;
+ ColumnRanges leftChildRange = null;
+ ColumnRanges rightChildRange = null;
+ switch (compoundPredicate.getOp()) {
+ case AND:
+ leftChildRange = expressionToRanges(compoundPredicate.getChild(0), desc);
+ rightChildRange = expressionToRanges(compoundPredicate.getChild(1), desc);
+ return leftChildRange.intersectRanges(rightChildRange);
+ case OR:
+ leftChildRange = expressionToRanges(compoundPredicate.getChild(0), desc);
+ rightChildRange = expressionToRanges(compoundPredicate.getChild(1), desc);
+ return leftChildRange.unionRanges(rightChildRange);
+ case NOT:
+ leftChildRange = expressionToRanges(compoundPredicate.getChild(0), desc);
+ return leftChildRange.complementOfRanges();
+ default:
+ throw new RuntimeException("unknown OP in compound predicate: "
+ + compoundPredicate.getOp().toString());
+ }
}
if (result.isEmpty()) {
@@ -377,6 +399,67 @@ public abstract class ScanNode extends PlanNode {
return IS_NULL;
}
+ public ColumnRanges complementOfRanges() {
+ if (type == Type.CONVERT_SUCCESS) {
+ RangeSet<ColumnBound> rangeSet = TreeRangeSet.create();
+ rangeSet.addAll(ranges);
+ return create(Lists.newArrayList(rangeSet.complement().asRanges()));
+ }
+ return CONVERT_FAILURE;
+ }
+
+ public ColumnRanges intersectRanges(ColumnRanges other) {
+ // intersect ranges can handle isnull
+ switch (this.type) {
+ case IS_NULL:
+ return createIsNull();
+ case CONVERT_FAILURE:
+ return createFailure();
+ case CONVERT_SUCCESS:
+ switch (other.type) {
+ case IS_NULL:
+ return createIsNull();
+ case CONVERT_FAILURE:
+ return createFailure();
+ case CONVERT_SUCCESS:
+ RangeSet<ColumnBound> rangeSet = TreeRangeSet.create();
+ rangeSet.addAll(this.ranges);
+ RangeSet<ColumnBound> intersectSet = TreeRangeSet.create();
+
+ other.ranges.forEach(range -> intersectSet.addAll(rangeSet.subRangeSet(range)));
+ return create(Lists.newArrayList(intersectSet.asRanges()));
+ default:
+ return createFailure();
+ }
+ default:
+ return createFailure();
+ }
+ }
+
+ public ColumnRanges unionRanges(ColumnRanges other) {
+ switch (this.type) {
+ case IS_NULL:
+ case CONVERT_FAILURE:
+ return createFailure();
+ case CONVERT_SUCCESS:
+ switch (other.type) {
+ case IS_NULL:
+ case CONVERT_FAILURE:
+ return createFailure();
+ case CONVERT_SUCCESS:
+ RangeSet<ColumnBound> rangeSet = TreeRangeSet.create();
+ rangeSet.addAll(this.ranges);
+ rangeSet.addAll(other.ranges);
+ List<Range<ColumnBound>> unionRangeList = Lists.newArrayList(rangeSet.asRanges());
+ return create(unionRangeList);
+ default:
+ return createFailure();
+ }
+ default:
+ return createFailure();
+ }
+ }
+
public static ColumnRanges createFailure() {
return CONVERT_FAILURE;
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/analysis/RangePartitionPruneTest.java b/fe/fe-core/src/test/java/org/apache/doris/analysis/RangePartitionPruneTest.java
index 6ab003e5fd..8c60f543f4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/analysis/RangePartitionPruneTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/analysis/RangePartitionPruneTest.java
@@ -185,6 +185,14 @@ public class RangePartitionPruneTest extends PartitionPruneTestBase {
// maybe something goes wrong with ExtractCommonFactorsRule.
addCase("select * from test.t1 where ((dt=20211123 and k1=1) or (dt=20211125 and k1=3)) and k2>0", "partitions=8/8", "partitions=8/8");
addCase("select * from test.t2 where k1 > 10 or k2 < 1", "partitions=9/9", "partitions=9/9");
+ // add some cases for CompoundPredicate
+ addCase("select * from test.t1 where (dt >= 20211121 and dt <= 20211122) or (dt >= 20211123 and dt <= 20211125)",
+ "partitions=8/8", "partitions=5/8");
+ addCase("select * from test.t1 where (dt between 20211121 and 20211122) or (dt between 20211123 and 20211125)",
+ "partitions=8/8", "partitions=5/8");
+ addCase("select * from test.t1 where (dt between 20211121 and 20211122) or dt is null or (dt between 20211123 and 20211125)",
+ "partitions=8/8", "partitions=6/8");
+
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 19/21: [enhancement](signal) output query_id when 'be' core dumped #15080
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit d79b6017963ac1907a61bc13e955fe097d7b52b3
Author: Tiewei Fang <43...@users.noreply.github.com>
AuthorDate: Mon Dec 19 09:47:38 2022 +0800
[enhancement](signal) output query_id when 'be' core dumped #15080
---
be/src/runtime/fragment_mgr.cpp | 5 -----
be/src/runtime/thread_context.cpp | 3 +++
2 files changed, 3 insertions(+), 5 deletions(-)
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index b4e0595451..9bf0a361cd 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -26,7 +26,6 @@
#include "agent/cgroups_mgr.h"
#include "common/object_pool.h"
-#include "common/signal_handler.h"
#include "gen_cpp/FrontendService.h"
#include "gen_cpp/PaloInternalService_types.h"
#include "gen_cpp/PlanNodes_types.h"
@@ -486,10 +485,6 @@ void FragmentMgr::_exec_actual(std::shared_ptr<FragmentExecState> exec_state, Fi
span->SetAttribute("query_id", print_id(exec_state->query_id()));
span->SetAttribute("instance_id", print_id(exec_state->fragment_instance_id()));
- // these two are used to output query_id when be cored dump.
- doris::signal::query_id_hi = exec_state->query_id().hi;
- doris::signal::query_id_lo = exec_state->query_id().lo;
-
LOG_INFO(func_name)
.tag("query_id", exec_state->query_id())
.tag("instance_id", exec_state->fragment_instance_id())
diff --git a/be/src/runtime/thread_context.cpp b/be/src/runtime/thread_context.cpp
index 6caafdc5db..d30d836064 100644
--- a/be/src/runtime/thread_context.cpp
+++ b/be/src/runtime/thread_context.cpp
@@ -17,6 +17,7 @@
#include "runtime/thread_context.h"
+#include "common/signal_handler.h"
#include "runtime/runtime_state.h"
#include "util/doris_metrics.h"
@@ -44,6 +45,8 @@ AttachTask::AttachTask(const std::shared_ptr<MemTrackerLimiter>& mem_tracker,
}
AttachTask::AttachTask(RuntimeState* runtime_state) {
+ doris::signal::query_id_hi = runtime_state->query_id().hi;
+ doris::signal::query_id_lo = runtime_state->query_id().lo;
thread_context()->attach_task(print_id(runtime_state->query_id()),
runtime_state->fragment_instance_id(),
runtime_state->query_mem_tracker());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 09/21: [vectorized](join) add try catch in create thread (#15065)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 8853d515ae7ef22a19d1ee200aed8c54a5e95639
Author: zhangstar333 <87...@users.noreply.github.com>
AuthorDate: Fri Dec 16 19:55:09 2022 +0800
[vectorized](join) add try catch in create thread (#15065)
---
be/src/vec/exec/join/vjoin_node_base.cpp | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
diff --git a/be/src/vec/exec/join/vjoin_node_base.cpp b/be/src/vec/exec/join/vjoin_node_base.cpp
index d5d5280f57..2ebedff219 100644
--- a/be/src/vec/exec/join/vjoin_node_base.cpp
+++ b/be/src/vec/exec/join/vjoin_node_base.cpp
@@ -159,11 +159,16 @@ Status VJoinNodeBase::open(RuntimeState* state) {
RETURN_IF_CANCELLED(state);
std::promise<Status> thread_status;
- std::thread([this, state, thread_status_p = &thread_status,
- parent_span = opentelemetry::trace::Tracer::GetCurrentSpan()] {
- OpentelemetryScope scope {parent_span};
- this->_probe_side_open_thread(state, thread_status_p);
- }).detach();
+ try {
+ std::thread([this, state, thread_status_p = &thread_status,
+ parent_span = opentelemetry::trace::Tracer::GetCurrentSpan()] {
+ OpentelemetryScope scope {parent_span};
+ this->_probe_side_open_thread(state, thread_status_p);
+ }).detach();
+ } catch (const std::system_error& e) {
+ LOG(WARNING) << "In VJoinNodeBase::open create thread fail, " << e.what();
+ return Status::InternalError(e.what());
+ }
// Open the probe-side child so that it may perform any initialisation in parallel.
// Don't exit even if we see an error, we still need to wait for the build thread
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 06/21: [Bug](case function) do not crash if prepare failed (#15113)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit e68a49f3fe03c8a23ba67f4c5185080b545e58bb
Author: Gabriel <ga...@gmail.com>
AuthorDate: Fri Dec 16 11:02:05 2022 +0800
[Bug](case function) do not crash if prepare failed (#15113)
---
be/src/vec/exprs/vcase_expr.cpp | 12 ------------
be/src/vec/functions/function_case.h | 9 +--------
2 files changed, 1 insertion(+), 20 deletions(-)
diff --git a/be/src/vec/exprs/vcase_expr.cpp b/be/src/vec/exprs/vcase_expr.cpp
index d8f067f6b8..a5614333ba 100644
--- a/be/src/vec/exprs/vcase_expr.cpp
+++ b/be/src/vec/exprs/vcase_expr.cpp
@@ -63,23 +63,11 @@ Status VCaseExpr::open(RuntimeState* state, VExprContext* context,
FunctionContext::FunctionStateScope scope) {
RETURN_IF_ERROR(VExpr::open(state, context, scope));
RETURN_IF_ERROR(VExpr::init_function_context(context, scope, _function));
- if (scope == doris_udf::FunctionContext::FRAGMENT_LOCAL) {
- auto* case_state = new CaseState {_data_type};
- context->fn_context(_fn_context_index)
- ->set_function_state(FunctionContext::FRAGMENT_LOCAL, case_state);
- }
return Status::OK();
}
void VCaseExpr::close(RuntimeState* state, VExprContext* context,
FunctionContext::FunctionStateScope scope) {
- if (scope == doris_udf::FunctionContext::FRAGMENT_LOCAL) {
- auto* case_state = reinterpret_cast<CaseState*>(
- context->fn_context(_fn_context_index)
- ->get_function_state(FunctionContext::FRAGMENT_LOCAL));
- delete case_state;
- }
-
VExpr::close_function_context(context, scope, _function);
VExpr::close(state, context, scope);
}
diff --git a/be/src/vec/functions/function_case.h b/be/src/vec/functions/function_case.h
index f3961c4eef..5c2b39d7f8 100644
--- a/be/src/vec/functions/function_case.h
+++ b/be/src/vec/functions/function_case.h
@@ -27,10 +27,6 @@
namespace doris::vectorized {
-struct CaseState {
- DataTypePtr result_type = nullptr;
-};
-
template <bool has_case, bool has_else>
struct FunctionCaseName;
@@ -375,10 +371,7 @@ public:
Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
size_t result, size_t input_rows_count) override {
- auto* case_state = reinterpret_cast<CaseState*>(
- context->get_function_state(FunctionContext::FRAGMENT_LOCAL));
-
- return execute_get_type(case_state->result_type, block, arguments, result,
+ return execute_get_type(block.get_by_position(result).type, block, arguments, result,
input_rows_count);
}
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 05/21: [fix](explain) fix explain output format problem (#15019)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 315f0396857aaa5d93e062b27062417bb24a97e2
Author: luozenglin <37...@users.noreply.github.com>
AuthorDate: Fri Dec 16 10:53:05 2022 +0800
[fix](explain) fix explain output format problem (#15019)
---
.../src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java
index eb09d679e6..7a34084ec8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java
@@ -244,7 +244,7 @@ public class NestedLoopJoinNode extends JoinNodeBase {
if (!runtimeFilters.isEmpty()) {
output.append(detailPrefix).append("runtime filters: ");
output.append(getRuntimeFilterExplainString(true));
- output.append("isOutputLeftSideOnly: ").append(isOutputLeftSideOnly).append("\n");
+ output.append(detailPrefix).append("is output left side only: ").append(isOutputLeftSideOnly).append("\n");
}
output.append(detailPrefix).append(String.format("cardinality=%,d", cardinality)).append("\n");
// todo unify in plan node
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 14/21: [vectorized](jdbc) fix external table of oracle with condition about … (#15092)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit cc139837665d6f1fb3403642978f7ecd09ecca57
Author: zhangstar333 <87...@users.noreply.github.com>
AuthorDate: Fri Dec 16 10:48:17 2022 +0800
[vectorized](jdbc) fix external table of oracle with condition about … (#15092)
* [vectorized](jdbc) fix external table of oracle with condition about datetime report error
* formatter
---
be/src/vec/exec/scan/new_jdbc_scan_node.cpp | 2 +-
be/src/vec/exec/scan/new_jdbc_scanner.cpp | 7 ++++++-
be/src/vec/exec/scan/new_jdbc_scanner.h | 2 +-
be/src/vec/exec/scan/new_odbc_scan_node.cpp | 2 +-
be/src/vec/exec/scan/new_odbc_scanner.cpp | 6 +++++-
be/src/vec/exec/scan/new_odbc_scanner.h | 2 +-
.../src/main/java/org/apache/doris/planner/OdbcScanNode.java | 8 ++++++++
7 files changed, 23 insertions(+), 6 deletions(-)
diff --git a/be/src/vec/exec/scan/new_jdbc_scan_node.cpp b/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
index e86060ad41..15ddd607ba 100644
--- a/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
@@ -52,7 +52,7 @@ Status NewJdbcScanNode::_init_scanners(std::list<VScanner*>* scanners) {
NewJdbcScanner* scanner =
new NewJdbcScanner(_state, this, _limit_per_scanner, _tuple_id, _query_string);
_scanner_pool.add(scanner);
- RETURN_IF_ERROR(scanner->prepare(_state));
+ RETURN_IF_ERROR(scanner->prepare(_state, _vconjunct_ctx_ptr.get()));
scanners->push_back(static_cast<VScanner*>(scanner));
return Status::OK();
}
diff --git a/be/src/vec/exec/scan/new_jdbc_scanner.cpp b/be/src/vec/exec/scan/new_jdbc_scanner.cpp
index b91300bf47..ef9004530d 100644
--- a/be/src/vec/exec/scan/new_jdbc_scanner.cpp
+++ b/be/src/vec/exec/scan/new_jdbc_scanner.cpp
@@ -27,8 +27,13 @@ NewJdbcScanner::NewJdbcScanner(RuntimeState* state, NewJdbcScanNode* parent, int
_query_string(query_string),
_tuple_desc(nullptr) {}
-Status NewJdbcScanner::prepare(RuntimeState* state) {
+Status NewJdbcScanner::prepare(RuntimeState* state, VExprContext** vconjunct_ctx_ptr) {
VLOG_CRITICAL << "NewJdbcScanner::Prepare";
+ if (vconjunct_ctx_ptr != nullptr) {
+ // Copy vconjunct_ctx_ptr from scan node to this scanner's _vconjunct_ctx.
+ RETURN_IF_ERROR((*vconjunct_ctx_ptr)->clone(state, &_vconjunct_ctx));
+ }
+
if (_is_init) {
return Status::OK();
}
diff --git a/be/src/vec/exec/scan/new_jdbc_scanner.h b/be/src/vec/exec/scan/new_jdbc_scanner.h
index 24c2649dfe..f5584cd5d0 100644
--- a/be/src/vec/exec/scan/new_jdbc_scanner.h
+++ b/be/src/vec/exec/scan/new_jdbc_scanner.h
@@ -32,7 +32,7 @@ public:
Status close(RuntimeState* state) override;
public:
- Status prepare(RuntimeState* state);
+ Status prepare(RuntimeState* state, VExprContext** vconjunct_ctx_ptr);
protected:
Status _get_block_impl(RuntimeState* state, Block* block, bool* eos) override;
diff --git a/be/src/vec/exec/scan/new_odbc_scan_node.cpp b/be/src/vec/exec/scan/new_odbc_scan_node.cpp
index 571566b88a..db8f8bfc68 100644
--- a/be/src/vec/exec/scan/new_odbc_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_odbc_scan_node.cpp
@@ -53,7 +53,7 @@ Status NewOdbcScanNode::_init_scanners(std::list<VScanner*>* scanners) {
}
NewOdbcScanner* scanner = new NewOdbcScanner(_state, this, _limit_per_scanner, _odbc_scan_node);
_scanner_pool.add(scanner);
- RETURN_IF_ERROR(scanner->prepare(_state));
+ RETURN_IF_ERROR(scanner->prepare(_state, _vconjunct_ctx_ptr.get()));
scanners->push_back(static_cast<VScanner*>(scanner));
return Status::OK();
}
diff --git a/be/src/vec/exec/scan/new_odbc_scanner.cpp b/be/src/vec/exec/scan/new_odbc_scanner.cpp
index 59c94cf887..71b1a8d525 100644
--- a/be/src/vec/exec/scan/new_odbc_scanner.cpp
+++ b/be/src/vec/exec/scan/new_odbc_scanner.cpp
@@ -36,8 +36,12 @@ NewOdbcScanner::NewOdbcScanner(RuntimeState* state, NewOdbcScanNode* parent, int
_tuple_id(odbc_scan_node.tuple_id),
_tuple_desc(nullptr) {}
-Status NewOdbcScanner::prepare(RuntimeState* state) {
+Status NewOdbcScanner::prepare(RuntimeState* state, VExprContext** vconjunct_ctx_ptr) {
VLOG_CRITICAL << NEW_SCANNER_TYPE << "::prepare";
+ if (vconjunct_ctx_ptr != nullptr) {
+ // Copy vconjunct_ctx_ptr from scan node to this scanner's _vconjunct_ctx.
+ RETURN_IF_ERROR((*vconjunct_ctx_ptr)->clone(state, &_vconjunct_ctx));
+ }
if (_is_init) {
return Status::OK();
diff --git a/be/src/vec/exec/scan/new_odbc_scanner.h b/be/src/vec/exec/scan/new_odbc_scanner.h
index 0b8d28ed99..012fc8b3c1 100644
--- a/be/src/vec/exec/scan/new_odbc_scanner.h
+++ b/be/src/vec/exec/scan/new_odbc_scanner.h
@@ -34,7 +34,7 @@ public:
Status close(RuntimeState* state) override;
public:
- Status prepare(RuntimeState* state);
+ Status prepare(RuntimeState* state, VExprContext** vconjunct_ctx_ptr);
protected:
Status _get_block_impl(RuntimeState* state, Block* block, bool* eos) override;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OdbcScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OdbcScanNode.java
index e3876c803b..bfec4e871f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OdbcScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OdbcScanNode.java
@@ -18,6 +18,7 @@
package org.apache.doris.planner;
import org.apache.doris.analysis.Analyzer;
+import org.apache.doris.analysis.DateLiteral;
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.ExprSubstitutionMap;
import org.apache.doris.analysis.FunctionCallExpr;
@@ -53,6 +54,7 @@ public class OdbcScanNode extends ScanNode {
// Now some database have different function call like doris, now doris do not
// push down the function call except MYSQL
+ // TODO: maybe add a config to decide whether to pushdown the function of MYSQL
public static boolean shouldPushDownConjunct(TOdbcTableType tableType, Expr expr) {
if (!tableType.equals(TOdbcTableType.MYSQL)) {
List<FunctionCallExpr> fnExprList = Lists.newArrayList();
@@ -60,6 +62,12 @@ public class OdbcScanNode extends ScanNode {
if (!fnExprList.isEmpty()) {
return false;
}
+ //oracle date type is not push down https://github.com/apache/doris/discussions/15069
+ //TODO: Now not sure how to rewrite the needed expr and doesn't affect other normal
+ //if we could rewrite the expr according to the format, then can pushdown this.
+ if (tableType.equals(TOdbcTableType.ORACLE) && expr.contains(DateLiteral.class)) {
+ return false;
+ }
}
return true;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 03/21: [Bug](datev2) Fix wrong result when use datev2 as partition key (#15094)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit a282d720c920610365393fb8edc10056b0199538
Author: Gabriel <ga...@gmail.com>
AuthorDate: Thu Dec 15 21:27:05 2022 +0800
[Bug](datev2) Fix wrong result when use datev2 as partition key (#15094)
---
be/src/runtime/raw_value.h | 11 +---
.../org/apache/doris/analysis/DateLiteral.java | 52 ++++++++++++---
.../org/apache/doris/catalog/PartitionKey.java | 12 ++--
.../java/org/apache/doris/load/DeleteHandler.java | 6 +-
regression-test/data/delete_p0/test_delete.out | 16 ++---
.../data/partition_p0/test_datev2_partition.out | 39 +++++++++++
.../suites/delete_p0/test_delete.groovy | 22 +++----
.../partition_p0/test_datev2_partition.groovy | 76 ++++++++++++++++++++++
8 files changed, 188 insertions(+), 46 deletions(-)
diff --git a/be/src/runtime/raw_value.h b/be/src/runtime/raw_value.h
index 70d12c60d5..ceb722e847 100644
--- a/be/src/runtime/raw_value.h
+++ b/be/src/runtime/raw_value.h
@@ -519,18 +519,11 @@ inline uint32_t RawValue::zlib_crc32(const void* v, size_t len, const TypeDescri
}
case TYPE_DATEV2: {
- auto* date_v2_val = (const vectorized::DateV2Value<doris::vectorized::DateV2ValueType>*)v;
- char buf[64];
- int date_v2_len = date_v2_val->to_buffer(buf);
- return HashUtil::zlib_crc_hash(buf, date_v2_len, seed);
+ return HashUtil::zlib_crc_hash(v, 4, seed);
}
case TYPE_DATETIMEV2: {
- auto* date_v2_val =
- (const vectorized::DateV2Value<doris::vectorized::DateTimeV2ValueType>*)v;
- char buf[64];
- int date_v2_len = date_v2_val->to_buffer(buf);
- return HashUtil::zlib_crc_hash(buf, date_v2_len, seed);
+ return HashUtil::zlib_crc_hash(v, 8, seed);
}
case TYPE_DECIMALV2: {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java
index b73cc7cd24..2a6efebd40 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteral.java
@@ -486,9 +486,9 @@ public class DateLiteral extends LiteralExpr {
return (year * 10000 + month * 100 + day) * 1000000L + hour * 10000 + minute * 100 + second;
} else if (type.equals(Type.DATEV2)) {
return (year << 9) | (month << 5) | day;
- } else if (type.equals(Type.DATETIMEV2)) {
- return (year << 50) | (month << 46) | (day << 41) | (hour << 36)
- | (minute << 30) | (second << 24) | microsecond;
+ } else if (type.isDatetimeV2()) {
+ return (year << 46) | (month << 42) | (day << 37) | (hour << 32)
+ | (minute << 26) | (second << 20) | (microsecond % (1 << 20));
} else {
Preconditions.checkState(false, "invalid date type: " + type);
return -1L;
@@ -505,8 +505,8 @@ public class DateLiteral extends LiteralExpr {
buffer.order(ByteOrder.LITTLE_ENDIAN);
buffer.putInt(value);
} else if (type == PrimitiveType.DATETIMEV2) {
- long value = (year << 50) | (month << 46) | (day << 41) | (hour << 36)
- | (minute << 30) | (second << 24) | microsecond;
+ long value = (year << 46) | (month << 42) | (day << 37) | (hour << 32)
+ | (minute << 26) | (second << 20) | (microsecond % (1 << 20));
buffer = ByteBuffer.allocate(8);
buffer.order(ByteOrder.LITTLE_ENDIAN);
buffer.putLong(value);
@@ -672,9 +672,40 @@ public class DateLiteral extends LiteralExpr {
return packedDatetime;
}
+ private void fromPackedDatetimeV2(long packedTime) {
+ microsecond = (packedTime % (1L << 20));
+ long ymdhms = (packedTime >> 20);
+ long ymd = ymdhms >> 17;
+ day = ymd % (1 << 5);
+ long ym = ymd >> 5;
+ month = ym % (1 << 4);
+ year = ym >> 4;
+
+ long hms = ymdhms % (1 << 17);
+ second = hms % (1 << 6);
+ minute = (hms >> 6) % (1 << 6);
+ hour = (hms >> 12);
+ // set default date literal type to DATETIME
+ // date literal read from meta will set type by flag bit;
+ this.type = Type.DATETIMEV2;
+ }
+
+ private void fromPackedDateV2(long packedTime) {
+ day = packedTime % (1 << 5);
+ long ym = packedTime >> 5;
+ month = ym % (1 << 4);
+ year = ym >> 4;
+
+ this.type = Type.DATEV2;
+ }
+
private long makePackedDatetimeV2() {
- return (year << 50) | (month << 46) | (day << 41) | (hour << 36)
- | (minute << 30) | (second << 24) | microsecond;
+ return (year << 46) | (month << 42) | (day << 37) | (hour << 32)
+ | (minute << 26) | (second << 20) | (microsecond % (1 << 20));
+ }
+
+ private long makePackedDateV2() {
+ return ((year << 9) | (month << 5) | day);
}
@Override
@@ -693,7 +724,7 @@ public class DateLiteral extends LiteralExpr {
out.writeInt(((ScalarType) this.type).getScalarScale());
} else if (this.type.equals(Type.DATEV2)) {
out.writeShort(DateLiteralType.DATEV2.value());
- out.writeLong(makePackedDatetimeV2());
+ out.writeLong(makePackedDateV2());
} else {
throw new IOException("Error date literal type : " + type);
}
@@ -721,14 +752,17 @@ public class DateLiteral extends LiteralExpr {
public void readFields(DataInput in) throws IOException {
super.readFields(in);
short dateLiteralType = in.readShort();
- fromPackedDatetime(in.readLong());
if (dateLiteralType == DateLiteralType.DATETIME.value()) {
+ fromPackedDatetime(in.readLong());
this.type = Type.DATETIME;
} else if (dateLiteralType == DateLiteralType.DATE.value()) {
+ fromPackedDatetime(in.readLong());
this.type = Type.DATE;
} else if (dateLiteralType == DateLiteralType.DATETIMEV2.value()) {
+ fromPackedDatetimeV2(in.readLong());
this.type = ScalarType.createDatetimeV2Type(in.readInt());
} else if (dateLiteralType == DateLiteralType.DATEV2.value()) {
+ fromPackedDateV2(in.readLong());
this.type = Type.DATEV2;
} else {
throw new IOException("Error date literal type : " + type);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
index f773209cd1..68b49c2974 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
@@ -59,7 +59,7 @@ public class PartitionKey implements Comparable<PartitionKey>, Writable {
throws AnalysisException {
PartitionKey partitionKey = new PartitionKey();
for (Column column : columns) {
- partitionKey.keys.add(LiteralExpr.createInfinity(Type.fromPrimitiveType(column.getDataType()), isMax));
+ partitionKey.keys.add(LiteralExpr.createInfinity(column.getType(), isMax));
partitionKey.types.add(column.getDataType());
}
return partitionKey;
@@ -71,15 +71,13 @@ public class PartitionKey implements Comparable<PartitionKey>, Writable {
Preconditions.checkArgument(keys.size() <= columns.size());
int i;
for (i = 0; i < keys.size(); ++i) {
- partitionKey.keys.add(keys.get(i).getValue(
- Type.fromPrimitiveType(columns.get(i).getDataType())));
+ partitionKey.keys.add(keys.get(i).getValue(columns.get(i).getType()));
partitionKey.types.add(columns.get(i).getDataType());
}
// fill the vacancy with MIN
for (; i < columns.size(); ++i) {
- Type type = Type.fromPrimitiveType(columns.get(i).getDataType());
- partitionKey.keys.add(LiteralExpr.createInfinity(type, false));
+ partitionKey.keys.add(LiteralExpr.createInfinity(columns.get(i).getType(), false));
partitionKey.types.add(columns.get(i).getDataType());
}
@@ -337,7 +335,9 @@ public class PartitionKey implements Comparable<PartitionKey>, Writable {
throw new IOException("type[" + type.name() + "] not supported: ");
}
}
- literal.setType(Type.fromPrimitiveType(type));
+ if (type != PrimitiveType.DATETIMEV2) {
+ literal.setType(Type.fromPrimitiveType(type));
+ }
keys.add(literal);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
index d4b3e3901b..51db514127 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
@@ -714,7 +714,7 @@ public class DeleteHandler implements Writable {
binaryPredicate.setChild(1, LiteralExpr.create(value,
ScalarType.createDatetimeV2Type(ScalarType.MAX_DATETIMEV2_SCALE)));
}
- LiteralExpr.create(value, Type.fromPrimitiveType(column.getDataType()));
+ LiteralExpr.create(value, column.getType());
} catch (AnalysisException e) {
// ErrorReport.reportDdlException(ErrorCode.ERR_INVALID_VALUE, value);
throw new DdlException("Invalid column value[" + value + "] for column " + columnName);
@@ -730,10 +730,10 @@ public class DeleteHandler implements Writable {
|| column.getDataType() == PrimitiveType.DATEV2
|| column.getDataType() == PrimitiveType.DATETIMEV2) {
DateLiteral dateLiteral = new DateLiteral(value,
- Type.fromPrimitiveType(column.getDataType()));
+ column.getType());
value = dateLiteral.getStringValue();
inPredicate.setChild(i, LiteralExpr.create(value,
- Type.fromPrimitiveType(column.getDataType())));
+ column.getType()));
} else {
LiteralExpr.create(value,
Type.fromPrimitiveType(column.getDataType()));
diff --git a/regression-test/data/delete_p0/test_delete.out b/regression-test/data/delete_p0/test_delete.out
index 59317fdefc..37c01de74e 100644
--- a/regression-test/data/delete_p0/test_delete.out
+++ b/regression-test/data/delete_p0/test_delete.out
@@ -14,29 +14,29 @@
-- !sql --
8
--- !sql --
+-- !sql1 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
--- !sql --
+-- !sql2 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
--- !sql --
+-- !sql3 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
abcdef 2022-08-16 2022-08-16T11:11:11 2022-08-16T11:11:11.111 2022-08-16 2022-08-16T11:11:11 2022-08-16T11:11:11.111
--- !sql --
+-- !sql4 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
--- !sql --
+-- !sql5 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
--- !sql --
+-- !sql6 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
--- !sql --
+-- !sql7 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
abcdef 2022-08-16 2022-08-16T11:11:11 2022-08-16T11:11:11.111 2022-08-16 2022-08-16T11:11:11 2022-08-16T11:11:11.111
--- !sql --
+-- !sql8 --
abcdef 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111 2022-08-12 2022-08-16T12:11:11 2022-08-16T12:11:11.111
diff --git a/regression-test/data/partition_p0/test_datev2_partition.out b/regression-test/data/partition_p0/test_datev2_partition.out
new file mode 100644
index 0000000000..224711371f
--- /dev/null
+++ b/regression-test/data/partition_p0/test_datev2_partition.out
@@ -0,0 +1,39 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+2022-12-14
+2022-12-15
+2022-12-16
+2022-12-17
+2022-12-18
+2022-12-19
+2022-12-20
+
+-- !select --
+2022-12-15
+
+-- !select --
+2022-12-16
+2022-12-17
+2022-12-18
+2022-12-19
+2022-12-20
+
+-- !select --
+2022-12-14T22:22:22.222
+2022-12-15T22:22:22.222
+2022-12-16T22:22:22.222
+2022-12-17T22:22:22.222
+2022-12-18T22:22:22.222
+2022-12-19T22:22:22.222
+2022-12-20T22:22:22.222
+
+-- !select --
+2022-12-15T22:22:22.222
+
+-- !select --
+2022-12-16T22:22:22.222
+2022-12-17T22:22:22.222
+2022-12-18T22:22:22.222
+2022-12-19T22:22:22.222
+2022-12-20T22:22:22.222
+
diff --git a/regression-test/suites/delete_p0/test_delete.groovy b/regression-test/suites/delete_p0/test_delete.groovy
index 10f75c4695..90ecb6db02 100644
--- a/regression-test/suites/delete_p0/test_delete.groovy
+++ b/regression-test/suites/delete_p0/test_delete.groovy
@@ -39,36 +39,36 @@ suite("test_delete") {
sql """ INSERT INTO delete_regression_test VALUES ('abcdef','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111'),('abcdef','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111'); """
sql """ delete from ${tableName} where k2 = '2022-08-16' """
- qt_sql """select * from ${tableName};"""
- sql """ delete from delete_regression_test where k1 = 'abcdef' """
+ qt_sql1 """select * from ${tableName} ORDER BY k2;"""
+ sql """ delete from ${tableName} where k1 = 'abcdef' """
sql """ INSERT INTO ${tableName} VALUES ('abcdef','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111'),('abcdef','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111'); """
- sql """ delete from delete_regression_test where k3 = '2022-08-16 11:11:11' """
- qt_sql """select * from ${tableName};"""
- sql """ delete from delete_regression_test where k1 = 'abcdef' """
+ sql """ delete from ${tableName} where k3 = '2022-08-16 11:11:11' """
+ qt_sql2 """select * from ${tableName} ORDER BY k2;"""
+ sql """ delete from ${tableName} where k1 = 'abcdef' """
sql """ INSERT INTO ${tableName} VALUES ('abcdef','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111'),('abcdef','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111'); """
sql """ delete from ${tableName} where k4 = '2022-08-16 11:11:11' """
- qt_sql """select * from ${tableName};"""
+ qt_sql3 """select * from ${tableName} ORDER BY k2;"""
sql """ delete from ${tableName} where k4 = '2022-08-16 11:11:11.111' """
- qt_sql """select * from ${tableName};"""
+ qt_sql4 """select * from ${tableName} ORDER BY k2;"""
sql """ delete from delete_regression_test where k1 = 'abcdef' """
sql """ INSERT INTO ${tableName} VALUES ('abcdef','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111'),('abcdef','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111'); """
sql """ delete from ${tableName} where v1 = '2022-08-16' """
- qt_sql """select * from ${tableName};"""
+ qt_sql5 """select * from ${tableName} ORDER BY k2;"""
sql """ delete from delete_regression_test where k1 = 'abcdef' """
sql """ INSERT INTO ${tableName} VALUES ('abcdef','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111'),('abcdef','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111'); """
sql """ delete from ${tableName} where v2 = '2022-08-16 11:11:11' """
- qt_sql """select * from ${tableName};"""
+ qt_sql6 """select * from ${tableName} ORDER BY k2;"""
sql """ delete from delete_regression_test where k1 = 'abcdef' """
sql """ INSERT INTO ${tableName} VALUES ('abcdef','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111','2022-08-16','2022-08-16 11:11:11.111111','2022-08-16 11:11:11.111111'),('abcdef','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111','2022-08-12','2022-08-16 12:11:11.111111','2022-08-16 12:11:11.111111'); """
sql """ delete from ${tableName} where v3 = '2022-08-16 11:11:11' """
- qt_sql """select * from ${tableName};"""
+ qt_sql7 """select * from ${tableName} ORDER BY k2;"""
sql """ delete from ${tableName} where v3 = '2022-08-16 11:11:11.111' """
- qt_sql """select * from ${tableName};"""
+ qt_sql8 """select * from ${tableName} ORDER BY k2;"""
sql """ delete from delete_regression_test where k1 = 'abcdef' """
sql """ DROP TABLE IF EXISTS ${tableName} """
diff --git a/regression-test/suites/partition_p0/test_datev2_partition.groovy b/regression-test/suites/partition_p0/test_datev2_partition.groovy
new file mode 100644
index 0000000000..600b820684
--- /dev/null
+++ b/regression-test/suites/partition_p0/test_datev2_partition.groovy
@@ -0,0 +1,76 @@
+// 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_datev2_partition") {
+ def tblName1 = "test_datev2_partition1"
+ sql "drop table if exists ${tblName1}"
+ sql """
+ CREATE TABLE `${tblName1}` (
+ `TIME_STAMP` datev2 NOT NULL COMMENT '采集日期'
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`TIME_STAMP`)
+ COMMENT 'OLAP'
+ PARTITION BY RANGE(`TIME_STAMP`)
+ (PARTITION p20221214 VALUES [('2022-12-14'), ('2022-12-15')),
+ PARTITION p20221215 VALUES [('2022-12-15'), ('2022-12-16')),
+ PARTITION p20221216 VALUES [('2022-12-16'), ('2022-12-17')),
+ PARTITION p20221217 VALUES [('2022-12-17'), ('2022-12-18')),
+ PARTITION p20221218 VALUES [('2022-12-18'), ('2022-12-19')),
+ PARTITION p20221219 VALUES [('2022-12-19'), ('2022-12-20')),
+ PARTITION p20221220 VALUES [('2022-12-20'), ('2022-12-21')),
+ PARTITION p20221221 VALUES [('2022-12-21'), ('2022-12-22')))
+ DISTRIBUTED BY HASH(`TIME_STAMP`) BUCKETS 10
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+ sql """ insert into ${tblName1} values ('2022-12-14'), ('2022-12-15'), ('2022-12-16'), ('2022-12-17'), ('2022-12-18'), ('2022-12-19'), ('2022-12-20') """
+
+ qt_select """ select * from ${tblName1} order by TIME_STAMP """
+ qt_select """ select * from ${tblName1} WHERE TIME_STAMP = '2022-12-15' order by TIME_STAMP """
+ qt_select """ select * from ${tblName1} WHERE TIME_STAMP > '2022-12-15' order by TIME_STAMP """
+ sql "drop table ${tblName1}"
+
+ def tblName2 = "test_datev2_partition2"
+ sql "drop table if exists ${tblName2}"
+ sql """
+ CREATE TABLE `${tblName2}` (
+ `TIME_STAMP` datetimev2(3) NOT NULL COMMENT '采集日期'
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`TIME_STAMP`)
+ COMMENT 'OLAP'
+ PARTITION BY RANGE(`TIME_STAMP`)
+ (PARTITION p20221214 VALUES [('2022-12-14 11:11:11.111'), ('2022-12-15 11:11:11.111')),
+ PARTITION p20221215 VALUES [('2022-12-15 11:11:11.111'), ('2022-12-16 11:11:11.111')),
+ PARTITION p20221216 VALUES [('2022-12-16 11:11:11.111'), ('2022-12-17 11:11:11.111')),
+ PARTITION p20221217 VALUES [('2022-12-17 11:11:11.111'), ('2022-12-18 11:11:11.111')),
+ PARTITION p20221218 VALUES [('2022-12-18 11:11:11.111'), ('2022-12-19 11:11:11.111')),
+ PARTITION p20221219 VALUES [('2022-12-19 11:11:11.111'), ('2022-12-20 11:11:11.111')),
+ PARTITION p20221220 VALUES [('2022-12-20 11:11:11.111'), ('2022-12-21 11:11:11.111')),
+ PARTITION p20221221 VALUES [('2022-12-21 11:11:11.111'), ('2022-12-22 11:11:11.111')))
+ DISTRIBUTED BY HASH(`TIME_STAMP`) BUCKETS 10
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+ sql """ insert into ${tblName2} values ('2022-12-14 22:22:22.222'), ('2022-12-15 22:22:22.222'), ('2022-12-16 22:22:22.222'), ('2022-12-17 22:22:22.222'), ('2022-12-18 22:22:22.222'), ('2022-12-19 22:22:22.222'), ('2022-12-20 22:22:22.222') """
+
+ qt_select """ select * from ${tblName2} order by TIME_STAMP """
+ qt_select """ select * from ${tblName2} WHERE TIME_STAMP = '2022-12-15 22:22:22.222' order by TIME_STAMP """
+ qt_select """ select * from ${tblName2} WHERE TIME_STAMP > '2022-12-15 22:22:22.222' order by TIME_STAMP """
+ sql "drop table ${tblName2}"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 21/21: [fix](merge-on-write) delete all rows with same key in all pre segments (#14995)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 95aab248b03ed82b384cf0c2d3510f07b944dad3
Author: xueweizhang <zx...@163.com>
AuthorDate: Mon Dec 19 10:08:38 2022 +0800
[fix](merge-on-write) delete all rows with same key in all pre segments (#14995)
---
be/src/olap/tablet.cpp | 51 ++++++++++++++++++++++++++------------------------
be/src/olap/tablet.h | 3 ++-
2 files changed, 29 insertions(+), 25 deletions(-)
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index 11d125035e..e15bd61610 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -2032,10 +2032,10 @@ Status Tablet::calc_delete_bitmap(RowsetId rowset_id,
RowLocation loc;
// first check if exist in pre segment
if (check_pre_segments) {
- auto st = _check_pk_in_pre_segments(pre_segments, *key, dummy_version,
- delete_bitmap, &loc);
+ auto st = _check_pk_in_pre_segments(rowset_id, pre_segments, *key,
+ dummy_version, delete_bitmap, &loc);
if (st.ok()) {
- delete_bitmap->add({loc.rowset_id, loc.segment_id, dummy_version.first},
+ delete_bitmap->add({rowset_id, loc.segment_id, dummy_version.first},
loc.row_id);
++row_id;
continue;
@@ -2045,23 +2045,27 @@ Status Tablet::calc_delete_bitmap(RowsetId rowset_id,
continue;
}
}
- auto st = lookup_row_key(*key, specified_rowset_ids, &loc, dummy_version.first - 1);
- CHECK(st.ok() || st.is_not_found() || st.is_already_exist());
- if (st.is_not_found()) {
- ++row_id;
- continue;
- }
- // sequence id smaller than the previous one, so delete current row
- if (st.is_already_exist()) {
- loc.rowset_id = rowset_id;
- loc.segment_id = seg->id();
- loc.row_id = row_id;
- }
+ if (!specified_rowset_ids->empty()) {
+ auto st = lookup_row_key(*key, specified_rowset_ids, &loc,
+ dummy_version.first - 1);
+ CHECK(st.ok() || st.is_not_found() || st.is_already_exist());
+ if (st.is_not_found()) {
+ ++row_id;
+ continue;
+ }
+ // sequence id smaller than the previous one, so delete current row
+ if (st.is_already_exist()) {
+ loc.rowset_id = rowset_id;
+ loc.segment_id = seg->id();
+ loc.row_id = row_id;
+ }
+
+ delete_bitmap->add({loc.rowset_id, loc.segment_id, dummy_version.first},
+ loc.row_id);
+ }
++row_id;
- delete_bitmap->add({loc.rowset_id, loc.segment_id, dummy_version.first},
- loc.row_id);
}
remaining -= num_read;
}
@@ -2077,15 +2081,15 @@ Status Tablet::calc_delete_bitmap(RowsetId rowset_id,
}
Status Tablet::_check_pk_in_pre_segments(
- const std::vector<segment_v2::SegmentSharedPtr>& pre_segments, const Slice& key,
- const Version& version, DeleteBitmapPtr delete_bitmap, RowLocation* loc) {
+ RowsetId rowset_id, const std::vector<segment_v2::SegmentSharedPtr>& pre_segments,
+ const Slice& key, const Version& version, DeleteBitmapPtr delete_bitmap, RowLocation* loc) {
for (auto it = pre_segments.rbegin(); it != pre_segments.rend(); ++it) {
auto st = (*it)->lookup_row_key(key, loc);
CHECK(st.ok() || st.is_not_found() || st.is_already_exist());
if (st.is_not_found()) {
continue;
} else if (st.ok() && _schema->has_sequence_col() &&
- delete_bitmap->contains({loc->rowset_id, loc->segment_id, version.first},
+ delete_bitmap->contains({rowset_id, loc->segment_id, version.first},
loc->row_id)) {
// if has sequence col, we continue to compare the sequence_id of
// all segments, util we find an existing key.
@@ -2159,10 +2163,9 @@ Status Tablet::update_delete_bitmap(const RowsetSharedPtr& rowset, DeleteBitmapP
for (const auto& to_del : rowset_ids_to_del) {
delete_bitmap->remove({to_del, 0, 0}, {to_del, UINT32_MAX, INT64_MAX});
}
- if (!rowset_ids_to_add.empty()) {
- RETURN_IF_ERROR(calc_delete_bitmap(rowset->rowset_id(), segments, &rowset_ids_to_add,
- delete_bitmap, cur_version - 1, true));
- }
+
+ RETURN_IF_ERROR(calc_delete_bitmap(rowset->rowset_id(), segments, &rowset_ids_to_add,
+ delete_bitmap, cur_version - 1, true));
// update version without write lock, compaction and publish_txn
// will update delete bitmap, handle compaction with _rowset_update_lock
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 318fdc8748..31e31d2958 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -385,7 +385,8 @@ private:
bool _reconstruct_version_tracker_if_necessary();
void _init_context_common_fields(RowsetWriterContext& context);
- Status _check_pk_in_pre_segments(const std::vector<segment_v2::SegmentSharedPtr>& pre_segments,
+ Status _check_pk_in_pre_segments(RowsetId rowset_id,
+ const std::vector<segment_v2::SegmentSharedPtr>& pre_segments,
const Slice& key, const Version& version,
DeleteBitmapPtr delete_bitmap, RowLocation* loc);
void _rowset_ids_difference(const RowsetIdUnorderedSet& cur, const RowsetIdUnorderedSet& pre,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 08/21: [config](load) enable new load scan node by default (#14808)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 3fe708cf56c3edcafa36f9e489e1d849e0fdf75a
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Fri Dec 16 09:41:43 2022 +0800
[config](load) enable new load scan node by default (#14808)
Set FE `enable_new_load_scan_node` to true by default.
So that all load tasks(broker load, stream load, routine load, insert into) will use FileScanNode instead of BrokerScanNode
to read data
1. Support loading parquet file in stream load with new load scan node.
2. Fix bug that new parquet reader can not read column without logical or converted type.
3. Change jsonb parser function to "jsonb_parse_error_to_null"
So that if the input string is not a valid json string, it will return null for jsonb column in load task.
---
be/src/http/action/stream_load.cpp | 1 +
be/src/util/jsonb_parser.h | 25 +++++---
be/src/vec/exec/format/parquet/schema_desc.cpp | 14 +++--
be/src/vec/exec/scan/vfile_scanner.cpp | 6 +-
.../org/apache/doris/analysis/DataDescription.java | 27 ++++++--
.../main/java/org/apache/doris/common/Config.java | 2 +-
.../org/apache/doris/load/BrokerFileGroup.java | 2 +
.../org/apache/doris/planner/LoadScanNode.java | 5 +-
.../apache/doris/planner/StreamLoadPlanner.java | 14 ++++-
.../planner/external/ExternalFileScanNode.java | 21 ++++++-
.../doris/planner/external/FileGroupInfo.java | 15 +++++
.../doris/planner/external/LoadScanProvider.java | 3 +-
.../apache/doris/service/FrontendServiceImpl.java | 3 +-
.../java/org/apache/doris/task/LoadTaskInfo.java | 5 ++
.../java/org/apache/doris/task/StreamLoadTask.java | 11 ++++
.../doris/planner/StreamLoadPlannerTest.java | 71 ----------------------
.../apache/doris/utframe/TestWithFeService.java | 8 ++-
gensrc/thrift/FrontendService.thrift | 1 +
.../table_valued_function/test_hdfs_tvf.groovy | 2 -
.../suites/export_p0/test_outfile_parquet.groovy | 2 +-
.../external_catalog_p0/hive/test_hive_orc.groovy | 27 --------
.../hive/test_hive_other.groovy | 25 --------
.../hive/test_hive_parquet.groovy | 28 ---------
.../jsonb_p0/test_jsonb_load_and_function.groovy | 26 +++++++-
.../test_jsonb_load_unique_key_and_function.groovy | 2 +
.../load_p0/broker_load/test_array_load.groovy | 8 ---
.../load_p0/broker_load/test_broker_load.groovy | 21 -------
...n_column_exclude_schema_without_jsonpath.groovy | 2 -
.../stream_load/load_json_null_to_nullable.groovy | 6 --
.../stream_load/load_json_with_jsonpath.groovy | 7 ---
.../load_p0/stream_load/test_hdfs_json_load.groovy | 6 --
.../load_p0/stream_load/test_json_load.groovy | 13 ----
.../stream_load/test_txt_special_delimiter.groovy | 7 ---
.../test_streamload_perfomance.groovy | 2 +-
.../multi_catalog_query/hive_catalog_orc.groovy | 25 --------
.../hive_catalog_parquet.groovy | 25 --------
36 files changed, 164 insertions(+), 304 deletions(-)
diff --git a/be/src/http/action/stream_load.cpp b/be/src/http/action/stream_load.cpp
index 69bfeabe48..b2acc0cedb 100644
--- a/be/src/http/action/stream_load.cpp
+++ b/be/src/http/action/stream_load.cpp
@@ -408,6 +408,7 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req, StreamLoadContext*
RETURN_IF_ERROR(file_sink->open());
request.__isset.path = true;
request.fileType = TFileType::FILE_LOCAL;
+ request.__set_file_size(ctx->body_bytes);
ctx->body_sink = file_sink;
}
if (!http_req->header(HTTP_COLUMNS).empty()) {
diff --git a/be/src/util/jsonb_parser.h b/be/src/util/jsonb_parser.h
index c050fd305c..f4711f9a62 100644
--- a/be/src/util/jsonb_parser.h
+++ b/be/src/util/jsonb_parser.h
@@ -65,6 +65,7 @@
#include "jsonb_document.h"
#include "jsonb_error.h"
#include "jsonb_writer.h"
+#include "string_parser.hpp"
namespace doris {
@@ -894,8 +895,12 @@ private:
}
*pbuf = 0; // set null-terminator
- int64_t val = strtol(num_buf_, NULL, 10);
- if (errno == ERANGE) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ int64_t val =
+ StringParser::string_to_int<int64_t>(num_buf_, pbuf - num_buf_, &parse_result);
+ if (parse_result != StringParser::PARSE_SUCCESS) {
+ VLOG_ROW << "debug string_to_int error for " << num_buf_ << " val=" << val
+ << " parse_result=" << parse_result;
err_ = JsonbErrType::E_DECIMAL_OVERFLOW;
return false;
}
@@ -950,7 +955,7 @@ private:
}
*pbuf = 0; // set null-terminator
- return internConvertBufferToDouble();
+ return internConvertBufferToDouble(num_buf_, pbuf - num_buf_);
}
// parse the exponent part of a double number
@@ -990,15 +995,17 @@ private:
}
*pbuf = 0; // set null-terminator
- return internConvertBufferToDouble();
+ return internConvertBufferToDouble(num_buf_, pbuf - num_buf_);
}
// call system function to parse double to string
- bool internConvertBufferToDouble() {
- double val = strtod(num_buf_, NULL);
-
- if (errno == ERANGE) {
- err_ = JsonbErrType::E_DOUBLE_OVERFLOW;
+ bool internConvertBufferToDouble(char* num_buf_, int len) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double val = StringParser::string_to_float<double>(num_buf_, len, &parse_result);
+ if (parse_result != StringParser::PARSE_SUCCESS) {
+ VLOG_ROW << "debug string_to_float error for " << num_buf_ << " val=" << val
+ << " parse_result=" << parse_result;
+ err_ = JsonbErrType::E_DECIMAL_OVERFLOW;
return false;
}
diff --git a/be/src/vec/exec/format/parquet/schema_desc.cpp b/be/src/vec/exec/format/parquet/schema_desc.cpp
index 6d07886e89..2af4d40ea2 100644
--- a/be/src/vec/exec/format/parquet/schema_desc.cpp
+++ b/be/src/vec/exec/format/parquet/schema_desc.cpp
@@ -168,20 +168,24 @@ TypeDescriptor FieldDescriptor::get_doris_type(const tparquet::SchemaElement& ph
switch (physical_schema.type) {
case tparquet::Type::BOOLEAN:
type.type = TYPE_BOOLEAN;
- return type;
+ break;
case tparquet::Type::INT32:
type.type = TYPE_INT;
- return type;
+ break;
case tparquet::Type::INT64:
case tparquet::Type::INT96:
type.type = TYPE_BIGINT;
- return type;
+ break;
case tparquet::Type::FLOAT:
type.type = TYPE_FLOAT;
- return type;
+ break;
case tparquet::Type::DOUBLE:
type.type = TYPE_DOUBLE;
- return type;
+ break;
+ case tparquet::Type::BYTE_ARRAY:
+ case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
+ type.type = TYPE_STRING;
+ break;
default:
break;
}
diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp b/be/src/vec/exec/scan/vfile_scanner.cpp
index b478d5c96e..6d170a6130 100644
--- a/be/src/vec/exec/scan/vfile_scanner.cpp
+++ b/be/src/vec/exec/scan/vfile_scanner.cpp
@@ -209,7 +209,11 @@ Status VFileScanner::_init_src_block(Block* block) {
data_type = DataTypeFactory::instance().create_data_type(it->second, true);
}
if (data_type == nullptr) {
- return Status::NotSupported(fmt::format("Not support arrow type:{}", slot->col_name()));
+ return Status::NotSupported(
+ fmt::format("Not support data type:{} for column: {}",
+ (it == _name_to_col_type.end() ? slot->type().debug_string()
+ : it->second.debug_string()),
+ slot->col_name()));
}
MutableColumnPtr data_column = data_type->create_column();
_src_block.insert(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java
index 79a8cf7840..b67574e743 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java
@@ -222,9 +222,14 @@ public class DataDescription {
public DataDescription(String tableName, LoadTaskInfo taskInfo) {
this.tableName = tableName;
this.partitionNames = taskInfo.getPartitions();
- // Add a dummy path to just make analyze() happy.
- // Stream load does not need this field.
- this.filePaths = Lists.newArrayList("dummy");
+
+ if (!Strings.isNullOrEmpty(taskInfo.getPath())) {
+ this.filePaths = Lists.newArrayList(taskInfo.getPath());
+ } else {
+ // Add a dummy path to just make analyze() happy.
+ this.filePaths = Lists.newArrayList("dummy");
+ }
+
this.fileFieldNames = taskInfo.getColumnExprDescs().getFileColNames();
this.columnSeparator = taskInfo.getColumnSeparator();
this.lineDelimiter = taskInfo.getLineDelimiter();
@@ -259,7 +264,20 @@ public class DataDescription {
// the compress type is saved in "compressType"
this.fileFormat = "csv";
} else {
- this.fileFormat = "json";
+ switch (type) {
+ case FORMAT_ORC:
+ this.fileFormat = "orc";
+ break;
+ case FORMAT_PARQUET:
+ this.fileFormat = "parquet";
+ break;
+ case FORMAT_JSON:
+ this.fileFormat = "json";
+ break;
+ default:
+ this.fileFormat = "unknown";
+ break;
+ }
}
}
// get compress type
@@ -1019,3 +1037,4 @@ public class DataDescription {
return toSql();
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/Config.java b/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
index c424350946..2555d8efaa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
@@ -1755,7 +1755,7 @@ public class Config extends ConfigBase {
* Temp config, should be removed when new file scan node is ready.
*/
@ConfField(mutable = true)
- public static boolean enable_new_load_scan_node = false;
+ public static boolean enable_new_load_scan_node = true;
/**
* Max data version of backends serialize block.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java b/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
index a89f48e52b..497f0c2d23 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
@@ -155,6 +155,7 @@ public class BrokerFileGroup implements Writable {
this.deleteCondition = dataDescription.getDeleteCondition();
this.mergeType = dataDescription.getMergeType();
this.sequenceCol = dataDescription.getSequenceCol();
+ this.filePaths = dataDescription.getFilePaths();
}
// NOTE: DBLock will be held
@@ -598,3 +599,4 @@ public class BrokerFileGroup implements Writable {
return fileGroup;
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/LoadScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/LoadScanNode.java
index 0f8eb77ffb..bf4c66dccb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/LoadScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/LoadScanNode.java
@@ -207,6 +207,8 @@ public abstract class LoadScanNode extends ScanNode {
expr.analyze(analyzer);
}
+ // for jsonb type, use jsonb_parse_xxx to parse src string to jsonb.
+ // and if input string is not a valid json string, return null.
PrimitiveType dstType = destSlotDesc.getType().getPrimitiveType();
PrimitiveType srcType = expr.getType().getPrimitiveType();
if (dstType == PrimitiveType.JSONB
@@ -217,7 +219,7 @@ public abstract class LoadScanNode extends ScanNode {
if (destSlotDesc.getIsNullable() || expr.isNullable()) {
nullable = "nullable";
}
- String name = "jsonb_parse_" + nullable + "_error_to_invalid";
+ String name = "jsonb_parse_" + nullable + "_error_to_null";
expr = new FunctionCallExpr(name, args);
expr.analyze(analyzer);
} else {
@@ -251,3 +253,4 @@ public abstract class LoadScanNode extends ScanNode {
planNode.setBrokerScanNode(brokerScanNode);
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
index b55bb8f402..81e9009f65 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
@@ -52,6 +52,7 @@ import org.apache.doris.task.LoadTaskInfo;
import org.apache.doris.thrift.PaloInternalServiceVersion;
import org.apache.doris.thrift.TBrokerFileStatus;
import org.apache.doris.thrift.TExecPlanFragmentParams;
+import org.apache.doris.thrift.TFileType;
import org.apache.doris.thrift.TLoadErrorHubInfo;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TPlanFragmentExecParams;
@@ -180,9 +181,15 @@ public class StreamLoadPlanner {
fileGroup.parse(db, dataDescription);
// 2. create dummy file status
TBrokerFileStatus fileStatus = new TBrokerFileStatus();
- fileStatus.setPath("");
- fileStatus.setIsDir(false);
- fileStatus.setSize(-1); // must set to -1, means stream.
+ if (taskInfo.getFileType() == TFileType.FILE_LOCAL) {
+ fileStatus.setPath(taskInfo.getPath());
+ fileStatus.setIsDir(false);
+ fileStatus.setSize(taskInfo.getFileSize()); // must set to -1, means stream.
+ } else {
+ fileStatus.setPath("");
+ fileStatus.setIsDir(false);
+ fileStatus.setSize(-1); // must set to -1, means stream.
+ }
fileScanNode.setLoadInfo(loadId, taskInfo.getTxnId(), destTable, BrokerDesc.createForStreamLoad(),
fileGroup, fileStatus, taskInfo.isStrictMode(), taskInfo.getFileType());
scanNode = fileScanNode;
@@ -323,3 +330,4 @@ public class StreamLoadPlanner {
return null;
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
index ab0b25b08d..c42c97041b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
@@ -456,7 +456,25 @@ public class ExternalFileScanNode extends ExternalScanNode {
expr = new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY, expr, new IntLiteral(-1));
expr.analyze(analyzer);
}
- expr = castToSlot(destSlotDesc, expr);
+
+ // for jsonb type, use jsonb_parse_xxx to parse src string to jsonb.
+ // and if input string is not a valid json string, return null.
+ PrimitiveType dstType = destSlotDesc.getType().getPrimitiveType();
+ PrimitiveType srcType = expr.getType().getPrimitiveType();
+ if (dstType == PrimitiveType.JSONB
+ && (srcType == PrimitiveType.VARCHAR || srcType == PrimitiveType.STRING)) {
+ List<Expr> args = Lists.newArrayList();
+ args.add(expr);
+ String nullable = "notnull";
+ if (destSlotDesc.getIsNullable() || expr.isNullable()) {
+ nullable = "nullable";
+ }
+ String name = "jsonb_parse_" + nullable + "_error_to_null";
+ expr = new FunctionCallExpr(name, args);
+ expr.analyze(analyzer);
+ } else {
+ expr = castToSlot(destSlotDesc, expr);
+ }
params.putToExprOfDestSlot(destSlotDesc.getId().asInt(), expr.treeToThrift());
}
params.setDestSidToSrcSidWithoutTrans(destSidToSrcSidWithoutTrans);
@@ -550,3 +568,4 @@ public class ExternalFileScanNode extends ExternalScanNode {
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileGroupInfo.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileGroupInfo.java
index eb99c88ed9..f3cb9a19f9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileGroupInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileGroupInfo.java
@@ -44,6 +44,7 @@ import org.apache.doris.thrift.TScanRangeLocation;
import org.apache.doris.thrift.TScanRangeLocations;
import org.apache.doris.thrift.TUniqueId;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -140,6 +141,10 @@ public class FileGroupInfo {
return loadParallelism;
}
+ public TFileType getFileType() {
+ return fileType;
+ }
+
public String getExplainString(String prefix) {
StringBuilder sb = new StringBuilder();
sb.append("file scan\n");
@@ -303,6 +308,15 @@ public class FileGroupInfo {
rangeDesc.setSize(rangeBytes);
rangeDesc.setColumnsFromPath(columnsFromPath);
} else {
+ // for stream load
+ if (getFileType() == TFileType.FILE_LOCAL) {
+ // when loading parquet via stream, there will be a local file saved on BE
+ // so to read it as a local file.
+ Preconditions.checkState(fileGroup.getFilePaths().size() == 1);
+ rangeDesc.setPath(fileGroup.getFilePaths().get(0));
+ rangeDesc.setStartOffset(0);
+ rangeDesc.setSize(fileStatus.size);
+ }
rangeDesc.setLoadId(loadId);
rangeDesc.setSize(fileStatus.size);
}
@@ -310,3 +324,4 @@ public class FileGroupInfo {
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/LoadScanProvider.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/LoadScanProvider.java
index 7502fb0749..8f7ea2bb6e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/LoadScanProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/LoadScanProvider.java
@@ -111,7 +111,7 @@ public class LoadScanProvider implements FileScanProviderIf {
TFileAttributes fileAttributes = new TFileAttributes();
setFileAttributes(ctx.fileGroup, fileAttributes);
params.setFileAttributes(fileAttributes);
- params.setFileType(fileGroupInfo.getBrokerDesc().getFileType());
+ params.setFileType(fileGroupInfo.getFileType());
ctx.params = params;
initColumns(ctx, analyzer);
@@ -252,3 +252,4 @@ public class LoadScanProvider implements FileScanProviderIf {
return fileGroupInfo.getTargetTable();
}
}
+
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 c7ab68ab0c..e4d0936fd1 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
@@ -899,7 +899,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
} catch (Throwable e) {
LOG.warn("catch unknown result.", e);
status.setStatusCode(TStatusCode.INTERNAL_ERROR);
- status.addToErrorMsgs(Strings.nullToEmpty(e.getMessage()));
+ status.addToErrorMsgs(e.getClass().getSimpleName() + ": " + Strings.nullToEmpty(e.getMessage()));
return result;
}
return result;
@@ -1230,3 +1230,4 @@ public class FrontendServiceImpl implements FrontendService.Iface {
return result;
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
index dd97602003..642597246f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
@@ -71,6 +71,10 @@ public interface LoadTaskInfo {
String getPath();
+ default long getFileSize() {
+ return 0;
+ }
+
double getMaxFilterRatio();
ImportColumnDescs getColumnExprDescs();
@@ -118,3 +122,4 @@ public interface LoadTaskInfo {
}
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
index 300a93ca19..59393b7d01 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
@@ -68,6 +68,7 @@ public class StreamLoadTask implements LoadTaskInfo {
private Separator lineDelimiter;
private PartitionNames partitions;
private String path;
+ private long fileSize = 0;
private boolean negative;
private boolean strictMode = false; // default is false
private String timezone = TimeUtils.DEFAULT_TIME_ZONE;
@@ -159,6 +160,11 @@ public class StreamLoadTask implements LoadTaskInfo {
return path;
}
+ @Override
+ public long getFileSize() {
+ return fileSize;
+ }
+
public boolean getNegative() {
return negative;
}
@@ -234,6 +240,7 @@ public class StreamLoadTask implements LoadTaskInfo {
return !Strings.isNullOrEmpty(sequenceCol);
}
+
@Override
public String getSequenceCol() {
return sequenceCol;
@@ -249,6 +256,9 @@ public class StreamLoadTask implements LoadTaskInfo {
request.getFileType(), request.getFormatType(),
request.getCompressType());
streamLoadTask.setOptionalFromTSLPutRequest(request);
+ if (request.isSetFileSize()) {
+ streamLoadTask.fileSize = request.getFileSize();
+ }
return streamLoadTask;
}
@@ -416,3 +426,4 @@ public class StreamLoadTask implements LoadTaskInfo {
return maxFilterRatio;
}
}
+
diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/StreamLoadPlannerTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/StreamLoadPlannerTest.java
index d8ddcfe909..870dbbeb55 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/planner/StreamLoadPlannerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/planner/StreamLoadPlannerTest.java
@@ -17,90 +17,19 @@
package org.apache.doris.planner;
-import org.apache.doris.analysis.Analyzer;
import org.apache.doris.analysis.CompoundPredicate;
import org.apache.doris.analysis.ImportColumnsStmt;
import org.apache.doris.analysis.ImportWhereStmt;
import org.apache.doris.analysis.SqlParser;
import org.apache.doris.analysis.SqlScanner;
-import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.Database;
-import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.Partition;
-import org.apache.doris.catalog.PrimitiveType;
-import org.apache.doris.common.UserException;
import org.apache.doris.common.util.SqlParserUtils;
-import org.apache.doris.task.StreamLoadTask;
-import org.apache.doris.thrift.TFileFormatType;
-import org.apache.doris.thrift.TFileType;
-import org.apache.doris.thrift.TStreamLoadPutRequest;
-import org.apache.doris.thrift.TUniqueId;
-import com.google.common.collect.Lists;
-import mockit.Expectations;
-import mockit.Injectable;
-import mockit.Mocked;
import org.junit.Assert;
import org.junit.Test;
import java.io.StringReader;
-import java.util.Arrays;
-import java.util.List;
public class StreamLoadPlannerTest {
- @Injectable
- Database db;
-
- @Injectable
- OlapTable destTable;
-
- @Mocked
- StreamLoadScanNode scanNode;
-
- @Mocked
- OlapTableSink sink;
-
- @Mocked
- Partition partition;
-
- @Test
- public void testNormalPlan() throws UserException {
- List<Column> columns = Lists.newArrayList();
- Column c1 = new Column("c1", PrimitiveType.BIGINT, false);
- columns.add(c1);
- Column c2 = new Column("c2", PrimitiveType.BIGINT, true);
- columns.add(c2);
- new Expectations() {
- {
- destTable.getBaseSchema();
- minTimes = 0;
- result = columns;
- destTable.getPartitions();
- minTimes = 0;
- result = Arrays.asList(partition);
- scanNode.init((Analyzer) any);
- minTimes = 0;
- scanNode.getChildren();
- minTimes = 0;
- result = Lists.newArrayList();
- scanNode.getId();
- minTimes = 0;
- result = new PlanNodeId(5);
- partition.getId();
- minTimes = 0;
- result = 0;
- }
- };
- TStreamLoadPutRequest request = new TStreamLoadPutRequest();
- request.setTxnId(1);
- request.setLoadId(new TUniqueId(2, 3));
- request.setFileType(TFileType.FILE_STREAM);
- request.setFormatType(TFileFormatType.FORMAT_CSV_PLAIN);
- StreamLoadTask streamLoadTask = StreamLoadTask.fromTStreamLoadPutRequest(request);
- StreamLoadPlanner planner = new StreamLoadPlanner(db, destTable, streamLoadTask);
- planner.plan(streamLoadTask.getId());
- }
-
@Test
public void testParseStmt() throws Exception {
String sql = new String("COLUMNS (k1, k2, k3=abc(), k4=default_value())");
diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java
index 90b0e4658d..979e143bd9 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java
@@ -96,6 +96,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Comparator;
+import java.util.ConcurrentModificationException;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -492,7 +493,12 @@ public abstract class TestWithFeService {
}
public void createTable(String sql) throws Exception {
- createTables(sql);
+ try {
+ createTables(sql);
+ } catch (ConcurrentModificationException e) {
+ e.printStackTrace();
+ throw e;
+ }
}
public void dropTable(String table, boolean force) throws Exception {
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 9cf6ffb6ee..47de9fcdc3 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -544,6 +544,7 @@ struct TStreamLoadPutRequest {
38: optional string header_type
39: optional string hidden_columns
40: optional PlanNodes.TFileCompressType compress_type
+ 41: optional i64 file_size // only for stream load with parquet or orc
}
struct TStreamLoadPutResult {
diff --git a/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy b/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy
index 65996cca81..07b8363b0b 100644
--- a/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy
+++ b/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy
@@ -26,7 +26,6 @@ suite("test_hdfs_tvf") {
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
try {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "true");"""
// test csv foramt
uri = "${defaultFS}" + "/user/doris/preinstalled_data/csv_format_test/all_types.csv"
@@ -193,7 +192,6 @@ suite("test_hdfs_tvf") {
assertTrue(result2[0][0] == 5, "Insert should update 12 rows")
qt_insert """ select * from test_hdfs_tvf order by id; """
} finally {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
}
}
}
diff --git a/regression-test/suites/export_p0/test_outfile_parquet.groovy b/regression-test/suites/export_p0/test_outfile_parquet.groovy
index 8b1944d2fb..9be0c8fdc4 100644
--- a/regression-test/suites/export_p0/test_outfile_parquet.groovy
+++ b/regression-test/suites/export_p0/test_outfile_parquet.groovy
@@ -22,7 +22,7 @@ import java.nio.file.Files
import java.nio.file.Paths
suite("test_outfile_parquet") {
- def dbName = "test_query_db"
+ def dbName = "test_outfile_parquet"
sql "CREATE DATABASE IF NOT EXISTS ${dbName}"
sql "USE $dbName"
StringBuilder strBuilder = new StringBuilder()
diff --git a/regression-test/suites/external_catalog_p0/hive/test_hive_orc.groovy b/regression-test/suites/external_catalog_p0/hive/test_hive_orc.groovy
index 7e1f8b78fd..2f8d610200 100644
--- a/regression-test/suites/external_catalog_p0/hive/test_hive_orc.groovy
+++ b/regression-test/suites/external_catalog_p0/hive/test_hive_orc.groovy
@@ -66,37 +66,12 @@ suite("test_hive_orc", "all_types") {
qt_only_partition_col """select count(p1_col), count(p2_col) from orc_all_types;"""
}
- def set_be_config = { flag ->
- String[][] backends = sql """ show backends; """
- assertTrue(backends.size() > 0)
- for (String[] backend in backends) {
- StringBuilder setConfigCommand = new StringBuilder();
- setConfigCommand.append("curl -X POST http://")
- setConfigCommand.append(backend[2])
- setConfigCommand.append(":")
- setConfigCommand.append(backend[5])
- setConfigCommand.append("/api/update_config?")
- String command1 = setConfigCommand.toString() + "enable_new_load_scan_node=$flag"
- logger.info(command1)
- String command2 = setConfigCommand.toString() + "enable_new_file_scanner=$flag"
- logger.info(command2)
- def process1 = command1.execute()
- int code = process1.waitFor()
- assertEquals(code, 0)
- def process2 = command2.execute()
- code = process1.waitFor()
- assertEquals(code, 0)
- }
- }
-
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
try {
String hms_port = context.config.otherConfigs.get("hms_port")
String catalog_name = "hive_test_orc"
sql """admin set frontend config ("enable_multi_catalog" = "true")"""
- sql """admin set frontend config ("enable_new_load_scan_node" = "true");"""
- set_be_config.call('true')
sql """drop catalog if exists ${catalog_name}"""
sql """
create catalog if not exists ${catalog_name} properties (
@@ -114,8 +89,6 @@ suite("test_hive_orc", "all_types") {
only_partition_col()
} finally {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
- set_be_config.call('false')
}
}
}
diff --git a/regression-test/suites/external_catalog_p0/hive/test_hive_other.groovy b/regression-test/suites/external_catalog_p0/hive/test_hive_other.groovy
index f8401d70e9..71c184b872 100644
--- a/regression-test/suites/external_catalog_p0/hive/test_hive_other.groovy
+++ b/regression-test/suites/external_catalog_p0/hive/test_hive_other.groovy
@@ -50,36 +50,11 @@ suite("test_hive_other", "p0") {
}
- def set_be_config = { ->
- String[][] backends = sql """ show backends; """
- assertTrue(backends.size() > 0)
- for (String[] backend in backends) {
- // No need to set this config anymore, but leave this code sample here
- // StringBuilder setConfigCommand = new StringBuilder();
- // setConfigCommand.append("curl -X POST http://")
- // setConfigCommand.append(backend[2])
- // setConfigCommand.append(":")
- // setConfigCommand.append(backend[5])
- // setConfigCommand.append("/api/update_config?")
- // String command1 = setConfigCommand.toString() + "enable_new_load_scan_node=true"
- // logger.info(command1)
- // String command2 = setConfigCommand.toString() + "enable_new_file_scanner=true"
- // logger.info(command2)
- // def process1 = command1.execute()
- // int code = process1.waitFor()
- // assertEquals(code, 0)
- // def process2 = command2.execute()
- // code = process1.waitFor()
- // assertEquals(code, 0)
- }
- }
-
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
String hms_port = context.config.otherConfigs.get("hms_port")
String hdfs_port = context.config.otherConfigs.get("hdfs_port")
String catalog_name = "hive_test_other"
- set_be_config.call()
sql """admin set frontend config ("enable_multi_catalog" = "true")"""
sql """drop catalog if exists ${catalog_name}"""
diff --git a/regression-test/suites/external_catalog_p0/hive/test_hive_parquet.groovy b/regression-test/suites/external_catalog_p0/hive/test_hive_parquet.groovy
index 8cb89baec5..d644699f00 100644
--- a/regression-test/suites/external_catalog_p0/hive/test_hive_parquet.groovy
+++ b/regression-test/suites/external_catalog_p0/hive/test_hive_parquet.groovy
@@ -139,38 +139,12 @@ suite("test_hive_parquet", "p0") {
"""
}
-
- def set_be_config = { flag ->
- String[][] backends = sql """ show backends; """
- assertTrue(backends.size() > 0)
- for (String[] backend in backends) {
- StringBuilder setConfigCommand = new StringBuilder();
- setConfigCommand.append("curl -X POST http://")
- setConfigCommand.append(backend[2])
- setConfigCommand.append(":")
- setConfigCommand.append(backend[5])
- setConfigCommand.append("/api/update_config?")
- String command1 = setConfigCommand.toString() + "enable_new_load_scan_node=$flag"
- logger.info(command1)
- String command2 = setConfigCommand.toString() + "enable_new_file_scanner=$flag"
- logger.info(command2)
- def process1 = command1.execute()
- int code = process1.waitFor()
- assertEquals(code, 0)
- def process2 = command2.execute()
- code = process1.waitFor()
- assertEquals(code, 0)
- }
- }
-
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
try {
String hms_port = context.config.otherConfigs.get("hms_port")
String catalog_name = "hive_test_parquet"
sql """admin set frontend config ("enable_multi_catalog" = "true")"""
- sql """admin set frontend config ("enable_new_load_scan_node" = "true");"""
- set_be_config.call('true')
sql """drop catalog if exists ${catalog_name}"""
sql """
create catalog if not exists ${catalog_name} properties (
@@ -201,8 +175,6 @@ suite("test_hive_parquet", "p0") {
q19()
q20()
} finally {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
- set_be_config.call('false')
}
}
}
diff --git a/regression-test/suites/jsonb_p0/test_jsonb_load_and_function.groovy b/regression-test/suites/jsonb_p0/test_jsonb_load_and_function.groovy
index b410251cc4..9c4abca487 100644
--- a/regression-test/suites/jsonb_p0/test_jsonb_load_and_function.groovy
+++ b/regression-test/suites/jsonb_p0/test_jsonb_load_and_function.groovy
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
suite("test_jsonb_load_and_function", "p0") {
// define a sql table
def testTable = "tbl_test_jsonb"
@@ -35,12 +37,12 @@ suite("test_jsonb_load_and_function", "p0") {
"""
// load the jsonb data from csv file
- // fail by default for invalid data rows
streamLoad {
table testTable
file dataFile // import csv file
time 10000 // limit inflight 10s
+ set 'strict_mode', 'true'
// if declared a check callback, the default check condition will ignore.
// So you must check all condition
@@ -50,12 +52,23 @@ suite("test_jsonb_load_and_function", "p0") {
}
log.info("Stream load result: ${result}".toString())
def json = parseJson(result)
+
+ StringBuilder sb = new StringBuilder()
+ sb.append("curl -X GET " + json.ErrorURL)
+ String command = sb.toString()
+ def process = command.execute()
+ def code = process.waitFor()
+ def err = IOGroovyMethods.getText(new BufferedReader(new InputStreamReader(process.getErrorStream())))
+ def out = process.getText()
+ log.info("error result: " + out)
+
assertEquals("fail", json.Status.toLowerCase())
assertEquals("too many filtered rows", json.Message)
assertEquals(25, json.NumberTotalRows)
assertEquals(18, json.NumberLoadedRows)
assertEquals(7, json.NumberFilteredRows)
assertTrue(json.LoadBytes > 0)
+ log.info("url: " + json.ErrorURL)
}
}
@@ -68,6 +81,7 @@ suite("test_jsonb_load_and_function", "p0") {
set 'max_filter_ratio', '0.3'
file dataFile // import csv file
time 10000 // limit inflight 10s
+ set 'strict_mode', 'true'
// if declared a check callback, the default check condition will ignore.
// So you must check all condition
@@ -77,6 +91,16 @@ suite("test_jsonb_load_and_function", "p0") {
}
log.info("Stream load result: ${result}".toString())
def json = parseJson(result)
+
+ StringBuilder sb = new StringBuilder()
+ sb.append("curl -X GET " + json.ErrorURL)
+ String command = sb.toString()
+ def process = command.execute()
+ def code = process.waitFor()
+ def err = IOGroovyMethods.getText(new BufferedReader(new InputStreamReader(process.getErrorStream())))
+ def out = process.getText()
+ log.info("error result: " + out)
+
assertEquals("success", json.Status.toLowerCase())
assertEquals(25, json.NumberTotalRows)
assertEquals(18, json.NumberLoadedRows)
diff --git a/regression-test/suites/jsonb_p0/test_jsonb_load_unique_key_and_function.groovy b/regression-test/suites/jsonb_p0/test_jsonb_load_unique_key_and_function.groovy
index 4c3bd82e83..0789d48c4e 100644
--- a/regression-test/suites/jsonb_p0/test_jsonb_load_unique_key_and_function.groovy
+++ b/regression-test/suites/jsonb_p0/test_jsonb_load_unique_key_and_function.groovy
@@ -41,6 +41,7 @@ suite("test_jsonb_unique_load_and_function", "p0") {
file dataFile // import csv file
time 10000 // limit inflight 10s
+ set 'strict_mode', 'true'
// if declared a check callback, the default check condition will ignore.
// So you must check all condition
@@ -68,6 +69,7 @@ suite("test_jsonb_unique_load_and_function", "p0") {
set 'max_filter_ratio', '0.3'
file dataFile // import csv file
time 10000 // limit inflight 10s
+ set 'strict_mode', 'true'
// if declared a check callback, the default check condition will ignore.
// So you must check all condition
diff --git a/regression-test/suites/load_p0/broker_load/test_array_load.groovy b/regression-test/suites/load_p0/broker_load/test_array_load.groovy
index 7f5d109952..271050f3ca 100644
--- a/regression-test/suites/load_p0/broker_load/test_array_load.groovy
+++ b/regression-test/suites/load_p0/broker_load/test_array_load.groovy
@@ -202,13 +202,6 @@ suite("test_array_load", "load_p0") {
try {
for ( i in 0..1 ) {
- // should be deleted after new_load_scan is ready
- if (i == 1) {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
- } else {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "true");"""
- }
-
// case1: import array data in json format and enable vectorized engine
try {
sql "DROP TABLE IF EXISTS ${testTable}"
@@ -280,7 +273,6 @@ suite("test_array_load", "load_p0") {
}
}
} finally {
- try_sql("""ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");""")
}
diff --git a/regression-test/suites/load_p0/broker_load/test_broker_load.groovy b/regression-test/suites/load_p0/broker_load/test_broker_load.groovy
index e1f16676dd..e67397e821 100644
--- a/regression-test/suites/load_p0/broker_load/test_broker_load.groovy
+++ b/regression-test/suites/load_p0/broker_load/test_broker_load.groovy
@@ -192,28 +192,8 @@ suite("test_broker_load", "p0") {
logger.info("Submit load with lable: $uuid, table: $table, path: $path")
}
- def set_be_config = { flag->
- String[][] backends = sql """ show backends; """
- assertTrue(backends.size() > 0)
- for (String[] backend in backends) {
- // No need to set this config anymore, but leave this code sample here
- // StringBuilder setConfigCommand = new StringBuilder();
- // setConfigCommand.append("curl -X POST http://")
- // setConfigCommand.append(backend[2])
- // setConfigCommand.append(":")
- // setConfigCommand.append(backend[5])
- // setConfigCommand.append("/api/update_config?")
- // String command1 = setConfigCommand.toString() + "enable_new_load_scan_node=$flag"
- // logger.info(command1)
- // def process1 = command1.execute()
- // int code = process1.waitFor()
- // assertEquals(code, 0)
- }
- }
-
if (enabled != null && enabled.equalsIgnoreCase("true")) {
def uuids = []
- set_be_config.call('true')
try {
def i = 0
for (String table in tables) {
@@ -258,7 +238,6 @@ suite("test_broker_load", "p0") {
order_qt_parquet_s3_case9 """ select * from parquet_s3_case9"""
} finally {
- set_be_config.call('false')
for (String table in tables) {
sql new File("""${context.file.parent}/ddl/${table}_drop.sql""").text
}
diff --git a/regression-test/suites/load_p0/stream_load/load_json_column_exclude_schema_without_jsonpath.groovy b/regression-test/suites/load_p0/stream_load/load_json_column_exclude_schema_without_jsonpath.groovy
index 760af3344e..d00316c37a 100644
--- a/regression-test/suites/load_p0/stream_load/load_json_column_exclude_schema_without_jsonpath.groovy
+++ b/regression-test/suites/load_p0/stream_load/load_json_column_exclude_schema_without_jsonpath.groovy
@@ -51,8 +51,6 @@ suite("test_load_json_column_exclude_schema_without_jsonpath", "p0") {
def load_array_data = {new_json_reader_flag, table_name, strip_flag, read_flag, format_flag, exprs, json_paths,
json_root, where_expr, fuzzy_flag, column_sep, file_name ->
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "${new_json_reader_flag}");"""
// load the json data
streamLoad {
diff --git a/regression-test/suites/load_p0/stream_load/load_json_null_to_nullable.groovy b/regression-test/suites/load_p0/stream_load/load_json_null_to_nullable.groovy
index f934c038a2..be0baf42b8 100644
--- a/regression-test/suites/load_p0/stream_load/load_json_null_to_nullable.groovy
+++ b/regression-test/suites/load_p0/stream_load/load_json_null_to_nullable.groovy
@@ -42,9 +42,6 @@ suite("test_load_json_null_to_nullable", "p0") {
def load_array_data = {new_json_reader_flag, table_name, strip_flag, read_flag, format_flag, exprs, json_paths,
json_root, where_expr, fuzzy_flag, column_sep, file_name ->
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "${new_json_reader_flag}");"""
-
// load the json data
streamLoad {
table table_name
@@ -77,9 +74,6 @@ suite("test_load_json_null_to_nullable", "p0") {
assertTrue(json.NumberLoadedRows > 0 && json.LoadBytes > 0)
}
}
-
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
}
def check_data_correct = {table_name ->
diff --git a/regression-test/suites/load_p0/stream_load/load_json_with_jsonpath.groovy b/regression-test/suites/load_p0/stream_load/load_json_with_jsonpath.groovy
index 02ffd808e2..f48b41be79 100644
--- a/regression-test/suites/load_p0/stream_load/load_json_with_jsonpath.groovy
+++ b/regression-test/suites/load_p0/stream_load/load_json_with_jsonpath.groovy
@@ -42,10 +42,6 @@ suite("test_load_json_with_jsonpath", "p0") {
def load_array_data = {new_json_reader_flag, table_name, strip_flag, read_flag, format_flag, exprs, json_paths,
json_root, where_expr, fuzzy_flag, column_sep, file_name ->
-
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "${new_json_reader_flag}");"""
-
// load the json data
streamLoad {
table table_name
@@ -78,9 +74,6 @@ suite("test_load_json_with_jsonpath", "p0") {
assertTrue(json.NumberLoadedRows > 0 && json.LoadBytes > 0)
}
}
-
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
}
def check_data_correct = {table_name ->
diff --git a/regression-test/suites/load_p0/stream_load/test_hdfs_json_load.groovy b/regression-test/suites/load_p0/stream_load/test_hdfs_json_load.groovy
index 5c79cf1b97..6565357e96 100644
--- a/regression-test/suites/load_p0/stream_load/test_hdfs_json_load.groovy
+++ b/regression-test/suites/load_p0/stream_load/test_hdfs_json_load.groovy
@@ -44,9 +44,6 @@ suite("test_hdfs_json_load", "p0") {
def load_from_hdfs1 = {new_json_reader_flag, strip_flag, fuzzy_flag, testTablex, label, fileName,
fsPath, hdfsUser, exprs, jsonpaths, json_root, columns_parameter, where ->
- // should be delete after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "${new_json_reader_flag}");"""
-
def hdfsFilePath = "${fsPath}/user/doris/preinstalled_data/json_format_test/${fileName}"
def result1= sql """
LOAD LABEL ${label} (
@@ -76,9 +73,6 @@ suite("test_hdfs_json_load", "p0") {
assertTrue(result1.size() == 1)
assertTrue(result1[0].size() == 1)
assertTrue(result1[0][0] == 0, "Query OK, 0 rows affected")
-
- // should be delete after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
}
def check_load_result = {checklabel, testTablex ->
diff --git a/regression-test/suites/load_p0/stream_load/test_json_load.groovy b/regression-test/suites/load_p0/stream_load/test_json_load.groovy
index e066467e3e..513d3e14dd 100644
--- a/regression-test/suites/load_p0/stream_load/test_json_load.groovy
+++ b/regression-test/suites/load_p0/stream_load/test_json_load.groovy
@@ -116,8 +116,6 @@ suite("test_json_load", "p0") {
def load_json_data = {new_json_reader_flag, label, strip_flag, read_flag, format_flag, exprs, json_paths,
json_root, where_expr, fuzzy_flag, file_name, ignore_failure=false ->
- // should be delete after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "${new_json_reader_flag}");"""
// load the json data
streamLoad {
@@ -150,9 +148,6 @@ suite("test_json_load", "p0") {
assertTrue(json.NumberLoadedRows > 0 && json.LoadBytes > 0)
}
}
-
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
}
def load_from_hdfs1 = {testTablex, label, hdfsFilePath, format, brokerName, hdfsUser, hdfsPasswd ->
@@ -529,8 +524,6 @@ suite("test_json_load", "p0") {
try {
sql "DROP TABLE IF EXISTS ${testTable}"
create_test_table3.call(testTable)
- // should be delete after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
// load the json data
streamLoad {
table "${testTable}"
@@ -557,16 +550,12 @@ suite("test_json_load", "p0") {
assertTrue(json.NumberLoadedRows > 0 && json.LoadBytes > 0)
}
}
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
sql "sync"
qt_select13 "select * from ${testTable} order by id"
sql "DROP TABLE IF EXISTS ${testTable}"
create_test_table3.call(testTable)
- // should be delete after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "true");"""
// load the json data
streamLoad {
table "${testTable}"
@@ -593,8 +582,6 @@ suite("test_json_load", "p0") {
assertTrue(json.NumberLoadedRows > 0 && json.LoadBytes > 0)
}
}
- // should be deleted after new_load_scan is ready
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
sql "sync"
qt_select13 "select * from ${testTable} order by id"
diff --git a/regression-test/suites/load_p0/stream_load/test_txt_special_delimiter.groovy b/regression-test/suites/load_p0/stream_load/test_txt_special_delimiter.groovy
index fff343078b..0e762f31d3 100644
--- a/regression-test/suites/load_p0/stream_load/test_txt_special_delimiter.groovy
+++ b/regression-test/suites/load_p0/stream_load/test_txt_special_delimiter.groovy
@@ -31,13 +31,6 @@ suite("test_txt_special_delimiter", "p0") {
PROPERTIES ("replication_allocation" = "tag.location.default: 1");
"""
for ( i in 0..1 ) {
- // should be deleted after new_load_scan is ready
- if (i == 1) {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "false");"""
- } else {
- sql """ADMIN SET FRONTEND CONFIG ("enable_new_load_scan_node" = "true");"""
- }
-
// test special_delimiter success
streamLoad {
table "${tableName}"
diff --git a/regression-test/suites/performance_p0/test_streamload_perfomance.groovy b/regression-test/suites/performance_p0/test_streamload_perfomance.groovy
index ef7e689e1d..adb0cafb87 100644
--- a/regression-test/suites/performance_p0/test_streamload_perfomance.groovy
+++ b/regression-test/suites/performance_p0/test_streamload_perfomance.groovy
@@ -37,7 +37,7 @@ suite("test_streamload_perfomance") {
streamLoad {
table tableName
- time 5000
+ time 10000
inputIterator rowIt
}
} finally {
diff --git a/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_orc.groovy b/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_orc.groovy
index 13b12e0027..e2c2cf8b1e 100644
--- a/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_orc.groovy
+++ b/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_orc.groovy
@@ -797,35 +797,10 @@ order by
"""
}
- def set_be_config = { ->
- String[][] backends = sql """ show backends; """
- assertTrue(backends.size() > 0)
- for (String[] backend in backends) {
- // No need to set this config anymore, but leave this code sample here
- // StringBuilder setConfigCommand = new StringBuilder();
- // setConfigCommand.append("curl -X POST http://")
- // setConfigCommand.append(backend[2])
- // setConfigCommand.append(":")
- // setConfigCommand.append(backend[5])
- // setConfigCommand.append("/api/update_config?")
- // String command1 = setConfigCommand.toString() + "enable_new_load_scan_node=true"
- // logger.info(command1)
- // String command2 = setConfigCommand.toString() + "enable_new_file_scanner=true"
- // logger.info(command2)
- // def process1 = command1.execute()
- // int code = process1.waitFor()
- // assertEquals(code, 0)
- // def process2 = command2.execute()
- // code = process1.waitFor()
- // assertEquals(code, 0)
- }
- }
-
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
String hms_port = context.config.otherConfigs.get("hms_port")
String catalog_name = "test_catalog_hive_orc"
- set_be_config.call()
sql """admin set frontend config ("enable_multi_catalog" = "true")"""
sql """drop catalog if exists ${catalog_name}"""
diff --git a/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_parquet.groovy b/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_parquet.groovy
index ce36a181a4..01530b738a 100644
--- a/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_parquet.groovy
+++ b/regression-test/suites/tpch_sf1_p0/multi_catalog_query/hive_catalog_parquet.groovy
@@ -797,35 +797,10 @@ order by
"""
}
- def set_be_config = { ->
- String[][] backends = sql """ show backends; """
- assertTrue(backends.size() > 0)
- for (String[] backend in backends) {
- // No need to set this config anymore, but leave this code sample here
- // StringBuilder setConfigCommand = new StringBuilder();
- // setConfigCommand.append("curl -X POST http://")
- // setConfigCommand.append(backend[2])
- // setConfigCommand.append(":")
- // setConfigCommand.append(backend[5])
- // setConfigCommand.append("/api/update_config?")
- // String command1 = setConfigCommand.toString() + "enable_new_load_scan_node=true"
- // logger.info(command1)
- // String command2 = setConfigCommand.toString() + "enable_new_file_scanner=true"
- // logger.info(command2)
- // def process1 = command1.execute()
- // int code = process1.waitFor()
- // assertEquals(code, 0)
- // def process2 = command2.execute()
- // code = process1.waitFor()
- // assertEquals(code, 0)
- }
- }
-
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
String hms_port = context.config.otherConfigs.get("hms_port")
String catalog_name = "test_catalog_hive_parquet"
- set_be_config.call()
sql """admin set frontend config ("enable_multi_catalog" = "true")"""
sql """drop catalog if exists ${catalog_name}"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 01/21: [fix](merge-on-write) calc delete bitmap need all segments which _do_flush in one memtable (#15018)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit b027ae3490627ed6bb780ee1a7d064eda23ec4b3
Author: xueweizhang <zx...@163.com>
AuthorDate: Thu Dec 15 20:44:49 2022 +0800
[fix](merge-on-write) calc delete bitmap need all segments which _do_flush in one memtable (#15018)
when some case(need modify be.conf), a memtable may flush many segments and then calc delete bitmap with new data. but now, it just only load one segment with max sgement id and this bug will not cala delte bitmap with all data of all segment of one memtable, and will get many rows with same key from merge-on-write table.
---
be/src/olap/memtable.cpp | 19 ++++++++++++------
be/src/olap/memtable.h | 3 ++-
be/src/olap/rowset/beta_rowset.cpp | 35 ++++++++++++++++++++-------------
be/src/olap/rowset/beta_rowset.h | 3 ++-
be/src/olap/rowset/beta_rowset_writer.h | 2 ++
be/src/olap/rowset/rowset_writer.h | 2 ++
6 files changed, 42 insertions(+), 22 deletions(-)
diff --git a/be/src/olap/memtable.cpp b/be/src/olap/memtable.cpp
index 892592b797..ab1a0f5209 100644
--- a/be/src/olap/memtable.cpp
+++ b/be/src/olap/memtable.cpp
@@ -415,7 +415,8 @@ bool MemTable::need_to_agg() {
: memory_usage() >= config::memtable_max_buffer_size;
}
-Status MemTable::_generate_delete_bitmap() {
+Status MemTable::_generate_delete_bitmap(int64_t atomic_num_segments_before_flush,
+ int64_t atomic_num_segments_after_flush) {
// generate delete bitmap, build a tmp rowset and load recent segment
if (!_tablet->enable_unique_key_merge_on_write()) {
return Status::OK();
@@ -423,12 +424,11 @@ Status MemTable::_generate_delete_bitmap() {
auto rowset = _rowset_writer->build_tmp();
auto beta_rowset = reinterpret_cast<BetaRowset*>(rowset.get());
std::vector<segment_v2::SegmentSharedPtr> segments;
- segment_v2::SegmentSharedPtr segment;
- if (beta_rowset->num_segments() == 0) {
+ if (atomic_num_segments_before_flush >= atomic_num_segments_after_flush) {
return Status::OK();
}
- RETURN_IF_ERROR(beta_rowset->load_segment(beta_rowset->num_segments() - 1, &segment));
- segments.push_back(segment);
+ RETURN_IF_ERROR(beta_rowset->load_segments(atomic_num_segments_before_flush,
+ atomic_num_segments_after_flush, &segments));
std::shared_lock meta_rlock(_tablet->get_header_lock());
// tablet is under alter process. The delete bitmap will be calculated after conversion.
if (_tablet->tablet_state() == TABLET_NOTREADY &&
@@ -445,8 +445,15 @@ Status MemTable::flush() {
VLOG_CRITICAL << "begin to flush memtable for tablet: " << tablet_id()
<< ", memsize: " << memory_usage() << ", rows: " << _rows;
int64_t duration_ns = 0;
+ // For merge_on_write table, it must get all segments in this flush.
+ // The id of new segment is set by the _num_segment of beta_rowset_writer,
+ // and new segment ids is between [atomic_num_segments_before_flush, atomic_num_segments_after_flush),
+ // and use the ids to load segment data file for calc delete bitmap.
+ int64_t atomic_num_segments_before_flush = _rowset_writer->get_atomic_num_segment();
RETURN_NOT_OK(_do_flush(duration_ns));
- RETURN_NOT_OK(_generate_delete_bitmap());
+ int64_t atomic_num_segments_after_flush = _rowset_writer->get_atomic_num_segment();
+ RETURN_NOT_OK(_generate_delete_bitmap(atomic_num_segments_before_flush,
+ atomic_num_segments_after_flush));
DorisMetrics::instance()->memtable_flush_total->increment(1);
DorisMetrics::instance()->memtable_flush_duration_us->increment(duration_ns / 1000);
VLOG_CRITICAL << "after flush memtable for tablet: " << tablet_id()
diff --git a/be/src/olap/memtable.h b/be/src/olap/memtable.h
index 52589f35cc..e7f59ff151 100644
--- a/be/src/olap/memtable.h
+++ b/be/src/olap/memtable.h
@@ -146,7 +146,8 @@ private:
void _insert_one_row_from_block(RowInBlock* row_in_block);
void _aggregate_two_row_in_block(RowInBlock* new_row, RowInBlock* row_in_skiplist);
- Status _generate_delete_bitmap();
+ Status _generate_delete_bitmap(int64_t atomic_num_segments_before_flush,
+ int64_t atomic_num_segments_after_flush);
private:
TabletSharedPtr _tablet;
diff --git a/be/src/olap/rowset/beta_rowset.cpp b/be/src/olap/rowset/beta_rowset.cpp
index 8ed48fdf70..f04fbf5922 100644
--- a/be/src/olap/rowset/beta_rowset.cpp
+++ b/be/src/olap/rowset/beta_rowset.cpp
@@ -130,20 +130,27 @@ Status BetaRowset::load_segments(std::vector<segment_v2::SegmentSharedPtr>* segm
return Status::OK();
}
-Status BetaRowset::load_segment(int64_t seg_id, segment_v2::SegmentSharedPtr* segment) {
- DCHECK(seg_id >= 0);
- auto fs = _rowset_meta->fs();
- if (!fs || _schema == nullptr) {
- return Status::OLAPInternalError(OLAP_ERR_INIT_FAILED);
- }
- auto seg_path = segment_file_path(seg_id);
- auto cache_path = segment_cache_path(seg_id);
- auto s = segment_v2::Segment::open(fs, seg_path, cache_path, seg_id, rowset_id(), _schema,
- segment);
- if (!s.ok()) {
- LOG(WARNING) << "failed to open segment. " << seg_path << " under rowset " << unique_id()
- << " : " << s.to_string();
- return Status::OLAPInternalError(OLAP_ERR_ROWSET_LOAD_FAILED);
+Status BetaRowset::load_segments(int64_t seg_id_begin, int64_t seg_id_end,
+ std::vector<segment_v2::SegmentSharedPtr>* segments) {
+ int64_t seg_id = seg_id_begin;
+ while (seg_id < seg_id_end) {
+ DCHECK(seg_id >= 0);
+ auto fs = _rowset_meta->fs();
+ if (!fs || _schema == nullptr) {
+ return Status::OLAPInternalError(OLAP_ERR_INIT_FAILED);
+ }
+ auto seg_path = segment_file_path(seg_id);
+ auto cache_path = segment_cache_path(seg_id);
+ std::shared_ptr<segment_v2::Segment> segment;
+ auto s = segment_v2::Segment::open(fs, seg_path, cache_path, seg_id, rowset_id(), _schema,
+ &segment);
+ if (!s.ok()) {
+ LOG(WARNING) << "failed to open segment. " << seg_path << " under rowset "
+ << unique_id() << " : " << s.to_string();
+ return Status::OLAPInternalError(OLAP_ERR_ROWSET_LOAD_FAILED);
+ }
+ segments->push_back(std::move(segment));
+ seg_id++;
}
return Status::OK();
}
diff --git a/be/src/olap/rowset/beta_rowset.h b/be/src/olap/rowset/beta_rowset.h
index 7cd792bf69..1087fdcd18 100644
--- a/be/src/olap/rowset/beta_rowset.h
+++ b/be/src/olap/rowset/beta_rowset.h
@@ -87,7 +87,8 @@ public:
Status load_segments(std::vector<segment_v2::SegmentSharedPtr>* segments);
- Status load_segment(int64_t seg_id, segment_v2::SegmentSharedPtr* segment);
+ Status load_segments(int64_t seg_id_begin, int64_t seg_id_end,
+ std::vector<segment_v2::SegmentSharedPtr>* segments);
protected:
BetaRowset(TabletSchemaSPtr schema, const std::string& tablet_path,
diff --git a/be/src/olap/rowset/beta_rowset_writer.h b/be/src/olap/rowset/beta_rowset_writer.h
index 65662ecf1a..5f4077d4b5 100644
--- a/be/src/olap/rowset/beta_rowset_writer.h
+++ b/be/src/olap/rowset/beta_rowset_writer.h
@@ -80,6 +80,8 @@ public:
void compact_segments(SegCompactionCandidatesSharedPtr segments);
+ int32_t get_atomic_num_segment() const override { return _num_segment.load(); }
+
private:
template <typename RowType>
Status _add_row(const RowType& row);
diff --git a/be/src/olap/rowset/rowset_writer.h b/be/src/olap/rowset/rowset_writer.h
index 2713b3c60c..b56262643e 100644
--- a/be/src/olap/rowset/rowset_writer.h
+++ b/be/src/olap/rowset/rowset_writer.h
@@ -84,6 +84,8 @@ public:
return Status::NotSupported("to be implemented");
}
+ virtual int32_t get_atomic_num_segment() const = 0;
+
private:
DISALLOW_COPY_AND_ASSIGN(RowsetWriter);
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 18/21: [fix](schema) Fix create table error if Colocate tables not equal to bucket num (#15071)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 62342915065f41170b18be2b72bd4981e5fd4a71
Author: yuxuan-luo <11...@users.noreply.github.com>
AuthorDate: Mon Dec 19 09:24:14 2022 +0800
[fix](schema) Fix create table error if Colocate tables not equal to bucket num (#15071)
Co-authored-by: hugoluo <hu...@tencent.com>
---
.../java/org/apache/doris/catalog/ColocateGroupSchema.java | 12 ++++++++++++
.../src/main/java/org/apache/doris/common/ErrorCode.java | 2 ++
.../java/org/apache/doris/datasource/InternalCatalog.java | 1 +
3 files changed, 15 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateGroupSchema.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateGroupSchema.java
index 7d76a25cd0..ff1c957cd4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateGroupSchema.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateGroupSchema.java
@@ -30,6 +30,7 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
/*
@@ -113,6 +114,17 @@ public class ColocateGroupSchema implements Writable {
}
}
+ public void checkDynamicPartition(Map<String, String> properties,
+ DistributionInfo distributionInfo) throws DdlException {
+ if (properties.get(DynamicPartitionProperty.BUCKETS) != null) {
+ HashDistributionInfo info = (HashDistributionInfo) distributionInfo;
+ if (info.getBucketNum() != Integer.parseInt(properties.get(DynamicPartitionProperty.BUCKETS))) {
+ ErrorReport.reportDdlException(
+ ErrorCode.ERR_DYNAMIC_PARTITION_MUST_HAS_SAME_BUCKET_NUM_WITH_COLOCATE_TABLE, bucketsNum);
+ }
+ }
+ }
+
public static ColocateGroupSchema read(DataInput in) throws IOException {
ColocateGroupSchema schema = new ColocateGroupSchema();
schema.readFields(in);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java b/fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java
index 4f21d77a0c..a6a78d054b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java
@@ -1635,6 +1635,8 @@ public enum ErrorCode {
ERR_COLOCATE_NOT_COLOCATE_TABLE(5064, new byte[]{'4', '2', '0', '0', '0'},
"Table %s is not a colocated table"),
ERR_INVALID_OPERATION(5065, new byte[]{'4', '2', '0', '0', '0'}, "Operation %s is invalid"),
+ ERR_DYNAMIC_PARTITION_MUST_HAS_SAME_BUCKET_NUM_WITH_COLOCATE_TABLE(5063, new byte[]{'4', '2', '0', '0', '0'},
+ "Dynamic partition buckets must equal the distribution buckets if creating a colocate table: %s"),
ERROR_DYNAMIC_PARTITION_TIME_UNIT(5065, new byte[]{'4', '2', '0', '0', '0'},
"Unsupported time unit %s. Expect HOUR/DAY/WEEK/MONTH."),
ERROR_DYNAMIC_PARTITION_START_ZERO(5066, new byte[]{'4', '2', '0', '0', '0'},
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 22bf478c7b..6466a29e12 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -1970,6 +1970,7 @@ public class InternalCatalog implements CatalogIf<Database> {
if (groupSchema != null) {
// group already exist, check if this table can be added to this group
groupSchema.checkColocateSchema(olapTable);
+ groupSchema.checkDynamicPartition(properties, olapTable.getDefaultDistributionInfo());
}
// add table to this group, if group does not exist, create a new one
Env.getCurrentColocateIndex()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 04/21: [fix](memtracker) Fix load channel memory tracker are not refreshed in time (#15048)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 12192e3f139833ee9c88fcd804f4923a977c215e
Author: Xinyi Zou <zo...@gmail.com>
AuthorDate: Fri Dec 16 10:43:03 2022 +0800
[fix](memtracker) Fix load channel memory tracker are not refreshed in time (#15048)
---
be/src/common/daemon.cpp | 9 +++++++--
1 file changed, 7 insertions(+), 2 deletions(-)
diff --git a/be/src/common/daemon.cpp b/be/src/common/daemon.cpp
index 497edd3037..fb306c21a0 100644
--- a/be/src/common/daemon.cpp
+++ b/be/src/common/daemon.cpp
@@ -202,6 +202,9 @@ void Daemon::memory_maintenance_thread() {
int64_t interval_milliseconds = config::memory_maintenance_sleep_time_ms;
while (!_stop_background_threads_latch.wait_for(
std::chrono::milliseconds(interval_milliseconds))) {
+ if (!MemInfo::initialized()) {
+ continue;
+ }
// Refresh process memory metrics.
doris::PerfCounters::refresh_proc_status();
doris::MemInfo::refresh_proc_meminfo();
@@ -240,8 +243,10 @@ void Daemon::load_channel_tracker_refresh_thread() {
// Refresh the memory statistics of the load channel tracker more frequently,
// which helps to accurately control the memory of LoadChannelMgr.
while (!_stop_background_threads_latch.wait_for(
- std::chrono::seconds(config::load_channel_memory_refresh_sleep_time_ms))) {
- doris::ExecEnv::GetInstance()->load_channel_mgr()->refresh_mem_tracker();
+ std::chrono::milliseconds(config::load_channel_memory_refresh_sleep_time_ms))) {
+ if (ExecEnv::GetInstance()->initialized()) {
+ doris::ExecEnv::GetInstance()->load_channel_mgr()->refresh_mem_tracker();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 13/21: [fix](type) forbid time type when creating table (#15093)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit f6936f256d7b162947512837715305aeccc8b674
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Fri Dec 16 21:54:35 2022 +0800
[fix](type) forbid time type when creating table (#15093)
---
.../src/main/java/org/apache/doris/analysis/CreateTableStmt.java | 4 ++++
fe/fe-core/src/test/java/org/apache/doris/alter/AlterTest.java | 5 +++--
2 files changed, 7 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
index 7f33428da3..2a4f3759c7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
@@ -401,6 +401,10 @@ public class CreateTableStmt extends DdlStmt {
}
}
+ if (columnDef.getType().isTime() || columnDef.getType().isTimeV2()) {
+ throw new AnalysisException("Time type is not supported for olap table");
+ }
+
if (columnDef.getType().isObjectStored()) {
hasObjectStored = true;
objectStoredColumn = columnDef.getName();
diff --git a/fe/fe-core/src/test/java/org/apache/doris/alter/AlterTest.java b/fe/fe-core/src/test/java/org/apache/doris/alter/AlterTest.java
index d8c6f1e7fb..b5d435203e 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/alter/AlterTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/alter/AlterTest.java
@@ -118,7 +118,8 @@ public class AlterTest {
+ "PROPERTIES('replication_num' = '1');");
createTable(
- "CREATE TABLE test.tbl6\n" + "(\n" + " k1 datetime(3),\n" + " k2 time(3),\n" + " v1 int \n,"
+ "CREATE TABLE test.tbl6\n" + "(\n" + " k1 datetime(3),\n" + " k2 datetime(3),\n"
+ + " v1 int \n,"
+ " v2 datetime(3)\n" + ") ENGINE=OLAP\n" + "UNIQUE KEY (k1,k2)\n"
+ "PARTITION BY RANGE(k1)\n" + "(\n"
+ " PARTITION p1 values less than('2020-02-01 00:00:00'),\n"
@@ -410,7 +411,7 @@ public class AlterTest {
alterTable(stmt, true);
// no conflict
- stmt = "alter table test.tbl6 add column k3 int, add column k4 time(6)";
+ stmt = "alter table test.tbl6 add column k3 int, add column k4 datetime(6)";
alterTable(stmt, false);
waitSchemaChangeJobDone(false);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 20/21: [bug](udaf) fix java udaf incorrect get null value with row (#15151)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 6766c5135319b6902e2d94480de3ffc31ff4ff36
Author: zhangstar333 <87...@users.noreply.github.com>
AuthorDate: Mon Dec 19 10:07:12 2022 +0800
[bug](udaf) fix java udaf incorrect get null value with row (#15151)
---
fe/java-udf/src/main/java/org/apache/doris/udf/UdafExecutor.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/fe/java-udf/src/main/java/org/apache/doris/udf/UdafExecutor.java b/fe/java-udf/src/main/java/org/apache/doris/udf/UdafExecutor.java
index 598886a26b..f684363bf6 100644
--- a/fe/java-udf/src/main/java/org/apache/doris/udf/UdafExecutor.java
+++ b/fe/java-udf/src/main/java/org/apache/doris/udf/UdafExecutor.java
@@ -361,7 +361,8 @@ public class UdafExecutor {
for (int i = 0; i < argTypes.length; ++i) {
// skip the input column of current row is null
if (UdfUtils.UNSAFE.getLong(null, UdfUtils.getAddressAtOffset(inputNullsPtrs, i)) != -1
- && UdfUtils.UNSAFE.getByte(null, UdfUtils.getAddressAtOffset(inputNullsPtrs, i) + row) == 1) {
+ && (UdfUtils.UNSAFE.getByte(null, UdfUtils.UNSAFE.getLong(null,
+ UdfUtils.getAddressAtOffset(inputNullsPtrs, i)) + row) == 1)) {
inputObjects[i] = null;
continue;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 17/21: [fix](multi-catalog) hidden password for show create jdbc catalog (#15145)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit afbe5ba198f198550185fb474f9fa021e49cc406
Author: xueweizhang <zx...@163.com>
AuthorDate: Sat Dec 17 17:20:17 2022 +0800
[fix](multi-catalog) hidden password for show create jdbc catalog (#15145)
when show create catalog of jdbc, it will show 'jdbc.password' plain text. fix it like other code that hidden password.
---
.../org/apache/doris/common/util/PrintableMap.java | 1 +
.../org/apache/doris/datasource/CatalogMgr.java | 2 +-
.../query_p0/show/test_show_create_catalog.out | 4 ++
.../query_p0/show/test_show_create_catalog.groovy | 46 ++++++++++++++++++++++
4 files changed, 52 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/PrintableMap.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/PrintableMap.java
index 5f6412bf96..571fa92975 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/PrintableMap.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/PrintableMap.java
@@ -38,6 +38,7 @@ public class PrintableMap<K, V> {
SENSITIVE_KEY.add("password");
SENSITIVE_KEY.add("kerberos_keytab_content");
SENSITIVE_KEY.add("bos_secret_accesskey");
+ SENSITIVE_KEY.add("jdbc.password");
}
public PrintableMap(Map<K, V> map, String keyValueSeparator,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
index f3b9299bc8..8ea37fd245 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
@@ -387,7 +387,7 @@ public class CatalogMgr implements Writable, GsonPostProcessable {
.append("`");
if (catalog.getProperties().size() > 0) {
sb.append(" PROPERTIES (\n");
- sb.append(new PrintableMap<>(catalog.getProperties(), "=", true, true, false));
+ sb.append(new PrintableMap<>(catalog.getProperties(), "=", true, true, true));
sb.append("\n);");
}
diff --git a/regression-test/data/query_p0/show/test_show_create_catalog.out b/regression-test/data/query_p0/show/test_show_create_catalog.out
new file mode 100644
index 0000000000..92fa98548a
--- /dev/null
+++ b/regression-test/data/query_p0/show/test_show_create_catalog.out
@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+test_show_create_mysql_jdbc_catalog CREATE CATALOG `test_show_create_mysql_jdbc_catalog` PROPERTIES (\n"jdbc.password" = "*XXX",\n"jdbc.driver_class" = "com.mysql.cj.jdbc.Driver",\n"jdbc.user" = "root",\n"jdbc.jdbc_url" = "jdbc:mysql://127.0.0.1:3316/doris_test?useSSL=false",\n"jdbc.driver_url" = "https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/jdbc_driver/mysql-connector-java-8.0.25.jar",\n"type" = "jdbc"\n);
+
diff --git a/regression-test/suites/query_p0/show/test_show_create_catalog.groovy b/regression-test/suites/query_p0/show/test_show_create_catalog.groovy
new file mode 100644
index 0000000000..a30adab3c5
--- /dev/null
+++ b/regression-test/suites/query_p0/show/test_show_create_catalog.groovy
@@ -0,0 +1,46 @@
+// 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_show_create_catalog", "query") {
+ String catalog_name = "test_show_create_mysql_jdbc_catalog";
+ try {
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ String mysql_port = context.config.otherConfigs.get("mysql_57_port");
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ sql """admin set frontend config ("enable_multi_catalog" = "true")"""
+
+ sql """drop catalog if exists ${catalog_name} """
+
+ // if use 'com.mysql.cj.jdbc.Driver' here, it will report: ClassNotFound
+ sql """ CREATE CATALOG ${catalog_name} PROPERTIES (
+ "type"="jdbc",
+ "jdbc.user"="root",
+ "jdbc.password"="123456",
+ "jdbc.jdbc_url" = "jdbc:mysql://127.0.0.1:${mysql_port}/doris_test?useSSL=false",
+ "jdbc.driver_url" = "https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/jdbc_driver/mysql-connector-java-8.0.25.jar",
+ "jdbc.driver_class" = "com.mysql.cj.jdbc.Driver");
+ """
+
+ qt_select "show create catalog `${catalog_name}`"
+
+ }
+ } finally {
+
+ try_sql("DROP CATALOG IF EXISTS `${catalog_name}`")
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org