@Test public void testDoWriteOperationWithoutUserDefinedBulkInsertPartitioner() throws HoodieException { when(hoodieWriteClient.getConfig()).thenReturn(config); DataSourceUtils.doWriteOperation(hoodieWriteClient, hoodieRecords, "test-time", WriteOperationType.BULK_INSERT); verify(hoodieWriteClient, times(1)).bulkInsert(any(hoodieRecords.getClass()), anyString(), optionCaptor.capture()); assertThat(optionCaptor.getValue(), is(equalTo(Option.empty()))); }
public static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords, String instantTime, WriteOperationType operation) throws HoodieException { switch (operation) { case BULK_INSERT: Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner = createUserDefinedBulkInsertPartitioner(client.getConfig()); return new HoodieWriteResult(client.bulkInsert(hoodieRecords, instantTime, userDefinedBulkInsertPartitioner)); case INSERT: return new HoodieWriteResult(client.insert(hoodieRecords, instantTime)); case UPSERT: return new HoodieWriteResult(client.upsert(hoodieRecords, instantTime)); case INSERT_OVERWRITE: return client.insertOverwrite(hoodieRecords, instantTime); default: throw new HoodieException("Not a valid operation type for doWriteOperation: " + operation.toString()); } }
DataSourceUtils { public static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords, String instantTime, WriteOperationType operation) throws HoodieException { switch (operation) { case BULK_INSERT: Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner = createUserDefinedBulkInsertPartitioner(client.getConfig()); return new HoodieWriteResult(client.bulkInsert(hoodieRecords, instantTime, userDefinedBulkInsertPartitioner)); case INSERT: return new HoodieWriteResult(client.insert(hoodieRecords, instantTime)); case UPSERT: return new HoodieWriteResult(client.upsert(hoodieRecords, instantTime)); case INSERT_OVERWRITE: return client.insertOverwrite(hoodieRecords, instantTime); default: throw new HoodieException("Not a valid operation type for doWriteOperation: " + operation.toString()); } } }
DataSourceUtils { public static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords, String instantTime, WriteOperationType operation) throws HoodieException { switch (operation) { case BULK_INSERT: Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner = createUserDefinedBulkInsertPartitioner(client.getConfig()); return new HoodieWriteResult(client.bulkInsert(hoodieRecords, instantTime, userDefinedBulkInsertPartitioner)); case INSERT: return new HoodieWriteResult(client.insert(hoodieRecords, instantTime)); case UPSERT: return new HoodieWriteResult(client.upsert(hoodieRecords, instantTime)); case INSERT_OVERWRITE: return client.insertOverwrite(hoodieRecords, instantTime); default: throw new HoodieException("Not a valid operation type for doWriteOperation: " + operation.toString()); } }  }
DataSourceUtils { public static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords, String instantTime, WriteOperationType operation) throws HoodieException { switch (operation) { case BULK_INSERT: Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner = createUserDefinedBulkInsertPartitioner(client.getConfig()); return new HoodieWriteResult(client.bulkInsert(hoodieRecords, instantTime, userDefinedBulkInsertPartitioner)); case INSERT: return new HoodieWriteResult(client.insert(hoodieRecords, instantTime)); case UPSERT: return new HoodieWriteResult(client.upsert(hoodieRecords, instantTime)); case INSERT_OVERWRITE: return client.insertOverwrite(hoodieRecords, instantTime); default: throw new HoodieException("Not a valid operation type for doWriteOperation: " + operation.toString()); } }  static String getTablePath(FileSystem fs, Path[] userProvidedPaths); static KeyGenerator createKeyGenerator(TypedProperties props); static AbstractHoodieDateTimeParser createDateTimeParser(TypedProperties props, String parserClass); static HoodieRecordPayload createPayload(String payloadClass, GenericRecord record, Comparable orderingVal); static HoodieRecordPayload createPayload(String payloadClass, GenericRecord record); static void checkRequiredProperties(TypedProperties props, List<String> checkPropNames); static HoodieWriteConfig createHoodieConfig(String schemaStr, String basePath,
      String tblName, Map<String, String> parameters); static SparkRDDWriteClient createHoodieClient(JavaSparkContext jssc, String schemaStr, String basePath,
                                                       String tblName, Map<String, String> parameters); static String getCommitActionType(WriteOperationType operation, HoodieTableType tableType); static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords,
                                                   String instantTime, WriteOperationType operation); static HoodieWriteResult doDeleteOperation(SparkRDDWriteClient client, JavaRDD<HoodieKey> hoodieKeys,
      String instantTime); static HoodieRecord createHoodieRecord(GenericRecord gr, Comparable orderingVal, HoodieKey hKey,
      String payloadClass); static HoodieRecord createHoodieRecord(GenericRecord gr, HoodieKey hKey,
                                                String payloadClass); @SuppressWarnings("unchecked") static JavaRDD<HoodieRecord> dropDuplicates(JavaSparkContext jssc, JavaRDD<HoodieRecord> incomingHoodieRecords,
      HoodieWriteConfig writeConfig); @SuppressWarnings("unchecked") static JavaRDD<HoodieRecord> dropDuplicates(JavaSparkContext jssc, JavaRDD<HoodieRecord> incomingHoodieRecords,
      Map<String, String> parameters); static HiveSyncConfig buildHiveSyncConfig(TypedProperties props, String basePath, String baseFileFormat); }
DataSourceUtils { public static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords, String instantTime, WriteOperationType operation) throws HoodieException { switch (operation) { case BULK_INSERT: Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner = createUserDefinedBulkInsertPartitioner(client.getConfig()); return new HoodieWriteResult(client.bulkInsert(hoodieRecords, instantTime, userDefinedBulkInsertPartitioner)); case INSERT: return new HoodieWriteResult(client.insert(hoodieRecords, instantTime)); case UPSERT: return new HoodieWriteResult(client.upsert(hoodieRecords, instantTime)); case INSERT_OVERWRITE: return client.insertOverwrite(hoodieRecords, instantTime); default: throw new HoodieException("Not a valid operation type for doWriteOperation: " + operation.toString()); } }  static String getTablePath(FileSystem fs, Path[] userProvidedPaths); static KeyGenerator createKeyGenerator(TypedProperties props); static AbstractHoodieDateTimeParser createDateTimeParser(TypedProperties props, String parserClass); static HoodieRecordPayload createPayload(String payloadClass, GenericRecord record, Comparable orderingVal); static HoodieRecordPayload createPayload(String payloadClass, GenericRecord record); static void checkRequiredProperties(TypedProperties props, List<String> checkPropNames); static HoodieWriteConfig createHoodieConfig(String schemaStr, String basePath,
      String tblName, Map<String, String> parameters); static SparkRDDWriteClient createHoodieClient(JavaSparkContext jssc, String schemaStr, String basePath,
                                                       String tblName, Map<String, String> parameters); static String getCommitActionType(WriteOperationType operation, HoodieTableType tableType); static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client, JavaRDD<HoodieRecord> hoodieRecords,
                                                   String instantTime, WriteOperationType operation); static HoodieWriteResult doDeleteOperation(SparkRDDWriteClient client, JavaRDD<HoodieKey> hoodieKeys,
      String instantTime); static HoodieRecord createHoodieRecord(GenericRecord gr, Comparable orderingVal, HoodieKey hKey,
      String payloadClass); static HoodieRecord createHoodieRecord(GenericRecord gr, HoodieKey hKey,
                                                String payloadClass); @SuppressWarnings("unchecked") static JavaRDD<HoodieRecord> dropDuplicates(JavaSparkContext jssc, JavaRDD<HoodieRecord> incomingHoodieRecords,
      HoodieWriteConfig writeConfig); @SuppressWarnings("unchecked") static JavaRDD<HoodieRecord> dropDuplicates(JavaSparkContext jssc, JavaRDD<HoodieRecord> incomingHoodieRecords,
      Map<String, String> parameters); static HiveSyncConfig buildHiveSyncConfig(TypedProperties props, String basePath, String baseFileFormat);  }