InitElement.java

  1. /*
  2.  * Copyright © 2014 - 2021 Leipzig University (Database Research Group)
  3.  *
  4.  * Licensed under the Apache License, Version 2.0 (the "License");
  5.  * you may not use this file except in compliance with the License.
  6.  * You may obtain a copy of the License at
  7.  *
  8.  *     http://www.apache.org/licenses/LICENSE-2.0
  9.  *
  10.  * Unless required by applicable law or agreed to in writing, software
  11.  * distributed under the License is distributed on an "AS IS" BASIS,
  12.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13.  * See the License for the specific language governing permissions and
  14.  * limitations under the License.
  15.  */
  16. package org.gradoop.flink.io.impl.graph.functions;

  17. import org.apache.flink.api.common.typeinfo.TypeInformation;
  18. import org.gradoop.common.model.impl.pojo.EPGMElement;

  19. import java.io.Serializable;

  20. import static com.google.common.base.Preconditions.checkNotNull;

  21. /**
  22.  * Abstract base class for initializing EPGM elements from import elements.
  23.  *
  24.  * @param <EL> EPGM element type
  25.  * @param <K>  Import EPGMEdge/EPGMVertex identifier type
  26.  */
  27. public abstract class InitElement
  28.   <EL extends EPGMElement, K extends Comparable<K>> implements Serializable {

  29.   /**
  30.    * Decides if lineage info (original identifier) shall be stored
  31.    */
  32.   private final boolean keepLineage;

  33.   /**
  34.    * Used to store the lineage information at the resulting EPGM element
  35.    */
  36.   private final String lineagePropertyKey;

  37.   /**
  38.    * Type information for the import element identifier
  39.    */
  40.   private final TypeInformation<K> keyTypeInfo;

  41.   /**
  42.    * Constructor.
  43.    *
  44.    * If the given property key is {@code null}, no lineage info will be stored.
  45.    *
  46.    * @param lineagePropertyKey  property key to store lineage info at EPGM
  47.    *                            element (can be {@code null})
  48.    * @param keyTypeInfo         type info for import element identifier
  49.    */
  50.   public InitElement(String lineagePropertyKey,
  51.     TypeInformation<K> keyTypeInfo) {
  52.     this.keepLineage        = lineagePropertyKey != null;
  53.     this.lineagePropertyKey = lineagePropertyKey;
  54.     this.keyTypeInfo        = checkNotNull(keyTypeInfo, "key type was null");
  55.   }

  56.   /**
  57.    * Updates the lineage info of the given EPGM element if necessary.
  58.    *
  59.    * @param element   EPGM element
  60.    * @param importKey import element identifier
  61.    * @return updated EPGM element
  62.    */
  63.   protected EL updateLineage(EL element, K importKey) {
  64.     if (keepLineage) {
  65.       element.setProperty(lineagePropertyKey, importKey);
  66.     }
  67.     return element;
  68.   }

  69.   /**
  70.    * Returns type information for the import element identifier.
  71.    *
  72.    * @return import element identifier type info
  73.    */
  74.   protected TypeInformation<K> getKeyTypeInfo() {
  75.     return keyTypeInfo;
  76.   }
  77. }