CDAP-16708 add API for autojoiner

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/batch/BatchAutoJoiner.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.etl.api.batch;
+
+import io.cdap.cdap.api.annotation.Beta;
+import io.cdap.cdap.etl.api.MultiInputPipelineConfigurable;
+import io.cdap.cdap.etl.api.MultiInputPipelineConfigurer;
+import io.cdap.cdap.etl.api.join.AutoJoiner;
+
+/**
+ * Joins input data, leaving implementation details up to the application.
+ */
+@Beta
+public abstract class BatchAutoJoiner extends MultiInputBatchConfigurable<BatchJoinerContext>
+  implements AutoJoiner, MultiInputPipelineConfigurable {
+  public static final String PLUGIN_TYPE = BatchJoiner.PLUGIN_TYPE;
+
+  /**
+   * Configure the pipeline. This is run once when the pipeline is being published.
+   * This is where you perform any static logic, like creating required datasets, performing schema validation,
+   * setting output schema, and things of that nature.
+   *
+   * @param multiInputPipelineConfigurer the configurer used to add required datasets and streams
+   */
+  @Override
+  public void configurePipeline(MultiInputPipelineConfigurer multiInputPipelineConfigurer) {
+    // no-op
+  }
+
+  /**
+   * Prepare a pipeline run. This is run every time before a pipeline runs in order to help set up the run.
+   * This is where you would set things like the number of partitions to use when joining, and setting the
+   * join key class if they are not known at compile time.
+   *
+   * @param context batch execution context
+   * @throws Exception
+   */
+  @Override
+  public void prepareRun(BatchJoinerContext context) throws Exception {
+    //no-op
+  }
+}

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/batch/BatchJoiner.java

@@ -35,7 +35,6 @@
  * If the joiner is being used in spark, both the join key and input record must implement the
  * {@link java.io.Serializable} interface.
  *
- *
  * @param <JOIN_KEY> type of join key. Must be a supported type
  * @param <INPUT_RECORD> type of input record. Must be a supported type
  * @param <OUT> type of output object

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/join/AutoJoiner.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.etl.api.join;
+
+import io.cdap.cdap.api.annotation.Beta;
+
+import javax.annotation.Nullable;
+
+/**
+ * Joins data from two or more input stages together.
+ *
+ * For example, the following represents a left outer join on 'purchases'.'user_id' = 'users'.'id':
+ *
+ * <pre>
+ *   {@code
+ *     JoinDefinition define(AutoJoinerContext context) {
+ *       JoinStage purchases = JoinStage.builder(context.getInputStages().get("purchases"))
+ *         .isRequired()
+ *         .build();
+ *       JoinStage users = JoinStage.builder(context.getInputStages().get("users"))
+ *         .isOptional()
+ *         .build();
+ *
+ *       JoinCondition condition = JoinCondition.onKeys()
+ *         .addKey(new JoinKey("users", Arrays.asList("id")))
+ *         .addKey(new JoinKey("purchases", Arrays.asList("user_id")))
+ *         .build();
+ *
+ *       return JoinDefinition.builder()
+ *         // stage name, field name, optional alias
+ *         .select(new Field("purchases", "id", "purchase_id"),
+ *                 new Field("purchases", "user_id"),
+ *                 new Field("users", "email"))
+ *         .from(purchases, users)
+ *         .on(condition)
+ *         .build()
+ *     }
+ *   }
+ * </pre>
+ */
+@Beta
+public interface AutoJoiner {
+
+  /**
+   * @return definition about what type of join to execute, or null if the definition cannot be created due to
+   *   macro values not being evaluated yet.
+   */
+  @Nullable
+  JoinDefinition define(AutoJoinerContext context);
+
+}

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/join/AutoJoinerContext.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.etl.api.join;
+
+import java.util.Map;
+
+/**
+ * Context for an AutoJoin.
+ */
+public interface AutoJoinerContext {
+
+  /**
+   * @return all input stages
+   */
+  Map<String, JoinStage> getInputStages();
+}

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/join/Field.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.etl.api.join;
+
+import io.cdap.cdap.api.annotation.Beta;
+
+import java.util.Objects;
+import javax.annotation.Nullable;
+
+/**
+ * The name of a field and an optional alias to rename it to.
+ */
+@Beta
+public class Field {
+  private final String stageName;
+  private final String fieldName;
+  private final String alias;
+
+  public Field(String stageName, String fieldName) {
+    this(stageName, fieldName, null);
+  }
+
+  public Field(String stageName, String fieldName, @Nullable String alias) {
+    this.stageName = stageName;
+    this.fieldName = fieldName;
+    this.alias = alias;
+  }
+
+  public String getStageName() {
+    return stageName;
+  }
+
+  public String getFieldName() {
+    return fieldName;
+  }
+
+  @Nullable
+  public String getAlias() {
+    return alias;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    Field field1 = (Field) o;
+    return Objects.equals(stageName, field1.stageName) &&
+      Objects.equals(fieldName, field1.fieldName) &&
+      Objects.equals(alias, field1.alias);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(stageName, fieldName, alias);
+  }
+
+  @Override
+  public String toString() {
+    return "Field{" +
+      "stage='" + stageName + '\'' +
+      ", field='" + fieldName + '\'' +
+      ", alias='" + alias + '\'' +
+      '}';
+  }
+}

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/join/InvalidJoinConditionException.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.etl.api.join;
+
+/**
+ * Thrown when the join condition is invalid.
+ */
+public class InvalidJoinConditionException extends InvalidJoinException {
+
+  public InvalidJoinConditionException(String message) {
+    super(message);
+  }
+
+  public InvalidJoinConditionException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

cdap-app-templates/cdap-etl/cdap-etl-api/src/main/java/io/cdap/cdap/etl/api/join/InvalidJoinException.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.etl.api.join;
+
+/**
+ * Thrown when a join definition is invalid.
+ */
+public class InvalidJoinException extends RuntimeException {
+
+  public InvalidJoinException(String message) {
+    super(message);
+  }
+
+  public InvalidJoinException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}