This is an automated email from the ASF dual-hosted git repository. caiconghui pushed a commit to branch tablet_index in repository https://gitbox.apache.org/repos/asf/doris.git
commit e45536fd1dca2bddc6bcc5b7e55b0018f3b0a3a2 Author: caiconghui1 <[email protected]> AuthorDate: Wed Nov 8 15:35:31 2023 +0800 [enhancement](random_sink) change tablet search algorithm from random to round-robin for random distribution table fix --- be/src/exec/tablet_info.h | 14 +-- be/src/olap/base_compaction.cpp | 2 +- be/src/vec/sink/vtablet_finder.cpp | 16 ++- be/src/vec/sink/vtablet_finder.h | 15 +-- be/src/vec/sink/writer/vtablet_writer.cpp | 1 - .../main/java/org/apache/doris/catalog/Env.java | 12 +-- .../org/apache/doris/planner/OlapTableSink.java | 18 ++-- .../doris/planner/SingleTabletLoadRecorderMgr.java | 112 --------------------- .../doris/planner/TabletLoadIndexRecorderMgr.java | 89 ++++++++++++++++ gensrc/thrift/Descriptors.thrift | 2 +- 10 files changed, 132 insertions(+), 149 deletions(-) diff --git a/be/src/exec/tablet_info.h b/be/src/exec/tablet_info.h index bb9fbd8bc60..9b735cc0117 100644 --- a/be/src/exec/tablet_info.h +++ b/be/src/exec/tablet_info.h @@ -127,7 +127,7 @@ struct VOlapTablePartition { int64_t num_buckets = 0; std::vector<OlapTableIndexTablets> indexes; bool is_mutable; - // -1 indicates load_to_single_tablet = false + // -1 indicates partition with hash distribution int64_t load_tablet_idx = -1; VOlapTablePartition(vectorized::Block* partition_block) @@ -184,7 +184,7 @@ public: ALWAYS_INLINE void find_tablets( vectorized::Block* block, const std::vector<uint32_t>& indexes, - const std::vector<VOlapTablePartition*>& partitions, + std::vector<VOlapTablePartition*>& partitions, std::vector<uint32_t>& tablet_indexes /*result*/, /*TODO: check if flat hash map will be better*/ std::map<int64_t, int64_t>* partition_tablets_buffer = nullptr) const { @@ -212,10 +212,9 @@ public: compute_function = [](vectorized::Block* block, uint32_t row, const VOlapTablePartition& partition) -> uint32_t { if (partition.load_tablet_idx == -1) { - // load_to_single_tablet = false, just do random + // for compatible with old version, just do random return butil::fast_rand() % partition.num_buckets; } - // load_to_single_tablet = ture, do round-robin return partition.load_tablet_idx % partition.num_buckets; }; } @@ -230,10 +229,11 @@ public: if (auto it = partition_tablets_buffer->find(partition_id); it != partition_tablets_buffer->end()) { tablet_indexes[index] = it->second; // tablet - } - // compute and save in buffer - (*partition_tablets_buffer)[partition_id] = tablet_indexes[index] = + } else { + // compute and save in buffer + (*partition_tablets_buffer)[partition_id] = tablet_indexes[index] = compute_function(block, index, *partitions[index]); + } } } } diff --git a/be/src/olap/base_compaction.cpp b/be/src/olap/base_compaction.cpp index eab41c73acb..6ae006709a7 100644 --- a/be/src/olap/base_compaction.cpp +++ b/be/src/olap/base_compaction.cpp @@ -98,7 +98,7 @@ Status BaseCompaction::execute_compact_impl() { void BaseCompaction::_filter_input_rowset() { // if dup_key and no delete predicate - // we skip big files too save resources + // we skip big files to save resources if (_tablet->keys_type() != KeysType::DUP_KEYS) { return; } diff --git a/be/src/vec/sink/vtablet_finder.cpp b/be/src/vec/sink/vtablet_finder.cpp index f01add4b22e..25109bb87f3 100644 --- a/be/src/vec/sink/vtablet_finder.cpp +++ b/be/src/vec/sink/vtablet_finder.cpp @@ -96,8 +96,20 @@ Status OlapTabletFinder::find_tablets(RuntimeState* state, Block* block, int row if (_find_tablet_mode == FindTabletMode::FIND_TABLET_EVERY_ROW) { _vpartition->find_tablets(block, qualified_rows, partitions, tablet_index); } else { - _vpartition->find_tablets(block, qualified_rows, partitions, tablet_index, - &_partition_to_tablet_map); + // for random distribution + _vpartition->find_tablets(block, qualified_rows, partitions, tablet_index, &_partition_to_tablet_map); + if (_find_tablet_mode == FindTabletMode::FIND_TABLET_EVERY_BATCH) { + for (auto row_index : qualified_rows) { + auto partition = partitions[row_index]; + if (_partition_to_tablet_map.find(partition->id) != _partition_to_tablet_map.end()) { + // do round-robin for next batch + if (partition->load_tablet_idx != -1) { + partition->load_tablet_idx++; + } + _partition_to_tablet_map.erase(partition->id); + } + } + } } return Status::OK(); diff --git a/be/src/vec/sink/vtablet_finder.h b/be/src/vec/sink/vtablet_finder.h index 3426f7cb67d..b44fb4f1fea 100644 --- a/be/src/vec/sink/vtablet_finder.h +++ b/be/src/vec/sink/vtablet_finder.h @@ -30,12 +30,13 @@ namespace doris::vectorized { class OlapTabletFinder { public: - // FIND_TABLET_EVERY_ROW is used for both hash and random distribution info, which indicates that we + // FIND_TABLET_EVERY_ROW is used for hash distribution info, which indicates that we // should compute tablet index for every row - // FIND_TABLET_EVERY_BATCH is only used for random distribution info, which indicates that we should + // FIND_TABLET_EVERY_BATCH is used for random distribution info, which indicates that we should // compute tablet index for every row batch - // FIND_TABLET_EVERY_SINK is only used for random distribution info, which indicates that we should - // only compute tablet index in the corresponding partition once for the whole time in olap table sink + // FIND_TABLET_EVERY_SINK is used for random distribution info when load_to_single_tablet set to true, + // which indicates that we should only compute tablet index in the corresponding partition once for the + // whole time in olap table sink enum FindTabletMode { FIND_TABLET_EVERY_ROW, FIND_TABLET_EVERY_BATCH, FIND_TABLET_EVERY_SINK }; OlapTabletFinder(VOlapTablePartitionParam* vpartition, FindTabletMode mode) @@ -50,12 +51,6 @@ public: return _find_tablet_mode == FindTabletMode::FIND_TABLET_EVERY_SINK; } - void clear_for_new_batch() { - if (_find_tablet_mode == FindTabletMode::FIND_TABLET_EVERY_BATCH) { - _partition_to_tablet_map.clear(); - } - } - bool is_single_tablet() { return _partition_to_tablet_map.size() == 1; } const vectorized::flat_hash_set<int64_t>& partition_ids() { return _partition_ids; } diff --git a/be/src/vec/sink/writer/vtablet_writer.cpp b/be/src/vec/sink/writer/vtablet_writer.cpp index 786c0e21105..6285736601a 100644 --- a/be/src/vec/sink/writer/vtablet_writer.cpp +++ b/be/src/vec/sink/writer/vtablet_writer.cpp @@ -1616,7 +1616,6 @@ Status VTabletWriter::append_block(doris::vectorized::Block& input_block) { _state->update_num_bytes_load_total(bytes); DorisMetrics::instance()->load_rows->increment(rows); DorisMetrics::instance()->load_bytes->increment(bytes); - // Random distribution and the block belongs to a single tablet, we could optimize to append the whole // block into node channel. bool load_block_to_single_tablet = diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java index 8c028bc8fc9..ae047fea74f 100755 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @@ -205,7 +205,7 @@ import org.apache.doris.persist.TruncateTableInfo; import org.apache.doris.persist.meta.MetaHeader; import org.apache.doris.persist.meta.MetaReader; import org.apache.doris.persist.meta.MetaWriter; -import org.apache.doris.planner.SingleTabletLoadRecorderMgr; +import org.apache.doris.planner.TabletLoadIndexRecorderMgr; import org.apache.doris.plugin.PluginInfo; import org.apache.doris.plugin.PluginMgr; import org.apache.doris.policy.PolicyMgr; @@ -334,7 +334,7 @@ public class Env { private LoadManager loadManager; private ProgressManager progressManager; private StreamLoadRecordMgr streamLoadRecordMgr; - private SingleTabletLoadRecorderMgr singleTabletLoadRecorderMgr; + private TabletLoadIndexRecorderMgr tabletLoadIndexRecorderMgr; private RoutineLoadManager routineLoadManager; private SqlBlockRuleMgr sqlBlockRuleMgr; private ExportMgr exportMgr; @@ -693,7 +693,7 @@ public class Env { this.progressManager = new ProgressManager(); this.streamLoadRecordMgr = new StreamLoadRecordMgr("stream_load_record_manager", Config.fetch_stream_load_record_interval_second * 1000L); - this.singleTabletLoadRecorderMgr = new SingleTabletLoadRecorderMgr(); + this.tabletLoadIndexRecorderMgr = new TabletLoadIndexRecorderMgr(); this.loadEtlChecker = new LoadEtlChecker(loadManager); this.loadLoadingChecker = new LoadLoadingChecker(loadManager); this.routineLoadScheduler = new RoutineLoadScheduler(routineLoadManager); @@ -1561,7 +1561,7 @@ public class Env { cooldownConfHandler.start(); } streamLoadRecordMgr.start(); - singleTabletLoadRecorderMgr.start(); + tabletLoadIndexRecorderMgr.start(); getInternalCatalog().getIcebergTableCreationRecordMgr().start(); new InternalSchemaInitializer().start(); if (Config.enable_hms_events_incremental_sync) { @@ -3782,8 +3782,8 @@ public class Env { return streamLoadRecordMgr; } - public SingleTabletLoadRecorderMgr getSingleTabletLoadRecorderMgr() { - return singleTabletLoadRecorderMgr; + public TabletLoadIndexRecorderMgr getTabletLoadIndexRecorderMgr() { + return tabletLoadIndexRecorderMgr; } public IcebergTableCreationRecordMgr getIcebergTableCreationRecordMgr() { diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java index 23bec446f4b..fa3b4abee9a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java @@ -23,6 +23,7 @@ import org.apache.doris.analysis.SlotDescriptor; import org.apache.doris.analysis.TupleDescriptor; import org.apache.doris.catalog.Column; import org.apache.doris.catalog.DistributionInfo; +import org.apache.doris.catalog.DistributionInfo.DistributionInfoType; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.HashDistributionInfo; import org.apache.doris.catalog.Index; @@ -109,8 +110,6 @@ public class OlapTableSink extends DataSink { private boolean isStrictMode = false; - private boolean loadToSingleTablet; - public OlapTableSink(OlapTable dstTable, TupleDescriptor tupleDescriptor, List<Long> partitionIds, boolean singleReplicaLoad) { this.dstTable = dstTable; @@ -134,7 +133,6 @@ public class OlapTableSink extends DataSink { "if load_to_single_tablet set to true," + " the olap table must be with random distribution"); } tSink.setLoadToSingleTablet(loadToSingleTablet); - this.loadToSingleTablet = loadToSingleTablet; tDataSink = new TDataSink(getDataSinkType()); tDataSink.setOlapTableSink(tSink); @@ -344,11 +342,12 @@ public class OlapTableSink extends DataSink { tPartition.setNumBuckets(index.getTablets().size()); } tPartition.setIsMutable(table.getPartitionInfo().getIsMutable(partitionId)); - if (loadToSingleTablet) { - int tabletIndex = Env.getCurrentEnv().getSingleTabletLoadRecorderMgr() - .getCurrentLoadTabletIndex(dbId, table.getId(), partitionId); + if (partition.getDistributionInfo().getType() == DistributionInfoType.RANDOM) { + int tabletIndex = Env.getCurrentEnv().getTabletLoadIndexRecorderMgr() + .getCurrentTabletLoadIndex(dbId, table.getId(), partition); tPartition.setLoadTabletIdx(tabletIndex); } + partitionParam.addToPartitions(tPartition); DistributionInfo distInfo = partition.getDistributionInfo(); @@ -403,9 +402,10 @@ public class OlapTableSink extends DataSink { index.getTablets().stream().map(Tablet::getId).collect(Collectors.toList())))); tPartition.setNumBuckets(index.getTablets().size()); } - if (loadToSingleTablet) { - int tabletIndex = Env.getCurrentEnv().getSingleTabletLoadRecorderMgr() - .getCurrentLoadTabletIndex(dbId, table.getId(), partition.getId()); + + if (partition.getDistributionInfo().getType() == DistributionInfoType.RANDOM) { + int tabletIndex = Env.getCurrentEnv().getTabletLoadIndexRecorderMgr() + .getCurrentTabletLoadIndex(dbId, table.getId(), partition); tPartition.setLoadTabletIdx(tabletIndex); } partitionParam.addToPartitions(tPartition); diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SingleTabletLoadRecorderMgr.java b/fe/fe-core/src/main/java/org/apache/doris/planner/SingleTabletLoadRecorderMgr.java deleted file mode 100644 index 89f21c47dc2..00000000000 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/SingleTabletLoadRecorderMgr.java +++ /dev/null @@ -1,112 +0,0 @@ -// 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.planner; - -import org.apache.doris.catalog.Env; -import org.apache.doris.catalog.MaterializedIndex; -import org.apache.doris.catalog.OlapTable; -import org.apache.doris.catalog.TableIf; -import org.apache.doris.common.UserException; -import org.apache.doris.common.util.MasterDaemon; - -import lombok.Getter; -import org.apache.commons.lang3.tuple.Triple; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - -import java.util.concurrent.ConcurrentHashMap; - -public class SingleTabletLoadRecorderMgr extends MasterDaemon { - private static final Logger LOG = LogManager.getLogger(SingleTabletLoadRecorderMgr.class); - private static final long EXPIRY_TIME_INTERVAL_MS = 86400000; // 1 * 24 * 60 * 60 * 1000, 1 days - - // <<db_id, table_id, partition_id> -> load_tablet_record> - // 0 =< load_tablet_index < number_buckets - private final ConcurrentHashMap<Triple<Long, Long, Long>, TabletUpdateRecord> loadTabletRecordMap = - new ConcurrentHashMap<>(); - - public SingleTabletLoadRecorderMgr() { - super("single_tablet_load_recorder", EXPIRY_TIME_INTERVAL_MS); - } - - @Override - protected void runAfterCatalogReady() { - long expiryTime = System.currentTimeMillis() - EXPIRY_TIME_INTERVAL_MS; - loadTabletRecordMap.entrySet().removeIf(entry -> - entry.getValue().getUpdateTimestamp() < expiryTime - ); - LOG.info("Remove expired load tablet record successfully."); - } - - public int getCurrentLoadTabletIndex(long dbId, long tableId, long partitionId) throws UserException { - Triple<Long, Long, Long> key = Triple.of(dbId, tableId, partitionId); - TabletUpdateRecord record = loadTabletRecordMap.get(key); - int numBuckets = -1; - if (record == null) { - numBuckets = getNumBuckets(dbId, tableId, partitionId); - } - return createOrUpdateLoadTabletRecord(key, numBuckets); - } - - private int getNumBuckets(long dbId, long tableId, long partitionId) throws UserException { - OlapTable olapTable = (OlapTable) Env.getCurrentInternalCatalog().getDb(dbId) - .flatMap(db -> db.getTable(tableId)).filter(t -> t.getType() == TableIf.TableType.OLAP) - .orElse(null); - if (olapTable == null) { - throw new UserException("Olap table[" + dbId + "." + tableId + "] is not exist."); - } - return olapTable.getPartition(partitionId) - .getMaterializedIndices(MaterializedIndex.IndexExtState.ALL) - .get(0).getTablets().size(); - } - - private int createOrUpdateLoadTabletRecord(Triple<Long, Long, Long> key, int numBuckets) { - TabletUpdateRecord record = loadTabletRecordMap.compute(key, (k, existingRecord) -> { - if (existingRecord == null) { - return new TabletUpdateRecord(0, numBuckets); - } else { - existingRecord.updateRecord(); - return existingRecord; - } - }); - return record.getTabletIndex(); - } - - static class TabletUpdateRecord { - @Getter - // 0 =< load_tablet_index < number_buckets - int tabletIndex; - int numBuckets; - @Getter - long updateTimestamp = System.currentTimeMillis(); - - TabletUpdateRecord(int tabletIndex, int numBuckets) { - this.tabletIndex = tabletIndex; - this.numBuckets = numBuckets; - } - - public synchronized void updateRecord() { - this.tabletIndex = this.tabletIndex + 1 >= numBuckets ? 0 : this.tabletIndex + 1; - // To reduce the compute time cost, only update timestamp when index is 0 - if (this.tabletIndex == 0) { - this.updateTimestamp = System.currentTimeMillis(); - } - } - - } -} diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/TabletLoadIndexRecorderMgr.java b/fe/fe-core/src/main/java/org/apache/doris/planner/TabletLoadIndexRecorderMgr.java new file mode 100644 index 00000000000..6302017551b --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/TabletLoadIndexRecorderMgr.java @@ -0,0 +1,89 @@ +// 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.planner; + +import org.apache.doris.catalog.Partition; +import org.apache.doris.common.UserException; +import org.apache.doris.common.util.MasterDaemon; + +import lombok.Getter; +import org.apache.commons.lang3.tuple.Triple; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.concurrent.ConcurrentHashMap; + +public class TabletLoadIndexRecorderMgr extends MasterDaemon { + private static final Logger LOG = LogManager.getLogger(TabletLoadIndexRecorderMgr.class); + private static final long TABLET_LOAD_INDEX_KEEP_MAX_TIME_MS = 86400000; // 1 * 24 * 60 * 60 * 1000, 1 days + private static final long TABLET_LOAD_INDEX_EXPIRE_CHECK_INTERVAL_MS = 3600000; // 1 hour + private static final int TIMES_FOR_UPDATE_TIMESTAMP = 1000; + + // <<db_id, table_id, partition_id> -> load_tablet_record> + // 0 =< load_tablet_index < number_buckets + private final ConcurrentHashMap<Triple<Long, Long, Long>, TabletLoadIndexRecord> loadTabletRecordMap = + new ConcurrentHashMap<>(); + + public TabletLoadIndexRecorderMgr() { + super("tablet_load_index_recorder", TABLET_LOAD_INDEX_EXPIRE_CHECK_INTERVAL_MS); + } + + @Override + protected void runAfterCatalogReady() { + int originRecordSize = loadTabletRecordMap.size(); + long expireTime = System.currentTimeMillis() - TABLET_LOAD_INDEX_KEEP_MAX_TIME_MS; + loadTabletRecordMap.entrySet().removeIf(entry -> + entry.getValue().getUpdateTimestamp() < expireTime + ); + int currentRecordSize = loadTabletRecordMap.size(); + LOG.info("Remove expired load tablet index record successfully, before {}, current {}", + originRecordSize, currentRecordSize); + } + + public int getCurrentTabletLoadIndex(long dbId, long tableId, Partition partition) throws UserException { + Triple<Long, Long, Long> key = Triple.of(dbId, tableId, partition.getId()); + return loadTabletRecordMap.compute(key, (k, existingRecord) -> + existingRecord == null ? new TabletLoadIndexRecord(partition.getVisibleVersion() - 1, + partition.getDistributionInfo().getBucketNum()) : existingRecord).getAndIncrement(); + } + + static class TabletLoadIndexRecord { + int loadIndex; + int numBuckets; + @Getter + long updateTimestamp = System.currentTimeMillis(); + + TabletLoadIndexRecord(long initialIndex, int numBuckets) { + this.loadIndex = (int) (initialIndex % numBuckets); + this.numBuckets = numBuckets; + } + + public synchronized int getAndIncrement() { + int tabletLoadIndex = loadIndex % numBuckets; + loadIndex++; + // To reduce the compute time cost, only update timestamp when load index is + // larger than TIMES_FOR_UPDATE_TIMESTAMP + if (loadIndex >= TIMES_FOR_UPDATE_TIMESTAMP) { + loadIndex = loadIndex - TIMES_FOR_UPDATE_TIMESTAMP; + this.updateTimestamp = System.currentTimeMillis(); + } + return tabletLoadIndex; + } + + } +} diff --git a/gensrc/thrift/Descriptors.thrift b/gensrc/thrift/Descriptors.thrift index 21bea8cac59..abaf8f8967d 100644 --- a/gensrc/thrift/Descriptors.thrift +++ b/gensrc/thrift/Descriptors.thrift @@ -178,7 +178,7 @@ struct TOlapTablePartition { 9: optional bool is_mutable = true // only used in List Partition 10: optional bool is_default_partition; - // only used in load_to_single_tablet + // only used in random distribution scenario to make data distributed even 11: optional i64 load_tablet_idx } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
