@Test(expectedExceptions = IllegalStateException.class) public void testPartitionBeginBegoreTableBegin() throws Exception { SourceState state = new SourceState(); PartitionLevelWatermarker watermarker = new PartitionLevelWatermarker(state); Table table = mockTable("test_dataset_urn"); Partition partition = mockPartition(table, ImmutableList.of("")); watermarker.onPartitionProcessBegin(partition, 0l, 0l); }
@Override public void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime) { Preconditions.checkNotNull(partition); Preconditions.checkNotNull(partition.getTable()); if (!this.expectedHighWatermarks.hasPartitionWatermarks(tableKey(partition.getTable()))) { throw new IllegalStateException(String.format( "onPartitionProcessBegin called before onTableProcessBegin for table: %s, partitions: %s", tableKey(partition.getTable()), partitionKey(partition))); } Collection<String> droppedPartitions = Collections2.transform(AbstractAvroToOrcConverter.getDropPartitionsDDLInfo(partition), new Function<Map<String, String>, String>() { @Override public String apply(Map<String, String> input) { return PARTITION_VALUES_JOINER.join(input.values()); } }); this.expectedHighWatermarks.removePartitionWatermarks(tableKey(partition.getTable()), droppedPartitions); this.expectedHighWatermarks.addPartitionWatermark(tableKey(partition.getTable()), partitionKey(partition), partitionUpdateTime); }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime) { Preconditions.checkNotNull(partition); Preconditions.checkNotNull(partition.getTable()); if (!this.expectedHighWatermarks.hasPartitionWatermarks(tableKey(partition.getTable()))) { throw new IllegalStateException(String.format( "onPartitionProcessBegin called before onTableProcessBegin for table: %s, partitions: %s", tableKey(partition.getTable()), partitionKey(partition))); } Collection<String> droppedPartitions = Collections2.transform(AbstractAvroToOrcConverter.getDropPartitionsDDLInfo(partition), new Function<Map<String, String>, String>() { @Override public String apply(Map<String, String> input) { return PARTITION_VALUES_JOINER.join(input.values()); } }); this.expectedHighWatermarks.removePartitionWatermarks(tableKey(partition.getTable()), droppedPartitions); this.expectedHighWatermarks.addPartitionWatermark(tableKey(partition.getTable()), partitionKey(partition), partitionUpdateTime); } }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime) { Preconditions.checkNotNull(partition); Preconditions.checkNotNull(partition.getTable()); if (!this.expectedHighWatermarks.hasPartitionWatermarks(tableKey(partition.getTable()))) { throw new IllegalStateException(String.format( "onPartitionProcessBegin called before onTableProcessBegin for table: %s, partitions: %s", tableKey(partition.getTable()), partitionKey(partition))); } Collection<String> droppedPartitions = Collections2.transform(AbstractAvroToOrcConverter.getDropPartitionsDDLInfo(partition), new Function<Map<String, String>, String>() { @Override public String apply(Map<String, String> input) { return PARTITION_VALUES_JOINER.join(input.values()); } }); this.expectedHighWatermarks.removePartitionWatermarks(tableKey(partition.getTable()), droppedPartitions); this.expectedHighWatermarks.addPartitionWatermark(tableKey(partition.getTable()), partitionKey(partition), partitionUpdateTime); } PartitionLevelWatermarker(State state); }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime) { Preconditions.checkNotNull(partition); Preconditions.checkNotNull(partition.getTable()); if (!this.expectedHighWatermarks.hasPartitionWatermarks(tableKey(partition.getTable()))) { throw new IllegalStateException(String.format( "onPartitionProcessBegin called before onTableProcessBegin for table: %s, partitions: %s", tableKey(partition.getTable()), partitionKey(partition))); } Collection<String> droppedPartitions = Collections2.transform(AbstractAvroToOrcConverter.getDropPartitionsDDLInfo(partition), new Function<Map<String, String>, String>() { @Override public String apply(Map<String, String> input) { return PARTITION_VALUES_JOINER.join(input.values()); } }); this.expectedHighWatermarks.removePartitionWatermarks(tableKey(partition.getTable()), droppedPartitions); this.expectedHighWatermarks.addPartitionWatermark(tableKey(partition.getTable()), partitionKey(partition), partitionUpdateTime); } PartitionLevelWatermarker(State state); @Override void onTableProcessBegin(Table table, long tableProcessTime); @Override void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime); @Override LongWatermark getPreviousHighWatermark(Table table); @Override LongWatermark getPreviousHighWatermark(Partition partition); @Override void onGetWorkunitsEnd(List<WorkUnit> workunits); @Override LongWatermark getExpectedHighWatermark(Table table, long tableProcessTime); @Override LongWatermark getExpectedHighWatermark(Partition partition, long tableProcessTime, long partitionProcessTime); @Override void setActualHighWatermark(WorkUnitState wus); @VisibleForTesting static String tableKey(Table table); @VisibleForTesting static String partitionKey(Partition partition); }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime) { Preconditions.checkNotNull(partition); Preconditions.checkNotNull(partition.getTable()); if (!this.expectedHighWatermarks.hasPartitionWatermarks(tableKey(partition.getTable()))) { throw new IllegalStateException(String.format( "onPartitionProcessBegin called before onTableProcessBegin for table: %s, partitions: %s", tableKey(partition.getTable()), partitionKey(partition))); } Collection<String> droppedPartitions = Collections2.transform(AbstractAvroToOrcConverter.getDropPartitionsDDLInfo(partition), new Function<Map<String, String>, String>() { @Override public String apply(Map<String, String> input) { return PARTITION_VALUES_JOINER.join(input.values()); } }); this.expectedHighWatermarks.removePartitionWatermarks(tableKey(partition.getTable()), droppedPartitions); this.expectedHighWatermarks.addPartitionWatermark(tableKey(partition.getTable()), partitionKey(partition), partitionUpdateTime); } PartitionLevelWatermarker(State state); @Override void onTableProcessBegin(Table table, long tableProcessTime); @Override void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime); @Override LongWatermark getPreviousHighWatermark(Table table); @Override LongWatermark getPreviousHighWatermark(Partition partition); @Override void onGetWorkunitsEnd(List<WorkUnit> workunits); @Override LongWatermark getExpectedHighWatermark(Table table, long tableProcessTime); @Override LongWatermark getExpectedHighWatermark(Partition partition, long tableProcessTime, long partitionProcessTime); @Override void setActualHighWatermark(WorkUnitState wus); @VisibleForTesting static String tableKey(Table table); @VisibleForTesting static String partitionKey(Partition partition); static final String IS_WATERMARK_WORKUNIT_KEY; }