You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@phoenix.apache.org by "rahulLiving (via GitHub)" <gi...@apache.org> on 2026/04/16 13:56:23 UTC
[PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
rahulLiving opened a new pull request, #2411:
URL: https://github.com/apache/phoenix/pull/2411
(no comment)
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3138392840
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
It is only retrying on exception case (not on null value), besides, the retry doesn't happen with reload=true either.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#issuecomment-4313089779
>If you mean, you want to address it in the very next PR, I am fine with it, but I don't think this should be considered as optional.
yes as part of very next PR, we have couple of more PR incoming for repair phase.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3152606018
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
How about to maintain a UNKNOWN_SERVER split for all such occurence of locating region fails. And this could be assigned as one mapper ?
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "tkhurana (via GitHub)" <gi...@apache.org>.
tkhurana commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3113419216
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputSplit.java:
##########
@@ -39,6 +39,7 @@ public class PhoenixInputSplit extends InputSplit implements Writable {
private List<Scan> scans;
private KeyRange keyRange;
+ private List<KeyRange> keyRanges;
Review Comment:
keyRange is redundant now. It's just the span from `keyRanges.get(0).getLowerRange()` to `keyRanges.get(last).getUpperRange()`. You can keep only keyRanges and derive the overall bounds on demand:
```suggestion
public KeyRange getKeyRange() {
return KeyRange.getKeyRange(
keyRanges.get(0).getLowerRange(),
keyRanges.get(keyRanges.size() - 1).getUpperRange());
}
```
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "tkhurana (via GitHub)" <gi...@apache.org>.
tkhurana commented on PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#issuecomment-4284553955
With coalescing each map task will span multiple table regions which means that each mapper will run longer. We could run the risk of map task timing out and getting killed. To get around that we might need to set a higher timeout value.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#issuecomment-4297860974
>With coalescing each map task will span multiple table regions which means that each mapper will run longer. We could run the risk of map task timing out and getting killed. To get around that we might need to set a higher timeout value.
We have `phoenix.sync.table.query.timeout` config to override MRJobConfig.TASK_TIMEOUT in case we are coalescing. Does that sounds good enough ?
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "tkhurana (via GitHub)" <gi...@apache.org>.
tkhurana commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3113938212
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableMapper.java:
##########
@@ -88,15 +87,15 @@ public enum SyncCounters {
private Connection globalConnection;
private PTable pTable;
private byte[] physicalTableName;
- private byte[] mapperRegionStart;
- private byte[] mapperRegionEnd;
+ private List<KeyRange> regionKeyRanges;
private PhoenixSyncTableOutputRepository syncTableOutputRepository;
private Timestamp mapperStartTime;
@Override
protected void setup(Context context) throws InterruptedException {
try {
super.setup(context);
+ context.getCounter(SyncCounters.TASK_CREATED).increment(1);
Review Comment:
Isn't this just the number of mappers created ? YARN already maintains that information. We can simply use the corresponding counter from YARN rather than maintaining our own.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#issuecomment-4312521700
> I saw progress being reported incrementally from 0->100. I think progress is being reported based on how many mapper pending/in-progress/completed. **But you are right that, any mapper which is running will be considered as 0/1 for job completion. There won't be a middle ground.**
This (the highlighted) is exactly what I am saying! Also, we wouldn't know in advance how many bytes would be encountered and so there is noe meaningful way to give a percentage based on bytes processed, but this simplification was fine before this change, but not anymore.
> I see the formulae at mapper level, Progress = Number of bytes read/Total bytes in Input Split.
> Will have to see if we can update bytes read manually in RecordReader.
My point is, we know how many splits will be processed by the mapper so we can easily generate a progress based on the number of splits completed.
> Do you think we should take this up as follow up ?
If you mean, you want to address it in the very next PR, I am fine with it, but I don't think this should be considered as optional.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3152490862
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
I was thinking if we should integrate retry as part of `ConnectionQueryServicesImpl.getTableRegionLocation` itself with reload=true, I see there are 4 callers for same ?
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3099675545
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
I am thinking that this should never see a `null` if all regions are fully online and client cache is up to date. However (per what AI says), if the region the key maps to is currently offline due to a RIT event (say, in the middle of a split), then the return value will be null, so shouldn't we retry to ride over such RITs, just like how hbase-client does during data path such as scans? Otherwise, there is a high chance of hitting a RIT and treating it as an error, correct?
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#issuecomment-4280088881
>Currently, the mapper doesn't report meaningful progress during execution, it would effectively shows 0% until completion, then jump to 100%, isn't it?
No that is not the case currently. From my past execution, I saw progress being reported incrementally from 0->100. I think progress is being reported based on how many mapper pending/in-progress/completed. But you are right that, any mapper which is running will be considered as 0/1 for job completion. There won't be a middle ground.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "tkhurana (via GitHub)" <gi...@apache.org>.
tkhurana commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3113387227
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -94,15 +101,36 @@ public List<InputSplit> getSplits(JobContext context) throws IOException, Interr
} catch (SQLException e) {
throw new RuntimeException(e);
}
- if (completedRegions.isEmpty()) {
- LOGGER.info("No completed regions for table {} - processing all {} splits", tableName,
- allSplits.size());
- return allSplits;
- }
List<InputSplit> unprocessedSplits = filterCompletedSplits(allSplits, completedRegions);
LOGGER.info("Found {} completed mapper regions for table {}, {} unprocessed splits remaining",
completedRegions.size(), tableName, unprocessedSplits.size());
+
+ boolean enableSplitCoalescing =
+ conf.getBoolean(PhoenixSyncTableTool.PHOENIX_SYNC_TABLE_SPLIT_COALESCING,
+ PhoenixSyncTableTool.DEFAULT_PHOENIX_SYNC_TABLE_SPLIT_COALESCING);
+ LOGGER.info("Split coalescing enabled: {}, for table {}", enableSplitCoalescing, tableName);
+
+ if (enableSplitCoalescing && unprocessedSplits.size() > 1) {
+ try (Connection conn = ConnectionUtil.getInputConnection(conf)) {
+ PhoenixConnection pConn = conn.unwrap(PhoenixConnection.class);
+ byte[] physicalTableName = pConn.getTable(tableName).getPhysicalName().getBytes();
+ org.apache.hadoop.hbase.client.Connection hbaseConn =
+ pConn.getQueryServices().getAdmin().getConnection();
Review Comment:
There is a resource leak here. Admin object is not being closed. Also hconnection is also not being closed. We already have similar APIs available to us in ConnectionQueryServicesImpl class. My suggestion would be to use `getTableRegionLocation`
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3153262155
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
Yes, grouping them together actually makes sense to me and any RIT recovery would be handled by hbase-client as part of mapper execution itself and help improve the overall tool resiliency.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3137621225
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
I see with @tkhurana suggestion to use [ConnectionQueryServiceImpl.getTableRegionLocation](https://github.com/apache/phoenix/blob/master/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java#L6773), we will have one retry by default
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3135716155
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
It is possible that the split generation logic ended up caching the metadata with some of the regions in transition so when we try to locate them, it would end up using the same cached metadata and fail to lookup. If we introduce at least 1 retry with a metadata reload, it can help avoid the IOException in majority of the scenarios. WDYT?
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv merged PR #2411:
URL: https://github.com/apache/phoenix/pull/2411
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7799 Coalesce splits by region server to avoid hotspotting from concurrent mappers [phoenix]
Posted by "rahulLiving (via GitHub)" <gi...@apache.org>.
rahulLiving commented on code in PR #2411:
URL: https://github.com/apache/phoenix/pull/2411#discussion_r3110903395
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableInputFormat.java:
##########
@@ -211,4 +242,93 @@ List<InputSplit> filterCompletedSplits(List<InputSplit> allSplits,
}
return unprocessedSplits;
}
+
+ /**
+ * Coalesces multiple region splits from the same RegionServer into single InputSplits. All
+ * regions from the same server are coalesced into one split, regardless of count or size. This
+ * reduces mapper count and avoids hot spotting when many concurrent mappers hit the same server.
+ * @param unprocessedSplits Splits remaining after filtering completed regions
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Coalesced splits with all regions per server combined into one split
+ */
+ List<InputSplit> coalesceSplits(List<InputSplit> unprocessedSplits, RegionLocator regionLocator)
+ throws IOException, InterruptedException {
+ // Group splits by RegionServer location
+ Map<String, List<PhoenixInputSplit>> splitsByServer =
+ groupSplitsByServer(unprocessedSplits, regionLocator);
+
+ List<InputSplit> coalescedSplits = new ArrayList<>();
+
+ // For each RegionServer, create one coalesced split with ALL regions from that server
+ for (Map.Entry<String, List<PhoenixInputSplit>> entry : splitsByServer.entrySet()) {
+ String serverName = entry.getKey();
+ List<PhoenixInputSplit> serverSplits = entry.getValue();
+
+ // Sort splits by start key for sequential processing
+ serverSplits.sort((s1, s2) -> Bytes.compareTo(s1.getKeyRange().getLowerRange(),
+ s2.getKeyRange().getLowerRange()));
+ // Create single coalesced split with ALL regions from this server
+ coalescedSplits.add(createCoalescedSplit(serverSplits, serverName));
+ }
+
+ return coalescedSplits;
+ }
+
+ /**
+ * Groups splits by RegionServer location for locality-aware coalescing. Uses HBase RegionLocator
+ * API to determine which server hosts each region.
+ * @param splits List of splits to group
+ * @param regionLocator HBase RegionLocator for querying region locations
+ * @return Map of server name to list of splits hosted on that server
+ */
+ private Map<String, List<PhoenixInputSplit>> groupSplitsByServer(List<InputSplit> splits,
+ RegionLocator regionLocator) throws IOException {
+ Map<String, List<PhoenixInputSplit>> splitsByServer = new HashMap<>();
+ for (InputSplit split : splits) {
+ PhoenixInputSplit pSplit = (PhoenixInputSplit) split;
+ KeyRange keyRange = pSplit.getKeyRange();
+ HRegionLocation regionLocation =
+ regionLocator.getRegionLocation(keyRange.getLowerRange(), false);
+ if (regionLocation == null) {
+ throw new IOException("Could not determine region location for key: "
+ + Bytes.toStringBinary(keyRange.getLowerRange()));
+ }
Review Comment:
Even during RIT, I believe meta would return cached location which might be outdated. It could give a false colocation but that is expected for some of regions as sync tool runs.
--
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: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org