@Override
    public Output run(RunContext runContext) throws Exception {

        URI from = new URI(runContext.render(this.from));

        final PebbleFieldExtractor keyExtractor = getKeyExtractor(runContext);

        final Map<String, Long> index = new HashMap<>(); // can be replaced by small-footprint Map implementation

        // 1st iteration: build a map of key->offset
        try (final BufferedReader reader = newBufferedReader(runContext, from)) {
            long offset = 0L;
            String item;
            while ((item = reader.readLine()) != null) {
                String key = keyExtractor.apply(item);
                index.put(key, offset);
                offset++;
            }
        }

        // metrics
        long processedItemsTotal = 0L;
        long droppedItemsTotal = 0L;
        long numKeys = index.size();

        final Path path = runContext.workingDir().createTempFile(".ion");
        // 2nd iteration: write deduplicate
        try (final BufferedWriter writer = Files.newBufferedWriter(path);
             final BufferedReader reader = newBufferedReader(runContext, from)) {
            long offset = 0L;
            String item;
            while ((item = reader.readLine()) != null) {
                String key = keyExtractor.apply(item);
                Long lastOffset = index.get(key);
                if (lastOffset != null && lastOffset == offset) {
                    writer.write(item);
                    writer.newLine();
                } else {
                    droppedItemsTotal++;
                }
                offset++;
                processedItemsTotal++;
            }
        }
        URI uri = runContext.storage().putFile(path.toFile());
        index.clear();
        return Output
            .builder()
            .uri(uri)
            .numKeys(numKeys)
            .processedItemsTotal(processedItemsTotal)
            .droppedItemsTotal(droppedItemsTotal)
            .build();
    }
@Test
    void shouldDeduplicateFileGivenKeyExpression() throws Exception {
        // Given
        RunContext runContext = runContextFactory.of();

        List<KeyValue1> values = List.of(
            new KeyValue1("k1", "v1"),
            new KeyValue1("k2", "v1"),
            new KeyValue1("k3", "v1"),
            new KeyValue1("k1", "v2"),
            new KeyValue1("k2", "v2"),
            new KeyValue1("k2", null),
            new KeyValue1("k3", "v2"),
            new KeyValue1("k1", "v3")
        );

        DeduplicateItems task = DeduplicateItems
            .builder()
            .from(generateKeyValueFile(values, runContext).toString())
            .expr(" {{ key }} ")
            .build();

        // When
        DeduplicateItems.Output output = task.run(runContext);

        // Then
        Assertions.assertNotNull(output);
        Assertions.assertNotNull(output.getUri());
        Assertions.assertEquals(3, output.getNumKeys());
        Assertions.assertEquals(5, output.getDroppedItemsTotal());
        Assertions.assertEquals(8, output.getProcessedItemsTotal());

        List<KeyValue1> expected = List.of(
            new KeyValue1("k2", null),
            new KeyValue1("k3", "v2"),
            new KeyValue1("k1", "v3")
        );
        assertSimpleCompactedFile(runContext, output, expected, KeyValue1.class);
    }