[lake/tiering] add table dropped handling and nullable write results by beryllw · Pull Request #2920 · apache/fluss

beryllw · 2026-03-24T08:48:40Z

Purpose

Linked issue: close #2498

This PR improves Lake Tiering's handling of dropped tables and adds cancellation support for tiering operations.

Brief change log

Dropped Table Handling

When a table is dropped during active tiering, the TieringSplitReader now gracefully completes all
in-progress splits with empty results instead of failing
Added handleTableDropped() to mark dropped tables and trigger force completion
Dropped tables are properly cleaned up from LakeTableTieringManager to prevent resource leaks

Cancellation Support

Added cancelled flag to TableBucketWriteResult to distinguish between normal and cancelled tiering rounds
Committers skip commit processing for cancelled results and report back to the coordinator
Lake writers are closed without completing when tiering is cancelled, discarding uncommitted data

Tests

API and Format

Documentation

leekeiabstraction

TY for the PR, left a comment.

...flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResult.java

luoyuxia

Haven't finish review, but left some comments

...uss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

- Removed redundant TableNotExistException catch block that re-threw the same exception - Consolidated exception handling to reduce code duplication - Maintained same error propagation behavior for all other exceptions - Improved code readability by removing unnecessary exception wrapping

luoyuxia

@beryllw Thanks for the pr. Left some comments. PTAL

luoyuxia · 2026-03-31T07:15:49Z

fluss-common/src/main/java/org/apache/fluss/lake/writer/LakeWriter.java

     * Completes the writing process and returns the write result.
     *
-     * @return the write result
+     * @return the write result, or null if no data was written (empty write scenario)


I'm curious about when it'll be null(not data written)

luoyuxia · 2026-03-31T07:22:35Z

.../src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

+                // Check if the error indicates table doesn't exist or tiering epoch is fenced
+                // (which happens when table is dropped and recreated)
+                if (errors == Errors.TABLE_NOT_EXIST
+                        || errors == Errors.UNKNOWN_TABLE_OR_BUCKET_EXCEPTION


not to handle UNKNOWN_TABLE_OR_BUCKET_EXCEPTION and FENCED_TIERING_EPOCH_EXCEPTION

luoyuxia · 2026-03-31T07:27:57Z

.../src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

+    @VisibleForTesting
+    protected void handleTableDropped(long tableId) {
+        // Remove from tiering table epochs
+        Long tieringEpoch = tieringTableEpochs.remove(tableId);


I feel like it's error-prone in here to remove from tieringTableEpochs, and put failedTableEpochs.
Another way is consider table drop as another fail tiering event (send FailedTieringEvent), so that the whole path is unified.

luoyuxia · 2026-03-31T07:30:49Z

...on/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResultSerializer.java

            ThreadLocal.withInitial(() -> new DataOutputSerializer(64));

-    private static final int CURRENT_VERSION = 1;
+    private static final int CURRENT_VERSION = 2;


Acutally, we may not need to bump the version since it's only needed for shuffle... For code clean, we can still keep it as CURRENT_VERSION = 1

luoyuxia · 2026-03-31T07:32:20Z

...-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java


                    @Override
-                    public void wakeUp() {
-                        // do nothing


please still keep the comment

luoyuxia · 2026-03-31T07:47:27Z

...uss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

+        }
+        lakeWriters.clear();
+
+        // Also handle pending snapshot splits for this table


we don't need to handle pending snapshot splits again since it must be in currentTableSplitsByBucket. See method

private void addSplitToCurrentTable(TieringSplit split) { this.currentTableSplitsByBucket.put(split.getTableBucket(), split); if (split.isTieringSnapshotSplit()) { this.currentPendingSnapshotSplits.add((TieringSnapshotSplit) split); } else if (split.isTieringLogSplit()) { subscribeLog((TieringLogSplit) split); } }

luoyuxia · 2026-03-31T07:52:09Z

...uss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

+            }
+        }
+
+        // Close any remaining lake writers that don't have corresponding splits


do we really need it?
When it happen that we have lake writer but no corresponding splits?

luoyuxia · 2026-03-31T07:53:26Z

...e/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java

        } catch (Exception e) {
            throw new IOException("Failed to complete Paimon write.", e);
        }
+        if (commitMessage == null) {


iceberg and lance also need to do same handle?

luoyuxia · 2026-03-31T07:56:49Z

...uss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

+                currentTableSplitsByBucket.put(split.getTableBucket(), split);
+            }
+            LOG.info(
+                    "Skipping RPC for dropped table {} (path: {}), will force complete in next fetch cycle.",


nit:

Suggested change

"Skipping RPC for dropped table {} (path: {}), will force complete in next fetch cycle.",

Table {} is already marked dropped; skip opening table {} and force complete in next fetch cycle.

Skipping RPC looks strange to me

beryllw force-pushed the tiering-droptbl branch 3 times, most recently from c74237d to ef27034 Compare March 25, 2026 05:04

beryllw added 2 commits March 25, 2026 17:55

[lake/tiering] add table dropped handling and nullable write results

827ea64

properly handle TableNotExistException in metadata operations

2e0a904

beryllw force-pushed the tiering-droptbl branch from ef27034 to 2e0a904 Compare March 25, 2026 09:56

leekeiabstraction reviewed Mar 25, 2026

View reviewed changes

...flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResult.java Show resolved Hide resolved

luoyuxia reviewed Mar 26, 2026

View reviewed changes

...uss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java Show resolved Hide resolved

...uss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java Outdated Show resolved Hide resolved

beryllw added 3 commits March 30, 2026 22:46

remove TableNotExistException handling from metadata updater

ca316be

correct version usage in serializer and simplify table drop logic

1c15b83

beryllw requested a review from luoyuxia March 31, 2026 02:13

add test for table dropped handling in source enumerator

ef52bd9

luoyuxia reviewed Mar 31, 2026

View reviewed changes

	"Skipping RPC for dropped table {} (path: {}), will force complete in next fetch cycle.",
	Table {} is already marked dropped; skip opening table {} and force complete in next fetch cycle.

Conversation

beryllw commented Mar 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

leekeiabstraction left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

beryllw commented Mar 24, 2026 •

edited

Loading