TimestampPartitioning implements Partitioning { @Override public TaskSource configure(ConfigSource config, Schema schema, int outputTaskCount) { PartitioningTask task = config.loadConfig(PartitioningTask.class); Column column = findColumnByName(schema, task.getColumn()); if (!task.getTimeZone().equals(DateTimeZone.UTC)) { throw new ConfigException("Timestamp partitioner supports only UTC time zone for now"); } Unit.of(task.getUnit()); if (column.getType() instanceof TimestampType) { } else if (column.getType() instanceof LongType) { UnixTimestampUnit.of(task.getUnixTimestamp()); } else { throw new ConfigException( String.format("Partitioning column '%s' must be timestamp or long but got '%s'", column.getName(), column.getType())); } task.setTargetColumn(column); return task.dump(); }  @Override TaskSource configure(ConfigSource config, Schema schema, int outputTaskCount); @Override Partitioner newPartitioner(TaskSource taskSource);  }
@Test public void validateConfigSource() throws IOException { { ConfigSource config = runtime.getExec().newConfigSource() .set("column", "_c0").set("unit", "hour").set("timezone", "UTC"); Schema schema = Schema.builder().add("not_included", Types.TIMESTAMP).build(); try { tp.configure(config, schema, 0); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } } { ConfigSource config = runtime.getExec().newConfigSource() .set("column", "_c0").set("unit", "hour").set("timezone", "PDT"); Schema schema = Schema.builder().add("_c0", Types.TIMESTAMP).build(); try { tp.configure(config, schema, 0); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } } { ConfigSource config = runtime.getExec().newConfigSource() .set("column", "_c0").set("unit", "invalid").set("timezone", "UTC"); Schema schema = Schema.builder().add("_c0", Types.TIMESTAMP).build(); try { tp.configure(config, schema, 0); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } } { ConfigSource config = runtime.getExec().newConfigSource() .set("column", "_c0").set("unit", "hour").set("timezone", "UTC"); Schema schema = Schema.builder().add("_c0", Types.STRING).build(); try { tp.configure(config, schema, 0); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } } { ConfigSource config = runtime.getExec().newConfigSource() .set("column", "_c0").set("unit", "hour").set("timezone", "UTC").set("unix_timestamp_unit", "invalid"); Schema schema = Schema.builder().add("_c0", Types.LONG).build(); try { tp.configure(config, schema, 0); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } } }