Make ConnectorAwareNodeManager#getWorkerNodes not include coordinator

[sopel39]

No description provided.

[erichwang]

Wait, the original code is operating as expected. Don't check this in yet.

The filtering for the coordinator should be handled here:
https://github.com/trinodb/trino/blob/master/core/trino-main/src/main/java/io/trino/execution/scheduler/TopologyAwareNodeSelectorFactory.java#L150

And all throughout this class:
https://github.com/trinodb/trino/blob/master/core/trino-main/src/main/java/io/trino/execution/scheduler/NodeScheduler.java#L49

If you put this check in, it will invalidate the effectiveness of the node-scheduler.include-coordinator config option.

If one of those original handling sites are not working properly, we should see if we can fix those rather than introduce a new filtering location.

[sopel39]

@erichwang ConnectorAwareNodeManager is used before scheduling happens. It can be used by connectors to generate splits (which are later scheduled) and assign them to nodes. ConnectorAwareNodeManager wasn't honoring node-scheduler.include-coordinator at all and returned coordinator as part of getWorkersNodes result set.

[erichwang]

Oh you are right, I was looking at the other flow.

But this does raise an important question on how this was originally being used. For example:

• trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/rubix/TrinoClusterManager.java

  Line 52 in f6422d0

  .filter(node -> !node.isCoordinator())

  expects the output of getWorkers() to include coordinators since it is filtering it out.

• trino/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopSplitManager.java

  Line 70 in 03d8e5a

  for (Node node : nodeManager.getWorkerNodes()) {

  This changes the behavior of the atop connector which was previously executing on all nodes (but it should probably be using the allNodes() instead).
• Was this code functioning correctly?

  trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveNodePartitioningProvider.java

  Line 86 in 930b0b0

  return createBucketNodeMap(nodeManager.getRequiredWorkerNodes().size() * PARTITIONED_BUCKETS_PER_NODE);

  If we were always including the coordinator here, the bucket calculation should be wrong if include-coordinator=false (which it is in most prod cases).

[sopel39]

@erichwang

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/rubix/TrinoClusterManager.java

Line 52 in f6422d0

.filter(node -> !node.isCoordinator())

That one I wrote myself and yet I did skipped the problem then somehow ;). TrinoClusterManager requires nodes list without coordinator so this change is ok. In fact, in newer Rubix code it no longer filters out coordinator (assuming that getWorkerNodes works correctly, but it doesn't)

trino/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopSplitManager.java

Line 70 in 03d8e5a

for (Node node : nodeManager.getWorkerNodes()) {

Judging by commit message that introduced getWorkerNodes (previously it was getAllNodes():

Fix atop connector to work with dedicated coordinator  Currently, if node-scheduler.include-coordinator=false, a split is still generated for the coordinator, which will then not be able to process it, failing the query.

It seems that we don't want to return coordinator as part of getWorkerNodes() when node-scheduler.include-coordinator=false

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveNodePartitioningProvider.java

Line 86 in 930b0b0

return createBucketNodeMap(nodeManager.getRequiredWorkerNodes().size() * PARTITIONED_BUCKETS_PER_NODE);

This code was introduced recently and it works correctly. It generates arbitrary number (large enough) for bucket count. Number should be large enough to ensure parallelism on all worker nodes during insert.

[erichwang]

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/rubix/TrinoClusterManager.java

Line 52 in f6422d0

.filter(node -> !node.isCoordinator())

That one I wrote myself and yet I did skipped the problem then somehow ;). TrinoClusterManager requires nodes list without coordinator so this change is ok. In fact, in newer Rubix code it no longer filters out coordinator (assuming that getWorkerNodes works correctly, but it doesn't)

In that case, we probably want to follow up afterwards to remove the coordinator filter so that it doesn't confuse future readers.

trino/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopSplitManager.java

Line 70 in 03d8e5a

for (Node node : nodeManager.getWorkerNodes()) {

Judging by commit message that introduced getWorkerNodes (previously it was getAllNodes():

Fix atop connector to work with dedicated coordinator  Currently, if node-scheduler.include-coordinator=false, a split is still generated for the coordinator, which will then not be able to process it, failing the query.

It seems that we don't want to return coordinator as part of getWorkerNodes() when node-scheduler.include-coordinator=false

I see the Atop connector as similar to the Jmx connector, which is designed to return stats about the nodes themselves. As a consumer for this type of data, I would definitely prefer to see the coordinator included in the output as well in those cases (barring any technical challenges). Also note, the Jmx connector reaches out explicitly to allNodes when generating splits. Anyways, this is just my preference for what I believe to be useful, but the more important thing is to be consistent.

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveNodePartitioningProvider.java

Line 86 in 930b0b0

return createBucketNodeMap(nodeManager.getRequiredWorkerNodes().size() * PARTITIONED_BUCKETS_PER_NODE);

This code was introduced recently and it works correctly. It generates arbitrary number (large enough) for bucket count. Number should be large enough to ensure parallelism on all worker nodes during insert.

I see, it's not actually used for strict bucket calculation.

[erichwang]

See the previous comments I made, but I don't believe any of those are blocking these changes.

[sopel39]

In that case, we probably want to follow up afterwards to remove the coordinator filter so that it doesn't confuse future readers.

@erichwang it's removed, see https://github.com/qubole/rubix/blob/3149b6385f6685f5fe934551126b6593f59da9c8/rubix-prestosql/src/main/java/com/qubole/rubix/prestosql/ClusterManagerNodeGetter.java#L29