Handle the partitioning mismatch between table config and stream #6031

Jackie-Jiang · 2020-09-17T22:57:24Z

Description

Fix for #6029

In consuming segment, update the partition info when ingesting new records. Log a warning and emit a metric REALTIME_PARTITION_MISMATCH when the partition is not aligned with the stream partition. The updated partition info will be persisted in the segment metadata, and when the segment is committed, also update the partition info stored in the segment ZK metadata.

Added SegmentPartitionLLCRealtimeClusterIntegrationTest to test the expected behavior.

NOTE: With the fix, the consuming segment can still be pruned out incorrectly if the partition info in the table config does not align with the stream. To fix that, we can only persist the partition info for the completed segments, but not the consuming segments. Need some perf test to verify the performance penalty.

mayankshriv · 2020-09-18T16:18:01Z

...va/org/apache/pinot/integration/tests/SegmentPartitionLLCRealtimeClusterIntegrationTest.java

+
+
+/**
+ * Integration test that enables segment partition for the LLC real-time table.


Do we also test partition function and num partitions change?

As discussed offline, currently because of the existing behavior, the partition routing won't give correct result when stream is not properly partitioned. Added a TODO with the reason why it does not give correct result now, and will add a test in the next PR which will address that issue.

mcvsubbu · 2020-09-18T16:12:56Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

@@ -510,6 +511,11 @@ private LLCRealtimeSegmentZKMetadata updateCommittingSegmentZKMetadata(String re
    committingSegmentZKMetadata.setIndexVersion(segmentMetadata.getVersion());
    committingSegmentZKMetadata.setTotalDocs(segmentMetadata.getTotalDocs());

+    // Update the partition metadata based on the segment metadata
+    // NOTE: When the stream partition changes, or the records are not properly partitioned from the stream, the
+    //       partition of the segment can be different from the stream partition.


Suggested change

// partition of the segment can be different from the stream partition.

// partition of the segment is undefined

It is not undefined, but based on the actual consumed records. Updated the comment to reflect that.

mcvsubbu · 2020-09-18T16:14:16Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

@@ -560,7 +566,8 @@ private void createNewSegmentZKMetadata(TableConfig tableConfig, PartitionLevelS
  }

  @Nullable
-  private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(TableConfig tableConfig, int partitionId) {
+  private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(TableConfig tableConfig, int numPartitions,


Suggested change

private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(TableConfig tableConfig, int numPartitions,

private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(TableConfig tableConfig, int numStreamPartitions,

Reverted this part to keep minimum change and keep the current behavior. Will handle this in the next PR which requires some perf testing

mcvsubbu · 2020-09-18T16:15:18Z

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java

@@ -39,6 +39,7 @@
  REALTIME_CONSUMPTION_EXCEPTIONS("exceptions", true),
  REALTIME_OFFSET_COMMITS("commits", true),
  REALTIME_OFFSET_COMMIT_EXCEPTIONS("exceptions", false),
+  REALTIME_PARTITION_MISMATCH("mismatch", false),


Can we add the metric on the controller instead? If it happens for one stream partition, it is highly likely that it will happen to all partitions, so might as well reduce noise

I take that back. We need the metric to fire as soon as we detect a mismatch, so it has to be on the server.

mcvsubbu · 2020-09-18T16:26:35Z

.../src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java

+        // NOTE: Here we compare the number of partitions from the config and the stream, and log a warning and emit a
+        //       metric when they don't match, but use the one from the stream. The mismatch could happen when the
+        //       stream partitions are changed, but the table config has not been updated to reflect the change. In such
+        //       case, picking the number of partitions from the stream can keep the segment properly partitioned as


We don't recognize new partitions instantly. Everytime the realtime validation manager runs, it checks if the number of aprtitions have changed, and if so, starts a new consuming partition.
Let us say at time T1 we checked and the partition number did not change
At time T1 + 10, the partition numbers changed, but we did not know. The stream divided the records into a different partitioning system, thus having mismatched rows in (most likely) all partitions . At time T1 + 50, we check again, and create the new consuming segment for the new partition we detected.
In this case, all the segments that have the mismatched rows should be marked as not belonging to any partition.

I am not sure this condition is being handled

This class in on server side, so it does not rely on the validation manager (which runs on controller side) to detect the partitions change.
I did not change the logic for this class, but only added some notes for readability.

mcvsubbu · 2020-09-18T21:11:32Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+      //       changed, but the table config has not been updated to reflect the change. In such case, picking the
+      //       number of partitions from the stream can keep the segment properly partitioned as long as the partition
+      //       function is not changed.
+      if (columnPartitionConfig.getNumPartitions() != numPartitions) {


For realtime table, maybe we should not have the num partitions in the tableconfig. It can be set to -1, to indicate that it needs to come from the stream.

Good point, or we just don't read it. While we can still use it when the server failed to get the numPartitions from the stream (the existing behavior).

Jackie-Jiang · 2020-09-18T21:40:51Z

@mayankshriv @mcvsubbu Addressed all the comments, and minimize the change in this PR to just bring back the old behavior where we update the partition info when committing the segment.
Will have a separate PR to handle the partition pruning for consuming segment, but that will bring some overhead to the server, so will need some perf test.

mcvsubbu

lgtm other than a minor comment

Jackie-Jiang force-pushed the realtime_partition_id branch from 2d5338a to fcf2b7a Compare September 17, 2020 23:20

Jackie-Jiang requested a review from mayankshriv September 17, 2020 23:21

Jackie-Jiang mentioned this pull request Sep 17, 2020

Partitioning mis-match can lead to bad query results in Realtime #6029

Closed

mayankshriv reviewed Sep 18, 2020

View reviewed changes

mcvsubbu reviewed Sep 18, 2020

View reviewed changes

Jackie-Jiang force-pushed the realtime_partition_id branch from fcf2b7a to 6a2fc33 Compare September 18, 2020 21:18

Handle the partitioning mismatch between table config and stream

750a985

Jackie-Jiang force-pushed the realtime_partition_id branch from 6a2fc33 to 750a985 Compare September 18, 2020 22:04

mcvsubbu approved these changes Sep 18, 2020

View reviewed changes

Jackie-Jiang merged commit 919f407 into apache:master Sep 19, 2020

Jackie-Jiang deleted the realtime_partition_id branch September 19, 2020 18:56

saxenakshitiz mentioned this pull request Oct 27, 2023

[Snyk] Security upgrade axios from 0.27.2 to 1.6.0 hypertrace/incubator-pinot#372

Closed

saxenakshitiz mentioned this pull request Dec 27, 2023

[Snyk] Security upgrade axios from 0.27.2 to 1.6.3 hypertrace/incubator-pinot#382

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Handle the partitioning mismatch between table config and stream #6031

Handle the partitioning mismatch between table config and stream #6031

Jackie-Jiang commented Sep 17, 2020 •

edited

Loading

mayankshriv Sep 18, 2020

Jackie-Jiang Sep 18, 2020

mcvsubbu Sep 18, 2020

Jackie-Jiang Sep 18, 2020

mcvsubbu Sep 18, 2020

Jackie-Jiang Sep 18, 2020

mcvsubbu Sep 18, 2020

mcvsubbu Sep 18, 2020

mcvsubbu Sep 18, 2020

Jackie-Jiang Sep 18, 2020

mcvsubbu Sep 18, 2020

Jackie-Jiang Sep 18, 2020

Jackie-Jiang commented Sep 18, 2020

mcvsubbu left a comment



		/**
		* Integration test that enables segment partition for the LLC real-time table.

	// partition of the segment can be different from the stream partition.
	// partition of the segment is undefined

	private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(TableConfig tableConfig, int numPartitions,
	private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(TableConfig tableConfig, int numStreamPartitions,

Handle the partitioning mismatch between table config and stream #6031

Handle the partitioning mismatch between table config and stream #6031

Conversation

Jackie-Jiang commented Sep 17, 2020 • edited Loading

Description

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang commented Sep 18, 2020

mcvsubbu left a comment

Choose a reason for hiding this comment

Jackie-Jiang commented Sep 17, 2020 •

edited

Loading