fix(lake): use watermark for offset in lake sink path

[rawkode]

Summary

The lake sink code path was hardcoding offset to 0, causing all parquet files to have the same name (based on partition-offset). This caused Iceberg to reject subsequent writes as duplicate files after the first batch.

Problem

In tansu-storage/src/dynostore.rs, the lake sink branch had:

let offset = 0;  // Always 0!

This offset is used in the parquet filename, so every batch tried to write to the same file.

Fix

Reuse the same watermark logic from the normal storage path to calculate proper incrementing offsets:

let watermark = self.watermarks.lock()...;
let offset = watermark.with_mut(&self.object_store, |watermark| {
    let offset = watermark.high.unwrap_or_default();
    // update watermark...
    Ok(offset)
}).await?;

Test Plan

• Existing tests pass (cargo test -p tansu-storage)
• Deploy with lake sink enabled, verify multiple batches write to different parquet files

[Copilot]

Pull request overview

This PR fixes a critical bug in the lake sink code path where the offset was hardcoded to 0, causing all parquet files to have identical names. This resulted in Iceberg rejecting subsequent writes as duplicates after the first batch. The fix reuses the same watermark-based offset calculation logic from the normal storage path to generate proper incrementing offsets.

• Replaces hardcoded offset = 0 with watermark-based offset calculation
• Ensures each batch writes to a uniquely named parquet file based on incrementing offset
• Reuses existing watermark management pattern from the normal storage path

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The watermark calculation logic here (lines 729-754) is duplicated from the normal storage path (lines 856-881). Consider extracting this into a helper method to avoid code duplication and ensure both paths remain consistent if the watermark logic needs to be updated in the future.