Skip to content

Refine IndexHandler methods a bit to make them reentrant #9440

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Jackie-Jiang merged 1 commit into from
Sep 21, 2022
Merged

Refine IndexHandler methods a bit to make them reentrant #9440

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions ...va/org/apache/pinot/segment/local/segment/index/loader/invertedindex/FSTIndexHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,22 +74,23 @@ public class FSTIndexHandler implements IndexHandler {
public FSTIndexHandler(SegmentMetadata segmentMetadata, IndexLoadingConfig indexLoadingConfig) {
_segmentMetadata = segmentMetadata;
_fstType = indexLoadingConfig.getFSTIndexType();
_columnsToAddIdx = new HashSet<>(indexLoadingConfig.getFSTIndexColumns());
_columnsToAddIdx = indexLoadingConfig.getFSTIndexColumns();
Copy link
Contributor

@vvivekiyer vvivekiyer Sep 20, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(nit) Isn't it probably cleaner and safer (from future changes) to not have _columnsToAddIdx as a private member? Instead just have _indexLoadingConfig as a private member and compute columnsToAddIdx in the respective methods?

This is the pattern followed in other handlers like BloomFilterHandler,H3IndexHandler.

Copy link
Contributor Author

@klsince klsince Sep 20, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see BF/H3 hander doesn't hold indexLoadingConfig reference either, but configs for creating BF/H3.
The handlers in this PR don't have specific configs from indexLoadingConfig either, so not hold its reference.

}

@Override
public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.FST_INDEX);
// Check if any existing index need to be removed.
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Need to remove existing FST index from segment: {}, column: {}", segmentName, column);
return true;
}
}
// Check if any new index need to be added.
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateFSTIndex(columnMetadata)) {
LOGGER.info("Need to create new FST index for segment: {}, column: {}", segmentName, column);
Expand All @@ -104,15 +105,16 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter, IndexCreatorPro
throws Exception {
// Remove indices not set in table config any more
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentWriter.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.FST_INDEX);
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Removing existing FST index from segment: {}, column: {}", segmentName, column);
segmentWriter.removeIndex(column, ColumnIndexType.FST_INDEX);
LOGGER.info("Removed existing FST index from segment: {}, column: {}", segmentName, column);
}
}
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateFSTIndex(columnMetadata)) {
createFSTIndexForColumn(segmentWriter, columnMetadata, indexCreatorProvider);
Expand Down
14 changes: 8 additions & 6 deletions ...g/apache/pinot/segment/local/segment/index/loader/invertedindex/InvertedIndexHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,27 +47,28 @@ public class InvertedIndexHandler implements IndexHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(InvertedIndexHandler.class);

private final SegmentMetadata _segmentMetadata;
private final HashSet<String> _columnsToAddIdx;
private final Set<String> _columnsToAddIdx;

public InvertedIndexHandler(SegmentMetadata segmentMetadata, IndexLoadingConfig indexLoadingConfig) {
_segmentMetadata = segmentMetadata;
_columnsToAddIdx = new HashSet<>(indexLoadingConfig.getInvertedIndexColumns());
_columnsToAddIdx = indexLoadingConfig.getInvertedIndexColumns();
}

@Override
public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns =
segmentReader.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.INVERTED_INDEX);
// Check if any existing index need to be removed.
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Need to remove existing inverted index from segment: {}, column: {}", segmentName, column);
return true;
}
}
// Check if any new index need to be added.
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateInvertedIndex(columnMetadata)) {
LOGGER.info("Need to create new inverted index for segment: {}, column: {}", segmentName, column);
Expand All @@ -82,16 +83,17 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter, IndexCreatorPro
throws IOException {
// Remove indices not set in table config any more.
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns =
segmentWriter.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.INVERTED_INDEX);
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Removing existing inverted index from segment: {}, column: {}", segmentName, column);
segmentWriter.removeIndex(column, ColumnIndexType.INVERTED_INDEX);
LOGGER.info("Removed existing inverted index from segment: {}, column: {}", segmentName, column);
}
}
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateInvertedIndex(columnMetadata)) {
createInvertedIndexForColumn(segmentWriter, columnMetadata, indexCreatorProvider);
Expand Down
14 changes: 8 additions & 6 deletions ...a/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,26 +50,27 @@ public class JsonIndexHandler implements IndexHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(JsonIndexHandler.class);

private final SegmentMetadata _segmentMetadata;
private final HashSet<String> _columnsToAddIdx;
private final Set<String> _columnsToAddIdx;

public JsonIndexHandler(SegmentMetadata segmentMetadata, IndexLoadingConfig indexLoadingConfig) {
_segmentMetadata = segmentMetadata;
_columnsToAddIdx = new HashSet<>(indexLoadingConfig.getJsonIndexColumns());
_columnsToAddIdx = indexLoadingConfig.getJsonIndexColumns();
}

@Override
public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.JSON_INDEX);
// Check if any existing index need to be removed.
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Need to remove existing json index from segment: {}, column: {}", segmentName, column);
return true;
}
}
// Check if any new index need to be added.
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateJsonIndex(columnMetadata)) {
LOGGER.info("Need to create new json index for segment: {}, column: {}", segmentName, column);
Expand All @@ -84,15 +85,16 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter, IndexCreatorPro
throws Exception {
// Remove indices not set in table config any more
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentWriter.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.JSON_INDEX);
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Removing existing json index from segment: {}, column: {}", segmentName, column);
segmentWriter.removeIndex(column, ColumnIndexType.JSON_INDEX);
LOGGER.info("Removed existing json index from segment: {}, column: {}", segmentName, column);
}
}
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateJsonIndex(columnMetadata)) {
createJsonIndexForColumn(segmentWriter, columnMetadata, indexCreatorProvider);
Expand Down
12 changes: 7 additions & 5 deletions .../org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,23 +52,24 @@ public class RangeIndexHandler implements IndexHandler {

public RangeIndexHandler(SegmentMetadata segmentMetadata, IndexLoadingConfig indexLoadingConfig) {
_segmentMetadata = segmentMetadata;
_columnsToAddIdx = new HashSet<>(indexLoadingConfig.getRangeIndexColumns());
_columnsToAddIdx = indexLoadingConfig.getRangeIndexColumns();
_rangeIndexVersion = indexLoadingConfig.getRangeIndexVersion();
}

@Override
public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.RANGE_INDEX);
// Check if any existing index need to be removed.
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Need to remove existing range index from segment: {}, column: {}", segmentName, column);
return true;
}
}
// Check if any new index need to be added.
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateRangeIndex(columnMetadata)) {
LOGGER.info("Need to create new range index for segment: {}, column: {}", segmentName, column);
Expand All @@ -83,15 +84,16 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter, IndexCreatorPro
throws IOException {
// Remove indices not set in table config any more
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentWriter.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.RANGE_INDEX);
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Removing existing range index from segment: {}, column: {}", segmentName, column);
segmentWriter.removeIndex(column, ColumnIndexType.RANGE_INDEX);
LOGGER.info("Removed existing range index from segment: {}, column: {}", segmentName, column);
}
}
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateRangeIndex(columnMetadata)) {
createRangeIndexForColumn(segmentWriter, columnMetadata, indexCreatorProvider);
Expand Down
12 changes: 7 additions & 5 deletions ...a/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,22 +91,23 @@ public class TextIndexHandler implements IndexHandler {
public TextIndexHandler(SegmentMetadata segmentMetadata, IndexLoadingConfig indexLoadingConfig) {
_segmentMetadata = segmentMetadata;
_fstType = indexLoadingConfig.getFSTIndexType();
_columnsToAddIdx = new HashSet<>(indexLoadingConfig.getTextIndexColumns());
_columnsToAddIdx = indexLoadingConfig.getTextIndexColumns();
}

@Override
public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.TEXT_INDEX);
// Check if any existing index need to be removed.
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Need to remove existing text index from segment: {}, column: {}", segmentName, column);
return true;
}
}
// Check if any new index need to be added.
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateTextIndex(columnMetadata)) {
LOGGER.info("Need to create new text index for segment: {}, column: {}", segmentName, column);
Expand All @@ -121,15 +122,16 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter, IndexCreatorPro
throws Exception {
// Remove indices not set in table config any more
String segmentName = _segmentMetadata.getName();
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
Set<String> existingColumns = segmentWriter.toSegmentDirectory().getColumnsWithIndex(ColumnIndexType.TEXT_INDEX);
for (String column : existingColumns) {
if (!_columnsToAddIdx.remove(column)) {
if (!columnsToAddIdx.remove(column)) {
LOGGER.info("Removing existing text index from segment: {}, column: {}", segmentName, column);
segmentWriter.removeIndex(column, ColumnIndexType.TEXT_INDEX);
LOGGER.info("Removed existing text index from segment: {}, column: {}", segmentName, column);
}
}
for (String column : _columnsToAddIdx) {
for (String column : columnsToAddIdx) {
ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);
if (shouldCreateTextIndex(columnMetadata)) {
createTextIndexForColumn(segmentWriter, columnMetadata, indexCreatorProvider);
Expand Down