[HUDI-1936] Introduce a optional property for conditional upsert

hudi-common/src/main/java/org/apache/hudi/common/model/OverwriteWithCustomAvroPayload.java

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hudi.common.model;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.ColumnNotFoundException;
+import org.apache.hudi.exception.UpdateKeyNotFoundException;
+import org.apache.hudi.exception.WriteOperationException;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+/**
+ * subclass of OverwriteWithLatestAvroPayload used for delta streamer.
+ *
+ * <ol>
+ * <li> combineAndGetUpdateValue - Accepts the column names to be updated;
+ * <li> splitKeys - Split keys based upon keys;
+ * </ol>
+ */
+public class OverwriteWithCustomAvroPayload extends OverwriteWithLatestAvroPayload {
+
+  public OverwriteWithCustomAvroPayload(GenericRecord record, Comparable orderingVal) {
+    super(record, orderingVal);
+  }
+
+  /**
+   * split keys over.
+   */
+  public List<String> splitKeys(String keys) throws UpdateKeyNotFoundException {
+    if (keys == null) {
+      throw new UpdateKeyNotFoundException("keys cannot be null");
+    } else if (keys.equals("")) {
+      throw new UpdateKeyNotFoundException("keys cannot be blank");
+    } else {
+      return Arrays.stream(keys.split(",")).collect(Collectors.toList());
+    }
+  }
+
+  /**
+   * check column exi.
+   */
+  public boolean checkColumnExists(List<String> keys, Schema schema) {
+    List<Schema.Field> field = schema.getFields();
+    List<Schema.Field> common = new ArrayList<>();
+    for (Schema.Field columns : field) {
+      if (keys.contains(columns.name())) {
+        common.add(columns);
+      }
+    }
+    return common.size() == keys.size();
+  }
+
+  @Override
+  public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Properties properties)
+      throws WriteOperationException, IOException, ColumnNotFoundException, UpdateKeyNotFoundException {
+
+    if (!properties.getProperty("hoodie.datasource.write.operation").equals("upsert")) {
+      throw new WriteOperationException("write should be upsert");
+    }
+
+    Option<IndexedRecord> recordOption = getInsertValue(schema);
+
+    if (!recordOption.isPresent()) {
+      return Option.empty();
+    }
+
+    GenericRecord existingRecord = (GenericRecord) currentValue;
+    GenericRecord incomingRecord = (GenericRecord) recordOption.get();
+    List<String> keys = splitKeys(properties.getProperty("hoodie.update.keys"));
+
+    if (checkColumnExists(keys, schema)) {
+      for (String key : keys) {
+        Object value = incomingRecord.get(key);
+        existingRecord.put(key, value);
+      }
+      return Option.of(existingRecord);
+    } else {
+      throw new ColumnNotFoundException("Update key not present please check the names");
+    }
+  }
+
+}

hudi-common/src/main/java/org/apache/hudi/exception/ColumnNotFoundException.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hudi.exception;
+
+public class ColumnNotFoundException extends HoodieIOException {
+  public ColumnNotFoundException(String message) {
+    super(message);
+  }
+}

hudi-common/src/main/java/org/apache/hudi/exception/UpdateKeyNotFoundException.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hudi.exception;
+
+public class UpdateKeyNotFoundException extends HoodieIOException {
+  public UpdateKeyNotFoundException(String message) {
+    super(message);
+  }
+}

hudi-common/src/main/java/org/apache/hudi/exception/WriteOperationException.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hudi.exception;
+
+public class WriteOperationException extends HoodieException {
+  public WriteOperationException(String message) {
+    super(message);
+  }
+}

hudi-common/src/test/java/org/apache/hudi/common/model/TestOverwriteWithCustomAvroPayload.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hudi.common.model;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Unit tests {@link OverwriteWithCustomAvroPayload}.
+ */
+public class TestOverwriteWithCustomAvroPayload {
+
+  private Schema schema;
+
+  @BeforeEach
+  public void setUp() throws Exception {
+    schema = Schema.createRecord(Arrays.asList(
+        new Schema.Field("id", Schema.create(Schema.Type.STRING), "", null),
+        new Schema.Field("partition", Schema.create(Schema.Type.STRING), "", null),
+        new Schema.Field("ts", Schema.create(Schema.Type.LONG), "", null),
+        new Schema.Field("_hoodie_is_deleted", Schema.create(Schema.Type.BOOLEAN), "", false)
+    ));
+  }
+
+  @Test
+  public void testsplitKeys() throws IOException {
+    GenericRecord record1 = new GenericData.Record(schema);
+    record1.put("id", "1");
+    record1.put("partition", "partition0");
+    record1.put("ts", 0L);
+    record1.put("_hoodie_is_deleted", false);
+    OverwriteWithCustomAvroPayload payload1 = new OverwriteWithCustomAvroPayload(record1, 1);
+    assertEquals(payload1.splitKeys("id,ts").size(), Arrays.asList("id", "ts").size());
+    assertEquals(payload1.splitKeys("id,ts"), Arrays.asList("id", "ts"));
+  }
+
+}
+

hudi-sync/hudi-hive-sync/pom.xml

@@ -99,7 +99,7 @@
       <artifactId>hadoop-hdfs</artifactId>
       <classifier>tests</classifier>
     </dependency>
-
+    <!-- https://mvnrepository.com/artifact/org.pentaho/pentaho-aggdesigner-algorithm -->
     <!-- Hive -->
     <dependency>
       <groupId>${hive.groupid}</groupId>

pom.xml

@@ -1017,6 +1017,10 @@
         <enabled>false</enabled>
       </snapshots>
     </repository>
+    <repository>
+      <id>pentaho.org</id>
+      <url>https://public.nexus.pentaho.org/repository/proxy-public-3rd-party-release/</url>
+    </repository>
     <repository>
       <id>cloudera-repo-releases</id>
       <url>https://repository.cloudera.com/artifactory/public/</url>