You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by sn...@apache.org on 2020/06/16 04:17:28 UTC
[incubator-pinot] branch master updated: Replace
SegmentMergeLineage to SegmentLineage (#5534)
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 b3a8368 Replace SegmentMergeLineage to SegmentLineage (#5534)
b3a8368 is described below
commit b3a83688e66e9522e54f81cb9872865db2d30e73
Author: Seunghyun Lee <sn...@linkedin.com>
AuthorDate: Mon Jun 15 21:17:19 2020 -0700
Replace SegmentMergeLineage to SegmentLineage (#5534)
* Replace SegmentMergeLineage to SegmentLineage
With the updated design (#2715) of segment merge, the metadata format
for segment lineage is changed.
1. Added segment lineage class which is a wrapper of ZNRecord
2. Added lineage entry class that represents m->n segment replacement
3. Added unit test
* Addressed the comments
---
.../apache/pinot/common/lineage/LineageEntry.java | 75 +++++
.../pinot/common/lineage/LineageEntryState.java | 26 ++
.../apache/pinot/common/lineage/SegmentGroup.java | 75 -----
.../pinot/common/lineage/SegmentLineage.java | 141 +++++++++
...Helper.java => SegmentLineageAccessHelper.java} | 45 ++-
.../pinot/common/lineage/SegmentMergeLineage.java | 352 ---------------------
.../pinot/common/metadata/ZKMetadataProvider.java | 6 +-
.../pinot/common/lineage/SegmentLineageTest.java | 88 ++++++
.../common/lineage/SegmentMergeLineageTest.java | 147 ---------
9 files changed, 355 insertions(+), 600 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/lineage/LineageEntry.java b/pinot-common/src/main/java/org/apache/pinot/common/lineage/LineageEntry.java
new file mode 100644
index 0000000..6cda88e
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/lineage/LineageEntry.java
@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.lineage;
+
+import java.util.List;
+import java.util.Objects;
+
+
+/**
+ * Class to represent the lineage entry.
+ *
+ */
+public class LineageEntry {
+ private final List<String> _segmentsFrom;
+ private final List<String> _segmentsTo;
+ private final LineageEntryState _state;
+ private final long _timestamp;
+
+ public LineageEntry(List<String> segmentsFrom, List<String> segmentsTo, LineageEntryState state, long timestamp) {
+ _segmentsFrom = segmentsFrom;
+ _segmentsTo = segmentsTo;
+ _state = state;
+ _timestamp = timestamp;
+ }
+
+ public List<String> getSegmentsFrom() {
+ return _segmentsFrom;
+ }
+
+ public List<String> getSegmentsTo() {
+ return _segmentsTo;
+ }
+
+ public LineageEntryState getState() {
+ return _state;
+ }
+
+ public long getTimestamp() {
+ return _timestamp;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ LineageEntry that = (LineageEntry) o;
+ return _timestamp == that._timestamp && _segmentsFrom.equals(that._segmentsFrom) && _segmentsTo.equals(
+ that._segmentsTo) && _state == that._state;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(_segmentsFrom, _segmentsTo, _state, _timestamp);
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/lineage/LineageEntryState.java b/pinot-common/src/main/java/org/apache/pinot/common/lineage/LineageEntryState.java
new file mode 100644
index 0000000..74407c0
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/lineage/LineageEntryState.java
@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.lineage;
+
+/**
+ * Enum for represent the state of lineage entry
+ */
+public enum LineageEntryState {
+ IN_PROGRESS, COMPLETED
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentGroup.java b/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentGroup.java
deleted file mode 100644
index ce6beda..0000000
--- a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentGroup.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.common.lineage;
-
-import java.util.List;
-import java.util.Set;
-
-
-/**
- * Class to represent segment group
- */
-public class SegmentGroup {
-
- private String _groupId;
- private int _groupLevel;
- private SegmentGroup _parentGroup;
- private List<SegmentGroup> _childrenGroups;
- private Set<String> _segments;
-
- public String getGroupId() {
- return _groupId;
- }
-
- public void setGroupId(String groupId) {
- _groupId = groupId;
- }
-
- public SegmentGroup getParentGroup() {
- return _parentGroup;
- }
-
- public void setParentGroup(SegmentGroup parentGroup) {
- _parentGroup = parentGroup;
- }
-
- public List<SegmentGroup> getChildrenGroups() {
- return _childrenGroups;
- }
-
- public void setChildrenGroups(List<SegmentGroup> childrenGroups) {
- _childrenGroups = childrenGroups;
- }
-
- public Set<String> getSegments() {
- return _segments;
- }
-
- public void setSegments(Set<String> segments) {
- _segments = segments;
- }
-
- public int getGroupLevel() {
- return _groupLevel;
- }
-
- public void setGroupLevel(int groupLevel) {
- _groupLevel = groupLevel;
- }
-}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentLineage.java b/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentLineage.java
new file mode 100644
index 0000000..ee61f11
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentLineage.java
@@ -0,0 +1,141 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.lineage;
+
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.helix.ZNRecord;
+
+
+/**
+ * Class to represent segment lineage information.
+ *
+ * Segment lineage keeps the metadata required for supporting m -> n segment replacement. Segment lineage is serialized
+ * into a znode and stored in a helix property store (zookeeper). This metadata will be used by brokers to make sure
+ * that the routing does not pick the segments with the duplicate data.
+ *
+ * NOTE: Update for the underlying segment lineage znode needs to happen with read-modify-write block to guarantee the
+ * atomic update because this metadata can be modified concurrently (e.g. task scheduler tries to add entries after
+ * scheduling new tasks while minion task tries to update the state of the existing entry)
+ */
+public class SegmentLineage {
+ private static final String COMMA_SEPARATOR = ",";
+
+ private final String _tableNameWithType;
+ private final Map<String, LineageEntry> _lineageEntries;
+
+ public SegmentLineage(String tableNameWithType) {
+ _tableNameWithType = tableNameWithType;
+ _lineageEntries = new HashMap<>();
+ }
+
+ public SegmentLineage(String tableNameWithType, Map<String, LineageEntry> lineageEntries) {
+ _tableNameWithType = tableNameWithType;
+ _lineageEntries = lineageEntries;
+ }
+
+ public String getTableNameWithType() {
+ return _tableNameWithType;
+ }
+
+ /**
+ * Add lineage entry to the segment lineage metadata
+ * @param lineageEntry a lineage entry
+ * @return the id for the input lineage entry for the access
+ */
+ public String addLineageEntry(LineageEntry lineageEntry) {
+ String lineageId = generateLineageId();
+ _lineageEntries.put(lineageId, lineageEntry);
+ return lineageId;
+ }
+
+ /**
+ * Retrieve lineage entry
+ * @param lineageEntryId the id for the lineage entry
+ * @return the lineage entry for the given lineage entry id
+ */
+ public LineageEntry getLineageEntry(String lineageEntryId) {
+ return _lineageEntries.get(lineageEntryId);
+ }
+
+ /**
+ * Retrieve the lineage ids for all lineage entries
+ * @return lineage entry ids
+ */
+ public Set<String> getLineageEntryIds() {
+ return _lineageEntries.keySet();
+ }
+
+ /**
+ * Delete lineage entry
+ * @param lineageEntryId the id for the lineage entry
+ */
+ public void deleteLineageEntry(String lineageEntryId) {
+ _lineageEntries.remove(lineageEntryId);
+ }
+
+ /**
+ * Convert ZNRecord to segment lineage
+ * @param record ZNRecord representation of the segment lineage
+ * @return the segment lineage object
+ */
+ public static SegmentLineage fromZNRecord(ZNRecord record) {
+ String tableNameWithType = record.getId();
+ Map<String, LineageEntry> lineageEntries = new HashMap<>();
+ Map<String, List<String>> listFields = record.getListFields();
+ for (Map.Entry<String, List<String>> listField : listFields.entrySet()) {
+ String lineageId = listField.getKey();
+ List<String> value = listField.getValue();
+ Preconditions.checkState(value.size() == 4);
+ List<String> segmentsFrom = Arrays.asList(value.get(0).split(COMMA_SEPARATOR));
+ List<String> segmentsTo = Arrays.asList(value.get(1).split(COMMA_SEPARATOR));
+ LineageEntryState state = LineageEntryState.valueOf(value.get(2));
+ long timestamp = Long.parseLong(value.get(3));
+ lineageEntries.put(lineageId, new LineageEntry(segmentsFrom, segmentsTo, state, timestamp));
+ }
+ return new SegmentLineage(tableNameWithType, lineageEntries);
+ }
+
+ /**
+ * Convert the segment lineage object to the ZNRecord
+ * @return ZNRecord representation of the segment lineage
+ */
+ public ZNRecord toZNRecord() {
+ ZNRecord znRecord = new ZNRecord(_tableNameWithType);
+ for (Map.Entry<String, LineageEntry> entry : _lineageEntries.entrySet()) {
+ LineageEntry lineageEntry = entry.getValue();
+ String segmentsFrom = String.join(",", lineageEntry.getSegmentsFrom());
+ String segmentsTo = String.join(",", lineageEntry.getSegmentsTo());
+ String state = lineageEntry.getState().toString();
+ String timestamp = Long.toString(lineageEntry.getTimestamp());
+ List<String> listEntry = Arrays.asList(segmentsFrom, segmentsTo, state, timestamp);
+ znRecord.setListField(entry.getKey(), listEntry);
+ }
+ return znRecord;
+ }
+
+ private String generateLineageId() {
+ return UUID.randomUUID().toString();
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentMergeLineageAccessHelper.java b/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentLineageAccessHelper.java
similarity index 55%
rename from pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentMergeLineageAccessHelper.java
rename to pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentLineageAccessHelper.java
index 1fb8cf7..b724c95 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentMergeLineageAccessHelper.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentLineageAccessHelper.java
@@ -26,56 +26,55 @@ import org.apache.zookeeper.data.Stat;
/**
- * Class to help to read, write segment merge lineage
+ * Class to help to read, write segment lineage metadata
*/
-public class SegmentMergeLineageAccessHelper {
-
+public class SegmentLineageAccessHelper {
/**
- * Read the segment merge lineage ZNRecord from the property store
+ * Read the segment lineage ZNRecord from the property store
*
* @param propertyStore a property store
* @param tableNameWithType a table name with type
* @return a ZNRecord of segment merge lineage, return null if znode does not exist
*/
- public static ZNRecord getSegmentMergeLineageZNRecord(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ public static ZNRecord getSegmentLineageZNRecord(ZkHelixPropertyStore<ZNRecord> propertyStore,
String tableNameWithType) {
- String path = ZKMetadataProvider.constructPropertyStorePathForSegmentMergeLineage(tableNameWithType);
+ String path = ZKMetadataProvider.constructPropertyStorePathForSegmentLineage(tableNameWithType);
Stat stat = new Stat();
- ZNRecord segmentMergeLineageZNRecord = propertyStore.get(path, stat, AccessOption.PERSISTENT);
- if (segmentMergeLineageZNRecord != null) {
- segmentMergeLineageZNRecord.setVersion(stat.getVersion());
+ ZNRecord segmentLineageZNRecord = propertyStore.get(path, stat, AccessOption.PERSISTENT);
+ if (segmentLineageZNRecord != null) {
+ segmentLineageZNRecord.setVersion(stat.getVersion());
}
- return segmentMergeLineageZNRecord;
+ return segmentLineageZNRecord;
}
/**
- * Read the segment merge lineage from the property store
+ * Read the segment lineage from the property store
*
* @param propertyStore a property store
* @param tableNameWithType a table name with type
- * @return a segment merge lineage, return null if znode does not exist
+ * @return a segment lineage, return null if znode does not exist
*/
- public static SegmentMergeLineage getSegmentMergeLineage(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ public static SegmentLineage getSegmentLineage(ZkHelixPropertyStore<ZNRecord> propertyStore,
String tableNameWithType) {
- ZNRecord znRecord = getSegmentMergeLineageZNRecord(propertyStore, tableNameWithType);
- SegmentMergeLineage segmentMergeLineage = null;
+ ZNRecord znRecord = getSegmentLineageZNRecord(propertyStore, tableNameWithType);
+ SegmentLineage segmentMergeLineage = null;
if (znRecord != null) {
- segmentMergeLineage = SegmentMergeLineage.fromZNRecord(znRecord);
+ segmentMergeLineage = SegmentLineage.fromZNRecord(znRecord);
}
return segmentMergeLineage;
}
/**
- * Write the segment merge lineage to the property store
+ * Write the segment lineage to the property store
*
* @param propertyStore a property store
- * @param segmentMergeLineage a segment merge lineage
+ * @param segmentLineage a segment lineage
* @return true if update is successful. false otherwise.
*/
- public static boolean writeSegmentMergeLineage(ZkHelixPropertyStore<ZNRecord> propertyStore,
- SegmentMergeLineage segmentMergeLineage, int expectedVersion) {
- String tableNameWithType = segmentMergeLineage.getTableName();
- String path = ZKMetadataProvider.constructPropertyStorePathForSegmentMergeLineage(tableNameWithType);
- return propertyStore.set(path, segmentMergeLineage.toZNRecord(), expectedVersion, AccessOption.PERSISTENT);
+ public static boolean writeSegmentLineage(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ SegmentLineage segmentLineage, int expectedVersion) {
+ String tableNameWithType = segmentLineage.getTableNameWithType();
+ String path = ZKMetadataProvider.constructPropertyStorePathForSegmentLineage(tableNameWithType);
+ return propertyStore.set(path, segmentLineage.toZNRecord(), expectedVersion, AccessOption.PERSISTENT);
}
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentMergeLineage.java b/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentMergeLineage.java
deleted file mode 100644
index 2ba3a7d..0000000
--- a/pinot-common/src/main/java/org/apache/pinot/common/lineage/SegmentMergeLineage.java
+++ /dev/null
@@ -1,352 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.common.lineage;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import org.apache.helix.ZNRecord;
-import org.apache.pinot.common.exception.InvalidConfigException;
-import org.apache.pinot.spi.utils.EqualityUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Class to represent segment merge lineage information.
- *
- * Segment merge lineage information is serialized into a znode and stored in a helix property store (zookeeper). This
- * information will be used by the broker, segment merge task generator, and retention manager.
- *
- * For each segment group, we are storing the following information:
- * 1. group id
- * - group identifier (will be stored in time based uuid format)
- * 2. group level
- * - segment level allows us to have a hierarchical representation of the segment lineage. When we assign the merge
- * task, we will only merge/roll-up segments with the same level.
- * (e.g. If hourly segment groups are in level 0, daily segment groups will belong to level 1)
- * 3. segments
- * - segments that belong to a particular segment group
- * 4. lineage information
- * - If a segment group is created by merging multiple children segment groups, we write the lineage information
- * (e.g. segment group C is merged from segment group A, B)
- *
- *
- * Example)
- * Let's say that we have 3 segments to begin with (S1, S2, S3). For original segments, we treat them specially by
- * regarding each segment as a group. So, our lineage info will have the following data.
- *
- * _parentGroupToChildrenGroupsMap = empty
- * _levelToGroupToSegmentsMap = { level_0 -> { G1 -> S1, G2 -> S2, G3 -> S3} }
- *
- * Let's say that a merge task merges S1, S2, S3 into S4, S5. Then, we now have a new group G4 (S4,S5). The lineage
- * now would be the following:
- *
- * _parentGroupToChildrenGroupsMap = { G4 -> { G1, G2, G3} }
- * _levelToGroupToSegmentsMap = { level_0 -> { G1 -> S1, G2 -> S2, G3 -> S3}
- * level_1 -> { G4 -> S4, S5} }
- *
- */
-public class SegmentMergeLineage {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(SegmentMergeLineage.class);
-
- private static final String LEVEL_KEY_PREFIX = "level_";
- private static final String ROOT_NODE_GROUP_ID = "root";
- private static final String SEGMENT_DELIMITER = ",";
- private static final int DEFAULT_GROUP_LEVEL = 0;
-
- private String _tableNameWithType;
-
- // Mapping of group id to children group ids
- private Map<String, List<String>> _parentGroupToChildrenGroupsMap;
-
- // Mapping of group level to group id to segments that belong to a group
- private Map<Integer, Map<String, List<String>>> _levelToGroupToSegmentsMap;
-
- public SegmentMergeLineage(String tableNameWithType) {
- _tableNameWithType = tableNameWithType;
- _parentGroupToChildrenGroupsMap = new HashMap<>();
- _levelToGroupToSegmentsMap = new HashMap<>();
- }
-
- public SegmentMergeLineage(String tableNameWithType, Map<String, List<String>> segmentGroupLineageMap,
- Map<Integer, Map<String, List<String>>> levelToGroupToSegmentMap) {
- _tableNameWithType = tableNameWithType;
- _parentGroupToChildrenGroupsMap = segmentGroupLineageMap;
- _levelToGroupToSegmentsMap = levelToGroupToSegmentMap;
- }
-
- public String getTableName() {
- return _tableNameWithType;
- }
-
- public static SegmentMergeLineage fromZNRecord(ZNRecord record) {
- String tableNameWithType = record.getId();
- Map<String, List<String>> segmentGroupLineageMap = record.getListFields();
-
- Map<Integer, Map<String, List<String>>> groupToSegmentsMap = new HashMap<>();
- for (Map.Entry<String, Map<String, String>> entry : record.getMapFields().entrySet()) {
- String levelKey = entry.getKey();
- Integer level = Integer.parseInt(levelKey.substring(LEVEL_KEY_PREFIX.length()));
- Map<String, List<String>> groupToSegmentsForLevel = new HashMap<>();
- for (Map.Entry<String, String> groupEntry : entry.getValue().entrySet()) {
- String groupId = groupEntry.getKey();
- String segmentsString = groupEntry.getValue();
- List<String> segments = Arrays.asList(segmentsString.split(SEGMENT_DELIMITER));
- groupToSegmentsForLevel.put(groupId, new ArrayList<>(segments));
- }
- groupToSegmentsMap.put(level, groupToSegmentsForLevel);
- }
- return new SegmentMergeLineage(tableNameWithType, segmentGroupLineageMap, groupToSegmentsMap);
- }
-
- public ZNRecord toZNRecord() {
- ZNRecord record = new ZNRecord(_tableNameWithType);
- record.setListFields(_parentGroupToChildrenGroupsMap);
- Map<String, Map<String, String>> groupToSegmentsMap = new HashMap<>();
-
- for (Map.Entry<Integer, Map<String, List<String>>> entry : _levelToGroupToSegmentsMap.entrySet()) {
- String key = LEVEL_KEY_PREFIX + entry.getKey();
- Map<String, String> groupSegmentsForLevel = new HashMap<>();
- for (Map.Entry<String, List<String>> groupEntry : entry.getValue().entrySet()) {
- String groupId = groupEntry.getKey();
- String segments = String.join(SEGMENT_DELIMITER, groupEntry.getValue());
- groupSegmentsForLevel.put(groupId, segments);
- }
- groupToSegmentsMap.put(key, groupSegmentsForLevel);
- }
- record.setMapFields(groupToSegmentsMap);
-
- return record;
- }
-
- /**
- * Add segment merge lineage information
- *
- * @param groupId a group id
- * @param currentGroupSegments a list of segments that belongs to the group
- * @param childrenGroups a list of children groups that the current group covers. All children group ids has to be
- * from the same group level.
- */
- public void addSegmentGroup(String groupId, List<String> currentGroupSegments, List<String> childrenGroups)
- throws InvalidConfigException {
- // Get group level
- Integer groupLevel = getGroupLevel(childrenGroups);
-
- Map<String, List<String>> groupToSegmentMap =
- _levelToGroupToSegmentsMap.computeIfAbsent(groupLevel, k -> new HashMap<>());
- if (groupToSegmentMap.containsKey(groupId) || _parentGroupToChildrenGroupsMap.containsKey(groupId)) {
- throw new InvalidConfigException("Group id : " + groupId + " already exists for table " + _tableNameWithType);
- }
-
- // Update group to segments map
- groupToSegmentMap.put(groupId, new ArrayList<>(currentGroupSegments));
- _levelToGroupToSegmentsMap.put(groupLevel, groupToSegmentMap);
-
- // Update segment group lineage map
- if (groupLevel > DEFAULT_GROUP_LEVEL) {
- _parentGroupToChildrenGroupsMap.put(groupId, new ArrayList<>(childrenGroups));
- }
-
- LOGGER.info("New group has been added successfully to the segment lineage. (tableName: {}, groupId: {}, "
- + "currentGroupSegments: {}, childrenGroups: {}", _tableNameWithType, groupId, currentGroupSegments,
- childrenGroups);
- }
-
- /**
- * Remove segment merge information given a group id
- *
- * @param groupId a group id
- */
- public void removeSegmentGroup(String groupId) {
- // Clean up the group id from parent to children group mapping
- _parentGroupToChildrenGroupsMap.remove(groupId);
- for (List<String> childrenGroups : _parentGroupToChildrenGroupsMap.values()) {
- childrenGroups.remove(groupId);
- }
-
- // Clean up the group id from group to segments mapping
- for (Map<String, List<String>> groupToSegments : _levelToGroupToSegmentsMap.values()) {
- groupToSegments.remove(groupId);
- }
-
- LOGGER.info("Group {} has been successfully removed for table {}.", groupId, _tableNameWithType);
- }
-
- /**
- * Construct a lineage tree and returns the root node
- *
- * @return a root node for lineage tree
- */
- public SegmentGroup getMergeLineageRootSegmentGroup() {
- // Create group nodes
- Map<String, SegmentGroup> groupNodes = new HashMap<>();
- for (Map.Entry<Integer, Map<String, List<String>>> groupEntryForLevel : _levelToGroupToSegmentsMap.entrySet()) {
- Integer level = groupEntryForLevel.getKey();
- Map<String, List<String>> groupToSegmentsForLevel = groupEntryForLevel.getValue();
- for (Map.Entry<String, List<String>> entry : groupToSegmentsForLevel.entrySet()) {
- String groupId = entry.getKey();
- List<String> segments = entry.getValue();
- SegmentGroup groupNode = new SegmentGroup();
- groupNode.setGroupId(groupId);
- groupNode.setSegments(new HashSet<>(segments));
- groupNode.setGroupLevel(level);
- groupNodes.put(groupId, groupNode);
- }
- }
-
- // Add edges by updating children & parent group information
- for (Map.Entry<String, List<String>> lineageEntry : _parentGroupToChildrenGroupsMap.entrySet()) {
- String parentGroupId = lineageEntry.getKey();
- List<String> childrenGroupIds = lineageEntry.getValue();
- List<SegmentGroup> childrenGroups = new ArrayList<>();
- SegmentGroup parentNode = groupNodes.get(parentGroupId);
- for (String groupId : childrenGroupIds) {
- SegmentGroup childNode = groupNodes.get(groupId);
- if (childNode != null) {
- childrenGroups.add(childNode);
- childNode.setParentGroup(parentNode);
- }
- }
- parentNode.setChildrenGroups(childrenGroups);
- }
-
- // Create a root node
- SegmentGroup root = new SegmentGroup();
- root.setGroupId(ROOT_NODE_GROUP_ID);
- List<SegmentGroup> childrenForRoot = new ArrayList<>();
- for (SegmentGroup group : groupNodes.values()) {
- if (group.getParentGroup() == null) {
- group.setParentGroup(root);
- childrenForRoot.add(group);
- }
- }
- root.setChildrenGroups(childrenForRoot);
-
- return root;
- }
-
- /**
- * Get a list of segments for a given group id
- *
- * @param groupId a group id
- * @return a list of segments that belongs to the given group id, null if the group does not exist
- */
- public List<String> getSegmentsForGroup(String groupId) {
- for (Map<String, List<String>> groupToSegmentMap : _levelToGroupToSegmentsMap.values()) {
- List<String> segments = groupToSegmentMap.get(groupId);
- if (segments != null) {
- return segments;
- }
- }
- return null;
- }
-
- /**
- * Get a list of children group ids for a given group id
- *
- * @param groupId a group id
- * @return a list of children groups that are covered by the given group id, null if the group does not exist
- */
- public List<String> getChildrenForGroup(String groupId) {
- return _parentGroupToChildrenGroupsMap.get(groupId);
- }
-
- /**
- * Get a list of all group levels
- *
- * @return a list of all group levels
- */
- public List<Integer> getAllGroupLevels() {
- List<Integer> groupLevels = new ArrayList<>(_levelToGroupToSegmentsMap.keySet());
- Collections.sort(groupLevels);
- return groupLevels;
- }
-
- /**
- * Get a list of group ids for a given group level
- *
- * @param groupLevel a group level
- * @return a list of group ids that belongs to the given group level, null if the group level does not exist
- */
- public List<String> getGroupIdsForGroupLevel(int groupLevel) {
- Map<String, List<String>> groupToSegmentsMap = _levelToGroupToSegmentsMap.get(groupLevel);
- if (groupToSegmentsMap != null) {
- return new ArrayList<>(groupToSegmentsMap.keySet());
- }
- return null;
- }
-
- /**
- * Helper function to compute group level given children groups
- *
- * @param childrenGroups a list of children group ids
- * @return group level
- */
- private Integer getGroupLevel(List<String> childrenGroups)
- throws InvalidConfigException {
- // If no children exists, the group belongs to the base level.
- if (childrenGroups == null || childrenGroups.isEmpty()) {
- return DEFAULT_GROUP_LEVEL;
- }
-
- for (Map.Entry<Integer, Map<String, List<String>>> entry : _levelToGroupToSegmentsMap.entrySet()) {
- Integer currentLevel = entry.getKey();
- Map<String, List<String>> currentLevelGroupToSegmentsMap = entry.getValue();
- if (currentLevelGroupToSegmentsMap.keySet().containsAll(childrenGroups)) {
- return currentLevel + 1;
- }
- }
-
- // At this point, not all children groups are covered, cannot add group
- throw new InvalidConfigException("Cannot compute group level because not all children groups exist "
- + "in the segment merge lineage, table name: " + _tableNameWithType + ", children groups: " + childrenGroups
- + "table");
- }
-
- @Override
- public boolean equals(Object o) {
- if (EqualityUtils.isSameReference(this, o)) {
- return true;
- }
-
- if (EqualityUtils.isNullOrNotSameClass(this, o)) {
- return false;
- }
-
- SegmentMergeLineage that = (SegmentMergeLineage) o;
-
- return EqualityUtils.isEqual(_tableNameWithType, that._tableNameWithType) && EqualityUtils
- .isEqual(_parentGroupToChildrenGroupsMap, that._parentGroupToChildrenGroupsMap) && EqualityUtils
- .isEqual(_levelToGroupToSegmentsMap, that._levelToGroupToSegmentsMap);
- }
-
- @Override
- public int hashCode() {
- int result = EqualityUtils.hashCodeOf(_tableNameWithType);
- result = EqualityUtils.hashCodeOf(result, _parentGroupToChildrenGroupsMap);
- result = EqualityUtils.hashCodeOf(result, _levelToGroupToSegmentsMap);
- return result;
- }
-}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
index 9256fc6..b3f007a 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
@@ -57,7 +57,7 @@ public class ZKMetadataProvider {
private static final String PROPERTYSTORE_TABLE_CONFIGS_PREFIX = "/CONFIGS/TABLE";
private static final String PROPERTYSTORE_INSTANCE_CONFIGS_PREFIX = "/CONFIGS/INSTANCE";
private static final String PROPERTYSTORE_CLUSTER_CONFIGS_PREFIX = "/CONFIGS/CLUSTER";
- private static final String PROPERTYSTORE_SEGMENT_MERGE_LINEAGE = "/SEGMENT_MERGE_LINEAGE";
+ private static final String PROPERTYSTORE_SEGMENT_LINEAGE = "/SEGMENT_LINEAGE";
public static void setRealtimeTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String realtimeTableName,
ZNRecord znRecord) {
@@ -111,8 +111,8 @@ public class ZKMetadataProvider {
return StringUtil.join("/", PROPERTYSTORE_CLUSTER_CONFIGS_PREFIX, controllerConfigKey);
}
- public static String constructPropertyStorePathForSegmentMergeLineage(String tableNameWithType) {
- return StringUtil.join("/", PROPERTYSTORE_SEGMENT_MERGE_LINEAGE, tableNameWithType);
+ public static String constructPropertyStorePathForSegmentLineage(String tableNameWithType) {
+ return StringUtil.join("/", PROPERTYSTORE_SEGMENT_LINEAGE, tableNameWithType);
}
public static boolean isSegmentExisted(ZkHelixPropertyStore<ZNRecord> propertyStore, String resourceNameForResource,
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/lineage/SegmentLineageTest.java b/pinot-common/src/test/java/org/apache/pinot/common/lineage/SegmentLineageTest.java
new file mode 100644
index 0000000..e3ce29e
--- /dev/null
+++ b/pinot-common/src/test/java/org/apache/pinot/common/lineage/SegmentLineageTest.java
@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.lineage;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.ZNRecord;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class SegmentLineageTest {
+ @Test
+ public void testSegmentLineage() throws Exception {
+ SegmentLineage segmentLineage = new SegmentLineage("test_OFFLINE");
+ String id = segmentLineage.addLineageEntry(
+ new LineageEntry(Arrays.asList("s1", "s2", "s3"), Arrays.asList("s4", "s5"), LineageEntryState.COMPLETED,
+ 11111L));
+ LineageEntry lineageEntry = segmentLineage.getLineageEntry(id);
+ Assert.assertEquals(lineageEntry.getSegmentsFrom(), Arrays.asList("s1", "s2", "s3"));
+ Assert.assertEquals(lineageEntry.getSegmentsTo(), Arrays.asList("s4", "s5"));
+ Assert.assertEquals(lineageEntry.getState(), LineageEntryState.COMPLETED);
+ Assert.assertEquals(lineageEntry.getTimestamp(), 11111L);
+
+ String id2 = segmentLineage.addLineageEntry(
+ new LineageEntry(Arrays.asList("s6", "s6", "s8"), Arrays.asList("s9", "s10"), LineageEntryState.COMPLETED,
+ 22222L));
+ LineageEntry lineageEntry2 = segmentLineage.getLineageEntry(id2);
+ Assert.assertEquals(lineageEntry2.getSegmentsFrom(), Arrays.asList("s6", "s6", "s8"));
+ Assert.assertEquals(lineageEntry2.getSegmentsTo(), Arrays.asList("s9", "s10"));
+ Assert.assertEquals(lineageEntry2.getState(), LineageEntryState.COMPLETED);
+ Assert.assertEquals(lineageEntry2.getTimestamp(), 22222L);
+
+ String id3 = segmentLineage.addLineageEntry(
+ new LineageEntry(Arrays.asList("s5", "s9"), Arrays.asList("s11"), LineageEntryState.IN_PROGRESS, 33333L));
+ LineageEntry lineageEntry3 = segmentLineage.getLineageEntry(id3);
+ Assert.assertEquals(lineageEntry3.getSegmentsFrom(), Arrays.asList("s5", "s9"));
+ Assert.assertEquals(lineageEntry3.getSegmentsTo(), Arrays.asList("s11"));
+ Assert.assertEquals(lineageEntry3.getState(), LineageEntryState.IN_PROGRESS);
+ Assert.assertEquals(lineageEntry3.getTimestamp(), 33333L);
+
+ // Test the convesion from the segment lineage to the znRecord
+ ZNRecord znRecord = segmentLineage.toZNRecord();
+ Assert.assertEquals(znRecord.getId(), "test_OFFLINE");
+
+ Map<String, List<String>> listFields = znRecord.getListFields();
+ List<String> entry = listFields.get(id);
+ Assert.assertEquals(entry.get(0), String.join(",", Arrays.asList("s1", "s2", "s3")));
+ Assert.assertEquals(entry.get(1), String.join(",", Arrays.asList("s4", "s5")));
+ Assert.assertEquals(entry.get(2), LineageEntryState.COMPLETED.toString());
+ Assert.assertEquals(entry.get(3), Long.toString(11111L));
+
+ List<String> entry2 = listFields.get(id2);
+ Assert.assertEquals(entry2.get(0), String.join(",", Arrays.asList("s6", "s6", "s8")));
+ Assert.assertEquals(entry2.get(1), String.join(",", Arrays.asList("s9", "s10")));
+ Assert.assertEquals(entry2.get(2), LineageEntryState.COMPLETED.toString());
+ Assert.assertEquals(entry2.get(3), Long.toString(22222L));
+
+ List<String> entry3 = listFields.get(id3);
+ Assert.assertEquals(entry3.get(0), String.join(",", Arrays.asList("s5", "s9")));
+ Assert.assertEquals(entry3.get(1), String.join(",", Arrays.asList("s11")));
+ Assert.assertEquals(entry3.get(2), LineageEntryState.IN_PROGRESS.toString());
+ Assert.assertEquals(entry3.get(3), Long.toString(33333L));
+
+ // Test the conversion from the znRecord to the segment lineage
+ SegmentLineage segmentLineageFromZNRecord = SegmentLineage.fromZNRecord(segmentLineage.toZNRecord());
+ Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id), lineageEntry);
+ Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id2), lineageEntry2);
+ Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id3), lineageEntry3);
+ }
+}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/lineage/SegmentMergeLineageTest.java b/pinot-common/src/test/java/org/apache/pinot/common/lineage/SegmentMergeLineageTest.java
deleted file mode 100644
index 29fe07a..0000000
--- a/pinot-common/src/test/java/org/apache/pinot/common/lineage/SegmentMergeLineageTest.java
+++ /dev/null
@@ -1,147 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.common.lineage;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import org.apache.pinot.common.exception.InvalidConfigException;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-
-public class SegmentMergeLineageTest {
-
- @Test
- public void testSegmentMergeLineage()
- throws Exception {
- SegmentMergeLineage segmentMergeLineage = new SegmentMergeLineage("test_OFFLINE");
-
- // Start with 3 original segments.
- String groupId1 = "G1";
- List<String> groupSegments1 = Arrays.asList(new String[]{"segment1"});
- segmentMergeLineage.addSegmentGroup(groupId1, groupSegments1, null);
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId1), groupSegments1);
-
- String groupId2 = "G2";
- List<String> groupSegments2 = Arrays.asList(new String[]{"segment2"});
- segmentMergeLineage.addSegmentGroup(groupId2, groupSegments2, null);
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId2), groupSegments2);
-
- String groupId3 = "G3";
- List<String> groupSegments3 = Arrays.asList(new String[]{"segment3"});
- segmentMergeLineage.addSegmentGroup(groupId3, groupSegments3, null);
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId3), groupSegments3);
-
- // Simulating when a merge task merges segment 1,2 into segment 4.
- String groupId4 = "G4";
- List<String> groupSegments4 = Arrays.asList(new String[]{"segment4"});
- segmentMergeLineage.addSegmentGroup(groupId4, groupSegments4, Arrays.asList(new String[]{groupId1, groupId2}));
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId4), groupSegments4);
- Assert.assertEquals(segmentMergeLineage.getChildrenForGroup(groupId4),
- Arrays.asList(new String[]{groupId1, groupId2}));
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId4), Arrays.asList(new String[]{"segment4"}));
-
- // 2 more original segments gets uploaded.
- String groupId5 = "G5";
- List<String> groupSegments5 = Arrays.asList(new String[]{"segment 5"});
- segmentMergeLineage.addSegmentGroup(groupId5, groupSegments5, null);
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId5), groupSegments5);
-
- String groupId6 = "G6";
- List<String> groupSegments6 = Arrays.asList(new String[]{"segment 6"});
- segmentMergeLineage.addSegmentGroup(groupId6, groupSegments6, null);
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId6), groupSegments6);
-
- // Let's say a merge task merges segment 3,5,6 into 7,8.
- String groupId7 = "G7";
- List<String> groupSegments7 = Arrays.asList(new String[]{"segment7", "segment8"});
- segmentMergeLineage
- .addSegmentGroup(groupId7, groupSegments7, Arrays.asList(new String[]{groupId3, groupId5, groupId6}));
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId7), groupSegments7);
- Assert.assertEquals(segmentMergeLineage.getChildrenForGroup(groupId7),
- Arrays.asList(new String[]{groupId3, groupId5, groupId6}));
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId7),
- Arrays.asList(new String[]{"segment7", "segment8"}));
-
- // Check available APIs
- Assert.assertEquals(segmentMergeLineage.getTableName(), "test_OFFLINE");
- Assert.assertEquals(segmentMergeLineage.getAllGroupLevels(), Arrays.asList(new Integer[]{0, 1}));
- Assert.assertTrue(segmentMergeLineage.equals(SegmentMergeLineage.fromZNRecord(segmentMergeLineage.toZNRecord())));
- Assert.assertEquals(segmentMergeLineage.getGroupIdsForGroupLevel(0),
- Arrays.asList(new String[]{groupId1, groupId2, groupId3, groupId5, groupId6}));
- Assert.assertEquals(new HashSet<>(segmentMergeLineage.getGroupIdsForGroupLevel(1)),
- new HashSet<>(Arrays.asList(new String[]{groupId4, groupId7})));
- validateSegmentGroup(segmentMergeLineage);
-
- // Check ZNRecord conversion
- Assert.assertEquals(segmentMergeLineage, SegmentMergeLineage.fromZNRecord(segmentMergeLineage.toZNRecord()));
-
- // Test removing groups
- segmentMergeLineage.removeSegmentGroup(groupId1);
- Assert.assertNull(segmentMergeLineage.getChildrenForGroup(groupId1));
- Assert.assertNull(segmentMergeLineage.getSegmentsForGroup(groupId1));
- Assert.assertFalse(segmentMergeLineage.getGroupIdsForGroupLevel(0).contains(groupId1));
- }
-
- @Test
- public void testUpdateWithDuplicateGroupId()
- throws Exception {
- SegmentMergeLineage segmentMergeLineage = new SegmentMergeLineage("test_OFFLINE");
- String groupId1 = "G1";
- List<String> groupSegments1 = Arrays.asList(new String[]{"segment1, segment2, segment3"});
- segmentMergeLineage.addSegmentGroup(groupId1, groupSegments1, null);
- Assert.assertEquals(segmentMergeLineage.getSegmentsForGroup(groupId1), groupSegments1);
-
- List<String> groupSegments2 = Arrays.asList(new String[]{"segment4, segment5, segment6"});
- try {
- segmentMergeLineage.addSegmentGroup(groupId1, groupSegments2, null);
- Assert.fail();
- } catch (InvalidConfigException e) {
- // expected
- }
- }
-
- private void validateSegmentGroup(SegmentMergeLineage segmentMergeLineage) {
- SegmentGroup segmentGroup = segmentMergeLineage.getMergeLineageRootSegmentGroup();
- for (SegmentGroup child : segmentGroup.getChildrenGroups()) {
- validateSegmentGroupNode(child, segmentMergeLineage);
- }
- }
-
- private void validateSegmentGroupNode(SegmentGroup segmentGroup, SegmentMergeLineage segmentMergeLineage) {
- String groupId = segmentGroup.getGroupId();
- Assert.assertEquals(segmentGroup.getSegments(), new HashSet<>(segmentMergeLineage.getSegmentsForGroup(groupId)));
- Assert.assertTrue(segmentMergeLineage.getGroupIdsForGroupLevel(segmentGroup.getGroupLevel()).contains(groupId));
-
- List<SegmentGroup> childrenGroups = segmentGroup.getChildrenGroups();
- if (childrenGroups != null) {
- List<String> childrenGroupIds = new ArrayList<>();
- for (SegmentGroup child : childrenGroups) {
- childrenGroupIds.add(child.getGroupId());
- }
- Assert.assertEquals(childrenGroupIds, segmentMergeLineage.getChildrenForGroup(groupId));
-
- for (SegmentGroup child : segmentGroup.getChildrenGroups()) {
- validateSegmentGroupNode(child, segmentMergeLineage);
- }
- }
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org