You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "Tanya-W (via GitHub)" <gi...@apache.org> on 2023/02/02 08:32:09 UTC
[GitHub] [doris] Tanya-W opened a new pull request, #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Tanya-W opened a new pull request, #16371:
URL: https://github.com/apache/doris/pull/16371
# Proposed changes
Issue Number: Step5 of [DSIP-023: Add inverted index for full text search](https://cwiki.apache.org/confluence/display/DORIS/DSIP-023%3A+Add+inverted+index+for+full+text+search?src=contextnavpagetreemode)
implementation for add/drop inverted index.
dependency pr: https://github.com/apache/doris/pull/14211 https://github.com/apache/doris/pull/15823 https://github.com/apache/doris/pull/14207 https://github.com/apache/doris/pull/15821
## Problem summary
1. Support create multiple inverted indexes at the same time
2. When execute alter inverted index, only update fe's meta, no need to modified be's meta, read/write base on fe's meta
## Checklist(Required)
1. Does it affect the original behavior:
- [ ] Yes
- [ ] No
- [ ] I don't know
2. Has unit tests been added:
- [ ] Yes
- [ ] No
- [ ] No Need
3. Has document been added or modified:
- [ ] Yes
- [ ] No
- [ ] No Need
4. Does it need to update dependencies:
- [ ] Yes
- [ ] No
5. Are there any changes that cannot be rolled back:
- [ ] Yes (If Yes, please explain WHY)
- [ ] No
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] xiaokang commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "xiaokang (via GitHub)" <gi...@apache.org>.
xiaokang commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1094499716
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -574,6 +574,7 @@ struct TOlapScanNode {
11: optional bool enable_unique_key_merge_on_write
12: optional TPushAggOp push_down_agg_type_opt
13: optional bool use_topn_opt
+ 16: optional list<Descriptors.TOlapTableIndex> indexes_desc
Review Comment:
seq 14
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -1767,24 +1773,54 @@ public int getAsInt() {
return;
}
lightSchemaChange = false;
+
+ CreateIndexClause createIndexClause = (CreateIndexClause) alterClause;
+ if (createIndexClause.getIndexDef().isInvertedIndex()) {
+ alterInvertedIndexes.add(createIndexClause.getIndex());
+ isDropInvertedIndex = false;
+ lightSchemaChangeWithInvertedIndex = true;
+ }
} else if (alterClause instanceof DropIndexClause) {
if (processDropIndex((DropIndexClause) alterClause, olapTable, newIndexes)) {
return;
}
lightSchemaChange = false;
+
+ DropIndexClause dropIndexClause = (DropIndexClause) alterClause;
+ List<Index> existedIndexes = olapTable.getIndexes();
+ Index found = null;
+ for (Index existedIdx : existedIndexes) {
+ if (existedIdx.getIndexName().equalsIgnoreCase(dropIndexClause.getIndexName())) {
+ found = existedIdx;
+ break;
+ }
+ }
+ IndexDef.IndexType indexType = found.getIndexType();
+ if (indexType == IndexType.INVERTED) {
Review Comment:
more generic isIndependent or something else
##########
gensrc/thrift/Types.thrift:
##########
@@ -205,7 +205,8 @@ enum TTaskType {
STORAGE_MEDIUM_MIGRATE_V2,
NOTIFY_UPDATE_STORAGE_POLICY, // deprecated
PUSH_COOLDOWN_CONF,
- PUSH_STORAGE_POLICY
+ PUSH_STORAGE_POLICY,
+ ALTER_INVERTED_INDEX
Review Comment:
It's better to use a generic name for further use by other index
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -1767,24 +1773,54 @@ public int getAsInt() {
return;
}
lightSchemaChange = false;
+
+ CreateIndexClause createIndexClause = (CreateIndexClause) alterClause;
+ if (createIndexClause.getIndexDef().isInvertedIndex()) {
Review Comment:
more generic isIndependent or something else
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -2367,4 +2403,212 @@ public void replayModifyTableAddOrDropColumns(TableAddOrDropColumnsInfo info) th
olapTable.writeUnlock();
}
}
+
+ // the invoker should keep table's write lock
+ public void modifyTableAddOrDropInvertedIndices(Database db, OlapTable olapTable,
+ Map<Long, LinkedList<Column>> indexSchemaMap, Map<String, String> propertyMap,
+ List<Index> indexes, List<Index> alterInvertedIndexes,
+ boolean isDropInvertedIndex, List<Index> oriIndexes,
+ long jobId, boolean isReplay) throws UserException {
+ LOG.info("begin to modify table's meta for add or drop inverted index. table: {}, job: {}",
+ olapTable.getName(), jobId);
+ LOG.info("indexSchemaMap:{}, indexes:{}, alterInvertedIndexes:{}, isDropInvertedIndex: {}",
+ indexSchemaMap, indexes, alterInvertedIndexes, isDropInvertedIndex);
+ if (olapTable.getState() == OlapTableState.ROLLUP) {
Review Comment:
It's covered by the following checkState.
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -2367,4 +2403,212 @@ public void replayModifyTableAddOrDropColumns(TableAddOrDropColumnsInfo info) th
olapTable.writeUnlock();
}
}
+
+ // the invoker should keep table's write lock
+ public void modifyTableAddOrDropInvertedIndices(Database db, OlapTable olapTable,
+ Map<Long, LinkedList<Column>> indexSchemaMap, Map<String, String> propertyMap,
+ List<Index> indexes, List<Index> alterInvertedIndexes,
+ boolean isDropInvertedIndex, List<Index> oriIndexes,
+ long jobId, boolean isReplay) throws UserException {
+ LOG.info("begin to modify table's meta for add or drop inverted index. table: {}, job: {}",
+ olapTable.getName(), jobId);
+ LOG.info("indexSchemaMap:{}, indexes:{}, alterInvertedIndexes:{}, isDropInvertedIndex: {}",
+ indexSchemaMap, indexes, alterInvertedIndexes, isDropInvertedIndex);
+ if (olapTable.getState() == OlapTableState.ROLLUP) {
+ throw new DdlException("Table[" + olapTable.getName() + "]'s is doing ROLLUP job");
+ }
+
+ // for now table's state can only be NORMAL
+ Preconditions.checkState(olapTable.getState() == OlapTableState.NORMAL, olapTable.getState().name());
+
+ boolean hasInvertedIndexChange = false;
+ if (!alterInvertedIndexes.isEmpty()) {
+ hasInvertedIndexChange = true;
+ }
+
+ // begin checking each table
+ // ATTN: DO NOT change any meta in this loop
+ Map<Long, List<Column>> changedIndexIdToSchema = Maps.newHashMap();
+ for (Long alterIndexId : indexSchemaMap.keySet()) {
+ if (!hasInvertedIndexChange) {
Review Comment:
hasInvertedIndexChange is not changed in loop, so should be check before loop
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java:
##########
@@ -337,6 +366,10 @@ protected void runPendingJob() throws AlterCancelException {
}
private void addShadowIndexToCatalog(OlapTable tbl) {
+ if (invertedIndexChange) {
Review Comment:
add comment
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java:
##########
@@ -499,6 +590,13 @@ protected void runRunningJob() throws AlterCancelException {
LOG.info("schema change tasks not finished. job: {}", jobId);
List<AgentTask> tasks = schemaChangeBatchTask.getUnfinishedTasks(2000);
for (AgentTask task : tasks) {
+ if (invertedIndexChange) {
+ // TODO(wy): more elegant implementation
Review Comment:
Is this TODO must be fixed?
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -2367,4 +2403,208 @@ public void replayModifyTableAddOrDropColumns(TableAddOrDropColumnsInfo info) th
olapTable.writeUnlock();
}
}
+
+ // the invoker should keep table's write lock
+ public void modifyTableAddOrDropInvertedIndices(Database db, OlapTable olapTable,
Review Comment:
modifyTableAddOrDropInvertedIndices is almost the same as modifyTableAddOrDropColumns, can we try to merge them to reuse code as much as possible.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -1095,9 +1098,15 @@ protected void toThrift(TPlanNode msg) {
}
}
+ for (Index index : olapTable.getIndexes()) {
+ TOlapTableIndex tIndex = index.toThrift();
+ indexDesc.add(tIndex);
+ }
+
msg.node_type = TPlanNodeType.OLAP_SCAN_NODE;
msg.olap_scan_node = new TOlapScanNode(desc.getId().asInt(), keyColumnNames, keyColumnTypes, isPreAggregation);
msg.olap_scan_node.setColumnsDesc(columnsDesc);
+ msg.olap_scan_node.setIndexesDesc(indexDesc);
Review Comment:
Adding indexDesc to thrift will add load for query rpc, can we limit it to query with inverted index?
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1094198749
##########
be/src/olap/schema_change.cpp:
##########
@@ -586,6 +592,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
Review Comment:
warning: initializer for base class 'doris::SchemaChange' is redundant [readability-redundant-member-init]
```suggestion
: ,
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,42 @@ class VSchemaChangeWithSorting : public SchemaChange {
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
Review Comment:
warning: prefer using 'override' or (rarely) 'final' instead of 'virtual' [modernize-use-override]
```suggestion
~SchemaChangeForInvertedIndex() override;
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -586,6 +592,248 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:741:
```diff
- else
+ } else
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -586,6 +592,248 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
else {
break;
}
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,42 @@
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
Review Comment:
warning: 'virtual' is redundant since the function is already declared 'override' [modernize-use-override]
```suggestion
Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,42 @@
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet, TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) override;
+
+private:
+ DISALLOW_COPY_AND_ASSIGN(SchemaChangeForInvertedIndex);
+ Status _write_inverted_index(int32_t segment_idx, vectorized::Block* block);
+ Status _add_data(const std::string& column_name, const std::string& index_writer_sign,
+ Field* field, const uint8_t** ptr, size_t num_rows);
+ Status _add_nullable(const std::string& column_name, const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr,
+ size_t num_rows);
+
+private:
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/olap/schema_change.h:189:** previously declared here
```cpp
private:
^
```
##########
be/src/olap/tablet_schema.cpp:
##########
@@ -492,6 +492,33 @@ void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
}
}
+void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
+ const std::vector<int32_t>& column_uids) {
+ _index_id = index.index_id;
+ _index_name = index.index_name;
+ _col_unique_ids = std::move(column_uids);
Review Comment:
warning: std::move of the const variable 'column_uids' has no effect; remove std::move() or make the variable non-const [performance-move-const-arg]
```suggestion
_col_unique_ids = column_uids;
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098514695
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +582,236 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {
+ _olap_data_convertor = std::make_unique<vectorized::OlapBlockDataConvertor>();
+}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+ _olap_data_convertor.reset();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ _olap_data_convertor->reserve(_alter_inverted_indexs.size());
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto index_id = inverted_index.index_id;
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto column_idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(column_idx);
+ _olap_data_convertor->add_column_data_convertor(column);
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ auto writer_sign = std::make_pair(i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ // load segments
+ SegmentCacheHandle segment_cache_handle;
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), block.get()) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ block->clear_column_data();
+ } while (true);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ auto writer_sign = std::make_pair(i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(
+ const std::string& column_name, const std::pair<int64_t, int64_t>& index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:728:
```diff
- else
+ } else
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +178,45 @@ class VSchemaChangeWithSorting : public SchemaChange {
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet, TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) override;
+
+private:
+ DISALLOW_COPY_AND_ASSIGN(SchemaChangeForInvertedIndex);
+ Status _write_inverted_index(int32_t segment_idx, vectorized::Block* block);
+ Status _add_data(const std::string& column_name,
+ const std::pair<int64_t, int64_t>& index_writer_sign, Field* field,
+ const uint8_t** ptr, size_t num_rows);
+ Status _add_nullable(const std::string& column_name,
+ const std::pair<int64_t, int64_t>& index_writer_sign, Field* field,
+ const uint8_t* null_map, const uint8_t** ptr, size_t num_rows);
+
+private:
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/olap/schema_change.h:190:** previously declared here
```cpp
private:
^
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +582,236 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {
+ _olap_data_convertor = std::make_unique<vectorized::OlapBlockDataConvertor>();
+}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+ _olap_data_convertor.reset();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ _olap_data_convertor->reserve(_alter_inverted_indexs.size());
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto index_id = inverted_index.index_id;
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto column_idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(column_idx);
+ _olap_data_convertor->add_column_data_convertor(column);
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ auto writer_sign = std::make_pair(i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ // load segments
+ SegmentCacheHandle segment_cache_handle;
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), block.get()) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ block->clear_column_data();
+ } while (true);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ auto writer_sign = std::make_pair(i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(
+ const std::string& column_name, const std::pair<int64_t, int64_t>& index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
+ }
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
else {
break;
}
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098516172
##########
be/src/olap/schema_change.cpp:
##########
@@ -958,6 +1232,270 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
return res;
}
+Status SchemaChangeHandler::_do_process_alter_inverted_index(
+ TabletSharedPtr tablet, const TAlterInvertedIndexReq& request) {
+ Status res = Status::OK();
+ // TODO(wy): check whether the tablet's max continuous version == request.version
+ if (tablet->tablet_state() == TABLET_TOMBSTONED || tablet->tablet_state() == TABLET_STOPPED ||
+ tablet->tablet_state() == TABLET_SHUTDOWN) {
+ LOG(WARNING) << "tablet's state=" << tablet->tablet_state()
+ << " cannot alter inverted index";
+ return Status::Error<ErrorCode::INTERNAL_ERROR>();
+ }
+
+ std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
+ if (!base_migration_rlock.owns_lock()) {
+ return Status::Error<TRY_LOCK_FAILED>();
+ }
+
+ TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->copy_from(*tablet->tablet_schema());
+ if (!request.columns.empty() && request.columns[0].col_unique_id >= 0) {
+ tablet_schema->clear_columns();
+ for (const auto& column : request.columns) {
+ tablet_schema->append_column(TabletColumn(column));
+ }
+ }
+
+ // get rowset reader
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ DeleteHandler delete_handler;
+ RETURN_IF_ERROR(
+ _get_rowset_readers(tablet, tablet_schema, request, &rs_readers, &delete_handler));
+ if (request.__isset.is_drop_op && request.is_drop_op) {
+ // drop index
+ res = _drop_inverted_index(rs_readers, tablet_schema, tablet, request);
+ } else {
+ // add index
+ res = _add_inverted_index(rs_readers, &delete_handler, tablet_schema, tablet, request);
+ }
+
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to alter tablet. tablet=" << tablet->full_name();
+ return res;
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeHandler::_get_rowset_readers(TabletSharedPtr tablet,
+ const TabletSchemaSPtr& tablet_schema,
+ const TAlterInvertedIndexReq& request,
+ std::vector<RowsetReaderSharedPtr>* rs_readers,
+ DeleteHandler* delete_handler) {
+ Status res = Status::OK();
+ std::vector<Version> versions_to_be_changed;
+ std::vector<ColumnId> return_columns;
+ std::vector<TOlapTableIndex> alter_inverted_indexs;
+
+ if (request.__isset.alter_inverted_indexes) {
+ alter_inverted_indexs = request.alter_inverted_indexes;
+ }
+
+ for (auto& inverted_index : alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // obtain base tablet's push lock and header write lock to prevent loading data
+ {
+ std::lock_guard<std::mutex> tablet_lock(tablet->get_push_lock());
+ std::lock_guard<std::shared_mutex> tablet_wlock(tablet->get_header_lock());
+
+ do {
Review Comment:
The main function used here is `capture_rs_readers`, other logic is to prepare for rs_reader's init param which maybe set differently according to different use scenarios.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098264968
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -2367,4 +2403,212 @@ public void replayModifyTableAddOrDropColumns(TableAddOrDropColumnsInfo info) th
olapTable.writeUnlock();
}
}
+
+ // the invoker should keep table's write lock
+ public void modifyTableAddOrDropInvertedIndices(Database db, OlapTable olapTable,
+ Map<Long, LinkedList<Column>> indexSchemaMap, Map<String, String> propertyMap,
+ List<Index> indexes, List<Index> alterInvertedIndexes,
+ boolean isDropInvertedIndex, List<Index> oriIndexes,
+ long jobId, boolean isReplay) throws UserException {
+ LOG.info("begin to modify table's meta for add or drop inverted index. table: {}, job: {}",
+ olapTable.getName(), jobId);
+ LOG.info("indexSchemaMap:{}, indexes:{}, alterInvertedIndexes:{}, isDropInvertedIndex: {}",
+ indexSchemaMap, indexes, alterInvertedIndexes, isDropInvertedIndex);
+ if (olapTable.getState() == OlapTableState.ROLLUP) {
+ throw new DdlException("Table[" + olapTable.getName() + "]'s is doing ROLLUP job");
+ }
+
+ // for now table's state can only be NORMAL
+ Preconditions.checkState(olapTable.getState() == OlapTableState.NORMAL, olapTable.getState().name());
+
+ boolean hasInvertedIndexChange = false;
+ if (!alterInvertedIndexes.isEmpty()) {
+ hasInvertedIndexChange = true;
+ }
+
+ // begin checking each table
+ // ATTN: DO NOT change any meta in this loop
+ Map<Long, List<Column>> changedIndexIdToSchema = Maps.newHashMap();
+ for (Long alterIndexId : indexSchemaMap.keySet()) {
+ if (!hasInvertedIndexChange) {
Review Comment:
yes, updated
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1094196033
##########
be/src/olap/schema_change.cpp:
##########
@@ -586,6 +592,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
Review Comment:
warning: initializer for base class 'doris::SchemaChange' is redundant [readability-redundant-member-init]
```suggestion
: , _alter_inverted_indexs(alter_inverted_indexs), _tablet_schema(tablet_schema) {
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -586,6 +592,248 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:736:
```diff
- else
+ } else
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,42 @@ class VSchemaChangeWithSorting : public SchemaChange {
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
Review Comment:
warning: prefer using 'override' or (rarely) 'final' instead of 'virtual' [modernize-use-override]
```suggestion
~SchemaChangeForInvertedIndex() override;
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,42 @@
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet, TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) override;
+
+private:
+ DISALLOW_COPY_AND_ASSIGN(SchemaChangeForInvertedIndex);
+ Status _write_inverted_index(int32_t segment_idx, vectorized::Block* block);
+ Status _add_data(const std::string& column_name, const std::string& index_writer_sign,
+ Field* field, const uint8_t** ptr, size_t num_rows);
+ Status _add_nullable(const std::string& column_name, const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr,
+ size_t num_rows);
+
+private:
+ std::vector<TOlapTableIndex> _alter_inverted_indexs;
+ TabletSchemaSPtr _tablet_schema;
+
+ // "segment_id_unique_id" -> InvertedIndexColumnWriter
+ std::unordered_map<std::string, std::unique_ptr<segment_v2::InvertedIndexColumnWriter>>
+ _inverted_index_builders;
+ std::vector<std::unique_ptr<TabletIndex>> _index_metas;
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/olap/schema_change.h:189:** previously declared here
```cpp
private:
^
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -586,6 +592,248 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
else {
break;
}
```
##########
be/src/olap/task/engine_alter_tablet_task.h:
##########
@@ -39,4 +39,18 @@
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
}; // EngineTask
+class EngineAlterInvertedIndexTask : public EngineTask {
+public:
+ virtual Status execute();
+
+public:
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/olap/task/engine_alter_tablet_task.h:42:** previously declared here
```cpp
public:
^
```
##########
be/src/olap/tablet_schema.cpp:
##########
@@ -492,6 +492,33 @@ void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
}
}
+void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
+ const std::vector<int32_t>& column_uids) {
+ _index_id = index.index_id;
+ _index_name = index.index_name;
Review Comment:
warning: std::move of the const variable 'column_uids' has no effect; remove std::move() or make the variable non-const [performance-move-const-arg]
```suggestion
_col_unique_ids = column_uids;
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,42 @@
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet, TabletSharedPtr base_tablet,
Review Comment:
warning: 'virtual' is redundant since the function is already declared 'override' [modernize-use-override]
```suggestion
Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer,
```
##########
be/src/olap/task/engine_alter_tablet_task.h:
##########
@@ -39,4 +39,18 @@ class EngineAlterTabletTask : public EngineTask {
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
}; // EngineTask
+class EngineAlterInvertedIndexTask : public EngineTask {
+public:
+ virtual Status execute();
Review Comment:
warning: prefer using 'override' or (rarely) 'final' instead of 'virtual' [modernize-use-override]
```suggestion
Status execute() override;
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098265163
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -574,6 +574,7 @@ struct TOlapScanNode {
11: optional bool enable_unique_key_merge_on_write
12: optional TPushAggOp push_down_agg_type_opt
13: optional bool use_topn_opt
+ 16: optional list<Descriptors.TOlapTableIndex> indexes_desc
Review Comment:
updated
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098574523
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java:
##########
@@ -499,6 +590,13 @@ protected void runRunningJob() throws AlterCancelException {
LOG.info("schema change tasks not finished. job: {}", jobId);
List<AgentTask> tasks = schemaChangeBatchTask.getUnfinishedTasks(2000);
for (AgentTask task : tasks) {
+ if (invertedIndexChange) {
+ // TODO(wy): more elegant implementation
Review Comment:
yes, need to do, but not in this pr, "more elegant implementation" need implement specify partition to rebuild index data
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] hello-stephen commented on pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #16371:
URL: https://github.com/apache/doris/pull/16371#issuecomment-1413808938
TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 34.59 seconds
load time: 505 seconds
storage size: 17171308895 Bytes
https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/tmp/20230202141208_clickbench_pr_89574.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098264089
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
Review Comment:
the loops to get `return_columns` can be merged, but here two loops is needed, because every segment need to change all inverted index in `_alter_inverted_indexs`
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring merged pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring merged PR #16371:
URL: https://github.com/apache/doris/pull/16371
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1095440190
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:732:
```diff
- else
+ } else
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098397282
##########
be/src/olap/schema_change.cpp:
##########
@@ -958,6 +1232,270 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
return res;
}
+Status SchemaChangeHandler::_do_process_alter_inverted_index(
+ TabletSharedPtr tablet, const TAlterInvertedIndexReq& request) {
+ Status res = Status::OK();
+ // TODO(wy): check whether the tablet's max continuous version == request.version
+ if (tablet->tablet_state() == TABLET_TOMBSTONED || tablet->tablet_state() == TABLET_STOPPED ||
+ tablet->tablet_state() == TABLET_SHUTDOWN) {
+ LOG(WARNING) << "tablet's state=" << tablet->tablet_state()
+ << " cannot alter inverted index";
+ return Status::Error<ErrorCode::INTERNAL_ERROR>();
+ }
+
+ std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
+ if (!base_migration_rlock.owns_lock()) {
+ return Status::Error<TRY_LOCK_FAILED>();
+ }
+
+ TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->copy_from(*tablet->tablet_schema());
+ if (!request.columns.empty() && request.columns[0].col_unique_id >= 0) {
+ tablet_schema->clear_columns();
+ for (const auto& column : request.columns) {
+ tablet_schema->append_column(TabletColumn(column));
+ }
+ }
+
+ // get rowset reader
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ DeleteHandler delete_handler;
+ RETURN_IF_ERROR(
+ _get_rowset_readers(tablet, tablet_schema, request, &rs_readers, &delete_handler));
+ if (request.__isset.is_drop_op && request.is_drop_op) {
+ // drop index
+ res = _drop_inverted_index(rs_readers, tablet_schema, tablet, request);
+ } else {
+ // add index
+ res = _add_inverted_index(rs_readers, &delete_handler, tablet_schema, tablet, request);
+ }
+
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to alter tablet. tablet=" << tablet->full_name();
+ return res;
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeHandler::_get_rowset_readers(TabletSharedPtr tablet,
+ const TabletSchemaSPtr& tablet_schema,
+ const TAlterInvertedIndexReq& request,
+ std::vector<RowsetReaderSharedPtr>* rs_readers,
+ DeleteHandler* delete_handler) {
+ Status res = Status::OK();
+ std::vector<Version> versions_to_be_changed;
+ std::vector<ColumnId> return_columns;
+ std::vector<TOlapTableIndex> alter_inverted_indexs;
+
+ if (request.__isset.alter_inverted_indexes) {
+ alter_inverted_indexs = request.alter_inverted_indexes;
+ }
+
+ for (auto& inverted_index : alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // obtain base tablet's push lock and header write lock to prevent loading data
+ {
+ std::lock_guard<std::mutex> tablet_lock(tablet->get_push_lock());
+ std::lock_guard<std::shared_mutex> tablet_wlock(tablet->get_header_lock());
+
+ do {
+ RowsetSharedPtr max_rowset;
+ // get history data to rebuild inverted index and it will check if there is hold in base tablet
+ res = _get_versions_to_be_changed(tablet, &versions_to_be_changed, &max_rowset);
+ if (!res.ok()) {
+ LOG(WARNING) << "fail to get version to be rebuild inverted index. res=" << res;
+ break;
+ }
+
+ // should check the max_version >= request.alter_version, if not the rebuild index is useless
+ if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) {
+ LOG(WARNING) << "base tablet's max version="
+ << (max_rowset == nullptr ? 0 : max_rowset->end_version())
+ << " is less than request version=" << request.alter_version;
+ res = Status::InternalError(
+ "base tablet's max version={} is less than request version={}",
+ (max_rowset == nullptr ? 0 : max_rowset->end_version()),
+ request.alter_version);
+ break;
+ }
+
+ // init one delete handler
+ int64_t end_version = -1;
+ for (auto& version : versions_to_be_changed) {
+ end_version = std::max(end_version, version.second);
+ }
+
+ auto& all_del_preds = tablet->delete_predicates();
+ for (auto& delete_pred : all_del_preds) {
+ if (delete_pred->version().first > end_version) {
+ continue;
+ }
+ tablet_schema->merge_dropped_columns(tablet->tablet_schema(delete_pred->version()));
+ }
+ res = delete_handler->init(tablet_schema, all_del_preds, end_version);
+ if (!res) {
+ LOG(WARNING) << "init delete handler failed. tablet=" << tablet->full_name()
+ << ", end_version=" << end_version;
+ break;
+ }
+
+ // acquire data sources correspond to history versions
+ tablet->capture_rs_readers(versions_to_be_changed, rs_readers);
+ if (rs_readers->size() < 1) {
+ LOG(WARNING) << "fail to acquire all data sources. "
+ << "version_num=" << versions_to_be_changed.size()
+ << ", data_source_num=" << rs_readers->size();
+ res = Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>();
+ break;
+ }
+
+ // reader_context is stack variables, it's lifetime should keep the same with rs_readers
+ RowsetReaderContext reader_context;
+ reader_context.reader_type = READER_ALTER_TABLE;
+ reader_context.tablet_schema = tablet_schema;
+ reader_context.need_ordered_result = true;
+ reader_context.delete_handler = delete_handler;
Review Comment:
I will add more regression test case
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098318886
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
+ }
+ return step;
+ };
+
+ try {
+ do {
+ auto step = next_run_step();
+ if (null_map[offset]) {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_nulls(step));
+ *ptr += field->size() * step;
+ } else {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, step));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, step));
+ }
+ }
+ offset += step;
+ } while (offset < num_rows);
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_add_data(const std::string& column_name,
+ const std::string& index_writer_sign, Field* field,
+ const uint8_t** ptr, size_t num_rows) {
+ try {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, num_rows));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, num_rows));
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_write_inverted_index(int32_t segment_idx,
+ vectorized::Block* block) {
+ LOG(INFO) << "begin to write inverted index";
Review Comment:
updated
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098490345
##########
be/src/olap/schema_change.cpp:
##########
@@ -958,6 +1232,270 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
return res;
}
+Status SchemaChangeHandler::_do_process_alter_inverted_index(
+ TabletSharedPtr tablet, const TAlterInvertedIndexReq& request) {
+ Status res = Status::OK();
+ // TODO(wy): check whether the tablet's max continuous version == request.version
+ if (tablet->tablet_state() == TABLET_TOMBSTONED || tablet->tablet_state() == TABLET_STOPPED ||
+ tablet->tablet_state() == TABLET_SHUTDOWN) {
+ LOG(WARNING) << "tablet's state=" << tablet->tablet_state()
+ << " cannot alter inverted index";
+ return Status::Error<ErrorCode::INTERNAL_ERROR>();
+ }
+
+ std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
+ if (!base_migration_rlock.owns_lock()) {
+ return Status::Error<TRY_LOCK_FAILED>();
+ }
+
+ TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->copy_from(*tablet->tablet_schema());
+ if (!request.columns.empty() && request.columns[0].col_unique_id >= 0) {
+ tablet_schema->clear_columns();
+ for (const auto& column : request.columns) {
+ tablet_schema->append_column(TabletColumn(column));
+ }
+ }
+
+ // get rowset reader
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ DeleteHandler delete_handler;
+ RETURN_IF_ERROR(
+ _get_rowset_readers(tablet, tablet_schema, request, &rs_readers, &delete_handler));
+ if (request.__isset.is_drop_op && request.is_drop_op) {
+ // drop index
+ res = _drop_inverted_index(rs_readers, tablet_schema, tablet, request);
+ } else {
+ // add index
+ res = _add_inverted_index(rs_readers, &delete_handler, tablet_schema, tablet, request);
+ }
+
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to alter tablet. tablet=" << tablet->full_name();
+ return res;
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeHandler::_get_rowset_readers(TabletSharedPtr tablet,
+ const TabletSchemaSPtr& tablet_schema,
+ const TAlterInvertedIndexReq& request,
+ std::vector<RowsetReaderSharedPtr>* rs_readers,
+ DeleteHandler* delete_handler) {
+ Status res = Status::OK();
+ std::vector<Version> versions_to_be_changed;
+ std::vector<ColumnId> return_columns;
+ std::vector<TOlapTableIndex> alter_inverted_indexs;
+
+ if (request.__isset.alter_inverted_indexes) {
+ alter_inverted_indexs = request.alter_inverted_indexes;
+ }
+
+ for (auto& inverted_index : alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // obtain base tablet's push lock and header write lock to prevent loading data
+ {
+ std::lock_guard<std::mutex> tablet_lock(tablet->get_push_lock());
+ std::lock_guard<std::shared_mutex> tablet_wlock(tablet->get_header_lock());
+
+ do {
+ RowsetSharedPtr max_rowset;
+ // get history data to rebuild inverted index and it will check if there is hold in base tablet
+ res = _get_versions_to_be_changed(tablet, &versions_to_be_changed, &max_rowset);
+ if (!res.ok()) {
+ LOG(WARNING) << "fail to get version to be rebuild inverted index. res=" << res;
+ break;
+ }
+
+ // should check the max_version >= request.alter_version, if not the rebuild index is useless
+ if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) {
+ LOG(WARNING) << "base tablet's max version="
+ << (max_rowset == nullptr ? 0 : max_rowset->end_version())
+ << " is less than request version=" << request.alter_version;
+ res = Status::InternalError(
+ "base tablet's max version={} is less than request version={}",
+ (max_rowset == nullptr ? 0 : max_rowset->end_version()),
+ request.alter_version);
+ break;
+ }
+
+ // init one delete handler
+ int64_t end_version = -1;
+ for (auto& version : versions_to_be_changed) {
+ end_version = std::max(end_version, version.second);
+ }
+
+ auto& all_del_preds = tablet->delete_predicates();
+ for (auto& delete_pred : all_del_preds) {
+ if (delete_pred->version().first > end_version) {
+ continue;
+ }
+ tablet_schema->merge_dropped_columns(tablet->tablet_schema(delete_pred->version()));
+ }
+ res = delete_handler->init(tablet_schema, all_del_preds, end_version);
+ if (!res) {
+ LOG(WARNING) << "init delete handler failed. tablet=" << tablet->full_name()
+ << ", end_version=" << end_version;
+ break;
+ }
+
+ // acquire data sources correspond to history versions
+ tablet->capture_rs_readers(versions_to_be_changed, rs_readers);
+ if (rs_readers->size() < 1) {
+ LOG(WARNING) << "fail to acquire all data sources. "
+ << "version_num=" << versions_to_be_changed.size()
+ << ", data_source_num=" << rs_readers->size();
+ res = Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>();
+ break;
+ }
+
+ // reader_context is stack variables, it's lifetime should keep the same with rs_readers
+ RowsetReaderContext reader_context;
+ reader_context.reader_type = READER_ALTER_TABLE;
+ reader_context.tablet_schema = tablet_schema;
+ reader_context.need_ordered_result = true;
Review Comment:
need_ordered_result = false is right
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098741343
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +582,232 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {
+ _olap_data_convertor = std::make_unique<vectorized::OlapBlockDataConvertor>();
+}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+ _olap_data_convertor.reset();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ _olap_data_convertor->reserve(_alter_inverted_indexs.size());
+
+ // load segments
+ SegmentCacheHandle segment_cache_handle;
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), seg_ptr->id());
+ std::vector<ColumnId> return_columns;
+ std::vector<std::pair<int64_t, int64_t>> inverted_index_writer_signs;
+ // create inverted index writer
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto index_id = inverted_index.index_id;
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto column_idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(column_idx);
+ _olap_data_convertor->add_column_data_convertor(column);
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ auto writer_sign = std::make_pair(seg_ptr->id(), index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ inverted_index_writer_signs.push_back(writer_sign);
+ }
+ }
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), block.get()) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ block->clear_column_data();
+ } while (true);
+
+ // finish write inverted index, flush data to compound file
+ for (auto& writer_sign : inverted_index_writer_signs) {
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(
+ const std::string& column_name, const std::pair<int64_t, int64_t>& index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:723:
```diff
- else
+ } else
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098731824
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java:
##########
@@ -2367,4 +2403,208 @@ public void replayModifyTableAddOrDropColumns(TableAddOrDropColumnsInfo info) th
olapTable.writeUnlock();
}
}
+
+ // the invoker should keep table's write lock
+ public void modifyTableAddOrDropInvertedIndices(Database db, OlapTable olapTable,
Review Comment:
yes, the same code will be encapsulated into functions
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] xiaokang commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "xiaokang (via GitHub)" <gi...@apache.org>.
xiaokang commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1095475224
##########
be/src/olap/schema_change.cpp:
##########
@@ -958,6 +1232,270 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
return res;
}
+Status SchemaChangeHandler::_do_process_alter_inverted_index(
+ TabletSharedPtr tablet, const TAlterInvertedIndexReq& request) {
+ Status res = Status::OK();
+ // TODO(wy): check whether the tablet's max continuous version == request.version
+ if (tablet->tablet_state() == TABLET_TOMBSTONED || tablet->tablet_state() == TABLET_STOPPED ||
+ tablet->tablet_state() == TABLET_SHUTDOWN) {
+ LOG(WARNING) << "tablet's state=" << tablet->tablet_state()
+ << " cannot alter inverted index";
+ return Status::Error<ErrorCode::INTERNAL_ERROR>();
+ }
+
+ std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
+ if (!base_migration_rlock.owns_lock()) {
+ return Status::Error<TRY_LOCK_FAILED>();
+ }
+
+ TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->copy_from(*tablet->tablet_schema());
+ if (!request.columns.empty() && request.columns[0].col_unique_id >= 0) {
+ tablet_schema->clear_columns();
+ for (const auto& column : request.columns) {
+ tablet_schema->append_column(TabletColumn(column));
+ }
+ }
+
+ // get rowset reader
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ DeleteHandler delete_handler;
+ RETURN_IF_ERROR(
+ _get_rowset_readers(tablet, tablet_schema, request, &rs_readers, &delete_handler));
+ if (request.__isset.is_drop_op && request.is_drop_op) {
+ // drop index
+ res = _drop_inverted_index(rs_readers, tablet_schema, tablet, request);
+ } else {
+ // add index
+ res = _add_inverted_index(rs_readers, &delete_handler, tablet_schema, tablet, request);
+ }
+
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to alter tablet. tablet=" << tablet->full_name();
+ return res;
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeHandler::_get_rowset_readers(TabletSharedPtr tablet,
+ const TabletSchemaSPtr& tablet_schema,
+ const TAlterInvertedIndexReq& request,
+ std::vector<RowsetReaderSharedPtr>* rs_readers,
+ DeleteHandler* delete_handler) {
+ Status res = Status::OK();
+ std::vector<Version> versions_to_be_changed;
+ std::vector<ColumnId> return_columns;
+ std::vector<TOlapTableIndex> alter_inverted_indexs;
+
+ if (request.__isset.alter_inverted_indexes) {
+ alter_inverted_indexs = request.alter_inverted_indexes;
+ }
+
+ for (auto& inverted_index : alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // obtain base tablet's push lock and header write lock to prevent loading data
+ {
+ std::lock_guard<std::mutex> tablet_lock(tablet->get_push_lock());
+ std::lock_guard<std::shared_mutex> tablet_wlock(tablet->get_header_lock());
+
+ do {
Review Comment:
is there any function can be reused here to create rs_readers?
##########
be/src/olap/task/engine_alter_tablet_task.cpp:
##########
@@ -45,4 +45,34 @@ Status EngineAlterTabletTask::execute() {
return res;
} // execute
+EngineAlterInvertedIndexTask::EngineAlterInvertedIndexTask(
+ const TAlterInvertedIndexReq& alter_inverted_index_request)
+ : _alter_inverted_index_req(alter_inverted_index_request) {
+ _mem_tracker = std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::SCHEMA_CHANGE,
+ fmt::format("EngineAlterInvertedIndexTask#tabletId={}",
+ std::to_string(_alter_inverted_index_req.tablet_id)),
+ config::memory_limitation_per_thread_for_schema_change_bytes);
+}
+
+Status EngineAlterInvertedIndexTask::execute() {
+ SCOPED_ATTACH_TASK(_mem_tracker);
+ DorisMetrics::instance()->create_rollup_requests_total->increment(1);
Review Comment:
metric name is not suitable
##########
be/src/olap/schema_change.cpp:
##########
@@ -958,6 +1232,270 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
return res;
}
+Status SchemaChangeHandler::_do_process_alter_inverted_index(
+ TabletSharedPtr tablet, const TAlterInvertedIndexReq& request) {
+ Status res = Status::OK();
+ // TODO(wy): check whether the tablet's max continuous version == request.version
+ if (tablet->tablet_state() == TABLET_TOMBSTONED || tablet->tablet_state() == TABLET_STOPPED ||
+ tablet->tablet_state() == TABLET_SHUTDOWN) {
+ LOG(WARNING) << "tablet's state=" << tablet->tablet_state()
+ << " cannot alter inverted index";
+ return Status::Error<ErrorCode::INTERNAL_ERROR>();
+ }
+
+ std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
+ if (!base_migration_rlock.owns_lock()) {
+ return Status::Error<TRY_LOCK_FAILED>();
+ }
+
+ TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->copy_from(*tablet->tablet_schema());
+ if (!request.columns.empty() && request.columns[0].col_unique_id >= 0) {
+ tablet_schema->clear_columns();
+ for (const auto& column : request.columns) {
+ tablet_schema->append_column(TabletColumn(column));
+ }
+ }
+
+ // get rowset reader
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ DeleteHandler delete_handler;
+ RETURN_IF_ERROR(
+ _get_rowset_readers(tablet, tablet_schema, request, &rs_readers, &delete_handler));
+ if (request.__isset.is_drop_op && request.is_drop_op) {
+ // drop index
+ res = _drop_inverted_index(rs_readers, tablet_schema, tablet, request);
+ } else {
+ // add index
+ res = _add_inverted_index(rs_readers, &delete_handler, tablet_schema, tablet, request);
+ }
+
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to alter tablet. tablet=" << tablet->full_name();
+ return res;
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeHandler::_get_rowset_readers(TabletSharedPtr tablet,
+ const TabletSchemaSPtr& tablet_schema,
+ const TAlterInvertedIndexReq& request,
+ std::vector<RowsetReaderSharedPtr>* rs_readers,
+ DeleteHandler* delete_handler) {
+ Status res = Status::OK();
+ std::vector<Version> versions_to_be_changed;
+ std::vector<ColumnId> return_columns;
+ std::vector<TOlapTableIndex> alter_inverted_indexs;
+
+ if (request.__isset.alter_inverted_indexes) {
+ alter_inverted_indexs = request.alter_inverted_indexes;
+ }
+
+ for (auto& inverted_index : alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // obtain base tablet's push lock and header write lock to prevent loading data
+ {
+ std::lock_guard<std::mutex> tablet_lock(tablet->get_push_lock());
+ std::lock_guard<std::shared_mutex> tablet_wlock(tablet->get_header_lock());
+
+ do {
+ RowsetSharedPtr max_rowset;
+ // get history data to rebuild inverted index and it will check if there is hold in base tablet
+ res = _get_versions_to_be_changed(tablet, &versions_to_be_changed, &max_rowset);
+ if (!res.ok()) {
+ LOG(WARNING) << "fail to get version to be rebuild inverted index. res=" << res;
+ break;
+ }
+
+ // should check the max_version >= request.alter_version, if not the rebuild index is useless
+ if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) {
+ LOG(WARNING) << "base tablet's max version="
+ << (max_rowset == nullptr ? 0 : max_rowset->end_version())
+ << " is less than request version=" << request.alter_version;
+ res = Status::InternalError(
+ "base tablet's max version={} is less than request version={}",
+ (max_rowset == nullptr ? 0 : max_rowset->end_version()),
+ request.alter_version);
+ break;
+ }
+
+ // init one delete handler
+ int64_t end_version = -1;
+ for (auto& version : versions_to_be_changed) {
+ end_version = std::max(end_version, version.second);
+ }
+
+ auto& all_del_preds = tablet->delete_predicates();
+ for (auto& delete_pred : all_del_preds) {
+ if (delete_pred->version().first > end_version) {
+ continue;
+ }
+ tablet_schema->merge_dropped_columns(tablet->tablet_schema(delete_pred->version()));
+ }
+ res = delete_handler->init(tablet_schema, all_del_preds, end_version);
+ if (!res) {
+ LOG(WARNING) << "init delete handler failed. tablet=" << tablet->full_name()
+ << ", end_version=" << end_version;
+ break;
+ }
+
+ // acquire data sources correspond to history versions
+ tablet->capture_rs_readers(versions_to_be_changed, rs_readers);
+ if (rs_readers->size() < 1) {
+ LOG(WARNING) << "fail to acquire all data sources. "
+ << "version_num=" << versions_to_be_changed.size()
+ << ", data_source_num=" << rs_readers->size();
+ res = Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>();
+ break;
+ }
+
+ // reader_context is stack variables, it's lifetime should keep the same with rs_readers
+ RowsetReaderContext reader_context;
+ reader_context.reader_type = READER_ALTER_TABLE;
+ reader_context.tablet_schema = tablet_schema;
+ reader_context.need_ordered_result = true;
Review Comment:
index should keep consistent with segment data file row by row. So is it right to set need_ordered_result = true?
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
Review Comment:
The three loops over segments can be merged to one. And there will be less opened _inverted_index_builders.
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
+ }
+ return step;
+ };
+
+ try {
+ do {
+ auto step = next_run_step();
+ if (null_map[offset]) {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_nulls(step));
+ *ptr += field->size() * step;
+ } else {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, step));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, step));
+ }
+ }
+ offset += step;
+ } while (offset < num_rows);
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_add_data(const std::string& column_name,
+ const std::string& index_writer_sign, Field* field,
+ const uint8_t** ptr, size_t num_rows) {
+ try {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, num_rows));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, num_rows));
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_write_inverted_index(int32_t segment_idx,
+ vectorized::Block* block) {
+ LOG(INFO) << "begin to write inverted index";
Review Comment:
verbose debug log
##########
be/src/vec/olap/olap_data_convertor.h:
##########
@@ -46,13 +46,16 @@ class IOlapColumnDataAccessor {
class OlapBlockDataConvertor {
public:
+ OlapBlockDataConvertor() = default;
Review Comment:
why not use OlapBlockDataConvertor(const TabletSchema* tablet_schema)?
##########
be/src/olap/schema_change.cpp:
##########
@@ -958,6 +1232,270 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
return res;
}
+Status SchemaChangeHandler::_do_process_alter_inverted_index(
+ TabletSharedPtr tablet, const TAlterInvertedIndexReq& request) {
+ Status res = Status::OK();
+ // TODO(wy): check whether the tablet's max continuous version == request.version
+ if (tablet->tablet_state() == TABLET_TOMBSTONED || tablet->tablet_state() == TABLET_STOPPED ||
+ tablet->tablet_state() == TABLET_SHUTDOWN) {
+ LOG(WARNING) << "tablet's state=" << tablet->tablet_state()
+ << " cannot alter inverted index";
+ return Status::Error<ErrorCode::INTERNAL_ERROR>();
+ }
+
+ std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
+ if (!base_migration_rlock.owns_lock()) {
+ return Status::Error<TRY_LOCK_FAILED>();
+ }
+
+ TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->copy_from(*tablet->tablet_schema());
+ if (!request.columns.empty() && request.columns[0].col_unique_id >= 0) {
+ tablet_schema->clear_columns();
+ for (const auto& column : request.columns) {
+ tablet_schema->append_column(TabletColumn(column));
+ }
+ }
+
+ // get rowset reader
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ DeleteHandler delete_handler;
+ RETURN_IF_ERROR(
+ _get_rowset_readers(tablet, tablet_schema, request, &rs_readers, &delete_handler));
+ if (request.__isset.is_drop_op && request.is_drop_op) {
+ // drop index
+ res = _drop_inverted_index(rs_readers, tablet_schema, tablet, request);
+ } else {
+ // add index
+ res = _add_inverted_index(rs_readers, &delete_handler, tablet_schema, tablet, request);
+ }
+
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to alter tablet. tablet=" << tablet->full_name();
+ return res;
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeHandler::_get_rowset_readers(TabletSharedPtr tablet,
+ const TabletSchemaSPtr& tablet_schema,
+ const TAlterInvertedIndexReq& request,
+ std::vector<RowsetReaderSharedPtr>* rs_readers,
+ DeleteHandler* delete_handler) {
+ Status res = Status::OK();
+ std::vector<Version> versions_to_be_changed;
+ std::vector<ColumnId> return_columns;
+ std::vector<TOlapTableIndex> alter_inverted_indexs;
+
+ if (request.__isset.alter_inverted_indexes) {
+ alter_inverted_indexs = request.alter_inverted_indexes;
+ }
+
+ for (auto& inverted_index : alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // obtain base tablet's push lock and header write lock to prevent loading data
+ {
+ std::lock_guard<std::mutex> tablet_lock(tablet->get_push_lock());
+ std::lock_guard<std::shared_mutex> tablet_wlock(tablet->get_header_lock());
+
+ do {
+ RowsetSharedPtr max_rowset;
+ // get history data to rebuild inverted index and it will check if there is hold in base tablet
+ res = _get_versions_to_be_changed(tablet, &versions_to_be_changed, &max_rowset);
+ if (!res.ok()) {
+ LOG(WARNING) << "fail to get version to be rebuild inverted index. res=" << res;
+ break;
+ }
+
+ // should check the max_version >= request.alter_version, if not the rebuild index is useless
+ if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) {
+ LOG(WARNING) << "base tablet's max version="
+ << (max_rowset == nullptr ? 0 : max_rowset->end_version())
+ << " is less than request version=" << request.alter_version;
+ res = Status::InternalError(
+ "base tablet's max version={} is less than request version={}",
+ (max_rowset == nullptr ? 0 : max_rowset->end_version()),
+ request.alter_version);
+ break;
+ }
+
+ // init one delete handler
+ int64_t end_version = -1;
+ for (auto& version : versions_to_be_changed) {
+ end_version = std::max(end_version, version.second);
+ }
+
+ auto& all_del_preds = tablet->delete_predicates();
+ for (auto& delete_pred : all_del_preds) {
+ if (delete_pred->version().first > end_version) {
+ continue;
+ }
+ tablet_schema->merge_dropped_columns(tablet->tablet_schema(delete_pred->version()));
+ }
+ res = delete_handler->init(tablet_schema, all_del_preds, end_version);
+ if (!res) {
+ LOG(WARNING) << "init delete handler failed. tablet=" << tablet->full_name()
+ << ", end_version=" << end_version;
+ break;
+ }
+
+ // acquire data sources correspond to history versions
+ tablet->capture_rs_readers(versions_to_be_changed, rs_readers);
+ if (rs_readers->size() < 1) {
+ LOG(WARNING) << "fail to acquire all data sources. "
+ << "version_num=" << versions_to_be_changed.size()
+ << ", data_source_num=" << rs_readers->size();
+ res = Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>();
+ break;
+ }
+
+ // reader_context is stack variables, it's lifetime should keep the same with rs_readers
+ RowsetReaderContext reader_context;
+ reader_context.reader_type = READER_ALTER_TABLE;
+ reader_context.tablet_schema = tablet_schema;
+ reader_context.need_ordered_result = true;
+ reader_context.delete_handler = delete_handler;
Review Comment:
If delete_handler filter rows that's marked deleted, the index will get rows not exactly the same as segment data file, and then MATCH query will produce wrong result. This need to be tested carefully.
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
Review Comment:
It's more efficient to use pair as key instead of string format. Furter more, if the three loops merged into one and process segment one by one, index_id can be used as key.
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
+ }
+ return step;
+ };
+
+ try {
+ do {
+ auto step = next_run_step();
+ if (null_map[offset]) {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_nulls(step));
+ *ptr += field->size() * step;
+ } else {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, step));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, step));
+ }
+ }
+ offset += step;
+ } while (offset < num_rows);
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_add_data(const std::string& column_name,
+ const std::string& index_writer_sign, Field* field,
+ const uint8_t** ptr, size_t num_rows) {
+ try {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, num_rows));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, num_rows));
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_write_inverted_index(int32_t segment_idx,
+ vectorized::Block* block) {
+ LOG(INFO) << "begin to write inverted index";
+
+ // converter block data
+ vectorized::OlapBlockDataConvertor olap_data_convertor;
Review Comment:
It's not efficient to create an olap_data_convertor for each block.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16371:
URL: https://github.com/apache/doris/pull/16371#issuecomment-1421906996
PR approved by anyone and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098264279
##########
be/src/olap/task/engine_alter_tablet_task.cpp:
##########
@@ -45,4 +45,34 @@ Status EngineAlterTabletTask::execute() {
return res;
} // execute
+EngineAlterInvertedIndexTask::EngineAlterInvertedIndexTask(
+ const TAlterInvertedIndexReq& alter_inverted_index_request)
+ : _alter_inverted_index_req(alter_inverted_index_request) {
+ _mem_tracker = std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::SCHEMA_CHANGE,
+ fmt::format("EngineAlterInvertedIndexTask#tabletId={}",
+ std::to_string(_alter_inverted_index_req.tablet_id)),
+ config::memory_limitation_per_thread_for_schema_change_bytes);
+}
+
+Status EngineAlterInvertedIndexTask::execute() {
+ SCOPED_ATTACH_TASK(_mem_tracker);
+ DorisMetrics::instance()->create_rollup_requests_total->increment(1);
Review Comment:
updated
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098264569
##########
fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java:
##########
@@ -337,6 +366,10 @@ protected void runPendingJob() throws AlterCancelException {
}
private void addShadowIndexToCatalog(OlapTable tbl) {
+ if (invertedIndexChange) {
Review Comment:
updated
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098271701
##########
be/src/vec/olap/olap_data_convertor.h:
##########
@@ -46,13 +46,16 @@ class IOlapColumnDataAccessor {
class OlapBlockDataConvertor {
public:
+ OlapBlockDataConvertor() = default;
Review Comment:
`OlapBlockDataConvertor(const TabletSchema* tablet_schema)` will insert all columns in `tablet_schema` into `_convertors`, we only need columns in `_alter_inverted_indexs`, if use `OlapBlockDataConvertor(const TabletSchema* tablet_schema)`, will assert failed in function `set_source_content`
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098508480
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
Review Comment:
you are right, use pair as key is better.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] Tanya-W commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "Tanya-W (via GitHub)" <gi...@apache.org>.
Tanya-W commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098289372
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +583,248 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(idx);
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ SegmentCacheHandle segment_cache_handle;
+ // load segments
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ block->clear_column_data();
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // copy block
+ auto ref_block = *block;
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), &ref_block) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ } while (block->rows() != 0);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ std::string writer_sign = fmt::format("{}_{}", i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(const std::string& column_name,
+ const std::string& index_writer_sign,
+ Field* field, const uint8_t* null_map,
+ const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
+ }
+ return step;
+ };
+
+ try {
+ do {
+ auto step = next_run_step();
+ if (null_map[offset]) {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_nulls(step));
+ *ptr += field->size() * step;
+ } else {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, step));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, step));
+ }
+ }
+ offset += step;
+ } while (offset < num_rows);
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_add_data(const std::string& column_name,
+ const std::string& index_writer_sign, Field* field,
+ const uint8_t** ptr, size_t num_rows) {
+ try {
+ if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+ DCHECK(field->get_sub_field_count() == 1);
+ const auto* col_cursor = reinterpret_cast<const CollectionValue*>(*ptr);
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_array_values(
+ field->get_sub_field(0)->size(), col_cursor, num_rows));
+ } else {
+ RETURN_IF_ERROR(_inverted_index_builders[index_writer_sign]->add_values(
+ column_name, *ptr, num_rows));
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ return Status::OK();
+}
+
+Status SchemaChangeForInvertedIndex::_write_inverted_index(int32_t segment_idx,
+ vectorized::Block* block) {
+ LOG(INFO) << "begin to write inverted index";
+
+ // converter block data
+ vectorized::OlapBlockDataConvertor olap_data_convertor;
Review Comment:
yes, you are right, I will fix it.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16371: [Feature-WIP](inverted index) Implementation for alter inverted index.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16371:
URL: https://github.com/apache/doris/pull/16371#discussion_r1098518535
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +582,236 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {
+ _olap_data_convertor = std::make_unique<vectorized::OlapBlockDataConvertor>();
+}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+ _olap_data_convertor.reset();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() || rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ std::vector<ColumnId> return_columns;
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ _olap_data_convertor->reserve(_alter_inverted_indexs.size());
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(), i);
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto index_id = inverted_index.index_id;
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto column_idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(column_idx);
+ _olap_data_convertor->add_column_data_convertor(column);
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index, *_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter> inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder, segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ auto writer_sign = std::make_pair(i, index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign, std::move(inverted_index_builder)));
+ }
+ }
+ }
+
+ // load segments
+ SegmentCacheHandle segment_cache_handle;
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()), &segment_cache_handle,
+ false));
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(), return_columns);
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+ std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for inverted index.");
+ }
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), block.get()) != Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ block->clear_column_data();
+ } while (true);
+ }
+
+ // finish write inverted index, flush data to compound file
+ for (auto i = 0; i < rowset_meta->num_segments(); ++i) {
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto index_id = inverted_index.index_id;
+ auto writer_sign = std::make_pair(i, index_id);
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" << rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(
+ const std::string& column_name, const std::pair<int64_t, int64_t>& index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr, size_t num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:727:
```diff
- else
+ } else
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org