code: static Map<String, Set<LineageEventBuilder>> load(State state) {
      String name = state.getProp(getKey(NAME_KEY));
      Descriptor source = Descriptor.fromJson(state.getProp(getKey(LineageEventBuilder.SOURCE)));
  
      String branchedPrefix = getKey(BRANCH, "");
      Map<String, Set<LineageEventBuilder>> events = Maps.newHashMap();
      if (source == null) {
        return events;
      }
  
      for (Map.Entry<Object, Object> entry : state.getProperties().entrySet()) {
        String key = entry.getKey().toString();
        if (!key.startsWith(branchedPrefix)) {
          continue;
        }
  
        String[] parts = key.substring(branchedPrefix.length()).split("\\.");
        assert parts.length == 2;
        String branchId = parts[0];
        Set<LineageEventBuilder> branchEvents = events.computeIfAbsent(branchId, k -> new HashSet<>());
  
        switch (parts[1]) {
          case LineageEventBuilder.DESTINATION:
            List<Descriptor> descriptors = Descriptor.fromJsonList(entry.getValue().toString());
            for (Descriptor descriptor : descriptors) {
              LineageEventBuilder event = new LineageEventBuilder(name);
              event.setSource(source);
              event.setDestination(descriptor);
              branchEvents.add(event);
            }
            break;
          default:
            throw new RuntimeException("Unsupported lineage key: " + key);
        }
      }
  
      return events;
    }
docs: Load all lineage info from a {@link State}
  
  @return A map from branch to its lineage info. If there is no destination info, return an empty map