This is an automated email from the ASF dual-hosted git repository.
snlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 4bed91b Add segment selector for merged segments (#3583)
4bed91b is described below
commit 4bed91b6fa53365adf55b12a465555c2a14a9f45
Author: Seunghyun Lee <[email protected]>
AuthorDate: Tue Dec 4 17:56:54 2018 -0800
Add segment selector for merged segments (#3583)
* Add segment selector for merged segments
Once the merge lineage information is written to the property store,
broker will read the lineage and pick only segments that are required
for answering the query. This PR adds the segment selector for merged
segments.
1. Add segment selector interface
2. Add merged segment selector implementation
3. Add unit tests
* Addressed comments
---
.../routing/HelixExternalViewBasedRouting.java | 25 +++-
.../BasePartitionAwareRoutingTableBuilder.java | 8 +-
.../routing/builder/BaseRoutingTableBuilder.java | 11 +-
.../builder/DefaultOfflineRoutingTableBuilder.java | 8 +-
.../DefaultRealtimeRoutingTableBuilder.java | 11 +-
.../routing/builder/RoutingTableBuilder.java | 8 +-
.../routing/selector/MergedSegmentSelector.java | 92 +++++++++++++
.../broker/routing/selector/SegmentSelector.java | 46 +++++++
.../routing/selector/SegmentSelectorProvider.java | 44 ++++++
.../pinot/broker/routing/RoutingTableTest.java | 1 +
.../BalancedRandomRoutingTableBuilderTest.java | 2 +-
.../HighLevelConsumerRoutingTableBuilderTest.java | 2 +-
...rtitionAwareOfflineRoutingTableBuilderTest.java | 9 +-
...titionAwareRealtimeRoutingTableBuilderTest.java | 10 +-
.../selector/MergedSegmentSelectorTest.java | 149 +++++++++++++++++++++
.../{routing => util}/FakePropertyStore.java | 12 +-
16 files changed, 404 insertions(+), 34 deletions(-)
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/HelixExternalViewBasedRouting.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/HelixExternalViewBasedRouting.java
index c3db57d..840c06a 100644
---
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/HelixExternalViewBasedRouting.java
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/HelixExternalViewBasedRouting.java
@@ -17,6 +17,8 @@ package com.linkedin.pinot.broker.routing;
import com.google.common.collect.Sets;
import com.linkedin.pinot.broker.routing.builder.RoutingTableBuilder;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelector;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelectorProvider;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.config.TableNameBuilder;
import com.linkedin.pinot.common.metrics.BrokerMeter;
@@ -67,6 +69,7 @@ public class HelixExternalViewBasedRouting implements
RoutingTable {
private final Map<String, Map<String, InstanceConfig>>
_lastKnownInstanceConfigsForTable = new ConcurrentHashMap<>();
private final Map<String, InstanceConfig> _lastKnownInstanceConfigs = new
ConcurrentHashMap<>();
private final Map<String, Set<String>> _tablesForInstance = new
ConcurrentHashMap<>();
+ private final Map<String, SegmentSelector> _segmentSelectorMap = new
ConcurrentHashMap<>();
private final HelixExternalViewBasedTimeBoundaryService _timeBoundaryService;
private final HelixManager _helixManager;
@@ -76,25 +79,25 @@ public class HelixExternalViewBasedRouting implements
RoutingTable {
private Configuration _configuration;
- private ZkHelixPropertyStore<ZNRecord> _propertyStore;
-
private RoutingTableBuilderFactory _routingTableBuilderFactory;
+ private SegmentSelectorProvider _segmentSelectorProvider;
+
public HelixExternalViewBasedRouting(ZkHelixPropertyStore<ZNRecord>
propertyStore, HelixManager helixManager,
Configuration configuration) {
- _propertyStore = propertyStore;
_configuration = configuration;
_timeBoundaryService = new
HelixExternalViewBasedTimeBoundaryService(propertyStore);
_routingTableBuilderMap = new HashMap<>();
_helixManager = helixManager;
_routingTableBuilderFactory = new
RoutingTableBuilderFactory(_configuration, propertyStore);
+ _segmentSelectorProvider = new SegmentSelectorProvider(propertyStore);
}
@Override
public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request) {
String tableName = request.getTableName();
RoutingTableBuilder routingTableBuilder =
_routingTableBuilderMap.get(tableName);
- return routingTableBuilder.getRoutingTable(request);
+ return routingTableBuilder.getRoutingTable(request,
_segmentSelectorMap.get(tableName));
}
@Override
@@ -116,6 +119,13 @@ public class HelixExternalViewBasedRouting implements
RoutingTable {
tableName);
_routingTableBuilderMap.put(tableName, routingTableBuilder);
+ // Initialize segment selector
+ SegmentSelector segmentSelector =
_segmentSelectorProvider.getSegmentSelector(tableConfig);
+ if (segmentSelector != null) {
+ LOGGER.info("Initialized segmentSelector: {} for table {}",
segmentSelector.getClass().getName(), tableName);
+ _segmentSelectorMap.put(tableName, segmentSelector);
+ }
+
// Build the routing table
if (externalView == null) {
// It is possible for us to get a request to serve a table for which
there is no external view. In this case, just
@@ -236,8 +246,15 @@ public class HelixExternalViewBasedRouting implements
RoutingTable {
try {
Map<String, InstanceConfig> relevantInstanceConfigs = new HashMap<>();
+ // Update routing table builder
routingTableBuilder.computeOnExternalViewChange(tableNameWithType,
externalView, instanceConfigs);
+ // Update segment selector
+ SegmentSelector segmentSelector =
_segmentSelectorMap.get(tableNameWithType);
+ if (segmentSelector != null) {
+ segmentSelector.computeOnExternalViewChange();
+ }
+
// Keep track of the instance configs that are used in that routing table
updateInstanceConfigsMapFromExternalView(relevantInstanceConfigs,
instanceConfigs, externalView);
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BasePartitionAwareRoutingTableBuilder.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BasePartitionAwareRoutingTableBuilder.java
index 8d1f568..fe15602 100644
---
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BasePartitionAwareRoutingTableBuilder.java
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BasePartitionAwareRoutingTableBuilder.java
@@ -18,6 +18,7 @@ package com.linkedin.pinot.broker.routing.builder;
import com.linkedin.pinot.broker.pruner.SegmentPrunerContext;
import com.linkedin.pinot.broker.pruner.SegmentZKMetadataPrunerService;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelector;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.metadata.segment.SegmentZKMetadata;
import com.linkedin.pinot.common.metrics.BrokerMeter;
@@ -82,14 +83,17 @@ public abstract class BasePartitionAwareRoutingTableBuilder
implements RoutingTa
}
@Override
- public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request) {
+ public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request, SegmentSelector segmentSelector) {
// Copy the reference for the current segment to replica to server mapping
for snapshot
Map<String, Map<Integer, String>> segmentToReplicaToServerMap =
_segmentToReplicaToServerMap;
// Get all available segments for table
Set<String> segmentsToQuery = segmentToReplicaToServerMap.keySet();
- // TODO: add the selection logic here
+ // Selecting segments only required for processing a query
+ if (segmentSelector != null) {
+ segmentsToQuery = segmentSelector.selectSegments(request,
segmentsToQuery);
+ }
Map<String, List<String>> routingTable = new HashMap<>();
SegmentPrunerContext prunerContext = new
SegmentPrunerContext(request.getBrokerRequest());
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BaseRoutingTableBuilder.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BaseRoutingTableBuilder.java
index c00d30d..6b7993a 100644
---
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BaseRoutingTableBuilder.java
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/BaseRoutingTableBuilder.java
@@ -16,6 +16,7 @@
package com.linkedin.pinot.broker.routing.builder;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelector;
import com.linkedin.pinot.common.config.RoutingConfig;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.metrics.BrokerMeter;
@@ -123,16 +124,16 @@ public abstract class BaseRoutingTableBuilder implements
RoutingTableBuilder {
}
}
- @Override
- public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request) {
+ public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request, SegmentSelector segmentSelector) {
if (_enableDynamicComputing) {
// Copy the pointer for snapshot since the pointer for segment to
servers map can change at anytime
Map<String, List<String>> segmentToServersMap = _segmentToServersMap;
- // Get all existing segments
+ // Selecting segments only required for processing a query
Set<String> segmentsToQuery = segmentToServersMap.keySet();
-
- // TODO: add the selection logic here
+ if (segmentSelector != null) {
+ segmentsToQuery = segmentSelector.selectSegments(request,
segmentsToQuery);
+ }
// Compute the final routing table
return computeDynamicRoutingTable(segmentToServersMap, segmentsToQuery);
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultOfflineRoutingTableBuilder.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultOfflineRoutingTableBuilder.java
index 07cea6b..2ed57c4 100644
---
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultOfflineRoutingTableBuilder.java
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultOfflineRoutingTableBuilder.java
@@ -16,6 +16,7 @@
package com.linkedin.pinot.broker.routing.builder;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelector;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.metrics.BrokerMetrics;
import java.util.HashSet;
@@ -47,7 +48,8 @@ public class DefaultOfflineRoutingTableBuilder implements
RoutingTableBuilder {
private int _minReplicaCountForLargeCluster = 4;
@Override
- public void init(Configuration configuration, TableConfig tableConfig,
ZkHelixPropertyStore<ZNRecord> propertyStore, BrokerMetrics brokerMetrics) {
+ public void init(Configuration configuration, TableConfig tableConfig,
ZkHelixPropertyStore<ZNRecord> propertyStore,
+ BrokerMetrics brokerMetrics) {
_largeClusterRoutingTableBuilder = new LargeClusterRoutingTableBuilder();
_smallClusterRoutingTableBuilder = new BalancedRandomRoutingTableBuilder();
if (configuration.containsKey("minServerCountForLargeCluster")) {
@@ -121,8 +123,8 @@ public class DefaultOfflineRoutingTableBuilder implements
RoutingTableBuilder {
}
@Override
- public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request) {
- return _routingTableBuilder.getRoutingTable(request);
+ public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request, SegmentSelector segmentSelector) {
+ return _routingTableBuilder.getRoutingTable(request, segmentSelector);
}
@Override
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultRealtimeRoutingTableBuilder.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultRealtimeRoutingTableBuilder.java
index d78de04..72a9a79 100644
---
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultRealtimeRoutingTableBuilder.java
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/DefaultRealtimeRoutingTableBuilder.java
@@ -16,6 +16,7 @@
package com.linkedin.pinot.broker.routing.builder;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelector;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.metrics.BrokerMetrics;
import com.linkedin.pinot.common.utils.SegmentName;
@@ -68,7 +69,7 @@ public class DefaultRealtimeRoutingTableBuilder implements
RoutingTableBuilder {
}
@Override
- public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request) {
+ public Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest
request, SegmentSelector segmentSelector) {
boolean forceLLC = false;
boolean forceHLC = false;
for (String routingOption : request.getRoutingOptions()) {
@@ -85,14 +86,14 @@ public class DefaultRealtimeRoutingTableBuilder implements
RoutingTableBuilder {
}
if (forceLLC) {
- return _realtimeLLCRoutingTableBuilder.getRoutingTable(request);
+ return _realtimeLLCRoutingTableBuilder.getRoutingTable(request,
segmentSelector);
} else if (forceHLC) {
- return _realtimeHLCRoutingTableBuilder.getRoutingTable(request);
+ return _realtimeHLCRoutingTableBuilder.getRoutingTable(request,
segmentSelector);
} else {
if (_hasLLC) {
- return _realtimeLLCRoutingTableBuilder.getRoutingTable(request);
+ return _realtimeLLCRoutingTableBuilder.getRoutingTable(request,
segmentSelector);
} else if (_hasHLC) {
- return _realtimeHLCRoutingTableBuilder.getRoutingTable(request);
+ return _realtimeHLCRoutingTableBuilder.getRoutingTable(request,
segmentSelector);
} else {
return Collections.emptyMap();
}
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/RoutingTableBuilder.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/RoutingTableBuilder.java
index 5a9c320..d27546a 100644
---
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/RoutingTableBuilder.java
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/builder/RoutingTableBuilder.java
@@ -16,6 +16,7 @@
package com.linkedin.pinot.broker.routing.builder;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.broker.routing.selector.SegmentSelector;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.metrics.BrokerMetrics;
import java.util.List;
@@ -45,9 +46,12 @@ public interface RoutingTableBuilder {
void computeOnExternalViewChange(String tableName, ExternalView
externalView, List<InstanceConfig> instanceConfigs);
/**
- * Get the routing table based on the given lookup request.
+ * Get the routing table based on the given lookup request and segment
selector
+ *
+ * TODO: we need to consider relocating segment selector into the routing
table builder instead of passing it
+ * from outside.
*/
- Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest request);
+ Map<String, List<String>> getRoutingTable(RoutingTableLookupRequest request,
SegmentSelector segmentSelector);
/**
* Get all pre-computed routing tables.
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/MergedSegmentSelector.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/MergedSegmentSelector.java
new file mode 100644
index 0000000..7d74146
--- /dev/null
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/MergedSegmentSelector.java
@@ -0,0 +1,92 @@
+/**
+ * Copyright (C) 2014-2018 LinkedIn Corp. ([email protected])
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.linkedin.pinot.broker.routing.selector;
+
+import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.common.config.TableConfig;
+import com.linkedin.pinot.common.lineage.SegmentGroup;
+import com.linkedin.pinot.common.lineage.SegmentMergeLineage;
+import com.linkedin.pinot.common.lineage.SegmentMergeLineageAccessHelper;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+
+
+/**
+ * Segment selector for merged segments
+ */
+public class MergedSegmentSelector implements SegmentSelector {
+ private String _tableNameWithType;
+ private ZkHelixPropertyStore<ZNRecord> _propertyStore;
+ private volatile SegmentGroup _rootSegmentGroup;
+
+ @Override
+ public void init(TableConfig tableConfig, ZkHelixPropertyStore<ZNRecord>
propertyStore) {
+ _tableNameWithType = tableConfig.getTableName();
+ _propertyStore = propertyStore;
+ }
+
+ @Override
+ public void computeOnExternalViewChange() {
+ SegmentMergeLineage segmentMergeLineage =
SegmentMergeLineageAccessHelper.getSegmentMergeLineage(_propertyStore,
+ _tableNameWithType);
+ _rootSegmentGroup = segmentMergeLineage.getMergeLineageRootSegmentGroup();
+ }
+
+ @Override
+ public Set<String> selectSegments(RoutingTableLookupRequest request,
Set<String> segmentsToQuery) {
+ Set<String> selectedSegments = new HashSet<>(segmentsToQuery);
+ for (SegmentGroup segmentGroup : _rootSegmentGroup.getChildrenGroups()) {
+ computeSelectionProcessForSegmentGroup(segmentGroup, selectedSegments,
segmentsToQuery);
+ }
+ return selectedSegments;
+ }
+
+ private void computeSelectionProcessForSegmentGroup(SegmentGroup
segmentGroup, Set<String> selectedSegments,
+ Set<String> availableSegments) {
+ Set<String> segmentsForGroup = segmentGroup.getSegments();
+
+ if (availableSegments.containsAll(segmentsForGroup)) {
+ // If we pick the current group node, we delete all segments covered by
children groups
+ if (segmentGroup.getChildrenGroups() == null ||
segmentGroup.getChildrenGroups().isEmpty()) {
+ return;
+ }
+ for (SegmentGroup child: segmentGroup.getChildrenGroups()) {
+ removeSegmentsForSegmentGroup(child, selectedSegments);
+ }
+ } else {
+ // If the current group is not picked, we compute the selection
recursively for children nodes
+ selectedSegments.removeAll(segmentsForGroup);
+ for (SegmentGroup child: segmentGroup.getChildrenGroups()) {
+ computeSelectionProcessForSegmentGroup(child, selectedSegments,
availableSegments);
+ }
+ }
+ }
+
+ private void removeSegmentsForSegmentGroup(SegmentGroup segmentGroup,
Set<String> selectedSegments) {
+ Set<String> segmentsForGroup = segmentGroup.getSegments();
+ selectedSegments.removeAll(segmentsForGroup);
+
+ if (segmentGroup.getChildrenGroups() == null ||
segmentGroup.getChildrenGroups().isEmpty()) {
+ return;
+ }
+
+ for (SegmentGroup child : segmentGroup.getChildrenGroups()) {
+ removeSegmentsForSegmentGroup(child, selectedSegments);
+ }
+ }
+}
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/SegmentSelector.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/SegmentSelector.java
new file mode 100644
index 0000000..9114c31
--- /dev/null
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/SegmentSelector.java
@@ -0,0 +1,46 @@
+/**
+ * Copyright (C) 2014-2018 LinkedIn Corp. ([email protected])
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.linkedin.pinot.broker.routing.selector;
+
+import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
+import com.linkedin.pinot.common.config.TableConfig;
+import java.util.Set;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+
+
+/**
+ * Interface for segment selector
+ */
+public interface SegmentSelector {
+
+ /**
+ * Initiate the segment selector
+ */
+ void init(TableConfig tableConfig, ZkHelixPropertyStore<ZNRecord>
propertyStore);
+
+ /**
+ * Compute and update the information required for a selector
+ * <p>Should be called whenever there is an external view change
+ */
+ void computeOnExternalViewChange();
+
+ /**
+ * Compute selection algorithm.
+ * <p>Should NOT modify input list of segments to query. Should copy it if
any modification is required.
+ */
+ Set<String> selectSegments(RoutingTableLookupRequest request, Set<String>
segmentsToQuery);
+}
diff --git
a/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/SegmentSelectorProvider.java
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/SegmentSelectorProvider.java
new file mode 100644
index 0000000..502ddfd
--- /dev/null
+++
b/pinot-broker/src/main/java/com/linkedin/pinot/broker/routing/selector/SegmentSelectorProvider.java
@@ -0,0 +1,44 @@
+/**
+ * Copyright (C) 2014-2018 LinkedIn Corp. ([email protected])
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.linkedin.pinot.broker.routing.selector;
+
+import com.linkedin.pinot.common.config.TableConfig;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+
+
+/**
+ * Factory for segment selector
+ */
+public class SegmentSelectorProvider {
+
+ private ZkHelixPropertyStore<ZNRecord> _propertyStore;
+
+ public SegmentSelectorProvider(ZkHelixPropertyStore<ZNRecord> propertyStore)
{
+ _propertyStore = propertyStore;
+ }
+
+ /**
+ * Get a segment selector based on table config. This method returns null
when no segment selector is required
+ *
+ * @param tableConfig a table config
+ * @return a segment selector or returns null if no selector is required
+ */
+ public SegmentSelector getSegmentSelector(TableConfig tableConfig) {
+ // TODO: add the support for merged segment selector once merge config is
updated.
+ return null;
+ }
+}
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/RoutingTableTest.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/RoutingTableTest.java
index 25600d3..3413711 100644
---
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/RoutingTableTest.java
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/RoutingTableTest.java
@@ -17,6 +17,7 @@ package com.linkedin.pinot.broker.routing;
import
com.linkedin.pinot.broker.routing.builder.HighLevelConsumerBasedRoutingTableBuilder;
import com.linkedin.pinot.broker.routing.builder.RoutingTableBuilder;
+import com.linkedin.pinot.broker.util.FakePropertyStore;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.config.TableConfig.Builder;
import com.linkedin.pinot.common.config.TableNameBuilder;
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/BalancedRandomRoutingTableBuilderTest.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/BalancedRandomRoutingTableBuilderTest.java
index 7bae71d..46ca40c 100644
---
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/BalancedRandomRoutingTableBuilderTest.java
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/BalancedRandomRoutingTableBuilderTest.java
@@ -79,7 +79,7 @@ public class BalancedRandomRoutingTableBuilderTest {
// Build routing table
routingTableBuilder.computeOnExternalViewChange("dummy", externalView,
instanceConfigList);
RoutingTableLookupRequest request = new
RoutingTableLookupRequest(tableNameWithType);
- Map<String, List<String>> routingTable =
routingTableBuilder.getRoutingTable(request);
+ Map<String, List<String>> routingTable =
routingTableBuilder.getRoutingTable(request, null);
Set<String> segmentsInRoutingTable = new HashSet<>();
for (List<String> segments : routingTable.values()) {
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/HighLevelConsumerRoutingTableBuilderTest.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/HighLevelConsumerRoutingTableBuilderTest.java
index 74f53c3..c2ef4e4 100644
---
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/HighLevelConsumerRoutingTableBuilderTest.java
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/HighLevelConsumerRoutingTableBuilderTest.java
@@ -95,7 +95,7 @@ public class HighLevelConsumerRoutingTableBuilderTest {
// Check if the routing table result is correct
for (int run = 0; run < MAX_NUM_GROUPS * 10; run++) {
RoutingTableLookupRequest request = new
RoutingTableLookupRequest(tableNameWithType);
- Map<String, List<String>> routingTable =
routingTableBuilder.getRoutingTable(request);
+ Map<String, List<String>> routingTable =
routingTableBuilder.getRoutingTable(request, null);
Set<String> coveredSegments = new HashSet<>();
for (List<String> segmentsForServer : routingTable.values()) {
coveredSegments.addAll(segmentsForServer);
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareOfflineRoutingTableBuilderTest.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareOfflineRoutingTableBuilderTest.java
index c110b6e..5a44a28 100644
---
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareOfflineRoutingTableBuilderTest.java
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareOfflineRoutingTableBuilderTest.java
@@ -15,7 +15,7 @@
*/
package com.linkedin.pinot.broker.routing.builder;
-import com.linkedin.pinot.broker.routing.FakePropertyStore;
+import com.linkedin.pinot.broker.util.FakePropertyStore;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
import com.linkedin.pinot.common.config.ReplicaGroupStrategyConfig;
import com.linkedin.pinot.common.config.RoutingConfig;
@@ -50,7 +50,6 @@ public class PartitionAwareOfflineRoutingTableBuilderTest {
private static final String PARTITION_COLUMN = "memberId";
private static final Pql2Compiler COMPILER = new Pql2Compiler();
-
private static final Random RANDOM = new Random();
private int NUM_REPLICA;
@@ -114,7 +113,7 @@ public class PartitionAwareOfflineRoutingTableBuilderTest {
// Check the query that requires to scan all segment.
String countStarQuery = "select count(*) from myTable";
Map<String, List<String>> routingTable =
-
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery));
+
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery),
null);
// Check that the number of servers picked are always equal or less than
the number of servers
// from a single replica group.
@@ -134,7 +133,7 @@ public class PartitionAwareOfflineRoutingTableBuilderTest {
for (int queryPartition = 0; queryPartition < 100; queryPartition++) {
String filterQuery = "select count(*) from myTable where " +
PARTITION_COLUMN + " = " + queryPartition;
routingTable =
-
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(filterQuery));
+
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(filterQuery),
null);
// Check that the number of servers picked are always equal or less
than the number of servers
// in a single replica group.
@@ -218,7 +217,7 @@ public class PartitionAwareOfflineRoutingTableBuilderTest {
for (int i = 0; i < 100; i++) {
String countStarQuery = "select count(*) from " + OFFLINE_TABLE_NAME;
Map<String, List<String>> routingTable =
-
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery));
+
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery),
null);
Assert.assertEquals(routingTable.keySet().size(), 1);
servers.add(routingTable.keySet().iterator().next());
}
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareRealtimeRoutingTableBuilderTest.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareRealtimeRoutingTableBuilderTest.java
index 7f349f8..7890b1b 100644
---
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareRealtimeRoutingTableBuilderTest.java
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/builder/PartitionAwareRealtimeRoutingTableBuilderTest.java
@@ -15,7 +15,7 @@
*/
package com.linkedin.pinot.broker.routing.builder;
-import com.linkedin.pinot.broker.routing.FakePropertyStore;
+import com.linkedin.pinot.broker.util.FakePropertyStore;
import com.linkedin.pinot.broker.routing.RoutingTableLookupRequest;
import com.linkedin.pinot.common.config.ColumnPartitionConfig;
import com.linkedin.pinot.common.config.RoutingConfig;
@@ -102,7 +102,7 @@ public class PartitionAwareRealtimeRoutingTableBuilderTest {
// Check the query that requires to scan all segment.
String countStarQuery = "select count(*) from myTable";
Map<String, List<String>> routingTable =
-
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery));
+
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery),
null);
// Check that all segments are covered exactly for once.
Set<String> assignedSegments = new HashSet<>();
@@ -117,7 +117,7 @@ public class PartitionAwareRealtimeRoutingTableBuilderTest {
// Check the broker side server and segment pruning.
for (int queryPartition = 0; queryPartition < 100; queryPartition++) {
String filterQuery = "select count(*) from myTable where " +
PARTITION_COLUMN + " = " + queryPartition;
- routingTable =
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(filterQuery));
+ routingTable =
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(filterQuery),
null);
int partition = queryPartition % NUM_PARTITION;
assignedSegments = new HashSet<>();
@@ -178,7 +178,7 @@ public class PartitionAwareRealtimeRoutingTableBuilderTest {
// Check the query that requires to scan all segment.
String countStarQuery = "select count(*) from myTable";
Map<String, List<String>> routingTable =
-
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery));
+
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery),
null);
// Check that all segments are covered exactly for once.
Set<String> assignedSegments = new HashSet<>();
@@ -253,7 +253,7 @@ public class PartitionAwareRealtimeRoutingTableBuilderTest {
for (int i = 0; i < 100; i++) {
String countStarQuery = "select count(*) from " + REALTIME_TABLE_NAME;
Map<String, List<String>> routingTable =
-
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery));
+
routingTableBuilder.getRoutingTable(buildRoutingTableLookupRequest(countStarQuery),
null);
Assert.assertEquals(routingTable.keySet().size(), 1);
servers.addAll(routingTable.keySet());
}
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/selector/MergedSegmentSelectorTest.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/selector/MergedSegmentSelectorTest.java
new file mode 100644
index 0000000..2edc891
--- /dev/null
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/selector/MergedSegmentSelectorTest.java
@@ -0,0 +1,149 @@
+/**
+ * Copyright (C) 2014-2018 LinkedIn Corp. ([email protected])
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.linkedin.pinot.broker.routing.selector;
+
+import com.linkedin.pinot.broker.util.FakePropertyStore;
+import com.linkedin.pinot.common.config.TableConfig;
+import com.linkedin.pinot.common.lineage.SegmentMergeLineage;
+import com.linkedin.pinot.common.lineage.SegmentMergeLineageAccessHelper;
+import com.linkedin.pinot.common.utils.CommonConstants;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class MergedSegmentSelectorTest {
+ private static final String TEST_TABLE_NAME = "test_OFFLINE";
+
+ @Test
+ public void testMergedSegmentSelector() throws Exception {
+ MergedSegmentSelector segmentSelector = new MergedSegmentSelector();
+
+ TableConfig tableConfig =
+ new
TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(TEST_TABLE_NAME).build();
+ ZkHelixPropertyStore<ZNRecord> fakePropertyStore = new FakePropertyStore();
+ segmentSelector.init(tableConfig, fakePropertyStore);
+
+ // Add G0 (segment0), G1 (segment1), G2 (segment2), G3 (segment3) that
represent 4 base segments
+ SegmentMergeLineage segmentMergeLineage = new
SegmentMergeLineage(TEST_TABLE_NAME);
+ for (int i = 0; i < 4; i++) {
+ String groupId = "G" + i;
+ segmentMergeLineage.addSegmentGroup(groupId, Arrays.asList(new
String[]{"segment" + i}), null);
+ }
+
+ // Update segment merge lineage in the property store and update segment
selector
+
SegmentMergeLineageAccessHelper.writeSegmentMergeLineage(fakePropertyStore,
segmentMergeLineage, 0);
+ segmentSelector.computeOnExternalViewChange();
+
+ // Test the result of segment selection
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3"},
+ new String[]{"segment0", "segment1", "segment2", "segment3"});
+
+ // Add G4 (merged_0, merged_1) that was merged from G0, G1, G2
+ String mergedGroupId = "G4";
+ List<String> childrenGroups = Arrays.asList(new String[]{"G0", "G1",
"G2"});
+ segmentMergeLineage.addSegmentGroup(mergedGroupId,
+ Arrays.asList(new String[]{"merged0", "merged1"}), childrenGroups);
+
+ // Update segment merge lineage in the property store and update segment
selector
+
SegmentMergeLineageAccessHelper.writeSegmentMergeLineage(fakePropertyStore,
segmentMergeLineage, 0);
+ segmentSelector.computeOnExternalViewChange();
+
+ // Test the result of segment selection
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"merged0", "merged1"},
+ new String[]{"merged0", "merged1", "segment3"});
+
+ // Test the result of segment selection when the external view does not
have one of merged segments
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"merged0"},
+ new String[]{"segment0", "segment1", "segment2", "segment3"});
+
+ // Test the result of segment selection when one of the base segment is
missing.
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment2", "segment3", "merged0", "merged1"},
+ new String[]{"merged0", "merged1", "segment3"});
+
+
+ // Add G5 (segment4)
+ segmentMergeLineage.addSegmentGroup("G5", Arrays.asList(new
String[]{"segment4"}), null);
+
+ // Update segment merge lineage in the property store and update segment
selector
+
SegmentMergeLineageAccessHelper.writeSegmentMergeLineage(fakePropertyStore,
segmentMergeLineage, 0);
+ segmentSelector.computeOnExternalViewChange();
+
+ // Test the result of segment selection
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged0", "merged1"},
+ new String[]{"merged0", "merged1", "segment3", "segment4"});
+
+ // Add G6 (merged3) that is merged from G3 (segment3), G5 (segment4)
+ segmentMergeLineage.addSegmentGroup("G6", Arrays.asList(new
String[]{"merged2"}),
+ Arrays.asList(new String[]{"G3", "G5"}));
+
+ // Update segment merge lineage in the property store and update segment
selector
+
SegmentMergeLineageAccessHelper.writeSegmentMergeLineage(fakePropertyStore,
segmentMergeLineage, 0);
+ segmentSelector.computeOnExternalViewChange();
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged0", "merged1", "merged2"},
+ new String[]{"merged0", "merged1", "merged2"});
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged1", "merged2"},
+ new String[]{"segment0", "segment1", "segment2", "merged2"});
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged1"},
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4"});
+
+ // Add G7 (merged4) that is merged from G3, G6
+ segmentMergeLineage.addSegmentGroup("G7", Arrays.asList(new
String[]{"merged3"}),
+ Arrays.asList(new String[]{"G4", "G6"}));
+
+ // Update segment merge lineage in the property store and update segment
selector
+
SegmentMergeLineageAccessHelper.writeSegmentMergeLineage(fakePropertyStore,
segmentMergeLineage, 0);
+ segmentSelector.computeOnExternalViewChange();
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged0", "merged1", "merged2", "merged3"},
+ new String[]{"merged3"});
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged0", "merged2", "merged3"},
+ new String[]{"merged3"});
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged0", "merged1", "merged2"},
+ new String[]{"merged0", "merged1", "merged2"});
+
+ testSegmentSelector(segmentSelector,
+ new String[]{"segment0", "segment1", "segment2", "segment3",
"segment4", "merged0", "merged2"},
+ new String[]{"merged2", "segment0", "segment1", "segment2"});
+ }
+
+ private void testSegmentSelector(SegmentSelector segmentSelector, String[]
segmentsToQuery, String[] expectedResult) {
+ Set<String> segmentsToQuerySet = new
HashSet<>(Arrays.asList(segmentsToQuery));
+ Set<String> selectedSegments = segmentSelector.selectSegments(null,
segmentsToQuerySet);
+ Assert.assertEquals(selectedSegments, new
HashSet<>(Arrays.asList(expectedResult)));
+ }
+}
diff --git
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/FakePropertyStore.java
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/util/FakePropertyStore.java
similarity index 88%
rename from
pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/FakePropertyStore.java
rename to
pinot-broker/src/test/java/com/linkedin/pinot/broker/util/FakePropertyStore.java
index 6bb8322..70e5310 100644
---
a/pinot-broker/src/test/java/com/linkedin/pinot/broker/routing/FakePropertyStore.java
+++
b/pinot-broker/src/test/java/com/linkedin/pinot/broker/util/FakePropertyStore.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package com.linkedin.pinot.broker.routing;
+package com.linkedin.pinot.broker.util;
import java.util.HashMap;
import java.util.Map;
@@ -42,6 +42,16 @@ public class FakePropertyStore extends
ZkHelixPropertyStore<ZNRecord> {
}
@Override
+ public boolean set(String path, ZNRecord stat, int expectedVersion, int
options) {
+ try {
+ setContents(path, stat);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ @Override
public boolean set(String path, ZNRecord stat, int options) {
try {
setContents(path, stat);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
