[lake/paimon] Compare paimon schema and Fluss schema before alter table. #2331

loserwang1024 · 2026-01-08T13:22:35Z

Purpose

Linked issue: close #2330

Brief change log

Tests

API and Format

Documentation

Copilot

Pull request overview

This pull request enhances the Paimon lake catalog integration by comparing Paimon and Fluss schemas before altering tables. This prevents schema inconsistencies that can occur when a previous alter operation partially succeeds (e.g., Paimon succeeds but Fluss fails).

Key Changes:

Added getFlussSchema() method to LakeCatalog.Context interface to provide current Fluss schema during alter operations
Implemented schema validation logic in PaimonLakeCatalog to detect and handle schema mismatches between Paimon and Fluss
Updated MetadataManager and CoordinatorService to pass schema information via the context

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java	Added `getFlussSchema()` method to Context interface
fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java	Updated test context to support schema parameter
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Updated to create context with schema, and pass FlussPrincipal instead of pre-created context
fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java	Changed method signatures to accept FlussPrincipal and create context locally with current schema
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java	Added schema comparison logic to detect and handle retry scenarios
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java	Enhanced tests to cover schema mismatch scenarios and retry logic

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

Copilot · 2026-01-08T13:32:50Z

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

+            // columns or other operations next time.
+            String errorMsg =
+                    String.format(
+                            "Paimon table has more columns (%d) than Fluss schema (%d), thus need to add the diff columns at once rather than other table changes %s.",


The error message has a grammatical issue. "thus need to add" should be "thus needs to add" or better yet, rephrase to "therefore you need to add the diff columns all at once, rather than applying other table changes" for better clarity.

Suggested change

"Paimon table has more columns (%d) than Fluss schema (%d), thus need to add the diff columns at once rather than other table changes %s.",

"Paimon table has more columns (%d) than Fluss schema (%d); therefore you need to add the diff columns all at once, rather than applying other table changes: %s.",

Copilot · 2026-01-08T13:32:51Z

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

+            // if paimon column size is greater than expected fluss column size, meaning last add
+            // columns are failed.
+            // Thus, this time must be retried to keep the schema same, only then can add new
+            // columns or other operations next time.


The comment states "meaning last add columns are failed" which should be "meaning the last add columns operation failed" for better grammar. Also, "This time must be retried" could be clearer as "This operation must retry adding the same columns".

Suggested change

// if paimon column size is greater than expected fluss column size, meaning last add

// columns are failed.

// Thus, this time must be retried to keep the schema same, only then can add new

// columns or other operations next time.

// If Paimon column size is greater than the expected Fluss column size, it means the

// last add columns operation failed.

// Thus, this operation must retry adding the same columns to keep the schemas the

// same before adding new columns or performing other operations.

Copilot · 2026-01-08T13:32:51Z

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

+            for (int i = 0; i < paimonFields.size() - flussColumns.size(); i++) {
+                DataField paimonDataField = paimonFields.get(i + flussColumns.size());
+                TableChange tableChange = tableChanges.get(i);
+                if (!(tableChange instanceof TableChange.AddColumn
+                        && ((TableChange.AddColumn) tableChange).getPosition()
+                                == TableChange.ColumnPosition.last()
+                        && isColumnAlreadyExists(
+                                paimonDataField, (TableChange.AddColumn) tableChange))) {
+                    throw new InvalidAlterTableException(errorMsg);
                }
            }


The method checkAndFilterDuplicateTableChanges only handles TableChange.AddColumn operations in its retry logic (lines 186-196). If the method is called with property changes (like TableChange.SetOption or TableChange.ResetOption) when Paimon has more columns than Fluss, it will incorrectly throw an error at line 194. Consider checking if all tableChanges are AddColumn operations before entering the retry logic, or handle non-AddColumn changes appropriately.

Copilot · 2026-01-08T13:32:52Z

fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java

+        /**
+         * Get the current schema of fluss.
+         *
+         * @return the current schema of fluss,


The new method getFlussSchema() should include a @since annotation to document when it was added to the API, consistent with other methods in the interface that have @since annotations.

Suggested change

* @return the current schema of fluss,

* @return the current schema of fluss,

* @since 0.10

Copilot · 2026-01-08T13:32:52Z

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

+                                paimonFields.size(), flussColumns.size()));
+            }
+
+            // Validate  schema compatibility


There's an extra space in the comment. Should be "Validate schema compatibility" instead of "Validate schema compatibility".

Suggested change

// Validate schema compatibility

// Validate schema compatibility

Copilot · 2026-01-08T13:32:53Z

fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java

+         * Get the current schema of fluss.
+         *
+         * @return the current schema of fluss,


The JavaDoc comment is incomplete. It has "@return the current schema of fluss," but the sentence is not finished and has a trailing comma. Consider completing the description, for example: "Get the current schema of Fluss. @return the current schema of Fluss".

Suggested change

* Get the current schema of fluss.

*

* @return the current schema of fluss,

* Get the current schema of Fluss.

*

* @return the current schema of Fluss.

beryllw · 2026-01-09T09:36:53Z

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

+                TableChange tableChange = tableChanges.get(i);
+                if (!(tableChange instanceof TableChange.AddColumn
+                        && ((TableChange.AddColumn) tableChange).getPosition()
+                                == TableChange.ColumnPosition.last()


Maybe we could perform this check earlier—it’s not related to the existing Paimon schema. Also, it’s already validated here:

fluss/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java

Lines 101 to 104 in 256bd7f

TableChange.ColumnPosition position = addColumn.getPosition();

if (position != TableChange.ColumnPosition.last()) {

throw new IllegalArgumentException("Only support addColumn column at last now.");

}

beryllw · 2026-01-09T09:42:27Z

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

+                                == TableChange.ColumnPosition.last()
+                        && isColumnAlreadyExists(
+                                paimonDataField, (TableChange.AddColumn) tableChange))) {
+                    throw new InvalidAlterTableException(errorMsg);


Maybe we could split the checks and provide distinct error messages to help users more easily identify issues related to the Paimon schema.

[lake/paimon] Compare paimon schema and Fluss schema before alter table.

dcc741c

loserwang1024 requested a review from Copilot January 8, 2026 13:22

Copilot started reviewing on behalf of loserwang1024 January 8, 2026 13:23 View session

Copilot AI reviewed Jan 8, 2026

View reviewed changes

beryllw reviewed Jan 9, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[lake/paimon] Compare paimon schema and Fluss schema before alter table. #2331

[lake/paimon] Compare paimon schema and Fluss schema before alter table. #2331

loserwang1024 commented Jan 8, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Jan 8, 2026

Uh oh!

Copilot AI Jan 8, 2026

Uh oh!

Copilot AI Jan 8, 2026

Uh oh!

Copilot AI Jan 8, 2026

Uh oh!

Copilot AI Jan 8, 2026

Uh oh!

Copilot AI Jan 8, 2026

Uh oh!

beryllw Jan 9, 2026

Uh oh!

beryllw Jan 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	"Paimon table has more columns (%d) than Fluss schema (%d), thus need to add the diff columns at once rather than other table changes %s.",
	"Paimon table has more columns (%d) than Fluss schema (%d); therefore you need to add the diff columns all at once, rather than applying other table changes: %s.",

	* @return the current schema of fluss,
	* @return the current schema of fluss,
	* @since 0.10

	// Validate schema compatibility
	// Validate schema compatibility

	TableChange.ColumnPosition position = addColumn.getPosition();
	if (position != TableChange.ColumnPosition.last()) {
	throw new IllegalArgumentException("Only support addColumn column at last now.");
	}

[lake/paimon] Compare paimon schema and Fluss schema before alter table. #2331

Are you sure you want to change the base?

[lake/paimon] Compare paimon schema and Fluss schema before alter table. #2331

Conversation

loserwang1024 commented Jan 8, 2026

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

beryllw Jan 9, 2026

Choose a reason for hiding this comment

Uh oh!

beryllw Jan 9, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants