perf: address scheduler crash on high partition SF data

[coderfender]

Which issue does this PR close?

Closes #1851

Rationale for this change

We currently have more than what is needed in PartitionMetadata causing scheduler OOM on high cardinality data

What changes are included in this PR?

Implement PartitionLocation a lean, address-only struct instead of the full metadata:

  #[derive(Debug, Clone, PartialEq, Eq)]
  pub struct PartitionLocationMetadata {
      pub id: String,
      pub host: String,
      pub port: u16,
      pub grpc_port: u16,
  }

  pub struct PartitionLocation {
      ...
      pub executor_meta: Arc<PartitionLocationMetadata>,
  }

• New From<&ExecutorMetadata> for PartitionLocationMetadata is the single place
  that maps an executor to its location address.
• Arc-wrapped so all locations for the same executor (built in one
  partition_to_location call) share one allocation rather than N copies.
• Proto PartitionLocation now carries only the four address fields, shrinking
  the serialized payload sent to executors/clients. (no contract changes per say)

Are there any user-facing changes?

No

No

[coderfender]

@milenkovicm please take a look whenever you get a chance

[milenkovicm]

thanks @coderfender will have a look

[milenkovicm]

thanks @coderfender, i guess we have two goals with this effort

• reduce number of executor metadata instances keept in memory (we do not need to have exact one instance, but it would ne good not to have instance per partition location)
• reduce serialised payload as we're hitting gprc threshoulds with many partitions

[milenkovicm]

my idea was to have single instance of partition location and then Arc-ed behind shared pointers. At the moment, we have instance of executor metadata per partition location which for scenarios where we have just a few executor and many partitions may produce a lot of data (id, host ...). also when serialised we would serialize executor metadata for each of partition, which is hitting gprc issues. not sure if we can somehow reduce executor metadata instance count and its serialization

[coderfender]

Interesting

[coderfender]

Would it be okay to create a followup PR to handle the second part and port all common info in ShuffleReaderExec proto level and not PartitionLocation ?

[milenkovicm]

i think #1853 does that already if im not mistaken

[coderfender]

Yeah we had a chat in the original PR . I am okay to merge that PR ahead if that makes more sense this one. Although I think mine is a little bit different in terms of code paths changes

[coderfender]

Closing this PR in favor of #1853