xiangfu0 commented on code in PR #19738:
URL: https://github.com/apache/pinot/pull/19738#discussion_r4290446108
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java:
##########
@@ -78,9 +78,31 @@ private HelixHelper() {
public static final String BROKER_RESOURCE =
CommonConstants.Helix.BROKER_RESOURCE_INSTANCE;
+ /// Returns a deep copy of the IdealState with the same content as cloning
it through a serialization round trip of
+ /// its ZNRecord: id, simple fields, map fields, list fields and raw
payload, with the same container types and
+ /// iteration order. Like the round trip, the ZK stat fields (e.g. version)
are not copied. Unlike the round trip, the
+ /// serializer's write policies (list field bound, size limit) are not
applied; they still apply when the IdealState
+ /// is written.
+ ///
+ /// Copying the fields directly avoids serializing (and compressing) the
whole ZNRecord and parsing it back, which
+ /// takes seconds for an IdealState with hundreds of thousands of segments
and widens the window for version
+ /// conflicts when updating it.
public static IdealState cloneIdealState(IdealState idealState) {
- return new IdealState(
- (ZNRecord)
ZN_RECORD_SERIALIZER.deserialize(ZN_RECORD_SERIALIZER.serialize(idealState.getRecord())));
+ ZNRecord record = idealState.getRecord();
+ IdealState copy = new IdealState(record.getId());
+ ZNRecord copyRecord = copy.getRecord();
+ copyRecord.setSimpleFields(new TreeMap<>(record.getSimpleFields()));
+ TreeMap<String, Map<String, String>> mapFields = new
TreeMap<>(record.getMapFields());
+ mapFields.replaceAll((key, value) -> value != null ? new
LinkedHashMap<>(value) : null);
+ copyRecord.setMapFields(mapFields);
+ TreeMap<String, List<String>> listFields = new
TreeMap<>(record.getListFields());
+ listFields.replaceAll((key, value) -> value != null ? new
ArrayList<>(value) : null);
+ copyRecord.setListFields(listFields);
Review Comment:
Worth double-checking this intentional behavior change: the old round trip
applied the serializer's `LIST_FIELD_BOUND` truncation to list fields, and this
copy does not. For IdealState that's harmless since it never sets
`ZNRecord.LIST_FIELD_BOUND`, and the bound still gets applied on the actual
write. The docstring already calls this out, so I'd just suggest keeping an eye
on it if `cloneIdealState` ever gets reused for records that do set that bound.
##########
pinot-common/src/test/java/org/apache/pinot/common/utils/helix/HelixHelperTest.java:
##########
@@ -88,4 +106,105 @@ public void testRemoveDisabledPartitions() {
assertTrue(HelixHelper.removeDisabledPartitions(instanceConfig));
assertTrue(instanceConfig.getDisabledPartitionsMap().isEmpty());
}
+
+ @Test
+ public void testCloneIdealState() {
+ IdealState idealState = new IdealState("myTable_REALTIME");
+ ZNRecord record = idealState.getRecord();
+ record.setSimpleField("REPLICAS", "2");
+ record.setBooleanField("enableCompression", true);
+ // Insertion order that differs from the sorted order, as in a
deserialized ZNRecord
+ Map<String, Map<String, String>> mapFields = new LinkedHashMap<>();
+ Map<String, String> instanceStateMap = new LinkedHashMap<>();
+ instanceStateMap.put("Server_2", "CONSUMING");
+ instanceStateMap.put("Server_1", "CONSUMING");
+ mapFields.put("segment_2", instanceStateMap);
+ mapFields.put("segment_1", new LinkedHashMap<>(Map.of("Server_1",
"ONLINE")));
+ record.setMapFields(mapFields);
+ record.setListField("segment_2", new ArrayList<>(List.of("Server_2",
"Server_1")));
+ record.setRawPayload(new byte[]{1, 2, 3});
+ record.setVersion(5);
+
+ // Expected result of cloning through a serialization round trip
+ ZNRecordSerializer serializer = new ZNRecordSerializer();
+ ZNRecord expected =
+ new IdealState((ZNRecord)
serializer.deserialize(serializer.serialize(record))).getRecord();
+ ZNRecord copy = HelixHelper.cloneIdealState(idealState).getRecord();
+
+ // Same content and iteration order
+ assertEquals(copy.getId(), expected.getId());
+ assertEquals(copy, expected);
+ assertEquals(new ArrayList<>(copy.getMapFields().keySet()), new
ArrayList<>(expected.getMapFields().keySet()));
+ assertEquals(new ArrayList<>(copy.getMapField("segment_2").keySet()),
+ new ArrayList<>(expected.getMapField("segment_2").keySet()));
+ assertEquals(copy.getRawPayload(), expected.getRawPayload());
+ assertEquals(copy.getVersion(), expected.getVersion());
+
+ // Modifying the copy does not modify the original
+ assertNotSame(copy.getRawPayload(), record.getRawPayload());
+ copy.getSimpleFields().put("REPLICAS", "3");
+ copy.getMapField("segment_2").put("Server_2", "ONLINE");
+ copy.getMapFields().remove("segment_1");
+ copy.getListField("segment_2").add("Server_3");
+ assertEquals(record.getSimpleField("REPLICAS"), "2");
+ assertEquals(record.getMapField("segment_2").get("Server_2"), "CONSUMING");
+ assertTrue(record.getMapFields().containsKey("segment_1"));
+ assertEquals(record.getListField("segment_2"), List.of("Server_2",
"Server_1"));
+ }
+
+ /// cloneIdealState copies a ZNRecord field by field, so it must know every
field an IdealState stores in ZK. If a
+ /// Helix upgrade fails this test, make cloneIdealState copy the new state
(or leave it out if it is not stored, like
+ /// the ZK stat fields), then update the expected values.
+ @Test
+ public void testCloneIdealStateKnowsAllStoredFields()
+ throws IOException {
+ // Instance fields, including inherited ones, sorted
+ assertEquals(getInstanceFieldNames(ZNRecord.class),
+ List.of("_creationTime", "_deltaList", "_ephemeralOwner",
"_modifiedTime", "_serializer", "_version", "id",
+ "listFields", "mapFields", "rawPayload", "simpleFields"));
+ assertEquals(getInstanceFieldNames(IdealState.class), List.of("_record",
"_stat"));
+
+ // Of these, only the following are stored in ZK: the properties the
serializer's mapper can write, whatever their
+ // values, and the properties it writes for a record with all of them set
+ Set<String> storedFields = Set.of("id", "simpleFields", "mapFields",
"listFields", "rawPayload");
+ ObjectMapper mapper = ZNRecordSerializerMapper.get();
+
assertEquals(mapper.getSerializationConfig().introspect(mapper.constructType(ZNRecord.class)).findProperties()
+ .stream()
+ .filter(BeanPropertyDefinition::couldSerialize)
+ .map(BeanPropertyDefinition::getName)
+ .collect(Collectors.toSet()), storedFields);
+ ZNRecord record = new ZNRecord("myTable_REALTIME");
+ record.setSimpleField("REPLICAS", "2");
+ record.setMapField("segment_1", Map.of("Server_1", "ONLINE"));
+ record.setListField("segment_1", List.of("Server_1"));
+ record.setRawPayload(new byte[]{1});
+ byte[] bytes = new ZNRecordSerializer().serialize(record);
+ if (GZipCompressionUtil.isCompressed(bytes)) {
+ bytes = GZipCompressionUtil.uncompress(new ByteArrayInputStream(bytes));
+ }
+ Set<String> writtenFields = new HashSet<>();
+
JsonUtils.bytesToJsonNode(bytes).fieldNames().forEachRemaining(writtenFields::add);
+ assertEquals(writtenFields, storedFields);
Review Comment:
This is a thoughtful guard test — pinning the exact ZNRecord/IdealState
fields and the serialized property set means a Helix upgrade that adds a new
stored field will fail here loudly instead of silently dropping it from the
clone. The pairing of Jackson introspection (value-independent) with the
actually-written bytes is a nice touch. No changes needed.
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java:
##########
@@ -78,9 +78,31 @@ private HelixHelper() {
public static final String BROKER_RESOURCE =
CommonConstants.Helix.BROKER_RESOURCE_INSTANCE;
+ /// Returns a deep copy of the IdealState with the same content as cloning
it through a serialization round trip of
+ /// its ZNRecord: id, simple fields, map fields, list fields and raw
payload, with the same container types and
+ /// iteration order. Like the round trip, the ZK stat fields (e.g. version)
are not copied. Unlike the round trip, the
+ /// serializer's write policies (list field bound, size limit) are not
applied; they still apply when the IdealState
+ /// is written.
+ ///
+ /// Copying the fields directly avoids serializing (and compressing) the
whole ZNRecord and parsing it back, which
+ /// takes seconds for an IdealState with hundreds of thousands of segments
and widens the window for version
+ /// conflicts when updating it.
public static IdealState cloneIdealState(IdealState idealState) {
- return new IdealState(
- (ZNRecord)
ZN_RECORD_SERIALIZER.deserialize(ZN_RECORD_SERIALIZER.serialize(idealState.getRecord())));
+ ZNRecord record = idealState.getRecord();
+ IdealState copy = new IdealState(record.getId());
+ ZNRecord copyRecord = copy.getRecord();
+ copyRecord.setSimpleFields(new TreeMap<>(record.getSimpleFields()));
+ TreeMap<String, Map<String, String>> mapFields = new
TreeMap<>(record.getMapFields());
+ mapFields.replaceAll((key, value) -> value != null ? new
LinkedHashMap<>(value) : null);
+ copyRecord.setMapFields(mapFields);
+ TreeMap<String, List<String>> listFields = new
TreeMap<>(record.getListFields());
+ listFields.replaceAll((key, value) -> value != null ? new
ArrayList<>(value) : null);
+ copyRecord.setListFields(listFields);
Review Comment:
Worth double-checking this intentional behavior change: the old round trip
applied the serializer's `LIST_FIELD_BOUND` truncation to list fields, and this
copy does not. For IdealState that's harmless since it never sets
`ZNRecord.LIST_FIELD_BOUND`, and the bound still gets applied on the actual
write. The docstring already calls this out, so I'd just suggest keeping an eye
on it if `cloneIdealState` ever gets reused for records that do set that bound.
##########
pinot-common/src/test/java/org/apache/pinot/common/utils/helix/HelixHelperTest.java:
##########
@@ -88,4 +106,105 @@ public void testRemoveDisabledPartitions() {
assertTrue(HelixHelper.removeDisabledPartitions(instanceConfig));
assertTrue(instanceConfig.getDisabledPartitionsMap().isEmpty());
}
+
+ @Test
+ public void testCloneIdealState() {
+ IdealState idealState = new IdealState("myTable_REALTIME");
+ ZNRecord record = idealState.getRecord();
+ record.setSimpleField("REPLICAS", "2");
+ record.setBooleanField("enableCompression", true);
+ // Insertion order that differs from the sorted order, as in a
deserialized ZNRecord
+ Map<String, Map<String, String>> mapFields = new LinkedHashMap<>();
+ Map<String, String> instanceStateMap = new LinkedHashMap<>();
+ instanceStateMap.put("Server_2", "CONSUMING");
+ instanceStateMap.put("Server_1", "CONSUMING");
+ mapFields.put("segment_2", instanceStateMap);
+ mapFields.put("segment_1", new LinkedHashMap<>(Map.of("Server_1",
"ONLINE")));
+ record.setMapFields(mapFields);
+ record.setListField("segment_2", new ArrayList<>(List.of("Server_2",
"Server_1")));
+ record.setRawPayload(new byte[]{1, 2, 3});
+ record.setVersion(5);
+
+ // Expected result of cloning through a serialization round trip
+ ZNRecordSerializer serializer = new ZNRecordSerializer();
+ ZNRecord expected =
+ new IdealState((ZNRecord)
serializer.deserialize(serializer.serialize(record))).getRecord();
+ ZNRecord copy = HelixHelper.cloneIdealState(idealState).getRecord();
+
+ // Same content and iteration order
+ assertEquals(copy.getId(), expected.getId());
+ assertEquals(copy, expected);
+ assertEquals(new ArrayList<>(copy.getMapFields().keySet()), new
ArrayList<>(expected.getMapFields().keySet()));
+ assertEquals(new ArrayList<>(copy.getMapField("segment_2").keySet()),
+ new ArrayList<>(expected.getMapField("segment_2").keySet()));
+ assertEquals(copy.getRawPayload(), expected.getRawPayload());
+ assertEquals(copy.getVersion(), expected.getVersion());
+
+ // Modifying the copy does not modify the original
+ assertNotSame(copy.getRawPayload(), record.getRawPayload());
+ copy.getSimpleFields().put("REPLICAS", "3");
+ copy.getMapField("segment_2").put("Server_2", "ONLINE");
+ copy.getMapFields().remove("segment_1");
+ copy.getListField("segment_2").add("Server_3");
+ assertEquals(record.getSimpleField("REPLICAS"), "2");
+ assertEquals(record.getMapField("segment_2").get("Server_2"), "CONSUMING");
+ assertTrue(record.getMapFields().containsKey("segment_1"));
+ assertEquals(record.getListField("segment_2"), List.of("Server_2",
"Server_1"));
+ }
+
+ /// cloneIdealState copies a ZNRecord field by field, so it must know every
field an IdealState stores in ZK. If a
+ /// Helix upgrade fails this test, make cloneIdealState copy the new state
(or leave it out if it is not stored, like
+ /// the ZK stat fields), then update the expected values.
+ @Test
+ public void testCloneIdealStateKnowsAllStoredFields()
+ throws IOException {
+ // Instance fields, including inherited ones, sorted
+ assertEquals(getInstanceFieldNames(ZNRecord.class),
+ List.of("_creationTime", "_deltaList", "_ephemeralOwner",
"_modifiedTime", "_serializer", "_version", "id",
+ "listFields", "mapFields", "rawPayload", "simpleFields"));
+ assertEquals(getInstanceFieldNames(IdealState.class), List.of("_record",
"_stat"));
+
+ // Of these, only the following are stored in ZK: the properties the
serializer's mapper can write, whatever their
+ // values, and the properties it writes for a record with all of them set
+ Set<String> storedFields = Set.of("id", "simpleFields", "mapFields",
"listFields", "rawPayload");
+ ObjectMapper mapper = ZNRecordSerializerMapper.get();
+
assertEquals(mapper.getSerializationConfig().introspect(mapper.constructType(ZNRecord.class)).findProperties()
+ .stream()
+ .filter(BeanPropertyDefinition::couldSerialize)
+ .map(BeanPropertyDefinition::getName)
+ .collect(Collectors.toSet()), storedFields);
+ ZNRecord record = new ZNRecord("myTable_REALTIME");
+ record.setSimpleField("REPLICAS", "2");
+ record.setMapField("segment_1", Map.of("Server_1", "ONLINE"));
+ record.setListField("segment_1", List.of("Server_1"));
+ record.setRawPayload(new byte[]{1});
+ byte[] bytes = new ZNRecordSerializer().serialize(record);
+ if (GZipCompressionUtil.isCompressed(bytes)) {
+ bytes = GZipCompressionUtil.uncompress(new ByteArrayInputStream(bytes));
+ }
+ Set<String> writtenFields = new HashSet<>();
+
JsonUtils.bytesToJsonNode(bytes).fieldNames().forEachRemaining(writtenFields::add);
+ assertEquals(writtenFields, storedFields);
Review Comment:
This is a thoughtful guard test — pinning the exact ZNRecord/IdealState
fields and the serialized property set means a Helix upgrade that adds a new
stored field will fail here loudly instead of silently dropping it from the
clone. The pairing of Jackson introspection (value-independent) with the
actually-written bytes is a nice touch. No changes needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
