dataroaring commented on code in PR #19917: URL: https://github.com/apache/doris/pull/19917#discussion_r1223880507
########## fe/fe-core/src/main/java/org/apache/doris/catalog/AutoIncrementGenerator.java: ########## @@ -0,0 +1,104 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.catalog; + +import org.apache.doris.common.Pair; +import org.apache.doris.common.UserException; +import org.apache.doris.common.io.Text; +import org.apache.doris.common.io.Writable; +import org.apache.doris.persist.AutoIncrementIdUpdateLog; +import org.apache.doris.persist.EditLog; +import org.apache.doris.persist.gson.GsonUtils; + +import com.google.gson.annotations.SerializedName; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; + +public class AutoIncrementGenerator implements Writable { + private static final Logger LOG = LogManager.getLogger(AutoIncrementGenerator.class); + + public static final long NEXT_ID_INIT_VALUE = 1; + // _MIN_BATCH_SIZE = 4064 in load task + private static final long BATCH_ID_INTERVAL = 50000; + + @SerializedName(value = "dbId") + private Long dbId; + @SerializedName(value = "tableId") + private Long tableId; + @SerializedName(value = "columnId") + private Long columnId; + @SerializedName(value = "nextId") + private long nextId; + @SerializedName(value = "batchEndId") + private long batchEndId; + + private EditLog editLog; + + public AutoIncrementGenerator() { + } + + public AutoIncrementGenerator(long dbId, long tableId, long columnId) { + this.dbId = dbId; + this.tableId = tableId; + this.columnId = columnId; + } + + public void setEditLog(EditLog editLog) { + this.editLog = editLog; + } + + public synchronized void applyChange(long columnId, long batchNextId) { + if (this.columnId == columnId && batchEndId < batchNextId) { + nextId = batchNextId; + batchEndId = batchNextId; + } + } + + public synchronized Pair<Long, Long> getAutoIncrementRange(long columnId, + long length, long lowerBound) throws UserException { + LOG.info("[getAutoIncrementRange request][col:{}][length:{}], [{}]", columnId, length, this.columnId); + if (this.columnId != columnId) { + throw new UserException("column dosen't exist, columnId=" + columnId); + } + long startId = Math.max(nextId, lowerBound); + long endId = startId + length; + nextId = startId + length; + if (endId > batchEndId) { + batchEndId = (endId / BATCH_ID_INTERVAL + 1) * BATCH_ID_INTERVAL; + if (editLog != null) { + AutoIncrementIdUpdateLog info = new AutoIncrementIdUpdateLog(dbId, tableId, columnId, batchEndId); + editLog.logUpdateAutoIncrementId(info); + } Review Comment: We cannot go ahead if editLog == null. ########## be/src/vec/sink/vtablet_sink.cpp: ########## @@ -968,6 +977,96 @@ void VNodeChannel::mark_close() { _eos_is_produced = true; } +FetchAutoIncIDExecutor::FetchAutoIncIDExecutor() { + ThreadPoolBuilder("AsyncFetchAutoIncIDExecutor") + .set_min_threads(1) + .set_max_threads(3) + .set_max_queue_size(std::numeric_limits<int>::max()) + .build(&_pool); +} + +AutoIncIDBuffer::AutoIncIDBuffer(size_t batch_size, int64_t db_id, int64_t table_id, + TNetworkAddress master_addr) + : _batch_size(std::max(batch_size, MIN_BATCH_SIZE)), + _prefetch_size(_batch_size * 10), + _low_water_level_mark(_batch_size * 3), + _db_id(db_id), + _table_id(table_id), + _master_addr(master_addr), + _token(FetchAutoIncIDExecutor::GetInstance()->_pool->new_token( + ThreadPool::ExecutionMode::CONCURRENT)) {} + +void AutoIncIDBuffer::set_column_id(int64_t column_id) { + _column_id = column_id; +} + +Status AutoIncIDBuffer::consume(size_t length, bool eos, + std::vector<std::pair<size_t, size_t>>* result) { + result->reserve(2); + if (_front_buffer.second > 0) { + auto min_length = std::min(_front_buffer.second, length); + length -= min_length; + result->emplace_back(_front_buffer.first, min_length); + _front_buffer.first += min_length; + _front_buffer.second -= min_length; + } + if (length > 0) { + _token->wait(); + if (_rpc_status != Status::OK()) { + return _rpc_status; + } + + { + std::lock_guard<std::mutex> lock(_mutex); + std::swap(_front_buffer, _backend_buffer); + } + + DCHECK(length <= _front_buffer.second); + result->emplace_back(_front_buffer.first, length); + _front_buffer.first += length; + _front_buffer.second -= length; + } + if (!eos && _front_buffer.second <= _low_water_level_mark) { + prefetch_ids(_prefetch_size); + } + return Status::OK(); +} + +void AutoIncIDBuffer::prefetch_ids(size_t length) { + _token->submit_func([=, this]() { + TAutoIncrementRangeRequest request; + TAutoIncrementRangeResult result; + request.db_id = _db_id; + request.table_id = _table_id; + request.column_id = _column_id; + request.length = length; + + auto rpc_begin = std::chrono::system_clock::now(); + _rpc_status = ThriftRpcHelper::rpc<FrontendServiceClient>( + _master_addr.hostname, _master_addr.port, + [&request, &result](FrontendServiceConnection& client) { + client->getAutoIncrementRange(result, request); + }); + auto rpc_end = std::chrono::system_clock::now(); Review Comment: there are timers in a stop_watch.h . ########## docs/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE.md: ########## @@ -114,7 +114,12 @@ distribution_desc HLL_UNION：HLL 类型的列的聚合方式，通过 HyperLogLog 算法聚合。 BITMAP_UNION：BIMTAP 类型的列的聚合方式，进行位图的并集聚合。 ``` + * `AUTO_INCREMENT`(在Apache Doris 2.0.0 beta版本支持) Review Comment: it is not supported in 2.0.0 now. ########## be/src/vec/sink/vtablet_sink.h: ########## @@ -447,6 +449,73 @@ class IndexChannel { std::map<int64_t, std::vector<std::pair<int64_t, int64_t>>> _tablets_received_rows; }; +struct FetchAutoIncIDExecutor { + FetchAutoIncIDExecutor(); + + static FetchAutoIncIDExecutor* GetInstance() { + static FetchAutoIncIDExecutor instance; + return &instance; + } + + std::unique_ptr<ThreadPool> _pool; +}; + +// estimate the number of remaining rows based on the rows read and bytes read +struct RowsEstimator { + static constexpr size_t BATCH_INTERVAL = 5000; Review Comment: It should be a config. ########## fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java: ########## @@ -2242,6 +2245,41 @@ public TGetTabletReplicaInfosResult getTabletReplicaInfos(TGetTabletReplicaInfos return result; } + @Override + public TAutoIncrementRangeResult getAutoIncrementRange(TAutoIncrementRangeRequest request) { + String clientAddr = getClientAddrAsString(); + LOG.debug("receive get auto-increement range request: {}, backend: {}", request, clientAddr); + + TAutoIncrementRangeResult result = new TAutoIncrementRangeResult(); + TStatus status = new TStatus(TStatusCode.OK); + result.setStatus(status); + try { + Env env = Env.getCurrentEnv(); + Database db = env.getInternalCatalog().getDbOrMetaException(request.getDbId()); + OlapTable olapTable = (OlapTable) db.getTableOrMetaException(request.getTableId(), TableType.OLAP); + AutoIncrementGenerator autoIncrementGenerator = null; + autoIncrementGenerator = olapTable.getAutoIncrentGenerator(); Review Comment: Incrent -> Increment ########## docs/en/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE.md: ########## @@ -114,6 +114,11 @@ distribution_desc HLL_UNION: The aggregation method of HLL type columns, aggregated by HyperLogLog algorithm. BITMAP_UNION: The aggregation mode of BIMTAP type columns, which performs the union aggregation of bitmaps. ``` + * `AUTO_INCREMENT`(supported in 在Apache Doris 2.0.0 beta) Review Comment: supported in master. ########## be/src/vec/sink/vtablet_sink.cpp: ########## @@ -968,6 +977,96 @@ void VNodeChannel::mark_close() { _eos_is_produced = true; } +FetchAutoIncIDExecutor::FetchAutoIncIDExecutor() { + ThreadPoolBuilder("AsyncFetchAutoIncIDExecutor") + .set_min_threads(1) + .set_max_threads(3) + .set_max_queue_size(std::numeric_limits<int>::max()) + .build(&_pool); +} + +AutoIncIDBuffer::AutoIncIDBuffer(size_t batch_size, int64_t db_id, int64_t table_id, + TNetworkAddress master_addr) + : _batch_size(std::max(batch_size, MIN_BATCH_SIZE)), + _prefetch_size(_batch_size * 10), + _low_water_level_mark(_batch_size * 3), + _db_id(db_id), + _table_id(table_id), + _master_addr(master_addr), + _token(FetchAutoIncIDExecutor::GetInstance()->_pool->new_token( + ThreadPool::ExecutionMode::CONCURRENT)) {} + +void AutoIncIDBuffer::set_column_id(int64_t column_id) { + _column_id = column_id; +} + +Status AutoIncIDBuffer::consume(size_t length, bool eos, + std::vector<std::pair<size_t, size_t>>* result) { + result->reserve(2); + if (_front_buffer.second > 0) { + auto min_length = std::min(_front_buffer.second, length); + length -= min_length; + result->emplace_back(_front_buffer.first, min_length); + _front_buffer.first += min_length; + _front_buffer.second -= min_length; + } + if (length > 0) { + _token->wait(); + if (_rpc_status != Status::OK()) { + return _rpc_status; + } + + { + std::lock_guard<std::mutex> lock(_mutex); + std::swap(_front_buffer, _backend_buffer); + } + + DCHECK(length <= _front_buffer.second); + result->emplace_back(_front_buffer.first, length); + _front_buffer.first += length; + _front_buffer.second -= length; + } + if (!eos && _front_buffer.second <= _low_water_level_mark) { + prefetch_ids(_prefetch_size); + } + return Status::OK(); +} + +void AutoIncIDBuffer::prefetch_ids(size_t length) { + _token->submit_func([=, this]() { + TAutoIncrementRangeRequest request; + TAutoIncrementRangeResult result; + request.db_id = _db_id; + request.table_id = _table_id; + request.column_id = _column_id; + request.length = length; + + auto rpc_begin = std::chrono::system_clock::now(); + _rpc_status = ThriftRpcHelper::rpc<FrontendServiceClient>( + _master_addr.hostname, _master_addr.port, Review Comment: master addr may be changed due to fe's leader changed. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
