Add support for update by id

Author: t3t5u

README.md

@@ -26,6 +26,7 @@ kintone output plugin for Embulk stores app records from kintone.
 - **max_sort_memory**: Maximum memory usage for sorting input records (bytes in long, default is the estimated available memory, which is the approximate value of the JVM's current free memory)
 - **prefer_nulls**: Whether to set fields to null instead of default value of type when column is null (boolean, default is `false`)
 - **ignore_nulls**: Whether to completely ignore fields when column is null (boolean, default is `false`)
+- **skip_if_non_existing_id_or_update_key**: Skip policy if id or update key of record does not exist (string `auto`, `never` or `always`, default is `auto`)
 - **column_options** advanced: a key-value pairs where key is a column name and value is options for the column.
     - **field_code**: field code (string, required)
     - **type**: field type (string, required). See [this page](https://cybozu.dev/ja/kintone/docs/overview/field-types/#field-type-update) for list of available types. However, following types are not yet supported

src/main/java/org/embulk/output/kintone/KintoneClient.java

@@ -8,9 +8,12 @@
 import java.util.Map;
 import java.util.function.Supplier;
 import org.embulk.config.ConfigException;
+import org.embulk.output.kintone.record.Id;
 import org.embulk.output.kintone.util.Lazy;
 import org.embulk.spi.Column;
 import org.embulk.spi.Schema;
+import org.embulk.spi.type.Type;
+import org.embulk.spi.type.Types;
 
 public class KintoneClient implements AutoCloseable {
   private final PluginTask task;
@@ -49,19 +52,13 @@ private KintoneClient(PluginTask task, Schema schema) {
     KintoneMode.of(task).validate(task, this);
   }
 
-  public void validateUpdateKey(String columnName) {
+  public void validateIdOrUpdateKey(String columnName) {
     Column column = getColumn(columnName);
     if (column == null) {
       throw new ConfigException("The column '" + columnName + "' for update does not exist.");
     }
-    String fieldCode = getFieldCode(column);
-    FieldType fieldType = getFieldType(fieldCode);
-    if (fieldType == null) {
-      throw new ConfigException("The field '" + fieldCode + "' for update does not exist.");
-    }
-    if (fieldType != FieldType.SINGLE_LINE_TEXT && fieldType != FieldType.NUMBER) {
-      throw new ConfigException("The update_key must be 'SINGLE_LINE_TEXT' or 'NUMBER'.");
-    }
+    validateId(column);
+    validateUpdateKey(column);
   }
 
   public Column getColumn(String columnName) {
@@ -89,6 +86,30 @@ public void close() {
     }
   }
 
+  private void validateId(Column column) {
+    if (!column.getName().equals(Id.FIELD)) {
+      return;
+    }
+    Type type = column.getType();
+    if (!type.equals(Types.LONG)) {
+      throw new ConfigException("The id column must be 'long'.");
+    }
+  }
+
+  private void validateUpdateKey(Column column) {
+    if (column.getName().equals(Id.FIELD)) {
+      return;
+    }
+    String fieldCode = getFieldCode(column);
+    FieldType fieldType = getFieldType(fieldCode);
+    if (fieldType == null) {
+      throw new ConfigException("The field '" + fieldCode + "' for update does not exist.");
+    }
+    if (fieldType != FieldType.SINGLE_LINE_TEXT && fieldType != FieldType.NUMBER) {
+      throw new ConfigException("The update_key must be 'SINGLE_LINE_TEXT' or 'NUMBER'.");
+    }
+  }
+
   private String getFieldCode(Column column) {
     KintoneColumnOption option = task.getColumnOptions().get(column.getName());
     return option != null ? option.getFieldCode() : column.getName();

src/main/java/org/embulk/output/kintone/KintoneColumnVisitor.java

@@ -4,8 +4,12 @@
 import com.kintone.client.model.record.Record;
 import com.kintone.client.model.record.UpdateKey;
 import java.time.Instant;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.embulk.output.kintone.record.Id;
+import org.embulk.output.kintone.record.IdOrUpdateKey;
 import org.embulk.spi.Column;
 import org.embulk.spi.ColumnVisitor;
 import org.embulk.spi.PageReader;
@@ -14,6 +18,7 @@
 import org.msgpack.value.ValueFactory;
 
 public class KintoneColumnVisitor implements ColumnVisitor {
+  private static final List<String> BUILTIN_FIELD_CODES = Arrays.asList(Id.FIELD, "$revision");
   private final PageReader reader;
   private final Set<Column> derived;
   private final Map<String, KintoneColumnOption> options;
@@ -22,7 +27,7 @@ public class KintoneColumnVisitor implements ColumnVisitor {
   private final String reduceKeyName;
   private final String updateKeyName;
   private Record record;
-  private UpdateKey updateKey;
+  private IdOrUpdateKey idOrUpdateKey;
 
   public KintoneColumnVisitor(
       PageReader reader,
@@ -55,8 +60,8 @@ public void setRecord(Record record) {
     this.record = record;
   }
 
-  public void setUpdateKey(UpdateKey updateKey) {
-    this.updateKey = updateKey;
+  public void setIdOrUpdateKey(IdOrUpdateKey idOrUpdateKey) {
+    this.idOrUpdateKey = idOrUpdateKey;
   }
 
   @Override
@@ -68,6 +73,9 @@ public void booleanColumn(Column column) {
     UpdateKey updateKey = getUpdateKey(column);
     KintoneColumnType type = KintoneColumnType.getType(option, KintoneColumnType.NUMBER);
     String fieldCode = getFieldCode(column);
+    if (isBuiltin(fieldCode)) {
+      return;
+    }
     if (isPreferNull(column)) {
       setNull(type, fieldCode, updateKey);
     } else if (reader.isNull(column)) {
@@ -82,10 +90,18 @@ public void longColumn(Column column) {
     if (isReduced(column) || isIgnoreNull(column)) {
       return;
     }
+    Id id = getId(column);
+    if (id != null) {
+      setIdValue(id, column);
+      return;
+    }
     KintoneColumnOption option = getOption(column);
     UpdateKey updateKey = getUpdateKey(column);
     KintoneColumnType type = KintoneColumnType.getType(option, KintoneColumnType.NUMBER);
     String fieldCode = getFieldCode(column);
+    if (isBuiltin(fieldCode)) {
+      return;
+    }
     if (isPreferNull(column)) {
       setNull(type, fieldCode, updateKey);
     } else if (reader.isNull(column)) {
@@ -104,6 +120,9 @@ public void doubleColumn(Column column) {
     UpdateKey updateKey = getUpdateKey(column);
     KintoneColumnType type = KintoneColumnType.getType(option, KintoneColumnType.NUMBER);
     String fieldCode = getFieldCode(column);
+    if (isBuiltin(fieldCode)) {
+      return;
+    }
     if (isPreferNull(column)) {
       setNull(type, fieldCode, updateKey);
     } else if (reader.isNull(column)) {
@@ -124,6 +143,9 @@ public void stringColumn(Column column) {
         updateKey != null ? KintoneColumnType.SINGLE_LINE_TEXT : KintoneColumnType.MULTI_LINE_TEXT;
     KintoneColumnType type = KintoneColumnType.getType(option, defaultType);
     String fieldCode = getFieldCode(column);
+    if (isBuiltin(fieldCode)) {
+      return;
+    }
     if (isPreferNull(column)) {
       setNull(type, fieldCode, updateKey);
     } else if (reader.isNull(column)) {
@@ -142,6 +164,9 @@ public void timestampColumn(Column column) {
     UpdateKey updateKey = getUpdateKey(column);
     KintoneColumnType type = KintoneColumnType.getType(option, KintoneColumnType.DATETIME);
     String fieldCode = getFieldCode(column);
+    if (isBuiltin(fieldCode)) {
+      return;
+    }
     if (isPreferNull(column)) {
       setNull(type, fieldCode, updateKey);
     } else if (reader.isNull(column)) {
@@ -162,6 +187,9 @@ public void jsonColumn(Column column) {
         isDerived(column) ? KintoneColumnType.SUBTABLE : KintoneColumnType.MULTI_LINE_TEXT;
     KintoneColumnType type = KintoneColumnType.getType(option, defaultType);
     String fieldCode = getFieldCode(column);
+    if (isBuiltin(fieldCode)) {
+      return;
+    }
     if (isPreferNull(column)) {
       setNull(type, fieldCode, updateKey);
     } else if (reader.isNull(column)) {
@@ -171,6 +199,16 @@ public void jsonColumn(Column column) {
     }
   }
 
+  private void setIdValue(Id id, Column column) {
+    if (isPreferNull(column)) {
+      id.setValue(null);
+    } else if (reader.isNull(column)) {
+      id.setValue(0L);
+    } else {
+      id.setValue(reader.getLong(column));
+    }
+  }
+
   private void setNull(KintoneColumnType type, String fieldCode, UpdateKey updateKey) {
     if (updateKey != null) {
       type.setUpdateKey(updateKey, fieldCode);
@@ -265,8 +303,16 @@ private KintoneColumnOption getOption(Column column) {
     return options.get(column.getName());
   }
 
+  private Id getId(Column column) {
+    return column.getName().equals(updateKeyName) && updateKeyName.equals(Id.FIELD)
+        ? idOrUpdateKey.getId()
+        : null;
+  }
+
   private UpdateKey getUpdateKey(Column column) {
-    return updateKeyName != null && updateKeyName.equals(column.getName()) ? updateKey : null;
+    return column.getName().equals(updateKeyName) && !updateKeyName.equals(Id.FIELD)
+        ? idOrUpdateKey.getUpdateKey()
+        : null;
   }
 
   private boolean isReduced(Column column) {
@@ -284,4 +330,8 @@ private boolean isIgnoreNull(Column column) {
   private boolean isPreferNull(Column column) {
     return preferNulls && reader.isNull(column);
   }
+
+  private static boolean isBuiltin(String fieldCode) {
+    return BUILTIN_FIELD_CODES.contains(fieldCode);
+  }
 }

src/main/java/org/embulk/output/kintone/KintoneMode.java

@@ -1,6 +1,9 @@
 package org.embulk.output.kintone;
 
+import java.util.function.Consumer;
 import org.embulk.config.ConfigException;
+import org.embulk.output.kintone.record.Id;
+import org.embulk.output.kintone.record.Skip;
 import org.embulk.spi.Page;
 
 public enum KintoneMode {
@@ -13,35 +16,36 @@ public void validate(PluginTask task, KintoneClient client) {
     }
 
     @Override
-    public void add(Page page, KintonePageOutput output) {
+    public void add(Page page, Skip skip, KintonePageOutput output) {
       output.insertPage(page);
     }
   },
   UPDATE("update") {
     @Override
     public void validate(PluginTask task, KintoneClient client) {
-      if (!task.getUpdateKeyName().isPresent()) {
-        throw new ConfigException("When mode is update, require update_key.");
+      if (!task.getUpdateKeyName().isPresent() && client.getColumn(Id.FIELD) == null) {
+        throw new ConfigException("When mode is update, require update_key or id column.");
       }
-      client.validateUpdateKey(task.getUpdateKeyName().orElse(null));
+      client.validateIdOrUpdateKey(task.getUpdateKeyName().orElse(Id.FIELD));
     }
 
     @Override
-    public void add(Page page, KintonePageOutput output) {
-      output.updatePage(page);
+    public void add(Page page, Skip skip, KintonePageOutput output) {
+      Consumer<Page> consumer = skip == Skip.ALWAYS ? output::upsertPage : output::updatePage;
+      consumer.accept(page);
     }
   },
   UPSERT("upsert") {
     @Override
     public void validate(PluginTask task, KintoneClient client) {
-      if (!task.getUpdateKeyName().isPresent()) {
-        throw new ConfigException("When mode is upsert, require update_key.");
+      if (!task.getUpdateKeyName().isPresent() && client.getColumn(Id.FIELD) == null) {
+        throw new ConfigException("When mode is upsert, require update_key or id column.");
       }
-      client.validateUpdateKey(task.getUpdateKeyName().orElse(null));
+      client.validateIdOrUpdateKey(task.getUpdateKeyName().orElse(Id.FIELD));
     }
 
     @Override
-    public void add(Page page, KintonePageOutput output) {
+    public void add(Page page, Skip skip, KintonePageOutput output) {
       output.upsertPage(page);
     }
   };
@@ -53,7 +57,7 @@ public void add(Page page, KintonePageOutput output) {
 
   public abstract void validate(PluginTask task, KintoneClient client);
 
-  public abstract void add(Page page, KintonePageOutput output);
+  public abstract void add(Page page, Skip skip, KintonePageOutput output);
 
   @Override
   public String toString() {